给爬虫增加数据质量检查：拦住空结果、错字段与异常记录
建立页面级与记录级质量检查，把商品数据分成合格、隔离与重复三类，并解释空结果、字段失效和部分完成时如何处理。
采集任务返回 200、导出文件也存在，仍可能没有可用数据：页面实际是登录提示，价格字段取到了运费，或者某个分类一夜之间从几百条降到零条。质量检查要回答的是“这批数据能否进入下游”，而不是“程序有没有报异常”。
本文面向已有采集程序、准备把商品数据接入分析或业务系统的团队。我们在下载解析之后、正式入库之前增加一道检查，产出合格记录、隔离记录和批次报告。示例同时处理空列表、访问异常、缺字段、错误类型和重复商品。
给采集器和下游约定输入
页面级证据和记录级字段要一起交付。采集器至少提供来源 URL、HTTP 状态、页面识别结果和提取记录。页面识别结果应由目标站的实际标记确定：正常商品列表、明确的空列表、登录页、访问挑战或未知页面。
不要把“找不到商品选择器”直接标成正常空列表。正常空列表应有明确的空结果提示，或者站点 API 返回可解释的空集合与结束信息。未知页面进入隔离，保留材料供排查。
下面用 page_kind 表达采集器已经判断的页面类型。这是接口合同，不是质量检查函数自动识别所有网站的能力。接入现有爬虫时，需要把识别逻辑放在解析阶段，页面不明时填写 unknown，不能默认填写 products。
一份可以解释每条去向的示例
使用 Python 3.12 或更新版本，无第三方依赖。创建 quality_gate.py，放入以下完整代码。正文提供的页面包是教学输入，没有访问外部站点。
import json
from urllib.parse import urlsplit
base = {"sku": "sku-1", "name": "红茶 500g", "price_minor": 1290,
"currency": "CNY", "source_url": "https://example.16yun.cn/p/1"}
pages = [
{"url": "https://example.16yun.cn/list/1", "status": 200,
"page_kind": "products", "records": [base, {**base, "sku": "sku-2", "name": ""}]},
{"url": "https://example.16yun.cn/list/2", "status": 200,
"page_kind": "products", "records": [
base, {**base, "sku": "sku-3", "price_minor": -1},
{**base, "sku": "sku-4", "price_minor": "12.90"}]},
{"url": "https://example.16yun.cn/list/3", "status": 200,
"page_kind": "challenge", "records": []},
{"url": "https://example.16yun.cn/list/4", "status": 200,
"page_kind": "products", "records": []},
]
def reasons_for(row):
reasons = []
for field in ("sku", "name"):
if not isinstance(row.get(field), str) or not row[field].strip():
reasons.append(f"missing_or_invalid_{field}")
amount = row.get("price_minor")
if type(amount) is not int or amount < 0:
reasons.append("invalid_price_minor")
if row.get("currency") not in {"CNY", "USD", "GBP"}:
reasons.append("unsupported_currency")
source = row.get("source_url")
if not isinstance(source, str):
reasons.append("missing_source")
elif urlsplit(source).scheme not in {"http", "https"} or not urlsplit(source).hostname:
reasons.append("invalid_source_url")
return reasons
accepted, rejected, duplicates, page_errors = [], [], [], []
seen = {}
for page in pages:
if page["status"] != 200 or page["page_kind"] not in {"products", "empty"}:
page_errors.append({"url": page["url"], "reason": "unexpected_page"})
continue
if page["page_kind"] == "empty" and page["records"]:
page_errors.append({"url": page["url"], "reason": "inconsistent_empty_page"})
continue
if page["page_kind"] == "products" and not page["records"]:
page_errors.append({"url": page["url"], "reason": "unexplained_empty_result"})
continue
for row in page["records"]:
reasons = reasons_for(row)
if reasons:
rejected.append({"page": page["url"], "record": row, "reasons": reasons})
continue
key = (row["sku"], row["currency"])
if key in seen:
duplicates.append({
"key": list(key), "record": row,
"conflict": row != seen[key],
})
continue
seen[key] = row
accepted.append(row)
report = {
"accepted": len(accepted), "rejected": len(rejected),
"duplicates": len(duplicates), "page_errors": page_errors,
"publishable": not rejected and not page_errors
and not any(item["conflict"] for item in duplicates),
}
for filename, data in [
("accepted.json", accepted), ("rejected.json", rejected),
("duplicates.json", duplicates), ("quality-report.json", report),
]:
with open(filename, "w", encoding="utf-8") as stream:
json.dump(data, stream, ensure_ascii=False, indent=2)
print(json.dumps(report, ensure_ascii=False, indent=2))执行：
python quality_gate.py本例应保留一条合格商品、隔离三条坏记录，并识别一条重复记录。第三、四个页面分别因为挑战页面和无法解释的空结果进入页面错误清单。因此 publishable 为 false，并不因为有一条好数据就允许整批覆盖旧数据。
三种检查各自负责什么
字段类型检查阻止明显的结构变化，例如本来是便士整数的价格突然变成字符串 "12.90"。代码使用 type(amount) is int，避免把 Python 中属于整数子类的布尔值当作合法金额。
业务检查确认值的含义：这里允许零价格，但拒绝负价格；支持的币种是当前任务范围，而不是所有金融系统的完整列表。如果你的业务不接受免费商品，增加零值规则前先明确业务约定。
批次检查发现单条记录无法表达的问题：某页未成功处理、列表无法解释地变空、同一业务键出现冲突。示例采用保守发布策略：存在这些问题就不把整批作为完整快照发布。面向实时部分数据的系统可以接收合格分区，但必须向下游标明部分完成，不能冒充完整目录。
重复记录为什么不直接覆盖
同一个 SKU 与币种可能重复出现在两个列表中，完全相同的重复可以去重。如果两条价格不同，应先确定任务是否还缺少地区、规格或观测时间维度。
本例保存重复记录并标明 conflict，不采用“最后一条覆盖”。在没有时间排序约定时，最后完成的网络请求不一定包含最新价格。要选择最新报价，必须根据明确的观测时间与来源优先级排序。
业务键也需要随任务调整。多地区采集至少加入市场维度，多规格采集通常使用 SKU 而不是商品父 ID。不要把示例中的两字段键搬到所有任务里。
接到现有爬虫的哪个位置
下载和解析完成后先形成页面包，交给质量检查，再将合格记录送入暂存区。页面失败、字段隔离与最终发布是不同状态，不应该由一个 success 布尔值承担。
接入时保留原始响应或脱敏页面快照的路径，并把路径放进页面错误和坏记录中。本例输入直接在正文里，故没有文件路径；真实采集只保留错误字符串，通常不足以重现页面改版导致的错位。
如果出现 challenge，先判断访问条件；如果是 unexplained_empty_result，检查列表选择器与数据是否延迟加载；如果是 invalid_price_minor，检查价格解析和单位。分别修复后只补采受影响页面，再对组合后的整批结果重新检查。
检查报告还不能发现什么
一个错误字段如果恰好类型正确、范围合理，仍会通过规则。例如运费 1290 被当作商品价，单靠非负整数校验发现不了。应选择若干人工核对的来源记录，比较字段语义，或者增加页面字段标记与来源位置等证据。
同样，报告中的 publishable 只依据本次输入，无法知道采集器是否压根没发现第二页。因此目录完整性还需要预期页面集合、分页结束证据或来源数量等上游信息。质量检查不能替代发现范围管理。
对历史基线做异常检测时，先记录分母：合格率是按发现记录、处理记录还是全部预期记录计算。昨日 1000 条、今日 800 条可能来自真实下架，也可能是一次漏页，不能只设一个百分比阈值就自动删除历史数据。
常见问题与处理
坏记录越来越多，但程序没有报错。 检查发布流程是否实际读取 publishable，是否有人绕过暂存区直接写正式表。规则存在却不参与发布，不能保护数据。
空列表被误判为失败。 为真实空结果增加明确的 empty 识别依据；代码中这类页面可带空 records 通过。不要让所有零条页面都自动转成 empty。
同一商品出现冲突。 比较规格、地区、币种和观测时间。补齐缺失维度后再定义键；不要按先来或后来任意选择价格。
修复字段后仍然被隔离。 确认补采结果进入的是新的检查批次，且输入没有混入旧的失败文件。检查报告应绑定批次 ID，避免把上一次的缺陷计数当成本次结果。
部分成功是否全部重跑。 按错误来源保留补采清单。小任务可以重跑；大任务应补齐缺失部分后统一评估完整性，不能只检查刚补的记录就宣布整批正常。
这道质量检查的价值，是让每条数据和每个页面都有明确去向。它应帮助你决定修复哪里、哪些结果能交付，以及哪些历史数据暂时不能覆盖。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。