Scraper API Webhookとポーリングの比較：信頼性の高い結果パイプラインの構築

非同期のスクレイピングジョブは独自のスケジュールで完了しますが、問題はコードがそれをどのように検知するかです。 このガイドでは、WebスクレイピングのWebhookとステータスポーリングを比較し、それぞれに必要な要素を構築していきます。具体的には、バックオフと上限設定を備えたポーリングループ、署名を検証してジョブをキューに追加するFastAPIレシーバー、イデポテンシーテーブル、配信ログからのリプレイ、そしてWebhookが見逃したものを捕捉するフォールバックポーラーです。 このガイドを終える頃には、ジョブを紛失したり二重に処理したりしない結果処理パイプラインが完成しているはずです。

同期抽出と非同期抽出

Geonode 上の POST /v1/extract および Scraper API は、2 つのモードのいずれかで動作します。 processing_mode: "sync" の場合、リクエストはブロックされ、data.markdown または data.html に対しては 200 を返し、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"]

ポーリング：ジョブ数が少ない場合は問題ないが、多い場合はコストがかかる

ポーリングとは、statusがcompletedまたはfailedになるまで、GET /v1/extract/{job_id}を呼び出し続けることを指します。 ジョブステータスガイドには、queued、processing、completed、failedの4つの状態が記載されています。 多くの人が犯す間違いは、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)

これは、10個のURLを送信するスクリプトには問題ありませんが、2万件のURLを処理する毎晩のバッチ処理には不適切です。数秒ごとにポーリングを行う200件のジョブが実行中だと、1分あたり数千回の空のステータス呼び出しが発生し、レイテンシは依然としてポーリング間隔によって制限されます。

ポーリングが適している場合と、Webhookが適している場合

Webhookは、クライアントループからインフラストラクチャへと複雑さを移行させます。まだパブリックエンドポイントやキューを運用していない場合は、ポーリングが適切な最初の選択肢となります。すでに運用している場合は、Webスクレイピング用の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

1つのWebhookは1つのevent_typeを購読するため、抽出、バッチ、クロールを使用するパイプラインでは3つが登録されます。

検証とキューへの格納を行う最小限のレシーバー

Webhookレシーバーには、正確に3つの役割があります。それは、署名を検証すること、できるだけ速やかに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"}

重要な点が3つあります。署名は、再シリアライズされたJSONオブジェクトではなく、生のバイト列に対して計算されます。これは、空白の変更があると署名が破綻するためです。 processed_eventsへの挿入はエンキューの前に実行されるため、この2つの処理の間にクラッシュが発生すると、未処理の行が残ってしまいますが、これは後でフォールバックポーラーによって修正されます。また、ハンドラーは重複が発生した場合、エラーではなく2xxを返します。これは、エラーを返すとプラットフォームがすでに配信済みの処理を再試行してしまうためです。

冪等性と順不同の配信

どのWebhookシステムも「少なくとも1回」の配信を保証します。ネットワークの一時的な障害、ワーカーの処理遅延、レスポンスが失われるローリングデプロイなど、これらすべてが同じジョブの2回目の配信を引き起こします。 job_idを主キーとするprocessed_eventsテーブルが、この問題を完全に解決します。一意制約により、重複チェックがアトミックに行われますが、SELECTの後にINSERTが続くようなケースでは、そうはなりません。

順序の乱れは別の問題です。バッチ送信の記録が完了する前に batch_completed イベントが到着したり、関連するジョブの 2 つのイベントが逆の順序で到着したりする可能性があります。 ワーカーを順序に依存させてはいけません。ワーカーの唯一の入力は 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 がある場合、同じジョブの 2 つの並行した配信間で競合が発生します。job_id を主キーとし、2 回目の挿入をデータベースが拒否するようにしてください。

ジョブステータスよりもイベントペイロードを優先する

Webhookをデータではなく通知として扱います。ワーカーは常にGET /v1/extract/{job_id}を再取得し、実際のstatusに基づいて処理を行うべきです。これにより、リプレイや調整も容易になります。

調整を行わずに Webhook を実行する場合

Web スクレイピング型の Webhook は「少なくとも 1 回」の保証しかなく、「正確に 1 回」ではありません。また、発生しうるあらゆるサービス停止において「少なくとも 1 回」が保証されるわけでもありません。フォールバック用のポーラーがない場合、最終的にはデータの欠落に気づくことになりますが、その規模を把握する手段はありません。

次のステップ

上記のすべては、Geonode Scraper API上で実行されます。これには、processing_mode: "async"による非同期抽出、バッチおよびクロールジョブ、さらにローテーション可能なシークレットとクエリ可能な配信ログを備えたWebhookサブスクリプションが含まれます。 レシーバーの構築およびテスト用に、月間1.5Kリクエストまでの無料枠が用意されており、それを超えるとpay-as-you-goが10Kリクエストあたり3.50ドルから利用可能です。まずはポーリングから始め、ジョブ数に応じて必要になった時点でWebhookを追加し、いずれの場合もレコンサイラーは稼働させ続けてください。

よくある質問

ウェブスクレイピングのウェブフックとは何ですか？また、どのような場合に使用すべきですか？

ウェブスクレイピングのウェブフックとは、非同期ジョブが完了した際にスクレイパープラットフォームがサーバーに送信するHTTPコールバックのことです。これにより、ステータスを都度確認する必要がなくなります。 バッチジョブやクロールジョブを実行する場合、あるいは一度に数十件以上の非同期抽出を行う場合に使用してください。少数の短いジョブであれば、ポーリングの方がシンプルで十分です。

scraper APIのポーリングは悪い慣行ですか？

いいえ。ポーリングは、処理量が少なく、スクリプトやノートブックなど、数秒の遅延が問題にならない場合には適切な選択です。ただし、進行中のジョブが数百件ある場合、各ジョブごとにポーリング間隔ごとに1回のリクエストが発生し、その間隔よりも速く反応することはできないため、問題となります。そのような場合に、ウェブスクレイピングのウェブフックが威力を発揮します。

Webhookが本当にスクレイパープラットフォームから送信されたものであることを、どのように確認すればよいですか？

各Geonode Webhookは、作成時に署名用シークレットを返し、このシークレットはローテーション可能です。 受信側は、そのシークレットを使用して生のリクエスト本文に対する署名を再計算し、プラットフォームから送信されたヘッダーと定数時間で照合します。チェックに失敗したものはすべて、ペイロードを解析する前に拒否し、シークレットは決してログに記録しないでください。

ジョブが完了した際に、Webhookエンドポイントがダウンしていた場合はどうなりますか？

プラットフォームは配信試行を記録しており、GET /v1/webhooks/{webhook_id}/deliveries を通じて確認できます。このリクエストでは、試行ごとの attempt_number、status、および http_status が一覧表示されます。 パイプラインは再配信のみに依存してはいけません。閾値より古いジョブで、処理済みのイベントがないものを照合するフォールバックポーラーを実行し、その結果を直接取得するようにしてください。

Geonode および Scraper API ウェブフックは一般提供されていますか？

本稿執筆時点では、Webhookエンドポイントは docs.geonode.com に記載されていますが、本番環境ではまだ利用不可とされており、シグネチャヘッダー、ペイロードスキーマ、再試行スケジュールは公開されていません。 ドキュメントに記載された契約に基づいてレシーバーを構築し、ヘッダーとフィールド名は設定可能にしておき、リリース前にドキュメントと照らし合わせて確認してください。

ウェブスクレイピング用ウェブフックとポーリングの両方が必要ですか？

本番環境では、はい、両方必要です。Webhookは低遅延かつリクエスト数が少ないという利点があります。一方、低速なレコンサイレメントポーラーは、配信に失敗したジョブ、エンドポイントによって拒否されたジョブ、またはキューから削除されたジョブを捕捉します。これらを併用することで「少なくとも1回」の処理が保証され、job_idをキーとする冪等性テーブルによって、実質的に「1回」の処理が実現されます。