Python开发求助：为现有CSV处理脚本新增嵌套循环实现按批次读取未处理行

核心逻辑说明 #

• 用本地JSON文件持久化存储每个CSV的已处理进度，解决脚本重启后状态丢失的问题
• 所有可调参数统一放在代码开头，无需修改核心逻辑就能调整运行规则
• 完全兼容现有业务处理流程，不改动原有的函数调用逻辑
• 增加文件末尾容错，不会因为CSV剩余行数不足批次大小触发报错

先修正现有代码的运行错误 #

你当前的代码存在3个会直接触发运行失败的问题，先做基础修正：

• path_id_map列表内的字典条目末尾缺少逗号，会触发Python语法错误
• 业务函数调用时重复写了B_id=B_id=pair['id']，语法不合法
• 没有导入time模块，调用time.sleep()会抛出模块不存在的错误

完整可运行的增量处理代码 #

import json
import os
import time
from abc.zzz import xyz  # 保留你原有业务模块的导入

# ========== 自定义配置区（可直接修改参数） ==========
PROCESS_BATCH_SIZE = 2  # 单文件每次处理的行数，匹配你要的单批次2行需求
LINE_DELAY = 25          # 单行处理完成后的等待时长
FILE_SWITCH_DELAY = 30   # 单文件处理完切下一个文件的等待时长
STATE_STORE_PATH = "csv_process_state.json"  # 进度存储文件的路径
# ==================================================

path_id_map = [
    {'path':'file1.csv', 'id': '12345678'},
    {'path':'file2.csv', 'id': '44556677'},
    {'path':'file3.csv', 'id': '33377799'},
    {'path':'file4.csv', 'id': '66221144'}
]
s_id = None

# 读取上次的处理进度，首次运行自动初始化默认值
def load_state():
    if not os.path.exists(STATE_STORE_PATH):
        return {}
    with open(STATE_STORE_PATH, 'r', encoding='utf-8') as f:
        return json.load(f)

# 处理完成后写入最新进度
def save_state(state):
    with open(STATE_STORE_PATH, 'w', encoding='utf-8') as f:
        json.dump(state, f, indent=2, ensure_ascii=False)

# 加载全局处理进度
current_state = load_state()

for csv_info in path_id_map:
    csv_path = csv_info['path']
    # 获取当前文件上次处理到的最后行号，首次运行默认从第1条数据行开始（跳过表头）
    last_done_row = current_state.get(csv_path, 0)
    process_start_row = last_done_row + 1

    with open(csv_path, 'r', encoding='utf-8') as f:
        next(f)  # 跳过表头行，和你原有逻辑保持一致
        # 将文件指针移动到本次要处理的起始行
        for _ in range(process_start_row - 1):
            next(f)

        processed_count = 0
        # 循环处理直到达到批次大小，或读到文件末尾
        while processed_count < PROCESS_BATCH_SIZE:
            try:
                line = next(f)
            except StopIteration:
                # 读到文件最后一行，无更多数据，直接跳出循环
                break

            # 以下为原有业务处理逻辑，未做改动
            img_url, title_1, desc_1, link_1 = map(str.strip, line.split(';'))
            zzz.func1(
                img_url=img_url,
                title_1=title_1,
                desc_1=desc_1,
                link_1=link_1,
                B_id=csv_info['id'],
                s_id=s_id
            )
            time.sleep(LINE_DELAY)

            processed_count += 1
            last_done_row += 1

    # 更新当前文件的最新进度
    current_state[csv_path] = last_done_row
    # 单文件处理完成，延迟后切换下一个文件
    time.sleep(FILE_SWITCH_DELAY)

# 所有文件遍历完成，持久化保存最新进度
save_state(current_state)

实际运行效果 #

• 第一次运行：系统找不到进度文件，默认从每个CSV的第1、2条数据行开始处理，处理完成后自动生成csv_process_state.json，记录每个文件已处理到第2行
• 第二次运行：自动读取进度文件内容，每个CSV从第3行开始读取2行数据（即第3、4行）处理，完成后更新进度到第4行
• 后续每次运行：自动向后滚动到对应的未处理行批次，直到某个CSV所有数据行处理完成，下次运行时会自动跳过该文件，不会重复处理

注意事项 #

• 不要手动删除csv_process_state.json文件，删除后系统会默认重新从每个CSV的第1行开始处理
• 调整处理规则时，只需要修改代码开头自定义配置区的参数即可，无需改动核心逻辑
• 不要随意修改已处理过的CSV行内容，否则进度记录的行号会和实际内容错位

内容的提问来源于stack exchange，提问作者Dave99