验证抓取的数据并检测模式漂移

抓取的数据很少会明显出错。它会逐渐变质：页面重新设计导致价格被移至另一个元素，解析器返回 None，结果一万行价格为空的记录被加载到数据仓库中，却未触发任何异常。 本指南构建了一系列机制，可在模式漂移导致报告出错之前将其拦截：pydantic v2 记录模式、按批次进行的质量检查、采用滚动基准线的漂移检测器、隔离步骤、在金丝雀 URL 上进行的每日选择器契约测试，以及解析器版本控制——这样在修复问题后，您就可以进行数据补全。

为什么提取的数据会悄无声息地变质

几乎所有此类事件都源于三种模式漂移的原因。网站重新设计会改变 DOM，而使用 .get() 和 find() 编写的容错解析器会返回空结果，而不是抛出异常。 A/B 布局的情况更糟，因为只有一小部分请求会看到新模板，因此填充率会从 99% 降至 70%，而这一情况可能长达一周都无人察觉。 区域和地理位置的变更会改变格式而不改变结构：1,299.00 变为 1.299,00，日期格式从 ISO 格式切换为日优先格式，此时数据会被错误解析，而非完全无法解析。

这些情况的共同点在于解析器不会崩溃。记录级容错是正确的默认设置，但这意味着必须在其他地方强制保证正确性。这个“其他地方”就是网络爬虫管道中的批处理级数据验证，而实现起来成本很低。

使用 pydantic v2 定义记录模式

首先，明确有效记录应具备的结构。必填字段、类型、取值范围和格式将构成一个 pydantic 模型，每个解析后的字典在被传递到其他地方之前，都会先经过 model_validate 的处理。

from __future__ import annotations

import re
from datetime import date, datetime
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, field_validator

PRICE_RE = re.compile(r"^\d+(\.\d{1,2})?$")
SKU_RE = re.compile(r"^[A-Z0-9-]{4,32}$")

class ProductRecord(BaseModel):
    source_url: str = Field(min_length=10)
    sku: str
    title: str = Field(min_length=3, max_length=300)
    price: float = Field(gt=0, lt=100_000)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    in_stock: bool
    rating: Optional[float] = Field(default=None, ge=0, le=5)
    scraped_at: datetime
    listed_on: Optional[date] = None
    parser_version: str

    @field_validator("sku")
    @classmethod
    def sku_shape(cls, v: str) -> str:
        v = v.strip().upper()
        if not SKU_RE.match(v):
            raise ValueError(f"sku does not match pattern: {v!r}")
        return v

    @field_validator("price", mode="before")
    @classmethod
    def normalise_price(cls, v):
        # Accept "1,299.00" or "1299" from the parser; reject anything else.
        if isinstance(v, str):
            cleaned = v.replace(",", "").replace("$", "").strip()
            if not PRICE_RE.match(cleaned):
                raise ValueError(f"price not parseable: {v!r}")
            return float(cleaned)
        return v

    @field_validator("title")
    @classmethod
    def title_not_placeholder(cls, v: str) -> str:
        if v.strip().lower() in {"", "n/a", "untitled", "loading..."}:
            raise ValueError("title looks like a placeholder")
        return v.strip()

def validate_batch(rows: list[dict]) -> tuple[list[ProductRecord], list[dict]]:
    good: list[ProductRecord] = []
    bad: list[dict] = []
    for row in rows:
        try:
            good.append(ProductRecord.model_validate(row))
        except ValidationError as e:
            bad.append({"row": row, "errors": e.errors(include_url=False)})
    return good, bad

这里有两个关键选择。价格验证器上的 mode="before" 允许模式完成最后的规范化处理，从而使解析器保持简单。 而 e.errors() 会通过 loc 元组返回结构化错误，这使得您可以按字段而非按记录统计失败次数。 一个批次中，如果 40% 的行在 price 验证时失败而其他字段均通过，这与所有行在所有字段上均失败的情况是截然不同的问题。

按批次检查

验证会拒绝不符合条件的行。批次检查会综合考虑通过验证的行和未通过验证的行，判断整个批次是否合理。以下四项检查可捕获大部分模式漂移。

from collections import Counter
from statistics import median

def batch_metrics(good: list[ProductRecord], bad: list[dict], fields: list[str]) -> dict:
    total = len(good) + len(bad)
    failures: Counter[str] = Counter()
    for b in bad:
        for err in b["errors"]:
            if err["loc"]:
                failures[str(err["loc"][0])] += 1

    fill_rate = {}
    for f in fields:
        present = sum(1 for g in good if getattr(g, f, None) is not None)
        fill_rate[f] = present / total if total else 0.0

    skus = [g.sku for g in good]
    dup_rate = 1 - (len(set(skus)) / len(skus)) if skus else 0.0
    prices = [g.price for g in good]

    return {
        "record_count": total,
        "reject_rate": len(bad) / total if total else 0.0,
        "field_failures": dict(failures),
        "fill_rate": fill_rate,
        "duplicate_rate": dup_rate,
        "price_median": median(prices) if prices else None,
    }

def count_in_range(count: int, expected: int, tolerance: float = 0.20) -> bool:
    return abs(count - expected) <= expected * tolerance

每次运行后都要保留 batch_metrics。它是漂移检测器的原始数据，也是当有人询问“为什么周二的数据看起来很奇怪”时，您首先需要查看的内容。

采用滚动基线的漂移检测器

固定阈值会失效，因为不同来源的情况各不相同。在某个站点，填充率达到60%属于正常情况；但在另一个站点，这却可能意味着危机。因此，应将其与来源自身的近期历史数据进行比较：即基于最近N个合格批次的滚动基线。

from dataclasses import dataclass, field

@dataclass
class DriftAlert:
    field_name: str
    baseline: float
    current: float
    drop_points: float

@dataclass
class DriftDetector:
    window: int = 14
    drop_threshold: float = 10.0  # percentage points
    critical_fields: dict[str, float] = field(default_factory=lambda: {"price": 5.0})
    history: list[dict] = field(default_factory=list)

    def baseline_fill(self, field_name: str) -> float | None:
        rates = [m["fill_rate"].get(field_name) for m in self.history[-self.window:]]
        rates = [r for r in rates if r is not None]
        return median(rates) if rates else None

    def check(self, metrics: dict, expected_count: int) -> list[DriftAlert]:
        alerts: list[DriftAlert] = []
        for name, current in metrics["fill_rate"].items():
            base = self.baseline_fill(name)
            if base is None:
                continue
            drop = (base - current) * 100
            limit = self.critical_fields.get(name, self.drop_threshold)
            if drop >= limit:
                alerts.append(DriftAlert(name, base, current, drop))
        if not count_in_range(metrics["record_count"], expected_count):
            alerts.append(DriftAlert("record_count", expected_count, metrics["record_count"], 0.0))
        if metrics["duplicate_rate"] > 0.02:
            alerts.append(DriftAlert("duplicate_rate", 0.0, metrics["duplicate_rate"], 0.0))
        return alerts

    def accept(self, metrics: dict) -> None:
        # Only batches that passed become part of the baseline.
        self.history.append(metrics)

关键点在于 accept：只有合格批次才会纳入基线。如果让一个发生漂移的批次进入，基线就会随其漂移，导致检测器在理应发出警报时却保持静默。出于同样的原因，应使用中位数而非平均值；单个异常日不应改变参考基准。

隔离而非加载

当检测器触发时，不要加载该批次，也不要将其丢弃。将其连同相关指标和警报一起写入隔离位置，发送警报，并由人工或自动化规则进行裁决。 决策通常有三种：误报（将批次提升并纳入基线）、真实漂移（修复解析器，从存储的原始内容重新运行）或源数据中断（丢弃该批次，保持基线不变）。

import json
from pathlib import Path

def route_batch(batch_id: str, good, bad, metrics, alerts, detector: DriftDetector) -> str:
    if alerts:
        q = Path("quarantine") / batch_id
        q.mkdir(parents=True, exist_ok=True)
        (q / "metrics.json").write_text(json.dumps(metrics, default=str, indent=2))
        (q / "alerts.json").write_text(json.dumps([a.__dict__ for a in alerts], indent=2))
        (q / "rows.jsonl").write_text(
            "\n".join(g.model_dump_json() for g in good)
        )
        notify(f"batch {batch_id} quarantined: " + ", ".join(a.field_name for a in alerts))
        return "quarantined"
    load_to_warehouse(good)
    detector.accept(metrics)
    return "loaded"

def notify(message: str) -> None:
    # Wire to Slack, PagerDuty, email, or a log sink.
    print("ALERT:", message)

def load_to_warehouse(records) -> None:
    pass

应针对“发生了什么变化”发出警报，而非“验证失败”。一条显示为 price fill rate 0.98 -> 0.61 on source X 的消息具有可操作性；而泛泛的失败提示则没有。

针对“金丝雀”URL的选择器合约测试

批量检查只能在数据发生偏移后才告诉你发生了偏移。而选择器合约测试能更早地发现问题，通常在每日例行运行之前就能发现。 为每个数据源选择一组固定的“金丝雀”URL（即您已知其预期值的URL），每天通过 POST /v1/extract 获取这些URL，运行解析器，并验证每个合同字段是否存在且合理。当合同验证失败时，需在大型运行开始前对解析器进行排查。

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

CANARIES = [
    {"url": "https://example.com/product/123", "expect": {"sku": "EX-123", "currency": "USD"}},
    {"url": "https://example.com/product/456", "expect": {"sku": "EX-456", "currency": "USD"}},
]

CONTRACT_FIELDS = ["sku", "title", "price", "currency", "in_stock"]

def fetch_markdown(url: str) -> str:
    r = httpx.post(
        f"{BASE}/v1/extract",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "render_js": True,
            "processing_mode": "sync",
            "proxy": {"country": "US", "type": "residential"},
        },
        timeout=90.0,
    )
    r.raise_for_status()
    return r.json()["data"]["markdown"]

def parse_product(markdown: str) -> dict:
    # Your parser lives here; it should return a plain dict for the schema.
    raise NotImplementedError

def run_contract(parser_version: str) -> list[str]:
    failures: list[str] = []
    for c in CANARIES:
        md = fetch_markdown(c["url"])
        row = parse_product(md) | {"source_url": c["url"], "parser_version": parser_version}
        try:
            rec = ProductRecord.model_validate(row)
        except ValidationError as e:
            failures.append(f"{c['url']}: {[x['loc'] for x in e.errors()]}")
            continue
        for k, v in c["expect"].items():
            if getattr(rec, k) != v:
                failures.append(f"{c['url']}: {k} expected {v!r} got {getattr(rec, k)!r}")
    return failures

if __name__ == "__main__":
    problems = run_contract(parser_version="2026.10.1")
    if problems:
        notify("selector contract failed:\n" + "\n".join(problems))
        raise SystemExit(1)

每个数据源设置 5 到 10 个“金丝雀”URL 即可。如果网站支持本地化，请固定代理国家/地区，因为从不同地区获取的“金丝雀”URL 测试的页面可能不同。选择能够覆盖您关注布局的 URL：一个有库存的、一个缺货的、一个打折的、一个没有评分的。

为解析器添加版本号并存储原始页面

每一行都应包含 parser_version，且每当解析器逻辑发生变化时，版本号也应随之更新。这样能带来两点好处。首先，你可以按版本对任何指标进行切片分析，并立即判断漂移警报是否与部署时间吻合。 其次，当修复解析器时，您能准确知道需要重新生成哪些数据行。

只有保留了原始输入数据，才可能进行重新生成。 请存储每次抓取生成的原始 HTML 或 Markdown 内容，进行压缩，并以 source_url 和 scraped_at 作为键值进行存储。与重新抓取的成本相比，存储成本微乎其微；而且由于页面往往已再次发生变化，重新抓取通常甚至不是一个可行的选项。 当确认模式漂移后，工作流如下：修复解析器、提升版本号、对受影响时间窗口内的存储原始内容进行重新解析、通过同一模式进行验证，并加载这些内容以替换被隔离的批次。如果没有原始数据存储，数据缺口将永久存在。

import gzip
import hashlib
from pathlib import Path

def store_raw(source_url: str, scraped_at: datetime, content: str, fmt: str = "md") -> Path:
    key = hashlib.sha256(source_url.encode()).hexdigest()[:16]
    p = Path("raw") / scraped_at.strftime("%Y/%m/%d") / f"{key}.{fmt}.gz"
    p.parent.mkdir(parents=True, exist_ok=True)
    with gzip.open(p, "wt", encoding="utf-8") as f:
        f.write(content)
    return p

常见错误

仅对单条记录进行验证，而从未对整批数据进行验证

记录模式可以捕获格式错误的行；但它无法捕获这样一批数据：其中每行格式都正确，却有一半的价格数据缺失。应始终在批次层面计算填充率和计数。

允许异常批次进入基线

如果滚动基线中包含一个偏离的批次，那么下一个偏离的批次看起来就会正常。仅接受通过验证的批次并将其纳入历史数据，并使用中位数。

解析器容错性过高

返回 .get("price") 并生成 None 的情况尚可接受，但如果解析器在失败时返回 0.0 或空字符串，就会使填充率检查失效。对于缺失的数据，应返回 None，并由模式进行拒绝。

未固定数据获取条件

金丝雀测试和生产环境运行必须使用相同的渲染和代理设置。如果某项契约测试在 render_js: true 和美国代理的条件下通过，但这并不能证明在既不使用 render_js: true 也不使用美国代理的运行环境中同样有效。

下一步

本指南中所有数据获取操作均通过 Geonode Scraper API 进行，该服务通过单次对 POST /v1/extract 的调用，结合 JavaScript 渲染及按需启用的国家/地区级代理定位，返回经过处理的 Markdown 或 HTML 内容。 保持数据获取层的稳定性，可以消除架构漂移调查中的一个变量，因为发生变化的只是页面本身，而非获取方式。免费套餐每月提供 1.5K 次请求，足以在构建验证管道的同时，每天针对多个数据源运行金丝雀测试合约。

常见问题

网页抓取中的“模式漂移”是什么？

模式漂移是指解析器生成的数据结构发生渐进或突发的变化，这种变化是由源网站的变化而非您的代码引起的。 例如，网站改版将价格移至新元素中；A/B 测试导致部分请求的页面布局发生变化；或者语言环境切换导致数字格式改变。解析器虽然仍在运行，但字段内容可能变为空值或出现错误。

为什么抓取的数据会“悄无声息地变质”，而不是“高调报错”？

大多数解析器在设计上都具有容错性，因此当元素缺失时，会返回 None 而不是抛出异常。从单条记录来看这很合理，但从整批数据来看却灾难性——因为一万条价格为空的记录会无错误地加载。你往往要等到数周后，当下游报告看起来有问题时，才会发现这一问题。批处理级别的验证和漂移检查可以弥补这一漏洞。

如何使用 pydantic 验证抓取的数据？

为每种记录类型定义一个模型，其中包含必填字段、数据类型、数值范围和正则表达式约束；然后在 try 代码块中对每个已解析的字典调用 model_validate 方法，该代码块需捕获 ValidationError 异常。统计整个批次中各字段的失败次数。 某个字段的高拒收率是模式漂移的信号；所有字段均出现高拒收率通常意味着数据抓取本身失败。

触发模式漂移警报的阈值应设定为多少？

将当前批次中每个字段的填充率与近期有效批次的滚动基准值进行比较。 必填字段的填充率下降 10 个百分点或以上即为强信号；对于价格等高价值字段，阈值应设为 5 个百分点。同时检查记录数是否在预期值的正负 20% 范围内，并关注重复率。在收集一周数据后，根据数据源分别调整阈值。

是否应存储每次抓取生成的原始 HTML 或 Markdown 文件？

如果存储空间允许，建议存储。原始 HTML 或 Markdown 是修复解析器后进行数据补全的唯一依据，因为抓取时页面原貌一旦丢失便无法恢复。 与重新抓取的成本相比，压缩后的 HTML 文件体积很小，而且它还能为您提供诊断字段偏移原因所需的依据。

验证能否保证我的抓取数据是正确的？

不能。验证和漂移检测只能捕捉结构和分布的变化；它们无法验证解析出的值是否与人类在页面上所见的内容一致。某个字段可能仍被填充，但其含义却已发生变化。请将自动化检查与定期人工抽样相结合，并请记住，无论数据质量如何，抓取操作都必须遵守各网站的条款及适用法律。