跳转至内容
BakeHub
BakeHub 工科开源社群
  • 登录
  • 首页
  • 项目
  • 论坛
  • 活动
  • 关于
  • AI 助手
  • 首页
  • 项目
  • 论坛
  • 活动
  • 关于
  • AI 助手

BakeHub 外部 Markdown 发帖工具

已定时 已固定 已锁定 已移动 经验分享
pythonnodebb自动化
1 帖子 1 发布者 49 浏览 1 关注中
  • 从旧到新
  • 从新到旧
  • 最多赞同
回复
  • 在新帖中回复
登录后回复
此主题已被删除。只有拥有主题管理权限的用户可以查看。
  • KongCharK 离线
    KongCharK离线
    KongChar
    编写于最后由 KongChar 编辑
    #1

    背景:为什么要写一个外部发帖工具?

    BakeHub 是基于 NodeBB 搭建的工科开源社群,日常发帖通常要在网页编辑器里手敲 Markdown。但当你有一篇几千字的技术复盘、或者想批量迁移本地笔记库时,网页编辑器就成了瓶颈——没有本地版本控制、不能 vim 键位、不能 grep、不能 diff。于是我用 F12 抓了一下 NodeBB 的发帖请求,顺手写了一个 零依赖 的 Python 脚本,直接用本地 Markdown 文件发帖。这篇帖子既是工具发布,也是把抓包分析过程完整记录下来,方便后来人复用。

    第一步:F12 抓包分析 NodeBB 发帖流程

    打开 BakeHub 首页,按 F12 切到 Network 面板,勾上 Preserve log,然后手动走一遍「登录 → 进版块 → 发帖」的流程。关键请求按时间顺序如下:

    1. 登录请求

    POST /login HTTP/1.1
    Host: 203.195.163.189
    Content-Type: application/x-www-form-urlencoded
    x-csrf-token: <从 /api/config 拿到的 token>
    Referer: http://203.195.163.189/
    
    username=KongChar&password=********&remember=on&returnTo=
    

    服务器返回 {"next":"/"} 并 Set-Cookie: express.sid=...,这个 session cookie 就是后续所有请求的身份凭证。注意 CSRF token 必须通过 x-csrf-token 请求头 传回去,而不是放在 body 里——这是 NodeBB 的硬性要求,漏掉会直接 403。

    2. 获取 CSRF token

    GET /api/config HTTP/1.1
    Accept: application/json
    

    返回的 JSON 里关键字段:

    字段 含义
    csrf_token 写操作必须带的 token,每次会话变化
    loggedIn true/false,判断当前 session 是否已登录
    uid 当前用户 id,游客为 0
    maximumPostLength 单帖最大字符数(本站 32767)
    maximumTagsPerTopic 每帖最多标签数(本站 5)

    3. 发帖请求(NodeBB v3 API)

    POST /api/v3/topics HTTP/1.1
    Content-Type: application/json
    x-csrf-token: <同上>
    
    {
      "cid": 3,
      "title": "帖子标题",
      "content": "Markdown 正文",
      "tags": ["工具", "Python"]
    }
    

    返回 {"topic": {"tid": 123, "slug": "123/xxx", ...}},帖子 URL 就是 /topic/{tid}/{slug}。

    4. 列版块

    GET /api/categories
    

    返回 categories 数组,每项含 cid / name / slug / topic_count。本站当前版块:

    cid 名称 用途
    1 项目复盘 复盘总结
    2 开源求助 提问
    3 经验分享 技术分享(本帖所在)
    4 学习路线 路线规划

    第二步:踩过的坑

    抓包看似简单,实际写脚本时踩了三个坑,值得单独说一下:

    坑 1:Connection reset by peer。 服务器对短时间高频请求会直接 RST 连接,第一次跑 cats 子命令就挂了。解决办法是给底层请求加 指数退避重试,失败后 sleep 1.5s / 3s / 4.5s 再试。

    坑 2:Accept 头重复。 一开始在 opener.addheaders 里写了 Accept,又在 _request 里 append 一次,导致请求头里出现两个 Accept,服务器偶尔会拒绝。统一在 _request 里追加,addheaders 只放 User-Agent 和 Referer。

    坑 3:CSRF token 会过期。 长时间不操作后再发帖,会收到 403 [[error:invalid-csrf-token]]。所以每次发帖前都重新 GET /api/config 刷新一次 token,而不是登录时拿一次就缓存到底。

    第三步:完整工具代码

    下面是 bakehub-poster.py 的完整源码,零第三方依赖,只用 Python 标准库。把它存成文件直接 python3 bakehub-poster.py 就能跑。

    #!/usr/bin/env python3
    """
    bakehub-poster.py — BakeHub (NodeBB) 外部 Markdown 发帖工具
    
    用法:
        python bakehub-poster.py post --file post.md --title "帖子标题" --cid 3 --tags "工具,Python"
        python bakehub-poster.py cats                      # 列出所有版块
        python bakehub-poster.py login                     # 仅登录并保存会话
    
    依赖: 仅 Python 标准库 (urllib / http / json / argparse)
    作者: KongChar
    """
    
    import argparse
    import http.cookiejar
    import json
    import os
    import time
    import urllib.parse
    import urllib.request
    
    # ===== 配置区 =====
    BASE_URL = "http://203.195.163.189"
    DEFAULT_CID = 3  # 经验分享
    COOKIE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bakehub.cookies.txt")
    UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
          "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
    
    
    class BakeHubClient:
        def __init__(self, base_url=BASE_URL):
            self.base = base_url.rstrip("/")
            self.cj = http.cookiejar.MozillaCookieJar(COOKIE_FILE)
            if os.path.exists(COOKIE_FILE):
                self.cj.load(ignore_discard=True, ignore_expires=True)
            self.opener = urllib.request.build_opener(
                urllib.request.HTTPCookieProcessor(self.cj))
            self.opener.addheaders = [
                ("User-Agent", UA),
                ("Referer", self.base + "/"),
            ]
            self.csrf = ""
    
        # ---------- 底层请求(带重试) ----------
        def _request(self, path, method="GET", data=None, headers=None,
                     content_type="application/x-www-form-urlencoded",
                     accept_json=True, retries=3):
            url = self.base + path
            last_err = None
            for attempt in range(retries):
                body = None
                h = list(self.opener.addheaders)
                if accept_json:
                    h.append(("Accept", "application/json"))
                if headers:
                    h.extend(headers.items())
                if data is not None:
                    if isinstance(data, (dict, list)):
                        body = json.dumps(data).encode("utf-8")
                        h.append(("Content-Type", "application/json"))
                    elif isinstance(data, str):
                        body = data.encode("utf-8")
                        h.append(("Content-Type", content_type))
                    else:
                        body = data
                        h.append(("Content-Type", content_type))
                req = urllib.request.Request(url, data=body, method=method)
                for k, v in h:
                    req.add_header(k, v)
                try:
                    with self.opener.open(req, timeout=30) as r:
                        return r.status, r.read().decode("utf-8", errors="replace")
                except urllib.error.HTTPError as e:
                    return e.code, e.read().decode("utf-8", errors="replace")
                except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
                    last_err = e
                    if attempt < retries - 1:
                        time.sleep(1.5 * (attempt + 1))
                        continue
            raise RuntimeError(f"请求 {path} 失败(重试 {retries} 次): {last_err}")
    
        # ---------- 公开方法 ----------
        def fetch_csrf(self):
            """从 /api/config 拿 CSRF token"""
            status, text = self._request("/api/config")
            if status != 200:
                raise RuntimeError(f"获取 CSRF 失败: HTTP {status} {text[:200]}")
            cfg = json.loads(text)
            self.csrf = cfg.get("csrf_token", "")
            if not self.csrf:
                raise RuntimeError("CSRF token 为空")
            return self.csrf
    
        def login(self, username, password):
            """登录并保存 cookie"""
            self.fetch_csrf()
            form = urllib.parse.urlencode({
                "username": username,
                "password": password,
                "remember": "on",
                "returnTo": "",
            })
            status, text = self._request(
                "/login", method="POST", data=form,
                headers={"x-csrf-token": self.csrf})
            if status != 200:
                raise RuntimeError(f"登录失败: HTTP {status} {text[:200]}")
            try:
                resp = json.loads(text)
            except json.JSONDecodeError:
                resp = {"raw": text}
            # 保存 cookie
            self.cj.save(COOKIE_FILE, ignore_discard=True, ignore_expires=True)
            # 验证登录
            self.fetch_csrf()
            status2, text2 = self._request("/api/config")
            cfg = json.loads(text2)
            if not cfg.get("loggedIn"):
                raise RuntimeError("登录后仍为游客,请检查账号密码")
            return resp
    
        def list_categories(self):
            status, text = self._request("/api/categories")
            if status != 200:
                raise RuntimeError(f"获取版块失败: HTTP {status}")
            data = json.loads(text)
            return data.get("categories", [])
    
        def create_topic(self, cid, title, content, tags=None):
            """创建新主题。content 为 Markdown 文本"""
            if not self.csrf:
                self.fetch_csrf()
            payload = {
                "cid": int(cid),
                "title": title,
                "content": content,
                "tags": tags or [],
            }
            status, text = self._request(
                "/api/v3/topics", method="POST", data=payload,
                headers={"x-csrf-token": self.csrf})
            if status not in (200, 201):
                raise RuntimeError(f"发帖失败: HTTP {status} {text[:300]}")
            return json.loads(text)
    
        def reply(self, tid, content):
            if not self.csrf:
                self.fetch_csrf()
            payload = {"tid": int(tid), "content": content}
            status, text = self._request(
                "/api/v3/topics/" + str(tid), method="POST", data=payload,
                headers={"x-csrf-token": self.csrf})
            if status not in (200, 201):
                raise RuntimeError(f"回复失败: HTTP {status} {text[:300]}")
            return json.loads(text)
    
    
    # ===== CLI =====
    def cmd_login(args):
        c = BakeHubClient()
        r = c.login(args.user, args.password)
        print("[OK] 登录成功,会话已保存到", COOKIE_FILE)
        print("    返回:", r)
    
    
    def cmd_cats(args):
        c = BakeHubClient()
        cats = c.list_categories()
        print(f"{'cid':>4}  {'name':<20}  {'slug':<20}  topics")
        print("-" * 60)
        for cat in cats:
            print(f"{cat.get('cid'):>4}  {cat.get('name'):<20}  "
                  f"{cat.get('slug'):<20}  {cat.get('topic_count')}")
    
    
    def cmd_post(args):
        with open(args.file, "r", encoding="utf-8") as f:
            content = f.read()
        tags = [t.strip() for t in (args.tags or "").split(",") if t.strip()]
        c = BakeHubClient()
        # 自动登录(若会话失效)
        try:
            c.fetch_csrf()
        except Exception:
            c.login(args.user, args.password)
        # 二次确认登录态
        status, text = c._request("/api/config")
        cfg = json.loads(text)
        if not cfg.get("loggedIn"):
            print("[INFO] 会话已过期,重新登录...")
            c.login(args.user, args.password)
        result = c.create_topic(args.cid, args.title, content, tags)
        print("[OK] 发帖成功!")
        if isinstance(result, dict):
            topic = result.get("topic", result)
            slug = topic.get("slug") or topic.get("tid")
            print("    tid:", topic.get("tid"))
            print("    URL:", f"{BASE_URL}/topic/{topic.get('tid')}/{slug}")
    
    
    def main():
        p = argparse.ArgumentParser(
            description="BakeHub (NodeBB) 外部 Markdown 发帖工具")
        p.add_argument("--user", default="KongChar", help="用户名")
        p.add_argument("--password", default="csj123456", help="密码")
        sub = p.add_subparsers(dest="cmd", required=True)
    
        sub.add_parser("login", help="仅登录并保存会话").set_defaults(func=cmd_login)
        sub.add_parser("cats", help="列出所有版块").set_defaults(func=cmd_cats)
    
        pp = sub.add_parser("post", help="发布 Markdown 帖子")
        pp.add_argument("--file", required=True, help="Markdown 文件路径")
        pp.add_argument("--title", required=True, help="帖子标题")
        pp.add_argument("--cid", type=int, default=DEFAULT_CID, help="版块 cid")
        pp.add_argument("--tags", help="标签,逗号分隔")
        pp.set_defaults(func=cmd_post)
    
        args = p.parse_args()
        args.func(args)
    
    
    if __name__ == "__main__":
        main()
    

    第四步:使用方法

    把上面的代码存为 bakehub-poster.py,然后:

    1. 首次登录(保存会话,后续免登录):

    python3 bakehub-poster.py login
    # 输出: [OK] 登录成功,会话已保存到 .../bakehub.cookies.txt
    

    2. 查看版块列表:

    python3 bakehub-poster.py cats
    #  cid  name        slug          topics
    # ------------------------------------------------------------
    #     1  项目复盘    1/项目复盘       3
    #     2  开源求助    2/开源求助       5
    #     4  学习路线    4/学习路线       2
    #     3  经验分享    3/经验分享       6
    

    3. 用本地 Markdown 文件发帖:

    cat > my-post.md <<'EOF'
    ## 我的第一篇外部发帖
    
    这是用脚本发的帖子,正文支持完整 Markdown 语法。
    
    - 列表
    - **加粗** / *斜体* / `代码`
    - [链接](https://example.com)
    EOF
    
    python3 bakehub-poster.py post \
        --file my-post.md \
        --title "我的第一篇外部发帖" \
        --cid 3 \
        --tags "工具,Python,自动化"
    # 输出:
    # [OK] 发帖成功!
    #     tid: 17
    #     URL: http://203.195.163.189/topic/17/我的第一篇外部发帖
    

    这篇帖子本身就是用这个工具发出来的 —— 自举成功。脚本、Markdown 源文件、cookie 缓存都在本地,可以 git 管理,可以 diff 历史版本,再也不用在网页编辑器里反复粘贴了。

    设计取舍说明

    为什么不用 requests 库?因为想保证 零依赖,任何一台装了 Python 3 的机器都能直接跑,不用 pip install。urllib 的 API 确实啰嗦,但封装一层 _request 之后用起来和 requests 差不多。

    为什么 cookie 用 Netscape 格式落盘?因为 http.cookiejar.MozillaCookieJar 是标准库原生支持的,文件可以直接被 curl -b 复用,方便调试——脚本跑挂了可以手动 curl -b bakehub.cookies.txt ... 排查。

    为什么 CSRF token 每次都重新取?因为 NodeBB 的 token 会随 session 状态变化,缓存反而更容易踩 403。一次 GET /api/config 才几 KB,代价可以忽略。

    后续可扩展的方向

    当前版本只覆盖了「登录 + 列版块 + 发主题 + 回复」四个核心动作。如果有人想接着做,几个方向值得考虑:

    • 图片上传:NodeBB 的上传接口是 POST /api/v3/topics/upload,需要 multipart/form-data,目前脚本没实现,长帖里的图片还得手动传。
    • 草稿同步:把本地 .md 文件的 frontmatter(标题、tags、cid)解析出来,做到一篇文件 = 一篇帖子,git diff 就能看出改了啥。
    • 批量迁移:遍历一个目录下所有 .md 文件,按 frontmatter 里的 cid 自动分发到对应版块,适合从 Obsidian / Hexo 仓库一键搬迁。
    • WebSocket 监听:NodeBB 用 socket.io 推实时通知,接上之后可以做到「别人回复你的帖子时本地弹通知」。

    欢迎在评论区交流改进思路,或者直接 fork 这个脚本改一版发上来。

    1 条回复最后回复
    回复 引用
    顶贴1踩帖

    你好!看起来您对这段对话很感兴趣,但您还没有一个账号。

    厌倦了每次访问都刷到同样的帖子?您注册账号后,您每次返回时都能精准定位到您上次浏览的位置,并可选择接收新回复通知(通过邮件或推送通知)。您还能收藏书签、为帖子顶,向社区成员表达您的欣赏。

    有了你的建议,这篇帖子会更精彩哦 💗

    注册 登录

    BakeHub
    • 第一个帖子
      最后一个帖子
    0
    • 版块
    • 最新
    • 标签
    • 热门
    • 世界
    • 用户
    • 群组