Scraper API Webhooks vs Polling: Membangun Jalur Pengiriman Hasil yang Andal

Tugas pengikisan asinkron selesai sesuai jadwalnya sendiri, dan pertanyaannya adalah bagaimana kode Anda mengetahuinya. Panduan ini membandingkan webhook pengikisan web dengan polling status, kemudian membangun komponen-komponen yang Anda butuhkan untuk salah satu dari keduanya: loop polling dengan backoff dan batas maksimum, penerima FastAPI yang memverifikasi tanda tangan dan mengantrekan pekerjaan, tabel idempotensi, pemutaran ulang dari log pengiriman, serta poller cadangan yang menangkap apa pun yang terlewatkan oleh webhook. Pada akhirnya, Anda akan memiliki alur hasil yang tidak kehilangan tugas atau memprosesnya dua kali.

Ekstraksi sinkron vs asinkron

POST /v1/extract pada Geonode dan Scraper API berjalan dalam salah satu dari dua mode. Dengan processing_mode: "sync", permintaan akan terhenti dan mengembalikan kode status 200 dengan data.markdown atau data.html, sebuah objek metadata (url, http_status, duration_ms) dan tokens_charged. Dengan processing_mode: "async", permintaan mengembalikan kode status 202 bersama job_id, status: "queued", status_url, dan estimated_tokens, serta tugas dijalankan di latar belakang.

Sinkron cocok saat Anda membutuhkan jawaban secara langsung: panggilan alat agen, penangan permintaan, atau skrip satu kali. Asinkron cocok saat halaman memerlukan rendering JavaScript dan waktu tunggu yang lama, saat Anda mengirimkan banyak URL, atau saat pemanggil tidak boleh mempertahankan koneksi tetap terbuka. Batch (POST /v1/batch) dan crawl (POST /v1/crawl) selalu berupa tugas. Begitu Anda memiliki job_id, Anda harus memilih antara webhook polling dan web scraping.

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: cocok untuk beberapa tugas dan mahal untuk banyak tugas

Polling berarti memanggil GET /v1/extract/{job_id} hingga status menjadi completed atau failed. Panduan status pekerjaan mencantumkan empat status: queued, processing, completed, failed. Kesalahan yang paling sering dilakukan orang adalah menggunakan jeda tetap selama satu detik. Hal ini membebani API saat pekerjaan masih dalam antrian, dan dengan batas laju 429 (error.code "RATE_LIMITED"), loop polling mulai gagal sebelum pekerjaan itu sendiri gagal.

Gunakan backoff eksponensial dengan jitter penuh dan batasan ketat pada total waktu tunggu. Jitter menyebar kan pemantau bersamaan sehingga mereka tidak lagi mengantre pada detik yang sama.

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)

Ini tidak masalah untuk skrip yang mengirimkan sepuluh URL, tetapi tidak tepat untuk batch harian sebanyak dua puluh ribu. Dua ratus pekerjaan yang sedang berjalan dan melakukan polling setiap beberapa detik akan menghasilkan ribuan panggilan status kosong per menit, dan latensi masih dibatasi oleh interval polling.

Kapan polling lebih unggul dan kapan webhooks lebih unggul

Webhook memindahkan kompleksitas dari loop klien ke infrastruktur Anda. Jika Anda belum menjalankan titik akhir publik dan antrian, polling adalah versi awal yang tepat. Jika sudah, webhook untuk web scraping lebih hemat biaya, lebih cepat, dan dapat diskalakan berdasarkan penyelesaian, bukan berdasarkan detik.

Mendaftarkan webhook

Titik akhir webhook tercantum dalam referensi API. POST /v1/webhooks menerima url (wajib, sebuah URI), event_type (wajib, salah satu dari extract_completed, batch_completed, crawl_completed) dan description yang bersifat opsional. Respons 201 mengembalikan webhook id, url, description, is_active, event_type, cap waktu, dan secret. Kunci rahasia dikembalikan saat pembuatan dan melalui POST /v1/webhooks/{webhook_id}/rotate-secret; simpanlah di pengelola kunci rahasia pada saat itu, karena kunci inilah yang akan Anda gunakan untuk memverifikasi setiap pengiriman.

Perhatikan bahwa pada saat penulisan, dokumentasi menandai titik akhir webhook sebagai telah didokumentasikan tetapi belum tersedia di lingkungan produksi. Kontrak di bawah ini mengikuti referensi yang telah dipublikasikan; verifikasilah sebelum Anda melakukan deployment.

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

Satu webhook berlangganan ke satu event_type, sehingga sebuah pipeline yang menggunakan ekstraksi, batch, dan crawl akan mendaftarkan tiga webhook.

Penerima minimalis yang memverifikasi dan memasukkan ke antrian

Sebuah penerima webhook memiliki tiga tugas: memverifikasi tanda tangan, mengembalikan kode status 2xx secepat mungkin, dan menyerahkan peristiwa tersebut ke antrian. Jika penanganan permintaan memakan waktu lebih dari beberapa ratus milidetik, hal itu merupakan kesalahan desain, karena upaya pengiriman platform mungkin akan kedaluwarsa dan Anda akan melihat upaya ulang untuk pekerjaan yang telah Anda selesaikan.

Referensi tersebut belum mempublikasikan nama header tanda tangan, skema hashing, atau skema muatan, sehingga penerima di bawah ini menjaga hal-hal tersebut dapat dikonfigurasi dan memverifikasi HMAC-SHA256 atas badan mentah, yang merupakan pola umum. Pastikan header dan algoritma yang tepat dalam dokumentasi dan atur melalui variabel lingkungan.

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

Tiga hal penting perlu diperhatikan. Tanda tangan dihitung berdasarkan byte mentah, bukan objek JSON yang diserialisasi ulang, karena perubahan spasi kosong apa pun akan merusaknya. Penyisipan ke processed_events terjadi sebelum antrian, sehingga jika terjadi kegagalan antara keduanya, akan ada baris yang tidak diproses, yang kemudian diperbaiki oleh poller cadangan. Selain itu, handler mengembalikan kode 2xx pada duplikat alih-alih kesalahan, karena kesalahan akan memerintahkan platform untuk mencoba kembali apa yang telah dikirimkan.

Idempotensi dan pengiriman yang tidak berurutan

Setiap sistem webhook bersifat "setidaknya sekali". Gangguan jaringan, worker yang lambat, atau proses deployment bertahap yang gagal mengirimkan respons: semua hal ini dapat menyebabkan pengiriman kedua untuk tugas yang sama. Tabel processed_events dengan job_id sebagai kunci utama adalah solusi utuhnya. Konstrain unik membuat pemeriksaan duplikat menjadi atomik, yang tidak terjadi pada urutan SELECT diikuti oleh INSERT.

Urutan yang tidak teratur merupakan masalah yang berbeda. Sebuah peristiwa batch_completed mungkin tiba sebelum Anda selesai mencatat pengiriman batch, atau dua peristiwa untuk pekerjaan yang terkait mungkin tiba dalam urutan terbalik. Jangan membuat pekerja bergantung pada urutan. Satu-satunya masukan pekerja seharusnya adalah job_id; pekerja ini mengambil status pekerjaan saat ini dari API dan bertindak berdasarkan itu, yang selalu benar terlepas dari kapan peristiwa tersebut tiba.

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

Riwayat dari catatan pengiriman

GET /v1/webhooks/{webhook_id}/deliveries mencantumkan setiap upaya yang dilakukan platform. Setiap entri mencakup id, event_type, job_id, attempt_number, status, http_status, delivered_at, dan created_at, dengan page, page_size (maks. 100), serta filter opsional status sebagai parameter kueri. Nilai-nilai yang mungkin untuk status tidak tercantum dalam referensi, jadi gunakan filter http_status sebagai gantinya saat Anda ingin menemukan pengiriman yang ditolak oleh endpoint Anda.

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

Karena handle_job melewati tabel idempotensi yang sama, menjalankan ulang pekerjaan yang sebenarnya telah berhasil tidak akan menimbulkan masalah.

Poller cadangan

Log pengiriman memberikan informasi mengenai upaya pengiriman yang dilakukan platform. Log ini tidak dapat memberikan informasi mengenai peristiwa yang tidak pernah dihasilkan, atau webhook yang telah dinonaktifkan. Garis pertahanan terakhir adalah loop rekonsiliasi lambat: setiap beberapa menit, daftar pekerjaan dengan kode GET /v1/extract/jobs, dan untuk setiap pekerjaan yang lebih lama dari ambang batas Anda yang tidak memiliki baris di processed_events, prosesnya secara langsung.

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

Jalankan ini dengan frekuensi yang cukup lambat agar biayanya terjangkau, namun cukup cepat sehingga webhook yang hilang tidak menyebabkan celah data yang dapat disadari oleh pengguna Anda. Sepuluh menit adalah titik awal yang masuk akal.

Rotasi rahasia

Lakukan rotasi rahasia sesuai jadwal dan segera jika terjadi kebocoran. POST /v1/webhooks/{webhook_id}/rotate-secret akan menghasilkan secret yang baru dan, sesuai dokumentasi, membuat yang sebelumnya tidak berlaku lagi. Artinya, terdapat jeda waktu antara pergantian API dan pemuatan ulang konfigurasi oleh penerima, di mana pengiriman akan gagal verifikasi. Pastikan jeda waktu tersebut singkat: minta penerima membaca rahasia dari sumber yang dapat diperbarui tanpa perlu me-restart, lakukan rotasi, perbarui sumber tersebut, lalu periksa log pengiriman untuk mencari pengiriman yang menghasilkan kode status 401 dan ulangi pengirimannya.

Kesalahan umum

Melakukan pemrosesan di dalam penangan permintaan

Mengambil hasil, mem-parsing-nya, dan menulis ke database di dalam penangan webhook adalah kesalahan paling umum. Waktu pengiriman platform habis, sistem mencoba ulang, dan kini Anda memiliki data duplikat serta endpoint yang lambat. Cukup akui dan masukkan ke antrian, tidak lebih.

Memeriksa duplikat tanpa batasan unik

SELECT diikuti oleh INSERT menunjukkan adanya persaingan antara dua pengiriman bersamaan dari pekerjaan yang sama. Jadikan job_id sebagai kunci utama dan biarkan basis data menolak penyisipan kedua.

Mempercayai muatan peristiwa daripada status pekerjaan

Perlakukan webhook sebagai pemberitahuan, bukan sebagai data. Pekerja harus selalu mengambil ulang GET /v1/extract/{job_id} dan bertindak berdasarkan status yang sebenarnya, yang juga membuat pemutaran ulang dan rekonsiliasi menjadi sangat mudah.

Menjalankan webhook tanpa rekonsiliasi

Webhook pengikis web bersifat "setidaknya sekali", bukan "tepat sekali", dan tidak dijamin akan "setidaknya sekali" terjadi di setiap gangguan yang akan Anda alami. Tanpa poller cadangan, pada akhirnya Anda akan menemukan celah tanpa cara untuk mengetahui seberapa besar celah tersebut.

Langkah selanjutnya

Semua yang disebutkan di atas berjalan di atas Geonode Scraper API: ekstraksi asinkron dengan processing_mode: "async", pekerjaan batch dan crawling, serta langganan webhook dengan rahasia yang dapat diganti dan log pengiriman yang dapat dicari. Tersedia paket gratis sebesar 1.500 permintaan per bulan untuk membangun dan menguji penerima, serta pay-as-you-go mulai dari $3,50 per 10.000 permintaan setelah kuota gratis habis. Mulailah dengan polling, tambahkan webhook ketika jumlah pekerjaan sudah memadai, dan pastikan rekonsiliator tetap berjalan dalam kedua skenario tersebut.

Pertanyaan Umum

Apa itu webhook web scraping dan kapan sebaiknya saya menggunakannya?

Webhook web scraping adalah callback HTTP yang dikirimkan oleh platform scraper ke server Anda saat suatu tugas asinkron selesai, sehingga Anda tidak perlu terus-menerus memeriksa statusnya. Gunakan webhook ini saat Anda menjalankan pekerjaan batch atau crawling, atau lebih dari beberapa lusin ekstraksi asinkron sekaligus. Untuk beberapa pekerjaan singkat, polling lebih sederhana dan sudah cukup.

Apakah polling pada scraper API merupakan praktik yang buruk?

Tidak. Polling adalah pilihan yang tepat untuk volume kecil, skrip, notebook, dan apa pun di mana latensi beberapa detik tidak menjadi masalah. Hal ini menjadi masalah ketika Anda memiliki ratusan pekerjaan yang sedang berjalan, karena setiap pekerjaan menghabiskan satu permintaan per interval polling dan Anda tetap tidak dapat bereaksi lebih cepat dari interval tersebut. Di situlah webhook web scraping memberikan manfaatnya.

Bagaimana cara memverifikasi bahwa webhook benar-benar berasal dari platform scraper?

Setiap webhook Geonode mengembalikan rahasia penandatanganan saat dibuat, dan rahasia tersebut dapat diganti secara berkala. Penerima menghitung ulang tanda tangan atas isi permintaan mentah menggunakan rahasia tersebut dan membandingkannya dalam waktu konstan dengan header yang dikirimkan platform. Tolak apa pun yang gagal dalam pemeriksaan sebelum mengurai muatan, dan jangan pernah mencatat rahasia tersebut.

Apa yang terjadi jika titik akhir webhook saya tidak aktif saat suatu tugas selesai?

Platform mencatat upaya pengiriman yang dapat Anda periksa melalui GET /v1/webhooks/{webhook_id}/deliveries, yang mencantumkan attempt_number, status, dan http_status per upaya. Pipeline Anda tidak boleh bergantung pada pengiriman ulang saja. Jalankan poller cadangan yang menyinkronkan setiap pekerjaan yang lebih lama dari batas waktu tertentu tanpa peristiwa yang diproses, dan ambil hasilnya secara langsung.

Apakah webhook Geonode dan Scraper API sudah tersedia secara umum?

Pada saat penulisan, titik akhir webhook didokumentasikan di docs.geonode.com tetapi ditandai sebagai belum tersedia di lingkungan produksi, dan header tanda tangan, skema muatan, serta jadwal percobaan ulang belum dipublikasikan. Bangun penerima berdasarkan kontrak yang didokumentasikan, jadikan nama header dan bidang dapat dikonfigurasi, serta verifikasi sesuai dengan dokumentasi sebelum Anda meluncurkannya.

Apakah saya memerlukan webhook web scraping dan polling?

Dalam produksi, ya. Webhook memberi Anda latensi rendah dan volume permintaan rendah. Poller rekonsiliasi yang lambat menangkap pekerjaan yang pengirimannya hilang, titik akhir Anda ditolak, atau antrian Anda terlewat. Bersama-sama, keduanya memberi Anda pemrosesan setidaknya sekali, dan idempotensi yang diindeks berdasarkan job_id mengubahnya menjadi secara efektif sekali.