95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
from typing import Dict, List, Optional, Type
|
||
from base.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)]
|