gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refc：重构插件api，补全文档，合并expressor和replyer，分离reply和sender，新log浏览器

Browse Source
This commit is contained in:
SengokuCola
2025-06-19 20:20:34 +08:00
parent 7e05ede846
commit ab28b94e33
63 changed files with 5285 additions and 8316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
src/plugin_system/core/component_registry.py
View File

@@ -54,23 +54,43 @@ class ComponentRegistry:
"""
component_name = component_info.name
component_type = component_info.component_type
plugin_name = getattr(component_info, 'plugin_name', 'unknown')
if component_name in self._components:
logger.warning(f"组件 {component_name} 已存在，跳过注册")
# 🔥 系统级别自动区分：为不同类型的组件添加命名空间前缀
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}"
# 检查命名空间化的名称是否冲突
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}' 的注册"
)
return False
# 注册到通用注册表
self._components[component_name] = component_info
self._components_by_type[component_type][component_name] = component_info
self._component_classes[component_name] = component_class
# 注册到通用注册表（使用命名空间化的名称）
self._components[namespaced_name] = component_info
self._components_by_type[component_type][component_name] = component_info # 类型内部仍使用原名
self._component_classes[namespaced_name] = component_class
# 根据组件类型进行特定注册
# 根据组件类型进行特定注册（使用原始名称）
if component_type == ComponentType.ACTION:
self._register_action_component(component_info, component_class)
elif component_type == ComponentType.COMMAND:
self._register_command_component(component_info, component_class)
logger.debug(f"已注册{component_type.value}组件: {component_name} ({component_class.__name__})")
logger.debug(
f"已注册{component_type.value}组件: '{component_name}' -> '{namespaced_name}' "
f"({component_class.__name__}) [插件: {plugin_name}]"
)
return True
def _register_action_component(self, action_info: ActionInfo, action_class: Type):
@@ -94,13 +114,103 @@ class ComponentRegistry:
# === 组件查询方法 ===
def get_component_info(self, component_name: str) -> Optional[ComponentInfo]:
"""获取组件信息"""
return self._components.get(component_name)
def get_component_info(self, component_name: str, component_type: ComponentType = None) -> Optional[ComponentInfo]:
"""获取组件信息，支持自动命名空间解析
Args:
component_name: 组件名称，可以是原始名称或命名空间化的名称
component_type: 组件类型，如果提供则优先在该类型中查找
Returns:
Optional[ComponentInfo]: 组件信息或None
"""
# 1. 如果已经是命名空间化的名称，直接查找
if '.' in component_name:
return self._components.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._components.get(namespaced_name)
# 3. 如果没有指定类型，尝试在所有命名空间中查找
candidates = []
for namespace_prefix in ["action", "command"]:
namespaced_name = f"{namespace_prefix}.{component_name}"
component_info = self._components.get(namespaced_name)
if component_info:
candidates.append((namespace_prefix, namespaced_name, component_info))
if len(candidates) == 1:
# 只有一个匹配，直接返回
return candidates[0][2]
elif len(candidates) > 1:
# 多个匹配，记录警告并返回第一个
namespaces = [ns for ns, _, _ in candidates]
logger.warning(
f"组件名称 '{component_name}' 在多个命名空间中存在: {namespaces}，"
f"使用第一个匹配项: {candidates[0][1]}"
)
return candidates[0][2]
# 4. 都没找到
return None
def get_component_class(self, component_name: str) -> Optional[Type]:
"""获取组件类"""
return self._component_classes.get(component_name)
def get_component_class(self, component_name: str, component_type: ComponentType = None) -> Optional[Type]:
"""获取组件类，支持自动命名空间解析
Args:
component_name: 组件名称，可以是原始名称或命名空间化的名称
component_type: 组件类型，如果提供则优先在该类型中查找
Returns:
Optional[Type]: 组件类或None
"""
# 1. 如果已经是命名空间化的名称，直接查找
if '.' in component_name:
return self._component_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)
# 3. 如果没有指定类型，尝试在所有命名空间中查找
candidates = []
for namespace_prefix in ["action", "command"]:
namespaced_name = f"{namespace_prefix}.{component_name}"
component_class = self._component_classes.get(namespaced_name)
if component_class:
candidates.append((namespace_prefix, namespaced_name, component_class))
if len(candidates) == 1:
# 只有一个匹配，直接返回
namespace, full_name, cls = candidates[0]
logger.debug(f"自动解析组件: '{component_name}' -> '{full_name}'")
return cls
elif len(candidates) > 1:
# 多个匹配，记录警告并返回第一个
namespaces = [ns for ns, _, _ in candidates]
logger.warning(
f"组件名称 '{component_name}' 在多个命名空间中存在: {namespaces}，"
f"使用第一个匹配项: {candidates[0][1]}"
)
return candidates[0][2]
# 4. 都没找到
return None
def get_components_by_type(self, component_type: ComponentType) -> Dict[str, ComponentInfo]:
"""获取指定类型的所有组件"""
@@ -123,7 +233,7 @@ class ComponentRegistry:
def get_action_info(self, action_name: str) -> Optional[ActionInfo]:
"""获取Action信息"""
info = self.get_component_info(action_name)
info = self.get_component_info(action_name, ComponentType.ACTION)
return info if isinstance(info, ActionInfo) else None
# === Command特定查询方法 ===
@@ -138,7 +248,7 @@ class ComponentRegistry:
def get_command_info(self, command_name: str) -> Optional[CommandInfo]:
"""获取Command信息"""
info = self.get_component_info(command_name)
info = self.get_component_info(command_name, ComponentType.COMMAND)
return info if isinstance(info, CommandInfo) else None
def find_command_by_text(self, text: str) -> Optional[tuple[Type, dict, bool, str]]:
@@ -150,7 +260,9 @@ class ComponentRegistry:
Returns:
Optional[tuple[Type, dict, bool, str]]: (命令类, 匹配的命名组, 是否拦截消息, 插件名) 或 None
"""
for pattern, command_class in self._command_patterns.items():
match = pattern.match(text)
if match:
command_name = None
@@ -159,17 +271,18 @@ class ComponentRegistry:
if cls == command_class:
command_name = name
break
# 检查命令是否启用
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,
)
if command_info:
if command_info.enabled:
return (
command_class,
match.groupdict(),
command_info.intercept_message,
command_info.plugin_name,
)
return None
# === 插件管理方法 ===
@@ -227,26 +340,51 @@ class ComponentRegistry:
# === 状态管理方法 ===
def enable_component(self, component_name: str) -> bool:
"""启用组件"""
if component_name in self._components:
self._components[component_name].enabled = True
def enable_component(self, component_name: str, component_type: ComponentType = None) -> bool:
"""启用组件，支持命名空间解析"""
# 首先尝试找到正确的命名空间化名称
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，更新默认动作集
component_info = self._components[component_name]
if isinstance(component_info, ActionInfo):
self._default_actions[component_name] = component_info.description
logger.debug(f"已启用组件: {component_name}")
logger.debug(f"已启用组件: {component_name} -> {namespaced_name}")
return True
return False
def disable_component(self, component_name: str) -> bool:
"""禁用组件"""
if component_name in self._components:
self._components[component_name].enabled = False
def disable_component(self, component_name: str, component_type: ComponentType = None) -> bool:
"""禁用组件，支持命名空间解析"""
# 首先尝试找到正确的命名空间化名称
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，从默认动作集中移除
if component_name in self._default_actions:
del self._default_actions[component_name]
logger.debug(f"已禁用组件: {component_name}")
logger.debug(f"已禁用组件: {component_name} -> {namespaced_name}")
return True
return False