Les données extraites ne présentent que rarement des erreurs flagrantes. Elles se détériorent insidieusement : une refonte du site déplace le prix vers un élément différent, l’analyseur renvoie « None », et dix mille lignes avec des prix vides sont chargées dans l’entrepôt de données sans la moindre exception. Ce guide présente les couches qui permettent de détecter la dérive de schéma avant qu’elle ne se traduise par un rapport erroné : un schéma d’enregistrement pydantic v2, des contrôles de qualité par lot, un détecteur de dérive avec une base de référence glissante, une étape de mise en quarantaine, un test quotidien du contrat de sélection sur des URL « canary », et la gestion des versions de l’analyseur afin que vous puissiez effectuer un backfill une fois le problème résolu.
Pourquoi les données extraites se détériorent sans qu’on s’en aperçoive
Trois sources de dérive de schéma sont à l’origine de la quasi-totalité des incidents. Les refontes de sites modifient le DOM, et un analyseur tolérant écrit avec .get() et find() renvoie des valeurs vides plutôt que de générer une exception. Les mises en page A/B sont encore pires, car seule une fraction des requêtes voit le nouveau modèle ; votre taux de remplissage chute ainsi de 99 % à 70 %, sans que personne ne s’en aperçoive pendant une semaine. Les changements de paramètres régionaux et géographiques modifient les formats sans altérer la structure : 1,299.00 devient 1.299,00, un format de date passe de la norme ISO au format « jour en premier », et la valeur est mal analysée plutôt que de ne pas l’être du tout.
Le point commun est que l’analyseur ne plante pas. La tolérance au niveau des enregistrements est le bon paramètre par défaut, mais cela signifie que l’exactitude doit être garantie ailleurs. Cet « ailleurs », c’est la validation des données au niveau des lots dans les pipelines de web scraping, et elle est peu coûteuse à mettre en place.
Définir un schéma d'enregistrement avec pydantic v2
Commencez par définir la structure d'un enregistrement valide. Les champs obligatoires, les types, les plages de valeurs et les formats constituent un modèle pydantic, et chaque dictionnaire analysé passe par model_validate avant d'être traité ailleurs.
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
Deux choix sont ici déterminants. L’utilisation de mode="before" sur le validateur de prix permet au schéma d’effectuer la dernière étape de normalisation, ce qui simplifie le parseur. Et e.errors() renvoie des erreurs structurées sous la forme d’un tuple loc, ce qui vous permet de compter les échecs par champ plutôt que par enregistrement. Un lot dans lequel 40 % des lignes échouent sur price et rien d’autre pose un problème différent de celui où chaque ligne échoue sur tous les champs.
Contrôles par lot
La validation rejette les lignes. Les contrôles par lot examinent à la fois les lignes validées et celles qui ne le sont pas, et vérifient si le lot dans son ensemble est plausible. Quatre contrôles permettent de détecter la plupart des dérives de schéma.
| Contrôle | Signal | Seuil typique |
|---|---|---|
| Taux de valeurs nulles ou de rejet par champ | Mise en page A/B, élément déplacé | > 10 points au-dessus de la valeur de référence |
| Nombre d’enregistrements par rapport aux prévisions | Modification de la pagination ou de la liste, blocage | hors de la fourchette prévue de ±20 % |
| Taux de doublons sur la clé | Boucle de pagination, modification de l’URL canonique | > 2 % |
| Décalage de la distribution des valeurs | Changement de paramètres régionaux ou d’unités | décalage de la médiane > 30 % |
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
Conservez le fichier batch_metrics à chaque exécution. Il sert de matière première au détecteur de dérive, et c’est également ce que vous consulterez en premier lorsque quelqu’un vous demandera pourquoi les chiffres de mardi semblent anormaux.
Un détecteur de dérive avec une ligne de base glissante
Les seuils fixes ne sont pas fiables car les sources varient. Un champ rempli à 60 % sur un site est normal ; sur un autre, c'est une situation critique. Comparez plutôt avec l'historique récent de la source elle-même : une ligne de base glissante calculée à partir des N derniers lots corrects.
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 ligne importante est accept : seuls les lots conformes alimentent la base de référence. Si vous laissez passer un lot présentant une dérive, la base de référence suit cette dérive et le détecteur se tait précisément au moment où il devrait s'activer. Utilisez la médiane plutôt que la moyenne pour la même raison ; un jour atypique ne devrait pas modifier la référence.
Mise en quarantaine plutôt que chargement
Lorsque le détecteur se déclenche, ne chargez pas le lot, mais ne le rejetez pas non plus. Enregistrez-le dans un emplacement de quarantaine en y joignant ses métriques et ses alertes, envoyez l’alerte, puis laissez un opérateur humain ou une règle automatisée prendre la décision. La décision se résume généralement à l’une des trois options suivantes : fausse alerte (valider le lot et l’intégrer dans la ligne de base), dérive réelle (corriger l’analyseur syntaxique, relancer le traitement à partir du contenu brut stocké) ou interruption de la source (rejeter le lot, ne pas modifier la ligne de base).
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
Générez une alerte sur ce qui a changé, et non sur un « échec de validation ». Un message indiquant « price fill rate 0.98 -> 0.61 on source X » est exploitable ; un échec générique ne l’est pas.
Le test de contrat de sélection sur les URL « canary »
Les vérifications par lots vous signalent un écart après coup. Un test de contrat de sélection vous alerte plus tôt, souvent avant même l’exécution quotidienne. Choisissez un ensemble fixe d’URL « canary » par source, dont vous connaissez les valeurs attendues, récupérez-les chaque jour via POST /v1/extract, lancez l’analyseur et vérifiez que chaque champ défini dans le contrat est présent et plausible. Lorsque le contrat échoue, l’analyseur doit faire l’objet d’une attention particulière avant le début de l’exécution principale.
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)
Cinq à dix « canaris » par source suffisent. Définissez le pays du proxy si le site est localisé, car un « canari » récupéré depuis une autre région teste une page différente. Choisissez des URL qui couvrent les cas de figure qui vous intéressent : une en stock, une en rupture de stock, une en promotion, une sans note.
Versionner l'analyseur et stocker la page brute
Chaque ligne doit comporter la chaîne « parser_version », et la version doit changer chaque fois que la logique de l'analyseur évolue. Cela vous apporte deux avantages. Premièrement, vous pouvez segmenter n'importe quelle métrique par version et voir immédiatement si une alerte de dérive coïncide avec un déploiement. Deuxièmement, lorsque vous corrigez un analyseur, vous savez exactement quelles lignes régénérer.
La régénération n’est possible que si vous avez conservé les données d’entrée. Stockez le code HTML brut ou Markdown de chaque extraction, compressé, indexé par source_url et scraped_at. Le coût de stockage est négligeable par rapport à celui d’une nouvelle extraction, et cette dernière n’est souvent même pas envisageable car la page a de nouveau changé. Une fois la dérive du schéma confirmée, le workflow est le suivant : corriger le parseur, incrémenter la version, réanalyser le contenu brut stocké pour la période concernée, le valider à l’aide du même schéma, puis le charger pour remplacer les lots mis en quarantaine. Sans stockage des données brutes, le décalage est permanent.
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
Erreurs courantes
Valider en fonction des enregistrements et jamais par lot
Un schéma d’enregistrement permet de détecter les lignes mal formées ; il ne peut toutefois pas détecter un lot dans lequel toutes les lignes sont bien formées mais où la moitié des prix manque. Calculez toujours les taux de remplissage et les comptages au niveau du lot.
Laisser des lots erronés entrer dans la base de référence
Si la base de référence glissante inclut un lot présentant un écart, le lot suivant présentant un écart paraîtra normal. N’acceptez que les lots ayant été intégrés à l’historique, et utilisez une médiane.
Rendre l’analyseur syntaxique trop tolérant
.get("price") renvoyant None est acceptable, mais un analyseur qui renvoie 0.0 ou une chaîne vide en cas d’échec rend la vérification du taux de remplissage inefficace. Renvoyez None en cas de données manquantes, et laissez le schéma le rejeter.
Ne pas fixer les conditions de récupération
Les tests « canary » et les exécutions en production doivent utiliser les mêmes paramètres de rendu et de proxy. Un test de contrat qui réussit avec render_js: true et un proxy américain ne prouve rien concernant une exécution qui n’utilise ni l’un ni l’autre.
Étape suivante
Toutes les requêtes décrites dans ce guide passent par la fonction Geonode Scraper API, qui renvoie du Markdown ou du HTML « propre » à partir d’un seul appel à POST /v1/extract, avec rendu JavaScript et ciblage par proxy au niveau national lorsque cela est nécessaire. Le fait de maintenir la couche de récupération stable élimine une variable des analyses de dérive de schéma, car ce qui a changé, c’est la page, et non la manière dont vous l’avez obtenue. Le niveau gratuit couvre 1 500 requêtes par mois, ce qui est suffisant pour exécuter quotidiennement des contrats « canary » sur plusieurs sources pendant que vous mettez en place le pipeline de validation.
FAQ
Qu'est-ce que la « dérive de schéma » dans le web scraping ?
La dérive de schéma désigne le changement progressif ou soudain de la structure des données produites par votre analyseur, provoqué par une modification du site source et non par votre code. Une refonte du site déplace un prix vers un nouvel élément, un test A/B génère une mise en page différente pour certaines requêtes, ou un changement de paramètres régionaux modifie le format des nombres. L’analyseur continue de fonctionner, mais les champs se retrouvent vides ou erronés.
Pourquoi les données extraites se détériorent-elles silencieusement au lieu de signaler une erreur de manière évidente ?
La plupart des parseurs sont conçus pour être tolérants : ainsi, un élément manquant renvoie « None » au lieu de déclencher une exception. C’est raisonnable au niveau d’un enregistrement, mais désastreux au niveau d’un lot, car dix mille enregistrements avec un prix nul sont chargés sans erreur. Vous ne le détectez que lorsqu’un rapport en aval semble erroné, souvent plusieurs semaines plus tard. La validation au niveau du lot et les contrôles de dérive comblent cette lacune.
Comment valider les données extraites avec pydantic ?
Définissez un modèle par type d’enregistrement avec les champs obligatoires, les types, les plages numériques et les contraintes d’expressions régulières, puis appelez model_validate sur chaque dictionnaire analysé à l’intérieur d’un bloc try qui intercepte les ValidationError. Comptez le nombre d’échecs par champ sur l’ensemble du lot. Un taux de rejet élevé sur un champ est un signe de dérive du schéma ; un taux élevé sur l’ensemble des champs signifie généralement que la récupération elle-même a échoué.
Quel seuil doit déclencher une alerte de dérive du schéma ?
Comparez le taux de remplissage de chaque champ dans le lot actuel à une référence glissante établie à partir des derniers lots valides. Une baisse de dix points de pourcentage ou plus sur un champ obligatoire est un signal fort ; utilisez cinq points pour les champs à forte valeur ajoutée comme le prix. Vérifiez également que le nombre d’enregistrements se situe dans une fourchette de plus ou moins 20 % par rapport aux prévisions et surveillez le taux de doublons. Ajustez les seuils par source après une semaine de collecte de données.
Dois-je stocker le code HTML brut ou au format Markdown de chaque extraction ?
Oui, si l’espace de stockage le permet. Le code HTML brut ou au format Markdown est le seul moyen de reconstituer les données après avoir corrigé un analyseur syntaxique, car sinon, la page telle qu’elle était au moment de l’extraction a disparu. Le HTML compressé occupe peu d’espace par rapport au coût d’un nouveau scraping, et il vous fournit également les éléments nécessaires pour diagnostiquer les causes d’une dérive de champ.
La validation peut-elle garantir l’exactitude de mes données extraites ?
Non. La validation et la détection des dérives permettent de détecter les changements de structure et de répartition ; elles ne peuvent pas vérifier qu’une valeur analysée correspond à ce qu’un humain verrait sur la page. Un champ peut rester renseigné alors que sa signification change. Associez les contrôles automatisés à un échantillonnage manuel périodique, et n’oubliez pas que le scraping doit respecter les conditions d’utilisation de chaque site et la législation applicable, quelle que soit la qualité des données.