diff --git a/changes.md b/changes.md
index 9372fe9f4..0d6b507b9 100644
--- a/changes.md
+++ b/changes.md
@@ -15,21 +15,23 @@
- `python_dependencies`: 插件依赖的Python包列表,默认为空。**现在并不检查**
- `config_file_name`: 插件配置文件名,默认为`config.toml`。
- `config_schema`: 插件配置文件的schema,用于自动生成配置文件。
+4. 部分API的参数类型和返回值进行了调整
+ - `chat_api.py`中获取流的参数中可以使用一个特殊的枚举类型来获得所有平台的 ChatStream 了。
+ - `config_api.py`中的`get_global_config`和`get_plugin_config`方法现在支持嵌套访问的配置键名。
+ - `database_api.py`中的`db_query`方法调整了参数顺序以增强参数限制的同时,保证了typing正确;`db_get`方法增加了`single_result`参数,与`db_query`保持一致。
+5. 增加了`logging_api`,可以用`get_logger`来获取日志记录器。
# 插件系统修改
1. 现在所有的匹配模式不再是关键字了,而是枚举类。**(可能有遗漏)**
2. 修复了一下显示插件信息不显示的问题。同时精简了一下显示内容
-3. 修复了插件系统混用了`plugin_name`和`display_name`的问题。现在所有的插件信息都使用`display_name`来显示,而内部标识仍然使用`plugin_name`。**(可能有遗漏)**
-3. 部分API的参数类型和返回值进行了调整
- - `chat_api.py`中获取流的参数中可以使用一个特殊的枚举类型来获得所有平台的 ChatStream 了。
- - `config_api.py`中的`get_global_config`和`get_plugin_config`方法现在支持嵌套访问的配置键名。
- - `database_api.py`中的`db_query`方法调整了参数顺序以增强参数限制的同时,保证了typing正确;`db_get`方法增加了`single_result`参数,与`db_query`保持一致。
+3. 修复了插件系统混用了`plugin_name`和`display_name`的问题。现在所有的插件信息都使用`display_name`来显示,而内部标识仍然使用`plugin_name`。
4. 现在增加了参数类型检查,完善了对应注释
5. 现在插件抽象出了总基类 `PluginBase`
- - 基于`Action`和`Command`的插件基类现在为`BasePlugin`。
- - 基于`Event`的插件基类现在为`BaseEventPlugin`。
- - 所有的插件都继承自`PluginBase`。
- - 所有的插件都由`register_plugin`装饰器注册。
+ - 基于`Action`和`Command`的插件基类现在为`BasePlugin`。
+ - 基于`Event`的插件基类现在为`BaseEventPlugin`。
+ - 基于`Action`,`Command`和`Event`的插件基类现在为`BasePlugin`,所有插件都应该继承此基类。
+ - `BasePlugin`继承自`PluginBase`。
+ - 所有的插件类都由`register_plugin`装饰器注册。
6. 现在我们终于可以让插件有自定义的名字了!
- 真正实现了插件的`plugin_name`**不受文件夹名称限制**的功能。(吐槽:可乐你的某个小小细节导致我搞了好久……)
- 通过在插件类中定义`plugin_name`属性来指定插件内部标识符。
@@ -38,6 +40,7 @@
- 例如:`MaiMBot.plugins.example_plugin`而不是`example_plugin`。
- 仅在插件 import 失败时会如此,正常注册过程中失败的插件不会显示包名,而是显示插件内部标识符。(这是特性,但是基本上不可能出现这个情况)
7. 现在不支持单文件插件了,加载方式已经完全删除。
+8. 把`BaseEventPlugin`合并到了`BasePlugin`中,所有插件都应该继承自`BasePlugin`。
# 吐槽
diff --git a/plugins/hello_world_plugin/plugin.py b/plugins/hello_world_plugin/plugin.py
index 0376cbf2a..2d645616b 100644
--- a/plugins/hello_world_plugin/plugin.py
+++ b/plugins/hello_world_plugin/plugin.py
@@ -7,15 +7,13 @@ from src.plugin_system import (
ComponentInfo,
ActionActivationType,
ConfigField,
- BaseEventPlugin,
BaseEventHandler,
EventType,
+ MaiMessages,
)
-from src.plugin_system.base.component_types import MaiMessages
+
# ===== Action组件 =====
-
-
class HelloAction(BaseAction):
"""问候Action - 简单的问候动作"""
@@ -86,7 +84,7 @@ class TimeCommand(BaseCommand):
import datetime
# 获取当前时间
- time_format = self.get_config("time.format", "%Y-%m-%d %H:%M:%S")
+ time_format: str = self.get_config("time.format", "%Y-%m-%d %H:%M:%S") # type: ignore
now = datetime.datetime.now()
time_str = now.strftime(time_format)
@@ -140,6 +138,7 @@ class HelloWorldPlugin(BasePlugin):
"enable_emoji": ConfigField(type=bool, default=True, description="是否启用表情符号"),
},
"time": {"format": ConfigField(type=str, default="%Y-%m-%d %H:%M:%S", description="时间显示格式")},
+ "print_message": {"enabled": ConfigField(type=bool, default=True, description="是否启用打印")},
}
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
@@ -147,26 +146,27 @@ class HelloWorldPlugin(BasePlugin):
(HelloAction.get_action_info(), HelloAction),
(ByeAction.get_action_info(), ByeAction), # 添加告别Action
(TimeCommand.get_command_info(), TimeCommand),
+ (PrintMessage.get_handler_info(), PrintMessage),
]
-@register_plugin
-class HelloWorldEventPlugin(BaseEventPlugin):
- """Hello World事件插件 - 处理问候和告别事件"""
+# @register_plugin
+# class HelloWorldEventPlugin(BaseEPlugin):
+# """Hello World事件插件 - 处理问候和告别事件"""
- plugin_name = "hello_world_event_plugin"
- enable_plugin = False
- dependencies = []
- python_dependencies = []
- config_file_name = "event_config.toml"
-
- config_schema = {
- "plugin": {
- "name": ConfigField(type=str, default="hello_world_event_plugin", description="插件名称"),
- "version": ConfigField(type=str, default="1.0.0", description="插件版本"),
- "enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
- },
- }
+# plugin_name = "hello_world_event_plugin"
+# enable_plugin = False
+# dependencies = []
+# python_dependencies = []
+# config_file_name = "event_config.toml"
- def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
- return [(PrintMessage.get_handler_info(), PrintMessage)]
+# config_schema = {
+# "plugin": {
+# "name": ConfigField(type=str, default="hello_world_event_plugin", description="插件名称"),
+# "version": ConfigField(type=str, default="1.0.0", description="插件版本"),
+# "enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
+# },
+# }
+
+# def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
+# return [(PrintMessage.get_handler_info(), PrintMessage)]
diff --git a/src/plugin_system/__init__.py b/src/plugin_system/__init__.py
index c28ee6df9..491da7c1c 100644
--- a/src/plugin_system/__init__.py
+++ b/src/plugin_system/__init__.py
@@ -21,7 +21,6 @@ from .base import (
BaseEventHandler,
EventHandlerInfo,
EventType,
- BaseEventPlugin,
MaiMessages,
)
from .core import (
@@ -49,7 +48,6 @@ __all__ = [
"BasePlugin",
"BaseAction",
"BaseCommand",
- "BaseEventPlugin",
"BaseEventHandler",
# 类型定义
"ComponentType",
diff --git a/src/plugin_system/apis/plugin_register_api.py b/src/plugin_system/apis/plugin_register_api.py
index 4dafa3f72..879c09b32 100644
--- a/src/plugin_system/apis/plugin_register_api.py
+++ b/src/plugin_system/apis/plugin_register_api.py
@@ -2,13 +2,12 @@ from pathlib import Path
from src.common.logger import get_logger
-logger = get_logger("plugin_register")
+logger = get_logger("plugin_manager") # 复用plugin_manager名称
def register_plugin(cls):
from src.plugin_system.core.plugin_manager import plugin_manager
from src.plugin_system.base.base_plugin import BasePlugin
- from src.plugin_system.base.base_event_plugin import BaseEventPlugin
"""插件注册装饰器
@@ -19,13 +18,16 @@ def register_plugin(cls):
plugin_description = "我的插件"
...
"""
- if not issubclass(cls, BasePlugin) and not issubclass(cls, BaseEventPlugin):
+ if not issubclass(cls, BasePlugin):
logger.error(f"类 {cls.__name__} 不是 BasePlugin 的子类")
return cls
# 只是注册插件类,不立即实例化
# 插件管理器会负责实例化和注册
plugin_name: str = cls.plugin_name # type: ignore
+ if "." in plugin_name:
+ logger.error(f"插件名称 '{plugin_name}' 包含非法字符 '.',请使用下划线替代")
+ raise ValueError(f"插件名称 '{plugin_name}' 包含非法字符 '.',请使用下划线替代")
plugin_manager.plugin_classes[plugin_name] = cls
splitted_name = cls.__module__.split(".")
root_path = Path(__file__)
diff --git a/src/plugin_system/base/__init__.py b/src/plugin_system/base/__init__.py
index cfecb2df3..a95e05aed 100644
--- a/src/plugin_system/base/__init__.py
+++ b/src/plugin_system/base/__init__.py
@@ -7,7 +7,6 @@
from .base_plugin import BasePlugin
from .base_action import BaseAction
from .base_command import BaseCommand
-from .base_event_plugin import BaseEventPlugin
from .base_events_handler import BaseEventHandler
from .component_types import (
ComponentType,
@@ -39,7 +38,6 @@ __all__ = [
"ConfigField",
"EventHandlerInfo",
"EventType",
- "BaseEventPlugin",
"BaseEventHandler",
"MaiMessages",
]
diff --git a/src/plugin_system/base/base_action.py b/src/plugin_system/base/base_action.py
index 74ab22e67..a61a0339d 100644
--- a/src/plugin_system/base/base_action.py
+++ b/src/plugin_system/base/base_action.py
@@ -41,6 +41,7 @@ class BaseAction(ABC):
action_message: Optional[dict] = None,
**kwargs,
):
+ # sourcery skip: hoist-similar-statement-from-if, merge-else-if-into-elif, move-assign-in-block, swap-if-else-branches, swap-nested-ifs
"""初始化Action组件
Args:
@@ -355,7 +356,9 @@ class BaseAction(ABC):
# 从类属性读取名称,如果没有定义则使用类名自动生成
name = getattr(cls, "action_name", cls.__name__.lower().replace("action", ""))
-
+ if "." in name:
+ logger.error(f"Action名称 '{name}' 包含非法字符 '.',请使用下划线替代")
+ raise ValueError(f"Action名称 '{name}' 包含非法字符 '.',请使用下划线替代")
# 获取focus_activation_type和normal_activation_type
focus_activation_type = getattr(cls, "focus_activation_type", ActionActivationType.ALWAYS)
normal_activation_type = getattr(cls, "normal_activation_type", ActionActivationType.ALWAYS)
diff --git a/src/plugin_system/base/base_command.py b/src/plugin_system/base/base_command.py
index caf68567b..5387e01dd 100644
--- a/src/plugin_system/base/base_command.py
+++ b/src/plugin_system/base/base_command.py
@@ -219,7 +219,9 @@ class BaseCommand(ABC):
Returns:
CommandInfo: 生成的Command信息对象
"""
-
+ if "." in cls.command_name:
+ logger.error(f"Command名称 '{cls.command_name}' 包含非法字符 '.',请使用下划线替代")
+ raise ValueError(f"Command名称 '{cls.command_name}' 包含非法字符 '.',请使用下划线替代")
return CommandInfo(
name=cls.command_name,
component_type=ComponentType.COMMAND,
diff --git a/src/plugin_system/base/base_event_plugin.py b/src/plugin_system/base/base_event_plugin.py
deleted file mode 100644
index df67152a4..000000000
--- a/src/plugin_system/base/base_event_plugin.py
+++ /dev/null
@@ -1,57 +0,0 @@
-from abc import abstractmethod
-from typing import List, Tuple, Type
-
-from src.common.logger import get_logger
-from .plugin_base import PluginBase
-from .component_types import EventHandlerInfo
-from .base_events_handler import BaseEventHandler
-
-logger = get_logger("base_event_plugin")
-
-class BaseEventPlugin(PluginBase):
- """基于事件的插件基类
-
- 所有事件类型的插件都应该继承这个基类
- """
-
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
-
- @abstractmethod
- def get_plugin_components(self) -> List[Tuple[EventHandlerInfo, Type[BaseEventHandler]]]:
- """获取插件包含的事件组件
-
- 子类必须实现此方法,返回事件组件
-
- Returns:
- List[Tuple[ComponentInfo, Type]]: [(组件信息, 组件类), ...]
- """
- raise NotImplementedError("子类必须实现 get_plugin_components 方法")
-
- def register_plugin(self) -> bool:
- """注册事件插件"""
- from src.plugin_system.core.events_manager import events_manager
-
- components = self.get_plugin_components()
-
- # 检查依赖
- if not self._check_dependencies():
- logger.error(f"{self.log_prefix} 依赖检查失败,跳过注册")
- return False
-
- registered_components = []
- for handler_info, handler_class in components:
- handler_info.plugin_name = self.plugin_name
- if events_manager.register_event_subscriber(handler_info, handler_class):
- registered_components.append(handler_info)
- else:
- logger.error(f"{self.log_prefix} 事件处理器 {handler_info.name} 注册失败")
-
- self.plugin_info.components = registered_components
-
- if events_manager.register_plugins(self.plugin_info):
- logger.debug(f"{self.log_prefix} 插件注册成功,包含 {len(registered_components)} 个事件处理器")
- return True
- else:
- logger.error(f"{self.log_prefix} 插件注册失败")
- return False
\ No newline at end of file
diff --git a/src/plugin_system/base/base_events_handler.py b/src/plugin_system/base/base_events_handler.py
index 24d577a13..db6c20b62 100644
--- a/src/plugin_system/base/base_events_handler.py
+++ b/src/plugin_system/base/base_events_handler.py
@@ -38,9 +38,12 @@ class BaseEventHandler(ABC):
"""获取事件处理器的信息"""
# 从类属性读取名称,如果没有定义则使用类名自动生成
name: str = getattr(cls, "handler_name", cls.__name__.lower().replace("handler", ""))
+ if "." in name:
+ logger.error(f"事件处理器名称 '{name}' 包含非法字符 '.',请使用下划线替代")
+ raise ValueError(f"事件处理器名称 '{name}' 包含非法字符 '.',请使用下划线替代")
return EventHandlerInfo(
name=name,
- component_type=ComponentType.LISTENER,
+ component_type=ComponentType.EVENT_HANDLER,
description=getattr(cls, "handler_description", "events处理器"),
event_type=cls.event_type,
weight=cls.weight,
diff --git a/src/plugin_system/base/base_plugin.py b/src/plugin_system/base/base_plugin.py
index a93de5fab..1e6841eba 100644
--- a/src/plugin_system/base/base_plugin.py
+++ b/src/plugin_system/base/base_plugin.py
@@ -1,9 +1,12 @@
from abc import abstractmethod
-from typing import List, Type
+from typing import List, Type, Tuple, Union
from .plugin_base import PluginBase
from src.common.logger import get_logger
-from src.plugin_system.base.component_types import ComponentInfo
+from src.plugin_system.base.component_types import ComponentInfo, ActionInfo, CommandInfo, EventHandlerInfo
+from .base_action import BaseAction
+from .base_command import BaseCommand
+from .base_events_handler import BaseEventHandler
logger = get_logger("base_plugin")
@@ -21,7 +24,15 @@ class BasePlugin(PluginBase):
super().__init__(*args, **kwargs)
@abstractmethod
- def get_plugin_components(self) -> List[tuple[ComponentInfo, Type]]:
+ def get_plugin_components(
+ self,
+ ) -> List[
+ Union[
+ Tuple[ActionInfo, Type[BaseAction]],
+ Tuple[CommandInfo, Type[BaseCommand]],
+ Tuple[EventHandlerInfo, Type[BaseEventHandler]],
+ ]
+ ]:
"""获取插件包含的组件列表
子类必须实现此方法,返回组件信息和组件类的列表
diff --git a/src/plugin_system/base/component_types.py b/src/plugin_system/base/component_types.py
index 2b7636eb7..774daa598 100644
--- a/src/plugin_system/base/component_types.py
+++ b/src/plugin_system/base/component_types.py
@@ -11,7 +11,7 @@ class ComponentType(Enum):
ACTION = "action" # 动作组件
COMMAND = "command" # 命令组件
SCHEDULER = "scheduler" # 定时任务组件(预留)
- LISTENER = "listener" # 事件监听组件(预留)
+ EVENT_HANDLER = "event_handler" # 事件处理组件(预留)
def __str__(self) -> str:
return self.value
@@ -161,7 +161,7 @@ class EventHandlerInfo(ComponentInfo):
def __post_init__(self):
super().__post_init__()
- self.component_type = ComponentType.LISTENER
+ self.component_type = ComponentType.EVENT_HANDLER
@dataclass
diff --git a/src/plugin_system/core/component_registry.py b/src/plugin_system/core/component_registry.py
index 917069e11..29a45c604 100644
--- a/src/plugin_system/core/component_registry.py
+++ b/src/plugin_system/core/component_registry.py
@@ -1,16 +1,19 @@
-from typing import Dict, List, Optional, Any, Pattern, Tuple, Union, Type
import re
+
+from typing import Dict, List, Optional, Any, Pattern, Tuple, Union, Type
+
from src.common.logger import get_logger
from src.plugin_system.base.component_types import (
ComponentInfo,
ActionInfo,
CommandInfo,
+ EventHandlerInfo,
PluginInfo,
ComponentType,
)
-
from src.plugin_system.base.base_command import BaseCommand
from src.plugin_system.base.base_action import BaseAction
+from src.plugin_system.base.base_events_handler import BaseEventHandler
logger = get_logger("component_registry")
@@ -23,12 +26,11 @@ class ComponentRegistry:
def __init__(self):
# 组件注册表
- self._components: Dict[str, ComponentInfo] = {} # 组件名 -> 组件信息
- self._components_by_type: Dict[ComponentType, Dict[str, ComponentInfo]] = {
- ComponentType.ACTION: {},
- ComponentType.COMMAND: {},
- }
- self._component_classes: Dict[str, Union[Type[BaseCommand], Type[BaseAction]]] = {} # 组件名 -> 组件类
+ self._components: Dict[str, ComponentInfo] = {} # 命名空间式组件名 -> 组件信息
+ # 类型 -> 命名空间式名称 -> 组件信息
+ self._components_by_type: Dict[ComponentType, Dict[str, ComponentInfo]] = {types: {} for types in ComponentType}
+ # 命名空间式组件名 -> 组件类
+ self._components_classes: Dict[str, Type[Union[BaseCommand, BaseAction, BaseEventHandler]]] = {}
# 插件注册表
self._plugins: Dict[str, PluginInfo] = {} # 插件名 -> 插件信息
@@ -39,20 +41,43 @@ class ComponentRegistry:
# Command特定注册表
self._command_registry: Dict[str, Type[BaseCommand]] = {} # command名 -> command类
- self._command_patterns: Dict[Pattern, Type[BaseCommand]] = {} # 编译后的正则 -> command类
+ self._command_patterns: Dict[Pattern, str] = {} # 编译后的正则 -> command名
+
+ # EventHandler特定注册表
+ self._event_handler_registry: Dict[str, Type[BaseEventHandler]] = {} # event_handler名 -> event_handler类
+ self._enabled_event_handlers: Dict[str, Type[BaseEventHandler]] = {} # 启用的事件处理器
logger.info("组件注册中心初始化完成")
- # === 通用组件注册方法 ===
+ # == 注册方法 ==
+
+ def register_plugin(self, plugin_info: PluginInfo) -> bool:
+ """注册插件
+
+ Args:
+ plugin_info: 插件信息
+
+ Returns:
+ bool: 是否注册成功
+ """
+ plugin_name = plugin_info.name
+
+ if plugin_name in self._plugins:
+ logger.warning(f"插件 {plugin_name} 已存在,跳过注册")
+ return False
+
+ self._plugins[plugin_name] = plugin_info
+ logger.debug(f"已注册插件: {plugin_name} (组件数量: {len(plugin_info.components)})")
+ return True
def register_component(
- self, component_info: ComponentInfo, component_class: Union[Type[BaseCommand], Type[BaseAction]]
+ self, component_info: ComponentInfo, component_class: Type[Union[BaseCommand, BaseAction, BaseEventHandler]]
) -> bool:
"""注册组件
Args:
- component_info: 组件信息
- component_class: 组件类
+ component_info (ComponentInfo): 组件信息
+ component_class (Type[Union[BaseCommand, BaseAction, BaseEventHandler]]): 组件类
Returns:
bool: 是否注册成功
@@ -60,68 +85,110 @@ class ComponentRegistry:
component_name = component_info.name
component_type = component_info.component_type
plugin_name = getattr(component_info, "plugin_name", "unknown")
+ if "." in component_name:
+ logger.error(f"组件名称 '{component_name}' 包含非法字符 '.',请使用下划线替代")
+ return False
+ if "." in plugin_name:
+ logger.error(f"插件名称 '{plugin_name}' 包含非法字符 '.',请使用下划线替代")
+ return False
- # 🔥 系统级别自动区分:为不同类型的组件添加命名空间前缀
- if component_type == ComponentType.ACTION:
- namespaced_name = f"action.{component_name}"
- elif component_type == ComponentType.COMMAND:
- namespaced_name = f"command.{component_name}"
- else:
- # 未来扩展的组件类型
- namespaced_name = f"{component_type.value}.{component_name}"
+ namespaced_name = f"{component_type}.{component_name}"
- # 检查命名空间化的名称是否冲突
if namespaced_name in self._components:
existing_info = self._components[namespaced_name]
existing_plugin = getattr(existing_info, "plugin_name", "unknown")
logger.warning(
- f"组件冲突: {component_type.value}组件 '{component_name}' "
- f"已被插件 '{existing_plugin}' 注册,跳过插件 '{plugin_name}' 的注册"
+ f"组件名冲突: '{plugin_name}' 插件的 {component_type} 类型组件 '{component_name}' 已被插件 '{existing_plugin}' 注册,跳过此组件注册"
)
return False
- # 注册到通用注册表(使用命名空间化的名称)
- self._components[namespaced_name] = component_info
+ self._components[namespaced_name] = component_info # 注册到通用注册表(使用命名空间化的名称)
self._components_by_type[component_type][component_name] = component_info # 类型内部仍使用原名
- self._component_classes[namespaced_name] = component_class
+ self._components_classes[namespaced_name] = component_class
# 根据组件类型进行特定注册(使用原始名称)
- if component_type == ComponentType.ACTION:
- self._register_action_component(component_info, component_class) # type: ignore
- elif component_type == ComponentType.COMMAND:
- self._register_command_component(component_info, component_class) # type: ignore
+ match component_type:
+ case ComponentType.ACTION:
+ ret = self._register_action_component(component_info, component_class) # type: ignore
+ case ComponentType.COMMAND:
+ ret = self._register_command_component(component_info, component_class) # type: ignore
+ case ComponentType.EVENT_HANDLER:
+ ret = self._register_event_handler_component(component_info, component_class) # type: ignore
+ case _:
+ logger.warning(f"未知组件类型: {component_type}")
+ if not ret:
+ return False
logger.debug(
- f"已注册{component_type.value}组件: '{component_name}' -> '{namespaced_name}' "
+ f"已注册{component_type}组件: '{component_name}' -> '{namespaced_name}' "
f"({component_class.__name__}) [插件: {plugin_name}]"
)
return True
- def _register_action_component(self, action_info: ActionInfo, action_class: Type[BaseAction]):
- # -------------------------------- NEED REFACTORING --------------------------------
- # -------------------------------- LOGIC ERROR -------------------------------------
+ def _register_action_component(self, action_info: ActionInfo, action_class: Type[BaseAction]) -> bool:
"""注册Action组件到Action特定注册表"""
- action_name = action_info.name
+ if not (action_name := action_info.name):
+ logger.error(f"Action组件 {action_class.__name__} 必须指定名称")
+ return False
+ if not isinstance(action_info, ActionInfo) or not issubclass(action_class, BaseAction):
+ logger.error(f"注册失败: {action_name} 不是有效的Action")
+ return False
+
self._action_registry[action_name] = action_class
# 如果启用,添加到默认动作集
if action_info.enabled:
self._default_actions[action_name] = action_info
- def _register_command_component(self, command_info: CommandInfo, command_class: Type[BaseCommand]):
+ return True
+
+ def _register_command_component(self, command_info: CommandInfo, command_class: Type[BaseCommand]) -> bool:
"""注册Command组件到Command特定注册表"""
- command_name = command_info.name
+ if not (command_name := command_info.name):
+ logger.error(f"Command组件 {command_class.__name__} 必须指定名称")
+ return False
+ if not isinstance(command_info, CommandInfo) or not issubclass(command_class, BaseCommand):
+ logger.error(f"注册失败: {command_name} 不是有效的Command")
+ return False
+
self._command_registry[command_name] = command_class
- # 编译正则表达式并注册
- if command_info.command_pattern:
+ # 如果启用了且有匹配模式
+ if command_info.enabled and command_info.command_pattern:
pattern = re.compile(command_info.command_pattern, re.IGNORECASE | re.DOTALL)
- self._command_patterns[pattern] = command_class
+ if pattern not in self._command_patterns:
+ self._command_patterns[pattern] = command_name
+
+ logger.warning(f"'{command_name}' 对应的命令模式与 '{self._command_patterns[pattern]}' 重复,忽略此命令")
+
+ return True
+
+ def _register_event_handler_component(
+ self, handler_info: EventHandlerInfo, handler_class: Type[BaseEventHandler]
+ ) -> bool:
+ if not (handler_name := handler_info.name):
+ logger.error(f"EventHandler组件 {handler_class.__name__} 必须指定名称")
+ return False
+ if not isinstance(handler_info, EventHandlerInfo) or not issubclass(handler_class, BaseEventHandler):
+ logger.error(f"注册失败: {handler_name} 不是有效的EventHandler")
+ return False
+
+ self._event_handler_registry[handler_name] = handler_class
+
+ from .events_manager import events_manager # 延迟导入防止循环导入问题
+
+ if events_manager.register_event_subscriber(handler_info, handler_class):
+ self._enabled_event_handlers[handler_name] = handler_class
+ return True
+ else:
+ logger.error(f"注册事件处理器 {handler_name} 失败")
+ return False
# === 组件查询方法 ===
-
- def get_component_info(self, component_name: str, component_type: ComponentType = None) -> Optional[ComponentInfo]: # type: ignore
+ def get_component_info(
+ self, component_name: str, component_type: Optional[ComponentType] = None
+ ) -> Optional[ComponentInfo]:
# sourcery skip: class-extract-method
"""获取组件信息,支持自动命名空间解析
@@ -138,18 +205,12 @@ class ComponentRegistry:
# 2. 如果指定了组件类型,构造命名空间化的名称查找
if component_type:
- if component_type == ComponentType.ACTION:
- namespaced_name = f"action.{component_name}"
- elif component_type == ComponentType.COMMAND:
- namespaced_name = f"command.{component_name}"
- else:
- namespaced_name = f"{component_type.value}.{component_name}"
-
+ namespaced_name = f"{component_type}.{component_name}"
return self._components.get(namespaced_name)
# 3. 如果没有指定类型,尝试在所有命名空间中查找
candidates = []
- for namespace_prefix in ["action", "command"]:
+ for namespace_prefix in [types.value for types in ComponentType]:
namespaced_name = f"{namespace_prefix}.{component_name}"
if component_info := self._components.get(namespaced_name):
candidates.append((namespace_prefix, namespaced_name, component_info))
@@ -171,8 +232,8 @@ class ComponentRegistry:
def get_component_class(
self,
component_name: str,
- component_type: ComponentType = None, # type: ignore
- ) -> Optional[Union[Type[BaseCommand], Type[BaseAction]]]:
+ component_type: Optional[ComponentType] = None,
+ ) -> Optional[Union[Type[BaseCommand], Type[BaseAction], Type[BaseEventHandler]]]:
"""获取组件类,支持自动命名空间解析
Args:
@@ -184,29 +245,23 @@ class ComponentRegistry:
"""
# 1. 如果已经是命名空间化的名称,直接查找
if "." in component_name:
- return self._component_classes.get(component_name)
+ return self._components_classes.get(component_name)
# 2. 如果指定了组件类型,构造命名空间化的名称查找
if component_type:
- if component_type == ComponentType.ACTION:
- namespaced_name = f"action.{component_name}"
- elif component_type == ComponentType.COMMAND:
- namespaced_name = f"command.{component_name}"
- else:
- namespaced_name = f"{component_type.value}.{component_name}"
-
- return self._component_classes.get(namespaced_name)
+ namespaced_name = f"{component_type.value}.{component_name}"
+ return self._components_classes.get(namespaced_name)
# 3. 如果没有指定类型,尝试在所有命名空间中查找
candidates = []
- for namespace_prefix in ["action", "command"]:
+ for namespace_prefix in [types.value for types in ComponentType]:
namespaced_name = f"{namespace_prefix}.{component_name}"
- if component_class := self._component_classes.get(namespaced_name):
+ if component_class := self._components_classes.get(namespaced_name):
candidates.append((namespace_prefix, namespaced_name, component_class))
if len(candidates) == 1:
# 只有一个匹配,直接返回
- namespace, full_name, cls = candidates[0]
+ _, full_name, cls = candidates[0]
logger.debug(f"自动解析组件: '{component_name}' -> '{full_name}'")
return cls
elif len(candidates) > 1:
@@ -235,7 +290,7 @@ class ComponentRegistry:
"""获取Action注册表(用于兼容现有系统)"""
return self._action_registry.copy()
- def get_action_info(self, action_name: str) -> Optional[ActionInfo]:
+ def get_registered_action_info(self, action_name: str) -> Optional[ActionInfo]:
"""获取Action信息"""
info = self.get_component_info(action_name, ComponentType.ACTION)
return info if isinstance(info, ActionInfo) else None
@@ -247,18 +302,18 @@ class ComponentRegistry:
# === Command特定查询方法 ===
def get_command_registry(self) -> Dict[str, Type[BaseCommand]]:
- """获取Command注册表(用于兼容现有系统)"""
+ """获取Command注册表"""
return self._command_registry.copy()
- def get_command_patterns(self) -> Dict[Pattern, Type[BaseCommand]]:
- """获取Command模式注册表(用于兼容现有系统)"""
- return self._command_patterns.copy()
-
- def get_command_info(self, command_name: str) -> Optional[CommandInfo]:
+ def get_registered_command_info(self, command_name: str) -> Optional[CommandInfo]:
"""获取Command信息"""
info = self.get_component_info(command_name, ComponentType.COMMAND)
return info if isinstance(info, CommandInfo) else None
+ def get_command_patterns(self) -> Dict[Pattern, str]:
+ """获取Command模式注册表"""
+ return self._command_patterns.copy()
+
def find_command_by_text(self, text: str) -> Optional[Tuple[Type[BaseCommand], dict, bool, str]]:
# sourcery skip: use-named-expression, use-next
"""根据文本查找匹配的命令
@@ -270,47 +325,36 @@ class ComponentRegistry:
Tuple: (命令类, 匹配的命名组, 是否拦截消息, 插件名) 或 None
"""
- for pattern, command_class in self._command_patterns.items():
- if match := pattern.match(text):
- command_name = None
- # 查找对应的组件信息
- for name, cls in self._command_registry.items():
- if cls == command_class:
- command_name = name
- break
+ 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.intercept_message,
+ command_info.plugin_name,
+ )
- # 检查命令是否启用
- if command_name:
- command_info = self.get_command_info(command_name)
- if command_info and command_info.enabled:
- return (
- command_class,
- match.groupdict(),
- command_info.intercept_message,
- command_info.plugin_name,
- )
- return None
+ # === 事件处理器特定查询方法 ===
- # === 插件管理方法 ===
+ def get_event_handler_registry(self) -> Dict[str, Type[BaseEventHandler]]:
+ """获取事件处理器注册表"""
+ return self._event_handler_registry.copy()
- def register_plugin(self, plugin_info: PluginInfo) -> bool:
- """注册插件
+ def get_registered_event_handler_info(self, handler_name: str) -> Optional[EventHandlerInfo]:
+ """获取事件处理器信息"""
+ info = self.get_component_info(handler_name, ComponentType.EVENT_HANDLER)
+ return info if isinstance(info, EventHandlerInfo) else None
- Args:
- plugin_info: 插件信息
+ def get_enabled_event_handlers(self) -> Dict[str, Type[BaseEventHandler]]:
+ """获取启用的事件处理器"""
+ return self._enabled_event_handlers.copy()
- Returns:
- bool: 是否注册成功
- """
- plugin_name = plugin_info.name
-
- if plugin_name in self._plugins:
- logger.warning(f"插件 {plugin_name} 已存在,跳过注册")
- return False
-
- self._plugins[plugin_name] = plugin_info
- logger.debug(f"已注册插件: {plugin_name} (组件数量: {len(plugin_info.components)})")
- return True
+ # === 插件查询方法 ===
def get_plugin_info(self, plugin_name: str) -> Optional[PluginInfo]:
"""获取插件信息"""
@@ -344,82 +388,22 @@ class ComponentRegistry:
plugin_instance = plugin_manager.get_plugin_instance(plugin_name)
return plugin_instance.config if plugin_instance else None
- # === 状态管理方法 ===
-
- # def enable_component(self, component_name: str, component_type: ComponentType = None) -> bool:
- # # -------------------------------- NEED REFACTORING --------------------------------
- # # -------------------------------- LOGIC ERROR -------------------------------------
- # """启用组件,支持命名空间解析"""
- # # 首先尝试找到正确的命名空间化名称
- # component_info = self.get_component_info(component_name, component_type)
- # if not component_info:
- # return False
-
- # # 根据组件类型构造正确的命名空间化名称
- # if component_info.component_type == ComponentType.ACTION:
- # namespaced_name = f"action.{component_name}" if "." not in component_name else component_name
- # elif component_info.component_type == ComponentType.COMMAND:
- # namespaced_name = f"command.{component_name}" if "." not in component_name else component_name
- # else:
- # namespaced_name = (
- # f"{component_info.component_type.value}.{component_name}"
- # if "." not in component_name
- # else component_name
- # )
-
- # if namespaced_name in self._components:
- # self._components[namespaced_name].enabled = True
- # # 如果是Action,更新默认动作集
- # # ---- HERE ----
- # # if isinstance(component_info, ActionInfo):
- # # self._action_descriptions[component_name] = component_info.description
- # logger.debug(f"已启用组件: {component_name} -> {namespaced_name}")
- # return True
- # return False
-
- # def disable_component(self, component_name: str, component_type: ComponentType = None) -> bool:
- # # -------------------------------- NEED REFACTORING --------------------------------
- # # -------------------------------- LOGIC ERROR -------------------------------------
- # """禁用组件,支持命名空间解析"""
- # # 首先尝试找到正确的命名空间化名称
- # component_info = self.get_component_info(component_name, component_type)
- # if not component_info:
- # return False
-
- # # 根据组件类型构造正确的命名空间化名称
- # if component_info.component_type == ComponentType.ACTION:
- # namespaced_name = f"action.{component_name}" if "." not in component_name else component_name
- # elif component_info.component_type == ComponentType.COMMAND:
- # namespaced_name = f"command.{component_name}" if "." not in component_name else component_name
- # else:
- # namespaced_name = (
- # f"{component_info.component_type.value}.{component_name}"
- # if "." not in component_name
- # else component_name
- # )
-
- # if namespaced_name in self._components:
- # self._components[namespaced_name].enabled = False
- # # 如果是Action,从默认动作集中移除
- # # ---- HERE ----
- # # if component_name in self._action_descriptions:
- # # del self._action_descriptions[component_name]
- # logger.debug(f"已禁用组件: {component_name} -> {namespaced_name}")
- # return True
- # return False
-
def get_registry_stats(self) -> Dict[str, Any]:
"""获取注册中心统计信息"""
action_components: int = 0
command_components: int = 0
+ events_handlers: int = 0
for component in self._components.values():
if component.component_type == ComponentType.ACTION:
action_components += 1
elif component.component_type == ComponentType.COMMAND:
command_components += 1
+ elif component.component_type == ComponentType.EVENT_HANDLER:
+ events_handlers += 1
return {
"action_components": action_components,
"command_components": command_components,
+ "event_handlers": events_handlers,
"total_components": len(self._components),
"total_plugins": len(self._plugins),
"components_by_type": {
@@ -430,5 +414,4 @@ class ComponentRegistry:
}
-# 全局组件注册中心实例
component_registry = ComponentRegistry()
diff --git a/src/plugin_system/core/events_manager.py b/src/plugin_system/core/events_manager.py
index 5143d765c..2c48f9d6d 100644
--- a/src/plugin_system/core/events_manager.py
+++ b/src/plugin_system/core/events_manager.py
@@ -3,7 +3,7 @@ from typing import List, Dict, Optional, Type
from src.chat.message_receive.message import MessageRecv
from src.common.logger import get_logger
-from src.plugin_system.base.component_types import EventType, EventHandlerInfo, MaiMessages, PluginInfo
+from src.plugin_system.base.component_types import EventType, EventHandlerInfo, MaiMessages
from src.plugin_system.base.base_events_handler import BaseEventHandler
logger = get_logger("events_manager")
@@ -14,7 +14,6 @@ class EventsManager:
# 有权重的 events 订阅者注册表
self.events_subscribers: Dict[EventType, List[BaseEventHandler]] = {event: [] for event in EventType}
self.handler_mapping: Dict[str, Type[BaseEventHandler]] = {} # 事件处理器映射表
- self._plugins: Dict[str, PluginInfo] = {} # 插件注册表
def register_event_subscriber(self, handler_info: EventHandlerInfo, handler_class: Type[BaseEventHandler]) -> bool:
"""注册事件处理器
@@ -42,23 +41,6 @@ class EventsManager:
return self._insert_event_handler(handler_class)
- def register_plugins(self, plugin_info: PluginInfo) -> bool:
- """注册插件
-
- Args:
- plugin_info (PluginInfo): 插件信息
-
- Returns:
- bool: 是否注册成功
- """
- if plugin_info.name in self._plugins:
- logger.warning(f"插件 {plugin_info.name} 已存在,跳过注册")
- return False
-
- self._plugins[plugin_info.name] = plugin_info
- logger.debug(f"插件 {plugin_info.name} 注册成功")
- return True
-
async def handler_mai_events(
self,
event_type: EventType,
diff --git a/src/plugin_system/core/plugin_manager.py b/src/plugin_system/core/plugin_manager.py
index be91d9292..3ce9c9e52 100644
--- a/src/plugin_system/core/plugin_manager.py
+++ b/src/plugin_system/core/plugin_manager.py
@@ -457,13 +457,14 @@ class PluginManager:
stats = component_registry.get_registry_stats()
action_count = stats.get("action_components", 0)
command_count = stats.get("command_components", 0)
+ event_handler_count = stats.get("event_handlers", 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})"
+ f"📊 总览: {total_registered}个插件, {total_components}个组件 (Action: {action_count}, Command: {command_count}, EventHandler: {event_handler_count})"
)
# 显示详细的插件列表
@@ -492,8 +493,9 @@ class PluginManager:
# 组件列表
if plugin_info.components:
- action_components = [c for c in plugin_info.components if c.component_type.name == "ACTION"]
- command_components = [c for c in plugin_info.components if c.component_type.name == "COMMAND"]
+ action_components = [c for c in plugin_info.components if c.component_type == ComponentType.ACTION]
+ command_components = [c for c in plugin_info.components if c.component_type == ComponentType.COMMAND]
+ event_handler_components = [c for c in plugin_info.components if c.component_type == ComponentType.EVENT_HANDLER]
if action_components:
action_names = [c.name for c in action_components]
@@ -502,6 +504,10 @@ class PluginManager:
if command_components:
command_names = [c.name for c in command_components]
logger.info(f" ⚡ Command组件: {', '.join(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)}")
# 依赖信息
if plugin_info.dependencies: