批量文件处理的核心是理清“谁在什么时候对什么文件做了什么”,需明确处理边界、分层解耦逻辑、提供可控执行反馈、以配置驱动替代硬编码。

用python做批量文件处理,核心不是写多少代码,而是理清“谁在什么时候对什么文件做了什么”。系统性不强,脚本容易变成一次性玩具;流程不优化,小任务也会卡在IO、路径、编码或异常上。
明确处理边界:从“所有文件”到“该处理的文件”
盲目遍历整个目录树是常见误区。真正需要处理的往往只是特定类型、特定时间范围、或满足命名规则的文件。
- 用
pathlib.Path.rglob()替代os.walk(),链式调用更清晰,比如list(p.rglob("*.log"))直接获取所有日志文件 - 加时间过滤:读取
stat().st_mtime,转为datetime后对比,避免用os.listdir()再手动筛选 - 支持白名单/黑名单模式:配置一个
include_patterns = ["report_2024*.csv", "data_v2_*.xlsx"],用fnmatch或正则匹配,比硬编码路径更灵活
分层处理逻辑:解耦读、转、写、验
把“打开→解析→修改→保存→校验”揉成一个函数,调试难、复用差、出错难定位。应拆成可单独测试的小单元。
- 读取层:统一用
with open(...)或pandas.read_*(),自动处理编码(如encoding="utf-8-sig"防bom)、空文件、权限错误 - 转换层:纯数据操作,不碰文件系统。例如清洗excel列名、补全缺失日期、标准化手机号格式——输入DataFrame或字典,输出同结构数据
- 写入层:生成目标路径(建议用
target_dir / f"processed_{p.name}"),检查父目录是否存在(target_dir.mkdir(parents=True, exist_ok=True)),再保存 - 校验层:写完立刻验证行数是否一致、关键字段有无空值、MD5是否与预期模板匹配,失败则记录并跳过后续步骤
可控执行与进度反馈:别让脚本“黑着跑”
批量处理动辄几百个文件,没反馈=没掌控。用户需要知道卡在哪、还剩多久、哪些失败了。
立即学习“Python免费学习笔记(深入)”;
- 用
tqdm包装文件列表:for p in tqdm(file_list, desc="Processing"),终端实时显示进度条和预估时间 - 每完成10个文件,打印简要统计(成功/失败数、平均耗时),避免只等最后结果
- 失败时不中断整体流程,收集错误信息到
errors = [],最后统一输出jsON日志,含文件路径、错误类型、traceback片段 - 支持
--dry-run参数:只打印将要执行的操作,不真实读写,上线前必跑一次
流程可复用:配置驱动,而非硬编码
换一批文件,不该改代码,而应改配置。把变化点抽出来,系统就具备迁移能力。
- 用
pydantic定义配置模型,例如class ProcessConfig(BaseModel): input_dir: str; output_dir: str; encoding: str = "utf-8"; skip_if_exists: bool = True - 配置来源支持多级覆盖:默认值 ← YAML文件 ← 命令行参数(用
argparse或typer),命令行优先级最高 - 不同业务场景(如财务对账、日志归档)对应不同YAML,主程序只加载配置,不关心业务细节
- 加一个
--config example.yaml参数,就能切换整套行为,无需改一行逻辑代码
基本上就这些。不复杂,但容易忽略——边界不清导致误处理,逻辑紧耦合让维护变噩梦,没反馈让人怀疑脚本挂了,硬编码则让同一套代码无法服务多个项目。把这四点立住,你的批量处理系统就稳了一大半。