Geonode logo
Geonode Team

Geonode Team

更新于:2026年10月16日

发布于:2026年10月16日

使用 Python 检测网站变更:监控文档和政策更新

网站变更检测是指在供应商尚未正式公告之前,就发现其修改了服务条款、在文档中废弃了某个接口,或悄悄修改了隐私政策。 读完本指南后,您将拥有一段 Python 脚本:它会通过 Geonode 和 Scraper API 工具将页面列表以 Markdown 格式快照,过滤掉无关信息,将每个版本存储在 SQLite 中,生成易于阅读的差异报告,并且仅在内容真正发生变化时才向 Slack 发送通知。 该脚本可通过 cron 或 GitHub Actions 运行,并能应对不稳定的抓取情况。 本脚本刻意不作为价格监控工具。价格是位于已知位置的数值;而文档则是散文,其变更的位置和形式无法预先确定。以下的设计选择正是基于这一前提。

网站变更检测是指在供应商尚未正式公告之前,就发现其修改了服务条款、在文档中废弃了某个接口,或悄悄修改了隐私政策。 读完本指南后,您将拥有一段 Python 脚本:它会通过 Geonode 和 Scraper API 工具将页面列表以 Markdown 格式快照,过滤掉无关信息,将每个版本存储在 SQLite 中,生成易于阅读的差异报告,并且仅在内容确实发生变化时才向 Slack 发送通知。 该脚本可通过 cron 或 GitHub Actions 运行,并能应对不稳定的数据获取情况。

本脚本刻意不作为价格监控工具。价格是位于已知位置的数值;而文档则是散文,其变更的位置和形式无法预先确定。下文的设计选择正是基于这一前提。

监控对象及选择 Markdown 的原因

值得关注的页面具有以下共同特征:公开可见、以文本为主、更新频率较低,且一旦更新将产生重大影响。 API 参考页面、身份验证指南、速率限制表、条款与条件、数据处理协议、可接受使用政策、变更日志以及状态页面的故障历史记录均符合这一标准。

比较这些页面的原始 HTML 非常痛苦。资产 URL 中的哈希值、CSRF 令牌、A/B 测试类名以及注入的分析代码在每次加载时都会发生变化。 先转换为 Markdown 格式,几乎可以过滤掉所有这些内容,仅保留句子。/v1/extract 端点会直接返回 Markdown 格式,因此脚本无需调用任何 HTML 解析器。

使用 pip install requests 安装唯一的依赖项。其余部分均来自标准库。

截取快照

该抓取工具要求使用 Markdown 格式,默认禁用 JavaScript 渲染,但对于采用客户端渲染的文档网站,会按页面启用该功能。只有当页面在加载后才填充内容时,才需要 wait_config 代码块;若存在该代码块,系统会自动启用渲染。

import os
import requests

API = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_API_KEY"]}

def snapshot(session: requests.Session, url: str, render_js: bool = False,
             wait_for: str | None = None) -> tuple[int, str]:
    """Return (origin_http_status, markdown). Raises on API-level errors."""
    body = {"url": url, "formats": ["markdown"], "render_js": render_js,
            "processing_mode": "sync", "timeout": 60000}
    if wait_for:
        body["wait_config"] = {"wait_until": "domcontentloaded", "wait_for": wait_for,
                               "wait_timeout": 2000}
        body.pop("render_js")          # wait_config implies rendering; sending false is rejected
    r = session.post(f"{API}/v1/extract", json=body, headers=HEADERS, timeout=90)
    if r.status_code == 429:
        err = r.json().get("error", r.json())
        raise RuntimeError(f"rate limited ({err.get('code')}): {err.get('message')}")
    r.raise_for_status()
    payload = r.json()
    return payload["metadata"]["http_status"], (payload.get("data") or {}).get("markdown") or ""

源状态与正文内容同样重要。一个 503 维护页面转换后会成为完全有效的 Markdown;如果不检查 http_status,系统会将其存储为新版本并触发警报。

过滤噪音

按顺序应用两种机制。首先通过正则表达式过滤,移除已知会频繁变化的内容:日期、“最后更新”横幅、Cookie 提示、版权页脚以及“编辑此页面”链接。 随后,通过分段白名单仅保留您关心的标题,从而剔除导航栏、侧边栏和营销模块——这些内容因属于真实文本而被 Markdown 转换器保留了下来。

import re

NOISE = [
    re.compile(r"(?im)^.*\b(last (updated|modified|reviewed)|effective date|updated on)\b.*$"),
    re.compile(r"(?im)^.*\b(cookie|cookies)\b.*\b(accept|consent|preferences)\b.*$"),
    re.compile(r"(?im)^.*(©|copyright|all rights reserved).*$"),
    re.compile(r"(?im)^.*\b(edit this page|was this page helpful|on this page)\b.*$"),
    re.compile(r"\b(?:\d{1,2}\s)?(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4}\b", re.I),
    re.compile(r"\b\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2})?\S*)?\b"),
    re.compile(r"!\[[^\]]*\]\([^)]*\)"),                 # images
]

def strip_noise(md: str, keep_sections: list[str] | None = None) -> str:
    for rx in NOISE:
        md = rx.sub("", md)
    if keep_sections:
        kept, keep = [], False
        for line in md.splitlines():
            if line.startswith("#"):
                title = line.lstrip("#").strip().lower()
                keep = any(s.lower() in title for s in keep_sections)
            if keep:
                kept.append(line)
        md = "\n".join(kept)
    return normalize_ws(md)

def normalize_ws(text: str) -> str:
    lines = (re.sub(r"[ \t]+", " ", ln).strip() for ln in text.splitlines())
    return "\n".join(ln for ln in lines if ln)

keep_sections 是一组用于匹配标题的子字符串列表,因此 ["rate limit", "authentication"] 会保留这些 H2/H3 块及其下所有内容,直至下一个标题。对于政策页面(其中任何条款都至关重要),请保留 None 设置。

日期正则表达式被刻意设置得较为严格。政策页面中“自 2026 年 3 月 1 日起生效”这一行正是您希望看到的变更,因此如果这对您很重要,请删除第一个 NOISE 条目,让差异对比显示该内容。

哈希与差异比较

哈希能在常数时间内回答“是否有内容发生变化”。差异比较则能以人类可读的形式,在 Slack 消息中回答“发生了哪些变化”。大多数网站变更检测工具仅止步于哈希;建议两者兼用:哈希用于筛选,差异比较用于报告。

import difflib
import hashlib

def digest(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def unified(old: str, new: str, url: str, context: int = 2, max_lines: int = 120) -> str:
    diff = difflib.unified_diff(old.splitlines(), new.splitlines(),
                                fromfile=f"{url} (previous)", tofile=f"{url} (current)",
                                lineterm="", n=context)
    lines = list(diff)
    if len(lines) > max_lines:
        lines = lines[:max_lines] + [f"... {len(lines) - max_lines} more lines"]
    return "\n".join(lines)

difflib.unified_diff 属于标准库,相关文档见 docs.python.org。 n 参数用于设置上下文行数;对于散文类内容,两行已足够。请对输出进行限制,以免整页重写导致生成长达 5,000 行的警报。

存储快照

SQLite 会保留完整的历史记录,而不仅仅是最新版本。这使得我们可以与最近两个快照进行比较(详见下一节),并在数月后回答“该子句首次出现于何时”的问题。

import sqlite3
import time

SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshots (
  id          INTEGER PRIMARY KEY,
  url         TEXT NOT NULL,
  taken_at    REAL NOT NULL,
  http_status INTEGER,
  hash        TEXT,
  content     TEXT
);
CREATE INDEX IF NOT EXISTS idx_url_time ON snapshots(url, taken_at DESC);
"""

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

def last_snapshots(db: sqlite3.Connection, url: str, n: int = 2) -> list[tuple[str, str]]:
    """Most recent first: [(hash, content), ...] for successful (200) fetches only."""
    return db.execute("SELECT hash, content FROM snapshots WHERE url=? AND http_status=200 "
                      "ORDER BY taken_at DESC LIMIT ?", (url, n)).fetchall()

def save(db: sqlite3.Connection, url: str, status: int, h: str | None, content: str | None) -> None:
    db.execute("INSERT INTO snapshots (url, taken_at, http_status, hash, content) VALUES (?,?,?,?,?)",
               (url, time.time(), status, h, content))
    db.commit()

策略页面体积较小,因此每次运行时存储完整内容的成本很低。如果您每小时需要监控数百份长文档,则仅在哈希值发生变化时存储内容,否则仅保留哈希值行。

处理不稳定的数据获取

不稳定的数据获取是网站变更检测中误报的主要来源,因此有三种应对策略:在放弃之前,采用退避策略重试数据获取;将除 200 以外的所有状态码视为“无数据”,将该状态码记录在案,且不进行比较。 此外,应将新哈希值与最近的两个有效快照进行比对,而不仅仅是与最后一个:如果页面状态依次变为 A、B、A,那么 B 状态极有可能是暂时的变体(例如仍返回 200 状态码的错误页面,或缓存中提供的过期版本),不应触发警报。

def fetch_with_retry(session: requests.Session, url: str, attempts: int = 3, **kw) -> tuple[int, str]:
    delay = 5
    for i in range(attempts):
        try:
            status, md = snapshot(session, url, **kw)
            if status < 500 and md.strip():
                return status, md
        except (requests.RequestException, RuntimeError) as exc:
            if i == attempts - 1:
                raise
        time.sleep(delay)
        delay *= 3
    return status, md

def check(db: sqlite3.Connection, session: requests.Session, url: str,
          keep_sections: list[str] | None = None, **kw) -> str | None:
    """Returns a unified diff if a real change was detected, else None."""
    status, raw = fetch_with_retry(session, url, **kw)
    if status != 200:
        save(db, url, status, None, None)
        return None
    clean = strip_noise(raw, keep_sections)
    h = digest(clean)
    previous = last_snapshots(db, url, n=2)
    save(db, url, status, h, clean)
    if not previous:
        return None                                   # first run, nothing to compare
    if any(h == old_hash for old_hash, _ in previous):
        return None                                   # unchanged, or a revert of a blip
    return unified(previous[0][1], clean, url)

“第二个快照”规则存在一个代价:在单次运行内被回滚的真实变更将不会被报告。对于文档和策略监控而言,这几乎总是正确的权衡。

通知

Slack 的传入 webhook 接受包含 text 字段的 JSON 正文;请将差异内容包裹在代码块中,以确保其保持等宽字体显示。通过 smtplib 发送的电子邮件内容稍长几行,在无法使用 Slack 的情况下非常实用。

import json
import smtplib
from email.message import EmailMessage

SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL")

def notify_slack(url: str, diff: str) -> None:
    if not SLACK_WEBHOOK:
        return
    fence = "`" * 3                          # literal ``` would close the Markdown block
    text = f"*Change detected:* {url}\n{fence}\n{diff[:3500]}\n{fence}"
    requests.post(SLACK_WEBHOOK, data=json.dumps({"text": text}),
                  headers={"Content-Type": "application/json"}, timeout=15).raise_for_status()

def notify_email(url: str, diff: str, to: str) -> None:
    msg = EmailMessage()
    msg["Subject"] = f"Change detected: {url}"
    msg["From"] = os.environ["SMTP_FROM"]
    msg["To"] = to
    msg.set_content(diff)
    with smtplib.SMTP(os.environ["SMTP_HOST"], int(os.environ.get("SMTP_PORT", "587"))) as s:
        s.starttls()
        s.login(os.environ["SMTP_USER"], os.environ["SMTP_PASS"])
        s.send_message(msg)

Slack 会截断过长的消息,因此 diff 的长度上限为 3,500 个字符。对于较长的 diff,请发布第一屏内容,并将全文附在其他地方,或者在消息中指向已存储的快照 ID。

观察列表和运行器

配置存储在一个列表中,因此您可以将其保存在代码库中,并像审查代码一样审查其变更。每个条目包含 URL、是否需要渲染、一个可选的待处理选择器,以及一个可选的节允许列表。

WATCHLIST = [
    {"url": "https://example.com/legal/terms"},
    {"url": "https://example.com/legal/privacy"},
    {"url": "https://example.com/docs/api/authentication",
     "keep_sections": ["authentication", "rate limit"]},
    {"url": "https://example.com/changelog", "render_js": True, "wait_for": "main article"},
]

def main() -> None:
    db = open_db()
    session = requests.Session()
    for item in WATCHLIST:
        url = item["url"]
        try:
            diff = check(db, session, url, keep_sections=item.get("keep_sections"),
                         render_js=item.get("render_js", False), wait_for=item.get("wait_for"))
        except Exception as exc:
            print(f"[skip] {url}: {exc}")
            continue
        if diff:
            print(f"[changed] {url}")
            notify_slack(url, diff)
        else:
            print(f"[ok] {url}")

if __name__ == "__main__":
    main()

首次运行会初始化数据库,且不返回任何报告。从第二次运行开始,每次实际变更都会触发一条包含差异信息的 Slack 消息。

网站变更检测的调度

以下两种方案无需任何基础设施。只需在任何安装了 Python 并设置了环境变量的机器上添加一条 cron 任务:

# daily at 06:10, log to a file
10 6 * * * cd /opt/change-monitor && /usr/bin/python3 monitor.py >> monitor.log 2>&1

或者使用 GitHub Actions 工作流,这种方式非常方便,因为 SQLite 文件可以作为状态存储提交回代码库,而密钥则保存在代码库设置中:

name: change-monitor
on:
  schedule:
    - cron: "10 6 * * *"
  workflow_dispatch:
jobs:
  run:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install requests
      - run: python monitor.py
        env:
          GEONODE_API_KEY: ${{ secrets.GEONODE_API_KEY }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
      - run: |
          git config user.name "change-monitor"
          git config user.email "bot@users.noreply.github.com"
          git add changes.sqlite && git commit -m "snapshot $(date -u +%F)" || true
          git push

对于变更日志,每小时检查一次即可;对于法律页面,每天检查一次已足以满足大多数团队的需求。下表提供了一个合理的起点。

页面类型建议频率render_jskeep_sections
条款、隐私政策、使用条款每日否无(整页)
API 参考文档每 6 小时取决于文档引擎是,您使用的端点
变更日志/发布说明每小时经常无
状态页面事件历史每小时通常无

关于服务条款和机器人的说明

每天阅读一次公开的文档或政策页面属于网络的常规使用行为,这也正是浏览器所做的事情。不过,在将网站添加到关注列表之前,请务必查阅该网站的服务条款和robots.txt,保持合理的访问频率,并且切勿监控未经授权的登录页面。 如果网站提供了 RSS 源或变更日志 API,请优先使用:这对双方都更经济,且在页面布局发生变化时也不会导致数据中断。

常见错误

比较原始 HTML

现代文档网站的每次构建都会重写资源哈希值和类名。这样会导致差异对比结果难以阅读,且每次部署都会触发警报。请先将内容转换为 Markdown 格式,就像 snapshot() 那样。

首次非 200 状态码即触发警报

单次 503 错误,或是返回 200 状态码并显示“请验证您是人类”的机器人验证页面,不应被视为存储版本或触发警报。 请参考 http_status,要求内容不为空,并与最近两个有效的快照进行比对。

过滤范围过窄或过宽

如果差异报告中仅显示日期,请将其添加到 NOISE 中。如果因 keep_sections 删除了所属的标题而遗漏了子句变更,请扩大白名单范围。在完全信任该监控之前,请运行一周并根据生成的差异报告对两份列表进行微调。

允许监控列表与实际情况产生偏差

一个重定向到新路径的 URL,或者一个移至带版本前缀下的文档页面,会悄无声息地变成 404 错误或另一个页面。请明确记录非 200 状态码,并定期审查输出中的“跳过”行。

下一步

以上内容均基于 Python 标准库,每页仅需一次 HTTP 请求,这已足以满足文档网站变更检测的需求。 自行托管时最难实现的部分在于抓取过程本身:需要一个干净的 Markdown 转换功能,既能处理由 JavaScript 渲染的文档网站,又能返回源站的真实 http_status 值,并且在每天早晨检查四十个供应商页面时不会被封锁。 Geonode 和 Scraper API 正是提供了这样的调用服务,此外还有针对客户端渲染页面的 render_js 和 wait_config,并提供每月 1.5K 次请求的免费配额,足以轻松满足每日监控列表的需求。 有关套餐详情,请参阅 Scraper API 页面;相关使用模式,请参阅 网页抓取用例。

常见问题

什么是网站变更检测?它何时有用?

网站变更检测是指按计划抓取网页,将其与之前的快照进行比较,并在重要内容发生变化时发出警报。 该功能对那些您依赖但无法控制的页面最为有用:API 文档、服务条款、隐私政策、定价条款以及供应商变更日志等,因为这些内容的悄然修改可能会导致集成失效或合规性假设被打破。

为什么比较 Markdown 而不是原始 HTML?

HTML 包含布局、脚本、跟踪参数和构建哈希值,这些内容即使文本未变也会发生变化。先转换为 Markdown 可以将页面简化为纯内容,因此差异对比会显示被编辑的句子,而不是数百行不断变化的属性。 Geonode Scraper API 直接返回 Markdown 格式,从而保持处理流程简洁。

如何避免因日期、Cookie 提示和导航元素引发的误报?

使用一小套正则表达式(日期、“最后更新”行、Cookie 提示、页脚)过滤已知的干扰信息,然后通过标题白名单仅保留您关心的部分。哈希前对空格进行标准化处理。最后,与最近的两个快照进行比较,这样临时错误页面就不会单独触发警报。

应该对页面进行哈希计算还是比较差异?

两者都要做。哈希是一种快速的相等性检查,用于决定是否需要采取任何行动。 dflib 提供的统一差异报告(unified diff)是人类在警报中阅读的内容。先计算哈希值,仅在哈希值不同时才进行差异比较,并将差异报告随通知一并发送,以便读者无需打开页面即可判断变更情况。

网站变更检测应以何种频率运行?

将检测频率与页面的重要性及更新频率相匹配。 法律和政策页面一年通常只会更新几次,因此每天检查一次就足够了。活跃的 API 文档和变更日志则需要每小时检查一次。通过 cron 或 GitHub Actions 计划任务运行该检查,并确保间隔时间足够长,以便在报告异常前能对不稳定的抓取请求进行重试。

合规性和限制方面需要考虑哪些因素?

以适度的频率监控公开文档和政策页面属于正常网络使用行为,但请阅读网站的服务条款和 robots.txt 文件并予以遵守,切勿监控您无权访问的受身份验证保护的页面。 在 API 方面,Scraper API 免费套餐每月提供 1.5K 次请求配额;每天检查 40 个页面完全在该配额范围内。