diff --git a/docs/PLUS_COMMAND_GUIDE.md b/docs/PLUS_COMMAND_GUIDE.md new file mode 100644 index 000000000..9ea67f47d --- /dev/null +++ b/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类 | 手动处理 | +| 别名支持 | 内置支持 | 需要在正则中处理 | +| 代码复杂度 | 简单 | 复杂 | +| 学习曲线 | 平缓 | 陡峭 | + +增强命令系统让插件开发变得更加简单和高效，特别适合新手开发者快速上手。 diff --git a/src/chat/message_receive/bot.py b/src/chat/message_receive/bot.py index 661ceb38f..760d69062 100644 --- a/src/chat/message_receive/bot.py +++ b/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(): diff --git a/src/config/config.py b/src/config/config.py index bd69399f4..8ffeaa9ab 100644 --- a/src/config/config.py +++ b/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="反提示注入配置") diff --git a/src/config/official_configs.py b/src/config/official_configs.py index 08518e9c1..de8479bdd 100644 --- a/src/config/official_configs.py +++ b/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): """权限系统配置类""" diff --git a/src/plugin_system/__init__.py b/src/plugin_system/__init__.py index ecadc0e80..26cc166f6 100644 --- a/src/plugin_system/__init__.py +++ b/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", diff --git a/src/plugin_system/base/__init__.py b/src/plugin_system/base/__init__.py index bc63d35d1..83debab01 100644 --- a/src/plugin_system/base/__init__.py +++ b/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", ] diff --git a/src/plugin_system/base/base_plugin.py b/src/plugin_system/base/base_plugin.py index ea28c5143..57f131ba1 100644 --- a/src/plugin_system/base/base_plugin.py +++ b/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]], ] diff --git a/src/plugin_system/base/command_args.py b/src/plugin_system/base/command_args.py new file mode 100644 index 000000000..46a2b701b --- /dev/null +++ b/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()})" diff --git a/src/plugin_system/base/component_types.py b/src/plugin_system/base/component_types.py index 0d22bf63e..63b32dec7 100644 --- a/src/plugin_system/base/component_types.py +++ b/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): """工具组件信息""" diff --git a/src/plugin_system/base/plus_command.py b/src/plugin_system/base/plus_command.py new file mode 100644 index 000000000..16af685a1 --- /dev/null +++ b/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{prefixes_pattern})(?P{commands_pattern})(?P\\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 diff --git a/src/plugin_system/base/plus_plugin.py b/src/plugin_system/base/plus_plugin.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/plugin_system/core/component_registry.py b/src/plugin_system/core/component_registry.py index 69a2d2a3b..7e925e3f0 100644 --- a/src/plugin_system/core/component_registry.py +++ b/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": { diff --git a/src/plugin_system/core/plugin_manager.py b/src/plugin_system/core/plugin_manager.py index 775edd1d9..5da07369f 100644 --- a/src/plugin_system/core/plugin_manager.py +++ b/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)}")