DeOldify批量处理脚本编写：自动化运维与监控

DeOldify批量处理脚本编写自动化运维与监控给一张黑白老照片上色那种时光倒流的感觉确实很奇妙。但如果你手头有成百上千张照片一张张手动处理那感觉可能就不太妙了。想象一下你有一个装满家族老照片的文件夹里面还有各种子文件夹图片格式五花八门有JPG、PNG甚至还有BMP。手动操作不仅耗时还容易出错比如漏掉几张或者处理失败了自己都不知道。这时候一个能自动帮你搞定一切的脚本就显得格外重要了。今天我们就来聊聊如何用Python写一个给DeOldify模型用的批量处理脚本。这个脚本不仅能自动找到所有图片还能保持原来的文件夹结构处理完了放回原处并且清清楚楚地告诉你哪些成功了哪些失败了。这样一来你就能放心地去喝杯咖啡让机器替你完成那些重复性的工作。1. 环境准备与脚本框架在开始写代码之前我们得先把“舞台”搭好。这里假设你已经安装好了Python并且对DeOldify的基本使用有初步了解。我们的目标是写一个工具脚本而不是从零开始研究模型本身。1.1 核心工具包我们的脚本主要依赖几个Python库它们各自负责不同的任务Pillow (PIL)这是Python里处理图片的“瑞士军刀”用来读取、保存和各种图片格式转换。os 和 pathlib这两个是Python自带的专门用来和文件系统打交道比如遍历文件夹、创建新路径。logging也是Python自带的用来记录脚本运行时的信息比如哪里出错了处理了多少张图比直接用print打印出来要专业和方便得多。你不需要额外安装DeOldify的库来运行这个脚本因为我们的脚本是“调用者”它负责组织任务、准备数据然后把图片交给已经部署好的DeOldify服务比如通过API去处理。当然如果你是在本地直接调用DeOldify的Python代码那也需要确保相应的环境已配置好。1.2 脚本设计思路在动手敲代码前我们先理清脚本要干什么怎么干入口与配置告诉脚本去哪里找图片处理完放哪里以及DeOldify服务在哪里。图片发现者自动钻进你指定的文件夹包括里面所有子文件夹把各种格式的图片都找出来。任务执行者把找到的每张图片按照原来的文件夹结构送到DeOldify那里去上色然后保存到新位置。监控记录员全程记录谁处理成功了谁失败了为什么失败最后生成一份详细的报告。下面我们就一步步把这些想法变成代码。2. 构建图片自动发现模块首先我们要解决“找图”的问题。这个模块的任务是给定一个根目录它能像探险家一样探索每一个角落把所有支持的图片文件都列出来并且记住它们原本的位置。import os from pathlib import Path from typing import List, Tuple class ImageDiscoverer: 图片发现器负责遍历目录找出所有支持的图片文件。 # 支持的图片格式可以根据需要扩展 SUPPORTED_EXTENSIONS {‘.jpg‘, ‘.jpeg‘, ‘.png‘, ‘.bmp‘, ‘.tiff‘, ‘.tif‘} def __init__(self, root_dir: str): 初始化发现器。 Args: root_dir: 需要遍历的根目录路径。 self.root_dir Path(root_dir).resolve() # 转换为绝对路径并解析 if not self.root_dir.is_dir(): raise ValueError(f“指定的路径不是目录: {self.root_dir}”) def discover(self) - List[Tuple[Path, Path]]: 遍历根目录及其所有子目录发现图片文件。 Returns: 一个列表每个元素是一个元组 (图片绝对路径, 相对于根目录的路径)。 discovered_images [] # 使用 pathlib 的 rglob 递归遍历所有文件 for file_path in self.root_dir.rglob(‘*‘): if file_path.is_file() and file_path.suffix.lower() in self.SUPPORTED_EXTENSIONS: # 计算相对路径用于保持目录结构 relative_path file_path.relative_to(self.root_dir) discovered_images.append((file_path, relative_path)) return discovered_images # 简单测试一下 if __name__ “__main__“: discoverer ImageDiscoverer(‘./old_photos‘) # 假设你的老照片在这个文件夹 images discoverer.discover() print(f“共发现 {len(images)} 张图片“) for img_abs, img_rel in images[:3]: # 打印前3个看看 print(f“绝对路径: {img_abs}, 相对路径: {img_rel}“)这段代码定义了一个ImageDiscoverer类。你只需要在创建它的时候告诉它老照片文件夹在哪里比如‘./old_photos‘然后调用discover()方法它就会返回一个列表。列表里每一项都包含图片的绝对路径和相对于根目录的路径。后面我们就会用这个相对路径来维持文件夹结构。3. 实现批量处理与结构保持找到图片后下一步就是处理它们。这里的关键是处理后的图片要放到一个新文件夹里但里面的子文件夹结构要和原来一模一样。这样你找起来才方便。3.1 核心处理函数我们先写一个函数它负责处理单张图片。这里为了演示我们用了一个mock_process_image函数来模拟调用DeOldify的过程。在实际应用中你需要把它替换成真正调用DeOldify API或本地模型的代码。from pathlib import Path import logging from PIL import Image import time # 设置日志方便查看运行情况 logging.basicConfig(levellogging.INFO, format‘%(asctime)s - %(levelname)s - %(message)s‘) logger logging.getLogger(__name__) def mock_process_image(input_path: Path) - Image.Image: 模拟图片处理过程例如调用DeOldify API。 在实际使用中这里应替换为真实的DeOldify调用代码。 Args: input_path: 输入图片的路径。 Returns: 处理后的PIL Image对象。 # 模拟处理耗时 time.sleep(0.1) # 这里简单演示打开图片并返回实际应进行上色处理 try: img Image.open(input_path) logger.info(f“模拟处理图片: {input_path.name}“) # 此处应添加调用DeOldify的代码 # processed_img deoldify_colorizer.get_transformed_image(img, ...) # return processed_img return img # 暂时返回原图 except Exception as e: logger.error(f“打开图片失败 {input_path}: {e}“) raise def process_single_image(input_path: Path, output_path: Path) - bool: 处理单张图片并保存。 Args: input_path: 输入图片路径。 output_path: 输出图片路径。 Returns: 成功返回True失败返回False。 try: # 1. 确保输出目录存在 output_path.parent.mkdir(parentsTrue, exist_okTrue) # 2. 调用处理函数模拟或真实 processed_image mock_process_image(input_path) # 3. 保存处理后的图片 # 可以保持原格式或统一保存为JPEG if output_path.suffix.lower() in [‘.jpg‘, ‘.jpeg‘]: processed_image.save(output_path, ‘JPEG‘, quality95) else: processed_image.save(output_path) logger.info(f“图片处理成功并保存: {output_path}“) return True except Exception as e: logger.error(f“处理图片失败 {input_path} - {output_path}: {e}“) return False3.2 批量处理与结构保持现在我们把发现器和处理器结合起来实现完整的批量流程。class BatchColorizer: 批量上色处理器协调图片发现、处理、保存全过程。 def __init__(self, input_root: str, output_root: str): 初始化处理器。 Args: input_root: 输入图片的根目录。 output_root: 输出图片的根目录。 self.input_root Path(input_root) self.output_root Path(output_root) self.discoverer ImageDiscoverer(input_root) # 用于记录结果的列表 self.successful [] self.failed [] def run(self): 执行批量处理任务。 logger.info(f“开始批量处理输入目录: {self.input_root}, 输出目录: {self.output_root}“) # 1. 发现所有图片 images self.discoverer.discover() total len(images) logger.info(f“共发现 {total} 张待处理图片“) if total 0: logger.warning(“未发现任何支持的图片文件任务结束。“) return # 2. 逐个处理图片 for idx, (input_abs_path, relative_path) in enumerate(images, 1): # 构造输出路径保持原有目录结构 output_abs_path self.output_root / relative_path logger.info(f“正在处理 [{idx}/{total}]: {relative_path}“) # 3. 处理并记录结果 if process_single_image(input_abs_path, output_abs_path): self.successful.append((input_abs_path, output_abs_path)) else: self.failed.append((input_abs_path, output_abs_path)) # 4. 打印简要报告 self._print_summary() def _print_summary(self): 打印处理结果摘要。 logger.info(“” * 50) logger.info(“批量处理完成“) logger.info(f“成功处理: {len(self.successful)} 张“) logger.info(f“处理失败: {len(self.failed)} 张“) if self.failed: logger.warning(“失败列表:“) for inp, _ in self.failed: logger.warning(f“ - {inp}“) logger.info(“” * 50) # 如何使用这个类 if __name__ “__main__“: # 指定你的老照片文件夹和处理结果存放文件夹 colorizer BatchColorizer( input_root‘./old_photos‘, output_root‘./colorized_photos‘ ) colorizer.run()运行这个脚本它就会自动把./old_photos文件夹包括所有子文件夹里的图片处理后原样保存到./colorized_photos文件夹下。结构完全一致非常清晰。4. 增强脚本日志、监控与报告一个健壮的自动化脚本不能处理完就“沉默”了。我们需要更详细的运行记录和一份漂亮的报告。4.1 配置详细日志我们之前用了基础的logging现在来增强它让日志同时输出到控制台和文件。import logging from pathlib import Path import sys def setup_logging(log_dir: str “./logs“): 配置日志系统同时输出到控制台和文件。 Args: log_dir: 日志文件存放目录。 log_path Path(log_dir) log_path.mkdir(parentsTrue, exist_okTrue) # 生成带时间戳的日志文件名 from datetime import datetime log_file log_path / f“deoldify_batch_{datetime.now().strftime(‘%Y%m%d_%H%M%S‘)}.log“ # 获取根日志记录器 logger logging.getLogger() logger.setLevel(logging.INFO) # 清除已有的处理器避免重复 if logger.hasHandlers(): logger.handlers.clear() # 创建控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_format logging.Formatter(‘%(asctime)s - %(name)s - %(levelname)s - %(message)s‘) console_handler.setFormatter(console_format) # 创建文件处理器 file_handler logging.FileHandler(log_file, encoding‘utf-8‘) file_format logging.Formatter(‘%(asctime)s - %(name)s - %(levelname)s - %(message)s‘) file_handler.setFormatter(file_format) # 将处理器添加到日志记录器 logger.addHandler(console_handler) logger.addHandler(file_handler) logging.info(f“日志系统已初始化日志文件: {log_file}“) return str(log_file) # 在脚本主函数开始处调用 # setup_logging()4.2 生成HTML报告除了日志生成一个可视化的HTML报告会更直观。我们可以用Python内置的库简单生成一个。import json from pathlib import Path from datetime import datetime class ProcessingReporter: 处理报告生成器。 def __init__(self, output_root: Path): self.output_root output_root self.report_data { “start_time“: None, “end_time“: None, “total_images“: 0, “successful“: [], “failed“: [], “summary“: {} } def start_report(self): 开始记录标记开始时间。 self.report_data[“start_time“] datetime.now().isoformat() def add_success(self, input_path: Path, output_path: Path): 记录一条成功记录。 rel_path input_path.relative_to(input_path.parents[len(input_path.parents)-3]) # 简化路径显示 self.report_data[“successful“].append({ “input“: str(rel_path), “output“: str(output_path.relative_to(self.output_root)) }) def add_failed(self, input_path: Path, error_msg: str): 记录一条失败记录。 rel_path input_path.relative_to(input_path.parents[len(input_path.parents)-3]) self.report_data[“failed“].append({ “input“: str(rel_path), “error“: error_msg }) def finalize_report(self, total: int): 结束记录生成摘要并写入文件。 self.report_data[“end_time“] datetime.now().isoformat() self.report_data[“total_images“] total successful_count len(self.report_data[“successful“]) failed_count len(self.report_data[“failed“]) self.report_data[“summary“] { “total_processed“: total, “successful“: successful_count, “failed“: failed_count, “success_rate“: f“{(successful_count/total*100):.1f}%“ if total 0 else “0%“ } # 保存为JSON文件 report_path self.output_root / “processing_report.json“ with open(report_path, ‘w‘, encoding‘utf-8‘) as f: json.dump(self.report_data, f, indent2, ensure_asciiFalse) logging.info(f“详细处理报告已生成: {report_path}“) # 生成一个简单的HTML报告更易读 self._generate_html_report(report_path.parent / “processing_report.html“) def _generate_html_report(self, html_path: Path): 生成一个简单的HTML格式报告。 html_content f“““ !DOCTYPE html html head titleDeOldify批量处理报告/title style body {{ font-family: sans-serif; margin: 40px; }} .summary {{ background-color: #f0f0f0; padding: 20px; border-radius: 5px; margin-bottom: 30px; }} .success {{ color: green; }} .failed {{ color: red; }} table {{ border-collapse: collapse; width: 100%; margin-top: 20px; }} th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }} th {{ background-color: #f2f2f2; }} /style /head body h1DeOldify批量上色处理报告/h1 div class“summary“ h2处理摘要/h2 pstrong开始时间:/strong {self.report_data[‘start_time‘]}/p pstrong结束时间:/strong {self.report_data[‘end_time‘]}/p pstrong总图片数:/strong {self.report_data[‘summary‘][‘total_processed‘]}/p p class“success“strong成功:/strong {self.report_data[‘summary‘][‘successful‘]} 张/p p class“failed“strong失败:/strong {self.report_data[‘summary‘][‘failed‘]} 张/p pstrong成功率:/strong {self.report_data[‘summary‘][‘success_rate‘]}/p /div “““ # 添加失败列表如果有 if self.report_data[“failed“]: html_content “““ h2失败详情/h2 table trth图片路径/thth错误信息/th/tr “““ for item in self.report_data[“failed“]: html_content f“ trtd{item[‘input‘]}/tdtd{item[‘error‘]}/td/tr\n“ html_content “ /table\n“ html_content “““ /body /html “““ with open(html_path, ‘w‘, encoding‘utf-8‘) as f: f.write(html_content) logging.info(f“HTML报告已生成: {html_path}“)然后我们需要修改一下BatchColorizer类的run方法集成这个报告器。class BatchColorizer: def __init__(self, input_root: str, output_root: str): # ... 初始化代码同上 ... self.reporter ProcessingReporter(Path(output_root)) def run(self): 执行批量处理任务。 self.reporter.start_report() # 开始记录 # ... 原有的发现图片代码 ... total len(images) self.report_data[“total_images“] total # 暂时存储总数最后会更新 for idx, (input_abs_path, relative_path) in enumerate(images, 1): output_abs_path self.output_root / relative_path # ... 处理逻辑 ... try: if process_single_image(input_abs_path, output_abs_path): self.successful.append((input_abs_path, output_abs_path)) self.reporter.add_success(input_abs_path, output_abs_path) else: # 如果process_single_image返回False说明已知错误已记录 self.failed.append((input_abs_path, output_abs_path)) self.reporter.add_failed(input_abs_path, “处理函数返回失败“) except Exception as e: # 捕获未预料的异常 logger.exception(f“处理过程发生未预料错误: {input_abs_path}“) self.failed.append((input_abs_path, output_abs_path)) self.reporter.add_failed(input_abs_path, str(e)) # 最终化报告 self.reporter.finalize_report(total) self._print_summary()5. 脚本优化与使用建议现在一个功能完整的批量处理脚本已经有了。但在实际使用中我们还可以让它更强大、更友好。5.1 添加命令行参数每次都去改脚本里的路径太麻烦。我们可以用argparse库让脚本通过命令行参数来接收配置。import argparse def main(): parser argparse.ArgumentParser(description‘DeOldify图片批量上色处理脚本‘) parser.add_argument(‘-i‘, ‘--input‘, requiredTrue, help‘输入图片的根目录路径‘) parser.add_argument(‘-o‘, ‘--output‘, requiredTrue, help‘输出图片的根目录路径‘) parser.add_argument(‘--log-dir‘, default‘./logs‘, help‘日志文件存放目录默认: ./logs‘) args parser.parse_args() # 设置日志 log_file setup_logging(args.log_dir) logging.info(f“命令行参数: 输入目录{args.input}, 输出目录{args.output}“) # 运行批量处理器 try: colorizer BatchColorizer(args.input, args.output) colorizer.run() logging.info(f“批量处理任务完成。日志文件: {log_file}“) except Exception as e: logging.error(f“脚本运行失败: {e}“, exc_infoTrue) raise if __name__ “__main__“: main()这样你就可以在终端里这样运行脚本了python batch_colorizer.py -i ./my_old_photos -o ./colorized_results --log-dir ./my_logs5.2 错误处理与重试机制网络调用或模型处理有时可能不稳定。我们可以为process_single_image函数添加简单的重试逻辑。def process_single_image_with_retry(input_path: Path, output_path: Path, max_retries: int 2) - bool: 带重试机制的图片处理函数。 Args: input_path: 输入图片路径。 output_path: 输出图片路径。 max_retries: 最大重试次数。 Returns: 成功返回True失败返回False。 for attempt in range(max_retries 1): # 尝试一次 重试次数 try: if attempt 0: logging.warning(f“第{attempt}次重试处理图片: {input_path.name}“) success process_single_image(input_path, output_path) if success: if attempt 0: logging.info(f“图片 {input_path.name} 在第{attempt1}次尝试时处理成功“) return True # 如果process_single_image返回False但不抛异常也视为失败 else: logging.warning(f“图片 {input_path.name} 处理失败将重试。“) except Exception as e: logging.warning(f“图片 {input_path.name} 处理时发生异常尝试{attempt1}/{max_retries1}: {e}“) # 如果不是最后一次尝试等待一下再重试 if attempt max_retries: time.sleep(1 * (attempt 1)) # 等待时间逐渐增加 logging.error(f“图片 {input_path.name} 经过{max_retries1}次尝试后仍失败。“) return False5.3 使用建议与注意事项在实际部署和运行这个脚本时有几点建议先做小规模测试在处理整个家族相册前先创建一个包含几十张图片的小文件夹进行测试确保脚本和你的DeOldify环境工作正常。关注资源消耗批量处理大量高分辨率图片会消耗大量内存和显存如果使用GPU。确保你的机器有足够资源或者考虑在脚本中添加延迟避免同时处理太多图片。替换处理核心脚本中的mock_process_image函数需要替换成真实的DeOldify调用代码。这可能涉及加载模型、调用API使用requests库或运行本地推理。处理结果验证脚本运行完毕后不要只看报告。建议随机抽查一些处理后的图片确保上色效果符合预期。6. 总结写到这里一个具备自动化运维和监控能力的DeOldify批量处理脚本就初具雏形了。从自动发现图片、保持目录结构到详细的日志记录和HTML报告生成这个脚本已经能帮你把重复性的劳动交给机器而你只需要关注最终的结果和那些需要人工复核的异常情况。脚本的核心价值在于“解放生产力”。它把我们从繁琐的文件操作和重复的点击中解脱出来让处理成百上千张图片成为一件只需一条命令就能发起的事情。更重要的是通过完善的日志和报告整个过程变得透明、可追溯即使中途出错也能快速定位问题。当然这只是一个起点。你可以根据自己的需求继续扩展它比如添加进度条、支持更多图片格式、集成到更复杂的流水线中或者为报告增加图片对比预览功能。希望这个脚本能成为你处理大量老照片上色任务的一个得力助手。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。