Geonode logo
Geonode Team

Geonode Team

更新于：2026年10月16日

发布于：2026年10月16日

Scraper API Webhooks 与轮询：构建可靠的結果处理流程

异步抓取任务会按照自己的时间表完成，问题在于你的代码如何得知这一点。 本指南将网络爬虫的 webhook 与状态轮询进行了对比，随后构建了实现这两种方式所需的组件：一个带有重试策略和硬性上限的轮询循环、一个用于验证签名并将任务入队到队列的 FastAPI 接收器、一个幂等性表、基于交付日志的重放功能，以及一个用于捕获 webhook 遗漏任务的备用轮询器。 完成后，您将拥有一条既不会丢失任务，也不会重复处理任务的结果处理管道。

异步抓取任务会按照自己的时间表完成，问题在于你的代码如何得知这一点。 本指南将网络爬虫的 webhook 与状态轮询进行了对比，随后构建了实现这两种方式所需的组件：一个带有重试策略和硬性上限的轮询循环、一个用于验证签名并将任务入队到队列的 FastAPI 接收器、一个幂等性表、基于交付日志的重放功能，以及一个用于捕获 webhook 遗漏任务的备用轮询器。 完成后，您将拥有一条既不会丢失任务，也不会重复处理任务的结果处理管道。

同步与异步提取

POST /v1/extract 在 Geonode 和 Scraper API 上运行时，有两种模式之一。 当与 processing_mode: "sync" 配合使用时，请求会阻塞并返回 200 状态码；而当与 data.markdown 或 data.html 配合使用时，则返回一个 metadata 对象（url、http_status、 duration_ms）以及 tokens_charged。 对于 processing_mode: "async"，当返回 job_id、status: "queued"、status_url 和 estimated_tokens 时，会返回 202 状态码，且任务在后台运行。

当您需要立即获得结果时，应选择同步方式：例如代理工具调用、请求处理程序或一次性脚本。当页面需要 JavaScript 渲染且等待时间较长、提交大量 URL，或者调用方不应保持连接打开时，应选择异步方式。 批处理（POST /v1/batch）和爬取（POST /v1/crawl）始终是任务。一旦遇到 job_id，就必须在轮询和网页抓取 webhook 之间做出选择。

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

轮询：对少数任务来说可行，但对许多任务来说成本过高

轮询是指不断调用 GET /v1/extract/{job_id}，直到 status 的状态变为 completed 或 failed 为止。 作业状态指南列出了四种状态：queued、processing、completed、failed。 大多数人常犯的错误是使用固定的1秒延迟。这会在任务排队期间频繁调用API，而由于存在429速率限制（error.code "RATE_LIMITED"），轮询循环会在任务完成前就已开始失败。

请使用带完全抖动且对总等待时间设有硬性上限的指数退避算法。抖动能分散并发轮询者的时间间隔，使其不再集中在同一秒进行轮询。

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)

对于提交十个 URL 的脚本来说这没问题，但对于每晚处理两万个 URL 的批处理任务则不合适。两百个正在处理中的任务每隔几秒进行轮询，意味着每分钟会有数千次空状态调用，而延迟仍然受轮询间隔的限制。

何时应采用轮询，何时应采用 webhook

关注点轮询Web 抓取 webhook
正在处理的任务较少（约50个以下）数百至数千
任务持续时间秒分钟（批处理、爬取、渲染页面）
响应延迟轮询间隔近乎即时
请求量每个任务每次轮询一次每次完成一次
必须运行的基础设施无公共 HTTPS 端点、队列、幂等性存储
故障模式可控超时未送达或重复送达
可调试性非常简单需要交付日志和重放功能
最适合脚本、笔记本、低流量管道、批处理、爬取

Webhook 将复杂性从客户端循环转移到了您的基础设施中。如果您尚未运行公共端点和队列，轮询是正确的第一版方案。如果您已经运行了这些组件，那么用于网络爬虫的 Webhook 成本更低、速度更快，且其扩展性取决于任务完成数量而非时间。

注册 Webhook

Webhook 端点在 API 参考中列出。POST /v1/webhooks 接受 url（必填，一个 URI）、event_type（必填，取自 extract_completed、 batch_completed、crawl_completed）中的一个，以及可选的 description。201 响应将返回 webhook 的 id、url、 description、is_active、event_type、时间戳以及一个 secret。 密钥在创建时及通过 POST /v1/webhooks/{webhook_id}/rotate-secret 返回；请立即将其存储在密钥管理器中，因为您将使用它来验证每次交付。

请注意，在撰写本文时，文档中将这些 webhook 端点标记为“已记录但尚未在生产环境中可用”。 下面的合约遵循已发布的参考文档；部署前请务必进行验证。

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

一个 webhook 订阅一个 event_type，因此使用提取、批处理和爬取功能的管道会注册三个。

一个用于验证并入队的简易接收器

Webhook 接收器只有三项任务：验证签名、尽快返回 2xx 状态码，并将事件交由队列处理。如果请求处理程序的响应时间超过几百毫秒，那就是设计错误，因为平台的投递尝试可能会超时，导致你为已经处理过的工作再次进行重试。

参考文档尚未公布签名标头名称、哈希方案或有效载荷模式，因此下面的接收器将这些参数设为可配置，并针对原始正文验证 HMAC-SHA256 签名——这是常见的模式。请在文档中确认确切的标头和算法，并通过环境变量进行设置。

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

有三个细节需要注意。签名是基于原始字节计算的，而不是基于重新序列化的 JSON 对象，因为任何空格的变化都会导致签名失效。 将数据插入 processed_events 的操作发生在入队之前，因此如果两者之间发生崩溃，会留下一个没有处理的行，该行将由备用轮询器稍后修复。此外，处理程序在遇到重复数据时会返回 2xx 状态码而非错误，因为错误会指示平台重试已经交付的操作。

幂等性与乱序投递

任何 webhook 系统都遵循“至少一次”原则。网络波动、处理程序运行缓慢、滚动部署导致响应丢失：所有这些情况都会导致同一任务被重复投递。 以 job_id 为主键的 processed_events 表便是完整的解决方案。唯一约束使重复检查成为原子操作，而 SELECT 之后紧接着 INSERT 的情况则无法做到这一点。

顺序错乱则是另一个问题。可能在您完成批量提交的记录之前，batch_completed事件就已到达；或者两个相关任务的事件以相反的顺序到达。 请勿让工作进程依赖事件顺序。工作进程的唯一输入应为 job_id；它从 API 获取作业的当前状态并据此采取行动，无论事件何时到达，该状态始终是真实的。

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

交付日志回放

GET /v1/webhooks/{webhook_id}/deliveries 列出了平台所做的每次尝试。每项记录都包含 id、event_type、job_id、attempt_number、status、 http_status、delivered_at 和 created_at，并以 page、page_size（最多 100 个）以及可选的 status 过滤器作为查询参数。 由于参考文档中未列出 status 的可能取值，因此当您需要查找被端点拒绝的交付时，请改用 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

由于 handle_job 会经过相同的幂等性表，因此重放一个实际上已成功执行的作业是无害的。

备用轮询器

投递日志会记录平台尝试进行的投递操作。但它无法告知您那些从未生成的事件，也无法告知您已被停用的 webhook。 最后的防线是一个低频对账循环：每隔几分钟，列出包含在 GET /v1/extract/jobs 中的任务，对于任何超过阈值且在 processed_events 中没有对应行的任务，直接进行处理。

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

请以足够慢的频率运行此循环，以确保成本低廉；同时也要足够快，以免因 Webhook 丢失而导致用户察觉到的数据缺口。每 10 分钟是一个合理的起始点。

密钥轮换

应按计划轮换密钥，若发生泄露则应立即轮换。POST /v1/webhooks/{webhook_id}/rotate-secret 会返回新的 secret，并且根据文档说明，会使之前的密钥失效。 这意味着在 API 切换与接收端重新加载配置之间存在一个时间窗口，在此期间的投递将无法通过验证。请尽量缩短该时间窗口：让接收端从无需重启即可刷新源处读取密钥，完成轮换后更新该源，然后检查投递日志中是否存在返回 401 状态码的记录，并重新发送这些消息。

常见错误

在请求处理程序中执行操作

在 webhook 处理程序中获取结果、解析数据并写入数据库是最常见的失败原因。当平台的交付超时并触发重试时，就会导致数据重复且端点响应变慢。只需确认接收并将其加入队列，仅此而已。

未设置唯一约束时检查重复项

当先收到 SELECT 再收到 INSERT 时，说明同一任务的两个并发交付之间存在竞争条件。请将 job_id 设为主键，并让数据库拒绝第二次插入操作。

优先信任事件负载而非任务状态

将 webhook 视为通知，而非数据。处理程序应始终重新获取 GET /v1/extract/{job_id}，并根据真实的 status 采取行动，这也会使重放和对账变得非常简单。

未进行对账的 Webhook 运行

网络爬虫型 Webhook 遵循“至少一次”原则，而非“精确一次”原则，且无法保证在每次系统中断期间都能实现“至少一次”的处理。如果没有备用轮询器，最终会出现数据缺口，且无法确定缺口的大小。

下一步

上述所有内容均基于 Geonode Scraper API 运行：使用 processing_mode: "async" 进行异步提取、批处理和爬取任务，以及支持可轮换密钥和可查询交付日志的 webhook 订阅。 提供每月 1.5K 次请求的免费配额，用于构建和测试接收器；超出后，pay-as-you-go 的费用为每 10K 次请求 3.50 美元起。建议先从轮询开始，待任务数量达到一定规模后再添加 webhook，无论采用哪种方式，都应保持对账程序持续运行。

常见问题

什么是网络爬虫 webhook，何时应该使用它们？

网络爬虫 webhook 是爬虫平台在异步任务完成时发送到您服务器的 HTTP 回调，因此您无需反复查询任务状态。 在运行批处理或爬取任务，或同时进行数十个以上异步提取任务时，请使用它们。对于少量短任务，轮询方式更简单且已足够。

轮询 scraper API 是否属于不良实践？

不是。对于小批量任务、脚本、笔记本以及任何几秒延迟无伤大雅的情况，轮询是正确的选择。当您有数百个正在执行的任务时，轮询就会成为问题，因为每次轮询都会消耗一个请求，而且您的响应速度仍无法快于该轮询间隔。这就是网络爬虫 webhook 发挥作用的地方。

如何验证 Webhook 确实来自爬虫平台？

每个 Geonode Webhook 在创建时都会返回一个签名密钥，且该密钥可以轮换。 接收方会使用该密钥对原始请求正文重新计算签名，并在常数时间内将其与平台发送的头部进行比对。在解析有效载荷之前，请拒绝任何验证失败的请求，并且切勿记录该密钥。

如果任务完成时我的 webhook 端点不可用，会发生什么？

平台会记录一次投递尝试，您可以通过 GET /v1/webhooks/{webhook_id}/deliveries 进行查看，该接口会按每次尝试列出 attempt_number、status 和 http_status。 您的管道不应仅依赖于重发机制。请运行一个备用轮询程序，对超过阈值且未处理任何事件的旧任务进行核对，并直接获取其结果。

Geonode 和 Scraper API Webhook 是否已正式发布？

在撰写本文时，这些 webhook 端点已在 docs.geonode.com 上有相关文档，但标注为尚未在生产环境中可用，且签名头、有效负载架构和重试计划尚未发布。 请根据文档中描述的契约构建接收器，确保标头和字段名称可配置，并在发布前对照文档进行验证。

我是否需要同时使用网页抓取 webhook 和轮询？

在生产环境中，需要两者兼备。Webhook 能提供低延迟和低请求量。而慢速的对账轮询程序可捕获因投递丢失、端点拒绝或队列丢弃而未处理的任务。两者结合可实现“至少一次”处理，而以 job_id 为键的幂等性表则将其转化为“实际仅一次”处理。