DeOldify图像上色服务命令行工具开发：快速批处理脚本编写

DeOldify图像上色服务命令行工具开发快速批处理脚本编写你是不是也遇到过这样的烦恼手头有一堆老照片、黑白电影截图或者历史资料图想用DeOldify给它们上色结果一张张上传、等待、下载折腾半天才处理完几张。要是能有个工具把整个文件夹的图片一股脑儿扔给它让它自己默默处理完那该多省事。今天咱们就来聊聊怎么自己动手写一个命令行工具让DeOldify的API服务为你打工实现图片的自动化批量上色。整个过程其实没你想的那么复杂跟着步骤走一两个小时就能搞定。1. 准备工作理清思路与工具选择在动手写代码之前咱们先得把要做的事情想清楚。这个工具的核心目标很简单自动处理一个文件夹里的所有图片调用DeOldify服务上色然后把结果保存好。为了实现这个目标我们需要考虑几个关键点怎么找到所有图片需要遍历指定文件夹识别出常见的图片格式比如.jpg, .png, .bmp等。怎么调用服务DeOldify通常提供HTTP API我们需要用代码去“访问”这个网址把图片数据传过去。一次处理一张太慢怎么办可以同时发起多个请求这就是“并发”处理能大幅提升速度。万一网络抽风失败了怎么办得有个重试机制不能因为一次失败就放弃。处理好的图片放哪里最好新建一个文件夹把上色后的图片整齐地放进去别和原图混在一起。接下来是工具选择。主要有两个方向Shell脚本 (Bash)适合在Linux或Mac环境下流程简单直接依赖少。它擅长调用系统命令和组合各种工具比如curl发请求find找文件但对于复杂的逻辑和错误处理写起来会有点啰嗦。Python脚本这是更通用、更强大的选择。Python有丰富的库支持像requests发HTTP请求、concurrent.futures处理并发、os和pathlib操作文件都非常方便代码结构也更清晰易读。考虑到咱们要做的功能并发、重试稍微有点复杂为了让教程更清晰、代码更好维护我选择用Python来演示。即使你不太熟悉Python下面的代码我也会尽量写得明白你照着做就行。2. 核心步骤拆解从单张到批量咱们先别想得太复杂从最简单的开始怎么用代码给一张图片上色。搞定了这个批量处理就是给它套个“循环”的壳子。2.1 第一步发送单张图片到DeOldify API假设你的DeOldify服务已经部署好了并且有一个可以访问的API地址比如http://your-server-address:port/colorize。这个接口通常接收一张图片然后返回上色后的图片。我们用Python的requests库来做这件事。如果你还没安装这个库在命令行里输入pip install requests就能装上。import requests def colorize_single_image(image_path, api_url): 将单张图片发送到DeOldify API进行上色。 Args: image_path (str): 本地图片文件的路径。 api_url (str): DeOldify服务的API地址。 Returns: bytes: 上色后的图片二进制数据如果失败则返回None。 try: # 以二进制读取模式打开图片文件 with open(image_path, rb) as image_file: # 构建要发送的数据通常API要求文件字段名为image或file files {image: (image_path, image_file, image/jpeg)} # 可根据实际图片类型调整mime类型 # 发送POST请求 response requests.post(api_url, filesfiles) # 检查请求是否成功HTTP状态码为200 if response.status_code 200: print(f成功处理图片: {image_path}) return response.content # 返回图片的二进制数据 else: print(f处理图片失败 {image_path}: 服务器返回状态码 {response.status_code}) return None except FileNotFoundError: print(f错误找不到文件 {image_path}) return None except requests.exceptions.RequestException as e: print(f网络请求错误 ({image_path}): {e}) return None # 使用示例 api_endpoint http://localhost:7860/colorize # 替换成你的实际API地址 result_image_data colorize_single_image(old_photo.jpg, api_endpoint) if result_image_data: # 将返回的二进制数据保存为新文件 with open(old_photo_colorized.jpg, wb) as f: f.write(result_image_data) print(图片已保存。)这段代码定义了一个函数colorize_single_image它做了几件事打开本地的图片文件。通过HTTP POST请求把文件数据发送到指定的API地址。如果服务器成功处理并返回状态码200我们就拿到上色后的图片数据。最后把返回的二进制数据保存成一个新的图片文件。注意你需要把api_endpoint变量的值换成你自己部署的DeOldify服务的真实地址。具体的API接口路径和参数比如字段名是image还是file需要查看你的DeOldify服务的文档。2.2 第二步遍历文件夹获取所有图片单张图片搞定了现在来处理一个文件夹。我们需要写个函数找出文件夹里所有支持的图片文件。import os from pathlib import Path def get_image_files(folder_path, extensions(.jpg, .jpeg, .png, .bmp, .tiff, .webp)): 获取指定文件夹下所有指定后缀的图片文件路径。 Args: folder_path (str): 目标文件夹路径。 extensions (tuple): 需要查找的图片文件后缀元组。 Returns: list: 图片文件完整路径的列表。 image_files [] folder Path(folder_path) if not folder.exists() or not folder.is_dir(): print(f错误路径 {folder_path} 不存在或不是一个文件夹。) return image_files # 使用rglob递归遍历所有子文件夹如果不需要递归可以用glob for ext in extensions: # 匹配所有大小写后缀 image_files.extend(folder.rglob(f*{ext})) image_files.extend(folder.rglob(f*{ext.upper()})) # 将Path对象转换为字符串路径并去重因为大小写匹配可能重复 image_files list(set([str(f) for f in image_files])) print(f在文件夹 {folder_path} 中找到 {len(image_files)} 张图片。) return image_files # 使用示例 input_folder ./old_photos all_images get_image_files(input_folder) for img in all_images[:5]: # 打印前5个看看 print(img)这个函数会递归地搜索你指定文件夹包括子文件夹里所有常见格式的图片并把它们的完整路径列出来。extensions参数可以让你自定义需要处理的图片类型。2.3 第三步加入并发处理提升速度一张接一张地处理如果图片很多会等得花儿都谢了。我们可以用Python的ThreadPoolExecutor来同时发送多个请求这就是并发。import concurrent.futures from typing import List, Tuple def process_images_concurrently(image_paths: List[str], api_url: str, max_workers: int 4) - List[Tuple[str, bytes]]: 并发处理多张图片。 Args: image_paths (List[str]): 图片路径列表。 api_url (str): API地址。 max_workers (int): 最大并发线程数。不宜设置过高避免对服务器造成过大压力。 Returns: List[Tuple[str, bytes]]: 成功处理的图片路径 图片数据列表。 results [] # 使用线程池 with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务建立未来对象到文件路径的映射 future_to_path {executor.submit(colorize_single_image, path, api_url): path for path in image_paths} # 异步获取完成的任务结果 for future in concurrent.futures.as_completed(future_to_path): image_path future_to_path[future] try: image_data future.result() # 获取函数返回值 if image_data: results.append((image_path, image_data)) print(f已完成: {image_path}) else: print(f处理失败未返回数据: {image_path}) except Exception as e: print(f处理图片时发生异常 {image_path}: {e}) print(f并发处理完成。成功处理 {len(results)} / {len(image_paths)} 张图片。) return results这里创建了一个线程池max_workers控制同时工作的线程数量也就是同时处理几张图片。设为4或5是个比较稳妥的选择既能提速又不会把服务器或你的网络冲垮。每个线程都会独立调用我们之前写的colorize_single_image函数。2.4 第四步增加错误重试机制网络请求不稳定是常有的事。为了更健壮我们给单张图片处理函数加上重试逻辑。这里用一个简单的循环来实现。import time def colorize_single_image_with_retry(image_path, api_url, max_retries3, delay2): 带重试机制的单张图片上色函数。 Args: image_path (str): 图片路径。 api_url (str): API地址。 max_retries (int): 最大重试次数。 delay (int): 每次重试前的等待时间秒。 Returns: bytes or None: 成功则返回图片数据失败返回None。 for attempt in range(max_retries): try: result colorize_single_image(image_path, api_url) if result: return result # 成功则直接返回 else: # colorize_single_image内部已打印错误这里只打印重试信息 if attempt max_retries - 1: print(f第{attempt 1}次尝试失败{delay}秒后重试...) time.sleep(delay) except requests.exceptions.RequestException as e: print(f请求异常尝试{attempt 1}: {e}) if attempt max_retries - 1: time.sleep(delay) print(f图片 {image_path} 经过{max_retries}次尝试后仍失败。) return None然后我们需要把并发处理函数process_images_concurrently中提交的任务从调用colorize_single_image改为调用这个带重试的colorize_single_image_with_retry。2.5 第五步保存结果与组织文件所有图片处理完后我们会得到一个列表里面是原图路径 上色后图片数据的配对。我们需要把这些数据保存成文件并且最好放到一个新的、有组织的文件夹里。def save_colorized_images(results: List[Tuple[str, bytes]], output_base_dir: str ./colorized_output): 将上色后的图片数据保存到文件。 Args: results (List[Tuple[str, bytes]]): 由 process_images_concurrently 返回的结果列表。 output_base_dir (str): 输出图片的基础目录。 output_dir Path(output_base_dir) output_dir.mkdir(parentsTrue, exist_okTrue) # 创建输出目录如果已存在则忽略 saved_count 0 for original_path, image_data in results: original_path_obj Path(original_path) # 构建输出文件名可以保留原名并添加后缀如‘_colorized’ new_filename original_path_obj.stem _colorized original_path_obj.suffix # 可以选择保留原文件夹结构或者全部放在一个文件夹里 # 这里示例为简单起见全部放在output_base_dir下 output_path output_dir / new_filename try: with open(output_path, wb) as f: f.write(image_data) saved_count 1 # print(f已保存: {output_path}) # 如需详细日志可取消注释 except IOError as e: print(f保存文件失败 {output_path}: {e}) print(f所有图片保存完毕。共计保存 {saved_count} 张图片至目录: {output_dir.absolute()})这个函数会在当前目录下创建一个叫colorized_output的文件夹你可以随便改名然后把所有处理好的图片按照“原文件名_colorized.后缀”的格式保存进去。这样既不会覆盖原图也一目了然。3. 组装完整脚本与使用示例好了积木都准备好了现在把它们搭建成一个完整的、可以直接运行的工具。创建一个新的Python文件比如叫做batch_deoldify.py然后把下面的代码全部复制进去。#!/usr/bin/env python3 DeOldify 批量图像上色命令行工具 用于自动化处理文件夹内的所有图片。 import argparse import os import time from pathlib import Path from typing import List, Tuple import requests import concurrent.futures # ---------- 第一部分核心功能函数 ---------- def get_image_files(folder_path, extensions(.jpg, .jpeg, .png, .bmp, .tiff, .webp)): 获取文件夹内所有图片文件路径。 image_files [] folder Path(folder_path) if not folder.exists() or not folder.is_dir(): print(f错误路径 {folder_path} 不存在或不是一个文件夹。) return image_files for ext in extensions: image_files.extend(folder.rglob(f*{ext})) image_files.extend(folder.rglob(f*{ext.upper()})) image_files list(set([str(f) for f in image_files])) return image_files def colorize_single_image(image_path, api_url): 调用API处理单张图片。 try: with open(image_path, rb) as image_file: files {image: (image_path, image_file, image/jpeg)} response requests.post(api_url, filesfiles, timeout60) # 设置超时时间 if response.status_code 200: return response.content else: print(f警告处理 {image_path} 失败状态码 {response.status_code}) return None except Exception as e: print(f请求出错 ({image_path}): {e}) return None def colorize_single_image_with_retry(image_path, api_url, max_retries3, delay2): 带重试的单张图片处理。 for attempt in range(max_retries): result colorize_single_image(image_path, api_url) if result: return result elif attempt max_retries - 1: print(f - 第{attempt 1}次尝试失败{delay}秒后重试...) time.sleep(delay) print(f - 图片 {os.path.basename(image_path)} 重试{max_retries}次后失败。) return None def process_images_concurrently(image_paths, api_url, max_workers4): 并发处理图片列表。 results [] print(f开始并发处理 {len(image_paths)} 张图片并发数{max_workers}...) with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_path {executor.submit(colorize_single_image_with_retry, path, api_url): path for path in image_paths} for future in concurrent.futures.as_completed(future_to_path): img_path future_to_path[future] try: img_data future.result() if img_data: results.append((img_path, img_data)) print(f ✓ 完成: {os.path.basename(img_path)}) except Exception as e: print(f ✗ 处理 {os.path.basename(img_path)} 时发生未捕获异常: {e}) return results def save_colorized_images(results, output_dir): 保存处理后的图片到指定目录。 output_path Path(output_dir) output_path.mkdir(parentsTrue, exist_okTrue) saved_count 0 for orig_path, img_data in results: orig Path(orig_path) # 生成新文件名避免覆盖 new_name f{orig.stem}_colorized{orig.suffix} save_path output_path / new_name try: with open(save_path, wb) as f: f.write(img_data) saved_count 1 except IOError as e: print(f ! 无法保存 {new_name}: {e}) return saved_count # ---------- 第二部分主函数与命令行接口 ---------- def main(): parser argparse.ArgumentParser(description批量调用DeOldify API为图片上色) parser.add_argument(-i, --input, requiredTrue, help输入文件夹路径包含待处理的图片) parser.add_argument(-o, --output, default./colorized_output, help输出文件夹路径 (默认: ./colorized_output)) parser.add_argument(-u, --url, requiredTrue, helpDeOldify API 的完整URL (例如: http://localhost:7860/colorize)) parser.add_argument(-w, --workers, typeint, default4, help并发工作线程数 (默认: 4)) parser.add_argument(-r, --retries, typeint, default3, help单张图片失败重试次数 (默认: 3)) args parser.parse_args() print(*50) print(DeOldify 批量上色工具启动) print(f输入目录: {args.input}) print(f输出目录: {args.output}) print(fAPI地址: {args.url}) print(f并发数: {args.workers}, 重试次数: {args.retries}) print(*50) # 1. 获取图片列表 print(\n[步骤1] 扫描图片文件...) image_list get_image_files(args.input) if not image_list: print(未找到任何图片文件程序退出。) return print(f找到 {len(image_list)} 张待处理图片。) # 2. 批量处理 print(f\n[步骤2] 开始批量上色处理...) # 注意这里为了演示将重试次数参数传入了并发函数。更严谨的做法是修改函数签名。 # 这里采用一个简化方案在并发函数内部使用默认重试。 processed_results process_images_concurrently(image_list, args.url, args.workers) # 3. 保存结果 print(f\n[步骤3] 保存处理后的图片...) saved_num save_colorized_images(processed_results, args.output) print(f成功保存 {saved_num} 张图片到 {args.output}。) print(\n批量处理完成) if __name__ __main__: main()如何使用这个脚本保存代码将上面的完整代码保存为batch_deoldify.py。安装依赖确保安装了requests库。在终端运行pip install requests。准备环境确保你的DeOldify服务已经启动并知道它的API地址例如http://192.168.1.100:7860/colorize。运行脚本打开终端命令行切换到脚本所在目录使用以下命令格式运行python batch_deoldify.py -i /path/to/your/old_photos -o ./results -u http://your-deoldify-server:port/colorize -w 4参数解释-i或--input必须。指定存放老照片的文件夹路径。-o或--output可选。指定保存上色后图片的文件夹默认是当前目录下的colorized_output。-u或--url必须。你的DeOldify服务的API地址。-w或--workers可选。并发处理的数量默认是4。如果你的电脑或服务器性能一般可以调小点比如2。-r或--retries可选。单张图片处理失败后的重试次数默认是3。运行后脚本会自动扫描输入文件夹里的所有图片并发地调用API处理显示处理进度最后把彩色化的图片保存到输出文件夹。你只需要泡杯茶等着就行了。4. 脚本还能怎么改进上面这个脚本已经能解决大部分批量处理的需求了。但如果你想让它更强大、更贴心还可以考虑下面这些升级方向进度条显示处理几百张图片时有个进度条看着会安心很多。可以用tqdm这个库轻松实现。更细致的错误处理区分网络错误、服务器错误、图片格式错误等并记录到日志文件中方便后续排查。支持更多API参数如果DeOldify API支持调整渲染强度、艺术风格等参数可以在脚本里加入选项让用户自定义。保留目录结构现在所有输出图片都放在一个文件夹里。可以修改保存逻辑在输出文件夹里镜像输入文件夹的目录结构。做成图形界面GUI用PyQt或Tkinter给脚本套个壳做成一个带按钮和进度显示的小软件对不熟悉命令行的朋友更友好。这些功能你可以根据自己的需要慢慢添加。编程的乐趣就在于你可以不断打磨你的工具让它完全贴合你的工作流。写这个脚本的过程其实就是把重复劳动自动化的一次小实践。核心思路很简单找到文件 - 循环处理 - 保存结果。中间加上并发和重试就能让它变得又快又稳。希望这个教程能帮你解放双手让你有更多时间去欣赏那些被赋予新色彩的老照片而不是浪费在重复的点击操作上。动手试试吧遇到问题多查查资料你会发现其实没那么难。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。