from typing import Dict, List, Optional, Type from plugin_common.plugin_interface import PluginInterface, PluginStatus class PluginRegistry: """插件注册表,维护已加载插件的信息和状态""" _instance = None def __new__(cls): """单例模式""" if cls._instance is None: cls._instance = super(PluginRegistry, cls).__new__(cls) cls._instance._plugins = {} return cls._instance def register(self, plugin: PluginInterface) -> bool: """ 注册插件 Args: plugin: 插件实例 Returns: 注册是否成功 """ if plugin.name in self._plugins: print(f"插件 {plugin.name} 已存在") return False self._plugins[plugin.name] = plugin return True def unregister(self, plugin_name: str) -> bool: """ 注销插件 Args: plugin_name: 插件名称 Returns: 注销是否成功 """ if plugin_name not in self._plugins: print(f"插件 {plugin_name} 不存在") return False del self._plugins[plugin_name] return True def get_plugin(self, plugin_name: str) -> Optional[PluginInterface]: """ 获取插件实例 Args: plugin_name: 插件名称 Returns: 插件实例,不存在返回None """ return self._plugins.get(plugin_name) def get_all_plugins(self) -> Dict[str, PluginInterface]: """ 获取所有插件 Returns: 插件字典,键为插件名称,值为插件实例 """ return self._plugins.copy() def get_plugins_by_status(self, status: PluginStatus) -> List[PluginInterface]: """ 获取指定状态的插件 Args: status: 插件状态 Returns: 插件列表 """ return [p for p in self._plugins.values() if p.status == status] def get_plugins_by_type(self, plugin_type: Type) -> List[PluginInterface]: """ 获取指定类型的插件 Args: plugin_type: 插件类型 Returns: 插件列表 """ return [p for p in self._plugins.values() if isinstance(p, plugin_type)]