Inkrementelles Web-Crawling: Geänderte Seiten aktualisieren, ohne alles erneut zu crawlen

Beim inkrementellen Web-Crawling werden nur die Seiten erneut abgerufen, die sich geändert haben, nicht die gesamte Website. Am Ende dieses Leitfadens verfügen Sie über ein Python-Modul, das URLs aus einer Sitemap oder dem Geonode-Endpunkt /v1/map ermittelt, diese normalisiert, den Status pro URL in SQLite speichert, anhand von ETag und Inhalts-Hashes feststellt, was sich geändert hat, die Intervalle für erneute Besuche anpasst, Löschungen erkennt und die geänderten Seiten über /v1/extract oder /v1/batch abruft. Geonode ruft die Seiten ab; Ihr Skript verwaltet den Status und den Vergleich.

Warum vollständige Neuindizierungen das Budget verschwenden

Eine Website mit 50.000 Seiten, auf der sich 2 % pro Woche ändern, benötigt etwa 1.000 Abrufe pro Woche, um auf dem neuesten Stand zu bleiben. Ein wöchentlicher vollständiger Crawling-Durchlauf verbraucht 50.000. Das entspricht einem 50-fachen Unterschied bei API-Anfragen, Bandbreite und der Zeit, die Ihre Pipeline durch E/A-Vorgänge blockiert.

Die Verschwendung verstärkt sich bei JavaScript-gerenderten Seiten, bei denen jeder Abruf langsamer und aufwendiger ist, sowie bei Websites mit aggressiven Ratenbeschränkungen. Inkrementelles Web-Crawling kehrt das Problem um: Man speichert pro URL eine kleine Menge an Statusinformationen und lässt sich von diesem Status leiten, wohin die nächste Anfrage gerichtet werden soll.

Das untenstehende Modul umfasst etwa 120 Zeilen. Jeder Abschnitt fügt ein Element hinzu. Installiere die Abhängigkeit mit pip install requests (httpx funktioniert identisch, falls du es bevorzugst).

URLs ermitteln

Zwei Quellen, in der Reihenfolge ihrer Präferenz. Eine Sitemap liefert Ihnen URLs sowie optional ein lastmod -Feld, das das einfachste Signal für Änderungen darstellt. Wenn eine Website keine brauchbare Sitemap hat, ermittelt POST /v1/map URLs durch die Kombination von Sitemap-Analyse und HTML-Link-Extraktion und gibt jede URL zusammen mit ihrer source zurück.

import os
import xml.etree.ElementTree as ET
import requests

API = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_API_KEY"]}
SM_NS = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}

def urls_from_sitemap(sitemap_url: str, session: requests.Session) -> list[tuple[str, str | None]]:
    """Return (loc, lastmod) pairs. Follows sitemap index files one level."""
    root = ET.fromstring(session.get(sitemap_url, timeout=30).content)
    out: list[tuple[str, str | None]] = []
    for sm in root.findall("sm:sitemap", SM_NS):          # sitemap index
        out.extend(urls_from_sitemap(sm.findtext("sm:loc", namespaces=SM_NS), session))
    for u in root.findall("sm:url", SM_NS):
        out.append((u.findtext("sm:loc", namespaces=SM_NS),
                    u.findtext("sm:lastmod", namespaces=SM_NS)))
    return out

def urls_from_map(site_url: str, session: requests.Session) -> list[tuple[str, None]]:
    body = {"url": site_url, "include_subdomains": False, "ignore_query_parameters": True}
    r = session.post(f"{API}/v1/map", json=body, headers=HEADERS, timeout=120)
    r.raise_for_status()
    return [(link["url"], None) for link in r.json()["links"]]

Das

lastmod -Feld aus einer Sitemap ist ein Hinweis, keine Garantie. Einige CMS versehen jede URL mit dem Zeitpunkt der Veröffentlichung; andere aktualisieren diesen nie. Betrachten Sie dies als Grund, früher zu überprüfen, niemals jedoch als Grund, eine Überprüfung gänzlich zu überspringen.

URL-Normalisierung

Ohne Normalisierung werden https://Example.com/docs/?utm_source=x#top und https://example.com/docs zu zwei Zeilen und zwei Abrufen. Normalisieren Sie, bevor irgendetwas den State Store berührt.

from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode

TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
            "gclid", "fbclid", "ref", "mc_cid", "mc_eid"}

def normalize(url: str) -> str:
    parts = urlsplit(url.strip())
    scheme = parts.scheme.lower()
    host = parts.netloc.lower()
    if (scheme, host.rsplit(":", 1)[-1]) in {("http", "80"), ("https", "443")}:
        host = host.rsplit(":", 1)[0]
    query = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
             if k.lower() not in TRACKING]
    query.sort()
    path = parts.path or "/"
    if len(path) > 1 and path.endswith("/"):
        path = path[:-1]                       # pick one convention and stick to it
    return urlunsplit((scheme, host, path, urlencode(query), ""))  # fragment dropped

Die Regel bezüglich des abschließenden Schrägstrichs ist eine Frage der Vorgehensweise. Einige Frameworks stellen unterschiedliche Inhalte unter /docs und /docs/ bereit. Wenn die Zielseite zu diesen gehört, entfernen Sie diese Zeile und lassen Sie die beiden URLs als separate Zeilen bestehen.

Der staatliche Speicher

Eine einzige SQLite-Tabelle enthält alle Informationen, die der Crawler für seine Entscheidungen benötigt. Die Spalten entsprechen den im weiteren Verlauf dieses Leitfadens behandelten Signalen.

import sqlite3
import time

SCHEMA = """
CREATE TABLE IF NOT EXISTS pages (
  normalized_url TEXT PRIMARY KEY,
  url            TEXT NOT NULL,
  last_seen      REAL,
  last_fetched   REAL,
  next_check     REAL DEFAULT 0,
  interval_s     INTEGER DEFAULT 86400,
  content_hash   TEXT,
  etag           TEXT,
  last_modified  TEXT,
  sitemap_lastmod TEXT,
  missing_runs   INTEGER DEFAULT 0,
  gone           INTEGER DEFAULT 0
);
"""

def open_db(path: str = "crawl_state.sqlite") -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.executescript(SCHEMA)
    return db

def upsert_discovered(db: sqlite3.Connection, found: list[tuple[str, str | None]]) -> None:
    now = time.time()
    for raw, lastmod in found:
        n = normalize(raw)
        row = db.execute("SELECT sitemap_lastmod FROM pages WHERE normalized_url=?", (n,)).fetchone()
        if row is None:
            db.execute("INSERT INTO pages (normalized_url,url,last_seen,sitemap_lastmod) VALUES (?,?,?,?)",
                       (n, raw, now, lastmod))
        else:
            bump = lastmod is not None and lastmod != row[0]
            db.execute("UPDATE pages SET last_seen=?, missing_runs=0, sitemap_lastmod=?, "
                       "next_check=CASE WHEN ? THEN 0 ELSE next_check END WHERE normalized_url=?",
                       (now, lastmod, bump, n))
    db.commit()

last_seen erfasst den letzten Erkundungslauf, bei dem die URL aufgeführt wurde; dies ist unabhängig von last_fetched . Eine Änderung von sitemap_lastmod setzt next_check auf Null zurück, sodass die Seite in die nächste Änderungswarteschlange aufgenommen wird.

Bedingte Anfragen: ETag und If-Modified-Since

Wenn eine Website ETag oder Last-Modified sendet, ermöglicht eine bedingte Anfrage dem Ursprungsserver, einen 304-Status ohne Hauptteil zurückzugeben. Sie übermitteln diese Header über das Feld headers von /v1/extract und lesen den Status des Ursprungsservers aus metadata.http_status aus. Die Anfrage wird weiterhin auf Ihr Kontingent angerechnet, ist jedoch schneller als ein vollständiges Rendering.

Nicht jede Website berücksichtigt Validatoren, und CDNs entfernen sie manchmal. Der folgende Code sendet sie, sofern vorhanden, und weicht auf das Hashing aus, wenn die Antwort 200 lautet.

Inhalts-Hashing nach dem Entfernen flüchtiger Blöcke

Ein Roh-Hash der Markdown-Änderungen ändert sich bei jedem Durchlauf auf Seiten, die die aktuelle Uhrzeit, ein CSRF-Token oder einen Block mit „verwandten Artikeln“ ausgeben. Entfernen Sie diese vor dem Hashing. Bewahren Sie die Regeln zum Entfernen an einem Ort auf, damit Sie sie pro Website anpassen können.

import hashlib
import re

VOLATILE = [
    re.compile(r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}(:\d{2})?\S*"),   # ISO timestamps
    re.compile(r"(?i)(csrf|nonce|session)[_-]?token[^\s]*"),
    re.compile(r"(?i)^#+\s*related (posts|articles).*?(?=^#|\Z)", re.M | re.S),
    re.compile(r"[ \t]+"),
]

def content_fingerprint(markdown: str) -> str:
    text = markdown
    for rx in VOLATILE:
        text = rx.sub(" ", text)
    text = "\n".join(line.strip() for line in text.splitlines() if line.strip())
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def fetch_one(session: requests.Session, row: sqlite3.Row | tuple) -> dict:
    """Sync extract with conditional headers. Returns a dict the caller can store."""
    normalized_url, url, etag, last_modified = row
    cond = {}
    if etag:
        cond["If-None-Match"] = etag
    if last_modified:
        cond["If-Modified-Since"] = last_modified
    body = {"url": url, "formats": ["markdown"], "render_js": False,
            "processing_mode": "sync", "headers": cond or None, "timeout": 60000}
    r = session.post(f"{API}/v1/extract", json=body, headers=HEADERS, timeout=90)
    if r.status_code == 429:
        raise RuntimeError(f"rate limited: {r.json().get('error', r.json())}")
    r.raise_for_status()
    payload = r.json()
    status = payload["metadata"].get("http_status")
    md = (payload.get("data") or {}).get("markdown") or ""
    return {"http_status": status, "markdown": md,
            "hash": content_fingerprint(md) if status == 200 else None}

Ob die Antwort-Header ETag und Last-Modified des Ursprungs in metadata verfügbar sind, hängt von der verwendeten API-Version ab; überprüfen Sie die Antwort für Ihr Konto, bevor Sie sich darauf verlassen. Wenn sie nicht verfügbar sind, liefert Ihnen der Inhalts-Hash dennoch eine zuverlässige Antwort darauf, ob der Inhalt geändert wurde oder nicht – Sie verzichten lediglich auf die 304-Abkürzung.

Die Änderungswarteschlange und der adaptive erneute Besuch

Die Warteschlange ist das Herzstück des inkrementellen Web-Crawlings und stellt eine Abfrage dar, keine Datenstruktur: jede Zeile, deren next_check in der Vergangenheit liegt und die nicht als gone markiert ist. Nach jedem Abruf verschiebt sich das Intervall. Änderung festgestellt: Halbieren, auf eine Stunde abrunden. Keine Änderung: mit 1,5 multiplizieren, auf 30 Tage aufrunden. Im Laufe einiger Wochen passt sich der Zeitplan an die beobachtete Änderungshäufigkeit jeder Seite an.

MIN_S, MAX_S = 3600, 30 * 86400

def change_queue(db: sqlite3.Connection, limit: int = 500) -> list[tuple]:
    return db.execute(
        "SELECT normalized_url, url, etag, last_modified FROM pages "
        "WHERE gone=0 AND next_check<=? ORDER BY next_check LIMIT ?",
        (time.time(), limit)).fetchall()

def record_result(db: sqlite3.Connection, normalized_url: str, res: dict) -> str:
    now = time.time()
    old_hash, interval = db.execute(
        "SELECT content_hash, interval_s FROM pages WHERE normalized_url=?",
        (normalized_url,)).fetchone()
    if res["http_status"] in (404, 410):
        db.execute("UPDATE pages SET gone=1, last_fetched=? WHERE normalized_url=?", (now, normalized_url))
        db.commit()
        return "gone"
    changed = res["http_status"] == 200 and res["hash"] != old_hash
    interval = max(MIN_S, interval // 2) if changed else min(MAX_S, int(interval * 1.5))
    db.execute("UPDATE pages SET last_fetched=?, next_check=?, interval_s=?, "
               "content_hash=COALESCE(?, content_hash) WHERE normalized_url=?",
               (now, now + interval, interval, res["hash"], normalized_url))
    db.commit()
    return "changed" if changed else "unchanged"

Ein 304-Status wird mit http_status == 304 und ohne Hash empfangen, sodass die Zeile in den unveränderten Zweig fällt und das Intervall verlängert wird. Bei jedem 5xx- oder Netzwerkfehler sollte die Zeile unverändert bleiben, damit sie beim nächsten Durchlauf erneut abgefragt wird; um dies zu gewährleisten, sollten Sie fetch_one in Ihrem Runner in einen try/except-Block einbinden.

Erkennung gelöschter URLs

Zwei Signale, die zusammen verwendet werden. Ein direkter 404- oder 410-Fehler setzt gone in record_result sofort auf „true“. Eine URL, die aus der Sitemap verschwindet, ist ein schwächerer Hinweis, da Sitemaps zeitlich verzögert sind und manchmal Abschnitte auslassen. Zähle aufeinanderfolgende Erkennungsläufe, bei denen die URL fehlte, und erzwinge erst nach N solchen Läufen einen bestätigenden Abruf.

def mark_missing(db: sqlite3.Connection, run_started: float, n_runs: int = 3) -> None:
    """Call after upsert_discovered. Rows not seen this run get missing_runs+1."""
    db.execute("UPDATE pages SET missing_runs=missing_runs+1 WHERE last_seen<? AND gone=0", (run_started,))
    db.execute("UPDATE pages SET next_check=0 WHERE missing_runs>=? AND gone=0", (n_runs,))
    db.commit()

Zeilen, die als gone markiert sind, bleiben in der Tabelle. Dadurch wird verhindert, dass ein veralteter eingehender Link die URL erneut als neu einbringt. Wenn die Seite wieder verfügbar ist, wird sie bei einem späteren Erkennungslauf erneut aufgelistet; setze gone in upsert_discovered auf 0 zurück, wenn du dieses Verhalten wünschst.

Das Abrufen der geänderten Elemente im Massenverfahren

Für einige hundert URLs ist die oben beschriebene Synchronisierungsschleife ausreichend, und in einer ausgereiften, inkrementellen Webcrawling-Umgebung ist die Menge der geänderten Elemente in der Regel so klein. Bei Tausenden von URLs sollten Sie die Warteschlange als einen einzigen /v1/batch-Auftrag übermitteln und diesen abfragen. Der Batch-Body akzeptiert dieselben Felder formats, render_js, proxy, headers und wait_config wie ein einzelner Extrakt, die auf jede URL angewendet werden; daher können bedingte Header innerhalb eines Batches nicht je nach URL variieren. Verwenden Sie den Batch-Modus für den reinen Content-Hash-Pfad und behalten Sie den Synchronisierungspfad für URLs bei, für die Sie einen ETag besitzen.

def batch_fetch(session: requests.Session, urls: list[str], poll_s: int = 10) -> dict[str, dict]:
    body = {"urls": urls, "formats": ["markdown"], "render_js": False, "ignore_invalid_urls": True}
    job = session.post(f"{API}/v1/batch", json=body, headers=HEADERS, timeout=60).json()
    while True:
        st = session.get(f"{API}/v1/batch/{job['job_id']}", headers=HEADERS, timeout=60).json()
        if st["status"] in ("completed", "failed", "cancelled"):
            break
        time.sleep(poll_s)
    out = {}
    for res in st.get("results", []):
        if res["status"] != "completed":
            continue
        md = (res.get("data") or {}).get("markdown") or ""
        out[res["url"]] = {"http_status": res["metadata"].get("http_status"),
                           "markdown": md, "hash": content_fingerprint(md)}
    return out

Die Alternative ist processing_mode: "async" bei einzelnen /v1/extract-Aufrufen, was einen 202-Status mit einer job_id zurückgibt, die du unter GET /v1/extract/{job_id} abfragst. Dadurch bleiben die Header pro URL erhalten, es wird jedoch eine Anfrage pro URL benötigt. In beiden Fällen gibst du die Ergebnisse genau wie beim Sync-Pfad über record_result weiter.

Den Lauf zusammenstellen

Ein einzelner Lauf umfasst: Erkennen, Upsert, als fehlend markieren, die Warteschlange leeren, aufzeichnen. Führen Sie ihn über Cron oder einen Scheduler in der von Ihrem Aktualisierungsziel geforderten Häufigkeit aus; die adaptiven Intervalle halten die Anzahl der Abrufe gering, unabhängig davon, wie oft der Runner ausgelöst wird.

def run(sitemap_url: str | None, site_url: str) -> None:
    db = open_db()
    session = requests.Session()
    started = time.time()
    found = urls_from_sitemap(sitemap_url, session) if sitemap_url else urls_from_map(site_url, session)
    upsert_discovered(db, found)
    mark_missing(db, started)
    queue = change_queue(db)
    with_etag = [r for r in queue if r[2] or r[3]]
    plain = [r for r in queue if not (r[2] or r[3])]
    for row in with_etag:
        try:
            record_result(db, row[0], fetch_one(session, row))
        except Exception as exc:          # leave row for next run
            print("retry later:", row[1], exc)
    if plain:
        results = batch_fetch(session, [r[1] for r in plain])
        for row in plain:
            if row[1] in results:
                record_result(db, row[0], results[row[1]])

Dies ist das gesamte Modul für das inkrementelle Web-Crawling. Die folgende Tabelle fasst zusammen, welche Funktion die einzelnen Signale haben, sodass Sie nicht benötigte Teile weglassen können.

Häufige Fehler

Hashing vor dem Entfernen flüchtiger Inhalte

Eine Seite, die „Generiert um 14:02:11“ anzeigt, ändert sich bei jedem Abruf, sodass Ihr Crawler sein Intervall halbiert, bis er den Mindestwert erreicht und stündlich neu abruft. Sehen Sie sich den Diff zweier aufeinanderfolgender Abrufe für jede Seite an, die bei jedem Durchlauf Änderungen meldet, und fügen Sie das Muster zu VOLATILE hinzu.

Normalisierung erst nach dem ersten Eintrag

Wenn beim ersten Durchlauf die Roh-URLs gespeichert werden und die Normalisierung erst später erfolgt, ist jede Zeile ein potenzieller Duplikat. Normalisieren Sie daher von Anfang an und legen Sie normalized_url als Primärschlüssel fest, wie es das Schema vorsieht.

Sich allein auf das Fehlen einer Sitemap verlassen

Sitemaps hinken den Veröffentlichungen hinterher, sind auf Indexdateien aufgeteilt und lassen nach einer CMS-Änderung manchmal ganze Abschnitte weg. Eine Seite als nicht mehr vorhanden zu markieren, nur weil sie bei einem Sitemap-Abruf verschwunden ist, löscht Live-Inhalte aus Ihrem Index. Verlangen Sie eine bestätigende 404- oder 410-Fehlermeldung.

Batch-Verarbeitung mit URL-spezifischen bedingten Headern

/v1/batch wendet ein einziges headers-Objekt auf jede URL im Auftrag an. Das Senden eines für eine Seite bestimmten If-None-Match-Headers an tausend andere Seiten führt zu sinnlosen Ergebnissen. Leiten Sie Zeilen mit ETag über den Synchronisationspfad und den Rest über den Batch-Prozess weiter, wie es run() tut.

Nächster Schritt

Das oben genannte Modul verwaltet den Status, vergleicht Inhalte und plant erneute Aufrufe. Es benötigt eine Abrufebene, die für jede URL sauberes Markdown mit einem korrekten http_status-Status zurückgibt, einschließlich der 304er- und 404er-Fehler, auf die es angewiesen ist. Die Geonode-Scraper API stellt diese Ebene bereit: /v1/extract für einzelne Seiten mit Headern pro Anfrage, /v1/batch für den Massen-Änderungssatz und /v1/map, wenn keine Sitemap vorhanden ist. Die kostenlose Stufe umfasst 1.500 Anfragen pro Monat – genug, um die inkrementelle Schleife auf einer echten Website zu testen, bevor Sie sich für einen Tarif entscheiden. Informationen zu den Tarifen finden Sie auf der Scraper API-Seite und zu den Anforderungsstufen unter Preise.

FAQ

Was ist inkrementelles Web-Crawling?

Inkrementelles Web-Crawling bedeutet, dass nur die Seiten erneut besucht werden, bei denen seit dem letzten Durchlauf wahrscheinlich Änderungen vorgenommen wurden, anstatt jedes Mal die gesamte Website neu abzurufen. Der Crawler verwaltet einen Zustandsspeicher mit einem Fingerabdruck pro URL, ermittelt Kandidaten auf ressourcenschonende Weise (Sitemap-Lastmod, ETag, Last-Modified) und setzt das Abrufbudget für die Teilmenge ein, die es tatsächlich benötigt.

Wie kann ich feststellen, welche Seiten sich geändert haben, ohne sie herunterzuladen?

Ohne eine Anfrage lässt sich dies in der Regel nicht mit Sicherheit feststellen, aber Sie können Kandidaten nach Priorität ordnen. Das „Last-Modified“-Datum der Sitemap ist das kostengünstigste Signal. Eine bedingte Anfrage mit „If-None-Match“ oder „If-Modified-Since“ gibt bei Websites, die Validatoren unterstützen, den Status 304 zurück. In allen anderen Fällen rufst du den Inhalt ab und vergleichst einen Inhalts-Hash, der nach dem Entfernen flüchtiger Blöcke wie Zeitstempel und Session-Token berechnet wurde.

Übernimmt die Geonode-Scraper API für mich die Zustands- und Änderungserkennung?

Nein. Die „Scraper API“ ruft Seiten ab und gibt Markdown oder HTML sowie Metadaten wie den http_status zurück. Ihr Skript ist für den Status-Speicher, die URL-Normalisierung, das Hashing und die Zeitplanung zuständig. Diese Aufteilung ist bewusst gewählt: Die API bleibt zustandslos und kostengünstig im Aufruf, während Ihre Logik für das inkrementelle Web-Crawling in Code implementiert ist, den Sie kontrollieren und testen können.

Wie sollte ich Wiederholungsaufrufe für das inkrementelle Web-Crawling planen?

Beginnen Sie mit einem Standardintervall und passen Sie dieses dann pro URL an. Wenn bei einem Abruf eine Änderung festgestellt wird, halbieren Sie das Intervall bis auf einen Mindestwert; wenn keine Änderung festgestellt wird, vergrößern Sie es bis auf einen Höchstwert. Seiten mit hoher Fluktuation werden häufig überprüft, statische Seiten tendieren zur Obergrenze. Speichere next_check in der Statustabelle und frage nach Zeilen, deren next_check bereits abgelaufen ist.

Wie erkenne ich Seiten, die gelöscht wurden?

Verwende zwei Signale zusammen. Ein 404- oder 410-Fehler beim Abruf ist ein starkes Signal. Das Fehlen in der Sitemap bei N aufeinanderfolgenden Durchläufen ist ein schwächeres Signal, da Sitemaps oft unvollständig sind. Markiere die Zeile erst dann als entfernt, wenn ein direkter Abruf dies bestätigt, und behalte die Zeile bei, damit du die URL nicht über veraltete Links erneut entdeckst und abrufst.

Welche Grenzen und Compliance-Regeln sollte ich beachten?

Beachten Sie die robots.txt-Datei und die Nutzungsbedingungen der Website, halten Sie die Anfrageraten moderat und umgehen Sie niemals die Authentifizierung. Was die Scraper API betrifft, umfasst die kostenlose Stufe 1.500 Anfragen pro Monat, und kostenpflichtige Tarife reichen von 2 bis 100 Threads – passen Sie also Ihren geänderten Datensatz an Ihren Tarif an. Bedingte Anfragen zählen weiterhin als Anfragen.