BakeHub 外部 Markdown 发帖工具
-
背景:为什么要写一个外部发帖工具?
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,每次会话变化 loggedIntrue/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.txt2. 查看版块列表:
python3 bakehub-poster.py cats # cid name slug topics # ------------------------------------------------------------ # 1 项目复盘 1/项目复盘 3 # 2 开源求助 2/开源求助 5 # 4 学习路线 4/学习路线 2 # 3 经验分享 3/经验分享 63. 用本地 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 这个脚本改一版发上来。
- 图片上传:NodeBB 的上传接口是