増分型Webクロール：すべてを再クロールすることなく、変更されたページを更新する

増分ウェブクロールとは、サイト全体ではなく、変更されたページのみを再取得する手法です。このガイドを読み終える頃には、サイトマップや Geonode、/v1/map エンドポイントから URL を検出する Python モジュールが完成しているはずです。 それらを正規化し、URLごとの状態をSQLiteに保持し、ETagとコンテンツハッシュを用いて変更箇所を判定し、再訪問間隔を調整し、削除を検出し、変更されたページセットを/v1/extractまたは/v1/batch経由で取得するPythonモジュールが完成します。 Geonodeがページを取得します。状態の管理と比較処理は、スクリプト側で行います。

完全再クロールが予算を浪費する理由

50,000ページあり、毎週2%が更新されるサイトの場合、最新の状態を維持するには週に約1,000回のフェッチが必要です。週1回のフル再クロールでは50,000回のフェッチを消費します。これは、APIリクエスト数、帯域幅、そしてパイプラインがI/Oでブロックされる時間において、50倍の差に相当します。

この無駄は、1回のフェッチごとに処理が遅くコストも高いJavaScriptレンダリングページや、厳格なレート制限が設けられているサイトではさらに増大します。増分型ウェブクロールはこの問題を根本から解決します。URLごとに少量のステート（状態情報）を保持し、そのステートに基づいて次のリクエストをどこに割り当てるかを判断するのです。

以下のモジュールは約120行です。各セクションで1つの機能が追加されています。pip install requestsを使用して依存関係をインストールしてください（お好みであれば、httpxでも同様に動作します）。

URLの発見

優先順位の高い順に2つの情報源があります。サイトマップからはURLに加え、オプションでlastmodが得られます。lastmodは、最も安価な変更シグナルです。 サイトに利用可能なサイトマップがない場合、POST /v1/mapはサイトマップの解析とHTMLリンクの抽出を組み合わせてURLを検出するとともに、各URLにsourceを付与して返します。

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

サイトマップからのlastmodはあくまでヒントであり、保証ではありません。CMSによっては、すべてのURLにデプロイ日時を付与するものもあれば、一切更新しないものもあります。これを「早めに確認すべき理由」として捉え、決して「確認を完全に省略する理由」としては扱わないでください。

URLの正規化

正規化を行わない場合、「https://Example.com/docs/?utm_source=x#top」と「https://example.com/docs」は2つの行となり、2回のフェッチが必要になります。ステートストアにデータが書き込まれる前に正規化を行ってください。

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

末尾のスラッシュに関するルールは、ポリシー上の選択です。フレームワークによっては、/docs と /docs/ で異なるコンテンツを配信する場合があります。対象サイトがそのようなフレームワークを使用している場合は、その行を削除し、2つのURLを別々の行として残してください。

ステートストア

1つのSQLiteテーブルに、クローラーが判断を行うために必要なすべての情報が格納されています。各列は、このガイドの残りの部分で説明するシグナルに対応しています。

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には、そのURLがリストされた直近のディスカバリー実行が記録されます。これはlast_fetchedとは別のものです。sitemap_lastmodが変更されると、next_checkがゼロにリセットされ、そのページは次の変更キューに組み込まれます。

条件付きリクエスト：ETag および If-Modified-Since

サイトが ETag または Last-Modified を送信する場合、条件付きリクエストを使用することで、オリジンは本文なしで 304 を返すことができます。 これらのヘッダーは、/v1/extractのheadersフィールドを通じて送信し、metadata.http_statusからオリジンのステータスを読み取ります。このリクエストはプランの使用量にカウントされますが、フルレンダリングよりも高速です。

すべてのサイトがバリデータに対応しているわけではなく、CDNによってはそれらを削除してしまうこともあります。以下のコードは、バリデータが存在する場合にそれらを送信し、応答が 200 の場合はハッシュ処理に切り替えます。

揮発性ブロックの除去後のコンテンツハッシュ

現在時刻、CSRFトークン、または「関連記事」ブロックを表示するページでは、実行のたびにMarkdownの生のハッシュ値が変化します。ハッシュ化を行う前に、これらのブロックを除去してください。サイトごとにルールを調整できるよう、除去ルールは一箇所にまとめて管理してください。

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}

オリジンのレスポンスヘッダー「ETag」および「Last-Modified」が「metadata」で公開されるかどうかは、使用しているAPIのバージョンによって異なります。これらを前提とする前に、ご自身のアカウントのレスポンスを確認してください。 これらが公開されていない場合でも、コンテンツのハッシュ値から変更の有無を確実に判断できます。ただ、304ステータスによるショートカットが利用できなくなるだけです。

変更キューと適応型再訪問

このキューは、インクリメンタルWebクロールの核心であり、データ構造ではなくクエリです。つまり、next_checkが過去の日付であり、かつgoneのマークが付いていないすべての行が対象となります。各フェッチのたびに、この間隔は変動します。 変更が検出された場合：その値を半分にし、1時間を下限とする。変更がない場合：1.5倍にし、30日を上限とする。数週間かけて、スケジュールは各ページの観測された変更頻度に収束していく。

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"

http_status == 304でハッシュなしの304レスポンスが返ってきた場合、変更なしの分岐に分類され、間隔が拡大します。 5xxエラーやネットワークエラーが発生した場合は、その行をそのまま残し、次回の実行で再試行するようにします。そのため、ランナー内でfetch_oneをtry/exceptで囲んでください。

削除されたURLの検出

2つのシグナルを組み合わせて使用します。直接的な404または410エラーが発生すると、goneは直ちにrecord_resultに切り替わります。サイトマップからURLが消えた場合は、証拠としての信頼性が低くなります。これは、サイトマップの更新に遅れが生じたり、セクションが省略されたりすることがあるためです。 そのURLが存在しなかった連続した検出実行回数をカウントし、N回に達した場合にのみ確認のためのフェッチを強制します。

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()

goneとマークされた行はテーブルに残ります。これにより、古い被リンクによってそのURLが新規として再登録されるのを防ぎます。 ページが復活した場合、その後の検出実行で再びリストに追加されます。この動作を望む場合は、upsert_discovered内でgoneを0にリセットしてください。

変更されたセットを一括で取得する

URLが数百件程度であれば、上記の同期ループで問題ありません。また、十分に整備された増分型Webクロール環境では、通常、変更されたセットはこの程度の規模にとどまります。URLが数千件に及ぶ場合は、キューを1つの/v1/batchジョブとして送信し、それをポーリングしてください。 バッチ本体は、単一の抽出と同様に formats、render_js、proxy、headers、および wait_config フィールドを受け入れますが、 これらはすべての URL に適用されるため、バッチ内の URL ごとに条件付きヘッダーを変更することはできません。単純なコンテンツハッシュパスにはバッチを使用し、ETag を保持している URL には同期パスを維持してください。

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

別の方法として、個別の /v1/extract 呼び出しで processing_mode: "async" を使用する方法があります。この場合、job_id を返して 202 を返し、GET /v1/extract/{job_id} でポーリングを行います。これにより URL ごとのヘッダーを維持できますが、URL ごとに 1 回のリクエストが必要になります。 いずれの場合も、同期パスとまったく同じように、結果を record_result を通じて処理してください。

実行プロセスの構成

1回の実行は、次の手順で構成されます：検出、アップサート、欠落のマーク、キューの排出、記録。cronやスケジューラを使用して、データの鮮度目標に合わせて適切な間隔で実行してください。適応型間隔機能により、ランナーがどのくらいの頻度で実行されても、フェッチ回数は低く抑えられます。

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

これが増分型Webクロールモジュール全体です。以下の表は各シグナルの役割をまとめたものです。不要な部分は削除して構いません。

よくある間違い

変動するコンテンツを削除する前にハッシュ化してしまう

「Generated at 14:02:11」と表示されるページは、フェッチするたびに内容が変化するため、クローラーはフェッチ間隔を半分に縮め続け、最終的に最低間隔に達して1時間ごとに再フェッチすることになります。 実行のたびに変更が報告されるページについて、連続する2回のフェッチの差分を確認し、そのパターンをVOLATILEに追加してください。

最初の挿入後に正規化を行う

最初の実行で生のURLが保存され、後で正規化が追加されると、すべての行が重複する危険性があります。スキーマの指定通り、初日から正規化を行い、normalized_urlを主キーにしてください。

サイトマップの不在のみに依存すること

サイトマップはデプロイに遅れをとったり、インデックスファイル間で分割されたり、CMSの変更後にセクション全体が削除されたりすることがあります。1回のサイトマップ取得でページが消えたという理由だけでそのページを削除すると、インデックスから公開中のコンテンツが削除されてしまいます。確認として404または410のステータスコードを必須にしてください。

URLごとの条件付きヘッダーを使用してバッチを実行する

/v1/batchは、ジョブ内のすべてのURLに1つのheadersオブジェクトを適用します。1つのページを対象としたIf-None-Matchを他の1,000のページに送信すると、意味のない結果が生じます。 run() が行っているように、ETag を含む行は同期パスに、それ以外はバッチ処理に振り分けてください。

次のステップ

上記のモジュールは状態を保持し、コンテンツを比較し、再訪問をスケジュールしますが、これには、依存している304や404を含むすべてのURLに対して、正しいhttp_statusを含むクリーンなMarkdownを返すフェッチ層が必要です。 Geonode Scraper API はそのレイヤーを提供します。リクエストごとのヘッダーを含む単一ページ用には /v1/extract、一括変更セット用には /v1/batch、サイトマップがない場合は /v1/map を使用します。 無料プランでは月間1.5Kリクエストまでカバーされており、プランの規模を決定する前に、実際のサイトでインクリメンタルループを検証するには十分な量です。 プランについてはScraper APIのページを、リクエストのティアについては料金表をご参照ください。

よくある質問

インクリメンタルWebクローリングとは何ですか？

インクリメンタルWebクローリングとは、毎回サイト全体を再取得するのではなく、前回の実行以降に変更された可能性のあるページのみを再訪問することを指します。 クローラーはURLごとのフィンガープリントを格納した状態ストアを維持し、低コストで更新候補を特定（サイトマップのlastmod、ETag、Last-Modified）し、実際に必要なサブセットに対してのみフェッチリソースを割り当てます。

ページをダウンロードせずに、どのページが変更されたかを知るにはどうすればよいですか？

通常、リクエストを行わなければ確実に知ることはできませんが、候補をランク付けすることは可能です。 サイトマップの「lastmod」が最もコスト効率の良い指標です。If-None-Match や If-Modified-Since を使用した条件付きリクエストは、バリデータに対応しているサイトでは 304 を返します。それ以外の場合は、タイムスタンプやセッショントークンなどの揮発性ブロックを除去した後に計算されたコンテンツハッシュを取得し、比較します。

Geonode や Scraper API は、状態管理や変更検出を自動的に行ってくれますか？

いいえ。Scraper API はページを取得し、Markdown または HTML に加えて http_status などのメタデータを返します。 状態の保存、URLの正規化、ハッシュ化、およびスケジューリングは、スクリプト側で管理する必要があります。この役割分担は意図的なものです。APIはステートレスで呼び出しコストを低く抑えつつ、増分ウェブクロール処理のロジックは、ユーザーが管理・テスト可能なコード内に実装されるようにしています。

増分ウェブクロールの再訪問はどのようにスケジュールすべきですか？

まずはデフォルトの間隔から始め、URLごとに調整してください。フェッチで変更が検出された場合は、間隔を半分にして下限値まで短縮し、変更が検出されなかった場合は、間隔を倍増して上限値まで引き上げます。頻繁に更新されるページは頻繁にチェックされ、静的なページは上限値に近づいていきます。next_checkを状態テーブルに保存し、next_checkが過ぎた行をクエリで抽出します。

削除されたページをどのように検出すればよいですか？

2つのシグナルを組み合わせて使用します。フェッチ時に404または410が返されることは強力なシグナルです。サイトマップにN回連続で掲載されていないことは、サイトマップが不完全であることが多いため、比較的弱いシグナルとなります。 直接フェッチして確認できた場合にのみ、その行を「削除済み」としてマークし、古いリンクからそのURLを再発見して再フェッチすることがないよう、行は保持しておきます。

遵守すべき制限やコンプライアンス規則は何ですか？

robots.txt およびサイトの利用規約を遵守し、リクエストレートを適度な範囲に抑え、認証を迂回してはなりません。 Scraper API に関しては、無料プランでは月間 1.5K リクエストまで対応しており、有料プランでは 2 ～ 100 スレッドが利用可能です。そのため、変更対象セットの規模はご契約のプランに合わせて調整してください。条件付きリクエストもリクエストとしてカウントされます。