1. 准备工作：获取 API 密钥
在编写脚本前，您需要从 Discourse 管理后台获取授权凭证：
- 管理员登录：以管理员身份进入论坛。
- 进入 API 设置：依次点击 设置 (Admin) → API → API 密钥 (API Keys)。
- 新建密钥：点击 添加 API 密钥 (New API Key)。
- 配置权限：
- 描述：填写用途（如：数据采集）。
- 用户：通常选
system或指定管理员。 - 权限等级：根据需求选择（获取全站信息建议选“全局”）。
- 保存：点击生成并妥善保存
API-Key。
2. 自动化脚本实现
Discourse 对 API 请求有频率限制。下面的脚本通过 _request_json 方法自动处理 429 Too Many Requests 错误，确保大规模采集时不被中断。
import requests
class DiscourseClient:
def __init__(self, base_url, api_key, api_user):
self.base_url = base_url.rstrip("/")
self.headers = {
"Api-Key": api_key,
"Api-Username": api_user
}
def _request_json(self, url, params=None, max_retries=5):
"""带重试机制的请求方法"""
for i in range(max_retries):
resp = requests.get(url, headers=self.headers, params=params)
if resp.status_code == 429:
# 获取服务器建议的等待时间，如果没有则默认等待 5 秒
wait_time = int(resp.headers.get("Retry-After", 5))
print(f"⚠️ 触发速率限制，等待 {wait_time} 秒后重试 (尝试 {i+1}/{max_retries})...")
time.sleep(wait_time)
continue
resp.raise_for_status()
return resp.json()
raise Exception(f"在重试 {max_retries} 次后仍然失败。")
def get_all_categories(self):
"""获取论坛所有分类"""
url = f"{self.base_url}/categories.json"
data = self._request_json(url)
return data.get("category_list", {}).get("categories", [])
def _get_all_topics_by_category(self, slug, category_id):
"""获取指定分类下的所有话题（分页累加）"""
all_topics = []
page = 0
while True:
url = f"{self.base_url}/c/{slug}/{category_id}.json"
data = self._request_json(url, params={"page": page})
topics = data.get("topic_list", {}).get("topics", [])
if not topics:
break
all_topics.extend(topics)
page += 1
return all_topics
def get_all_topics(self):
"""获取论坛所有分类下的所有话题"""
all_topics = []
categories = self.get_all_categories()
for cat in categories:
slug = cat["slug"]
category_id = cat["id"]
topics = self._get_all_topics_by_category(slug, category_id)
all_topics.extend(topics)
print(f"话题：{slug}有{len(topics)}个帖子")
return all_topics
def get_topic(self, topic_id):
"""获取单个话题详情"""
url = f"{self.base_url}/t/{topic_id}.json"
return requests.get(url, headers=self.headers).json()
if __name__ == '__main__':
base_url = '网址信息'
api_key = 'api密钥'
api_user = '用户名'
discourse= DiscourseClient(base_url, api_key, api_user)
# 获取所有分类
categories = discourse.get_all_categories()
for cat in categories:
slug = cat["slug"] #获取分类名
category_id = cat["id"] #获取分类id
topics = discourse._get_all_topics_by_category(slug, category_id) #获取所有的话题
if not topics:
continue
for topic in topics:
topic_id = topic["id"] #获取话题的id
posts_data = discourse.get_posts_from_topic(topic_id) #获取话题中所有的帖子
if not posts_data:
continue
topic_title = posts_data.get('topic_title', '') #获取标题
main_post = posts_data.get('main_post') #获取主贴
author = posts_data['main_post']['author'] #获取作者
content = main_post['content'] #获取主贴内容
reply = posts_data.get('replies', []) #获取回复