Python函数怎样用生成器函数实现断点续传 Python函数生成器断点续传的简单教程​

import os
import time

def simulate_download_generator(file_size, chunk_size, start_offset=0):
    """
    模拟一个文件下载生成器，支持断点续传。
    file_size: 文件总大小
    chunk_size: 每次下载的块大小
    start_offset: 从哪个字节偏移量开始下载
    """
    current_offset = start_offset
    print(f"开始下载，从偏移量 {start_offset} 处恢复...")
    while current_offset < file_size:
        # 模拟网络延迟或文件读取
        time.sleep(0.1) 

        # 计算当前块的大小，确保不超过文件末尾
        actual_chunk_size = min(chunk_size, file_size - current_offset)

        if actual_chunk_size <= 0:
            break # 已经下载完成

        # 模拟下载数据
        downloaded_data = f"模拟数据块 {current_offset}-{current_offset + actual_chunk_size - 1}"

        current_offset += actual_chunk_size

        # 每次yield当前进度，外部可以保存这个进度
        yield current_offset 

        # 可以在这里加入模拟的错误或中断
        # if current_offset > file_size / 2 and current_offset < file_size / 2 + chunk_size:
        #     raise Exception("模拟网络中断")

    print("下载完成。")

# --- 外部调用和状态管理 ---
if __name__ == "__main__":
    total_file_size = 1000 # 模拟文件大小1000字节
    download_chunk_size = 100 # 每次下载100字节
    progress_file = "download_progress.txt"

    last_saved_offset = 0
    if os.path.exists(progress_file):
        with open(progress_file, "r") as f:
            try:
                last_saved_offset = int(f.read().strip())
                print(f"发现上次中断的进度：从 {last_saved_offset} 字节开始续传。")
            except ValueError:
                print("进度文件损坏或为空，从头开始。")
                last_saved_offset = 0

    download_gen = simulate_download_generator(total_file_size, download_chunk_size, last_saved_offset)

    try:
        for current_progress in download_gen:
            print(f"当前已下载：{current_progress} 字节")
            # 每次yield后，都把当前进度保存起来
            with open(progress_file, "w") as f:
                f.write(str(current_progress))

            # 模拟用户强制中断或程序崩溃
            if current_progress >= total_file_size / 2 and last_saved_offset < total_file_size / 2:
                print("模拟程序意外中断...")
                # 清理一下，模拟下次从中断点恢复
                # raise KeyboardInterrupt # 实际中断会抛出异常
                break # 直接跳出循环，模拟中断

    except Exception as e:
        print(f"下载过程中发生错误：{e}")
    except KeyboardInterrupt:
        print("\n用户中断下载。")
    finally:
        # 最终状态，可以在这里决定是否删除进度文件
        if current_progress >= total_file_size:
            print("文件已完全下载，清理进度文件。")
            if os.path.exists(progress_file):
                os.remove(progress_file)
        else:
            print(f"下载未完成，当前进度 {current_progress} 已保存。")

yield

yield

import os

def process_large_log_file(file_path, start_line=0):
    """
    一个生成器函数，用于逐行处理大日志文件，支持从指定行开始续传。
    file_path: 日志文件路径
    start_line: 从文件的哪一行开始处理 (0-indexed)
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"文件不存在: {file_path}")

    current_line_num = 0
    with open(file_path, 'r', encoding='utf-8') as f:
        # 跳过已处理的行
        for _ in range(start_line):
            next(f, None) # 尝试读取下一行，如果文件结束则返回None
            current_line_num += 1
            if current_line_num >= start_line and f.tell() == os.fstat(f.fileno()).st_size:
                # 如果跳过行数已经达到，但文件已读完，说明start_line超出了文件总行数
                print(f"警告：起始行 {start_line} 超出文件总行数，没有内容可处理。")
                return # 提前结束生成器

        # 从指定行开始处理
        for line in f:
            # 模拟处理一行数据
            processed_data = f"处理了第 {current_line_num} 行: {line.strip()}"

            # 每次处理完一行，yield当前的行号，作为断点
            yield current_line_num, processed_data
            current_line_num += 1

# --- 外部调用和状态管理示例 ---
if __name__ == "__main__":
    log_file = "sample_log.txt"
    progress_save_file = "log_process_progress.txt"

    # 创建一个模拟的日志文件
    with open(log_file, "w", encoding="utf-8") as f:
        for i in range(50):
            f.write(f"这是日志文件的第 {i} 行。\n")

    last_processed_line = 0
    if os.path.exists(progress_save_file):
        with open(progress_save_file, "r") as f:
            try:
                last_processed_line = int(f.read().strip()) + 1 # 从下一行开始
                print(f"发现上次中断的进度：从第 {last_processed_line} 行开始续传。")
            except ValueError:
                print("进度文件损坏或为空，从头开始处理。")
                last_processed_line = 0

    log_processor = process_large_log_file(log_file, start_line=last_processed_line)

    try:
        for line_num, data in log_processor:
            print(data)
            # 每次处理完，保存当前行号
            with open(progress_save_file, "w") as f:
                f.write(str(line_num))

            # 模拟处理到一半中断
            if line_num == 20 and last_processed_line <= 20:
                print("模拟程序在第20行中断...")
                break # 模拟中断，跳出循环

    except Exception as e:
        print(f"处理日志文件时发生错误：{e}")
    except KeyboardInterrupt:
        print("\n用户中断处理。")
    finally:
        # 检查是否处理完成
        with open(log_file, "r", encoding="utf-8") as f:
            total_lines = sum(1 for _ in f)

        if last_processed_line > 0 and line_num >= total_lines -1: # line_num 是0-indexed
            print("日志文件已完全处理，清理进度文件。")
            if os.path.exists(progress_save_file):
                os.remove(progress_save_file)
        else:
            print(f"日志处理未完成，当前进度（行号） {line_num} 已保存。")

process_large_log_file

start_line

yield