DeOldify结合Python爬虫：自动采集并上色网络历史图片

DeOldify结合Python爬虫自动采集并上色网络历史图片你有没有想过那些尘封在网络角落里的黑白老照片如果能一键恢复色彩会是什么样子对于历史爱好者、内容创作者或者家族记忆的守护者来说这曾经是个耗时耗力的手工活。但现在事情变得简单了。我最近就折腾了这么一套系统用Python爬虫去网上自动“淘”那些有故事的黑白照片然后交给AI上色模型DeOldify让它们自动焕发新生。整个过程从采集、处理到归档完全自动化。这不仅仅是技术上的缝合更像是在数字世界里搭建了一条小小的“时光修复流水线”。今天我就来聊聊怎么把这两项看起来不相关的技术——网络爬虫和AI图像处理——揉在一起做成一个真正能跑起来的自动化工具。你会发现当AIGC的创造力遇上爬虫的自动化能力能玩出不少实用的新花样。1. 为什么需要自动化历史图片上色在动手之前我们得先想清楚费这么大劲搞自动化到底图个啥手动下载几张图拖到DeOldify的网页版里点一下不是更省事吗还真不是。当你面对的不是几张而是几百、几千张可能散落在不同网站、不同页面的历史图片时手动操作就成了一场噩梦。效率低下不说还容易出错、遗漏。我最初的想法很简单作为一个历史内容的自媒体运营者我需要持续地寻找高质量的历史素材并进行视觉化修复。手动操作让我每天耗费数小时在重复劳动上。这套自动化系统的核心价值就在于把“寻找-下载-处理-保存”这个链条打通了。爬虫负责不知疲倦地、按规则地从源头获取图片DeOldify负责稳定、批量地进行高质量上色而我们的程序则作为调度中心管理整个流程。最终我们得到的是一个不断扩大的、已上色的历史图片库随时可以调用这比每次临时找图、处理要高效得多。2. 系统设计与核心思路在开始写代码前我们先画个蓝图看看这个系统到底由哪些部分组成它们是怎么协作的。整个系统的运行就像一个高效的小工厂。采集车间爬虫负责从原料库目标网站搬运原始物料黑白图片。调度中心主程序收到物料后将其送入加工车间DeOldify服务进行精加工上色。最后加工好的成品彩色图片被贴上标签存入成品仓库本地目录或数据库。任何一个环节卡住比如网络不通或者加工车间忙不过来调度中心都会记录下来稍后重试保证流水线不会彻底停工。2.1 技术组件选型明确了流程我们来看看需要哪些工具来实现它。爬虫框架Requests BeautifulSoup。这是Python里最经典、最易上手的组合。Requests库负责模拟浏览器去访问网页、下载数据BeautifulSoup则像一把手术刀帮我们从复杂的网页HTML代码中精准地提取出图片链接。对于大多数静态历史图库网站这个组合完全够用。上色引擎DeOldify。这是一个基于深度学习的老照片上色项目效果在开源方案中相当出色。它提供了训练好的模型我们不需要自己从头训练可以直接调用它的API或者命令行工具来给图片上色。为了简化我们假设你已经通过Docker或其他方式部署好了DeOldify的REST API服务它会在本地某个端口比如7860等待我们提交图片。任务管理Python内置队列。我们需要处理可能成百上千张图片不能一股脑同时扔给DeOldify那样会把它压垮。这里用一个简单的queue.Queue就能实现一个生产者-消费者模型。爬虫作为生产者不断把找到的图片URL放入队列另一组处理线程作为消费者从队列里取URL下载、提交上色、保存结果。存储方案文件系统 简易索引。上色后的图片按日期或主题分类保存到本地文件夹。同时用一个CSV文件或者轻量级的SQLite数据库记录每张图片的“元数据”原始URL、来源网站、上色时间、保存路径等。这样以后想根据主题搜索图片就非常方便了。3. 动手搭建从爬虫开始理论说得再多不如一行代码。我们先从系统的起点——爬虫开始。我们的目标是某个假设的历史图片网站example-history.com它的图片列表页结构简单一页显示20张缩略图。爬虫的任务就是遍历这些列表页把每张图片的大图地址找出来。import requests from bs4 import BeautifulSoup import time import os from urllib.parse import urljoin class HistoryImageCrawler: def __init__(self, base_url, start_page1, max_pages5, download_dirraw_images): 初始化爬虫 :param base_url: 图片列表页的基础URL例如 https://example-history.com/photos?page :param start_page: 起始页码 :param max_pages: 最大爬取页数防止过量 :param download_dir: 原始图片下载目录 self.base_url base_url self.current_page start_page self.max_pages max_pages self.download_dir download_dir self.image_urls [] # 用于存储收集到的图片URL # 创建下载目录 os.makedirs(self.download_dir, exist_okTrue) # 设置一个简单的请求头模拟浏览器访问 self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 } def fetch_page(self, page_num): 获取指定页码的网页内容 url f{self.base_url}{page_num} try: print(f正在抓取页面: {url}) response requests.get(url, headersself.headers, timeout10) response.raise_for_status() # 检查请求是否成功 return response.text except requests.RequestException as e: print(f抓取页面 {url} 失败: {e}) return None def parse_image_links(self, html_content): 从网页HTML中解析出图片详情页链接或直接的大图链接 if not html_content: return [] soup BeautifulSoup(html_content, html.parser) links [] # 假设图片被包裹在class为photo-item的div中的a标签里 # 你需要根据目标网站的实际结构修改这里的选择器 for item in soup.select(div.photo-item a): href item.get(href) if href: # 将相对URL转换为绝对URL full_url urljoin(self.base_url, href) links.append(full_url) return links def get_image_url_from_detail(self, detail_url): 从图片详情页中提取高清大图的直接URL try: html requests.get(detail_url, headersself.headers, timeout10).text soup BeautifulSoup(html, html.parser) # 假设高清大图的URL在img标签的src属性中且该img有特定id # 实际选择器需要根据网站调整 img_tag soup.select_one(img#high-res-image) if img_tag and img_tag.get(src): return urljoin(detail_url, img_tag[src]) except Exception as e: print(f从详情页 {detail_url} 提取图片失败: {e}) return None def run(self): 启动爬虫的主循环 print(历史图片爬虫开始工作...) while self.current_page self.max_pages: print(f\n--- 正在处理第 {self.current_page} 页 ---) # 1. 抓取列表页 html self.fetch_page(self.current_page) if not html: self.current_page 1 time.sleep(2) # 礼貌性延迟避免请求过快 continue # 2. 解析出详情页链接 detail_links self.parse_image_links(html) print(f在本页找到 {len(detail_links)} 个图片详情链接) # 3. 遍历每个详情页获取大图URL for d_link in detail_links: img_url self.get_image_url_from_detail(d_link) if img_url: self.image_urls.append(img_url) print(f 已收集大图: {os.path.basename(img_url)}) time.sleep(1) # 访问详情页的延迟 self.current_page 1 time.sleep(3) # 翻页延迟 print(f\n爬取结束共收集到 {len(self.image_urls)} 张图片URL。) return self.image_urls # 使用示例 if __name__ __main__: # 请务必替换成你要爬取的真实网站URL并遵守该网站的robots.txt协议 crawler HistoryImageCrawler( base_urlhttps://example-history.com/photos?page, start_page1, max_pages3, # 先爬3页试试水 download_dir./raw_images ) all_image_urls crawler.run()这段代码定义了一个基础的爬虫类。它做的事情很清晰一页一页地翻列表找到图片详情页的链接然后再去每个详情页里把高清大图的真实地址挖出来。这里有两个关键点需要注意一是User-Agent头它让我们的请求看起来更像普通浏览器减少被直接拒绝的风险二是代码中无处不在的time.sleep()这是最基本的“礼貌”给目标网站的服务器喘息的时间也是对抗反爬机制的第一步。请注意在实际使用中你必须将base_url替换为你要爬取的真实网站地址并且务必先检查该网站的robots.txt文件尊重网站关于爬虫的规则。此外网站的实际HTML结构千差万别代码中的div.photo-item a和img#high-res-image等选择器需要你用浏览器的开发者工具仔细分析后进行调整。4. 构建自动化处理流水线图片URL收集好了堆在列表里。接下来我们要设计一条流水线让这些URL自动排队经过下载、上色、保存这几道工序。这里我们会用到Python的threading和queue模块创建一个“生产者-消费者”模型。爬虫线程或主线程作为生产者不断向任务队列里投放图片URL。同时我们启动几个工作线程作为消费者它们从队列里领取任务执行完整的处理流程。import queue import threading import requests from PIL import Image import io import csv from datetime import datetime class ImageProcessingPipeline: def __init__(self, deoldify_api_urlhttp://localhost:7860/api/predict, output_dircolorized_images, max_workers2): 初始化处理流水线 :param deoldify_api_url: 部署好的DeOldify API地址 :param output_dir: 上色后图片的输出目录 :param max_workers: 并发处理的工作线程数 self.task_queue queue.Queue() self.deoldify_api_url deoldify_api_url self.output_dir output_dir self.max_workers max_workers self.lock threading.Lock() # 用于安全地写入日志和索引 os.makedirs(self.output_dir, exist_okTrue) # 初始化索引文件CSV格式 self.index_file os.path.join(output_dir, image_index.csv) if not os.path.exists(self.index_file): with open(self.index_file, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([original_url, source, colorized_time, file_path, status]) def add_tasks(self, image_url_list): 将爬虫收集到的图片URL添加到任务队列 for url in image_url_list: self.task_queue.put(url) print(f已添加 {len(image_url_list)} 个任务到队列。) def worker(self): 工作线程函数从队列中取任务并处理 while True: try: img_url self.task_queue.get(timeout5) # 等待5秒无任务则退出 if img_url is None: # 接收到终止信号 break print(f线程 {threading.current_thread().name} 正在处理: {img_url}) self.process_single_image(img_url) self.task_queue.task_done() # 标记任务完成 except queue.Empty: print(f线程 {threading.current_thread().name} 等待任务超时退出。) break except Exception as e: print(f处理 {img_url} 时发生错误: {e}) with self.lock: self._log_to_index(img_url, Failed, errorstr(e)) self.task_queue.task_done() def process_single_image(self, img_url): 处理单张图片的完整流程下载 - 调用DeOldify上色 - 保存 # 1. 下载原始图片 raw_image_data self.download_image(img_url) if not raw_image_data: with self.lock: self._log_to_index(img_url, Download_Failed) return # 2. 调用DeOldify API进行上色 colorized_image_data self.call_deoldify_api(raw_image_data) if not colorized_image_data: with self.lock: self._log_to_index(img_url, API_Failed) return # 3. 保存上色后的图片 saved_path self.save_colorized_image(img_url, colorized_image_data) if saved_path: with self.lock: self._log_to_index(img_url, Success, file_pathsaved_path) print(f 处理成功保存至: {saved_path}) else: with self.lock: self._log_to_index(img_url, Save_Failed) def download_image(self, url): 下载图片返回二进制数据 try: response requests.get(url, timeout15, headers{User-Agent: Mozilla/5.0}) response.raise_for_status() # 简单验证是否为图片 if image in response.headers.get(Content-Type, ): return response.content else: print(f 警告: {url} 返回的内容不是图片。) return None except Exception as e: print(f 下载图片失败 {url}: {e}) return None def call_deoldify_api(self, image_data): 调用DeOldify API提交图片并获取上色结果 try: # 假设DeOldify API接收multipart/form-data格式字段名为image files {image: (image.jpg, image_data, image/jpeg)} response requests.post(self.deoldify_api_url, filesfiles, timeout60) # 上色可能较慢超时设长 response.raise_for_status() result response.json() # 假设API返回的JSON中包含base64编码的图片数据字段为image import base64 if result.get(image): return base64.b64decode(result[image]) else: print(f API响应中未找到图片数据。) return None except Exception as e: print(f 调用DeOldify API失败: {e}) return None def save_colorized_image(self, original_url, image_data): 保存上色后的图片到本地文件名包含时间戳和源文件名 try: # 从URL中提取原始文件名或生成一个唯一文件名 filename os.path.basename(original_url).split(?)[0] # 去掉URL参数 if not filename or . not in filename: filename unknown_image.jpg # 生成新文件名避免重复 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) name_part, ext os.path.splitext(filename) new_filename f{name_part}_{timestamp}{ext} save_path os.path.join(self.output_dir, new_filename) with open(save_path, wb) as f: f.write(image_data) # 可选用PIL验证一下图片是否有效 Image.open(io.BytesIO(image_data)).verify() return save_path except Exception as e: print(f 保存图片失败: {e}) return None def _log_to_index(self, original_url, status, file_path, error): 将处理结果记录到CSV索引文件中 with open(self.index_file, a, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([ original_url, example-history.com, # 可以改为从URL解析 datetime.now().isoformat(), file_path, status, error ]) def start(self): 启动处理流水线 print(启动图片处理流水线...) threads [] for i in range(self.max_workers): t threading.Thread(targetself.worker, namefWorker-{i1}) t.start() threads.append(t) # 等待所有任务被处理完成 self.task_queue.join() # 通知所有工作线程退出 for _ in range(self.max_workers): self.task_queue.put(None) for t in threads: t.join() print(所有图片处理完成) # 主程序串联爬虫和流水线 if __name__ __main__: # 1. 运行爬虫收集图片URL crawler HistoryImageCrawler( base_urlhttps://example-history.com/photos?page, max_pages2, download_dir./raw_images ) image_urls crawler.run() if not image_urls: print(未收集到任何图片URL程序退出。) exit() # 2. 初始化并启动处理流水线 pipeline ImageProcessingPipeline( deoldify_api_urlhttp://localhost:7860/api/predict, # 替换为你的DeOldify API地址 output_dir./colorized_output, max_workers2 # 根据你的机器性能和DeOldify服务能力调整 ) # 3. 将任务加入队列并启动 pipeline.add_tasks(image_urls) pipeline.start() print(f处理流水线已关闭。结果保存在 {pipeline.output_dir} 中。) print(f处理日志见: {pipeline.index_file})这段代码构建了整个系统的核心。ImageProcessingPipeline类封装了从任务队列管理、多线程工作到最终保存的全过程。每个worker线程独立工作从共享队列里取URL然后按部就班地执行下载-API调用-保存这三步。用锁Lock来确保多个线程同时写日志文件时不会出错。最后用一个CSV文件记录每张图片的处理履历方便后续管理和检索。请注意代码中调用DeOldify API的部分call_deoldify_api方法是一个示例。实际部署的DeOldify服务其API接口格式接收的参数、返回的数据结构可能有所不同。你需要根据自己部署的服务文档来调整这部分代码。同样线程数max_workers需要根据你的电脑CPU核心数和DeOldify服务的处理能力来调整不是越多越好。5. 实际运行中的挑战与优化把代码跑起来你可能会遇到一些计划外的情况。这才是工程实践有趣的地方。下面是我在搭建过程中遇到的几个典型问题以及应对思路。反爬虫策略这是爬虫的永恒课题。除了代码中已经使用的设置User-Agent和添加延迟如果目标网站有更复杂的防护如验证码、请求频率限制、JavaScript动态加载你可能需要使用Session对象保持会话状态处理Cookie。轮换User-Agent和IP代理池模拟不同用户避免单一IP被封锁。可以使用一些免费的代理IP网站但稳定性和速度需要测试。处理动态内容如果图片链接是由JavaScript生成的BeautifulSoup就无能为力了。这时可以考虑使用Selenium或Playwright这类浏览器自动化工具来模拟真实用户操作获取渲染后的页面内容。但这会大大增加复杂性和运行时间。DeOldify服务稳定性本地部署的AI模型服务有时不太稳定长时间处理可能内存泄漏或崩溃。增加重试机制在call_deoldify_api方法里如果请求失败超时或返回错误可以加入一个重试循环比如重试3次每次间隔一段时间。健康检查在主程序里可以定时ping一下DeOldify服务的健康检查端点如果有的话如果发现服务挂了可以暂停任务队列并发送通知比如邮件或桌面通知。结果验证保存图片后用PIL库简单打开验证一下文件是否完整有效避免保存了损坏的图片。任务管理与监控当任务量很大时你需要知道进度如何哪里卡住了。进度可视化可以定期打印队列剩余任务数或者用tqdm库创建一个进度条。更详细的日志除了记录成功失败还可以记录每张图片的处理耗时、下载文件大小等便于后期分析性能瓶颈。断点续传将已成功处理的图片URL记录到一个“已完成”列表里。如果程序中途崩溃重启后可以先加载这个列表跳过已经处理过的任务从断点继续。6. 总结回过头来看这个项目其实是一个经典的“数据采集智能处理”的微缩案例。技术本身并不高深用到的都是Python生态里很成熟的库。它的价值在于思路和整合将面向特定领域的爬虫历史图片与通用的AIGC能力图像上色通过一个自动化的调度程序连接起来创造出一个能持续产生价值的工具。对我自己而言运行这套系统后素材收集和处理的效率提升了不止一个量级。我可以设置好规则让它在夜里自动运行早上起来就能看到一个新增了几十张彩色历史图片的素材库。更重要的是这个过程是可复现、可扩展的。今天的目标是历史图片上色明天完全可以将爬虫规则改成抓取设计草图然后对接一个线稿上色模型或者抓取商品白底图自动生成营销场景图。当然现阶段的系统还有很多可以打磨的地方比如加入更友好的图形界面来配置爬虫规则和监控任务状态或者将结果图片自动上传到云存储并生成在线相册。但最重要的是它已经能跑起来并且实实在在地解决问题了。技术学习的乐趣莫过于此——用代码将想法变成现实让机器帮你完成那些重复而有趣的工作。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。