企业官网建设流程全解析

做图片采集这类的爬虫工程，跟普通网页爬虫完全是两码事。网页爬虫拿到文本、存进数据库就算完事；图片采集要过三关：第一关是下载，得把图片真实落地到磁盘；第二关是去重，很多站点的图片会以不同 URL、不同尺寸反复出现，不下下来根本不知道它是同一张；第三关是入库，图片本身的二进制、来源 URL、尺寸、字节数、哈希值要一起记下来，这批数据才算真正可用。这篇文章围绕“下载 → 感知哈希去重 → 元数据入库”这条主线，把整个工程从设计到落地拆开讲透，适合有 Python 基础、想建图片素材库或者做内容型数据采集的朋友参考。

1. 图片采集工程整体设计：先想清楚再动手

1.1 一条清晰的采集链路，比代码本身更重要

我把成熟工程的骨架拉出来给你看。一个能长期跑的图片采集项目，绝不是“在循环里写个 requests.get 然后 save”就完事了，而是要拆成三段独立模块来设计。

第一段是下载。这步负责把远端图片变成本地文件。听起来简单，实际坑最多：请求头构造不对，对方直接 403；图片 URL 是动态签名的，几分钟就过期；网络一抖，下载到一半就断了；有些 URL 返回的其实是 HTML 错误页，而不是图片。下载模块的核心目标，是让“拿到 URL 就能稳定产出本地图片”，不管中间出什么幺蛾子，都能自动重试或者明确失败。

第二段是去重。下载成功不等于万事大吉，采集过程中最容易被忽略的就是重复图片。同一个素材在网站的不同列表页、不同分页、不同 CDN 上，URL 可能完全不同，但图是同一张。如果没有去重环节，磁盘和数据库很快会被重复内容塞满。去重要解决的核心问题是：两张图片的字节数据可能不一样（尺寸、压缩率都被改过），但内容上依然是同一张图，怎么把它们识别出来。

第三段是入库。图片文件只是原料，要让这批图能被检索、被使用，必须把每个文件的元信息记录在案：原始 URL、保存路径、文件大小、像素宽高、采集时间、来源站点，以及前面算出来的感知哈希值。有了这份数据库，后续无论是做素材检索、相似图片搜索，还是出数据报表，底层都是扎实的。

这三段之间是流水线关系：下载产出文件，去重筛掉重复文件，入库把剩余有效文件登记在册。把三件事分开设计的好处是每一段都能独立测试、独立优化。下载慢就调并发，去重误判就调阈值，入库写入慢就改批量提交，互不干扰。

1.2 去重方案为什么选感知哈希，而不是 MD5 或 URL 比较

“去重”这个环节的选择，直接决定整个项目的数据质量。我见过很多人一上来就用 MD5，理由是简单——但字节级去重在图片场景下几乎没用。同一张图，网站给缩略图和原图各保存了一份，MD5 完全不一样，查重查了个寂寞。

先看一张对比表。

做采集这几年，我最大的体会是：不要把去重押在一种方案上。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 文件长这样：

requests==2.31.0 Pillow==10.1.0 imagehash==4.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_workers=8) 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_ok=True) 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/*,*/*;q=0.8", } def make_session() -> requests.Session: session = requests.Session() session.headers.update(HEADERS) adapter = HTTPAdapter(pool_connections=20, pool_maxsize=20, max_retries=3) 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), stream=True) 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_size=8192)) 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 个差值，最终得到 8x8=64 位哈希。

第二步，把每个像素转换成灰度值，对每一行的 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, mask=img.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_workers=8) 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, url=url, file_path=file_path, file_size=file_size, width=width, height=height, phash=str(img_hash), source="example", ) 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 最常见的原因是请求头不完整，尤其 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，但强烈不建议；更好的做法是在配置里设定最大宽高阈值，超过直接丢弃。我在采集壁纸站点时遇到过全景图，一张几十 MB，convert时内存直接飙到几百 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 入库之后不要只用来去重，后续做相似图检索、素材聚类、版权溯源都靠它。如果你也在做图片采集，建议从第一版就把“下载 → 感知哈希去重 → 元数据入库”的链路搭完整，把断点续传做好，后面会省下大量的重复劳动。