网页增量采集与变化监测：区分新增、修改和暂时抓取失败
用 SQLite 保存商品快照与观测轮次，输出新增和字段变化，并通过完整目录与重复观测确认删除，避免失败抓取清空历史数据。
昨天能抓到、今天没抓到，不一定是商品删除。目录发现不完整、详情超时、字段规则失效，都可能让本轮结果少一条。若直接用新结果覆盖旧库，就会把访问问题误写成业务变化。
本文接收采集器输出的目录与详情状态，用 SQLite 维护已知商品，区分新增、修改、暂时失败和待确认删除。它位于抓取之后，不重新介绍网页下载。
明确发现范围与详情状态
输入分成两层：expected_ids 表示本轮目录发现的商品，results 表示这些商品的详情处理结果。discovery_complete 只能在上游确实遍历完约定目录且检查通过时设置 true，不能由“没有异常退出”推断。
目录仍列出商品、详情却失败，表示本轮看见了实体但没取到新内容，旧记录应保留。目录没有它，且本轮目录完整，才进入缺失计数。本例两次完整观测缺失后标为 deleted，仍保存旧数据供回查，不物理删除。
准备输入与环境
使用 Python 3.12，只依赖标准库。创建 baseline.json：
{
"discovery_complete": true,
"expected_ids": ["p1", "p2"],
"results": [
{"id":"p1","status":"ok","name":"红茶","price_minor":100,"currency":"CNY"},
{"id":"p2","status":"ok","name":"绿茶","price_minor":200,"currency":"CNY"}
]
}再创建 update.json：
{
"discovery_complete": true,
"expected_ids": ["p1", "p2", "p3"],
"results": [
{"id":"p1","status":"ok","name":"红茶","price_minor":110,"currency":"CNY"},
{"id":"p2","status":"failed"},
{"id":"p3","status":"ok","name":"乌龙茶","price_minor":300,"currency":"CNY"}
]
}这是正文提供的教学快照，不是线上采集记录。p2 仍在目录里，因此它的详情失败不应触发删除。
完整比较与状态更新
创建 monitor.py：
import argparse
import hashlib
import json
import sqlite3
from pathlib import Path
parser = argparse.ArgumentParser()
parser.add_argument("snapshot")
parser.add_argument("--run-id", required=True)
args = parser.parse_args()
raw = Path(args.snapshot).read_text(encoding="utf-8")
batch = json.loads(raw)
input_hash = hashlib.sha256(raw.encode()).hexdigest()
expected = batch["expected_ids"]
if len(set(expected)) != len(expected) or any(not isinstance(x, str) or not x for x in expected):
raise ValueError("目录 ID 缺失或重复")
if not expected and not batch.get("empty_confirmed"):
raise ValueError("空目录需要明确确认，不能自动覆盖")
results = {}
for item in batch["results"]:
key = item["id"]
if key not in expected or key in results:
raise ValueError("详情 ID 不属于目录或重复")
if item["status"] not in {"ok", "failed"}:
raise ValueError("未知详情状态")
if item["status"] == "ok":
if not item.get("name") or type(item.get("price_minor")) is not int:
raise ValueError("详情结构不符合约定")
if item["price_minor"] < 0 or item.get("currency") != "CNY":
raise ValueError("金额或币种不符合约定")
results[key] = item
db = sqlite3.connect("monitor.sqlite")
db.executescript("""
CREATE TABLE IF NOT EXISTS items(
id TEXT PRIMARY KEY,payload TEXT NOT NULL,misses INTEGER NOT NULL,state TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS runs(
id TEXT PRIMARY KEY,input_hash TEXT NOT NULL,report TEXT NOT NULL
);
""")
try:
prior = db.execute("SELECT input_hash,report FROM runs WHERE id=?", (args.run_id,)).fetchone()
if prior:
if prior[0] != input_hash:
raise ValueError("同一 run-id 不能对应不同输入")
print(prior[1])
else:
old = {row[0]: row for row in db.execute("SELECT id,payload,misses,state FROM items")}
events = []
with db:
for key in expected:
item = results.get(key)
previous = old.get(key)
if item is None or item["status"] == "failed":
events.append({"id": key, "change": "fetch_failed"})
if previous:
db.execute("""UPDATE items SET misses=0,state=CASE
WHEN state IN ('deleted','restore_pending') THEN 'restore_pending'
ELSE 'active' END WHERE id=?""", (key,))
continue
payload = json.dumps({field: item[field] for field in
("name", "price_minor", "currency")}, ensure_ascii=False, sort_keys=True)
change = "added" if previous is None else (
"restored" if previous[3] in {"deleted", "restore_pending"} else
"modified" if previous[1] != payload else "unchanged"
)
events.append({"id": key, "change": change,
"before": json.loads(previous[1]) if previous else None,
"after": json.loads(payload)})
db.execute("""INSERT INTO items VALUES (?,?,0,'active')
ON CONFLICT(id) DO UPDATE SET payload=excluded.payload,misses=0,state='active'""",
(key, payload))
if batch["discovery_complete"] is True:
for key, previous in old.items():
if key not in expected:
misses = previous[2] + 1
state = "deleted" if misses >= 2 else "pending_delete"
db.execute("UPDATE items SET misses=?,state=? WHERE id=?", (misses, state, key))
events.append({"id": key, "change": (
state if previous[3] != state else "still_absent"
), "state": state})
report = json.dumps(events, ensure_ascii=False, indent=2)
db.execute("INSERT INTO runs VALUES (?,?,?)", (args.run_id, input_hash, report))
print(report)
finally:
db.close()运行两轮：
python monitor.py baseline.json --run-id baseline
python monitor.py update.json --run-id update-1预期第一轮新增 p1、p2，第二轮 p1 修改、p3 新增、p2 抓取失败但保留旧值。再次使用同一个 run-id 和同样输入，只返回已存报告，不增加缺失次数。
演示待确认删除
创建 missing.json，表示完整目录不再列出 p2：
{
"discovery_complete": true,
"expected_ids": ["p1", "p3"],
"results": [
{"id":"p1","status":"ok","name":"红茶","price_minor":110,"currency":"CNY"},
{"id":"p3","status":"ok","name":"乌龙茶","price_minor":300,"currency":"CNY"}
]
}按两个不同教学轮次处理：
python monitor.py missing.json --run-id missing-demo-1
python monitor.py missing.json --run-id missing-demo-2逻辑上先得到 pending_delete，再得到 deleted。这里重复材料用于演示状态变化，不能在真实任务里把同一份缓存响应换个 run-id 就当成两次独立观测。真正确认删除应结合来源更新频率、访问状态与业务风险决定等待间隔和证据。
deleted 仍是数据库状态，不是文件或历史数据物理删除。之后重新发现并成功读取 p2，会记录 restored。
已删除商品重新出现在目录、详情却失败时，状态改为 restore_pending，保留旧值并等待成功刷新；只有成功取得详情才报告 restored。连续缺失且已经处于同一删除状态时报告 still_absent，避免把同一次删除反复当作新变化。
为什么不对整个 HTML 做变化判断
导航时间、广告和推荐内容会造成原始 HTML 摘要变化，却不一定影响商品。本文只比较名称、价格和币种，原数据里抓取时间等字段不参与业务变化。
但这也意味着库存或规格变化不会被当前合同发现。需要监测哪些字段，就把它们明确放进规范化 payload，并解释缺失值与零值；不能宣称一个摘要监测了所有变化。
事务将商品状态与该轮报告一起保存，避免状态已经变更却找不到对应报告。参数化写入与事务行为见 Python SQLite 文档。
常见问题与修复
一次网络故障后大量商品 deleted。 检查上游是否错误把不完整目录标成 complete。先修复完整性判断，再根据历史轮次恢复状态，不只提高缺失阈值。
详情失败却被判下架。 检查 expected_ids 是否仍包含目录看到的商品。发现层与详情层不能用一个成功集合代替。
重复运行增加删除次数。 使用稳定的 run-id 管理一次观测，而不是每次处理程序启动都生成新观测身份。
所有记录每天都 modified。 检查 payload 是否混入时间戳、随机字段、格式不稳定的金额或无序集合，先统一字段含义再比较。
明明恢复了却仍显示 pending。 看是否仅目录恢复而详情还失败。本文保留旧值并重置缺失，来源新鲜度还需单独标注；不能把保留旧值当作已刷新。
完整目录确实为空。 必须由上游明确确认空状态后提供 empty_confirmed，而不是把空响应自动接受。确认依据需要保留，不由比较器猜测。
增量监测的价值，是把“业务变化”与“本轮没有取得数据”分开。状态和证据分清后，后续告警与补采才有可靠依据。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。