Validar los datos extraídos y detectar desviaciones en el esquema

Los datos extraídos rara vez fallan de forma evidente. Se deterioran: un rediseño desplaza un precio a un elemento diferente, el analizador devuelve None y se cargan en el almacén de datos diez mil filas con precios vacíos sin que se produzca ni una sola excepción. Esta guía establece las capas que detectan la deriva del esquema antes de que se traduzca en un informe erróneo: un esquema de registros pydantic v2, comprobaciones de calidad por lotes, un detector de deriva con una línea de base móvil, un paso de cuarentena, una prueba diaria del contrato de selección en URL «canario» y el control de versiones del analizador, para que puedas rellenar los datos una vez solucionado el problema.

Por qué los datos extraídos se deterioran sin que nos demos cuenta

Hay tres causas principales de la deriva del esquema que explican casi todos los incidentes. Los rediseños de las páginas web modifican el DOM, y un analizador sintáctico tolerante escrito con .get() y find() devuelve valores vacíos en lugar de generar un error. Los diseños A/B son aún peores, ya que solo una fracción de las solicitudes ve la nueva plantilla, por lo que la tasa de llenado cae del 99 % al 70 % y nadie se da cuenta hasta pasada una semana. Los cambios de configuración regional y geográfica alteran los formatos sin modificar la estructura: 1,299.00 se convierte en 1.299,00, una fecha pasa del formato ISO al formato con el día primero, y el valor se analiza de forma errónea en lugar de no analizarse en absoluto.

La característica común es que el analizador no se bloquea. La tolerancia a nivel de registro es el valor predeterminado adecuado, pero implica que la corrección debe garantizarse en otro lugar. Ese lugar es la validación de datos a nivel de lote en los procesos de extracción de datos web, y su implementación resulta económica.

Definir un esquema de registros con pydantic v2

Empieza por describir cómo es un registro válido. Los campos obligatorios, los tipos, los rangos y los formatos se convierten en un modelo pydantic, y cada diccionario analizado pasa por model_validate antes de ir a cualquier otro sitio.

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

Aquí hay dos decisiones importantes. mode="before" en el validador de precios permite que el esquema realice la última parte de la normalización, de modo que el analizador pueda seguir siendo sencillo. Y e.errors() devuelve errores estructurados con una tupla loc, lo que permite contar los errores por campo en lugar de por registro. Un lote en el que el 40 % de las filas fallan en price y en ningún otro campo es un problema distinto al de uno en el que todas las filas fallan en todos los campos.

Comprobaciones por lote

La validación rechaza las filas. Las comprobaciones por lote analizan conjuntamente las filas que han superado la validación y las que no, y determinan si el lote en su conjunto es plausible. Cuatro comprobaciones detectan la mayor parte de las desviaciones del 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

Conserva batch_metrics en cada ejecución. Es la materia prima para el detector de desviaciones y también es lo primero que consultarás cuando alguien pregunte por qué las cifras del martes parecen extrañas.

Un detector de desviaciones con una línea de referencia móvil

Los umbrales fijos fallan porque las fuentes varían. Un campo que está lleno al 60 % en un sitio es normal; en otro, supone una crisis. En su lugar, compáralo con el historial reciente de la propia fuente: una línea de referencia móvil basada en los últimos N lotes correctos.

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)

La línea importante es accept: solo los lotes correctos alimentan la línea de referencia. Si se deja pasar un lote con desviación, la línea de referencia sigue la desviación y el detector se queda en silencio justo cuando debería dar la alarma. Utiliza la mediana en lugar de la media por la misma razón; un día atípico no debería alterar la referencia.

Cuarentena en lugar de carga

Cuando se active el detector, no cargues el lote, pero tampoco lo descartes. Guárdalo en una ubicación de cuarentena junto con sus métricas y alertas, envía la alerta y deja que sea una persona o una regla automatizada quien decida. La decisión suele ser una de estas tres: falsa alarma (promover el lote y aceptarlo en la línea de base), desviación real (corregir el analizador sintáctico y volver a ejecutarlo a partir del contenido sin procesar almacenado) o interrupción de la fuente (descartar el lote y mantener la línea de base sin cambios).

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

Alerta sobre lo que ha cambiado, no sobre «validación fallida». Un mensaje que indique price fill rate 0.98 -> 0.61 on source X permite actuar; un fallo genérico, no.

La prueba del contrato de selector en las URL de canario

Las comprobaciones por lotes te indican que algo se ha desviado una vez que ya se ha desviado. Una prueba del contrato de selector te lo indica antes, a menudo antes de la ejecución diaria. Elige un conjunto fijo de URL «canario» por fuente, aquellas cuyos valores esperados conozcas, recógelas a través de POST /v1/extract cada día, ejecuta el analizador y comprueba que cada campo del contrato esté presente y sea plausible. Cuando el contrato falla, hay que revisar el analizador antes de que comience la ejecución 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)

Bastan entre cinco y diez «canarios» por fuente. Fija el país del proxy si el sitio está localizado, ya que un «canario» recuperado desde una región diferente comprueba una página diferente. Elige URL que pongan a prueba los diseños que te interesan: una con existencias, otra sin existencias, otra en oferta y otra sin valoración.

Asignar una versión al analizador y almacenar la página sin procesar

Cada fila debe contener parser_version, y la versión debe cambiar cada vez que cambie la lógica del analizador. Esto te ofrece dos ventajas. En primer lugar, puedes desglosar cualquier métrica por versión y ver inmediatamente si una alerta de desviación coincide con una implementación. En segundo lugar, cuando corrijas el analizador, sabrás exactamente qué filas debes regenerar.

La regeneración solo es posible si has conservado los datos de entrada. Almacena el HTML sin procesar o el Markdown de cada recuperación, comprimido e indexado por source_url y scraped_at. El coste de almacenamiento es mínimo en comparación con el coste de volver a extraer los datos, y a menudo volver a extraerlos ni siquiera es una opción porque la página ha vuelto a cambiar. Cuando se confirma la desviación del esquema, el flujo de trabajo es el siguiente: corregir el analizador, actualizar la versión, volver a analizar el contenido sin procesar almacenado correspondiente al periodo afectado, validarlo mediante el mismo esquema y cargarlo para sustituir los lotes en cuarentena. Sin el almacenamiento del contenido sin procesar, la brecha es 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

Errores comunes

Validar por registro y nunca por lote

Un esquema de registros detecta las filas con formato incorrecto; no puede detectar un lote en el que todas las filas tengan un formato correcto pero falte la mitad de los precios. Calcula siempre las tasas de cobertura y los recuentos a nivel de lote.

Permitir que lotes defectuosos entren en la línea de base

Si la línea de base móvil incluye un lote con desviación, el siguiente lote con desviación parecerá normal. Acepta únicamente los lotes que hayan pasado a formar parte del historial y utiliza la mediana.

Hacer que el analizador sea demasiado tolerante

Que .get("price") devuelva None está bien, pero un analizador que devuelva 0.0 o una cadena vacía en caso de fallo anula la comprobación de la tasa de llenado. Devuelve None en caso de falta de datos y deja que el esquema lo rechace.

No fijar las condiciones de recuperación

Las pruebas canario y las ejecuciones en producción deben utilizar la misma configuración de renderizado y de proxy. Una prueba de contrato que se supere con render_js: true y un proxy de EE. UU. no demuestra nada sobre una ejecución que no utilice ninguno de los dos.

Siguiente paso

Todas las consultas de esta guía pasan por el Geonode Scraper API, que devuelve código Markdown o HTML limpio a partir de una única llamada a POST /v1/extract con renderizado en JavaScript y selección de proxy a nivel de país cuando sea necesario. Mantener estable la capa de recuperación elimina una variable de las investigaciones sobre desviaciones del esquema, ya que lo que ha cambiado es la página, no la forma de obtenerla. El nivel gratuito cubre 1.500 solicitudes al mes, suficientes para ejecutar contratos «canario» con varias fuentes a diario mientras se desarrolla el proceso de validación.

Preguntas frecuentes

¿Qué es la «deriva de esquema» en el web scraping?

La «deriva de esquema» es el cambio gradual o repentino en la estructura de los datos que genera tu analizador, provocado por modificaciones en el sitio web de origen y no por cambios en tu código. Un rediseño traslada un precio a un nuevo elemento, una prueba A/B genera un diseño diferente para algunas solicitudes o un cambio de configuración regional modifica el formato numérico. El analizador sigue funcionando, pero los campos quedan vacíos o contienen datos erróneos.

¿Por qué los datos extraídos se deterioran silenciosamente en lugar de generar un error evidente?

La mayoría de los analizadores están programados para ser tolerantes, por lo que un elemento que falta devuelve «None» en lugar de generar un error. Esto es sensato si se analiza por registro, pero desastroso si se analiza por lote, ya que diez mil registros con un precio nulo se cargan sin error alguno. Solo lo detectas cuando un informe posterior parece incorrecto, a menudo semanas más tarde. La validación a nivel de lote y las comprobaciones de deriva cierran esa brecha.

¿Cómo valido los datos extraídos con pydantic?

Define un modelo por tipo de registro con campos obligatorios, tipos, rangos numéricos y restricciones de expresiones regulares; a continuación, llama a model_validate sobre cada diccionario analizado dentro de un bloque try que capture ValidationError. Cuenta los fallos por campo en todo el lote. Una alta tasa de rechazo en un campo es una señal de desviación del esquema; una tasa elevada en todos los campos suele significar que la propia recuperación ha fallado.

¿Qué umbral debería activar una alerta de desviación del esquema?

Compara la tasa de cumplimentación de cada campo en el lote actual con una referencia móvil basada en lotes recientes válidos. Una caída de diez puntos porcentuales o más en un campo obligatorio es una señal clara; utiliza cinco para campos de alto valor, como el precio. Comprueba también el recuento de registros frente a lo esperado, con un margen de más o menos un 20 %, y vigila la tasa de duplicados. Ajusta los umbrales por fuente tras una semana de datos.

¿Debería almacenar el HTML sin procesar o el Markdown de cada extracción?

Sí, si el almacenamiento lo permite. El HTML sin procesar o el Markdown son lo único que te permite rellenar los datos posteriormente tras corregir un analizador sintáctico, ya que, de lo contrario, la página tal y como estaba en el momento de la extracción ya no existe. El HTML comprimido ocupa poco espacio en comparación con el coste de volver a realizar el rastreo, y además te proporciona la evidencia necesaria para diagnosticar por qué un campo se ha desviado.

¿Puede la validación garantizar que mis datos rastreados sean correctos?

No. La validación y la detección de desviaciones detectan cambios en la forma y la distribución; no pueden verificar que un valor analizado coincida con lo que un ser humano vería en la página. Un campo puede seguir rellenado aunque su significado cambie. Combina las comprobaciones automatizadas con muestreos manuales periódicos, y recuerda que el rastreo debe respetar los términos de cada sitio web y la legislación aplicable, independientemente de la calidad de los datos.