Merge branch 'dev' of https://github.com/MaiM-with-u/MaiBot into dev

plugins/hello_world_plugin/plugin.py

@@ -60,7 +60,7 @@ class ByeAction(BaseAction):
     async def execute(self) -> Tuple[bool, str]:
         bye_message = self.action_data.get("bye_message", "")
 
-        message = "再见！期待下次聊天！👋" + bye_message
+        message = f"再见！期待下次聊天！👋{bye_message}"
         await self.send_text(message)
         return True, "发送了告别消息"
 

src/chat/message_receive/bot.py

@@ -11,7 +11,7 @@ from src.chat.focus_chat.heartflow_message_processor import HeartFCMessageReceiv
 from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
 from src.config.config import global_config
 from src.plugin_system.core.component_registry import component_registry  # 导入新插件系统
-
+from src.plugin_system.base.base_command import BaseCommand
 # 定义日志配置
 
 
@@ -49,6 +49,7 @@ class ChatBot:
             logger.error(f"创建PFC聊天失败: {e}")
 
     async def _process_commands_with_new_system(self, message: MessageRecv):
+        # sourcery skip: use-named-expression
         """使用新插件系统处理命令"""
         try:
             text = message.processed_plain_text
@@ -62,7 +63,7 @@ class ChatBot:
                 plugin_config = component_registry.get_plugin_config(plugin_name)
 
                 # 创建命令实例
-                command_instance = command_class(message, plugin_config)
+                command_instance: BaseCommand = command_class(message, plugin_config)
                 command_instance.set_matched_groups(matched_groups)
 
                 try:
@@ -85,7 +86,7 @@ class ChatBot:
                     logger.error(traceback.format_exc())
 
                     try:
-                        await command_instance.send_reply(f"命令执行出错: {str(e)}")
+                        await command_instance.send_text(f"命令执行出错: {str(e)}")
                     except Exception as send_error:
                         logger.error(f"发送错误消息失败: {send_error}")
 

src/plugin_system/apis/send_api.py

@@ -22,7 +22,7 @@
 import traceback
 import time
 import difflib
-from typing import Optional
+from typing import Optional, Union
 from src.common.logger import get_logger
 
 # 导入依赖
@@ -44,7 +44,7 @@ logger = get_logger("send_api")
 
 async def _send_to_target(
     message_type: str,
-    content: str,
+    content: Union[str, dict],
     stream_id: str,
     display_message: str = "",
     typing: bool = False,
@@ -284,7 +284,7 @@ async def image_to_stream(image_base64: str, stream_id: str, storage_message: bo
     return await _send_to_target("image", image_base64, stream_id, "", typing=False, storage_message=storage_message)
 
 
-async def command_to_stream(command: str, stream_id: str, storage_message: bool = True) -> bool:
+async def command_to_stream(command: Union[str, dict], stream_id: str, storage_message: bool = True) -> bool:
     """向指定流发送命令
 
     Args:

src/plugin_system/base/base_action.py

@@ -53,6 +53,8 @@ class BaseAction(ABC):
             plugin_config: 插件配置字典
             **kwargs: 其他参数
         """
+        if plugin_config is None:
+            plugin_config = {}
         self.action_data = action_data
         self.reasoning = reasoning
         self.cycle_timers = cycle_timers

src/plugin_system/base/component_types.py

@@ -1,6 +1,6 @@
 from enum import Enum
 from typing import Dict, Any, List
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 # 组件类型枚举
@@ -64,7 +64,7 @@ class ComponentInfo:
     enabled: bool = True  # 是否启用
     plugin_name: str = ""  # 所属插件名称
     is_built_in: bool = False  # 是否为内置组件
-    metadata: Dict[str, Any] = None  # 额外元数据
+    metadata: Dict[str, Any] = field(default_factory=dict)  # 额外元数据
 
     def __post_init__(self):
         if self.metadata is None:
@@ -79,13 +79,13 @@ class ActionInfo(ComponentInfo):
     normal_activation_type: ActionActivationType = ActionActivationType.ALWAYS
     random_activation_probability: float = 0.0
     llm_judge_prompt: str = ""
-    activation_keywords: List[str] = None
+    activation_keywords: List[str] = field(default_factory=list)  # 激活关键词列表
     keyword_case_sensitive: bool = False
     mode_enable: ChatMode = ChatMode.ALL
     parallel_action: bool = False
-    action_parameters: Dict[str, Any] = None
+    action_parameters: Dict[str, Any] = field(default_factory=dict)  # 动作参数
-    action_require: List[str] = None
+    action_require: List[str] = field(default_factory=list)  # 动作需求说明
-    associated_types: List[str] = None
+    associated_types: List[str] = field(default_factory=list)  # 关联的消息类型
 
     def __post_init__(self):
         super().__post_init__()
@@ -106,7 +106,7 @@ class CommandInfo(ComponentInfo):
 
     command_pattern: str = ""  # 命令匹配模式（正则表达式）
     command_help: str = ""  # 命令帮助信息
-    command_examples: List[str] = None  # 命令使用示例
+    command_examples: List[str] = field(default_factory=list)  # 命令使用示例
     intercept_message: bool = True  # 是否拦截消息处理（默认拦截）
 
     def __post_init__(self):
@@ -126,18 +126,18 @@ class PluginInfo:
     author: str = ""  # 插件作者
     enabled: bool = True  # 是否启用
     is_built_in: bool = False  # 是否为内置插件
-    components: List[ComponentInfo] = None  # 包含的组件列表
+    components: List[ComponentInfo] = field(default_factory=list)  # 包含的组件列表
-    dependencies: List[str] = None  # 依赖的其他插件
+    dependencies: List[str] = field(default_factory=list)  # 依赖的其他插件
-    python_dependencies: List[PythonDependency] = None  # Python包依赖
+    python_dependencies: List[PythonDependency] = field(default_factory=list)  # Python包依赖
     config_file: str = ""  # 配置文件路径
-    metadata: Dict[str, Any] = None  # 额外元数据
+    metadata: Dict[str, Any] = field(default_factory=dict)  # 额外元数据
     # 新增：manifest相关信息
-    manifest_data: Dict[str, Any] = None  # manifest文件数据
+    manifest_data: Dict[str, Any] = field(default_factory=dict)  # manifest文件数据
     license: str = ""  # 插件许可证
     homepage_url: str = ""  # 插件主页
     repository_url: str = ""  # 插件仓库地址
-    keywords: List[str] = None  # 插件关键词
+    keywords: List[str] = field(default_factory=list)  # 插件关键词
-    categories: List[str] = None  # 插件分类
+    categories: List[str] = field(default_factory=list)  # 插件分类
     min_host_version: str = ""  # 最低主机版本要求
     max_host_version: str = ""  # 最高主机版本要求
 

src/plugin_system/core/component_registry.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Type, Optional, Any, Pattern
+from typing import Dict, List, Optional, Any, Pattern, Union
 import re
 from src.common.logger import get_logger
 from src.plugin_system.base.component_types import (
@@ -9,6 +9,9 @@ from src.plugin_system.base.component_types import (
     ComponentType,
 )
 
+from ..base.base_command import BaseCommand
+from ..base.base_action import BaseAction
+
 logger = get_logger("component_registry")
 
 
@@ -25,24 +28,26 @@ class ComponentRegistry:
             ComponentType.ACTION: {},
             ComponentType.COMMAND: {},
         }
-        self._component_classes: Dict[str, Type] = {}  # 组件名 -> 组件类
+        self._component_classes: Dict[str, Union[BaseCommand, BaseAction]] = {}  # 组件名 -> 组件类
 
         # 插件注册表
         self._plugins: Dict[str, PluginInfo] = {}  # 插件名 -> 插件信息
 
         # Action特定注册表
-        self._action_registry: Dict[str, Type] = {}  # action名 -> action类
+        self._action_registry: Dict[str, BaseAction] = {}  # action名 -> action类
         self._default_actions: Dict[str, str] = {}  # 启用的action名 -> 描述
 
         # Command特定注册表
-        self._command_registry: Dict[str, Type] = {}  # command名 -> command类
+        self._command_registry: Dict[str, BaseCommand] = {}  # command名 -> command类
-        self._command_patterns: Dict[Pattern, Type] = {}  # 编译后的正则 -> command类
+        self._command_patterns: Dict[Pattern, BaseCommand] = {}  # 编译后的正则 -> command类
 
         logger.info("组件注册中心初始化完成")
 
     # === 通用组件注册方法 ===
 
-    def register_component(self, component_info: ComponentInfo, component_class: Type) -> bool:
+    def register_component(
+        self, component_info: ComponentInfo, component_class: Union[BaseCommand, BaseAction]
+    ) -> bool:
         """注册组件
 
         Args:
@@ -93,7 +98,7 @@ class ComponentRegistry:
         )
         return True
 
-    def _register_action_component(self, action_info: ActionInfo, action_class: Type):
+    def _register_action_component(self, action_info: ActionInfo, action_class: BaseAction):
         """注册Action组件到Action特定注册表"""
         action_name = action_info.name
         self._action_registry[action_name] = action_class
@@ -102,7 +107,7 @@ class ComponentRegistry:
         if action_info.enabled:
             self._default_actions[action_name] = action_info.description
 
-    def _register_command_component(self, command_info: CommandInfo, command_class: Type):
+    def _register_command_component(self, command_info: CommandInfo, command_class: BaseCommand):
         """注册Command组件到Command特定注册表"""
         command_name = command_info.name
         self._command_registry[command_name] = command_class
@@ -115,6 +120,7 @@ class ComponentRegistry:
     # === 组件查询方法 ===
 
     def get_component_info(self, component_name: str, component_type: ComponentType = None) -> Optional[ComponentInfo]:
+        # sourcery skip: class-extract-method
         """获取组件信息，支持自动命名空间解析
 
         Args:
@@ -143,8 +149,7 @@ class ComponentRegistry:
         candidates = []
         for namespace_prefix in ["action", "command"]:
             namespaced_name = f"{namespace_prefix}.{component_name}"
-            component_info = self._components.get(namespaced_name)
+            if component_info := self._components.get(namespaced_name):
-            if component_info:
                 candidates.append((namespace_prefix, namespaced_name, component_info))
 
         if len(candidates) == 1:
@@ -161,7 +166,9 @@ class ComponentRegistry:
         # 4. 都没找到
         return None
 
-    def get_component_class(self, component_name: str, component_type: ComponentType = None) -> Optional[Type]:
+    def get_component_class(
+        self, component_name: str, component_type: ComponentType = None
+    ) -> Optional[Union[BaseCommand, BaseAction]]:
         """获取组件类，支持自动命名空间解析
 
         Args:
@@ -169,7 +176,7 @@ class ComponentRegistry:
             component_type: 组件类型，如果提供则优先在该类型中查找
 
         Returns:
-            Optional[Type]: 组件类或None
+            Optional[Union[BaseCommand, BaseAction]]: 组件类或None
         """
         # 1. 如果已经是命名空间化的名称，直接查找
         if "." in component_name:
@@ -190,8 +197,7 @@ class ComponentRegistry:
         candidates = []
         for namespace_prefix in ["action", "command"]:
             namespaced_name = f"{namespace_prefix}.{component_name}"
-            component_class = self._component_classes.get(namespaced_name)
+            if component_class := self._component_classes.get(namespaced_name):
-            if component_class:
                 candidates.append((namespace_prefix, namespaced_name, component_class))
 
         if len(candidates) == 1:
@@ -221,7 +227,7 @@ class ComponentRegistry:
 
     # === Action特定查询方法 ===
 
-    def get_action_registry(self) -> Dict[str, Type]:
+    def get_action_registry(self) -> Dict[str, BaseAction]:
         """获取Action注册表（用于兼容现有系统）"""
         return self._action_registry.copy()
 
@@ -236,11 +242,11 @@ class ComponentRegistry:
 
     # === Command特定查询方法 ===
 
-    def get_command_registry(self) -> Dict[str, Type]:
+    def get_command_registry(self) -> Dict[str, BaseCommand]:
         """获取Command注册表（用于兼容现有系统）"""
         return self._command_registry.copy()
 
-    def get_command_patterns(self) -> Dict[Pattern, Type]:
+    def get_command_patterns(self) -> Dict[Pattern, BaseCommand]:
         """获取Command模式注册表（用于兼容现有系统）"""
         return self._command_patterns.copy()
 
@@ -249,19 +255,19 @@ class ComponentRegistry:
         info = self.get_component_info(command_name, ComponentType.COMMAND)
         return info if isinstance(info, CommandInfo) else None
 
-    def find_command_by_text(self, text: str) -> Optional[tuple[Type, dict, bool, str]]:
+    def find_command_by_text(self, text: str) -> Optional[tuple[BaseCommand, dict, bool, str]]:
+        # sourcery skip: use-named-expression, use-next
         """根据文本查找匹配的命令
 
         Args:
             text: 输入文本
 
         Returns:
-            Optional[tuple[Type, dict, bool, str]]: (命令类, 匹配的命名组, 是否拦截消息, 插件名) 或 None
+            Optional[tuple[BaseCommand, dict, bool, str]]: (命令类, 匹配的命名组, 是否拦截消息, 插件名) 或 None
         """
 
         for pattern, command_class in self._command_patterns.items():
-            match = pattern.match(text)
+            if match := pattern.match(text):
-            if match:
                 command_name = None
                 # 查找对应的组件信息
                 for name, cls in self._command_registry.items():
@@ -272,14 +278,13 @@ class ComponentRegistry:
                 # 检查命令是否启用
                 if command_name:
                     command_info = self.get_command_info(command_name)
-                    if command_info:
+                    if command_info and command_info.enabled:
-                        if command_info.enabled:
+                        return (
-                            return (
+                            command_class,
-                                command_class,
+                            match.groupdict(),
-                                match.groupdict(),
+                            command_info.intercept_message,
-                                command_info.intercept_message,
+                            command_info.plugin_name,
-                                command_info.plugin_name,
+                        )
-                            )
         return None
 
     # === 插件管理方法 ===