DeOldify处理复杂数据结构：批量处理图片目录的工程化实践-程序员充电站

DeOldify处理复杂数据结构：批量处理图片目录的工程化实践

你是不是也遇到过这样的场景？手头有一个文件夹，里面塞满了各种老照片，文件夹套着子文件夹，结构乱七八糟，有的按年份分，有的按事件分。一张张手动上传到DeOldify去处理？光是想想就觉得头大。

我之前接手过一个项目，需要为一个文化机构修复上千张历史照片。这些照片的存放方式堪称“考古现场”——几十个主文件夹，每个下面又有好几层子目录，文件名也是五花八门。如果靠人工，不仅效率低下，还容易出错漏掉文件。这时候，一个能自动、批量、且能保持原有目录结构的处理脚本，就成了救命稻草。

今天，我就来分享一下如何用Python写一个“工程化”的批量处理脚本。它不仅能帮你自动遍历那些复杂的文件夹，调用DeOldify API给老照片上色，还能原封不动地把修复好的照片放回对应的位置，整个过程高效又省心。

1. 为什么需要工程化的批量处理？

在开始写代码之前，我们先聊聊为什么简单的“for循环”不够用，非得搞什么“工程化”处理。

想象一下，你有一个装满老照片的硬盘。里面的文件夹可能是这样的：

老照片库/ ├── 家庭合影/ │ ├── 1990年春节/ │ │ ├── 客厅合影.jpg │ │ └── 年夜饭.jpg │ └── 1995年暑假/ │ ├── 公园1.jpg │ └── 公园2.jpg └── 个人照片/ ├── 小学/ │ ├── 毕业照.jpg │ └── 运动会.jpg └── 中学/ └── 班级活动.jpg

面对这种结构，手动处理几乎是不可能的。一个合格的批量处理脚本需要解决几个核心问题：

自动遍历：脚本得自己会“钻”进每一个子文件夹，找到所有图片，不管它们藏得多深。
保持结构：修复后的照片不能全都堆在一个文件夹里，必须按照原来的样子放回去，这样你才知道哪张照片对应哪个事件。
高效稳定：成百上千张照片，一张张等太慢了。同时，网络可能会波动，API也可能偶尔出错，脚本不能因为一张照片失败就整个停下来。

这就是我们所说的“工程化”实践——不只是让代码跑起来，还要让它跑得快、跑得稳、结果清晰可管理。

2. 搭建你的批量处理工具箱

工欲善其事，必先利其器。我们先来准备一下脚本运行所需的环境和核心思路。

2.1 环境与依赖

首先，确保你的电脑上安装了Python（建议3.7或以上版本）。我们主要会用到几个标准库和第三方库，你可以通过pip来安装：

pip install requests pillow

requests：用来和DeOldify的API“对话”，发送图片并接收结果。
PIL(通过pillow安装)：一个强大的图像处理库，这里我们主要用它来确认找到的文件确实是图片。

假设你已经有一个可用的DeOldify API服务地址，比如http://your-deoldify-server:port/colorize。我们的脚本就是围绕这个端点展开工作的。

2.2 核心思路拆解

整个脚本的流程，可以想象成一个高效的流水线：

探索者（遍历模块）：从你指定的根目录出发，像探险一样记录下所有图片文件的路径，并且记住它们原本在文件夹结构中的位置。
调度员（任务队列）：把找到的所有图片路径，变成一个个待处理的“任务”，放进一个任务列表里。
工人小组（并发处理器）：创建多个“工人”（线程或协程），同时从任务列表里领取任务，调用API处理图片。
质检员与仓库管理员（结果处理）：工人处理好图片后，质检员（错误重试逻辑）要确保任务成功。然后，仓库管理员按照照片原本的目录结构，把修复好的新图片保存到对应的输出目录里。

接下来，我们就一步步把这个流水线搭建起来。

3. 第一步：智能的目录探索者

我们的首要任务是准确无误地找到所有图片。这里的关键是递归遍历和路径记录。

import os from pathlib import Path from PIL import Image def find_all_image_files(root_dir, output_base_dir): """ 递归探索目录，找出所有支持的图片文件，并生成对应的输出路径。 参数: root_dir: 需要处理的原始图片根目录。 output_base_dir: 修复后图片的输出根目录。 返回: 一个列表，里面每个元素都是一个元组 (原始图片路径, 目标输出路径)。 """ image_extensions = ('.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif', '.webp') task_list = [] # 使用 pathlib 的 rglob 进行递归遍历，更简洁 root_path = Path(root_dir) for img_path in root_path.rglob('*'): if img_path.suffix.lower() in image_extensions: try: # 快速用PIL打开一下，验证是否是有效图片文件，避免处理损坏文件 with Image.open(img_path) as img: img.verify() # 验证文件完整性 # 验证通过后，再构建输出路径 # 计算相对于根目录的相对路径 relative_path = img_path.relative_to(root_path) # 在输出根目录下，构建相同的子目录结构 output_path = Path(output_base_dir) / relative_path # 确保输出文件的目录存在 output_path.parent.mkdir(parents=True, exist_ok=True) # 可以给输出文件加个前缀或修改后缀，这里我们保持原名，但可以改为 .colorized.jpg output_path = output_path.with_stem(img_path.stem + '_colorized') task_list.append((str(img_path), str(output_path))) print(f"发现图片: {img_path} -> 将输出到: {output_path}") except (IOError, SyntaxError, Image.UnidentifiedImageError) as e: # 如果文件不是有效图片，跳过并记录日志 print(f"警告: 文件 {img_path} 可能不是有效图片或已损坏，已跳过。错误: {e}") continue print(f"探索完成。共发现 {len(task_list)} 张待处理图片。") return task_list

这个函数做了几件聪明事：

root_path.rglob('*')一行代码就递归遍历了所有子目录。
用PIL.Image.verify()提前检查文件，避免把损坏的图片发给API，白白浪费时间和资源。
最关键的一步是relative_to()和output_path.parent.mkdir()，它保证了输出路径的目录结构和输入一模一样。parents=True参数意味着它会自动创建所有不存在的中间目录。

4. 第二步：打造稳健的API处理工人

找到了所有图片，接下来就是处理它们。直接一个接一个处理太慢，我们需要并发。同时，网络世界充满不确定性，必须要有重试机制。

4.1 单个图片的处理与重试

我们先写一个负责单张图片处理的函数，它内置了简单的重试逻辑。

import requests import time def colorize_single_image(api_url, input_image_path, output_image_path, max_retries=3): """ 调用DeOldify API处理单张图片，并保存结果。 参数: api_url: DeOldify API的完整地址。 input_image_path: 待处理图片的本地路径。 output_image_path: 处理后图片的保存路径。 max_retries: 失败最大重试次数。 返回: (是否成功, 错误信息或None) """ for attempt in range(max_retries): try: with open(input_image_path, 'rb') as img_file: files = {'image': img_file} # 发送POST请求到API response = requests.post(api_url, files=files, timeout=30) # 设置超时 if response.status_code == 200: # 保存返回的图片数据 with open(output_image_path, 'wb') as f: f.write(response.content) print(f"成功: {input_image_path} -> {output_image_path}") return True, None else: error_msg = f"API返回错误状态码: {response.status_code}, 内容: {response.text[:200]}" print(f"尝试 {attempt+1}/{max_retries} 失败: {error_msg}") except requests.exceptions.Timeout: error_msg = f"请求超时" print(f"尝试 {attempt+1}/{max_retries} 失败: {error_msg}") except requests.exceptions.ConnectionError: error_msg = f"连接错误" print(f"尝试 {attempt+1}/{max_retries} 失败: {error_msg}") except Exception as e: error_msg = f"未知错误: {e}" print(f"尝试 {attempt+1}/{max_retries} 失败: {error_msg}") # 如果不是最后一次尝试，等待一下再重试 if attempt < max_retries - 1: wait_time = 2 ** attempt # 指数退避，等待1, 2, 4秒... print(f"等待 {wait_time} 秒后重试...") time.sleep(wait_time) # 所有重试都失败了 return False, error_msg

这个函数是一个“坚韧”的工人。如果一次请求失败了（可能是网络波动、API暂时不可用），它会休息一下（指数退避策略），然后换个时间再试几次，而不是轻易放弃。

4.2 组织工人团队：并发处理

有了能干的单个工人，我们现在要组织一个工人团队，让他们同时干活。这里我们用Python的concurrent.futures模块中的ThreadPoolExecutor，它非常适合这种I/O密集型任务（主要时间花在等待网络响应上）。

from concurrent.futures import ThreadPoolExecutor, as_completed def batch_colorize_images(api_url, task_list, max_workers=4): """ 并发批量处理图片任务。 参数: api_url: DeOldify API地址。 task_list: 由 find_all_image_files 生成的（输入路径，输出路径）列表。 max_workers: 最大并发线程数。根据你的API服务器性能和网络调整，通常4-8个即可。 """ success_count = 0 fail_count = 0 failed_tasks = [] print(f"开始批量处理 {len(task_list)} 张图片，使用 {max_workers} 个并发线程...") with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务到线程池，得到一个Future对象的映射 future_to_task = { executor.submit(colorize_single_image, api_url, inp, out): (inp, out) for inp, out in task_list } # 异步获取完成的任务结果 for future in as_completed(future_to_task): input_path, output_path = future_to_task[future] try: success, error_msg = future.result() # 获取任务结果 if success: success_count += 1 else: fail_count += 1 failed_tasks.append((input_path, error_msg)) print(f"处理失败: {input_path}, 错误: {error_msg}") except Exception as exc: fail_count += 1 failed_tasks.append((input_path, str(exc))) print(f"任务生成异常: {input_path}, 异常: {exc}") # 打印最终报告 print("\n" + "="*50) print("批量处理完成！") print(f"成功: {success_count} 张") print(f"失败: {fail_count} 张") if failed_tasks: print("\n失败任务列表:") for task, err in failed_tasks: print(f" - {task}: {err}") # 可以选择将失败列表保存到文件，方便后续手动重试 with open('failed_tasks.log', 'w') as f: for task, err in failed_tasks: f.write(f"{task}\t{err}\n") print("失败任务已保存至 'failed_tasks.log'")

这个函数是团队的“调度员”。它创建了一个包含多个工人的线程池，然后把所有任务分配下去。as_completed会按照任务完成的先后顺序返回结果，哪个工人先干完活，我们就先接收哪个的结果，这样效率最高。最后，它还会给你一份详细的处理报告，告诉你哪些成功了，哪些失败了，失败的原因是什么。

5. 第三步：将所有模块组装起来

现在，我们把探索者、调度员和工人小组组合成一个完整的脚本。

import argparse def main(): # 使用 argparse 让脚本可以通过命令行参数调用，更灵活 parser = argparse.ArgumentParser(description='批量处理目录中的老照片，使用DeOldify API上色。') parser.add_argument('input_dir', help='包含老照片的根目录路径') parser.add_argument('output_dir', help='修复后图片的输出根目录路径') parser.add_argument('--api', default='http://localhost:5000/colorize', help='DeOldify API 地址 (默认: http://localhost:5000/colorize)') parser.add_argument('--workers', type=int, default=4, help='并发处理线程数 (默认: 4)') args = parser.parse_args() input_dir = args.input_dir output_dir = args.output_dir api_url = args.api max_workers = args.workers # 步骤1: 探索目录，构建任务列表 print("正在扫描目录结构并寻找图片...") task_list = find_all_image_files(input_dir, output_dir) if not task_list: print("在指定目录中未找到任何支持的图片文件。请检查路径和文件格式。") return # 步骤2: 并发处理所有任务 batch_colorize_images(api_url, task_list, max_workers) print("所有任务执行完毕。") # 引入之前定义的函数 find_all_image_files, colorize_single_image, batch_colorize_images # ... (将前面定义的函数代码放在这里) if __name__ == '__main__': main()

这个主函数就像整个流水线的总控台。你可以通过命令行来运行它，非常方便：

python batch_deoldify.py /path/to/your/old_photos /path/to/output --api http://your-server:port/colorize --workers 6

6. 实践中的经验与优化建议

脚本跑起来之后，在实际项目中你可能还会遇到一些情况，这里分享几点经验：

控制并发数：--workers参数不是越大越好。设置太高可能会把你的API服务器或网络带宽压垮，导致大家都变慢甚至失败。从4开始，根据实际情况调整。
处理特殊文件：上面的脚本已经过滤了非图片文件，但你可能还会遇到文件名带特殊字符、路径过长等问题。pathlib库通常能很好地处理路径，但保持文件名简洁总没坏处。
日志与监控：对于超大批量任务（比如上万张），建议把打印信息 (print) 替换为更正式的日志模块 (logging)，可以输出到文件，方便事后查看。你甚至可以加入一个进度条库（如tqdm），实时查看处理进度。
结果校验：极端情况下，API可能返回了200状态码，但图片数据是空的或损坏的。更稳健的做法是在保存图片后，用PIL再打开验证一次文件完整性。
资源清理：如果处理过程中途被中断，可能会留下一些不完整的输出文件。一个健壮的脚本可以考虑在任务开始前清理旧的输出目录，或者设计更细粒度的任务状态记录，支持“断点续传”。