Add PlusCommand enhanced command system

Introduces the PlusCommand system for simplified command development, including new base classes, argument parsing utilities, and registration logic. Updates the plugin system, component registry, and bot message handling to support PlusCommand components alongside traditional commands. Adds documentation and configuration for command prefixes, and provides a developer guide for the new system.

docs/PLUS_COMMAND_GUIDE.md

@@ -0,0 +1,260 @@
+# 增强命令系统使用指南
+
+## 概述
+
+增强命令系统是MaiBot插件系统的一个扩展，让命令的定义和使用变得更加简单直观。你不再需要编写复杂的正则表达式，只需要定义命令名、别名和参数处理逻辑即可。
+
+## 核心特性
+
+- **无需正则表达式**：只需定义命令名和别名
+- **自动参数解析**：提供`CommandArgs`类处理参数
+- **命令别名支持**：一个命令可以有多个别名
+- **优先级控制**：支持命令优先级设置
+- **聊天类型限制**：可限制命令在群聊或私聊中使用
+- **消息拦截**：可选择是否拦截消息进行后续处理
+
+## 快速开始
+
+### 1. 创建基础命令
+
+```python
+from src.plugin_system import PlusCommand, CommandArgs, ChatType
+from typing import Tuple, Optional
+
+class EchoCommand(PlusCommand):
+    """Echo命令示例"""
+    
+    command_name = "echo"
+    command_description = "回显命令"
+    command_aliases = ["say", "repeat"]  # 可选：命令别名
+    priority = 5  # 可选：优先级，数字越大优先级越高
+    chat_type_allow = ChatType.ALL  # 可选：ALL, GROUP, PRIVATE
+    intercept_message = True  # 可选：是否拦截消息
+
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        """执行命令"""
+        if args.is_empty():
+            await self.send_text("❓ 请提供要回显的内容\\n用法: /echo <内容>")
+            return True, "参数不足", True
+        
+        content = args.get_raw()
+        await self.send_text(f"🔊 {content}")
+        
+        return True, "Echo命令执行成功", True
+```
+
+### 2. 在插件中注册命令
+
+```python
+from src.plugin_system import BasePlugin, create_plus_command_adapter, register_plugin
+
+@register_plugin
+class MyPlugin(BasePlugin):
+    plugin_name = "my_plugin"
+    enable_plugin = True
+    dependencies = []
+    python_dependencies = []
+    config_file_name = "config.toml"
+
+    def get_plugin_components(self):
+        components = []
+        
+        # 使用工厂函数创建适配器
+        echo_adapter = create_plus_command_adapter(EchoCommand)
+        components.append((EchoCommand.get_command_info(), echo_adapter))
+        
+        return components
+```
+
+## CommandArgs 类详解
+
+`CommandArgs`类提供了丰富的参数处理功能：
+
+### 基础方法
+
+```python
+# 获取原始参数字符串
+raw_text = args.get_raw()
+
+# 获取解析后的参数列表（按空格分割，支持引号）
+arg_list = args.get_args()
+
+# 检查是否有参数
+if args.is_empty():
+    # 没有参数的处理
+
+# 获取参数数量
+count = args.count()
+```
+
+### 获取特定参数
+
+```python
+# 获取第一个参数
+first_arg = args.get_first("默认值")
+
+# 获取指定索引的参数
+second_arg = args.get_arg(1, "默认值")
+
+# 获取从指定位置开始的剩余参数
+remaining = args.get_remaining(1)  # 从第2个参数开始
+```
+
+### 标志参数处理
+
+```python
+# 检查是否包含标志
+if args.has_flag("--verbose"):
+    # 处理verbose模式
+
+# 获取标志的值
+output_file = args.get_flag_value("--output", "default.txt")
+name = args.get_flag_value("--name", "Anonymous")
+```
+
+## 高级示例
+
+### 1. 带子命令的复杂命令
+
+```python
+class TestCommand(PlusCommand):
+    command_name = "test"
+    command_description = "测试命令，展示参数解析功能"
+    command_aliases = ["t"]
+
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        if args.is_empty():
+            await self.send_text("用法: /test <子命令> [参数]")
+            return True, "显示帮助", True
+        
+        subcommand = args.get_first().lower()
+        
+        if subcommand == "args":
+            result = f"""
+🔍 参数解析结果:
+原始字符串: '{args.get_raw()}'
+解析后参数: {args.get_args()}
+参数数量: {args.count()}
+第一个参数: '{args.get_first()}'
+剩余参数: '{args.get_remaining()}'
+            """
+            await self.send_text(result)
+            
+        elif subcommand == "flags":
+            result = f"""
+🏴 标志测试结果:
+包含 --verbose: {args.has_flag('--verbose')}
+包含 -v: {args.has_flag('-v')}
+--output 的值: '{args.get_flag_value('--output', '未设置')}'
+--name 的值: '{args.get_flag_value('--name', '未设置')}'
+            """
+            await self.send_text(result)
+            
+        else:
+            await self.send_text(f"❓ 未知的子命令: {subcommand}")
+        
+        return True, "Test命令执行成功", True
+```
+
+### 2. 聊天类型限制示例
+
+```python
+class PrivateOnlyCommand(PlusCommand):
+    command_name = "private"
+    command_description = "仅私聊可用的命令"
+    chat_type_allow = ChatType.PRIVATE
+
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        await self.send_text("这是一个仅私聊可用的命令")
+        return True, "私聊命令执行", True
+
+class GroupOnlyCommand(PlusCommand):
+    command_name = "group"
+    command_description = "仅群聊可用的命令"
+    chat_type_allow = ChatType.GROUP
+
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        await self.send_text("这是一个仅群聊可用的命令")
+        return True, "群聊命令执行", True
+```
+
+### 3. 配置驱动的命令
+
+```python
+class ConfigurableCommand(PlusCommand):
+    command_name = "config_cmd"
+    command_description = "可配置的命令"
+
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        # 从插件配置中获取设置
+        max_length = self.get_config("commands.max_length", 100)
+        enabled_features = self.get_config("commands.features", [])
+        
+        if args.is_empty():
+            await self.send_text("请提供参数")
+            return True, "无参数", True
+            
+        content = args.get_raw()
+        if len(content) > max_length:
+            await self.send_text(f"内容过长，最大允许 {max_length} 字符")
+            return True, "内容过长", True
+            
+        # 根据配置决定功能
+        if "uppercase" in enabled_features:
+            content = content.upper()
+            
+        await self.send_text(f"处理结果: {content}")
+        return True, "配置命令执行", True
+```
+
+## 支持的命令前缀
+
+系统支持以下命令前缀（在`config/bot_config.toml`中配置）：
+
+- `/` - 斜杠（默认）
+- `!` - 感叹号
+- `.` - 点号
+- `#` - 井号
+
+例如，对于echo命令，以下调用都是有效的：
+- `/echo Hello`
+- `!echo Hello`
+- `.echo Hello`
+- `#echo Hello`
+
+## 返回值说明
+
+`execute`方法需要返回一个三元组：
+
+```python
+return (执行成功标志, 可选消息, 是否拦截后续处理)
+```
+
+- **执行成功标志** (bool): True表示命令执行成功，False表示失败
+- **可选消息** (Optional[str]): 用于日志记录的消息
+- **是否拦截后续处理** (bool): True表示拦截消息，不进行后续处理
+
+## 最佳实践
+
+1. **命令命名**：使用简短、直观的命令名
+2. **别名设置**：为常用命令提供简短别名
+3. **参数验证**：总是检查参数的有效性
+4. **错误处理**：提供清晰的错误提示和使用说明
+5. **配置支持**：重要设置应该可配置
+6. **聊天类型**：根据命令功能选择合适的聊天类型限制
+
+## 完整示例
+
+完整的插件示例请参考 `plugins/echo_example/plugin.py` 文件。
+
+## 与传统BaseCommand的区别
+
+| 特性 | PlusCommand | BaseCommand |
+|------|-------------|-------------|
+| 正则表达式 | 自动生成 | 手动编写 |
+| 参数解析 | CommandArgs类 | 手动处理 |
+| 别名支持 | 内置支持 | 需要在正则中处理 |
+| 代码复杂度 | 简单 | 复杂 |
+| 学习曲线 | 平缓 | 陡峭 |
+
+增强命令系统让插件开发变得更加简单和高效，特别适合新手开发者快速上手。

src/chat/message_receive/bot.py

@@ -98,6 +98,118 @@ class ChatBot:
 
             self._started = True
 
+    async def _process_plus_commands(self, message: MessageRecv):
+        """独立处理PlusCommand系统"""
+        try:
+            text = message.processed_plain_text
+            
+            # 获取配置的命令前缀
+            from src.config.config import global_config
+            prefixes = global_config.command.command_prefixes
+            
+            # 检查是否以任何前缀开头
+            matched_prefix = None
+            for prefix in prefixes:
+                if text.startswith(prefix):
+                    matched_prefix = prefix
+                    break
+            
+            if not matched_prefix:
+                return False, None, True  # 不是命令，继续处理
+            
+            # 移除前缀
+            command_part = text[len(matched_prefix):].strip()
+            
+            # 分离命令名和参数
+            parts = command_part.split(None, 1)
+            if not parts:
+                return False, None, True  # 没有命令名，继续处理
+            
+            command_word = parts[0].lower()
+            args_text = parts[1] if len(parts) > 1 else ""
+            
+            # 查找匹配的PlusCommand
+            plus_command_registry = component_registry.get_plus_command_registry()
+            matching_commands = []
+            
+            for plus_command_name, plus_command_class in plus_command_registry.items():
+                plus_command_info = component_registry.get_registered_plus_command_info(plus_command_name)
+                if not plus_command_info:
+                    continue
+                
+                # 检查命令名是否匹配（命令名和别名）
+                all_commands = [plus_command_name.lower()] + [alias.lower() for alias in plus_command_info.command_aliases]
+                if command_word in all_commands:
+                    matching_commands.append((plus_command_class, plus_command_info, plus_command_name))
+            
+            if not matching_commands:
+                return False, None, True  # 没有找到匹配的PlusCommand，继续处理
+            
+            # 如果有多个匹配，按优先级排序
+            if len(matching_commands) > 1:
+                matching_commands.sort(key=lambda x: x[1].priority, reverse=True)
+                logger.warning(f"文本 '{text}' 匹配到多个PlusCommand: {[cmd[2] for cmd in matching_commands]}，使用优先级最高的")
+            
+            plus_command_class, plus_command_info, plus_command_name = matching_commands[0]
+            
+            # 检查命令是否被禁用
+            if (
+                message.chat_stream
+                and message.chat_stream.stream_id
+                and plus_command_name
+                in global_announcement_manager.get_disabled_chat_commands(message.chat_stream.stream_id)
+            ):
+                logger.info("用户禁用的PlusCommand，跳过处理")
+                return False, None, True
+            
+            message.is_command = True
+            
+            # 获取插件配置
+            plugin_config = component_registry.get_plugin_config(plus_command_name)
+            
+            # 创建PlusCommand实例
+            plus_command_instance = plus_command_class(message, plugin_config)
+            
+            try:
+                # 检查聊天类型限制
+                if not plus_command_instance.is_chat_type_allowed():
+                    is_group = hasattr(message, 'is_group_message') and message.is_group_message
+                    logger.info(f"PlusCommand {plus_command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}")
+                    return False, None, True  # 跳过此命令，继续处理其他消息
+                
+                # 设置参数
+                from src.plugin_system.base.command_args import CommandArgs
+                command_args = CommandArgs(args_text)
+                plus_command_instance.args = command_args
+                
+                # 执行命令
+                success, response, intercept_message = await plus_command_instance.execute(command_args)
+                
+                # 记录命令执行结果
+                if success:
+                    logger.info(f"PlusCommand执行成功: {plus_command_class.__name__} (拦截: {intercept_message})")
+                else:
+                    logger.warning(f"PlusCommand执行失败: {plus_command_class.__name__} - {response}")
+                
+                # 根据命令的拦截设置决定是否继续处理消息
+                return True, response, not intercept_message  # 找到命令，根据intercept_message决定是否继续
+                
+            except Exception as e:
+                logger.error(f"执行PlusCommand时出错: {plus_command_class.__name__} - {e}")
+                logger.error(traceback.format_exc())
+                
+                try:
+                    await plus_command_instance.send_text(f"命令执行出错: {str(e)}")
+                except Exception as send_error:
+                    logger.error(f"发送错误消息失败: {send_error}")
+                
+                # 命令出错时，根据命令的拦截设置决定是否继续处理消息
+                return True, str(e), False  # 出错时继续处理消息
+                
+        except Exception as e:
+            logger.error(f"处理PlusCommand时出错: {e}")
+            return False, None, True  # 出错时继续处理消息
+
     async def _process_commands_with_new_system(self, message: MessageRecv):
         # sourcery skip: use-named-expression
         """使用新插件系统处理命令"""
@@ -306,14 +418,24 @@ class ChatBot:
             ):
                 return
 
-            # 命令处理 - 使用新插件系统检查并处理命令
-            is_command, cmd_result, continue_process = await self._process_commands_with_new_system(message)
-
-            # 如果是命令且不需要继续处理，则直接返回
-            if is_command and not continue_process:
+            # 命令处理 - 首先尝试PlusCommand独立处理
+            is_plus_command, plus_cmd_result, plus_continue_process = await self._process_plus_commands(message)
+            
+            # 如果是PlusCommand且不需要继续处理，则直接返回
+            if is_plus_command and not plus_continue_process:
                 await MessageStorage.store_message(message, chat)
-                logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
+                logger.info(f"PlusCommand处理完成，跳过后续消息处理: {plus_cmd_result}")
                 return
+            
+            # 如果不是PlusCommand，尝试传统的BaseCommand处理
+            if not is_plus_command:
+                is_command, cmd_result, continue_process = await self._process_commands_with_new_system(message)
+                
+                # 如果是命令且不需要继续处理，则直接返回
+                if is_command and not continue_process:
+                    await MessageStorage.store_message(message, chat)
+                    logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
+                    return
 
             result = await event_manager.trigger_event(EventType.ON_MESSAGE,message=message)
             if not result.all_continue_process():

src/config/config.py

@@ -45,6 +45,7 @@ from src.config.official_configs import (
     MonthlyPlanSystemConfig,
     CrossContextConfig,
     PermissionConfig,
+    CommandConfig,
     MaizoneIntercomConfig,
 )
 
@@ -381,6 +382,7 @@ class Config(ValidatedConfigBase):
     voice: VoiceConfig = Field(..., description="语音配置")
     schedule: ScheduleConfig = Field(..., description="调度配置")
     permission: PermissionConfig = Field(..., description="权限配置")
+    command: CommandConfig = Field(..., description="命令系统配置")
     
     # 有默认值的字段放在后面
     anti_prompt_injection: AntiPromptInjectionConfig = Field(default_factory=lambda: AntiPromptInjectionConfig(), description="反提示注入配置")

src/config/official_configs.py

@@ -697,6 +697,12 @@ class MaizoneIntercomConfig(ValidatedConfigBase):
     groups: List[ContextGroup] = Field(default_factory=list, description="Maizone互通组列表")
 
 
+class CommandConfig(ValidatedConfigBase):
+    """命令系统配置类"""
+    
+    command_prefixes: List[str] = Field(default_factory=lambda: ['/', '!', '.', '#'], description="支持的命令前缀列表")
+
+
 class PermissionConfig(ValidatedConfigBase):
     """权限系统配置类"""
     

src/plugin_system/__init__.py

@@ -17,6 +17,7 @@ from .base import (
     ComponentInfo,
     ActionInfo,
     CommandInfo,
+    PlusCommandInfo,
     PluginInfo,
     ToolInfo,
     PythonDependency,
@@ -25,6 +26,12 @@ from .base import (
     EventType,
     MaiMessages,
     ToolParamType,
+    # 新增的增强命令系统
+    PlusCommand,
+    CommandArgs,
+    PlusCommandAdapter,
+    create_plus_command_adapter,
+    ChatType,
 )
 
 # 导入工具模块
@@ -81,10 +88,17 @@ __all__ = [
     "BaseCommand",
     "BaseTool",
     "BaseEventHandler",
+    # 增强命令系统
+    "PlusCommand",
+    "CommandArgs",
+    "PlusCommandAdapter", 
+    "create_plus_command_adapter", 
+    "create_plus_command_adapter",
     # 类型定义
     "ComponentType",
     "ActionActivationType",
     "ChatMode",
+    "ChatType",
     "ComponentInfo",
     "ActionInfo",
     "CommandInfo",

src/plugin_system/base/__init__.py

@@ -13,9 +13,11 @@ from .component_types import (
     ComponentType,
     ActionActivationType,
     ChatMode,
+    ChatType,
     ComponentInfo,
     ActionInfo,
     CommandInfo,
+    PlusCommandInfo,
     ToolInfo,
     PluginInfo,
     PythonDependency,
@@ -25,6 +27,8 @@ from .component_types import (
     ToolParamType,
 )
 from .config_types import ConfigField
+from .plus_command import PlusCommand, PlusCommandAdapter, create_plus_command_adapter
+from .command_args import CommandArgs
 
 __all__ = [
     "BasePlugin",
@@ -34,9 +38,11 @@ __all__ = [
     "ComponentType",
     "ActionActivationType",
     "ChatMode",
+    "ChatType",
     "ComponentInfo",
     "ActionInfo",
     "CommandInfo",
+    "PlusCommandInfo",
     "ToolInfo",
     "PluginInfo",
     "PythonDependency",
@@ -46,4 +52,9 @@ __all__ = [
     "BaseEventHandler",
     "MaiMessages",
     "ToolParamType",
+    # 增强命令系统
+    "PlusCommand",
+    "CommandArgs",
+    "PlusCommandAdapter",
+    "create_plus_command_adapter",
 ]

src/plugin_system/base/base_plugin.py

@@ -1,13 +1,14 @@
 from abc import abstractmethod
-from typing import List, Type, Tuple, Union
+from typing import List, Type, Tuple, Union, TYPE_CHECKING
 from .plugin_base import PluginBase
 
 from src.common.logger import get_logger
-from src.plugin_system.base.component_types import ActionInfo, CommandInfo, EventHandlerInfo, ToolInfo
+from src.plugin_system.base.component_types import ActionInfo, CommandInfo, PlusCommandInfo, EventHandlerInfo, ToolInfo
 from .base_action import BaseAction
 from .base_command import BaseCommand
 from .base_events_handler import BaseEventHandler
 from .base_tool import BaseTool
+from .plus_command import PlusCommand
 
 logger = get_logger("base_plugin")
 
@@ -31,6 +32,7 @@ class BasePlugin(PluginBase):
         Union[
             Tuple[ActionInfo, Type[BaseAction]],
             Tuple[CommandInfo, Type[BaseCommand]],
+            Tuple[PlusCommandInfo, Type[PlusCommand]],
             Tuple[EventHandlerInfo, Type[BaseEventHandler]],
             Tuple[ToolInfo, Type[BaseTool]],
         ]

src/plugin_system/base/command_args.py

@@ -0,0 +1,156 @@
+"""命令参数解析类
+
+提供简单易用的命令参数解析功能
+"""
+
+from typing import List, Optional
+import shlex
+
+
+class CommandArgs:
+    """命令参数解析类
+    
+    提供方便的方法来处理命令参数
+    """
+    
+    def __init__(self, raw_args: str = ""):
+        """初始化命令参数
+        
+        Args:
+            raw_args: 原始参数字符串
+        """
+        self._raw_args = raw_args.strip()
+        self._parsed_args: Optional[List[str]] = None
+    
+    def get_raw(self) -> str:
+        """获取完整的参数字符串
+        
+        Returns:
+            str: 原始参数字符串
+        """
+        return self._raw_args
+    
+    def get_args(self) -> List[str]:
+        """获取解析后的参数列表
+        
+        将参数按空格分割，支持引号包围的参数
+        
+        Returns:
+            List[str]: 参数列表
+        """
+        if self._parsed_args is None:
+            if not self._raw_args:
+                self._parsed_args = []
+            else:
+                try:
+                    # 使用shlex来正确处理引号和转义字符
+                    self._parsed_args = shlex.split(self._raw_args)
+                except ValueError:
+                    # 如果shlex解析失败，fallback到简单的split
+                    self._parsed_args = self._raw_args.split()
+        
+        return self._parsed_args
+    
+    def is_empty(self) -> bool:
+        """检查参数是否为空
+        
+        Returns:
+            bool: 如果没有参数返回True
+        """
+        return len(self.get_args()) == 0
+    
+    def get_arg(self, index: int, default: str = "") -> str:
+        """获取指定索引的参数
+        
+        Args:
+            index: 参数索引（从0开始）
+            default: 默认值
+            
+        Returns:
+            str: 参数值或默认值
+        """
+        args = self.get_args()
+        if 0 <= index < len(args):
+            return args[index]
+        return default
+    
+    def get_first(self, default: str = "") -> str:
+        """获取第一个参数
+        
+        Args:
+            default: 默认值
+            
+        Returns:
+            str: 第一个参数或默认值
+        """
+        return self.get_arg(0, default)
+    
+    def get_remaining(self, start_index: int = 1) -> str:
+        """获取从指定索引开始的剩余参数字符串
+        
+        Args:
+            start_index: 起始索引
+            
+        Returns:
+            str: 剩余参数组成的字符串
+        """
+        args = self.get_args()
+        if start_index < len(args):
+            return " ".join(args[start_index:])
+        return ""
+    
+    def count(self) -> int:
+        """获取参数数量
+        
+        Returns:
+            int: 参数数量
+        """
+        return len(self.get_args())
+    
+    def has_flag(self, flag: str) -> bool:
+        """检查是否包含指定的标志参数
+        
+        Args:
+            flag: 标志名（如 "--verbose" 或 "-v"）
+            
+        Returns:
+            bool: 如果包含该标志返回True
+        """
+        return flag in self.get_args()
+    
+    def get_flag_value(self, flag: str, default: str = "") -> str:
+        """获取标志参数的值
+        
+        查找 --key=value 或 --key value 形式的参数
+        
+        Args:
+            flag: 标志名（如 "--output"）
+            default: 默认值
+            
+        Returns:
+            str: 标志的值或默认值
+        """
+        args = self.get_args()
+        
+        # 查找 --key=value 形式
+        for arg in args:
+            if arg.startswith(f"{flag}="):
+                return arg[len(flag) + 1:]
+        
+        # 查找 --key value 形式
+        try:
+            flag_index = args.index(flag)
+            if flag_index + 1 < len(args):
+                return args[flag_index + 1]
+        except ValueError:
+            pass
+        
+        return default
+    
+    def __str__(self) -> str:
+        """字符串表示"""
+        return self._raw_args
+    
+    def __repr__(self) -> str:
+        """调试表示"""
+        return f"CommandArgs(raw='{self._raw_args}', parsed={self.get_args()})"

src/plugin_system/base/component_types.py

@@ -12,6 +12,7 @@ class ComponentType(Enum):
 
     ACTION = "action"  # 动作组件
     COMMAND = "command"  # 命令组件
+    PLUS_COMMAND = "plus_command"  # 增强命令组件
     TOOL = "tool"  # 工具组件
     SCHEDULER = "scheduler"  # 定时任务组件（预留）
     EVENT_HANDLER = "event_handler"  # 事件处理组件
@@ -164,6 +165,22 @@ class CommandInfo(ComponentInfo):
         self.component_type = ComponentType.COMMAND
 
 
+@dataclass
+class PlusCommandInfo(ComponentInfo):
+    """增强命令组件信息"""
+
+    command_aliases: List[str] = field(default_factory=list)  # 命令别名列表
+    priority: int = 0  # 命令优先级
+    chat_type_allow: ChatType = ChatType.ALL  # 允许的聊天类型
+    intercept_message: bool = False  # 是否拦截消息
+
+    def __post_init__(self):
+        super().__post_init__()
+        if self.command_aliases is None:
+            self.command_aliases = []
+        self.component_type = ComponentType.PLUS_COMMAND
+
+
 @dataclass
 class ToolInfo(ComponentInfo):
     """工具组件信息"""

src/plugin_system/base/plus_command.py

@@ -0,0 +1,459 @@
+"""增强版命令处理器
+
+提供更简单易用的命令处理方式，无需手写正则表达式
+"""
+
+from abc import ABC, abstractmethod
+from typing import Dict, Tuple, Optional, List
+import re
+
+from src.common.logger import get_logger
+from src.plugin_system.base.component_types import CommandInfo, PlusCommandInfo, ComponentType, ChatType
+from src.chat.message_receive.message import MessageRecv
+from src.plugin_system.apis import send_api
+from src.plugin_system.base.command_args import CommandArgs
+from src.plugin_system.base.base_command import BaseCommand
+from src.config.config import global_config
+
+logger = get_logger("plus_command")
+
+
+class PlusCommand(ABC):
+    """增强版命令基类
+    
+    提供更简单的命令定义方式，无需手写正则表达式
+    
+    子类只需要定义：
+    - command_name: 命令名称
+    - command_description: 命令描述  
+    - command_aliases: 命令别名列表（可选）
+    - priority: 优先级（可选，数字越大优先级越高）
+    - chat_type_allow: 允许的聊天类型（可选）
+    - intercept_message: 是否拦截消息（可选）
+    """
+
+    # 子类需要定义的属性
+    command_name: str = ""
+    """命令名称，如 'echo'"""
+    
+    command_description: str = ""
+    """命令描述"""
+    
+    command_aliases: List[str] = []
+    """命令别名列表，如 ['say', 'repeat']"""
+    
+    priority: int = 0
+    """命令优先级，数字越大优先级越高"""
+    
+    chat_type_allow: ChatType = ChatType.ALL
+    """允许的聊天类型"""
+    
+    intercept_message: bool = False
+    """是否拦截消息，不进行后续处理"""
+
+    def __init__(self, message: MessageRecv, plugin_config: Optional[dict] = None):
+        """初始化命令组件
+
+        Args:
+            message: 接收到的消息对象
+            plugin_config: 插件配置字典
+        """
+        self.message = message
+        self.plugin_config = plugin_config or {}
+        self.log_prefix = "[PlusCommand]"
+        
+        # 解析命令参数
+        self._parse_command()
+        
+        # 验证聊天类型限制
+        if not self._validate_chat_type():
+            is_group = hasattr(self.message, 'is_group_message') and self.message.is_group_message
+            logger.warning(
+                f"{self.log_prefix} 命令 '{self.command_name}' 不支持当前聊天类型: "
+                f"{'群聊' if is_group else '私聊'}, 允许类型: {self.chat_type_allow.value}"
+            )
+
+    def _parse_command(self) -> None:
+        """解析命令和参数"""
+        if not hasattr(self.message, 'plain_text') or not self.message.plain_text:
+            self.args = CommandArgs("")
+            return
+            
+        plain_text = self.message.plain_text.strip()
+        
+        # 获取配置的命令前缀
+        prefixes = global_config.command.command_prefixes
+        
+        # 检查是否以任何前缀开头
+        matched_prefix = None
+        for prefix in prefixes:
+            if plain_text.startswith(prefix):
+                matched_prefix = prefix
+                break
+        
+        if not matched_prefix:
+            self.args = CommandArgs("")
+            return
+        
+        # 移除前缀
+        command_part = plain_text[len(matched_prefix):].strip()
+        
+        # 分离命令名和参数
+        parts = command_part.split(None, 1)
+        if not parts:
+            self.args = CommandArgs("")
+            return
+            
+        command_word = parts[0].lower()
+        args_text = parts[1] if len(parts) > 1 else ""
+        
+        # 检查命令名是否匹配
+        all_commands = [self.command_name.lower()] + [alias.lower() for alias in self.command_aliases]
+        if command_word not in all_commands:
+            self.args = CommandArgs("")
+            return
+            
+        # 创建参数对象
+        self.args = CommandArgs(args_text)
+
+    def _validate_chat_type(self) -> bool:
+        """验证当前聊天类型是否允许执行此命令
+        
+        Returns:
+            bool: 如果允许执行返回True，否则返回False
+        """
+        if self.chat_type_allow == ChatType.ALL:
+            return True
+        
+        # 检查是否为群聊消息
+        is_group = hasattr(self.message, 'is_group_message') and self.message.is_group_message
+        
+        if self.chat_type_allow == ChatType.GROUP and is_group:
+            return True
+        elif self.chat_type_allow == ChatType.PRIVATE and not is_group:
+            return True
+        else:
+            return False
+
+    def is_chat_type_allowed(self) -> bool:
+        """检查当前聊天类型是否允许执行此命令
+        
+        Returns:
+            bool: 如果允许执行返回True，否则返回False
+        """
+        return self._validate_chat_type()
+
+    def is_command_match(self) -> bool:
+        """检查当前消息是否匹配此命令
+        
+        Returns:
+            bool: 如果匹配返回True
+        """
+        return not self.args.is_empty() or self._is_exact_command_call()
+    
+    def _is_exact_command_call(self) -> bool:
+        """检查是否是精确的命令调用（无参数）"""
+        if not hasattr(self.message, 'plain_text') or not self.message.plain_text:
+            return False
+            
+        plain_text = self.message.plain_text.strip()
+        
+        # 获取配置的命令前缀
+        prefixes = global_config.command.command_prefixes
+        
+        # 检查每个前缀
+        for prefix in prefixes:
+            if plain_text.startswith(prefix):
+                command_part = plain_text[len(prefix):].strip()
+                all_commands = [self.command_name.lower()] + [alias.lower() for alias in self.command_aliases]
+                if command_part.lower() in all_commands:
+                    return True
+        
+        return False
+
+    @abstractmethod
+    async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
+        """执行命令的抽象方法，子类必须实现
+
+        Args:
+            args: 解析后的命令参数
+
+        Returns:
+            Tuple[bool, Optional[str], bool]: (是否执行成功, 可选的回复消息, 是否拦截消息)
+        """
+        pass
+
+    def get_config(self, key: str, default=None):
+        """获取插件配置值，使用嵌套键访问
+
+        Args:
+            key: 配置键名，使用嵌套访问如 "section.subsection.key"
+            default: 默认值
+
+        Returns:
+            Any: 配置值或默认值
+        """
+        if not self.plugin_config:
+            return default
+
+        # 支持嵌套键访问
+        keys = key.split(".")
+        current = self.plugin_config
+
+        for k in keys:
+            if isinstance(current, dict) and k in current:
+                current = current[k]
+            else:
+                return default
+
+        return current
+
+    async def send_text(self, content: str, reply_to: str = "") -> bool:
+        """发送回复消息
+
+        Args:
+            content: 回复内容
+            reply_to: 回复消息，格式为"发送者:消息内容"
+
+        Returns:
+            bool: 是否发送成功
+        """
+        # 获取聊天流信息
+        chat_stream = self.message.chat_stream
+        if not chat_stream or not hasattr(chat_stream, "stream_id"):
+            logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
+            return False
+
+        return await send_api.text_to_stream(text=content, stream_id=chat_stream.stream_id, reply_to=reply_to)
+
+    async def send_type(
+        self, message_type: str, content: str, display_message: str = "", typing: bool = False, reply_to: str = ""
+    ) -> bool:
+        """发送指定类型的回复消息到当前聊天环境
+
+        Args:
+            message_type: 消息类型，如"text"、"image"、"emoji"等
+            content: 消息内容
+            display_message: 显示消息（可选）
+            typing: 是否显示正在输入
+            reply_to: 回复消息，格式为"发送者:消息内容"
+
+        Returns:
+            bool: 是否发送成功
+        """
+        # 获取聊天流信息
+        chat_stream = self.message.chat_stream
+        if not chat_stream or not hasattr(chat_stream, "stream_id"):
+            logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
+            return False
+
+        return await send_api.custom_to_stream(
+            message_type=message_type,
+            content=content,
+            stream_id=chat_stream.stream_id,
+            display_message=display_message,
+            typing=typing,
+            reply_to=reply_to,
+        )
+
+    async def send_emoji(self, emoji_base64: str) -> bool:
+        """发送表情包
+
+        Args:
+            emoji_base64: 表情包的base64编码
+
+        Returns:
+            bool: 是否发送成功
+        """
+        chat_stream = self.message.chat_stream
+        if not chat_stream or not hasattr(chat_stream, "stream_id"):
+            logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
+            return False
+
+        return await send_api.emoji_to_stream(emoji_base64, chat_stream.stream_id)
+
+    async def send_image(self, image_base64: str) -> bool:
+        """发送图片
+
+        Args:
+            image_base64: 图片的base64编码
+
+        Returns:
+            bool: 是否发送成功
+        """
+        chat_stream = self.message.chat_stream
+        if not chat_stream or not hasattr(chat_stream, "stream_id"):
+            logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
+            return False
+
+        return await send_api.image_to_stream(image_base64, chat_stream.stream_id)
+
+    @classmethod
+    def get_command_info(cls) -> "CommandInfo":
+        """从类属性生成CommandInfo
+
+        Returns:
+            CommandInfo: 生成的命令信息对象
+        """
+        if "." in cls.command_name:
+            logger.error(f"命令名称 '{cls.command_name}' 包含非法字符 '.'，请使用下划线替代")
+            raise ValueError(f"命令名称 '{cls.command_name}' 包含非法字符 '.'，请使用下划线替代")
+        
+        # 生成正则表达式模式来匹配命令
+        command_pattern = cls._generate_command_pattern()
+        
+        return CommandInfo(
+            name=cls.command_name,
+            component_type=ComponentType.COMMAND,
+            description=cls.command_description,
+            command_pattern=command_pattern,
+            chat_type_allow=getattr(cls, "chat_type_allow", ChatType.ALL),
+        )
+
+    @classmethod
+    def get_plus_command_info(cls) -> "PlusCommandInfo":
+        """从类属性生成PlusCommandInfo
+
+        Returns:
+            PlusCommandInfo: 生成的增强命令信息对象
+        """
+        if "." in cls.command_name:
+            logger.error(f"命令名称 '{cls.command_name}' 包含非法字符 '.'，请使用下划线替代")
+            raise ValueError(f"命令名称 '{cls.command_name}' 包含非法字符 '.'，请使用下划线替代")
+        
+        return PlusCommandInfo(
+            name=cls.command_name,
+            component_type=ComponentType.PLUS_COMMAND,
+            description=cls.command_description,
+            command_aliases=getattr(cls, "command_aliases", []),
+            priority=getattr(cls, "priority", 0),
+            chat_type_allow=getattr(cls, "chat_type_allow", ChatType.ALL),
+            intercept_message=getattr(cls, "intercept_message", False),
+        )
+
+    @classmethod
+    def _generate_command_pattern(cls) -> str:
+        """生成命令匹配的正则表达式
+        
+        Returns:
+            str: 正则表达式字符串
+        """
+        # 获取所有可能的命令名（主命令名 + 别名）
+        all_commands = [cls.command_name] + getattr(cls, 'command_aliases', [])
+        
+        # 转义特殊字符并创建选择组
+        escaped_commands = [re.escape(cmd) for cmd in all_commands]
+        commands_pattern = "|".join(escaped_commands)
+        
+        # 获取默认前缀列表（这里先用硬编码，后续可以优化为动态获取）
+        default_prefixes = ["/", "!", ".", "#"]
+        escaped_prefixes = [re.escape(prefix) for prefix in default_prefixes]
+        prefixes_pattern = "|".join(escaped_prefixes)
+        
+        # 生成完整的正则表达式
+        # 匹配: [前缀][命令名][可选空白][任意参数]
+        pattern = f"^(?P<prefix>{prefixes_pattern})(?P<command>{commands_pattern})(?P<args>\\s.*)?$"
+        
+        return pattern
+
+
+class PlusCommandAdapter(BaseCommand):
+    """PlusCommand适配器
+    
+    将PlusCommand适配到现有的插件系统，继承BaseCommand
+    """
+    
+    def __init__(self, plus_command_class, message: MessageRecv, plugin_config: Optional[dict] = None):
+        """初始化适配器
+        
+        Args:
+            plus_command_class: PlusCommand子类
+            message: 消息对象
+            plugin_config: 插件配置
+        """
+        # 先设置必要的类属性
+        self.command_name = plus_command_class.command_name
+        self.command_description = plus_command_class.command_description
+        self.command_pattern = plus_command_class._generate_command_pattern()
+        self.chat_type_allow = getattr(plus_command_class, "chat_type_allow", ChatType.ALL)
+        self.priority = getattr(plus_command_class, "priority", 0)
+        self.intercept_message = getattr(plus_command_class, "intercept_message", False)
+        
+        # 调用父类初始化
+        super().__init__(message, plugin_config)
+        
+        # 创建PlusCommand实例
+        self.plus_command = plus_command_class(message, plugin_config)
+    
+    async def execute(self) -> Tuple[bool, Optional[str], bool]:
+        """执行命令
+        
+        Returns:
+            Tuple[bool, Optional[str], bool]: 执行结果
+        """
+        # 检查命令是否匹配
+        if not self.plus_command.is_command_match():
+            return False, "命令不匹配", False
+            
+        # 检查聊天类型权限
+        if not self.plus_command.is_chat_type_allowed():
+            return False, "不支持当前聊天类型", self.intercept_message
+        
+        # 执行命令
+        try:
+            return await self.plus_command.execute(self.plus_command.args)
+        except Exception as e:
+            logger.error(f"执行命令时出错: {e}", exc_info=True)
+            return False, f"命令执行出错: {str(e)}", self.intercept_message
+
+
+def create_plus_command_adapter(plus_command_class):
+    """创建PlusCommand适配器的工厂函数
+    
+    Args:
+        plus_command_class: PlusCommand子类
+        
+    Returns:
+        适配器类
+    """
+    class AdapterClass(BaseCommand):
+        command_name = plus_command_class.command_name
+        command_description = plus_command_class.command_description
+        command_pattern = plus_command_class._generate_command_pattern()
+        chat_type_allow = getattr(plus_command_class, "chat_type_allow", ChatType.ALL)
+        
+        def __init__(self, message: MessageRecv, plugin_config: Optional[dict] = None):
+            super().__init__(message, plugin_config)
+            self.plus_command = plus_command_class(message, plugin_config)
+            self.priority = getattr(plus_command_class, "priority", 0)
+            self.intercept_message = getattr(plus_command_class, "intercept_message", False)
+        
+        async def execute(self) -> Tuple[bool, Optional[str], bool]:
+            """执行命令"""
+            # 从BaseCommand的正则匹配结果中提取参数
+            args_text = ""
+            if hasattr(self, 'matched_groups') and self.matched_groups:
+                # 从正则匹配组中获取参数部分
+                args_match = self.matched_groups.get('args', '')
+                if args_match:
+                    args_text = args_match.strip()
+            
+            # 创建CommandArgs对象
+            command_args = CommandArgs(args_text)
+            
+            # 检查聊天类型权限
+            if not self.plus_command.is_chat_type_allowed():
+                return False, "不支持当前聊天类型", self.intercept_message
+            
+            # 执行命令，传递正确解析的参数
+            try:
+                return await self.plus_command.execute(command_args)
+            except Exception as e:
+                logger.error(f"执行命令时出错: {e}", exc_info=True)
+                return False, f"命令执行出错: {str(e)}", self.intercept_message
+    
+    return AdapterClass
+
+
+# 兼容旧的命名
+PlusCommandAdapter = create_plus_command_adapter

src/plugin_system/core/component_registry.py

@@ -8,6 +8,7 @@ from src.plugin_system.base.component_types import (
     ActionInfo,
     ToolInfo,
     CommandInfo,
+    PlusCommandInfo,
     EventHandlerInfo,
     PluginInfo,
     ComponentType,
@@ -16,6 +17,7 @@ from src.plugin_system.base.base_command import BaseCommand
 from src.plugin_system.base.base_action import BaseAction
 from src.plugin_system.base.base_tool import BaseTool
 from src.plugin_system.base.base_events_handler import BaseEventHandler
+from src.plugin_system.base.plus_command import PlusCommand
 
 logger = get_logger("component_registry")
 
@@ -32,7 +34,7 @@ class ComponentRegistry:
         """组件注册表 命名空间式组件名 -> 组件信息"""
         self._components_by_type: Dict[ComponentType, Dict[str, ComponentInfo]] = {types: {} for types in ComponentType}
         """类型 -> 组件原名称 -> 组件信息"""
-        self._components_classes: Dict[str, Type[Union[BaseCommand, BaseAction, BaseTool, BaseEventHandler]]] = {}
+        self._components_classes: Dict[str, Type[Union[BaseCommand, BaseAction, BaseTool, BaseEventHandler, PlusCommand]]] = {}
         """命名空间式组件名 -> 组件类"""
 
         # 插件注册表
@@ -133,6 +135,10 @@ class ComponentRegistry:
                 assert isinstance(component_info, CommandInfo)
                 assert issubclass(component_class, BaseCommand)
                 ret = self._register_command_component(component_info, component_class)
+            case ComponentType.PLUS_COMMAND:
+                assert isinstance(component_info, PlusCommandInfo)
+                assert issubclass(component_class, PlusCommand)
+                ret = self._register_plus_command_component(component_info, component_class)
             case ComponentType.TOOL:
                 assert isinstance(component_info, ToolInfo)
                 assert issubclass(component_class, BaseTool)
@@ -192,6 +198,26 @@ class ComponentRegistry:
 
         return True
 
+    def _register_plus_command_component(self, plus_command_info: PlusCommandInfo, plus_command_class: Type[PlusCommand]) -> bool:
+        """注册PlusCommand组件到特定注册表"""
+        plus_command_name = plus_command_info.name
+
+        if not plus_command_name:
+            logger.error(f"PlusCommand组件 {plus_command_class.__name__} 必须指定名称")
+            return False
+        if not isinstance(plus_command_info, PlusCommandInfo) or not issubclass(plus_command_class, PlusCommand):
+            logger.error(f"注册失败: {plus_command_name} 不是有效的PlusCommand")
+            return False
+
+        # 创建专门的PlusCommand注册表（如果还没有）
+        if not hasattr(self, '_plus_command_registry'):
+            self._plus_command_registry: Dict[str, Type[PlusCommand]] = {}
+
+        self._plus_command_registry[plus_command_name] = plus_command_class
+
+        logger.debug(f"已注册PlusCommand组件: {plus_command_name}")
+        return True
+
     def _register_tool_component(self, tool_info: ToolInfo, tool_class: Type[BaseTool]) -> bool:
         """注册Tool组件到Tool特定注册表"""
         tool_name = tool_info.name
@@ -248,6 +274,12 @@ class ComponentRegistry:
                         self._command_patterns.pop(key, None)
                     logger.debug(f"已移除Command组件: {component_name} (清理了 {len(keys_to_remove)} 个模式)")
 
+                case ComponentType.PLUS_COMMAND:
+                    # 移除PlusCommand注册
+                    if hasattr(self, '_plus_command_registry'):
+                        self._plus_command_registry.pop(component_name, None)
+                    logger.debug(f"已移除PlusCommand组件: {component_name}")
+
                 case ComponentType.TOOL:
                     # 移除Tool注册
                     self._tool_registry.pop(component_name, None)
@@ -520,21 +552,23 @@ class ComponentRegistry:
             text: 输入文本
 
         Returns:
-            Tuple: (命令类, 匹配的命名组, 是否拦截消息, 插件名) 或 None
+            Tuple: (命令类, 匹配的命名组, 命令信息) 或 None
         """
 
+        # 只查找传统的BaseCommand
         candidates = [pattern for pattern in self._command_patterns if pattern.match(text)]
-        if not candidates:
-            return None
-        if len(candidates) > 1:
-            logger.warning(f"文本 '{text}' 匹配到多个命令模式: {candidates}，使用第一个匹配")
-        command_name = self._command_patterns[candidates[0]]
-        command_info: CommandInfo = self.get_registered_command_info(command_name)  # type: ignore
-        return (
-            self._command_registry[command_name],
-            candidates[0].match(text).groupdict(),  # type: ignore
-            command_info,
-        )
+        if candidates:
+            if len(candidates) > 1:
+                logger.warning(f"文本 '{text}' 匹配到多个命令模式: {candidates}，使用第一个匹配")
+            command_name = self._command_patterns[candidates[0]]
+            command_info: CommandInfo = self.get_registered_command_info(command_name)  # type: ignore
+            return (
+                self._command_registry[command_name],
+                candidates[0].match(text).groupdict(),  # type: ignore
+                command_info,
+            )
+            
+        return None
 
     # === Tool 特定查询方法 ===
     def get_tool_registry(self) -> Dict[str, Type[BaseTool]]:
@@ -557,6 +591,25 @@ class ComponentRegistry:
         info = self.get_component_info(tool_name, ComponentType.TOOL)
         return info if isinstance(info, ToolInfo) else None
 
+    # === PlusCommand 特定查询方法 ===
+    def get_plus_command_registry(self) -> Dict[str, Type[PlusCommand]]:
+        """获取PlusCommand注册表"""
+        if not hasattr(self, '_plus_command_registry'):
+            self._plus_command_registry: Dict[str, Type[PlusCommand]] = {}
+        return self._plus_command_registry.copy()
+
+    def get_registered_plus_command_info(self, command_name: str) -> Optional[PlusCommandInfo]:
+        """获取PlusCommand信息
+
+        Args:
+            command_name: 命令名称
+
+        Returns:
+            PlusCommandInfo: 命令信息对象，如果命令不存在则返回 None
+        """
+        info = self.get_component_info(command_name, ComponentType.PLUS_COMMAND)
+        return info if isinstance(info, PlusCommandInfo) else None
+
     # === EventHandler 特定查询方法 ===
 
     def get_event_handler_registry(self) -> Dict[str, Type[BaseEventHandler]]:
@@ -612,6 +665,7 @@ class ComponentRegistry:
         command_components: int = 0
         tool_components: int = 0
         events_handlers: int = 0
+        plus_command_components: int = 0
         for component in self._components.values():
             if component.component_type == ComponentType.ACTION:
                 action_components += 1
@@ -621,11 +675,14 @@ class ComponentRegistry:
                 tool_components += 1
             elif component.component_type == ComponentType.EVENT_HANDLER:
                 events_handlers += 1
+            elif component.component_type == ComponentType.PLUS_COMMAND:
+                plus_command_components += 1
         return {
             "action_components": action_components,
             "command_components": command_components,
             "tool_components": tool_components,
             "event_handlers": events_handlers,
+            "plus_command_components": plus_command_components,
             "total_components": len(self._components),
             "total_plugins": len(self._plugins),
             "components_by_type": {

src/plugin_system/core/plugin_manager.py

@@ -363,13 +363,14 @@ class PluginManager:
         command_count = stats.get("command_components", 0)
         tool_count = stats.get("tool_components", 0)
         event_handler_count = stats.get("event_handlers", 0)
+        plus_command_count = stats.get("plus_command_components", 0)
         total_components = stats.get("total_components", 0)
 
         # 📋 显示插件加载总览
         if total_registered > 0:
             logger.info("🎉 插件系统加载完成!")
             logger.info(
-                f"📊 总览: {total_registered}个插件, {total_components}个组件 (Action: {action_count}, Command: {command_count}, Tool: {tool_count}, EventHandler: {event_handler_count})"
+                f"📊 总览: {total_registered}个插件, {total_components}个组件 (Action: {action_count}, Command: {command_count}, Tool: {tool_count}, PlusCommand: {plus_command_count}, EventHandler: {event_handler_count})"
             )
 
             # 显示详细的插件列表
@@ -410,6 +411,9 @@ class PluginManager:
                         event_handler_components = [
                             c for c in plugin_info.components if c.component_type == ComponentType.EVENT_HANDLER
                         ]
+                        plus_command_components = [
+                            c for c in plugin_info.components if c.component_type == ComponentType.PLUS_COMMAND
+                        ]
 
                         if action_components:
                             action_names = [c.name for c in action_components]
@@ -421,6 +425,9 @@ class PluginManager:
                         if tool_components:
                             tool_names = [c.name for c in tool_components]
                             logger.info(f"    🛠️ Tool组件: {', '.join(tool_names)}")
+                        if plus_command_components:
+                            plus_command_names = [c.name for c in plus_command_components]
+                            logger.info(f"    ⚡ PlusCommand组件: {', '.join(plus_command_names)}")
                         if event_handler_components:
                             event_handler_names = [c.name for c in event_handler_components]
                             logger.info(f"    📢 EventHandler组件: {', '.join(event_handler_names)}")