前言
最近在使用RSS的过程中遇到即时性的问题，发现很多不提供RSS的网站使用公共解决方案时，RSS内容更新都不够及时，对于想要把RSS作为主要获取信息来源的我来说非常的不方便，中间也尝试过一些方案，RSSHub就不用说了，虽然功能强大，但对于我这种不想要深度了解的人来说，始终满足不了我想要的效果，一键制作RSS的工具也使用过，例如RssEverything等，不是收费太贵，就是没法对非公开页面进行一键处理，直到最近看到篇帖子，于是来了灵感，按照帖子的思路举一反三，便有了本文的方案。
原理
利用AI智能体抓取已登录的Edge浏览器会话，然后将网页内容转换成RSS内容，再部署到Cloudflare Pages。
准备
一台Windows服务器/云电脑/24小时开机的电脑/云龙虾（MiMo Claw、Kimi Claw等）。
一个Cloudflare账号。
一个域名（可选）。
教程
1.设备安装AI智能体软件，这里推荐WorkBuddy，轻量不花钱也能用。
2.登录Cloudflare账号，获取帐户API令牌。
路径：管理账户-创建令牌-搜索权限组“Cloudflare Pages”-勾选Read和Edit权限-审核令牌即可。
3.将下列代码发给AI智能体学习后，告知其需要制作RSS的真实网页，并一起发送上方的API令牌。
"""
论坛 RSS 生成器 — 纯案例文件（已脱敏）
========================================
通过浏览器 CDP 协议复用已登录会话，自动化抓取论坛帖子列表，
生成标准 RSS 2.0 XML，并通过 CLI 工具部署到 Pages 服务。
适用场景:
- 任意 Discuz! 系论坛版块的 RSS 订阅生成
- 需登录才能访问的内容RSS化
- 静态 Pages 托管 + CDP 抓取 的轻量方案
使用前请替换所有 <<PLACEHOLDER>> 为实际值。
"""
import asyncio
import json
import os
import sys
import urllib.request
import time
import subprocess
import tempfile
import shutil
from datetime import datetime
from email.utils import format_datetime
# ═══════════════════════════════════════════════════════════════════════
# 配置区 — 按需修改
# ═══════════════════════════════════════════════════════════════════════
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_FILE = os.path.join(SCRIPT_DIR, "rss_log.txt")
# Pages 服务配置
PAGES_API_TOKEN = "<<YOUR_PAGES_API_TOKEN>>" # API 令牌
PAGES_ACCOUNT_ID = "<<YOUR_ACCOUNT_ID>>" # 账户 ID
PAGES_PROJECT_NAME = "<<YOUR_PAGES_PROJECT_NAME>>" # Pages 项目名称
PAGES_DOMAIN = "<<YOUR_PROJECT.pages.dev>>" # Pages 域名
# 文件路径（使用随机字符串防止猜测）
SUBDIR_L1 = "<<RANDOM_PATH_LEVEL1>>" # 第一层随机目录
SUBDIR_L2 = "<<RANDOM_PATH_LEVEL2>>" # 第二层随机目录
RSS_FILENAME = "<<RANDOM_FILENAME>>.xml" # 随机文件名
RSS_DEPLOY_URL = f"https://{PAGES_DOMAIN}/{SUBDIR_L1}/{SUBDIR_L2}/{RSS_FILENAME}"
# 目标论坛配置
FORUM_URL = "<<TARGET_FORUM_URL>>" # 目标版块URL
FORUM_TITLE = "<<FORUM_TITLE>>" # RSS频道标题
FORUM_LINK = "<<FORUM_BASE_URL>>" # 论坛主页链接
FORUM_DESC = "<<FORUM_DESCRIPTION>>" # RSS频道描述
BASE_URL = "<<FORUM_BASE_URL>>" # 论坛根URL
# ═══════════════════════════════════════════════════════════════════════
# 日志工具
# ═══════════════════════════════════════════════════════════════════════
def log(msg: str):
"""统一日志输出：终端 + 文件"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{ts}] {msg}"
sys.stdout.buffer.write((line + "\n").encode("utf-8", errors="replace"))
sys.stdout.buffer.flush()
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# ═══════════════════════════════════════════════════════════════════════
# 日期解析 — 适配 Discuz! 论坛日期格式
# ═══════════════════════════════════════════════════════════════════════
def parse_forum_date(date_str: str) -> str:
"""
将论坛日期格式转为 RSS 标准 RFC 2822 格式
输入: "2026-6-20 07:41" / "2026-06-20 07:41"
输出: "Sat, 20 Jun 2026 07:41:00 +0800"
"""
date_str = date_str.strip()
for fmt in ["%Y-%m-%d %H:%M", "%Y-%m-%d"]:
try:
dt = datetime.strptime(date_str, fmt)
return format_datetime(dt)
except ValueError:
continue
log(f"[WARN] 无法解析日期: {date_str}，使用当前时间")
return format_datetime(datetime.now())
# ═══════════════════════════════════════════════════════════════════════
# 浏览器连接 — 通过 Chrome/Edge CDP 端口复用已登录会话
# ═══════════════════════════════════════════════════════════════════════
def ensure_browser_cdp(cdp_port: int = 9222) -> str:
"""
确保浏览器以调试模式运行，返回 WebSocket CDP URL
前置条件（手动操作一次即可，浏览器会保持登录状态）:
- 启动浏览器并添加参数: --remote-debugging-port=<cdp_port>
- 在浏览器中手动登录目标论坛
"""
try:
data = urllib.request.urlopen(f"http://localhost:{cdp_port}/json/version", timeout=3)
info = json.loads(data.read())
log(f"浏览器已运行，CDP 已连接")
return info["webSocketDebuggerUrl"]
except Exception:
log(f"[FAIL] 浏览器调试端口 {cdp_port} 未就绪。请先启动浏览器调试模式")
sys.exit(1)
# ═══════════════════════════════════════════════════════════════════════
# XML 工具
# ═══════════════════════════════════════════════════════════════════════
def escape_xml(text: str) -> str:
"""转义 XML 五种特殊字符"""
return (text
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
.replace("'", "'"))
# ═══════════════════════════════════════════════════════════════════════
# 自定义异常 — 登录状态检测
# ═══════════════════════════════════════════════════════════════════════
class LoginExpiredError(Exception):
"""检测到未登录状态时抛出，避免生成空RSS"""
pass
# ═══════════════════════════════════════════════════════════════════════
# 核心抓取逻辑 — Playwright CDP 模式
# ═══════════════════════════════════════════════════════════════════════
async def scrape_forum(cdp_port: int = 9222):
"""
通过 CDP 连接浏览器，抓取论坛帖子列表
返回: [{"tid", "title", "href", "category", "author", "pubDate"}, ...]
登录失效时抛 LoginExpiredError
适配: Discuz! 论坛模板 (tbody[id^='normalthread_'])
"""
from playwright.async_api import async_playwright
ws_url = ensure_browser_cdp(cdp_port)
async with async_playwright() as pw:
browser = await pw.chromium.connect_over_cdp(ws_url)
log("已连接浏览器")
ctx = browser.contexts[0]
# 复用已有论坛标签页
existing_page = None
for p in ctx.pages:
if "<<FORUM_DOMAIN_KEYWORD>>" in p.url: # 替换为论坛域名关键词
existing_page = p
log(f"复用已有标签")
break
if existing_page:
page = existing_page
else:
page = await ctx.new_page()
log("创建新标签页")
log(f"导航到 {FORUM_URL}")
await page.goto(FORUM_URL, wait_until="domcontentloaded", timeout=20000)
await asyncio.sleep(3) # 等待 JS 渲染完毕
# ── 登录状态检测 ──
page_text = await page.inner_text("body")
# 根据目标论坛的未登录特征调整检测条件
if ("<<LOGIN_REQUIRED_TEXT1>>" in page_text or
"<<LOGIN_REQUIRED_TEXT2>>" in page_text):
await browser.close()
raise LoginExpiredError("检测到未登录状态")
log(f"页面标题: {await page.title()}")
# ── 解析帖子列表（Discuz! 通用 DOM 结构）──
posts = await page.evaluate("""() => {
const posts = [];
const tbodies = document.querySelectorAll("tbody[id^='normalthread_']");
tbodies.forEach(tbody => {
try {
const th = tbody.querySelector("th.common, th.new");
if (!th) return;
// 标题和链接
const titleLink = th.querySelector("a.s.xst, a.xst");
if (!titleLink) return;
const title = titleLink.textContent.trim();
const href = titleLink.getAttribute("href");
// 分类标签
const catEl = th.querySelector("em a");
const category = catEl ? catEl.textContent.trim() : "";
// 作者和发布日期
const resTi = th.querySelector("p.res-ti");
let author = "";
let pubDate = "";
if (resTi) {
const authorLink = resTi.querySelector("a");
if (authorLink) author = authorLink.textContent.trim();
const text = resTi.textContent;
const parts = text.split("\u2022"); // • 分隔符
if (parts.length > 1) pubDate = parts[1].trim();
}
// fallback: 从 td.by 获取作者
if (!author) {
const byCite = tbody.querySelector("td.by cite a");
if (byCite) author = byCite.textContent.trim();
}
// fallback: 从 td.by em span 获取日期
if (!pubDate) {
const dateSpan = tbody.querySelector("td.by em span");
if (dateSpan) pubDate = dateSpan.textContent.trim();
}
// 从 tbody id 提取帖子ID
const tidMatch = tbody.id.match(/normalthread_(\\d+)/);
const tid = tidMatch ? tidMatch[1] : "";
posts.push({ tid, title, href, category, author, pubDate });
} catch(e) {}
});
return posts;
}""")
log(f"抓取到 {len(posts)} 条帖子")
if not existing_page:
await page.close()
await browser.close()
return posts
# ═══════════════════════════════════════════════════════════════════════
# RSS 生成 — 标准 RSS 2.0 XML
# ═══════════════════════════════════════════════════════════════════════
def build_rss_xml(posts: list) -> str:
"""根据帖子列表生成标准 RSS 2.0 XML"""
now_rfc2822 = format_datetime(datetime.now())
items_xml = []
for post in posts:
link = (post["href"] if post["href"].startswith("http")
else BASE_URL.rstrip("/") + "/" + post["href"].lstrip("/"))
pub_date = parse_forum_date(post["pubDate"]) if post["pubDate"] else now_rfc2822
category_prefix = f"[{post['category']}] " if post["category"] else ""
title_full = f"{category_prefix}{post['title']}"
description = f"<p>{escape_xml(title_full)}</p>"
if post["author"]:
description += f"<p>作者: {escape_xml(post['author'])}</p>"
description += f'<p><a href="{escape_xml(link)}">点击查看原文</a></p>'
item = f""" <item>
<title><![CDATA[{escape_xml(title_full)}]]></title>
<link>{escape_xml(link)}</link>
<guid isPermaLink="true">{escape_xml(link)}</guid>
<pubDate>{pub_date}</pubDate>
<description><![CDATA[{description}]]></description>
</item>"""
items_xml.append(item)
return f"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0"
xmlns:atom="http://www.w3.org/2005/Atom"
xmlns:content="http://purl.org/rss/1.0/modules/content/">
<channel>
<title><![CDATA[{escape_xml(FORUM_TITLE)}]]></title>
<link>{escape_xml(FORUM_LINK)}</link>
<description><![CDATA[{escape_xml(FORUM_DESC)}]]></description>
<language>zh-CN</language>
<lastBuildDate>{now_rfc2822}</lastBuildDate>
<pubDate>{now_rfc2822}</pubDate>
<generator>Forum RSS Generator (CDP)</generator>
<atom:link href="{escape_xml(RSS_DEPLOY_URL)}" rel="self" type="application/rss+xml"/>
{chr(10).join(items_xml)}
</channel>
</rss>"""
# ═══════════════════════════════════════════════════════════════════════
# 部署 — Pages 服务 CLI 部署
# ═══════════════════════════════════════════════════════════════════════
def deploy_to_pages(rss_content: str) -> bool:
"""
将文件部署到 Pages 服务（如 Cloudflare Pages / Netlify）
部署内容:
- index.html: 空白首页（防目录列表暴露）
- robots.txt: 全站禁止搜索引擎
- _headers: XML 文件 X-Robots-Tag: noindex
- {SUBDIR_L1}/{SUBDIR_L2}/{RSS_FILENAME}: RSS文件
"""
tmpdir = tempfile.mkdtemp(prefix="pages_deploy_")
# 空白首页
with open(os.path.join(tmpdir, "index.html"), "w", encoding="utf-8") as f:
f.write("""<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="utf-8"><title>RSS Service</title></head>
<body></body>
</html>""")
# robots.txt — 全站禁止搜索引擎爬取
with open(os.path.join(tmpdir, "robots.txt"), "w", encoding="utf-8") as f:
f.write("User-agent: *\nDisallow: /\n")
# _headers — XML 文件响应头防索引
with open(os.path.join(tmpdir, "_headers"), "w", encoding="utf-8") as f:
f.write(f"/{SUBDIR_L1}/{SUBDIR_L2}/{RSS_FILENAME}\n")
f.write(" X-Robots-Tag: noindex\n")
# 两层随机子目录 + RSS 文件
subdir_path = os.path.join(tmpdir, SUBDIR_L1, SUBDIR_L2)
os.makedirs(subdir_path, exist_ok=True)
with open(os.path.join(subdir_path, RSS_FILENAME), "w", encoding="utf-8") as f:
f.write(rss_content)
# ── 调用 Pages CLI 部署 ──
# 示例: Cloudflare Wrangler
# 替换为实际 CLI 路径和命令
log("通过 CLI 部署到 Pages 服务...")
try:
env = os.environ.copy()
env["PAGES_API_TOKEN"] = PAGES_API_TOKEN
env["PAGES_ACCOUNT_ID"] = PAGES_ACCOUNT_ID
# -- 替换为实际 CLI 可执行文件路径 --
node_path = "<<YOUR_NODE_PATH>>"
cli_path = "<<YOUR_CLI_PATH>>"
cmd = [
node_path, cli_path,
"pages", "deploy", tmpdir,
"--project-name", PAGES_PROJECT_NAME,
"--branch", "main"
]
result = subprocess.run(
cmd, env=env,
capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=120
)
output = (result.stdout or "") + (result.stderr or "")
log(output.strip())
if result.returncode != 0:
log(f"[FAIL] 部署失败 (exit {result.returncode})")
return False
log(f"[OK] 部署成功!")
log(f"[OK] RSS 地址: {RSS_DEPLOY_URL}")
except subprocess.TimeoutExpired:
log("[FAIL] 部署超时")
return False
except Exception as e:
log(f"[FAIL] 部署失败: {e}")
return False
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
return True
# ═══════════════════════════════════════════════════════════════════════
# 入口 — 含重试与异常处理
# ═══════════════════════════════════════════════════════════════════════
def main():
log("=" * 50)
log("论坛 RSS 生成器启动 (CDP + Pages 部署)")
log(f"目标: {RSS_DEPLOY_URL}")
try:
posts = None
for attempt in range(2):
try:
posts = asyncio.run(scrape_forum())
break
except LoginExpiredError as le:
if attempt == 0:
log(f"[WARN] 第1次抓取失败: {le}，5秒后重试...")
time.sleep(5)
else:
log(f"[FAIL] 第2次抓取仍失败: {le}")
log("[INFO] 登录状态失效，本次RSS更新已跳过")
sys.exit(0)
if not posts:
log("[FAIL] 未抓取到任何帖子")
sys.exit(1)
log(f"共抓取 {len(posts)} 条帖子，生成 RSS...")
rss_content = build_rss_xml(posts)
log(f"RSS 生成完毕，大小: {len(rss_content)} 字符")
# 本地备份
local_path = os.path.join(SCRIPT_DIR, RSS_FILENAME)
with open(local_path, "w", encoding="utf-8") as f:
f.write(rss_content)
log(f"本地备份: {local_path}")
# 部署
ok = deploy_to_pages(rss_content)
if ok:
log(f"[OK] 全部完成! RSS 地址: {RSS_DEPLOY_URL}")
else:
log("[FAIL] 部署失败")
sys.exit(1)
# 预览
log("\n--- 前5条帖子 ---")
for p in posts[:5]:
cat = f"[{p['category']}] " if p.get("category") else ""
log(f" {cat}{p['title']} | {p.get('author','?')} | {p.get('pubDate','?')}")
except Exception as e:
log(f"[FAIL] 脚本异常: {e}")
import traceback
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(traceback.format_exc())
sys.exit(1)
if __name__ == "__main__":
main()
注意
Cloudflare Pages默认域名可能无法直接访问，建议绑定自己的域名。