增量式网页爬取：无需重新爬取所有页面即可更新已更改的页面

增量式网页爬取是指仅重新抓取发生变化的页面，而非整个网站。 读完本指南后，您将获得一个 Python 模块，它能够从网站地图或 Geonode 的 /v1/map 端点中发现 URL，对它们进行规范化处理，将每个 URL 的状态保存在 SQLite 中，利用 ETag 和内容哈希值判断哪些内容发生了变化，调整重访间隔，检测页面删除情况，并通过 /v1/extract 或 /v1/batch 提取已更改的页面集。 Geonode 负责获取页面；您的脚本则负责管理状态和比较操作。

为什么全站重新抓取会浪费预算

一个拥有 50,000 个页面、每周更新率为 2% 的网站，每周需要大约 1,000 次抓取才能保持内容最新。 每周进行一次全站重新抓取则需要50,000次请求。这意味着API请求次数、带宽消耗以及管道因I/O操作而阻塞的时间都增加了50倍。

这种浪费在JavaScript渲染的页面上会进一步加剧——此类页面的每次抓取速度更慢、成本更高；而在实施严格速率限制的网站上，这种浪费同样显著。 增量网页爬取彻底扭转了这一局面：你为每个 URL 保留少量状态信息，并根据该状态决定下一次请求的目标。

下面的模块约有 120 行代码。每个部分都增加了一个功能。使用 pip install requests 安装依赖项（如果你更喜欢 httpx，其工作原理完全相同）。

发现 URL

有两个来源，按优先级排序。网站地图会提供 URL 以及可选的 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 将分别对应两行数据和两次数据获取操作。在任何操作触及状态存储之前，应先进行规范化。

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/ 提供不同的内容。如果目标站点属于此类，请删除该行，让这两个 URL 作为独立的行存在。

状态存储

一个 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}

源站是否在metadata中暴露ETag和Last-Modified响应头，取决于您使用的 API 版本；在依赖这些信息之前，请先检查您账户的响应。如果这些信息未被暴露，内容哈希仍能为您提供可靠的“已更改/未更改”判断结果，只是无法利用 304 状态码的快捷方式。

变更队列与自适应重访

队列是增量式网页爬取的核心，它是一种查询，而非数据结构：即所有next_check字段时间早于当前时间且未被标记为gone的行。每次抓取后，该时间间隔都会发生变化。若检测到变更：将其减半，并取整至一小时。 未发现变更：乘以 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

结合使用两种信号。直接返回 404 或 410 状态码时，会在 record_result 中立即将 gone 设为真。而从站点地图中消失的 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，上面的同步循环就足够了，而在成熟的增量网页爬取环境中，已更改的集合通常规模都很小。 若涉及数千个 URL，请将队列作为单个 /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"，该调用会返回 202 状态码并附带一个 job_id，您可通过 GET /v1/extract/{job_id} 进行轮询。这种方式可以保留每个 URL 的独立头部，但每个 URL 都需要消耗一次请求。无论采用哪种方式，都应像同步路径那样，将结果通过 record_result 进行处理。

构建运行流程

单次运行包括：发现、upsert、标记缺失、清空队列、记录。可通过 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]])

这就是整个增量网页爬取模块。下表总结了各信号的功能，以便您删除不需要的部分。

常见错误

在过滤易变内容前进行哈希计算

一个显示“生成于 14:02:11”的页面在每次抓取时都会发生变化，因此你的爬虫会将抓取间隔缩短一半，直到达到最低间隔，最终每小时重新抓取一次。对于任何在每次运行时都报告发生变化的页面，请查看两次连续抓取的差异，并将该模式添加到 VOLATILE 中。

首次插入后才进行规范化

如果首次运行时存储的是原始 URL，而规范化操作是在之后才添加的，那么每一行都可能成为重复记录。请从第一天开始就进行规范化，并按照模式设计将 normalized_url 设为主键。

仅凭网站地图缺失就判定页面不存在

网站地图的更新往往滞后于内容发布，且分散在多个索引文件中，有时在 CMS 变更后甚至会遗漏整个版块。仅因页面从某次网站地图抓取中消失就将其标记为已删除，将导致索引中活内容被误删。必须要求通过 404 或 410 状态码进行确认。

运行带按 URL 条件头信息的批处理

/v1/batch 会将一个 headers 对象应用于任务中的每个 URL。将本应针对单个页面的 If-None-Match 发送给其他一千个页面，会产生毫无意义的结果。请将带有 ETag 的行通过同步路径处理，其余行通过批处理处理，就像 run() 所做的那样。

下一步

上述模块负责维护状态、比较内容并安排重新访问，它需要一个数据获取层，该层能针对每个 URL 返回干净的 Markdown 内容，并附带真实的 http_status 状态码，包括其依赖的 304 和 404 状态码。 Geonode Scraper API 提供了这一层：/v1/extract 用于单页抓取（支持按请求设置头部），/v1/batch 用于批量抓取已更改的内容，/v1/map 则适用于没有站点地图的情况。免费套餐每月包含 1.5K 次请求，足以让您在选择付费方案前，在真实网站上验证增量循环的运行效果。 有关套餐信息，请参阅 Scraper API 页面；有关请求配额的详细信息，请参阅 定价。

常见问题

什么是增量式网页爬取？

增量式网页爬取是指仅重新访问自上次运行以来可能发生变化的页面，而不是每次都重新抓取整个网站。 爬虫会维护一个状态存储库，其中包含每个 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 已过期的行。

如何检测已被删除的页面？

结合使用两个信号。抓取时返回 404 或 410 状态码是强信号。 在连续 N 次运行中未出现在站点地图中则是一个较弱的信号，因为站点地图通常并不完整。仅当直接抓取确认后，才将该行标记为已删除，并保留该行，以免通过过期链接重新发现并抓取该 URL。

我应遵守哪些限制和合规规则？

请遵守 robots.txt 文件及网站的服务条款，保持适度的请求频率，切勿绕过身份验证。在 Scraper API 方面，免费套餐每月支持 1.5K 次请求，付费套餐支持 2 到 100 个线程，请根据您的套餐选择合适的变更集规模。条件请求仍计入请求总数。