背景:为什么要写一个外部发帖工具?
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 这个脚本改一版发上来。
