Scraper API Webhooks frente a sondeos: cómo crear un flujo de resultados fiable

Las tareas de scraping asíncronas finalizan según su propio calendario, y la cuestión es cómo lo detecta tu código. Esta guía compara los webhooks de scraping web con el sondeo de estado y, a continuación, crea los componentes necesarios para cualquiera de las dos opciones: un bucle de sondeo con retardo y un límite máximo, un receptor FastAPI que verifica las firmas y pone en cola el trabajo, una tabla de idempotencia, la reproducción a partir del registro de entregas y un sondeador de respaldo que recoge todo lo que se le haya pasado por alto a un webhook. Al final, dispondrás de un flujo de resultados que no pierde tareas ni las procesa dos veces.

Extracción sincrónica frente a asincrónica

POST /v1/extract en Geonode y Scraper API funciona en uno de estos dos modos. Con processing_mode: "sync", la solicitud se bloquea y devuelve un código 200 con data.markdown o data.html, un objeto metadata (url, http_status, duration_ms) y tokens_charged. Con processing_mode: "async" devuelve un código 202 junto con job_id, status: "queued", status_url y estimated_tokens, y la tarea se ejecuta en segundo plano.

La sincronización es adecuada cuando necesitas la respuesta en tiempo real: una llamada desde una herramienta de agente, un gestor de solicitudes o un script puntual. La asincronía es adecuada cuando la página necesita renderización con JavaScript y una espera prolongada, cuando se envían muchas URL o cuando el solicitante no debe mantener abierta una conexión. El procesamiento por lotes (POST /v1/batch) y el rastreo (POST /v1/crawl) son siempre tareas. En cuanto tienes un job_id, tienes que elegir entre webhooks de sondeo y de extracción de datos 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"]

Sondeo: adecuado para algunos trabajos y costoso para muchos

El sondeo consiste en llamar a GET /v1/extract/{job_id} hasta que status sea completed o failed. La guía de estado de los trabajos enumera cuatro estados: queued, processing, completed y failed. El error que comete la mayoría de la gente es establecer una pausa fija de un segundo. Esto satura la API mientras el trabajo está en cola y, con un límite de frecuencia 429 (error.code "RATE_LIMITED"), el bucle de sondeo empieza a fallar antes de que lo haga el trabajo.

Utiliza un retroceso exponencial con fluctuación máxima y un límite estricto en la espera total. La fluctuación distribuye los sondeos simultáneos para que dejen de alinearse en el mismo segundo.

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)

Esto está bien para un script que envía diez URL, pero no para un lote nocturno de veinte mil. Doscientos trabajos en curso que realizan sondeos cada pocos segundos suponen miles de llamadas de estado vacías por minuto, y la latencia sigue estando limitada por el intervalo de sondeo.

Cuándo es mejor el sondeo y cuándo los webhooks

Los webhooks trasladan la complejidad del bucle del cliente a tu infraestructura. Si aún no ejecutas un punto de conexión público ni una cola, el sondeo es la primera opción adecuada. Si ya lo haces, los webhooks para el rastreo web son más económicos, más rápidos y se escalan en función de las finalizaciones en lugar de en segundos.

Registro de un webhook

Los puntos finales de los webhooks se enumeran en la referencia de la API. POST /v1/webhooks admite url (obligatorio, un URI), event_type (obligatorio, uno de los siguientes: extract_completed, batch_completed, crawl_completed) y un description opcional. La respuesta 201 devuelve el webhook id, url, description, is_active, event_type, marcas de tiempo y un secret. El secreto se devuelve al crearlo y mediante POST /v1/webhooks/{webhook_id}/rotate-secret; guárdalo en un gestor de secretos en ese momento, ya que es lo que utilizarás para verificar cada entrega.

Ten en cuenta que, en el momento de redactar este documento, la documentación indica que los puntos finales de los webhooks están documentados, pero aún no están disponibles en producción. El contrato que figura a continuación sigue la referencia publicada; compruébalo antes de implementarlo.

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

Cada webhook se suscribe a un event_type, por lo que una canalización que utilice extracción, procesamiento por lotes y rastreo registra tres.

Un receptor mínimo que verifica y pone en cola

Un receptor de webhooks tiene exactamente tres funciones: verificar la firma, devolver un código de estado 2xx lo más rápido posible y enviar el evento a una cola. Cualquier retraso superior a unos pocos cientos de milisegundos en el gestor de solicitudes es un error de diseño, ya que el intento de entrega de la plataforma podría agotar el tiempo de espera y se producirían reintentos para un trabajo que ya se ha realizado.

La documentación de referencia aún no publica el nombre del encabezado de la firma, el esquema de hash ni el esquema de la carga útil, por lo que el receptor que se muestra a continuación mantiene esos parámetros configurables y verifica un HMAC-SHA256 sobre el cuerpo sin procesar, que es el patrón habitual. Confirma el encabezado y el algoritmo exactos en la documentación y configúralos mediante variables de entorno.

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"}

Hay tres detalles importantes. La firma se calcula sobre los bytes sin procesar, no sobre un objeto JSON reserializado, ya que cualquier cambio en los espacios en blanco la invalida. La inserción en processed_events se produce antes de la puesta en cola, por lo que un fallo entre ambos procesos deja una fila sin procesar, que el poller de reserva corrige posteriormente. Además, el controlador devuelve un código 2xx en caso de duplicado en lugar de un error, ya que un error indica a la plataforma que vuelva a intentar lo que ya ha entregado.

Idempotencia y entregas fuera de orden

Cualquier sistema de webhooks funciona según el principio «al menos una vez». Fallos puntuales en la red, un trabajador lento, una implementación progresiva que pierde una respuesta: todo ello provoca una segunda entrega para el mismo trabajo. La tabla processed_events con job_id como clave primaria es la solución completa. Una restricción de unicidad hace que la comprobación de duplicados sea atómica, algo que no ocurre con un SELECT seguido de un INSERT.

El desorden es un problema distinto. Un evento batch_completed puede llegar antes de que hayas terminado de registrar el envío por lotes, o dos eventos de trabajos relacionados pueden llegar en orden inverso. No hagas que el trabajador dependa del orden. La única entrada del trabajador debería ser un job_id; este recupera el estado actual del trabajo desde la API y actúa en función de ello, que es siempre el estado real independientemente de cuándo haya llegado el evento.

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

Repetición del registro de entregas

GET /v1/webhooks/{webhook_id}/deliveries recoge todos los intentos realizados por la plataforma. Cada entrada incluye id, event_type, job_id, attempt_number, status, http_status, delivered_at y created_at, junto con page, page_size (máximo 100) y un filtro opcional status como parámetros de consulta. Los posibles valores de status no se enumeran en la referencia, por lo que, si deseas encontrar las entregas que tu punto final ha rechazado, debes filtrar por http_status.

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

Dado que handle_job pasa por la misma tabla de idempotencia, volver a ejecutar un trabajo que, de hecho, se completó con éxito no supone ningún riesgo.

El sondeador de reserva

El registro de entregas te informa de las entregas que la plataforma ha intentado realizar. No puede informarte de eventos que nunca se generaron ni de un webhook que se haya desactivado. La última línea de defensa es un bucle de conciliación lento: cada pocos minutos, enumera los trabajos con GET /v1/extract/jobs y, para cualquier trabajo más antiguo que tu umbral que no tenga ninguna fila en processed_events, procésalo directamente.

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"])

Ejecuta esto a un ritmo lo suficientemente lento como para que resulte económico y lo suficientemente rápido como para que la pérdida de un webhook no se convierta en una laguna de datos que tus usuarios puedan detectar. Diez minutos es un punto de partida razonable.

Rotación del secreto

Rota el secreto según un calendario preestablecido y de forma inmediata si se produce una filtración. POST /v1/webhooks/{webhook_id}/rotate-secret devuelve un nuevo secret y, según la documentación, invalida el anterior. Esto significa que hay un intervalo de tiempo entre el cambio de la API y el momento en que el receptor recarga su configuración, durante el cual las entregas no superarán la verificación. Mantén ese intervalo lo más breve posible: haz que el receptor lea el secreto desde una fuente que pueda actualizar sin necesidad de reiniciar, rota el secreto, actualiza la fuente y, a continuación, comprueba el registro de entregas para ver si hay alguna que haya dado un error 401 y vuelve a enviarla.

Errores habituales

Realizar el trabajo dentro del gestor de solicitudes

Recuperar el resultado, analizarlo y escribirlo en la base de datos dentro del gestor de webhooks es el error más habitual. La plataforma agota el tiempo de espera de entrega, vuelve a intentarlo y, ahora, te encuentras con duplicados y un punto final lento. Acusa recibo y ponlo en cola, nada más.

Comprobar si hay duplicados sin una restricción de unicidad

Si se produce primero un SELECT y luego un INSERT, se produce una carrera entre dos entregas simultáneas del mismo trabajo. Establece job_id como clave primaria y deja que la base de datos rechace la segunda inserción.

Confiar en la carga útil del evento más que en el estado del trabajo

Trata el webhook como una notificación, no como datos. El trabajador siempre debe volver a recuperar GET /v1/extract/{job_id} y actuar sobre el status real, lo que además hace que la repetición y la reconciliación sean sencillas.

Ejecución de webhooks sin reconciliación

Los webhooks de scraping web son de tipo «al menos una vez», no «exactamente una vez», y no se garantiza que sean «al menos una vez» en todas las interrupciones del servicio que se produzcan. Sin un sondeador de respaldo, acabarás detectando una laguna sin poder saber cuál es su magnitud.

Siguiente paso

Todo lo anterior se ejecuta en Geonode Scraper API: extracción asíncrona con processing_mode: "async", tareas por lotes y de rastreo, y suscripciones a webhooks con secretos rotativos y un registro de entregas consultable. Existe un nivel gratuito de 1.500 solicitudes al mes para desarrollar y probar el receptor, y pay-as-you-go a partir de 3,50 $ por cada 10.000 solicitudes a partir de ahí. Empieza con el sondeo, añade webhooks cuando el número de trabajos lo justifique y mantén el reconciliador en funcionamiento en cualquier caso.

Preguntas frecuentes

¿Qué son los webhooks de web scraping y cuándo debo utilizarlos?

Los webhooks de web scraping son llamadas de retorno HTTP que la plataforma de scraping envía a tu servidor cuando finaliza un trabajo asíncrono, para que no tengas que estar consultando constantemente su estado. Úsalos cuando ejecutes trabajos por lotes o de rastreo, o más de unas pocas docenas de extracciones asíncronas a la vez. Para unos pocos trabajos cortos, el sondeo es más sencillo y suficiente.

¿Es el sondeo de un scraper API una mala práctica?

No. El sondeo es la opción adecuada para pequeños volúmenes, scripts, cuadernos de trabajo y cualquier situación en la que unos pocos segundos de latencia no supongan un problema. Se convierte en un problema cuando tienes cientos de trabajos en curso, ya que cada uno supone una solicitud por intervalo de sondeo y, aun así, no puedes reaccionar más rápido que ese intervalo. Ahí es donde los webhooks de web scraping resultan útiles.

¿Cómo puedo verificar que un webhook procede realmente de la plataforma de scraping?

Cada webhook Geonode devuelve un secreto de firma al crearse, y dicho secreto se puede rotar. El receptor vuelve a calcular una firma sobre el cuerpo de la solicitud sin procesar con ese secreto y la compara en tiempo constante con el encabezado que envía la plataforma. Rechaza cualquier cosa que no supere la comprobación antes de analizar la carga útil, y nunca registres el secreto.

¿Qué ocurre si mi punto final de webhook está inactivo cuando se completa un trabajo?

La plataforma registra un intento de entrega que puedes consultar mediante GET /v1/webhooks/{webhook_id}/deliveries, donde se enumeran el número de intento (attempt_number), el estado (status) y el estado HTTP (http_status) de cada intento. Tu pipeline no debería depender únicamente de la reentrega. Ejecuta un poller de respaldo que reconcilie cualquier trabajo anterior a un umbral sin ningún evento procesado y recupere su resultado directamente.

¿Están disponibles de forma general los webhooks Geonode y Scraper API?

En el momento de redactar este documento, los puntos finales de los webhooks están documentados en docs.geonode.com, pero marcados como «aún no disponibles en producción», y no se han publicado el encabezado de firma, el esquema de la carga útil ni la programación de reintentos. Crea el receptor según el contrato documentado, mantén configurables los nombres de los encabezados y los campos, y compruébalos con la documentación antes de lanzarlo.

¿Necesito tanto los webhooks de scraping web como el sondeo?

En producción, sí. Los webhooks ofrecen baja latencia y un volumen reducido de solicitudes. Un sondeo de conciliación lento detecta los trabajos cuya entrega se ha perdido, que tu punto final ha rechazado o que tu cola ha descartado. Juntos te proporcionan un procesamiento «al menos una vez», y una idempotencia basada en el job_id lo convierte en un procesamiento «efectivamente una vez».