React 网站自动更新 XML 网站地图方案（Python）
以下方案面向 JavaScript/React 网站，部署在 Linux VPS 环境，提供页面变更智能识别、XML 网站地图生成与根目录部署的自动化脚本与实践建议。
1. 依赖项清单
请在 Python 3.10+ 环境中安装以下依赖：
- playwright（支持执行 JavaScript 渲染，适合 React SPA）
- beautifulsoup4（内容解析与清洗）
- lxml（高性能 XML 生成）
- requests（拉取 robots.txt、健康检查）
- urllib3（可选，网络优化）
- sqlite3（标准库，用于持久化变更状态）
- typing-extensions（某些环境兼容）
- python-dotenv（可选，配置注入）
- logging（标准库，日志输出）
- cron（系统级定时器，Linux 环境使用）
安装示例：
pip install playwright beautifulsoup4 lxml requests python-dotenv
python -m playwright install
2. 脚本逻辑架构
整体流程：
- 初始化配置与数据库，加载上次运行记录。
- 读取 robots.txt，确定允许抓取的路径规则。
- 以 Playwright 启动无头浏览器，从站点入口开始 BFS 爬取：
- 对每个 URL 计算内容指纹（DOM 清洗 + 规范化文本哈希），对比历史记录识别新增与修改。
- 构建 XML 网站地图（urlset），添加 loc、lastmod、changefreq、priority。
- 原子化部署到网站根目录并可选生成 sitemap.gz。
- 写回数据库，输出日志与状态报告。
核心函数：
- load_config(): 载入基础配置（站点地址、部署目录、排除规则等）
- setup_db(): 初始化 SQLite 表结构与索引
- allowed_by_robots(url): 使用 robots.txt 判定是否允许抓取
- normalize_url(url): URL 归一化（去碎片、过滤追踪参数）
- crawl_site(base_url): Playwright 并发 BFS 抓取，收集候选 URL
- fingerprint_html(html): 内容清洗与哈希，降低误报
- detect_changes(urls): 与历史指纹对比，标注新增、修改、删除
- build_sitemap(entries): 生成标准 XML
- deploy_sitemap(xml_path, web_root): 原子写入与权限校验
- main(): 调度入口，支持 CLI
3. 变更检测策略
为降低 SPA 环境中的误报率与漏报：
- 渲染级抓取：使用 Playwright 执行 JS，避免仅静态 HTML 的不足。
- DOM 清洗指纹：
- 移除易变区域：script/style/noscript，以及可配置选择器（如广告、浮动通知、时间戳等）。
- 规范化文本：合并空白、去掉多余换行与非语义字符。
- 哈希算法：SHA-256。
- 查询参数归一化：
- 移除跟踪参数（utm_*, gclid, fbclid 等）。
- 可配置保留重要查询键（如分页 ?page=）。
- 路由发现多源融合：
- 爬虫 BFS 链接发现。
- 可选路由清单（例如构建期间导出 routes.json，由前端写入 /__routes.json）。
- 可选业务数据源（如 /api/posts 获取 slug 列表，拼装 /posts/{slug}）。
- 持久化与阈值：
- SQLite 存储 URL 与内容哈希、上次时间戳。
- 修改判断：哈希变化即判定为实质性修改（可扩展为相似度阈值）。
- 删除检测：
- 本次未发现但历史存在的 URL 视为下线，不再生成到 sitemap（可选保留以避免硬删除）。
4. 网站地图生成机制
生成的 XML 文件满足标准：
- 根元素：urlset，xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
- 每个 URL 元素包含：
- loc：完整绝对地址
- lastmod：ISO 8601 时间（UTC）
- changefreq：根据页面类型可配置（默认每日/每周）
- priority：默认 0.5，可根据深度/类型调整（首页更高）
- 对于大站：
- 可拆分多个 sitemapN.xml 并生成 sitemap_index.xml。
- 可选压缩：生成 sitemap.xml.gz。
5. 部署方案
- 原子写入：先写入临时文件（例如 sitemap.xml.tmp），完成后使用 mv 原子替换。
- 路径：部署到网站根目录（例如 /var/www/example.com/html/sitemap.xml）。
- 权限与所有者：确保 web 用户可读（例如 chmod 644，chown www-data:www-data）。
- 健康检查：完成后使用 HTTP HEAD 请求检测可访问性。
- CI/CD 集成：构建完成后触发脚本或在 VPS 上定时运行。
6. 定期执行建议
- 触发策略组合：
- 构建后：在前端构建发布脚本结束时运行（确保路由最新）。
- 定时：每天 02:00 运行（低流量时段），对于内容更新频繁的站点每 6 小时一次。
- Linux Cron 示例：
# 编辑 crontab
crontab -e
# 每天 02:00 执行
0 2 * * * /usr/bin/python3 /opt/sitemap/sitemap_updater.py \
--base-url https://example.com \
--web-root /var/www/example.com/html \
--max-depth 3 >> /var/log/sitemap_updater.log 2>&1
7. 性能优化方案
- 并发爬取：Playwright 多页面上下文 + 限流（Semaphore）。
- URL 去重与缓存：已抓取/失败 URL 集合；对静态资源忽略。
- 动态内容数据源：引入 API 端点列表，避免页面层级过深导致漏抓。
- 选择器排除优化：针对站点自定义排除动态区域（如实时组件、用户个性化模块），降低误报。
- 站点规模适配：
- 大型站分区：按路径前缀分批执行。
- Sitemap 切分与索引：避免单文件过大（>50,000 URL 或 >50MB）。
- 失败重试：对偶发网络错误进行退避重试。
- 监控与告警：运行失败或变更量异常时邮件/Webhook 报警。
Python 脚本示例
该脚本适配 React SPA，支持 Playwright 渲染抓取、变更检测与 XML 网站地图生成与部署。你可根据站点结构微调参数与排除选择器。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import argparse
import hashlib
import logging
import os
import re
import shutil
import sqlite3
import time
from datetime import datetime, timezone
from typing import Dict, List, Set, Tuple, Optional
from urllib.parse import urlparse, urljoin, urlunparse, parse_qsl, urlencode
import requests
from bs4 import BeautifulSoup
from lxml import etree
from urllib.robotparser import RobotFileParser
from playwright.async_api import async_playwright
# ----------------------------
# 配置与常量
# ----------------------------
DEFAULT_EXCLUDE_QUERY_KEYS = {
"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
"gclid", "fbclid", "msclkid", "ref", "ref_src"
}
DEFAULT_KEEP_QUERY_KEYS = {"page"} # 可根据业务保留分页等查询键
DEFAULT_EXCLUDE_SELECTORS = [
"script", "style", "noscript",
# 常见动态或噪声区域（可按需调整）
"header .timestamp", ".ad", ".ads", ".advert", ".cookie-banner",
".toast", ".notification", ".modal", ".overlay", ".live-counter"
]
DB_DIR = os.path.join(os.path.expanduser("~"), ".sitemap_cache")
DB_PATH = os.path.join(DB_DIR, "sitemap.db")
# ----------------------------
# 工具函数
# ----------------------------
def ensure_dir(path: str):
if not os.path.exists(path):
os.makedirs(path, exist_ok=True)
def isoformat_now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def normalize_url(url: str, base_netloc: str) -> Optional[str]:
"""归一化 URL：同源过滤、去碎片、过滤追踪参数，仅保留白名单查询键。"""
try:
parsed = urlparse(url)
if not parsed.scheme:
return None
if parsed.netloc != base_netloc:
return None
# 去 fragment
fragmentless = parsed._replace(fragment="")
# 过滤查询参数
query_pairs = parse_qsl(fragmentless.query, keep_blank_values=True)
filtered = []
for k, v in query_pairs:
if k in DEFAULT_EXCLUDE_QUERY_KEYS:
continue
if k in DEFAULT_KEEP_QUERY_KEYS:
filtered.append((k, v))
new_query = urlencode(filtered, doseq=True)
normalized = fragmentless._replace(query=new_query)
return urlunparse(normalized)
except Exception:
return None
def sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8", errors="ignore")).hexdigest()
def clean_html_to_text(html: str, extra_selectors: List[str]) -> str:
"""清洗 HTML，移除噪声，输出规范化文本用于指纹。"""
soup = BeautifulSoup(html, "html.parser")
# 统一移除默认 + 额外的选择器
for selector in set(DEFAULT_EXCLUDE_SELECTORS + extra_selectors):
for tag in soup.select(selector):
tag.decompose()
# 删除隐藏元素
for tag in soup.select("[hidden], [aria-hidden=true]"):
tag.decompose()
# 规范化文本
text = soup.get_text(separator=" ", strip=True)
text = re.sub(r"\s+", " ", text).strip()
return text
# ----------------------------
# 数据库持久化
# ----------------------------
def setup_db():
ensure_dir(DB_DIR)
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS pages (
url TEXT PRIMARY KEY,
content_hash TEXT,
last_seen TEXT,
last_modified TEXT
);
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_pages_last_seen ON pages(last_seen);")
conn.commit()
return conn
def load_existing(conn) -> Dict[str, Tuple[str, str, str]]:
cur = conn.execute("SELECT url, content_hash, last_seen, last_modified FROM pages;")
return {row[0]: (row[1], row[2], row[3]) for row in cur.fetchall()}
def upsert_page(conn, url: str, content_hash: str, last_modified: Optional[str] = None):
now = isoformat_now()
if last_modified is None:
last_modified = now
conn.execute("""
INSERT INTO pages(url, content_hash, last_seen, last_modified)
VALUES(?, ?, ?, ?)
ON CONFLICT(url) DO UPDATE SET
content_hash=excluded.content_hash,
last_seen=excluded.last_seen,
last_modified=CASE
WHEN pages.content_hash != excluded.content_hash THEN excluded.last_modified
ELSE pages.last_modified
END;
""", (url, content_hash, now, last_modified))
def mark_seen(conn, url: str):
now = isoformat_now()
conn.execute("UPDATE pages SET last_seen=? WHERE url=?;", (now, url))
# ----------------------------
# 机器人协议
# ----------------------------
def load_robots(base_url: str) -> RobotFileParser:
parsed = urlparse(base_url)
robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
rp = RobotFileParser()
try:
rp.set_url(robots_url)
rp.read()
except Exception:
# 不可用则默认允许
rp = RobotFileParser()
rp.parse([])
return rp
def allowed_by_robots(rp: RobotFileParser, url: str, ua: str = "SitemapUpdater"):
try:
return rp.can_fetch(ua, url)
except Exception:
return True
# ----------------------------
# Playwright 抓取
# ----------------------------
class Crawler:
def __init__(self, base_url: str, max_depth: int = 3, concurrency: int = 4, extra_selectors: Optional[List[str]] = None):
self.base_url = base_url.rstrip("/")
self.parsed_base = urlparse(self.base_url)
self.base_netloc = self.parsed_base.netloc
self.max_depth = max_depth
self.concurrency = concurrency
self.extra_selectors = extra_selectors or []
self.visited: Set[str] = set()
self.discovered: Set[str] = set()
self.sem = asyncio.Semaphore(concurrency)
self.rp = load_robots(self.base_url)
async def crawl(self) -> Dict[str, Dict]:
"""返回 URL -> {hash, lastmod} 的映射。"""
results: Dict[str, Dict] = {}
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
try:
# BFS 队列 (url, depth)
queue: List[Tuple[str, int]] = [(self.base_url, 0)]
while queue:
url, depth = queue.pop(0)
if depth > self.max_depth:
continue
normalized = normalize_url(url, self.base_netloc)
if not normalized:
continue
if normalized in self.visited:
continue
if not allowed_by_robots(self.rp, normalized):
logging.debug(f"Disallowed by robots: {normalized}")
continue
# 并发控制
await self.sem.acquire()
asyncio.create_task(self._process_url(context, normalized, depth, queue, results))
# 等待所有任务完成
while self.sem._value != self.concurrency:
await asyncio.sleep(0.05)
finally:
await context.close()
await browser.close()
return results
async def _process_url(self, context, url: str, depth: int, queue: List[Tuple[str, int]], results: Dict[str, Dict]):
page = await context.new_page()
try:
logging.info(f"Crawling: {url} (depth={depth})")
response = await page.goto(url, wait_until="networkidle", timeout=30000)
# 有些 React SPA 总是 200，但我们仍继续提取
html = await page.content()
# Canonical URL
canonical = await page.evaluate("""
() => {
const link = document.querySelector('link[rel="canonical"]');
return link ? link.href : null;
}
""")
target_url = normalize_url(canonical, self.base_netloc) or url
# 内容指纹
text = clean_html_to_text(html, self.extra_selectors)
content_hash = sha256_text(text)
results[target_url] = {
"hash": content_hash,
"lastmod": isoformat_now()
}
self.visited.add(target_url)
self.discovered.add(target_url)
# 发现新链接
links = await page.evaluate("""
() => Array.from(document.querySelectorAll('a[href]'))
.map(a => a.href)
""")
for href in set(links or []):
norm = normalize_url(href, self.base_netloc)
if not norm:
continue
if norm in self.visited:
continue
# 过滤常见静态资源与登录页等（可扩展）
if re.search(r"\.(jpg|jpeg|png|gif|svg|webp|pdf|zip|rar|7z|mp4|mp3)$", norm, re.I):
continue
if re.search(r"/login|/auth|/signup", norm, re.I):
continue
queue.append((norm, depth + 1))
except Exception as e:
logging.warning(f"Failed to crawl {url}: {e}")
finally:
await page.close()
self.sem.release()
# ----------------------------
# 变更检测与网站地图
# ----------------------------
def detect_changes(conn, crawled: Dict[str, Dict]) -> Tuple[List[str], List[str], List[str]]:
"""返回 (new_urls, modified_urls, removed_urls)"""
existing = load_existing(conn)
new_urls, modified_urls = [], []
current_set = set(crawled.keys())
previous_set = set(existing.keys())
for url, info in crawled.items():
h = info["hash"]
if url not in existing:
new_urls.append(url)
upsert_page(conn, url, h, last_modified=info["lastmod"])
else:
old_hash = existing[url][0]
if old_hash != h:
modified_urls.append(url)
upsert_page(conn, url, h, last_modified=info["lastmod"])
else:
mark_seen(conn, url)
removed_urls = sorted(list(previous_set - current_set))
return new_urls, modified_urls, removed_urls
def build_sitemap_xml(base_url: str, urls: List[Tuple[str, str]], default_changefreq: str = "daily") -> bytes:
"""
urls: List of (url, lastmod)
返回 XML 字节数据
"""
NS = "http://www.sitemaps.org/schemas/sitemap/0.9"
urlset = etree.Element("urlset", xmlns=NS)
# 可按路径设置优先级
def priority_for(u: str) -> str:
if u.rstrip("/") == base_url.rstrip("/"):
return "1.0"
depth = len(urlparse(u).path.strip("/").split("/")) if urlparse(u).path.strip("/") else 1
return "0.5" if depth <= 2 else "0.3"
for u, lastmod in sorted(urls):
url_el = etree.SubElement(urlset, "url")
loc = etree.SubElement(url_el, "loc"); loc.text = u
lm = etree.SubElement(url_el, "lastmod"); lm.text = lastmod
cf = etree.SubElement(url_el, "changefreq"); cf.text = default_changefreq
pr = etree.SubElement(url_el, "priority"); pr.text = priority_for(u)
return etree.tostring(urlset, xml_declaration=True, encoding="utf-8", pretty_print=True)
def deploy_sitemap(xml_bytes: bytes, web_root: str, filename: str = "sitemap.xml"):
ensure_dir(web_root)
tmp_path = os.path.join(web_root, f".{filename}.tmp")
final_path = os.path.join(web_root, filename)
with open(tmp_path, "wb") as f:
f.write(xml_bytes)
# 原子替换
shutil.move(tmp_path, final_path)
# 权限（按需调整）
os.chmod(final_path, 0o644)
logging.info(f"Sitemap deployed: {final_path}")
# 健康检查（可选）
try:
# 通过 base_url 拼接确认
# 如果不易确定，可跳过或传入完整 URL
logging.info("Health check skipped (requires public URL).")
except Exception as e:
logging.warning(f"Health check failed: {e}")
# ----------------------------
# 主流程
# ----------------------------
async def run(base_url: str, web_root: str, max_depth: int, concurrency: int, extra_selectors: List[str]):
logging.info("Initializing database...")
conn = setup_db()
logging.info("Starting crawler...")
crawler = Crawler(base_url, max_depth=max_depth, concurrency=concurrency, extra_selectors=extra_selectors)
crawled = await crawler.crawl()
logging.info(f"Crawled URLs: {len(crawled)}")
new_urls, modified_urls, removed_urls = detect_changes(conn, crawled)
conn.commit()
logging.info(f"New: {len(new_urls)}, Modified: {len(modified_urls)}, Removed: {len(removed_urls)}")
# 构建网站地图（仅包含本次发现的所有有效 URL）
url_list = [(u, crawled[u]["lastmod"]) for u in sorted(crawled.keys())]
xml_bytes = build_sitemap_xml(base_url, url_list, default_changefreq="daily")
# 部署
deploy_sitemap(xml_bytes, web_root)
conn.close()
logging.info("Done.")
def parse_args():
ap = argparse.ArgumentParser(description="React site XML sitemap auto updater")
ap.add_argument("--base-url", required=True, help="站点入口 URL，例如 https://example.com")
ap.add_argument("--web-root", required=True, help="网站根目录，例如 /var/www/example.com/html")
ap.add_argument("--max-depth", type=int, default=3, help="爬取最大深度，默认 3")
ap.add_argument("--concurrency", type=int, default=4, help="并发页面数，默认 4")
ap.add_argument("--exclude-selectors", nargs="*", default=[], help="额外排除的 CSS 选择器")
ap.add_argument("--log-level", default="INFO", choices=["DEBUG","INFO","WARNING","ERROR"])
return ap.parse_args()
if __name__ == "__main__":
args = parse_args()
logging.basicConfig(level=getattr(logging, args.log_level), format="%(asctime)s %(levelname)s %(message)s")
try:
asyncio.run(run(
base_url=args.base_url,
web_root=args.web_root,
max_depth=args.max_depth,
concurrency=args.concurrency,
extra_selectors=args.exclude_selectors
))
except KeyboardInterrupt:
logging.warning("Interrupted by user")
使用与集成建议
python /opt/sitemap/sitemap_updater.py \
--base-url https://example.com \
--web-root /var/www/example.com/html \
--max-depth 3 \
--concurrency 6 \
--exclude-selectors ".live-ticker" ".user-panel"
-
与 React 构建集成：
- 在构建流程中导出路由清单（如 routes.json），并由脚本定制读取该清单的 URL 追加到
crawler.discovered 集合，以减少对页面点击路径的依赖。
- 对动态页面（如博客、商品详情），建议提供 API 列表端点，脚本在抓取前预加载，拼装目标 URL 加入队列。
-
robots.txt：
如需进一步定制（例如多 sitemap 拆分、优先级策略、API 路由聚合、在 CI/CD 内触发执行），可告知具体站点结构与规模，我将提供针对性的增强脚本。
