feat: 添加 NodeBB API 客户端 SDK 以支持主题创建和回复功能
This commit is contained in:
Ching L
parent
1221a7c771
commit
15d43b1c78
149
nodebb_sdk.py
Normal file
149
nodebb_sdk.py
Normal file
@ -0,0 +1,149 @@
|
||||
import requests
|
||||
from typing import Dict, Any
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class NodeBBAPIError(Exception):
|
||||
"""自定义 NodeBB API 错误异常"""
|
||||
def __init__(self, message: str, status_code: int = None, response_text: str = None):
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
self.response_text = response_text
|
||||
|
||||
def __str__(self):
|
||||
if self.status_code:
|
||||
return f"{self.status_code}: {super().__str__()} | Response: {self.response_text}"
|
||||
return super().__str__()
|
||||
|
||||
class NodeBBClient:
|
||||
"""
|
||||
一个用于与 NodeBB v3 API 交互的客户端 SDK。
|
||||
"""
|
||||
def __init__(self, base_url: str, api_token: str):
|
||||
"""
|
||||
初始化 NodeBB 客户端。
|
||||
|
||||
:param base_url: 你的 NodeBB 论坛 URL (例如: "https://forum.example.com")
|
||||
:param api_token: 你在 NodeBB 后台生成的 API Bearer Token
|
||||
"""
|
||||
if not base_url or not api_token:
|
||||
raise ValueError("base_url 和 api_token 不能为空")
|
||||
|
||||
# 确保 URL 末尾没有斜杠，方便拼接
|
||||
self.base_url = base_url.rstrip('/')
|
||||
self._headers = {
|
||||
"Authorization": f"Bearer {api_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
一个内部辅助方法，用于发送 POST 请求。
|
||||
|
||||
:param endpoint: API 的端点 (例如: "/api/v3/topics")
|
||||
:param payload: 要发送的 JSON 数据
|
||||
:return: API 响应的 JSON 字典
|
||||
:raises NodeBBAPIError: 如果 API 请求失败
|
||||
"""
|
||||
api_url = f"{self.base_url}{endpoint}"
|
||||
try:
|
||||
response = requests.post(api_url, headers=self._headers, json=payload, timeout=15)
|
||||
# 检查 HTTP 状态码，如果不是 2xx，则抛出异常
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.exceptions.HTTPError as e:
|
||||
# 捕获 HTTP 错误，并用我们的自定义异常包装它
|
||||
raise NodeBBAPIError(
|
||||
message=f"API 请求返回了一个错误",
|
||||
status_code=e.response.status_code,
|
||||
response_text=e.response.text
|
||||
) from e
|
||||
except requests.exceptions.RequestException as e:
|
||||
# 捕获其他请求相关的错误 (如网络问题)
|
||||
raise NodeBBAPIError(f"请求失败: {e}") from e
|
||||
|
||||
def create_topic(self, cid: int, title: str, content: str) -> Dict[str, Any]:
|
||||
"""
|
||||
在指定板块创建一个新的主题 (发帖)。
|
||||
|
||||
:param cid: 板块 ID (Category ID)
|
||||
:param title: 主题标题
|
||||
:param content: 主题内容 (支持 Markdown)
|
||||
:return: 成功时返回 API 响应的 JSON 数据
|
||||
:raises NodeBBAPIError: 如果 API 请求失败
|
||||
"""
|
||||
logger.info(f"SDK: 正在向板块 (cid={cid}) 发布新主题: '{title}'...")
|
||||
payload = {
|
||||
"cid": cid,
|
||||
"title": title,
|
||||
"content": content
|
||||
}
|
||||
response_data = self._post("/api/v3/topics", payload)
|
||||
|
||||
logger.success("✅ SDK: 主题发布成功！")
|
||||
logger.info(f" - 主题 ID (tid): {response_data['response']['tid']}")
|
||||
logger.info(f" - 访问链接: {self.base_url}/topic/{response_data['response']['slug']}")
|
||||
return response_data['response']
|
||||
|
||||
def reply_to_topic(self, tid: int, content: str) -> Dict[str, Any]:
|
||||
"""
|
||||
回复一个已存在的主题 (跟帖)。
|
||||
|
||||
:param tid: 主题 ID (Topic ID)
|
||||
:param content: 回复内容 (支持 Markdown)
|
||||
:return: 成功时返回 API 响应的 JSON 数据
|
||||
:raises NodeBBAPIError: 如果 API 请求失败
|
||||
"""
|
||||
logger.info(f"SDK: 正在回复主题 (tid={tid})...")
|
||||
endpoint = f"/api/v3/topics/{tid}"
|
||||
payload = {
|
||||
"content": content
|
||||
}
|
||||
response_data = self._post(endpoint, payload)
|
||||
|
||||
logger.success("✅ SDK: 回复成功！")
|
||||
logger.info(f" - 新帖子 ID (pid): {response_data['response']['pid']}")
|
||||
logger.info(f" - 访问链接: {self.base_url}/post/{response_data['response']['pid']}")
|
||||
return response_data['response']
|
||||
|
||||
# --- 以下是如何使用这个 SDK 的示例 ---
|
||||
if __name__ == "__main__":
|
||||
# --- 配置区 ---
|
||||
NODEBB_FORUM_URL = "https://your-nodebb-forum.com" # 你的 NodeBB 论坛 URL
|
||||
NODEBB_API_TOKEN = "your_api_bearer_token_here" # 你的 API Bearer Token
|
||||
# --- 配置区结束 ---
|
||||
|
||||
# 1. 检查配置是否已填写
|
||||
if "your-nodebb-forum.com" in NODEBB_FORUM_URL or "your_api_bearer_token_here" in NODEBB_API_TOKEN:
|
||||
logger.error("❌ 请先在脚本中修改 'NODEBB_FORUM_URL' 和 'NODEBB_API_TOKEN' 的值。")
|
||||
else:
|
||||
# 2. 创建 NodeBBClient 实例
|
||||
try:
|
||||
client = NodeBBClient(base_url=NODEBB_FORUM_URL, api_token=NODEBB_API_TOKEN)
|
||||
|
||||
# --- 使用示例：发帖并立即回复 ---
|
||||
|
||||
# 3. 定义帖子内容
|
||||
category_id = 2 # 目标板块ID
|
||||
topic_title = "来自 NodeBB SDK 的测试"
|
||||
topic_content = "这是一个使用 `NodeBBClient` 类发送的主题。代码结构更清晰了！"
|
||||
|
||||
# 4. 使用 try...except 来捕获潜在的 API 错误
|
||||
try:
|
||||
# 5. 调用 create_topic 方法
|
||||
new_topic = client.create_topic(cid=category_id, title=topic_title, content=topic_content)
|
||||
|
||||
# 6. 从成功的响应中获取新主题的ID
|
||||
new_topic_id = new_topic['tid']
|
||||
|
||||
logger.success(f"\n新主题已创建 (tid={new_topic_id})，现在对其进行回复...")
|
||||
|
||||
# 7. 调用 reply_to_topic 方法
|
||||
client.reply_to_topic(tid=new_topic_id, content="这是来自 SDK 的自动沙发回复！🎉")
|
||||
|
||||
except NodeBBAPIError as e:
|
||||
logger.error("\n--- 操作失败 ---")
|
||||
logger.error(f"❌ 发生 API 错误: {e}")
|
||||
|
||||
except ValueError as e:
|
||||
logger.error(f"❌ 初始化客户端失败: {e}")
|
||||
Loading…
x
Reference in New Issue
Block a user