286 lines
9.2 KiB
Python
286 lines
9.2 KiB
Python
import importlib
|
||
import inspect
|
||
import os
|
||
import sys
|
||
from typing import Dict, List, Any, Optional, Type
|
||
|
||
from plugin_common.plugin_interface import PluginInterface, PluginStatus
|
||
from plugin_common.message_plugin_interface import MessagePluginInterface
|
||
from plugin_common.scheduled_plugin_interface import ScheduledPluginInterface
|
||
from plugin_common.plugin_registry import PluginRegistry
|
||
from plugin_common.event_system import EventSystem, EventType
|
||
|
||
|
||
class PluginManager:
|
||
"""插件管理器，负责插件的加载、初始化、启动、停止和卸载"""
|
||
|
||
def __init__(self, plugin_dir: str = "plugins"):
|
||
"""
|
||
初始化插件管理器
|
||
|
||
Args:
|
||
plugin_dir: 插件目录
|
||
"""
|
||
self.plugin_dir = plugin_dir
|
||
self.plugins: Dict[str, PluginInterface] = {} # 插件实例字典
|
||
self.plugin_modules = {} # 插件模块字典
|
||
self.system_context = {} # 系统上下文
|
||
|
||
# 确保插件目录存在
|
||
if not os.path.exists(self.plugin_dir):
|
||
os.makedirs(self.plugin_dir)
|
||
|
||
# 将插件目录添加到Python路径
|
||
if self.plugin_dir not in sys.path:
|
||
sys.path.insert(0, self.plugin_dir)
|
||
|
||
def set_system_context(self, context: Dict[str, Any]):
|
||
"""
|
||
设置系统上下文
|
||
|
||
Args:
|
||
context: 系统上下文
|
||
"""
|
||
self.system_context = context
|
||
|
||
def discover_plugins(self) -> List[str]:
|
||
"""
|
||
发现可用插件
|
||
|
||
Returns:
|
||
插件模块名称列表
|
||
"""
|
||
plugin_modules = []
|
||
|
||
# 遍历插件目录
|
||
for item in os.listdir(self.plugin_dir):
|
||
if os.path.isdir(os.path.join(self.plugin_dir, item)) and not item.startswith("__"):
|
||
# 检查是否有__init__.py文件
|
||
if os.path.exists(os.path.join(self.plugin_dir, item, "__init__.py")):
|
||
plugin_modules.append(item)
|
||
elif item.endswith(".py") and not item.startswith("__"):
|
||
# 单文件插件
|
||
plugin_modules.append(item[:-3])
|
||
|
||
return plugin_modules
|
||
|
||
def load_plugin(self, plugin_name: str) -> Optional[PluginInterface]:
|
||
"""
|
||
加载插件
|
||
|
||
Args:
|
||
plugin_name: 插件名称
|
||
|
||
Returns:
|
||
插件实例，加载失败返回None
|
||
"""
|
||
try:
|
||
# 如果插件已加载，直接返回
|
||
if plugin_name in self.plugins:
|
||
return self.plugins[plugin_name]
|
||
|
||
# 确定插件路径
|
||
if os.path.isdir(os.path.join(self.plugin_dir, plugin_name)):
|
||
plugin_path = os.path.join(self.plugin_dir, plugin_name)
|
||
|
||
# 优先从main.py加载插件
|
||
main_module_path = f"{plugin_name}.main"
|
||
if os.path.exists(os.path.join(plugin_path, "main.py")):
|
||
try:
|
||
module = importlib.import_module(main_module_path)
|
||
self.plugin_modules[plugin_name] = module
|
||
except ImportError:
|
||
# 如果main.py导入失败，尝试从__init__.py加载
|
||
module = importlib.import_module(plugin_name)
|
||
self.plugin_modules[plugin_name] = module
|
||
else:
|
||
# 如果没有main.py，从__init__.py加载
|
||
module = importlib.import_module(plugin_name)
|
||
self.plugin_modules[plugin_name] = module
|
||
else:
|
||
# 单文件插件
|
||
plugin_path = self.plugin_dir
|
||
module = importlib.import_module(plugin_name)
|
||
self.plugin_modules[plugin_name] = module
|
||
|
||
# 查找插件类
|
||
plugin_class = None
|
||
for name, obj in inspect.getmembers(module):
|
||
if (inspect.isclass(obj) and
|
||
issubclass(obj, PluginInterface) and
|
||
obj != PluginInterface
|
||
# 修改这里：不排除继承自 MessagePluginInterface 和 ScheduledPluginInterface 的类
|
||
# obj != MessagePluginInterface and
|
||
# obj != ScheduledPluginInterface
|
||
):
|
||
plugin_class = obj
|
||
break
|
||
|
||
if plugin_class is None:
|
||
print(f"插件 {plugin_name} 中未找到有效的插件类")
|
||
return None
|
||
|
||
# 实例化插件
|
||
plugin = plugin_class()
|
||
plugin.status = PluginStatus.LOADED
|
||
|
||
# 设置插件路径
|
||
plugin.set_plugin_path(plugin_path)
|
||
|
||
# 加载插件配置
|
||
if not plugin.load_config():
|
||
print(f"插件 {plugin_name} 加载配置失败")
|
||
return None
|
||
|
||
# 初始化插件
|
||
if not plugin.initialize(self.system_context):
|
||
print(f"插件 {plugin_name} 初始化失败")
|
||
return None
|
||
|
||
# 注册插件
|
||
PluginRegistry().register(plugin)
|
||
|
||
# 存储插件实例
|
||
self.plugins[plugin.name] = plugin
|
||
|
||
# 发布插件加载事件
|
||
EventSystem().publish(EventType.PLUGIN_LOADED, {"plugin": plugin})
|
||
|
||
return plugin
|
||
|
||
except Exception as e:
|
||
print(f"加载插件 {plugin_name} 失败: {e}")
|
||
return None
|
||
|
||
def load_all_plugins(self) -> Dict[str, PluginInterface]:
|
||
"""
|
||
加载所有插件
|
||
|
||
Returns:
|
||
插件实例字典
|
||
"""
|
||
plugin_modules = self.discover_plugins()
|
||
|
||
for module_name in plugin_modules:
|
||
self.load_plugin(module_name)
|
||
|
||
return self.plugins
|
||
|
||
def unload_plugin(self, plugin_name: str) -> bool:
|
||
"""
|
||
卸载插件
|
||
|
||
Args:
|
||
plugin_name: 插件名称
|
||
|
||
Returns:
|
||
卸载是否成功
|
||
"""
|
||
if plugin_name not in self.plugins:
|
||
print(f"插件 {plugin_name} 未加载")
|
||
return False
|
||
|
||
plugin = self.plugins[plugin_name]
|
||
|
||
# 停止插件
|
||
if plugin.status == PluginStatus.RUNNING:
|
||
if not plugin.stop():
|
||
print(f"停止插件 {plugin_name} 失败")
|
||
return False
|
||
|
||
# 清理插件资源
|
||
if not plugin.cleanup():
|
||
print(f"清理插件 {plugin_name} 资源失败")
|
||
return False
|
||
|
||
# 注销插件
|
||
PluginRegistry().unregister(plugin_name)
|
||
|
||
# 移除插件实例
|
||
del self.plugins[plugin_name]
|
||
|
||
# 发布插件卸载事件
|
||
EventSystem().publish(EventType.PLUGIN_UNLOADED, {"plugin_name": plugin_name})
|
||
|
||
return True
|
||
|
||
def start_plugin(self, plugin_name: str) -> bool:
|
||
"""
|
||
启动插件
|
||
|
||
Args:
|
||
plugin_name: 插件名称
|
||
|
||
Returns:
|
||
启动是否成功
|
||
"""
|
||
if plugin_name not in self.plugins:
|
||
print(f"插件 {plugin_name} 未加载")
|
||
return False
|
||
|
||
plugin = self.plugins[plugin_name]
|
||
|
||
if plugin.status == PluginStatus.RUNNING:
|
||
print(f"插件 {plugin_name} 已经在运行")
|
||
return True
|
||
|
||
if plugin.start():
|
||
plugin.status = PluginStatus.RUNNING
|
||
return True
|
||
else:
|
||
plugin.status = PluginStatus.ERROR
|
||
return False
|
||
|
||
def stop_plugin(self, plugin_name: str) -> bool:
|
||
"""
|
||
停止插件
|
||
|
||
Args:
|
||
plugin_name: 插件名称
|
||
|
||
Returns:
|
||
停止是否成功
|
||
"""
|
||
if plugin_name not in self.plugins:
|
||
print(f"插件 {plugin_name} 未加载")
|
||
return False
|
||
|
||
plugin = self.plugins[plugin_name]
|
||
|
||
if plugin.status != PluginStatus.RUNNING:
|
||
print(f"插件 {plugin_name} 未在运行")
|
||
return True
|
||
|
||
if plugin.stop():
|
||
plugin.status = PluginStatus.STOPPED
|
||
return True
|
||
else:
|
||
plugin.status = PluginStatus.ERROR
|
||
return False
|
||
|
||
def reload_plugin(self, plugin_name: str) -> Optional[PluginInterface]:
|
||
"""
|
||
重新加载插件
|
||
|
||
Args:
|
||
plugin_name: 插件名称
|
||
|
||
Returns:
|
||
插件实例，重新加载失败返回None
|
||
"""
|
||
# 卸载插件
|
||
if plugin_name in self.plugins:
|
||
if not self.unload_plugin(plugin_name):
|
||
print(f"卸载插件 {plugin_name} 失败")
|
||
return None
|
||
|
||
# 重新导入模块
|
||
if plugin_name in self.plugin_modules:
|
||
try:
|
||
importlib.reload(self.plugin_modules[plugin_name])
|
||
except Exception as e:
|
||
print(f"重新导入插件模块 {plugin_name} 失败: {e}")
|
||
return None
|
||
|
||
# 加载插件
|
||
return self.load_plugin(plugin_name) |