lklog | 龙鲲修改

前言

最近在使用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("&", "&amp;")
            .replace("<", "&lt;")
            .replace(">", "&gt;")
            .replace('"', "&quot;")
            .replace("'", "&apos;"))


# ═══════════════════════════════════════════════════════════════════════
# 自定义异常 — 登录状态检测
# ═══════════════════════════════════════════════════════════════════════

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()

前言

之前参加华为云的沃土计划，注册了华为云的开发者账号，后来发现华为云开发者每年会赠送800核时的云开发环境，可以免费使用开发桌面和虚拟机，也尝试了一下，感觉还挺不错的，但800核时算下来，用不了太长的时间，于是也就吃灰了。最近看到消息，说赠送的核时翻了10倍，于是就去看了一下，发现确实是的，感觉可玩性就变高了很多，这里简单记录一下，以备日后查阅。

领取

老用户（已经是华为云开发者）：

1.在华为云开发者官网进行登录（https://developer.huaweicloud.com/）。

2.鼠标移动到官网顶部的开发按钮，在弹出的页面中选择开发者空间。

3.在左边侧边栏的开发平台中选择云开发环境。

4.这时你会发现已经自动到账了8000核时（图片不足8000核时是因为我测试用了一点）。

如果是新用户（未注册华为云开发者），需要先在华为云开发者官网进行注册（https://developer.huaweicloud.com/）。注册好之后，进行开发者实名认证，大概流程为：选择个人开发者类型，再进行实名认证，之后的领取流程则和上面老用户一样。

使用

云开发环境支持创建的方式有三种：容器、虚拟机、开发桌面，标题所说的云电脑为创建开发桌面。

具体步骤如下：

1.在云开发环境中点击开发桌面，选择创建。

2.鼠标点击立即进入按钮，选择进入桌面（该桌面的开机、关机、删除在点击更多按钮后显示）。

关于核时的计费：

账单按照小时维度进行更新，账单单位为核时。

假设用户创建了一个四核的容器开发环境，并在开机后一小时后关机，那么将会产生 4（核）* 1（小时）= 4（核时）的消耗。

8000核时如果只使用4核的容器可以用2000小时，如果只使用2核时的容器可以用4000个小时（差不多6个月）。

注意：建议使用完记得及时关机以节省核时。

参考

《费用账单》