做图片采集这类的爬虫工程跟普通网页爬虫完全是两码事。网页爬虫拿到文本、存进数据库就算完事图片采集要过三关第一关是下载得把图片真实落地到磁盘第二关是去重很多站点的图片会以不同 URL、不同尺寸反复出现不下下来根本不知道它是同一张第三关是入库图片本身的二进制、来源 URL、尺寸、字节数、哈希值要一起记下来这批数据才算真正可用。这篇文章围绕“下载 → 感知哈希去重 → 元数据入库”这条主线把整个工程从设计到落地拆开讲透适合有 Python 基础、想建图片素材库或者做内容型数据采集的朋友参考。1. 图片采集工程整体设计先想清楚再动手1.1 一条清晰的采集链路比代码本身更重要我把成熟工程的骨架拉出来给你看。一个能长期跑的图片采集项目绝不是“在循环里写个 requests.get 然后 save”就完事了而是要拆成三段独立模块来设计。第一段是下载。这步负责把远端图片变成本地文件。听起来简单实际坑最多请求头构造不对对方直接 403图片 URL 是动态签名的几分钟就过期网络一抖下载到一半就断了有些 URL 返回的其实是 HTML 错误页而不是图片。下载模块的核心目标是让“拿到 URL 就能稳定产出本地图片”不管中间出什么幺蛾子都能自动重试或者明确失败。第二段是去重。下载成功不等于万事大吉采集过程中最容易被忽略的就是重复图片。同一个素材在网站的不同列表页、不同分页、不同 CDN 上URL 可能完全不同但图是同一张。如果没有去重环节磁盘和数据库很快会被重复内容塞满。去重要解决的核心问题是两张图片的字节数据可能不一样尺寸、压缩率都被改过但内容上依然是同一张图怎么把它们识别出来。第三段是入库。图片文件只是原料要让这批图能被检索、被使用必须把每个文件的元信息记录在案原始 URL、保存路径、文件大小、像素宽高、采集时间、来源站点以及前面算出来的感知哈希值。有了这份数据库后续无论是做素材检索、相似图片搜索还是出数据报表底层都是扎实的。这三段之间是流水线关系下载产出文件去重筛掉重复文件入库把剩余有效文件登记在册。把三件事分开设计的好处是每一段都能独立测试、独立优化。下载慢就调并发去重误判就调阈值入库写入慢就改批量提交互不干扰。1.2 去重方案为什么选感知哈希而不是 MD5 或 URL 比较“去重”这个环节的选择直接决定整个项目的数据质量。我见过很多人一上来就用 MD5理由是简单——但字节级去重在图片场景下几乎没用。同一张图网站给缩略图和原图各保存了一份MD5 完全不一样查重查了个寂寞。先看一张对比表。方案判定原理适用场景局限URL 去重URL 字符串完全相同同一站点的列表页去重同一图片多 URL、CDN 镜像均判不了MD5 / SHA1文件内容完全一致任何字节变化都会变本地文件去重、精确副本检测图片压缩、重编码、改尺寸即失效感知哈希提取图像内容特征计算特征距离内容相似图片去重、近似图检索对不同风格图片可能误判阈值要调做采集这几年我最大的体会是不要把去重押在一种方案上。URL 去重能挡掉最明显的重复速度快放在下载之前做MD5 在文件落地后立刻算一遍挡住字节级完全相同的重复感知哈希才是最后一道兜底用来识别那些“文件名不同、URL 不同、字节不同但内容长得一样”的图片。为什么很多教程重点推感知哈希因为图片采集场景里最常见的情况就是内容相同但格式和尺寸不同。同一个产品图网页里有缩略图、有原图、有 WebP 版本、有老 CDN 缓存版本MD5 全都不一样但人眼一看是同一张。感知哈希尤其是 pHash 和 dHash经过缩放、压缩、轻微裁剪哈希值依然保持接近可以按汉明距离判断相似度。这一点正好卡在图片采集的痛点需求上。顺便回应一个常见问题现在的 AI 图像理解技术当然比传统哈希更智能能识别“颜色不同但同款式的衣服”这种语义相似但那是特征向量检索的范畴要上模型、跑 GPU资源开销很大。传统感知哈希几十行代码、毫秒级耗时在采集管道里做初筛已经足够用了。真要追求极致语义相似再升级到向量检索方案也不迟但那就是另一个量级的项目了。2. 开工前准备环境、依赖与并发选型2.1 一份能直接跑起来的环境清单动手前先把环境备好。Python 版本建议 3.9 以上我自己测试用的是 3.10。依赖其实非常少核心就三个requests 负责 HTTP 下载Pillow 负责图片打开与基础校验imagehash 负责算感知哈希。sqlite3 是 Python 标准库直接 import 就行不需要额外安装如果以后数据量上到百万级可以换成 PostgreSQL架构不用变。requirements 文件长这样requests2.31.0 Pillow10.1.0 imagehash4.3.1安装命令还是老样子pip install -r requirements.txt工程目录我习惯按模块职责拆分后续换数据库、改下载方式都只动其中一个文件image_collector/ ├── main.py # 主入口串联下载-去重-入库 ├── config.py # 配置URL列表、并发数、阈值等 ├── downloader.py # 下载模块 ├── deduplicator.py # 感知哈希去重模块 ├── storage.py # 数据库模块 └── images/ # 下载文件存放目录如果你是刚配好 Python 环境提醒一句Pillow 和 imagehash 都有 C 扩展Windows 下要注意 Python 位数和包架构匹配64 位 Python 配 64 位包。遇到安装报错先查 Python 版本和 pip 版本是不是对应。2.2 并发设计线程池和协程到底选哪个图片下载是典型的 IO 密集型任务大部分时间都花在网络等待上。并发模型怎么选是爬虫工程师经常纠结的问题。我的结论很直接第一版用线程池数据量大了再考虑协程。理由有三条。第一线程池写起来简单、出错好排查concurrent.futures.ThreadPoolExecutor 十几行就能搞定协程要处理事件循环、await 链、任务调度对新手不友好调试成本也高。第二图片下载的瓶颈通常是对方网站的限速而不是你的网络能力线程数 8 到 16 足够跑满绝大多数场景的带宽协程带来的“高并发”优势在受限场景下并不明显。第三协程确实省内存但图片下载场景中内存大头在图片文件处理和拼接上协程省下的那部分开销杯水车薪。当然如果你的目标是几百个站点、每天上百万张图片、需要在单机上同时跑几千个任务那 asyncio aiohttp 是更好的选择。连接复用和低内存占用是线程池替代不了的。我的做法是在 downloader.py 里把下载动作抽象成单一函数线程池版本写得通协程改造时只需把主流程里的 ThreadPoolExecutor 换成 asyncio.gather底层下载函数不用大改。线程池的基本写法from concurrent.futures import ThreadPoolExecutor, as_completed def run_download(urls: list[str]) - None: with ThreadPoolExecutor(max_workers8) as executor: futures {executor.submit(download_one, url): url for url in urls} for future in as_completed(futures): url futures[future] try: result future.result() print(f下载成功: {url} - {result}) except Exception as e: print(f下载失败: {url}原因: {e})这个结构简单可读业务逻辑都在 download_one 里主流程不塞太多东西。3. 核心模块实现下载、去重、入库的完整代码拆解3.1 图片下载模块不能只写一个 requests.get很多新手写的下载代码就两行resp requests.get(url)然后open(a.jpg, wb).write(resp.content)。这种写法在本地测试一两张图没问题一上生产环境就各种翻车。一个能扛住真实采集压力的下载模块至少要考虑这几件事Session 复用连接池、完整请求头、超时控制、流式下载、响应类型校验。下面这份是我常用的实现可以直接抄import hashlib import imghdr from pathlib import Path import requests from requests.adapters import HTTPAdapter DOWNLOAD_DIR Path(images) DOWNLOAD_DIR.mkdir(exist_okTrue) HEADERS { User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ), Accept: image/avif,image/webp,image/apng,image/*,*/*;q0.8, } def make_session() - requests.Session: session requests.Session() session.headers.update(HEADERS) adapter HTTPAdapter(pool_connections20, pool_maxsize20, max_retries3) session.mount(http://, adapter) session.mount(https://, adapter) return session def download_one(session: requests.Session, url: str) - str | None: try: resp session.get(url, timeout(5, 30), streamTrue) resp.raise_for_status() # 第一步校验Content-Type content_type resp.headers.get(Content-Type, ) if not content_type.startswith(image/): return None # 第二步校验文件头 data b.join(resp.iter_content(chunk_size8192)) if imghdr.what(None, data) is None: return None # 文件名用 URL 的 MD5天然防重名也避开特殊字符 suffix Path(url.split(?)[0]).suffix.lower() if suffix not in (.jpg, .jpeg, .png, .gif, .webp, .bmp): suffix .jpg file_name hashlib.md5(url.encode()).hexdigest() suffix file_path DOWNLOAD_DIR / file_name file_path.write_bytes(data) return str(file_path) except requests.RequestException: return None这里有几个细节值得展开。第一timeout(5, 30)的写法5 秒是连接超时30 秒是读取超时两个都要设否则某个慢服务器能把你线程挂住半天。第二用iter_content分块拼接而不是直接resp.content大图不会一次性占满内存。第三Content-Type 校验和imghdr双重过滤能挡掉大量“201 状态码 HTML 错误页”的假图片。第四文件名用 URL 的 MD5这样做既避免两台 CDN 重复下载同一张图又避免原始文件名里的特殊字符和超长文件名导致的保存失败。3.2 感知哈希去重原理与实现去重是整个工程的灵魂。我拿 dHash差异哈希来讲解因为它最直观、速度最快、写起来也最短。dHash 的原理可以拆成四步。第一步把图片缩放到 9x8 像素的灰度图。注意是 9 列 8 行因为要比较“当前像素和右侧相邻像素”的亮度差9 列才能产生 8 个差值最终得到 8x864 位哈希。第二步把每个像素转换成灰度值对每一行的 9 个像素从左到右两两比较左侧比右侧亮记为 1否则记为 0。第三步把 64 个 0/1 按顺序拼成二进制串再转成十六进制字符串这就是这张图的 dHash。第四步两张图的 dHash 做异或统计结果中 1 的个数也就是汉明距离。距离越小图片越相似。自己实现一份并不难from PIL import Image def compute_dhash(image: Image.Image, hash_size: int 8) - str: 将 PIL Image 对象转换为 dHash 十六进制字符串 gray image.convert(L).resize((hash_size 1, hash_size), Image.LANCZOS) diff [] for row in range(hash_size): for col in range(hash_size): left gray.getpixel((col, row)) right gray.getpixel((col 1, row)) diff.append(1 if left right else 0) bits .join(diff) return .join(hex(int(bits[i : i 4], 2))[2:] for i in range(0, 64, 4))用 imagehash 库更省事一行就出结果import imagehash img_hash imagehash.dhash(img) # 或者 imagehash.phash(img)imagehash 库内部已经处理了灰度化、缩放的细节返回的是一个哈希对象可以直接和另一个哈希对象做减法得到的就是汉明距离。distance img_hash - another_hash print(distance)阈值怎么定我实测过很多场景dHash 的汉明距离阈值设在 10 以内判定为重复比较合适pHash 更稳定一点阈值放到 16 也没问题。阈值太小会漏掉相似图阈值太大又误杀内容不同的图片。最稳妥的做法是先用一批已标注的样本跑一遍画一个距离分布直方图来确定分界线而不是拍脑袋定数值。实际测试有个例子我跑某个新闻网站时同一张配图在列表页是 300x200 的缩略图详情页是 1200x800 的原图dHash 距离只有 2被正确判定为重复。这就是感知哈希在图片采集里最有价值的地方。这里还要提醒一个坑部分 PNG 带透明通道直接convert(L)在某些 Pillow 版本下会得到全黑图或者报错。处理办法是先把透明背景合到白色底上再转灰度def flatten_alpha(img: Image.Image, bg_color(255, 255, 255)) - Image.Image: if img.mode in (RGBA, LA, P): img img.convert(RGBA) background Image.new(RGB, img.size, bg_color) background.paste(img, maskimg.split()[-1]) return background return img.convert(RGB)去重模块的完整逻辑是下载完成后打开图片算哈希拿哈希去数据库查有没有距离小于阈值的记录如果有判定重复删除本地文件如果没有插入记录并保留文件。3.3 元数据入库表结构怎么设计才不返工图片文件只是中间产物真正能支撑业务的是数据库里的元数据。表结构一开始没设计好后面返工的痛苦我深有体会。先看建表语句CREATE TABLE IF NOT EXISTS images ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL, file_path TEXT NOT NULL, file_size INTEGER, width INTEGER, height INTEGER, phash TEXT NOT NULL, source TEXT DEFAULT , created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_images_phash ON images(phash); CREATE UNIQUE INDEX IF NOT EXISTS idx_images_url ON images(url);字段含义一目了然url 是原始来源地址file_path 是本地保存路径file_size、width、height 是文件基础属性phash 是感知哈希值source 记录来源站点created_at 记录采集时间。为什么 url 要建唯一索引同一个 URL 重复采集时直接跳过省一次 IO。为什么 phash 要建普通索引查重的时候需要按前缀或者范围筛选候选集有索引能省很多时间。这里有一个必须注意的性能关键点如果每张图都全表扫描一遍数据库去算汉明距离图片量一上来就完蛋。SQLite 没有内置汉明距离函数常见做法是截取 phash 的前 8 位十六进制也就是 32 位二进制做粗筛再在粗筛结果里算完整距离def is_duplicate(conn, phash: str, threshold: int 10) - bool: prefix phash[:4] # 取前 4 位十六进制做粗筛 rows conn.execute( SELECT phash FROM images WHERE substr(phash, 1, 4) ?, (prefix,), ).fetchall() for (existing, ) in rows: distance bin(int(phash, 16) ^ int(existing, 16)).count(1) if distance threshold: return True return False因为相似图片的 phash 前几位通常一致这样先把候选集从几十万缩小到个位数再逐条算完整距离性能会快非常非常多。在几十万图片量级的场景下这个方案完全够用。插入的时候不要每张图都 commit 一次攒够 50 到 100 条再批量提交速度能差一个数量级def insert_many(conn, rows: list[tuple]) - None: conn.executemany( INSERT OR IGNORE INTO images (url, file_path, file_size, width, height, phash, source) VALUES (?, ?, ?, ?, ?, ?, ?), rows, ) conn.commit()3.4 把整条流水线串起来主流程与断点续传模块写好了主流程就是把它们串起来跑。核心逻辑并不复杂但有两个设计要点一是已经处理过的 URL 要跳过二是失败任务要单独记录。主循环代码骨架import sqlite3 import os from concurrent.futures import ThreadPoolExecutor, as_completed from PIL import Image import imagehash def main(): session make_session() conn sqlite3.connect(images.db) init_db(conn) url_queue load_urls(urls.txt) # 待采集任务列表 visited load_done(conn) # 已入库的 URL 集合 with ThreadPoolExecutor(max_workers8) as executor: futures {} for url in url_queue: if url in visited: continue futures[executor.submit(download_one, session, url)] url for future in as_completed(futures): url futures[future] try: file_path future.result() if not file_path: mark_failed(url) continue with Image.open(file_path) as img: width, height img.size img_hash imagehash.dhash(Image.open(file_path)) if is_duplicate(conn, str(img_hash)): os.remove(file_path) continue file_size os.path.getsize(file_path) insert_one( conn, urlurl, file_pathfile_path, file_sizefile_size, widthwidth, heightheight, phashstr(img_hash), sourceexample, ) except Exception as e: log_error(url, e)断点续传这块我要专门强调。采集工程跑到一半断掉是常态网络波动、服务器重启、磁盘满都有可能。所以工程从第一天起就要支持断点续传否则每次重跑都从头开始时间成本太高。具体做法不复杂每个任务完成后无论是下载成功、判定重复还是入库完成都把 URL 状态写入 task_state 表主流程启动时先查这张表已经处理过的 URL 直接跳过。下载失败的任务要单独记到 failed 列表等主流程跑完对 failed 列表做 2 到 3 轮重试重试仍然失败的把失败原因记下来不要静默丢弃。这样才能保证整个工程在无人值守的情况下也能稳定跑完。4. 常见问题与排查技巧实录4.1 反爬与访问限制的合规应对做采集的第一原则是尊重目标网站的使用条款和 robots.txt合理控制请求频率不给对方服务器制造压力。图片采集本身带宽占用大更要克制。我的做法是单域名请求间隔控制在 1 秒以上并发数不超过 8如果对方网站明显响应变慢主动降速而不是硬扛。遇到 403 或 429 状态码我有一套排查顺序按这个顺序来十有八九能解决状态码含义应对措施403 Forbidden被拒绝访问检查请求头、Referer、User-Agent 是否完整URL 可能已过期404 Not Found图片不存在/URL 失效跳过并记录不要重试429 Too Many Requests请求频率过高指数退避增加延时降低并发5xx服务器端错误可重试 2-3 次仍失败则记录403 最常见的原因是请求头不完整尤其 Referer 和 User-Agent很多站点会校验这两项。429 就是在告诉你并发太高了得马上降速。针对单个 IP 的限流我的原则是“合法数据源放慢速度都能解决”没必要搞什么高深技巧。如果确实需要大量数据找授权渠道或申请开放接口才是正路。重试策略我习惯用指数退避思路是第一次失败等 1 秒第二次等 2 秒第三次等 4 秒给服务器恢复的时间同时避免加重对方压力。import time for attempt in range(3): try: result download_one(session, url) break except Exception: time.sleep(2 ** attempt)4.2 下载的图片打不开或损坏怎么自动处理下载过程中最让人崩溃的不是请求失败而是“200 OK”返回了存到本地打开却是一片黑或者直接报错。我梳理了四个高频原因。第一URL 带签名下载过程中签名过期拿到的是一段错误内容。第二CDN 默认错误图比如一张纯白 1x1 像素的图请求失败时服务器也返回 200 和 image 类型但内容毫无价值。第三WebP 或 AVIF 格式在旧版 Pillow 里打不开。第四网络中断导致文件不完整。对策是下载完立即做三重校验Content-Type 前缀校验、imghdr文件头校验、Pillow 打开并执行verify()。其中verify()是最关键的一步它只校验文件结构不加载像素数据速度非常快import io def validate_image(data: bytes) - bool: try: with Image.open(io.BytesIO(data)) as img: img.verify() return True except (IOError, OSError, Image.DecompressionBombError): return False校验不过的直接标记失败不要入库。注意一点verify()之后图片对象就不能再读取尺寸了要拿宽高必须重新打开文件。还有一个容易踩的坑有些图片虽然合法但像素尺寸特别夸张比如全景长图。Pillow 默认有像素上限保护约 1.78 亿像素超过会抛DecompressionBombError。处理方式有两种一是调大上限Image.MAX_IMAGE_PIXELS None但强烈不建议更好的做法是在配置里设定最大宽高阈值超过直接丢弃。我在采集壁纸站点时遇到过全景图一张几十 MBconvert时内存直接飙到几百 MB后来在配置里加了最大边长过滤问题立刻解决。4.3 去重阈值太严或太松怎么调才合适感知哈希去重最怕两件事该去掉的重复没去掉不该去掉的被误杀。阈值调太大会误杀调太小会漏网到底怎么定先给一组经验值通用场景下dHash 阈值 10 左右产品图、图标、界面截图这类内容简单、色彩单一的图阈值调到 6 到 8自然风景、复杂纹理的图阈值可以放到 12 到 16。不同图片风格对哈希距离的分布影响很大照搬别人的阈值不一定适用。最靠谱的方式还是数据驱动。具体做法随机从库里抽 200 对已知重复的图片和 200 对不同内容的图片分别算汉明距离画两条分布曲线看分界点在哪里阈值取中间值。这套方法比拍脑袋定阈值靠谱得多而且每换一个采集源就能快速重新标定一次。我还要补充一个规律同一张图经过不同压缩算法处理后pHash 比 dHash 更稳定但 dHash 计算更快。我一般这样分工下载管线里用 dHash 做实时去重等数据量大了之后离线再用 pHash 做一次全量清洗。两条哈希链各司其职效果比只用一种要扎实。4.4 并发越高越快吗线程数与数据库写冲突的坑有朋友问过我“线程数调到 50速度没变快数据库还频繁报锁错误。”这基本就是并发没控好加上 SQLite 写入方式不对。先说线程数。图片下载是 IO 密集型没错但对方站点会限速网络链路也有物理上限。并发太高不仅不会更快还可能触发对方反爬甚至把自己机器的文件描述符打满。我自己的经验到单个站点采集8 个线程左右就能跑满常见带宽多站点同时采集每个站点单独一个线程池总线程数控制在 CPU 核心数×2 到×4 之间。另外要做限速每个线程下载完一张图至少 sleep 0.2 到 0.5 秒给服务器和自己都留点余量。再说 SQLite 的写冲突。SQLite 同一时刻只允许一个写事务多个线程同时执行 insert 会报database is locked。解决办法也不难不要在下载线程里直接写库把入库操作全部放进一个队列由专门的消费者线程串行提交。这个模式简单可靠实测几十万条记录入库能稳定保持在每秒几千条完全够用。import queue from threading import Thread write_queue queue.Queue() def writer_thread(conn): while True: item write_queue.get() if item is None: # 结束信号 break insert_one(conn, item) write_queue.task_done()下载线程只负责把结果put进write_queue这样写库压力完全由单线程承担天然不会有锁冲突。主线程结束时往队列里丢一个None作为结束信号写线程就自然退出。这个采集工程做下来我最想分享的经验是一定要把“去重”和“入库”当成核心功能来设计而不是等数据量大了再回头补。我第一版只写了下载跑了三天磁盘堆了一万多张图一查重复率超过三成重新整理花的时间比重写一遍还长。另一个小技巧是phash 入库之后不要只用来去重后续做相似图检索、素材聚类、版权溯源都靠它。如果你也在做图片采集建议从第一版就把“下载 → 感知哈希去重 → 元数据入库”的链路搭完整把断点续传做好后面会省下大量的重复劳动。