企业官网建设流程全解析

1. 为什么 item 落库前必须做去重

做 Scrapy 爬虫的朋友大概率都遇到过这个场景：分页翻得好好的，商品也一条条抓下来了，结果一查数据库，同一个 SKU 出现了七八次。原因不复杂——列表页翻页时商品顺序会变、详情页被多个入口链接指向、断点续爬时又从头跑了一遍。这些都会让同一个 item 反复进入 pipeline。

如果只在数据库层加唯一索引，重复 item 到达时insert会直接抛异常，日志里一片红，虽然数据没脏，但爬虫的吞吐和可观测性都被拖垮了。更麻烦的是，你没法知道到底重复了多少、重复来自哪个环节。

所以正确的做法是在 item 入库前加一道指纹校验：给每个 item 算一个稳定指纹，先查这个指纹在不在，不在才写库。指纹可以基于 SKU、也可以基于「标题+价格+店铺」这类业务主键组合。校验层用 Redis 做高速缓存，数据库唯一索引做最后兜底，两层配合，重复率就能被压到接近零，而且可观测、可复现。

这篇就围绕 Scrapy 的pipelines.py和settings.py，把指纹生成、Redis 去重、数据库唯一索引这套组合拳拆开讲清楚。同时我会把模型调用通道统一到 TaoToken 上，用一个 Key 管理指纹辅助判断、异常归因这类需要模型能力的环节，避免在爬虫项目里散落多个 Key。

2. TaoToken 前置：统一 Key 与 API 通道

爬虫项目里什么时候会用到模型能力？举几个真实场景：指纹字段的语义归一化（比如「iPhone 15 Pro 256G 深空黑」和「苹果15pro 256g 黑色」要判成同款）、异常 item 的归因分类、以及去重命中后的抽样复核。这些环节如果各自接一家 API，Key 管理会非常乱。

TaoToken 在这里的角色是统一入口：一个 Key 走通模型对话、编码辅助和后续的 Agent 调用。你只需要在项目根目录放一个.env，把 Key 和 base_url 配好，Scrapy 里通过os.environ读取即可。

先拿到 Key。访问控制台创建：

https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console

创建后在 API Keys 页面复制，注意它只显示一次：

https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api-keys

接口地址统一用https://taotoken.net/api，不要带 UTM 参数，SDK 里配置 base_url 时直接用这个。如果你后续要做指纹语义比对，可以先用模型对话页验证一下模型返回是否符合预期：

https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model-chat

长期跑爬虫 + Agent 复核任务的话，Coding Plan 的额度模型更适合持续调用：

https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan

接入文档在这里，SDK 参数、错误码、限流说明都齐全：

https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc

注意：Key 只放在.env里，.gitignore必须包含.env。爬虫项目经常被推到公开仓库，Key 泄露是高频事故。

3. 可复制配置：settings.py 与 pipelines.py 骨架

3.1 settings.py 关键项

先看配置。Redis 去重和 MySQL 连接池的参数都放这里，方便复用：

# settings.py import os from dotenv import load_dotenv load_dotenv() BOT_NAME = "dedup_spider" SPIDER_MODULES = ["dedup_spider.spiders"] NEWSPIDER_MODULE = "dedup_spider.spiders" ROBOTSTXT_OBEY = True CONCURRENT_REQUESTS = 16 DOWNLOAD_DELAY = 0.5 # TaoToken 统一通道 TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = "https://taotoken.net/api" # Redis 去重 REDIS_HOST = "127.0.0.1" REDIS_PORT = 6379 REDIS_DB = 0 REDIS_DUP_KEY = "scrapy:item:fingerprint" # MySQL MYSQL_HOST = "127.0.0.1" MYSQL_DBNAME = "spider_db" MYSQL_USER = "root" MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD") MYSQL_PORT = 3306 ITEM_PIPELINES = { "dedup_spider.pipelines.FingerprintPipeline": 300, "dedup_spider.pipelines.RedisDedupPipeline": 400, "dedup_spider.pipelines.MysqlTwistedPipeline": 500, }

优先级数字很关键：指纹生成必须在去重之前，去重必须在入库之前。300 → 400 → 500 这个顺序不能乱。

3.2 指纹生成 pipeline

指纹要稳定、可复现。我用hashlib.sha1对归一化后的业务主键做哈希，取前 16 位足够：

# pipelines.py import hashlib import re import redis from scrapy.exceptions import DropItem class FingerprintPipeline: """基于业务主键生成稳定指纹""" def process_item(self, item, spider): raw = "{}-{}-{}".format( self._norm(item.get("good_sku", "")), self._norm(item.get("good_title", "")), self._norm(item.get("shop_name", "")), ) item["fingerprint"] = hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16] return item @staticmethod def _norm(text): text = str(text or "").strip().lower() text = re.sub(r"\s+", "", text) return text

_norm做了三件事：去空格、转小写、压缩连续空白。这样「iPhone 15 Pro」和「iphone15pro」会得到同一个指纹。

3.3 Redis 去重 pipeline

Redis 用SET NX做原子判断，返回 1 表示首次出现，返回 0 表示重复：

class RedisDedupPipeline: def __init__(self, redis_host, redis_port, redis_db, dup_key): self.redis = redis.Redis( host=redis_host, port=redis_port, db=redis_db, decode_responses=True ) self.dup_key = dup_key @classmethod def from_crawler(cls, crawler): return cls( redis_host=crawler.settings.get("REDIS_HOST"), redis_port=crawler.settings.get("REDIS_PORT"), redis_db=crawler.settings.get("REDIS_DB"), dup_key=crawler.settings.get("REDIS_DUP_KEY"), ) def process_item(self, item, spider): fp = item.get("fingerprint") if not fp: raise DropItem("missing fingerprint: {}".format(item)) is_new = self.redis.set(self.dup_key + ":" + fp, "1", nx=True, ex=86400 * 7) if not is_new: spider.crawler.stats.inc_value("dedup/redis_hit") raise DropItem("duplicate fingerprint: {}".format(fp)) spider.crawler.stats.inc_value("dedup/redis_pass") return item

ex=86400 * 7是给指纹设 7 天过期，避免 Redis 无限膨胀。如果你的爬虫周期更长，把这个值调大。

3.4 MySQL 入库 pipeline

入库沿用异步连接池，同时把唯一索引冲突也接住：

from twisted.enterprise import adbapi from pymysql import cursors class MysqlTwistedPipeline: def __init__(self, dbpool): self.dbpool = dbpool @classmethod def from_settings(cls, settings): dbparms = dict( host=settings["MYSQL_HOST"], db=settings["MYSQL_DBNAME"], user=settings["MYSQL_USER"], passwd=settings["MYSQL_PASSWORD"], port=settings["MYSQL_PORT"], charset="utf8", cursorclass=cursors.DictCursor, use_unicode=True, ) dbpool = adbapi.ConnectionPool("pymysql", **dbparms) return cls(dbpool) def process_item(self, item, spider): query = self.dbpool.runInteraction(self.do_insert, item) query.addErrback(self.handle_error, item, spider) return item def do_insert(self, cursor, item): cursor.execute( """INSERT INTO yurongfu (website, brand, shop_name, shop_url, good_sku, good_title, good_price, good_detail, good_color, good_tips, fingerprint) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (item["website"], item["brand"], item["shop_name"], item["shop_url"], item["good_sku"], item["good_title"], item["good_price"], item["good_detail"], item["good_color"], item["good_tips"], item["fingerprint"]), ) def handle_error(self, failure, item, spider): spider.logger.error("insert failed: %s", failure) def close_spider(self, spider): self.dbpool.close()

数据库表上给fingerprint加唯一索引，作为最后一道防线：

ALTER TABLE yurongfu ADD UNIQUE KEY uk_fingerprint (fingerprint);

4. 验证请求与成功结果

4.1 验证 Redis 去重是否生效

跑一次爬虫，观察 stats 输出：

scrapy crawl goods -s LOG_LEVEL=INFO

正常日志里会看到：

'dedup/redis_pass': 1280, 'dedup/redis_hit': 342, 'item_scraped_count': 1280,

redis_hit就是被拦下的重复 item 数。重复率 = 342 / (1280 + 342) ≈ 21%，这个数字可观测、可复现。

4.2 验证数据库唯一索引兜底

手动往 Redis 里删一个指纹，再跑一次，模拟 Redis 失效的情况：

redis-cli -n 0 DEL scrapy:item:fingerprint:<某个指纹>

再跑爬虫，这条 item 会穿过 Redis，但到 MySQL 时被唯一索引拦下，handle_error会打印：

insert failed: (1062, "Duplicate entry 'a1b2c3d4e5f6a7b8' for key 'uk_fingerprint'")

这说明两层防护都在工作。

4.3 用 TaoToken 做指纹语义复核

有些重复不是字面重复，而是语义重复。比如「苹果15pro 256g 黑色」和「iPhone 15 Pro 256GB 深空黑」。这种可以抽样丢给模型判断：

import requests def semantic_same(a, b): resp = requests.post( "https://taotoken.net/api/v1/chat/completions", headers={"Authorization": "Bearer " + os.getenv("TAOTOKEN_API_KEY")}, json={ "model": "claude-sonnet-4-5", "messages": [{ "role": "user", "content": "判断这两个商品是否同款，只回答 yes 或 no：\nA: {}\nB: {}".format(a, b) }], "temperature": 0, }, timeout=15, ) return resp.json()["choices"][0]["message"]["content"].strip().lower() == "yes"

这个函数不要放在主 pipeline 里逐条调用，成本太高。正确做法是：Redis 去重命中后，把疑似重复的 item 对写入一个待复核队列，离线批量跑语义判断，确认是真重复就永久拉黑指纹，是误判就把指纹从 Redis 删掉放行。

5. 本篇常见错排查

5.1 DropItem 后 stats 不计数

如果你在RedisDedupPipeline里直接return None而不是raise DropItem，Scrapy 不会把它算作丢弃，item_dropped_count一直是 0。必须用raise DropItem(...)，stats 才会正确累加。

5.2 指纹不稳定导致误杀

_norm里如果用了str(item.get("good_price"))，价格从99.0变成99时指纹会变，同款商品被当成两条。价格字段建议先float()再格式化，或者干脆不参与指纹，只用 SKU + 标题 + 店铺。

5.3 Redis 连接在爬虫关闭时未释放

RedisDedupPipeline里如果用了连接池，记得在close_spider里self.redis.close()。否则长时间跑批会出现连接数堆积，Redis 报max number of clients reached。

5.4 MySQL 唯一索引冲突日志刷屏

handle_error里如果直接print(failure)，重复 item 多的时候日志会爆炸。建议按错误码过滤，1062 单独计数，其他错误才打完整堆栈：

def handle_error(self, failure, item, spider): if failure.check(Exception) and "1062" in str(failure.value): spider.crawler.stats.inc_value("dedup/mysql_conflict") return spider.logger.error("insert failed: %s", failure)

5.5 TaoToken 调用超时拖慢爬虫

语义复核如果同步放在 pipeline 里，模型响应慢会直接卡住整个爬虫。务必异步化：用scrapy.Request发到独立队列，或者用 Celery 离线处理。爬虫主链路只做 Redis + MySQL 两层，模型能力放在旁路。

6. 下一步：把去重链路接到 Coding Plan

指纹校验这套跑通之后，你会发现真正耗时的不是去重本身，而是异常 item 的归因和指纹规则的迭代。比如某个店铺的标题格式突然变了，指纹大面积误判，你需要快速定位是哪个字段的归一化出了问题。

这类任务适合交给 Agent 持续跑：定时拉取dedup/redis_hit的样本，用模型判断误判率，自动调整_norm规则。长期跑这种任务，用 Coding Plan 的额度模型比按次调用更划算：

https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan

接入细节和 SDK 参数在文档里：

https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc

我自己的做法是：爬虫主链路保持纯 Redis + MySQL，零模型依赖；旁路开一个 Agent 任务，每天凌晨跑一次指纹质量报告，把误判样本和修正建议推到飞书。这样去重率可观测，规则迭代也有据可依，不会拍脑袋改代码。