Scraper API Webhooks ou interrogation régulière : mettre en place un pipeline de résultats fiable

Les tâches de scraping asynchrones se terminent selon leur propre calendrier, et la question est de savoir comment votre code peut le détecter. Ce guide compare les webhooks de scraping au sondage d’état, puis met en place les éléments dont vous avez besoin pour l’une ou l’autre solution : une boucle de sondage avec délai de réessai et limite stricte, un récepteur FastAPI qui vérifie les signatures et met les tâches en file d’attente, une table d’idempotence, la relecture à partir du journal de livraison et un sondage de secours qui récupère tout ce qu’un webhook a manqué. À la fin, vous disposerez d’un pipeline de résultats qui ne perdra aucune tâche et ne les traitera pas deux fois.

Extraction synchrone ou asynchrone

Le processus POST /v1/extract sur les instances Geonode et Scraper API fonctionne selon l'un des deux modes suivants. Avec le processing_mode: "sync", la requête se bloque et renvoie un code 200 avec le data.markdown ou le data.html, un objet metadata (url, http_status, duration_ms) et tokens_charged. Avec processing_mode: "async", il renvoie un code 202 avec job_id, status: "queued", status_url et estimated_tokens, et la tâche s’exécute en arrière-plan.

Le mode synchrone est adapté lorsque vous avez besoin de la réponse immédiatement : un appel depuis un outil d’agent, un gestionnaire de requêtes, un script ponctuel. Le mode asynchrone est adapté lorsque la page nécessite un rendu JavaScript et un long délai d’attente, lorsque vous soumettez de nombreuses URL, ou lorsque l’appelant ne doit pas maintenir une connexion ouverte. Le traitement par lots (POST /v1/batch) et l’exploration (POST /v1/crawl) sont toujours des tâches. Dès que vous avez un job_id, vous devez choisir entre les webhooks de sondage et ceux de scraping Web.

import os
import httpx

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

def submit(url: str) -> str:
    r = httpx.post(
        f"{BASE}/v1/extract",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "render_js": True,
            "processing_mode": "async",
        },
        timeout=30.0,
    )
    r.raise_for_status()  # 202 on success
    return r.json()["job_id"]

Interrogation : une solution acceptable pour quelques tâches, mais coûteuse pour beaucoup

L'interrogation consiste à appeler GET /v1/extract/{job_id} jusqu'à ce que status devienne completed ou failed. Le guide des états des tâches répertorie quatre états : queued, processing, completed, failed. L’erreur que la plupart des gens commettent consiste à utiliser un délai d’attente fixe d’une seconde. Cela sollicite excessivement l’API pendant que la tâche est en file d’attente, et avec une limite de débit 429 (error.code "RATE_LIMITED"), la boucle d’interrogation commence à échouer avant même que la tâche ne le fasse.

Utilisez un recul exponentiel avec une gigue maximale et une limite stricte sur le temps d’attente total. La gigue répartit les pollers simultanés afin qu’ils ne s’alignent plus tous à la même seconde.

import os
import random
import time
import httpx

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

class JobTimeout(Exception):
    pass

def poll_job(job_id: str, max_wait: float = 300.0) -> dict:
    deadline = time.monotonic() + max_wait
    delay = 1.0
    while True:
        r = httpx.get(f"{BASE}/v1/extract/{job_id}", headers=HEADERS, timeout=15.0)
        if r.status_code == 429:
            retry_after = float(r.headers.get("Retry-After", delay))
            time.sleep(retry_after)
            continue
        r.raise_for_status()
        job = r.json()
        if job["status"] in ("completed", "failed"):
            return job
        if time.monotonic() >= deadline:
            raise JobTimeout(job_id)
        # full jitter: sleep somewhere in [0, delay], then grow delay up to 30 s
        time.sleep(random.uniform(0, delay))
        delay = min(delay * 2, 30.0)

Cela convient pour un script qui soumet dix URL, mais ne convient pas pour un traitement par lots nocturne de vingt mille. Deux cents tâches en cours de traitement effectuant des interrogations toutes les quelques secondes représentent des milliers d’appels d’état vides par minute, et la latence reste limitée par l’intervalle d’interrogation.

Quand le polling l'emporte et quand les webhooks l'emportent

Les webhooks transfèrent la complexité de la boucle client vers votre infrastructure. Si vous n’exploitez pas déjà un point de terminaison public et une file d’attente, l’interrogation est la bonne solution de départ. Si c’est le cas, les webhooks de scraping web sont moins coûteux, plus rapides et s’adaptent au nombre de tâches terminées plutôt qu’au nombre de secondes.

Enregistrement d'un webhook

Les points de terminaison des webhooks sont répertoriés dans la documentation de l'API. POST /v1/webhooks prend en paramètre url (obligatoire, une URI), event_type (obligatoire, l'un des éléments suivants : extract_completed, batch_completed, crawl_completed) et un description facultatif. La réponse 201 renvoie les webhooks id, url, description, is_active, event_type, des horodatages et un secret. Le secret est renvoyé lors de la création et par le POST /v1/webhooks/{webhook_id}/rotate-secret ; enregistrez-le dans un gestionnaire de secrets à ce moment-là, car c’est ce que vous utiliserez pour vérifier chaque livraison.

Notez qu’au moment de la rédaction de ce document, la documentation indique que les points de terminaison des webhooks sont documentés mais pas encore disponibles en production. Le contrat ci-dessous suit la référence publiée ; vérifiez-le avant de procéder au déploiement.

import os
import httpx

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

def register_webhook(receiver_url: str, event_type: str = "extract_completed") -> dict:
    r = httpx.post(
        f"{BASE}/v1/webhooks",
        headers=HEADERS,
        json={
            "url": receiver_url,
            "event_type": event_type,
            "description": "results pipeline, prod",
        },
        timeout=15.0,
    )
    r.raise_for_status()  # 201
    hook = r.json()
    # hook["secret"] is only shown here and after rotate-secret: persist it now
    return hook

Un webhook s'abonne à un event_type ; ainsi, un pipeline utilisant l'extraction, le traitement par lots et l'exploration en enregistre trois.

Un récepteur minimal qui vérifie et met en file d'attente

Un récepteur de webhook a exactement trois tâches : vérifier la signature, renvoyer un code 2xx aussi rapidement que possible et transmettre l’événement à une file d’attente. Tout temps de traitement supérieur à quelques centaines de millisecondes dans le gestionnaire de requêtes constitue une erreur de conception, car la tentative de livraison de la plateforme risque d’expirer et vous verrez alors des tentatives de réessai pour un travail que vous avez déjà effectué.

La documentation de référence ne publie pas encore le nom de l’en-tête de signature, le schéma de hachage ni le schéma de la charge utile ; le récepteur ci-dessous laisse donc ces paramètres configurables et vérifie un HMAC-SHA256 sur le corps brut, ce qui correspond à la pratique courante. Vérifiez l’en-tête et l’algorithme exacts dans la documentation et définissez-les via des variables d’environnement.

import hashlib
import hmac
import json
import os
import sqlite3

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

WEBHOOK_SECRET = os.environ["GEONODE_WEBHOOK_SECRET"].encode()
SIGNATURE_HEADER = os.environ.get("GEONODE_SIGNATURE_HEADER", "X-Webhook-Signature")  # confirm header name in docs
DB = sqlite3.connect("events.db", check_same_thread=False)
DB.execute(
    "CREATE TABLE IF NOT EXISTS processed_events ("
    "  job_id TEXT PRIMARY KEY,"
    "  event_type TEXT NOT NULL,"
    "  delivery_id TEXT,"
    "  received_at TEXT DEFAULT CURRENT_TIMESTAMP"
    ")"
)

def verify_signature(raw_body: bytes, header_value: str | None) -> bool:
    if not header_value:
        return False
    expected = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()  # confirm algorithm in docs
    return hmac.compare_digest(expected, header_value.strip())

def enqueue(event: dict) -> None:
    # Replace with your queue client (Redis, SQS, Celery, etc.).
    print("enqueued", event.get("job_id"))

@app.post("/webhooks/geonode")
async def receive(request: Request):
    raw = await request.body()
    if not verify_signature(raw, request.headers.get(SIGNATURE_HEADER)):
        raise HTTPException(status_code=401, detail="bad signature")

    event = json.loads(raw)
    job_id = event.get("job_id")  # confirm field name in docs
    event_type = event.get("event_type")  # confirm field name in docs
    delivery_id = event.get("id")  # confirm field name in docs
    if not job_id:
        raise HTTPException(status_code=400, detail="missing job_id")

    try:
        DB.execute(
            "INSERT INTO processed_events (job_id, event_type, delivery_id) VALUES (?, ?, ?)",
            (job_id, event_type, delivery_id),
        )
        DB.commit()
    except sqlite3.IntegrityError:
        # Duplicate delivery: already recorded, acknowledge and do nothing.
        return {"status": "duplicate"}

    enqueue({"job_id": job_id, "event_type": event_type})
    return {"status": "accepted"}

Trois détails sont importants. La signature est calculée sur les octets bruts, et non sur un objet JSON resérialisé, car toute modification des espaces blancs la rendrait invalide. L’insertion dans processed_events a lieu avant la mise en file d’attente ; ainsi, un plantage entre ces deux opérations laisse une ligne sans traitement, ce que le poller de secours corrige ultérieurement. Enfin, le gestionnaire renvoie un code 2xx en cas de doublon plutôt qu’une erreur, car une erreur indiquerait à la plateforme de réessayer ce qu’elle a déjà traité.

Idempotence et livraisons hors ordre

Tout système de webhook fonctionne selon le principe « au moins une fois ». Des perturbations réseau, un worker lent, un déploiement progressif qui perd une réponse : tous ces éléments entraînent une deuxième livraison pour la même tâche. La table processed_events, avec job_id comme clé primaire, résout entièrement le problème. Une contrainte d’unicité rend la vérification des doublons atomique, ce qui n’est pas le cas avec un SELECT suivi d’un INSERT.

Le désordre d’exécution est un problème distinct. Un événement batch_completed peut arriver avant que vous ayez fini d’enregistrer la soumission par lot, ou deux événements pour des tâches liées peuvent arriver dans l’ordre inverse. Ne faites pas en sorte que le worker dépende de l’ordre. La seule entrée du worker doit être un job_id ; il récupère l’état actuel du job via l’API et agit en fonction de celui-ci, qui est toujours valable quel que soit le moment où l’événement est arrivé.

import os
import httpx

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

def store_result(job_id: str, job: dict) -> None: ...   # write to your store
def resubmit(job_id: str) -> None: ...                  # re-POST the original URL
def dead_letter(job_id: str, err: dict) -> None: ...    # park for manual review

def handle_job(job_id: str) -> None:
    r = httpx.get(f"{BASE}/v1/extract/{job_id}", headers=HEADERS, timeout=15.0)
    r.raise_for_status()
    job = r.json()
    if job["status"] == "completed":
        store_result(job_id, job)
    elif job["status"] == "failed":
        err = job.get("error", {})  # confirm field name in docs
        if err.get("retryable"):
            resubmit(job_id)
        else:
            dead_letter(job_id, err)
    else:
        # Event arrived before the job settled; let the fallback poller retry.
        pass

Historique des livraisons

GET /v1/webhooks/{webhook_id}/deliveries répertorie toutes les tentatives effectuées par la plateforme. Chaque élément porte les références id, event_type, job_id, attempt_number, status, http_status, delivered_at et created_at, avec page, page_size (max. 100) et un filtre status facultatif comme paramètres de requête. Les valeurs possibles de status ne sont pas répertoriées dans la documentation de référence ; utilisez donc plutôt le filtre http_status lorsque vous souhaitez rechercher les livraisons rejetées par votre point de terminaison.

import os
import httpx

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

def replay_failed(webhook_id: str) -> list[str]:
    missed: list[str] = []
    page = 1
    while True:
        r = httpx.get(
            f"{BASE}/v1/webhooks/{webhook_id}/deliveries",
            headers=HEADERS,
            params={"page": page, "page_size": 100},
            timeout=15.0,
        )
        r.raise_for_status()
        body = r.json()
        for d in body["items"]:
            ok = d.get("http_status") is not None and 200 <= d["http_status"] < 300
            if not ok and d["job_id"] not in missed:
                missed.append(d["job_id"])
        if page >= body["page_count"]:
            break
        page += 1
    for job_id in missed:
        handle_job(job_id)
    return missed

Étant donné que handle_job passe par la même table d’idempotence, la réexécution d’un travail qui a en réalité abouti est sans conséquence.

Le poller de secours

Le journal des envois vous informe des envois que la plateforme a tenté d'effectuer. Il ne peut toutefois pas vous renseigner sur les événements qui n'ont jamais été générés, ni sur un webhook qui a été désactivé. La dernière ligne de défense consiste en une boucle de réconciliation lente : toutes les quelques minutes, répertoriez les tâches figurant dans GET /v1/extract/jobs, et pour toute tâche plus ancienne que votre seuil et ne comportant aucune ligne dans processed_events, traitez-la directement.

import os
import httpx
from datetime import datetime, timedelta, timezone

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

def reconcile(older_than: timedelta = timedelta(minutes=10)) -> None:
    r = httpx.get(f"{BASE}/v1/extract/jobs", headers=HEADERS, timeout=30.0)
    r.raise_for_status()
    cutoff = datetime.now(timezone.utc) - older_than
    for job in r.json().get("items", []):  # confirm field name in docs
        created = datetime.fromisoformat(job["created_at"].replace("Z", "+00:00"))  # confirm field name in docs
        if created > cutoff or job["status"] not in ("completed", "failed"):
            continue
        seen = DB.execute(
            "SELECT 1 FROM processed_events WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        if seen is None:
            DB.execute(
                "INSERT INTO processed_events (job_id, event_type) VALUES (?, ?)",
                (job["job_id"], "reconciled"),
            )
            DB.commit()
            handle_job(job["job_id"])

Exécutez cette opération à une cadence suffisamment lente pour que cela reste peu coûteux, mais suffisamment rapide pour qu’un webhook manquant n’entraîne pas de lacune dans les données perceptible par vos utilisateurs. Dix minutes constituent un bon point de départ.

Rotation des secrets

Faites tourner le secret selon un calendrier défini et immédiatement en cas de fuite. POST /v1/webhooks/{webhook_id}/rotate-secret est remplacé par un nouveau secret et, conformément à la documentation, l'ancien est invalidé. Cela signifie qu’il existe un intervalle entre le basculement de l’API et le rechargement de la configuration par votre récepteur, pendant lequel les livraisons échoueront à la vérification. Réduisez cet intervalle au minimum : faites en sorte que le récepteur lise le secret à partir d’une source qu’il peut actualiser sans redémarrage, effectuez la rotation, mettez à jour la source, puis vérifiez le journal des livraisons pour repérer tout message ayant abouti à un code 401 et relancez-le.

Erreurs courantes

Effectuer le traitement au sein du gestionnaire de requêtes

Récupérer le résultat, l’analyser et l’enregistrer dans la base de données au sein du gestionnaire de webhooks est l’erreur la plus courante. La plateforme expire, elle réessaie, et vous vous retrouvez alors avec des doublons et un point de terminaison lent. Accusez réception et mettez en file d’attente, rien de plus.

Vérification des doublons sans contrainte d’unicité

Une séquence SELECT puis INSERT entraîne une situation de concurrence entre deux livraisons simultanées d’une même tâche. Définissez job_id comme clé primaire et laissez la base de données rejeter la deuxième insertion.

Privilégier la charge utile de l’événement plutôt que l’état de la tâche

Considérez le webhook comme une notification, et non comme des données. Le worker doit toujours récupérer à nouveau GET /v1/extract/{job_id} et agir sur le véritable status, ce qui simplifie également la relecture et le rapprochement.

Exécution de webhooks sans réconciliation

Les webhooks de scraping Web fonctionnent selon le principe « au moins une fois » (at-least-once), et non « exactement une fois » (exactly-once) ; il n’est pas garanti qu’ils soient exécutés « au moins une fois » lors de chaque interruption de service. Sans poller de secours, vous finirez par découvrir un écart sans pouvoir en évaluer l’ampleur.

Étape suivante

Tout ce qui précède s'exécute sur Geonode Scraper API : extraction asynchrone avec processing_mode: "async", tâches par lots et d'exploration, ainsi que des abonnements aux webhooks avec des secrets renouvelables et un journal de livraison consultable. Il existe une offre gratuite de 1 500 requêtes par mois pour développer et tester le récepteur, puis le tarif pay-as-you-go à partir de 3,50 $ pour 10 000 requêtes au-delà de ce seuil. Commencez par l'interrogation régulière, ajoutez des webhooks lorsque le nombre de tâches le justifie, et laissez le réconciliateur fonctionner dans tous les cas.

FAQ

Que sont les webhooks de web scraping et quand dois-je les utiliser ?

Les webhooks de web scraping sont des rappels HTTP que la plateforme de scraping envoie à votre serveur lorsqu’une tâche asynchrone est terminée, ce qui vous évite d’avoir à interroger sans cesse son statut. Utilisez-les lorsque vous exécutez des tâches par lots ou d’exploration, ou plus d’une douzaine d’extractions asynchrones à la fois. Pour quelques tâches courtes, l’interrogation régulière est plus simple et suffit amplement.

L’interrogation régulière d’un scraper API est-elle une mauvaise pratique ?

Non. L’interrogation régulière est le bon choix pour les petits volumes, les scripts, les notebooks et tout ce pour quoi quelques secondes de latence n’ont pas d’importance. Cela devient un problème lorsque vous avez des centaines de tâches en cours, car chacune d’entre elles coûte une requête par intervalle d’interrogation et vous ne pouvez toujours pas réagir plus rapidement que cet intervalle. C’est là que les webhooks de web scraping prennent tout leur sens.

Comment vérifier qu’un webhook provient bien de la plateforme de scraping ?

Chaque webhook Geonode renvoie une clé secrète de signature lors de sa création, et cette clé peut être renouvelée. Le destinataire recalcule une signature sur le corps brut de la requête à l’aide de ce secret et la compare en temps constant à l’en-tête envoyé par la plateforme. Rejetez tout ce qui échoue à cette vérification avant d’analyser la charge utile, et ne consignez jamais le secret.

Que se passe-t-il si mon point de terminaison de webhook est indisponible lorsqu’une tâche est terminée ?

La plateforme enregistre une tentative de livraison que vous pouvez consulter via GET /v1/webhooks/{webhook_id}/deliveries, qui répertorie le numéro de tentative (attempt_number), le statut (status) et le statut HTTP (http_status) pour chaque tentative. Votre pipeline ne doit pas dépendre uniquement de la redélivrance. Exécutez un poller de secours qui recoupe tout travail datant de plus d’un seuil défini et ne comportant aucun événement traité, puis récupérez directement son résultat.

Les webhooks Geonode et Scraper API sont-ils généralement disponibles ?

À la date de rédaction de ce document, les points de terminaison des webhooks sont documentés sur docs.geonode.com mais signalés comme n’étant pas encore disponibles en production ; l’en-tête de signature, le schéma de charge utile et le calendrier de réessais ne sont pas publiés. Implémentez le récepteur conformément au contrat documenté, veillez à ce que les noms d’en-tête et de champ restent configurables, et vérifiez-les par rapport à la documentation avant la mise en production.

Ai-je besoin à la fois des webhooks de scraping et de l’interrogation ?

En production, oui. Les webhooks offrent une faible latence et un faible volume de requêtes. Un poller de réconciliation lent détecte les tâches dont la livraison a échoué, qui ont été rejetées par votre point de terminaison ou qui ont été supprimées de votre file d’attente. Ensemble, ils garantissent un traitement « au moins une fois », et une table d’idempotence indexée sur le job_id transforme cela en un traitement « effectivement une fois ».