Scraper API Webhook’lar ve Polling: Güvenilir Bir Sonuç Akışı Oluşturma

Asenkron veri toplama işleri kendi zamanlamalarına göre tamamlanır ve asıl soru, kodunuzun bunu nasıl öğreneceğidir. Bu kılavuz, web kazıma webhook'larını durum sorgulamasıyla karşılaştırır, ardından her iki seçenek için de ihtiyacınız olan bileşenleri oluşturur: geri çekilme ve sabit sınır içeren bir sorgulama döngüsü, imzaları doğrulayan ve işleri kuyruğa alan bir FastAPI alıcısı, bir idempotans tablosu, teslimat günlüğünden yeniden oynatma ve webhook'un kaçırdığı her şeyi yakalayan bir yedek sorgulayıcı. Kılavuzun sonunda, işleri kaybetmeyen ve iki kez işlemeye almayan bir sonuç iş akışına sahip olacaksınız.

Senkron ve asenkron veri çıkarma

POST /v1/extract, Geonode ve Scraper API üzerinde iki moddan birinde çalışır. processing_mode: "sync" ile istek bloke olur ve data.markdown veya data.html ile 200 kodunu döndürür; bir metadata nesnesi (url, http_status, duration_ms) ve tokens_charged döndürür. processing_mode: "async" ile 202 kodunu döndürür; job_id, status: "queued", status_url ve estimated_tokens ile iş arka planda çalışır.

Cevabı satır içi olarak almanız gerektiğinde senkron (Sync) doğru seçimdir: bir ajan aracı çağrısı, bir istek işleyicisi, tek seferlik bir komut dosyası. Sayfanın JavaScript ile işlenmesi ve uzun bir bekleme süresi gerektirdiğinde, çok sayıda URL gönderdiğinizde veya çağıran tarafın bağlantıyı açık tutmaması gerektiğinde asenkron (Async) doğru seçimdir. Toplu iş (POST /v1/batch) ve tarama (POST /v1/crawl) her zaman işlerdir. job_id kullandığınız anda, yoklama ve web kazıma webhook'ları arasında seçim yapmanız gerekir.

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: Bazı işler için uygun, çoğu iş için pahalı

Polling, status değeri completed veya failed olana kadar GET /v1/extract/{job_id}’yi çağırmak anlamına gelir. İş durumu kılavuzunda dört durum listelenmiştir: queued, processing, completed, failed. Çoğu kişinin yaptığı hata, sabit bir saniyelik bekleme süresidir. Bu, iş kuyruğa alınmışken API'yi aşırı yükler ve 429 oran sınırı (error.code "RATE_LIMITED") nedeniyle, iş başarısız olmadan önce sorgulama döngüsü başarısız olmaya başlar.

Tam jitter ve toplam bekleme süresine sabit bir sınır ile üstel geri çekilme kullanın. Jitter, eşzamanlı sorgulayıcıları dağıtır, böylece aynı saniyede sıraya girmelerini engeller.

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)

Bu, on URL gönderen bir komut dosyası için uygundur, ancak her gece yirmi binlik bir toplu iş için yanlıştır. Birkaç saniyede bir sorgulama yapan iki yüz iş, dakikada binlerce boş durum çağrısı anlamına gelir ve gecikme süresi yine de sorgulama aralığıyla sınırlıdır.

Ne zaman sorgulama, ne zaman webhook'lar daha avantajlıdır?

Webhook'lar, karmaşıklığı istemci döngüsünden altyapınıza taşır. Henüz halka açık bir uç nokta ve kuyruk çalıştırmıyorsanız, sorgulama doğru ilk seçenektir. Eğer çalıştırıyorsanız, web kazıma webhook'ları daha ucuz, daha hızlıdır ve saniye yerine tamamlanma sayısına göre ölçeklenir.

Bir webhook kaydetme

Webhook uç noktaları API referansında listelenmiştir. POST /v1/webhooks, url (gerekli, bir URI), event_type (gerekli, extract_completed, batch_completed, crawl_completed) ve isteğe bağlı bir description alır. 201 yanıtı, webhook id, url, description, is_active, event_type webhook'larını, zaman damgalarını ve bir secret'yu döndürür. Gizli anahtar, oluşturulduğunda ve POST /v1/webhooks/{webhook_id}/rotate-secret tarafından döndürülür; her teslimatı doğrulamak için kullanacağınız bu anahtarı o anda bir gizli anahtar yöneticisinde saklayın.

Belgelerin yazıldığı tarihte, webhook uç noktalarının belgelenmiş ancak henüz üretim ortamında kullanıma açık olmadığına dikkat edin. Aşağıdaki sözleşme, yayınlanan referansa uygundur; dağıtım yapmadan önce bunu doğrulayın.

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

Bir webhook, bir event_type'ya abone olur; bu nedenle, ayıklama, toplu iş ve tarama işlemlerini kullanan bir boru hattı üç tane kaydeder.

Doğrulama ve kuyruğa ekleme işlevlerini yerine getiren minimal bir alıcı

Bir webhook alıcısının tam olarak üç görevi vardır: imzayı doğrulamak, mümkün olduğunca hızlı bir şekilde 2xx yanıtı döndürmek ve olayı bir kuyruğa aktarmak. İstek işleyicide birkaç yüz milisaniyeden daha yavaş olan her şey bir tasarım hatasıdır, çünkü platformun teslimat denemesi zaman aşımına uğrayabilir ve zaten yaptığınız iş için yeniden denemelerle karşılaşabilirsiniz.

Referans, imza başlığı adını, karma şemasını veya yük şemasını henüz yayınlamamıştır; bu nedenle aşağıdaki alıcı, bunları yapılandırılabilir tutar ve yaygın bir uygulama olan ham gövde üzerinde HMAC-SHA256'yı doğrular. Belgelerde tam başlığı ve algoritmayı teyit edin ve bunları ortam değişkenleri aracılığıyla ayarlayın.

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

Üç ayrıntı önemlidir. İmza, yeniden serileştirilmiş bir JSON nesnesi üzerinden değil, ham baytlar üzerinden hesaplanır; çünkü herhangi bir boşluk değişikliği imzasını bozar. processed_events'ya ekleme işlemi, kuyruğa ekleme işleminden önce gerçekleşir; bu nedenle, ikisi arasında bir çökme olması durumunda işlenmemiş bir satır kalır ve bu durum daha sonra yedek sorgulayıcı tarafından düzeltilir. Ayrıca, işleyici bir yineleme durumunda hata yerine 2xx kodunu döndürür; çünkü bir hata, platforma zaten teslim edilmiş olan işlemi yeniden denemesini söyler.

İdempotans ve sırası bozuk teslimatlar

Herhangi bir webhook sistemi “en az bir kez” ilkesine göre çalışır. Ağ kesintileri, yavaş çalışan bir işleyici, yanıtı düşüren aşamalı bir dağıtım: bunların hepsi aynı iş için ikinci bir teslimata yol açar. Birincil anahtarı job_id olan processed_events tablosu, sorunun tamamını çözer. Benzersizlik kısıtlaması, yinelenen kontrolü atomik hale getirir; buna karşın, SELECT'nun ardından INSERT gelmesi durumunda bu durum söz konusu değildir.

Sıranın bozulması ise farklı bir sorundur. Bir batch_completed olayı, toplu gönderimin kaydını tamamlamadan önce gelebilir veya ilgili işlere ait iki olay ters sırayla gelebilir. İşleyicinin sıraya bağlı kalmasını sağlamayın. İşleyicinin tek girişi bir job_id olmalıdır; bu, API'dan işin mevcut durumunu alır ve buna göre hareket eder; bu durum, olayın ne zaman geldiğine bakılmaksızın her zaman doğrudur.

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

Teslimat günlüğünden tekrar görüntüleme

GET /v1/webhooks/{webhook_id}/deliveries, platformun yaptığı her denemeyi listeler. Her bir öğe, id, event_type, job_id, attempt_number, status, http_status, delivered_at ve created_at değerlerini içerir; sorgu parametreleri olarak ise page, page_size (maks. 100) ve isteğe bağlı status filtresi kullanılır. Referans belgesinde olası status değerleri listelenmemiştir; bu nedenle, uç noktanızın reddettiği teslimatları bulmak istediğinizde bunun yerine http_status'ya göre filtreleme yapın.

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 aynı idempotans tablosundan geçtiği için, aslında başarılı olan bir işi yeniden çalıştırmanın bir zararı yoktur.

Yedek sorgulayıcı

Teslimat günlüğü, platformun gerçekleştirmeye çalıştığı teslimatlar hakkında bilgi verir. Hiç oluşturulmamış olaylar veya devre dışı bırakılmış bir webhook hakkında bilgi vermez. Son savunma hattı, yavaş bir mutabakat döngüsüdür: birkaç dakikada bir, GET /v1/extract/jobs'da bulunan işleri listeleyin ve eşik değerinizden daha eski olup processed_events'da satırı bulunmayan işleri doğrudan işleyin.

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

Bunu, maliyet açısından yeterince düşük ve kaybolan bir webhook'un kullanıcılarınızın fark edeceği bir veri boşluğuna dönüşmeyecek kadar hızlı bir sıklıkta çalıştırın. On dakika, makul bir başlangıç noktasıdır.

Gizli anahtarın değiştirilmesi

Gizli anahtarı belirli bir zamanlamaya göre ve sızıntı durumunda derhal değiştirin. POST /v1/webhooks/{webhook_id}/rotate-secret, yeni bir secret döndürür ve belgelere göre bir öncekini geçersiz kılar. Bu, API'nin değiştirilmesi ile alıcınızın yapılandırmasını yeniden yüklemesi arasında, teslimatların doğrulamadan geçemeyeceği bir zaman aralığı olduğu anlamına gelir. Bu aralığı kısa tutun: alıcının, yeniden başlatmaya gerek kalmadan yenileyebileceği bir kaynaktan gizli anahtarı okumasını sağlayın, anahtarı değiştirin, kaynağı güncelleyin, ardından teslimat günlüğünü kontrol ederek 401 hatasıyla sonuçlanan işlemleri tespit edin ve bunları tekrar gönderin.

Sık yapılan hatalar

İstek işleyicisi içinde iş yapmak

Webhook işleyicisi içinde sonucu almak, ayrıştırmak ve veritabanına yazmak en yaygın hatadır. Platformun teslim süresi dolduğunda yeniden deneme yapar ve sonuç olarak yinelenen kayıtlar ve yavaş bir uç nokta ile karşı karşıya kalırsınız. Onaylayın ve kuyruğa ekleyin, başka bir şey yapmayın.

Benzersizlik kısıtlaması olmadan yinelenen verileri kontrol etme

SELECT ve ardından INSERT, aynı işin iki eşzamanlı teslimatı arasında bir yarışa neden olur. job_id'yu birincil anahtar yapın ve veritabanının ikinci eklemeyi reddetmesine izin verin.

İş durumuna göre olay yüküne güvenme

Webhook'u veri olarak değil, bir bildirim olarak değerlendirin. İşleyici her zaman GET /v1/extract/{job_id}'yu yeniden almalı ve gerçek status'ya göre hareket etmelidir; bu da tekrar oynatma ve uzlaştırma işlemlerini çok basit hale getirir.

Uzlaşma yapmadan webhook'ları çalıştırma

Web kazıma webhook'ları "en az bir kez" (at-least-once) çalışır, "tam olarak bir kez" (exactly-once) değil; ayrıca, karşılaşabileceğiniz her kesinti durumunda "en az bir kez" çalışacakları garanti edilmez. Yedek bir sorgulama mekanizması olmadan, sonunda ne kadar büyük olduğunu bilmenin imkânı olmayan bir boşlukla karşılaşacaksınız.

Sonraki adım

Yukarıdaki her şey Geonode Scraper API üzerinde çalışır: processing_mode: "async" ile asenkron veri çıkarma, toplu iş ve tarama işleri ile döndürülebilir gizli anahtarlar ve sorgulanabilir teslimat günlüğü içeren webhook abonelikleri. Alıcıyı oluşturmak ve test etmek için aylık 1.5K isteklik ücretsiz bir kademesi mevcuttur; bunun ardından pay-as-you-go, 10K istek başına 3,50 $'dan başlayan fiyatlarla sunulur. Polling ile başlayın, iş sayısı gerektirdiğinde webhook'ları ekleyin ve her durumda uzlaştırıcıyı çalışır durumda tutun.

SSS

Web kazıma webhook’ları nedir ve ne zaman kullanmalıyım?

Web kazıma webhook’ları, asenkron bir iş tamamlandığında kazıma platformunun sunucunuza gönderdiği HTTP geri çağrılarıdır; böylece işin durumunu sürekli sorgulamanıza gerek kalmaz. Toplu işler veya tarama işleri çalıştırdığınızda ya da aynı anda birkaç düzineyi aşan asenkron veri çıkarma işlemi gerçekleştirdiğinizde bunları kullanın. Birkaç kısa iş için ise sorgulama yöntemi daha basit ve yeterlidir.

scraper API sorgulama yöntemi kötü bir uygulama mıdır?

Hayır. Polling, küçük hacimli işler, komut dosyaları, notebook'lar ve birkaç saniyelik gecikmenin önemi olmadığı her durum için doğru seçimdir. Yüzlerce işin aynı anda yürütülmesi durumunda sorun haline gelir; çünkü her bir iş, polling aralığı başına bir istek maliyeti oluşturur ve yine de bu aralıktan daha hızlı tepki veremezsiniz. İşte bu noktada web kazıma webhook'ları devreye girer.

Bir webhook'un gerçekten kazıyıcı platformundan geldiğini nasıl doğrularım?

Her Geonode webhook'u oluşturulduğunda bir imzalama gizli anahtarı döndürür ve bu anahtar değiştirilebilir. Alıcı, bu gizli anahtarla ham istek gövdesi üzerinde bir imza yeniden hesaplar ve bunu, platformun gönderdiği başlıkla sabit sürede karşılaştırır. Yükü ayrıştırmadan önce kontrolü geçemeyen her şeyi reddedin ve gizli anahtarı asla günlüğe kaydetmeyin.

Bir iş tamamlandığında webhook uç noktam çalışmıyorsa ne olur?

Platform, GET /v1/webhooks/{webhook_id}/deliveries komutu aracılığıyla inceleyebileceğiniz bir teslimat denemesi kaydeder; bu kayıt, her deneme için attempt_number, status ve http_status bilgilerini listeler. İş akışınız yalnızca yeniden teslimata bağlı olmamalıdır. Eşiğin üzerindeki, işlenmemiş olay içeren tüm işleri uzlaştıran bir yedek sorgulayıcı çalıştırın ve sonucunu doğrudan alın.

Geonode ve Scraper API webhook'ları genel kullanıma açık mı?

Bu yazının yazıldığı tarihte, webhook uç noktaları docs.geonode.com adresinde belgelenmiştir ancak üretim ortamında henüz kullanıma açık olmadığı belirtilmiştir; imza başlığı, yük şeması ve yeniden deneme zamanlaması yayınlanmamıştır. Alıcıyı belgelenen sözleşmeye göre oluşturun, başlık ve alan adlarını yapılandırılabilir tutun ve yayınlamadan önce belgelerle karşılaştırarak doğrulayın.

Hem web kazıma webhook'larına hem de yoklamaya ihtiyacım var mı?

Üretim ortamında, evet. Webhook'lar size düşük gecikme süresi ve düşük istek hacmi sağlar. Yavaş bir mutabakat sorgulayıcısı, teslimatı kaybolan, uç noktanız tarafından reddedilen veya kuyruğunuzdan düşen işleri yakalar. Bunlar birlikte size "en az bir kez" işleme olanağı sağlar ve job_id anahtarlı bir idempotans tablosu bunu "etkin olarak bir kez" işleme dönüştürür.