Scraper API Webhooks vs. Polling: Criar um fluxo de resultados fiável

As tarefas de scraping assíncronas terminam de acordo com o seu próprio calendário, e a questão é saber como é que o seu código fica a saber disso. Este guia compara os webhooks de scraping com a sondagem de estado e, em seguida, constrói os componentes necessários para qualquer uma das opções: um ciclo de sondagem com backoff e um limite máximo, um recetor FastAPI que verifica assinaturas e enfileira tarefas, uma tabela de idempotência, a reprodução a partir do registo de entregas e um sondador de fallback que capta tudo o que um webhook tenha deixado escapar. No final, terá um pipeline de resultados que não perde tarefas nem as processa duas vezes.

Extração síncrona vs. assíncrona

O POST /v1/extract no Geonode Scraper API funciona num de dois modos. Com o processing_mode: "sync", o pedido fica bloqueado e devolve o código 200 com o data.markdown ou o data.html, um objeto metadata (url, http_status, duration_ms) e tokens_charged. Com o processing_mode: "async", devolve o código 202 com o job_id, status: "queued", status_url e estimated_tokens, e a tarefa é executada em segundo plano.

A sincronização é adequada quando se precisa da resposta imediata: uma chamada de ferramenta de agente, um gestor de pedidos, um script pontual. A assincronia é adequada quando a página necessita de renderização em JavaScript e de uma longa espera, quando se estão a enviar muitos URLs ou quando o chamador não deve manter uma ligação aberta. O processamento em lote (POST /v1/batch) e o rastreamento (POST /v1/crawl) são sempre tarefas. Assim que tiver um job_id, terá de escolher entre webhooks de sondagem e de extração de dados da 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"]

Polling: adequado para algumas tarefas e dispendioso para muitas

Polling significa chamar o GET /v1/extract/{job_id} até que o status seja o completed ou o failed. O guia de estado das tarefas enumera quatro estados: queued, processing, completed, failed. O erro que a maioria das pessoas comete é definir um tempo de espera fixo de um segundo. Isso sobrecarrega a API enquanto a tarefa está na fila e, com um limite de taxa 429 (error.code "RATE_LIMITED"), o ciclo de sondagem começa a falhar antes mesmo da tarefa.

Utilize o backoff exponencial com jitter total e um limite máximo rígido para o tempo de espera total. O jitter distribui os polizadores simultâneos, para que deixem de se alinhar no mesmo 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)

Isto é adequado para um script que submete dez URLs, mas inadequado para um lote noturno de vinte mil. Duzentas tarefas em execução a fazer poling a cada poucos segundos representam milhares de chamadas de estado vazias por minuto, e a latência continua a ser limitada pelo intervalo de poling.

Quando o polling é a melhor opção e quando os webhooks são a melhor opção

Os webhooks transferem a complexidade do ciclo do cliente para a sua infraestrutura. Se ainda não tiver um ponto de extremidade público e uma fila em funcionamento, a sondagem é a primeira versão correta. Se tiver, os webhooks de web scraping são mais económicos, mais rápidos e escalam com base nas conclusões, em vez de segundos.

Registo de um webhook

Os pontos finais do webhook estão listados na referência da API. O POST /v1/webhooks aceita o url (obrigatório, um URI), o event_type (obrigatório, um dos seguintes: extract_completed, batch_completed, crawl_completed) e um description opcional. A resposta 201 devolve os webhooks id, url, description, is_active, event_type, carimbos de data/hora e um secret. O segredo é devolvido aquando da criação e pelo POST /v1/webhooks/{webhook_id}/rotate-secret; guarde-o num gestor de segredos nesse momento, pois é o que irá utilizar para verificar cada entrega.

Note que, à data da redação deste documento, a documentação indica que os pontos finais dos webhooks estão documentados, mas ainda não estão disponíveis em produção. O contrato abaixo segue a referência publicada; verifique-o antes de efetuar a implementação.

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

Um webhook subscreve um event_type, pelo que um pipeline que utilize extração, processamento em lote e rastreamento regista três.

Um recetor minimalista que verifica e coloca na fila

Um recetor de webhooks tem exatamente três tarefas: verificar a assinatura, devolver um código de estado 2xx o mais rapidamente possível e encaminhar o evento para uma fila. Qualquer tempo de resposta superior a algumas centenas de milissegundos no manipulador de pedidos constitui um erro de conceção, uma vez que a tentativa de entrega da plataforma pode atingir o tempo limite e irá observar novas tentativas para um trabalho que já realizou.

A documentação de referência ainda não publica o nome do cabeçalho da assinatura, o esquema de hash nem o esquema da carga útil; por isso, o recetor abaixo mantém esses parâmetros configuráveis e verifica um HMAC-SHA256 sobre o corpo bruto, o que é o padrão comum. Confirme o cabeçalho e o algoritmo exatos na documentação e defina-os através de variáveis de ambiente.

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

Há três detalhes importantes. A assinatura é calculada sobre os bytes brutos, e não sobre um objeto JSON resserializado, porque qualquer alteração nos espaços em branco a invalida. A inserção no processed_events ocorre antes da colocação na fila, pelo que uma falha entre os dois processos deixa uma linha sem processamento, o que o poller de fallback corrige posteriormente. E o manipulador devolve um código 2xx em caso de duplicação, em vez de um erro, porque um erro indica à plataforma para repetir o que já foi entregue.

Idempotência e entregas fora de ordem

Qualquer sistema de webhooks funciona segundo o princípio «pelo menos uma vez». Falhas de rede, um worker lento, uma implementação gradual que perde uma resposta: tudo isto resulta numa segunda entrega para o mesmo trabalho. A tabela processed_events com job_id como chave primária é a solução completa. Uma restrição de unicidade torna a verificação de duplicados atómica, o que não acontece com um SELECT seguido de um INSERT.

A desordenação é um problema diferente. Um evento batch_completed pode chegar antes de ter terminado de registar o envio em lote, ou dois eventos para tarefas relacionadas podem chegar na ordem inversa. Não faça com que o trabalhador dependa da ordem. A única entrada do trabalhador deve ser um job_id; este obtém o estado atual do trabalho a partir da API e age com base nisso, o que é sempre a verdade, independentemente de quando o evento chegou.

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

Reprodução a partir do registo de entregas

O GET /v1/webhooks/{webhook_id}/deliveries lista todas as tentativas efetuadas pela plataforma. Cada item inclui o id, o event_type, o job_id, o attempt_number, o status, http_status, delivered_at e created_at, com page, page_size (máximo de 100) e um filtro opcional status como parâmetros de consulta. Os valores possíveis para status não estão enumerados na referência; por isso, filtre por http_status quando pretender encontrar entregas que o seu ponto de extremidade tenha rejeitado.

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

Uma vez que o handle_job passa pela mesma tabela de idempotência, a repetição de um trabalho que, de facto, foi bem-sucedido é inofensiva.

O poller de reserva

O registo de entregas fornece informações sobre as entregas que a plataforma tentou efetuar. Não fornece informações sobre eventos que nunca foram gerados, nem sobre um webhook que tenha sido desativado. A última linha de defesa é um ciclo de reconciliação lento: a cada poucos minutos, liste os trabalhos com o código GET /v1/extract/jobs e, para qualquer trabalho mais antigo do que o seu limiar que não tenha uma linha no processed_events, processe-o diretamente.

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

Execute isto com uma frequência suficientemente lenta para ser económica e suficientemente rápida para que um webhook perdido não se torne uma lacuna de dados que os seus utilizadores notem. Dez minutos é um ponto de partida razoável.

Rotação do segredo

Efetue a rotação do segredo de acordo com um calendário e imediatamente em caso de fuga de informação. O POST /v1/webhooks/{webhook_id}/rotate-secret dá lugar a um novo secret e, de acordo com a documentação, invalida o anterior. Isso significa que existe um intervalo entre a mudança da API e a recarga da configuração do seu recetor, durante o qual as entregas não passarão na verificação. Mantenha esse intervalo curto: faça com que o recetor leia o segredo a partir de uma fonte que possa ser atualizada sem necessidade de reinício, faça a rotação, atualize a fonte e, em seguida, verifique o registo de entregas para identificar quaisquer mensagens que tenham recebido um código de erro 401 e reenvie-as.

Erros comuns

Executar o trabalho dentro do manipulador de pedidos

Obter o resultado, analisá-lo e gravá-lo na base de dados dentro do manipulador de webhooks é a falha mais comum. A plataforma atinge o tempo limite de entrega, tenta novamente e, agora, tem-se duplicados e um ponto final lento. Confirmar e colocar na fila, nada mais.

Verificar se há duplicados sem uma restrição de unicidade

Um SELECT seguido de um INSERT implica uma situação de corrida entre duas entregas simultâneas do mesmo trabalho. Defina job_id como chave primária e deixe que a base de dados rejeite a segunda inserção.

Confiar na carga útil do evento em vez do estado do trabalho

Trate o webhook como uma notificação, não como dados. O trabalhador deve sempre voltar a obter o GET /v1/extract/{job_id} e agir com base no verdadeiro status, o que também torna a repetição e a reconciliação triviais.

Executar webhooks sem reconciliação

Os webhooks de web scraping são do tipo «pelo menos uma vez», não «exatamente uma vez», e não há garantia de que sejam «pelo menos uma vez» em todas as interrupções de serviço que possam ocorrer. Sem um polador de fallback, acabará por descobrir uma lacuna sem ter como saber qual é a sua dimensão.

Próximo passo

Tudo o que foi descrito acima funciona com o Geonode Scraper API: extração assíncrona com o processing_mode: "async", tarefas em lote e de rastreamento, e subscrições de webhooks com segredos rotativos e um registo de entregas pesquisável. Existe um plano gratuito de 1,5 mil pedidos por mês para criar e testar o recetor, e o pay-as-you-go a partir de 3,50 $ por cada 10 mil pedidos a partir daí. Comece com o polling, adicione webhooks quando o número de tarefas o justificar e mantenha o reconciliador em execução em qualquer dos casos.

Perguntas frequentes

O que são webhooks de web scraping e quando devo utilizá-los?

Os webhooks de web scraping são callbacks HTTP que a plataforma de scraping envia para o seu servidor quando uma tarefa assíncrona termina, para que não tenha de estar constantemente a verificar o seu estado. Utilize-os quando executar tarefas em lote ou de rastreamento, ou mais do que algumas dezenas de extrações assíncronas ao mesmo tempo. Para um pequeno número de tarefas curtas, o polling é mais simples e suficiente.

O polling de um scraper API é uma má prática?

Não. O polling é a escolha certa para pequenos volumes, scripts, notebooks e qualquer situação em que alguns segundos de latência não sejam relevantes. Torna-se um problema quando tem centenas de tarefas em execução, porque cada uma custa um pedido por intervalo de polling e, mesmo assim, não consegue reagir mais rapidamente do que esse intervalo. É aí que os webhooks de web scraping se revelam vantajosos.

Como posso verificar se um webhook veio realmente da plataforma de scraping?

Cada webhook Geonode devolve um segredo de assinatura quando é criado, e esse segredo pode ser renovado. O destinatário recalcula uma assinatura sobre o corpo bruto da solicitação com esse segredo e compara-a, em tempo constante, com o cabeçalho que a plataforma envia. Rejeite tudo o que falhar na verificação antes de analisar a carga útil e nunca registe o segredo.

O que acontece se o meu ponto de extremidade do webhook estiver inativo quando uma tarefa for concluída?

A plataforma regista uma tentativa de entrega que pode ser inspecionada através de GET /v1/webhooks/{webhook_id}/deliveries, que lista o número da tentativa (attempt_number), o estado (status) e o estado HTTP (http_status) por tentativa. O seu pipeline não deve depender apenas da reentrega. Execute um polidor de fallback que reconcilie qualquer tarefa mais antiga do que um limiar sem nenhum evento processado e obtenha o seu resultado diretamente.

Os webhooks Geonode e Scraper API estão disponíveis para o público em geral?

À data da redação deste artigo, os pontos finais dos webhooks estão documentados em docs.geonode.com, mas marcados como ainda não disponíveis em produção, e o cabeçalho de assinatura, o esquema de carga útil e a programação de novas tentativas não foram publicados. Crie o recetor de acordo com o contrato documentado, mantenha os nomes dos cabeçalhos e campos configuráveis e verifique-os em relação à documentação antes de o lançar.

Preciso tanto dos webhooks de web scraping como do polling?

Em produção, sim. Os webhooks proporcionam baixa latência e baixo volume de pedidos. Um polador de reconciliação lento deteta tarefas cuja entrega se perdeu, cujo ponto de extremidade foi rejeitado ou cuja fila foi descartada. Juntos, garantem um processamento «pelo menos uma vez», e uma tabela de idempotência indexada por job_id transforma isso num processamento efetivamente único.