149 lines
6.1 KiB
Python
149 lines
6.1 KiB
Python
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}") |