from abc import abstractmethod from src.common.logger import get_logger from src.plugin_system.base.component_types import ( ActionInfo, AdapterInfo, CommandInfo, ComponentType, EventHandlerInfo, PlusCommandInfo, PromptInfo, ToolInfo, ) from .base_action import BaseAction from .base_adapter import BaseAdapter from .base_command import BaseCommand from .base_events_handler import BaseEventHandler from .base_prompt import BasePrompt from .base_tool import BaseTool from .plugin_base import PluginBase from .plus_command import PlusCommand logger = get_logger("base_plugin") class BasePlugin(PluginBase): """基于Action和Command的插件基类 所有上述类型的插件都应该继承这个基类,一个插件可以包含多种组件: - Action组件:处理聊天中的动作 - Command组件:处理命令请求 - 未来可扩展:Scheduler、Listener等 """ @classmethod def _get_component_info_from_class(cls, component_class: type, component_type: ComponentType): """从类获取组件信息 Args: component_class: 组件类 component_type: 组件类型 Returns: 对应的ComponentInfo对象 """ if component_type == ComponentType.COMMAND: if hasattr(component_class, "get_command_info"): return component_class.get_command_info() else: logger.warning(f"Command组件 {component_class.__name__} 缺少 get_command_info 方法") return None elif component_type == ComponentType.ACTION: if hasattr(component_class, "get_action_info"): return component_class.get_action_info() else: logger.warning(f"Action组件 {component_class.__name__} 缺少 get_action_info 方法") return None elif component_type == ComponentType.PLUS_COMMAND: # PlusCommand组件的get_info方法尚未实现 logger.warning("PlusCommand组件的get_info方法尚未实现") return None elif component_type == ComponentType.TOOL: # Tool组件的get_info方法尚未实现 logger.warning("Tool组件的get_info方法尚未实现") return None elif component_type == ComponentType.EVENT_HANDLER: # EventHandler组件的get_info方法尚未实现 logger.warning("EventHandler组件的get_info方法尚未实现") return None elif component_type == ComponentType.PROMPT: if hasattr(component_class, "get_prompt_info"): return component_class.get_prompt_info() else: logger.warning(f"Prompt组件 {component_class.__name__} 缺少 get_prompt_info 方法") return None elif component_type == ComponentType.ADAPTER: if hasattr(component_class, "get_adapter_info"): return component_class.get_adapter_info() else: logger.warning(f"Adapter�� {component_class.__name__} ȱ�� get_adapter_info ����") return None else: logger.error(f"不支持的组件类型: {component_type}") return None @classmethod def get_component_info(cls, component_class: type, component_type: ComponentType): """获取组件信息 Args: component_class: 组件类 component_type: 组件类型 Returns: 对应的ComponentInfo对象 """ return cls._get_component_info_from_class(component_class, component_type) @abstractmethod def get_plugin_components( self, ) -> list[ tuple[ActionInfo, type[BaseAction]] | tuple[AdapterInfo, type[BaseAdapter]] | tuple[CommandInfo, type[BaseCommand]] | tuple[PlusCommandInfo, type[PlusCommand]] | tuple[EventHandlerInfo, type[BaseEventHandler]] | tuple[ToolInfo, type[BaseTool]] | tuple[PromptInfo, type[BasePrompt]] ]: """获取插件包含的组件列表 子类必须实现此方法,返回组件信息和组件类的列表 Returns: List[tuple[ComponentInfo, Type]]: [(组件信息, 组件类), ...] """ ... def register_plugin(self) -> bool: """注册插件及其所有组件""" from src.plugin_system.core.component_registry import component_registry components = self.get_plugin_components() # 注册所有组件 registered_components = [] for component_info, component_class in components: component_info.plugin_name = self.plugin_name if component_registry.register_component(component_info, component_class): registered_components.append(component_info) else: logger.warning(f"{self.log_prefix} 组件 {component_info.name} 注册失败") # 更新插件信息中的组件列表 self.plugin_info.components = registered_components # 注册插件 if component_registry.register_plugin(self.plugin_info): logger.debug(f"{self.log_prefix} 插件注册成功,包含 {len(registered_components)} 个组件") return True else: logger.error(f"{self.log_prefix} 插件注册失败") return False