接手这个项目的第一天客户直接甩给我一串数字363636。我第一反应是订单号或者门店编号但聊下去才发现这是他们对一套批处理流程的约定叫法。拆开看其实是三层“36”36 个来源目录、每个目录下 36 个子项、每个子项需要执行 36 项规范化操作。三个 36 叠起来就是 46656 次独立操作。这种量级的工作纯手工几乎不可能完成哪怕每次操作只需要 5 秒也要连续干上 64 个小时而且中途只要手一抖后面全乱。所以项目接过来之后我只做了一件事把所有能自动化的环节全部写成脚本把人工干预降到最低。今天这篇文章就是这次“363636”批量处理项目的完整复盘从需求拆解到具体实现再到那些文档里不会写的坑一次性讲清楚。1. 项目背景与需求拆解1.1 “363636”到底是什么意思这类内部代号乍看像乱码但往往比正经项目名更能说明问题。拆解下来这个数字对应的是三层循环结构第一层“36”项目需要接收 36 个来源批次每个批次对应一个独立目录内容来自不同协作方。第二层“36”每个来源批次里固定包含 36 个子文件夹代表该批次下的细分条目。第三层“36”每条细分内容需要进行 36 项标准化动作包括文件重命名、编码转换、日期格式化、内容校验、归档复制等。所以 36 × 36 × 36 46656这才是项目真实的工作量。理解了这层关系后面的架构就清晰了外层循环管批次中层循环管子项内层循环管处理动作。三套循环互不干扰但每一层都需要独立的进度记录和错误隔离。这种命名方式在实际项目里很常见。它不描述“做什么”而是直接描述“工作量规模”。好处是团队一听 363636就知道复杂度在哪里坏处是如果没做好拆解这个数字就只是一个吓人的工作量而不是一个可执行的方案。1.2 批量任务为什么不能靠手动处理很多人面对 46656 次操作的第一想法是“招一批临时工慢慢做”但手工处理有三个致命问题。第一个问题是稳定性。人做重复操作前面 100 次可能很标准做到 500 次开始疲劳做到 1000 次就会开始出现漏网之鱼。文件名少个后缀、日期格式不统一、内容剪切位置偏移这些错误在单次操作里几乎不致命但放在四万多条数据里最后排查起来比重新做一遍还痛苦。第二个问题是成本。即使按每位操作员每小时能处理 360 次操作来算处理完全部也需要 130 人时。这不现实尤其在交付周期只有一两天的情况下。第三个问题是可追溯性。手工处理过的文件你很难说清楚“这一条是谁改的、改动之前是什么样、改了哪些字段”。但脚本不一样每条记录的每个动作都可以写进日志任何一条数据出了问题都能回溯到具体处理节点。所以这个项目从一开始就确定了原则体力活交给代码人只负责制定规则和检查规则。规则越清楚脚本越好写结果越可控。2. 工具选型与整体架构设计2.1 为什么最终选了 Python其实这类批量文件处理任务用 Shell 脚本也能做。文件重命名、格式转换、目录复制Shell 都有现成命令。但这次需求有几个特殊点促使我选了 Python。首先36 项处理动作里有相当一部分不是简单的“改名”或“移动”而是需要读取文件内容、判断编码、做字段级修改。比如时间字段要从2024/1/5修正成2024-01-05金额字段要去掉千分位符号再对齐小数位。这种内容层面的逻辑用 Shell 写会很别扭用 Python 则顺手很多。其次项目要求有完善的错误日志和断点续跑能力。Shell 脚本做循环和异常捕获不是不行但要处理复杂的嵌套逻辑和状态记录可读性会直线下降。Python 有现成的logging、json、pathlib等标准库开箱即用。最后是跨平台问题。虽然最终跑在处理服务器上但开发阶段我在自己电脑上调试处理服务器是 Windows 环境这两者之间的路径分隔符、编码规则都不同。Python 的pathlib能很好地屏蔽这些差异。当然Shell 也不是没有优势。如果只是做简单的批量重命名和归档一段for循环加mv命令可能 10 分钟就搞定了。但凡是处理逻辑超过三层、错误处理要求高、需要反复调整规则的场景我还是建议直接用 Python别在 Shell 上硬凑。2.2 目录结构与数据模型设计动手写代码之前我先把目录结构和目标格式定了下来这是整个项目最关键的准备工作。目录结构清晰了后面的循环和校验逻辑就简单很多。输入目录长这样data/input/ batch_01/ item_01/ 001.txt 002.txt ... item_02/ ... batch_02/ ... batch_36/输出目录没有沿用原来的批次结构而是在每个批次下按照“业务类型”进一步分类。因为下游使用方需要按照处理时间倒序找数据而不是按照原批次找。data/output/ batch_01/ normalized/ checked/ rejected/ batch_02/每个文件处理完之后的命名规则是YYYYMMDD_原批次号_序号_原文件名.ext。例如20240106_batch_01_023_contract.txt。这个格式看起来简单但好处非常多按时间排序就是处理顺序按批次号就能溯源序号保证文件名唯一不会互相覆盖。数据处理过程中我还额外维护了一个“状态表”记录每条记录处于哪个阶段。状态表本身是一个 JSON 文件跑批过程中实时更新。这样脚本如果中途挂掉下一次启动时能知道自己处理到哪里了不用从头再来。2.3 三层循环怎么拆才不混乱三层循环最怕写成一团浆糊外层没结束中层就乱套中层失败还会把整个批次拖垮。我的拆解原则是外层管进度中层管边界内层管动作。外层循环遍历 36 个批次每处理完一个批次就把“批量完成”的状态写入日志并把进度百分比打印出来方便观察整体跑批情况。中层循环遍历每个批次里的 36 个子项这一层有两个职责一是跳过状态表里已经标记为“成功”的项保证幂等二是隔离异常。某个子项处理失败时不能影响同一批次的其他子项所以异常要在这里捕获并记录而不是抛到外层。内层循环则是顺序执行 36 项处理动作。这一层不关心文件名不关心路径只关心“当前这一条记录的状态”。所有动作执行完后统一做一次完整性校验确认没有半成品文件残留。这套拆分逻辑的好处是任何一层出了问题修复起来都很快。内层动作出问题只影响单条记录的处理中层出问题最多导致某个子项需要重跑外层出问题大不了某个批次从零开始。层级之间互不污染排查成本就低。3. 核心脚本实现与执行过程3.1 第一阶段扫描与预检写主处理脚本之前我先做了一个预检脚本。这个脚本不处理任何文件只负责扫描输入目录核对三个关键指标实际批次目录数量是否为 36每个批次下的子目录数量是否为 36每个子目录下的待处理文件数量是否与清单一致这一步非常重要因为大部分协作方发来的数据都不会完全合规。有的批次少了几个子项有的子项是空目录有的文件命名不规范。如果在主处理流程中才发现这些问题还得暂停脚本去核实预检阶段就全暴露出来就能提前跟对方沟通补齐。import json from pathlib import Path BASE Path(data/input) EXPECTED_BATCH 36 EXPECTED_ITEMS_PER_BATCH 36 report {} for batch_dir in sorted([p for p in BASE.iterdir() if p.is_dir()]): items [p for p in batch_dir.iterdir() if p.is_dir()] report[batch_dir.name] { item_count: len(items), missing: [fitem_{i:02d} for i in range(1, EXPECTED_ITEMS_PER_BATCH 1) if not (batch_dir / fitem_{i:02d}).exists()], } with open(precheck_report.json, w, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2)这个脚本跑完直接就生成了预检报告。第一轮扫描发现36 个批次里有 3 个批次存在子项缺失一共缺了 7 个子目录。原因也比较常见协作方打包时漏传了一部分。预检脚本还做了一件重要的事把所有文件的编码类型检测了一遍。有些历史文件是 GBK 编码有些是 UTF-8还有些是混合编码。这个信息在后续处理中会用到提前摸底能避免处理到一半才发现乱码。3.2 第二阶段批量处理主循环主循环脚本是整个项目的核心代码量并不大200 行左右但每一行都有讲究。整体逻辑就是前面说的三层循环结构加上状态管理、异常隔离和日志记录。这里的关键设计是“幂等性”。意思是同一个脚本跑一遍和跑两遍最终结果必须一样。实现方式很简单每处理完一个子项就把该子项标记为“done”写入状态文件下次启动时先读状态文件跳过已经完成的子项。import json import shutil from pathlib import Path from datetime import datetime INPUT Path(data/input) OUTPUT Path(data/output) STATE_FILE Path(data/state.json) BACKUP Path(data/backup) def load_state(): if STATE_FILE.exists(): return json.loads(STATE_FILE.read_text(encodingutf-8)) return {finished_batches: [], done_items: [], failed_items: []} def save_state(state): STATE_FILE.write_text(json.dumps(state, ensure_asciiFalse, indent2), encodingutf-8) def normalize_file(src_file: Path, target: Path): raw src_file.read_bytes() # 省略具体的 36 项动作实际包括编码检测、格式修正等 target.write_bytes(raw) return True def process_batch(batch_dir: Path, state): batch_name batch_dir.name if batch_name in state[finished_batches]: return for item_dir in sorted(batch_dir.glob(item_*)): if str(item_dir) in state[done_items]: continue try: out_dir OUTPUT / batch_name / normalized out_dir.mkdir(parentsTrue, exist_okTrue) for src_file in sorted(item_dir.iterdir()): target out_dir / f{datetime.now():%Y%m%d}_{batch_name}_{src_file.stem}{src_file.suffix} normalize_file(src_file, target) state[done_items].append(str(item_dir)) except Exception as exc: state[failed_items].append({item: str(item_dir), error: str(exc)}) save_state(state) state[finished_batches].append(batch_name) save_state(state)代码里最值得注意的一点是每次处理完一个子项就调用save_state。这种做法看似多做了很多次磁盘写入但换来的是一旦脚本崩溃最多只会丢失一个子项的进度而不是一整批。跑批任务最怕的就是“跑了两小时突然断电一切从头开始”。实际跑批时我在循环里加了一个进度条每处理完一个批次就打印一次当前进度。日志大致长这样[09:12:03] batch_01 36/36 完成累计 36 项 [09:12:58] batch_02 36/36 完成累计 72 项 ... [10:41:27] batch_19 第 17 项处理失败文件名编码异常已跳过中间出现的失败项不会中断整体跑批但会被记录进状态文件。跑批结束后我统一查看失败项针对性地修复即可。3.3 第三阶段校验与结果导出主流程跑完不代表项目就结束了。还需要一个独立的校验脚本确保输出目录中的文件数量、命名格式、文件内容都符合预期。校验脚本做三件事第一核对数量。输出目录里“normalized”子目录下的文件总数应当等于预期值也就是 46656 减去预检阶段已经确认缺失的数量。第二抽检内容。按批次随机抽取 5% 的文件对比原始文件和处理后文件的差异确认规范化操作没有破坏内容。如果有条件尽量做全量校验而不是抽检因为批量操作一旦某个逻辑写错往往是系统性的抽检可以发现但全量校验更稳妥。第三生成校验报告。报告里包含每个批次的处理情况比如文件总数、成功数、失败数、平均耗时以及所有失败项的详细原因。这个报告既是交付给下游的数据支撑也是后续优化脚本的依据。校验脚本还顺带做了一个文件完整性校验。复制过程中如果文件大小不一致或者内容被截断就要从备份目录恢复。备份目录留给那些需要修改文件内容而不是直接复制的场景处理前先把原始文件复制一份到 backup处理后如果校验失败可以快速还原。3.4 执行效率与优化记录第一版脚本按照最简单的单进程、顺序执行方式跑全部跑完耗时大约 27 分钟。对于非实时的批处理任务来说27 分钟其实可以接受但我还是做了一轮优化把耗时压缩到了 6 分钟以内。优化思路很简单36 个批次之间是相互独立的完全可以并行处理。我用concurrent.futures.ProcessPoolExecutor把批次分配到多个进程里跑每个进程处理一个批次。文件 IO 密集型任务用多进程比多线程更合适因为可以绕过 Python 全局解释器锁的限制。并行度不是越高越好。在 Windows 服务器上跑文件系统并发写入太多反而会变成磁盘瓶颈。我试过 8 进程、16 进程、32 进程最终 16 进程效果最好再往上反而因为磁盘争用导致耗时增加。看一下优化前后的数据对比方案耗时说明单进程顺序执行27 分钟最稳但耗时偏长8 进程并行11 分钟磁盘负载中等16 进程并行6 分钟效果最好磁盘接近瓶颈32 进程并行7 分钟磁盘争用性能反而下降这组数据也说明了盲目增加并发数并不总是好事。如果你的任务也涉及大量文件读写建议先从小并发开始测试观察磁盘占用率找到自己的平衡点。4. 常见的坑与经验记录4.1 Windows 路径长度限制引发的批量失败项目跑批到一半突然出现一批文件处理失败日志里报的错误非常诡异提示找不到文件路径。排查下来发现根源是 Windows 的路径长度限制。原始目录结构本身并不深但处理后的文件命名规则加了一段时间前缀路径就变长了。当某个文件处理后的完整路径超过 260 个字符Windows 就会拒绝访问。最头疼的是这个问题是随机出现的文件名短的系统碰不到文件名长一点就崩溃。解决方案有两个层面。第一在 Windows 系统里开启长路径支持通过注册表把LongPathsEnabled设置为 1第二从根源上缩短输出路径。我两个都做了最终把输出目录从data/output/batch_01/normalized/缩短成out/b01/nrm/风格页面路径从源头降下来。如果你也遇到类似情况建议先检查一下是否符合 Windows 260 字符限制的报错特征。这个问题在 Linux 和 macOS 上基本不存在但国内很多企业服务器都是 Windows批量复制文件时很容易踩到这个坑。4.2 编码混战GBK 和 UTF-8 的相爱相杀这个项目里的文件来源五花八门有些来自老系统导出的 GBK 编码有些来自新系统默认的 UTF-8 编码还有一批文件内部同时包含两种编码。直接按 UTF-8 读取遇到 GBK 文件就会报编码错误按 GBK 读取又会把合法的 UTF-8 文件读成乱码。处理编码问题我总结出一个比较实用的流程先读取文件前几千字节用chardet或charset-normalizer做编码检测。根据检测结果选择合适的解码方式读取文件内容。将内容统一转为 UTF-8 后写回。要注意编码检测本身不是 100% 准确的尤其在短文件上。所以我加了一个兜底机制如果按检测出的编码解码时抛异常就回退到 GBK 和 UTF-8 依次尝试。最大的教训是不要只处理文件名的编码。文件名乱码和文件内容乱码是两回事我一开始只修了文件名结果打开内容还是乱码等于白干。两个地方都得处理。4.3 非幂等操作差点把数据弄丢这里要说一个我自己踩过的坑。跑批过程中我为了节省磁盘空间加了一步“处理完成后删除源文件”。但脚本写的逻辑不太严谨第二次跑批时同一批文件被再次扫描到结果把已经处理过的文件当成待处理文件又处理了一遍然后删除了源文件。还好当时备份目录还在最后从备份恢复了一部分数据。这次事故之后我再也不在正式跑批时删除源文件最多把它移动到archive目录保留三天确认无误再清理。更关键的是我把脚本改成了严格的幂等模式处理前检查目标文件是否存在如果存在并且内容一致就直接跳过。处理时不直接覆盖目标文件而是先写入临时文件成功后再用原子操作改名。这套机制看起来多写了几行代码但能避免绝大多数重复跑批导致的数据事故。4.4 中途崩溃与断点续跑跑批任务最怕的突发状况就是脚本跑到一半崩了。我遇到过服务器意外重启也遇到过磁盘写满。第一版脚本没有做状态持久化只能从头再跑时间损失很大。后来我加入了状态文件机制就是前面代码里已经展示的load_state()和save_state()。状态文件里记录了哪些批次已经完成、哪些子项已经处理完这样即使崩溃重启后只需要处理尚未完成的那部分。除了状态文件我还建议给日志加上“轮转”功能。长时间跑批任务会产生大量日志如果不做轮转日志文件会无限膨胀既占磁盘空间也影响排查效率。用 Python 的logging.handlers.RotatingFileHandler按大小或按天自动切割保留最近三天的日志就行。4.5 问题排查速查表最后整理一份速查表都是这次项目里实际遇到的问题症状可能原因解决方案某批次全部失败日志提示找不到路径Windows 路径过长 / 目录拼接错误启用长路径支持缩短输出目录层级处理后的文件打开是乱码文件编码检测错误或只处理了文件名编码用 charset-normalizer 检测统一转 UTF-8第二次跑批覆盖了正常文件脚本非幂等直接覆盖目标文件目标文件存在时跳过先写临时文件再改名脚本运行一段时间后内存暴涨日志未轮转或大批量文件列表载入内存日志轮转用迭代器逐条读取文件并发跑批时文件数对不上多进程同时写同一目录出现相互覆盖不同批次写不同输出目录进程间隔离源文件被误删脚本中“处理完成后删除源文件”逻辑不严谨改为移动到 archive 目录延迟清理最后再分享一个小技巧如果你也要处理类似的批量任务强烈建议给脚本加一个--dry-run参数。这个模式不执行任何实际写操作只扫描文件、模拟处理流程、输出将要执行的动作清单。跑一次--dry-run你能在几分钟内发现绝大多数问题文件数对不对、命名规则是否符合预期、目标路径是否需要调整。等所有检查确认通过再真正执行跑批。我在 363636 项目里就是先跑了两轮 dry-run第一轮发现命名规则里有重复序号第二轮确认修正后的规则无误最终正式跑批才一次通关。这个习惯后来被我带进了所有批量处理项目成本极低但省下来的排查时间非常可观。