网站变更检测是指在供应商尚未正式公告之前,就发现其修改了服务条款、在文档中废弃了某个接口,或悄悄修改了隐私政策。 读完本指南后,您将拥有一段 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_js | keep_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 个页面完全在该配额范围内。