Validar dados extraídos e detetar desvios no esquema

Os dados extraídos raramente apresentam falhas evidentes. A problema vai-se agravando: uma reformulação do design desloca um preço para um elemento diferente, o analisador devolve None e dez mil linhas com preços em branco são carregadas no armazém de dados sem uma única exceção. Este guia constrói as camadas que detetam a deriva do esquema antes que esta se traduza num relatório incorreto: um esquema de registos pydantic v2, verificações de qualidade por lote, um detetor de deriva com uma linha de base contínua, uma etapa de quarentena, um teste diário do contrato de seleção em URLs «canary» e o controlo de versões do analisador, para que possa preencher os dados retroativamente assim que corrigir o problema.

Por que é que os dados extraídos se deterioram silenciosamente

Três fontes de desvio de esquema são responsáveis por quase todos os incidentes. As reformulações do site alteram o DOM, e um analisador tolerante escrito com .get() e find() devolve valores vazios em vez de gerar erros. Os layouts A/B são ainda piores, porque apenas uma fração dos pedidos vê o novo modelo, pelo que a taxa de preenchimento cai de 99% para 70% e ninguém repara nisso durante uma semana. Alterações de localização e geográficas modificam os formatos sem alterar a estrutura: 1,299.00 torna-se 1.299,00, uma data muda do formato ISO para o formato com o dia primeiro e o valor é analisado incorretamente, em vez de não ser analisado de todo.

A característica comum é que o analisador não entra em falha. A tolerância ao nível do registo é a predefinição correta, mas isso significa que a exatidão tem de ser garantida noutro local. Esse local é a validação de dados ao nível do lote nos pipelines de web scraping, e é barato de implementar.

Definir um esquema de registo com o pydantic v2

Comece por definir como é um registo válido. Os campos obrigatórios, tipos, intervalos e formatos transformam-se num modelo pydantic, e cada dicionário analisado passa pelo model_validate antes de seguir para qualquer outro local.

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

Há aqui duas opções importantes. O mode="before" no validador de preço permite que o esquema realize a última etapa de normalização, para que o analisador possa permanecer simples. E o e.errors() devolve erros estruturados com uma tupla loc, o que permite contar as falhas por campo, em vez de por registo. Um lote em que 40 por cento das linhas falham no price e em mais nada é um problema diferente de um em que todas as linhas falham em todos os campos.

Verificações por lote

A validação rejeita linhas. As verificações por lote analisam, em conjunto, as linhas que foram aprovadas e as que não foram, e determinam se o lote, no seu conjunto, é plausível. Quatro verificações detetam a maior parte dos desvios do esquema.

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

Mantenha o batch_metrics em cada execução. É a matéria-prima do detetor de desvios e é também o que irá analisar em primeiro lugar quando alguém perguntar por que razão os números de terça-feira parecem estranhos.

Um detetor de desvio com uma linha de base móvel

Os limiares fixos falham porque as fontes diferem. Um campo preenchido a 60 por cento num local é normal; noutro, é uma crise. Em vez disso, compare com o próprio histórico recente da fonte: uma linha de base móvel dos últimos N lotes válidos.

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)

A linha importante é a accept: apenas os lotes sem anomalias alimentam a linha de base. Se deixar entrar um lote com desvio, a linha de base segue o desvio e o detetor fica silencioso precisamente quando deveria estar a alertar. Utilize a mediana em vez da média pela mesma razão; um dia atípico não deve alterar a referência.

Quarentena em vez de carregamento

Quando o detetor disparar, não carregue o lote, nem o descarte. Guarde-o num local de quarentena com as suas métricas e alertas anexados, envie o alerta e deixe que um utilizador ou uma regra automatizada decida. A decisão é normalmente uma das três seguintes: falso alarme (promova o lote e aceite-o na linha de base), desvio real (corrija o analisador, volte a executar a partir do conteúdo bruto armazenado) ou interrupção na fonte (descarte o lote, mantenha a linha de base inalterada).

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

Alerte sobre o que mudou, não sobre «falha na validação». Uma mensagem que indique price fill rate 0.98 -> 0.61 on source X é passível de ação; uma falha genérica não o é.

O teste do contrato de seletor em URLs canary

As verificações em lote só indicam que algo se desviou depois de o desvio ter ocorrido. Um teste do contrato de seletor avisa-o mais cedo, muitas vezes antes da execução diária. Escolha um conjunto fixo de URLs «canary» por fonte, cujos valores esperados conheça, recupere-os através do POST /v1/extract todos os dias, execute o analisador e verifique se cada campo previsto no contrato está presente e é plausível. Quando o contrato falha, o analisador precisa de atenção antes do início da execução principal.

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)

Cinco a dez canários por fonte são suficientes. Defina o país do proxy se o site for localizado, pois um canário obtido de uma região diferente testa uma página diferente. Escolha URLs que testem os layouts que lhe interessam: um com stock, um sem stock, um em promoção e um sem classificação.

Versionar o analisador e guardar a página em bruto

Cada linha deve conter parser_version, e a versão deve ser alterada sempre que a lógica do analisador mudar. Isso proporciona-lhe duas vantagens. Em primeiro lugar, pode segmentar qualquer métrica por versão e verificar imediatamente se um alerta de desvio coincide com uma implementação. Em segundo lugar, quando corrigir um analisador, saberá exatamente quais as linhas a regenerar.

A regeneração só é possível se tiver guardado os dados de entrada. Armazene o HTML bruto ou o Markdown de cada recolha, comprimido e indexado por source_url e scraped_at. O custo de armazenamento é reduzido em comparação com o custo de uma nova recolha, e muitas vezes a recolha não é sequer uma opção, porque a página já voltou a mudar. Quando a alteração do esquema for confirmada, o fluxo de trabalho é o seguinte: corrigir o analisador, atualizar a versão, voltar a analisar o conteúdo bruto armazenado para o período afetado, validá-lo através do mesmo esquema e carregá-lo para substituir os lotes em quarentena. Sem o armazenamento do conteúdo bruto, a lacuna é permanente.

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

Erros comuns

Validar por registo e nunca por lote

Um esquema de registo deteta linhas malformadas; não consegue detetar um lote em que todas as linhas estejam bem formadas, mas em que metade dos preços falte. Calcule sempre as taxas de preenchimento e as contagens ao nível do lote.

Permitir que lotes com erros entrem na linha de base

Se a linha de base móvel incluir um lote com desvios, o próximo lote com desvios parecerá normal. Aceite apenas lotes que tenham sido aprovados no histórico e utilize a mediana.

Tornar o analisador demasiado tolerante

.get("price") que devolve None está correto, mas um analisador que devolve 0.0 ou uma cadeia de caracteres vazia em caso de falha anula a verificação da taxa de preenchimento. Devolva None para dados em falta e deixe que o esquema o rejeite.

Não fixar as condições de recuperação

Os testes «canary» e as execuções em produção devem utilizar as mesmas definições de renderização e de proxy. Um teste de contrato que seja aprovado com render_js: true e um proxy dos EUA não prova nada sobre uma execução que não utilize nenhum dos dois.

Próximo passo

Todas as recuperações descritas neste guia passam pelo Geonode Scraper API, que devolve Markdown ou HTML limpo a partir de uma única chamada ao POST /v1/extract com renderização em JavaScript e direcionamento de proxy ao nível do país, quando necessário. Manter a camada de recuperação estável elimina uma variável das investigações sobre desvios de esquema, porque o que mudou foi a página, e não a forma como a obteve. O plano gratuito cobre 1,5 mil pedidos por mês, o suficiente para executar contratos «canary» em várias fontes diariamente, enquanto constrói o pipeline de validação.

Perguntas frequentes

O que é o desvio de esquema na extração de dados da Web?

O desvio de esquema é a alteração gradual ou repentina na estrutura dos dados que o seu analisador produz, causada por alterações no site de origem e não no seu código. Uma reformulação move um preço para um novo elemento, um teste A/B apresenta um layout diferente para alguns pedidos ou uma mudança de localização altera o formato numérico. O analisador continua a funcionar, mas os campos ficam vazios ou incorretos.

Por que razão os dados extraídos deterioram-se silenciosamente em vez de apresentarem erros evidentes?

A maioria dos analisadores é programada para ser tolerante, pelo que um elemento em falta devolve «None» em vez de gerar uma exceção. Isso faz sentido a nível de registo, mas é desastroso a nível de lote, porque dez mil registos com um preço nulo são carregados sem erro. Só o deteta quando um relatório a jusante parece errado, muitas vezes semanas mais tarde. A validação ao nível do lote e as verificações de desvio colmatam essa lacuna.

Como validar dados extraídos com o pydantic?

Defina um modelo por tipo de registo com campos obrigatórios, tipos, intervalos numéricos e restrições de expressões regulares; em seguida, chame model_validate em cada dicionário analisado dentro de um bloco try que capture ValidationError. Conte as falhas por campo em todo o lote. Uma elevada taxa de rejeição num campo é um sinal de desvio do esquema; uma taxa elevada em todos os campos significa, normalmente, que a própria recolha falhou.

Que limiar deve desencadear um alerta de desvio do esquema?

Compare a taxa de preenchimento de cada campo no lote atual com uma linha de base móvel de lotes recentes válidos. Uma queda de dez pontos percentuais ou mais num campo obrigatório é um sinal forte; utilize cinco para campos de alto valor, como o preço. Verifique também a contagem de registos em relação à expectativa, dentro de uma margem de mais ou menos 20 por cento, e observe a taxa de duplicados. Ajuste os limiares por fonte após uma semana de dados.

Devo armazenar HTML bruto ou Markdown de cada extração?

Sim, se o espaço de armazenamento o permitir. O HTML bruto ou o Markdown são a única forma de permitir o preenchimento retroativo após corrigir um analisador, porque, de outra forma, a página tal como estava no momento da extração já não existe. O HTML comprimido ocupa pouco espaço em comparação com o custo de repetir a extração e também fornece as provas necessárias para diagnosticar a razão pela qual um campo sofreu uma deriva.

A validação pode garantir que os meus dados extraídos estão corretos?

Não. A validação e a deteção de desvios detetam alterações na forma e na distribuição; não conseguem verificar se um valor analisado corresponde ao que um ser humano veria na página. Um campo pode permanecer preenchido mesmo que o seu significado mude. Combine verificações automatizadas com amostragens manuais periódicas e lembre-se de que a recolha de dados deve respeitar os termos de cada site e a legislação aplicável, independentemente da qualidade dos dados.