From 4ad222cbc7b6ccc8b9af75b43d4b4d932b1571a7 Mon Sep 17 00:00:00 2001 From: ninemine <1371605831@qq.com> Date: Thu, 6 Nov 2025 16:16:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=8F=92=E4=BB=B6=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=E4=B8=AD=E7=94=B1=E4=BA=8E=E5=BC=95=E7=94=A8=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E7=9A=84=E9=87=8D=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CoreModules/models.py | 99 +++------------------------------ CoreModules/plugin_interface.py | 10 +++- CoreRouters/callback.py | 26 ++++++++- 3 files changed, 42 insertions(+), 93 deletions(-) diff --git a/CoreModules/models.py b/CoreModules/models.py index f9d8ba7..1368018 100644 --- a/CoreModules/models.py +++ b/CoreModules/models.py @@ -1,95 +1,14 @@ """数据模型定义""" from pydantic import BaseModel, Field -from typing import Optional, Dict, Any, List +from typing import * class CallbackRequest(BaseModel): - """WPS Callback请求模型""" - chatid: int = Field(..., description="会话ID") - creator: int = Field(..., description="发送者ID") - content: str = Field(..., description="消息内容") - reply: Optional[Dict[str, Any]] = Field(None, description="回复内容") - robot_key: str = Field(..., description="机器人key") - url: str = Field(..., description="callback地址") - ctime: int = Field(..., description="发送时间") - - -class TextMessage(BaseModel): - """文本消息""" - msgtype: str = "text" - text: Dict[str, str] - - @classmethod - def create(cls, content: str): - """创建文本消息""" - return cls(text={"content": content}) - - -class MarkdownMessage(BaseModel): - """Markdown消息""" - msgtype: str = "markdown" - markdown: Dict[str, str] - - @classmethod - def create(cls, text: str): - """创建Markdown消息""" - return cls(markdown={"text": text}) - - -class LinkMessage(BaseModel): - """链接消息""" - msgtype: str = "link" - link: Dict[str, str] - - @classmethod - def create(cls, title: str, text: str, message_url: str = "", btn_title: str = "查看详情"): - """创建链接消息""" - return cls(link={ - "title": title, - "text": text, - "messageUrl": message_url, - "btnTitle": btn_title - }) - - -class GameState(BaseModel): - """游戏状态基类""" - game_type: str - created_at: int - updated_at: int - - -class GuessGameState(GameState): - """猜数字游戏状态""" - game_type: str = "guess" - target: int = Field(..., description="目标数字") - attempts: int = Field(0, description="尝试次数") - guesses: list[int] = Field(default_factory=list, description="历史猜测") - max_attempts: int = Field(10, description="最大尝试次数") - - -class QuizGameState(GameState): - """问答游戏状态""" - game_type: str = "quiz" - question_id: int = Field(..., description="问题ID") - question: str = Field(..., description="问题内容") - attempts: int = Field(0, description="尝试次数") - max_attempts: int = Field(3, description="最大尝试次数") - - -class PrivateMessageRequest(BaseModel): - """私聊消息请求模型""" - user_id: int = Field(..., description="目标用户ID") - content: str = Field(..., description="消息内容") - msg_type: str = Field(default="text", description="消息类型: text 或 markdown") - - -class CheckBatchRequest(BaseModel): - """批量检查请求模型""" - user_ids: List[int] = Field(..., description="用户ID列表") - - -class CheckBatchResponse(BaseModel): - """批量检查响应模型""" - results: Dict[int, bool] = Field(..., description="用户ID到是否有URL的映射") - + """Callback请求模型""" + chatid: int = Field(default=0, description="会话ID") + creator: int = Field(default=0, description="发送者ID") + content: str = Field(default="", description="消息内容") + reply: Dict[str, Any] = Field(default={}, description="回复内容") + robot_key: str = Field(default="", description="机器人key") + url: str = Field(default="", description="callback地址") + ctime: int = Field(default=0, description="发送时间") diff --git a/CoreModules/plugin_interface.py b/CoreModules/plugin_interface.py index 9632bd1..42f330a 100644 --- a/CoreModules/plugin_interface.py +++ b/CoreModules/plugin_interface.py @@ -77,7 +77,11 @@ class PluginInterface(ABC): ''' 将插件注册, 使其可以被命令匹配 ''' - PluginInterface.plugin_instances[command] = self + if command in PluginInterface.plugin_instances: + config.Log("Warning", f"插件{PluginInterface.plugin_instances[command].__class__.__name__}已注册命令{command}, 将被新插件{self.__class__.__name__}覆盖") + else: + config.Log("Info", f"插件{self.__class__.__name__}已注册命令{command}") + PluginInterface.plugin_instances[command] = self def register_db_model(self) -> DatabaseModel: ''' @@ -109,6 +113,7 @@ def ImportPlugins(app: FastAPI, plugin_dir:str = "Plugins") -> None: if plugin_tool_dir.IsDir() == False: config.Log("Error", f"插件目录不是目录: {plugin_tool_dir.GetFullPath()}") return + plugin_registered_class = set[type[PluginInterface]]() for dir_name, sub_dirs, files in plugin_tool_dir.DirWalk(): for file_name in files: module_file = ToolFile(dir_name)|file_name @@ -119,7 +124,8 @@ def ImportPlugins(app: FastAPI, plugin_dir:str = "Plugins") -> None: plugin_class = getattr(module, class_name) if not isinstance(plugin_class, type): continue - if issubclass(plugin_class, PluginInterface): + if issubclass(plugin_class, PluginInterface) and plugin_class not in plugin_registered_class: + plugin_registered_class.add(plugin_class) plugin = plugin_class() if plugin.is_enable_plugin() == False: continue diff --git a/CoreRouters/callback.py b/CoreRouters/callback.py index d3820cf..2835897 100644 --- a/CoreRouters/callback.py +++ b/CoreRouters/callback.py @@ -19,9 +19,33 @@ async def callback_verify(): return JSONResponse({"result": "ok"}) +@router.post("/callback/construct") +async def callback_receive_construct(callback_data: CallbackRequest): + """以构造好的Callback消息进行处理, 已知方式""" + try: + # 解析指令 + content = callback_data.content + command = content.split(" ")[0] + message = content[len(command):].strip() + config.Log("Info", f"识别指令: command={command}") + + # 处理指令 + result = await handle_command(command, message, callback_data.chatid, callback_data.creator) + if result: + return JSONResponse({"result": "ok", "message": result}) + else: + return JSONResponse({"result": "ok"}) + + except Exception as e: + config.Log("Error", f"处理Callback异常: {e}") + if ALWAYS_RETURN_OK: + return JSONResponse({"result": "ok"}) + else: + return JSONResponse({"result": "error", "message": str(e)}) + @router.post("/callback") async def callback_receive(request: Request): - """Callback消息""" + """接受未知的Callback消息并进行处理, 默认方式""" try: # 解析请求数据 data = await request.json()