""" 插件加载器 - 按需加载，惰性实例化 """ import importlib.util import sys import logging from pathlib import Path from typing import Dict, Any, Optional, List, Callable from fastapi import APIRouter, FastAPI from app.plugins.models import PluginInfo, PluginStatus from app.config.settings import settings logger = logging.getLogger(__name__) class DefaultPlugin: """默认插件实现 - 无生命周期钩子""" def on_enable(self): pass def on_disable(self): pass class PluginLoader: """插件加载器 - 仅加载启用插件，惰性实例化""" def __init__(self): self._loaded_plugins: Dict[str, Any] = {} self._plugin_routers: Dict[str, APIRouter] = {} self._plugin_factories: Dict[str, Dict] = {} # 存储工厂函数，延迟实例化 def load_plugin(self, plugin_info: PluginInfo) -> bool: """加载插件模块（仅导入，不实例化）""" plugin_name = plugin_info.metadata.name if plugin_name in self._plugin_factories: return True plugin_path = Path(plugin_info.install_path) # 1. 动态导入插件主模块 main_module = self._import_plugin_module(plugin_path, "main") if not main_module: logger.error(f"无法加载插件主模块: {plugin_name}") return False # 2. 存储工厂函数而非实例 factory = self._get_plugin_factory(main_module) self._plugin_factories[plugin_name] = { "factory": factory, "module": main_module, "info": plugin_info, } # 3. 仅对已启用插件执行实例化并调用 on_enable logger.info(f"插件状态检查: {plugin_name} - status={plugin_info.status}, is_enabled={plugin_info.status == PluginStatus.ENABLED}") if plugin_info.status == PluginStatus.ENABLED: instance = self._instantiate_plugin(plugin_name) logger.info(f"实例化结果: {plugin_name} - instance={instance is not None}, has_on_enable={hasattr(instance, 'on_enable') if instance else False}") if instance and hasattr(instance, "on_enable"): instance.on_enable() logger.info(f"已启用插件: {plugin_name}") logger.info(f"插件模块已加载: {plugin_name}") return True def load_plugin_from_dir(self, plugin_dir: Path) -> bool: """从目录加载插件模块（兼容新的 manager 接口） Args: plugin_dir: 插件目录路径 Returns: 是否加载成功 """ plugin_name = plugin_dir.name # 如果已加载，返回 True if plugin_name in self._plugin_factories: return True # 动态导入插件主模块 main_module = self._import_plugin_module(plugin_dir, "main") if not main_module: logger.error(f"无法加载插件主模块: {plugin_name}") return False # 存储工厂函数 factory = self._get_plugin_factory(main_module) self._plugin_factories[plugin_name] = { "factory": factory, "module": main_module, "info": None, # 不再依赖 PluginInfo } # 实例化并调用 on_enable instance = self._instantiate_plugin(plugin_name) if instance and hasattr(instance, "on_enable"): instance.on_enable() logger.info(f"已启用插件: {plugin_name}") logger.info(f"插件模块已加载: {plugin_name}") return True def _instantiate_plugin(self, plugin_name: str) -> Optional[Any]: """实例化插件""" factory_data = self._plugin_factories.get(plugin_name) logger.info(f"_instantiate_plugin: {plugin_name} - factory_data={factory_data is not None}") if not factory_data: return None if plugin_name in self._loaded_plugins: logger.info(f"_instantiate_plugin: {plugin_name} - 已存在实例") return self._loaded_plugins[plugin_name]["instance"] logger.info(f"_instantiate_plugin: {plugin_name} - 创建新实例, factory={factory_data['factory']}") instance = factory_data["factory"]() logger.info(f"_instantiate_plugin: {plugin_name} - instance={instance is not None}, type={type(instance).__name__ if instance else 'None'}") self._loaded_plugins[plugin_name] = { "instance": instance, "module": factory_data["module"], "info": factory_data["info"], } return instance def _get_plugin_factory(self, module) -> Callable: """获取插件工厂函数""" plugin_class = None if hasattr(module, "Plugin"): plugin_class = getattr(module, "Plugin") else: for attr_name in dir(module): if attr_name.endswith("Plugin") and attr_name != "Plugin": plugin_class = getattr(module, attr_name) break if plugin_class and callable(plugin_class): return plugin_class else: return DefaultPlugin def unload_plugin(self, plugin_name: str) -> bool: """卸载插件""" if plugin_name in self._loaded_plugins: try: plugin_data = self._loaded_plugins[plugin_name] if hasattr(plugin_data["instance"], "on_disable"): plugin_data["instance"].on_disable() del self._loaded_plugins[plugin_name] if plugin_name in self._plugin_routers: del self._plugin_routers[plugin_name] logger.info(f"成功卸载插件: {plugin_name}") return True except Exception as e: logger.error(f"卸载插件失败 {plugin_name}: {e}") return False return False def enable_plugin(self, plugin_name: str) -> bool: """启用插件 - 惰性实例化并调用 on_enable""" logger.info(f"enable_plugin: {plugin_name} - 已在loaded_plugins={plugin_name in self._loaded_plugins}") if plugin_name in self._loaded_plugins: instance = self._loaded_plugins[plugin_name]["instance"] logger.info(f"enable_plugin: {plugin_name} - 调用 on_enable (已存在实例)") if hasattr(instance, "on_enable"): instance.on_enable() logger.info(f"enable_plugin: {plugin_name} - on_enable 已调用") return True logger.info(f"enable_plugin: {plugin_name} - 调用 _instantiate_plugin") instance = self._instantiate_plugin(plugin_name) logger.info(f"enable_plugin: {plugin_name} - instance={instance is not None}") if instance: if hasattr(instance, "on_enable"): logger.info(f"enable_plugin: {plugin_name} - 调用 on_enable (新实例)") instance.on_enable() logger.info(f"插件 {plugin_name} 已启用") return True return False def disable_plugin(self, plugin_name: str) -> bool: """禁用插件 - 调用 on_disable""" if plugin_name in self._loaded_plugins: try: plugin_data = self._loaded_plugins[plugin_name] if hasattr(plugin_data["instance"], "on_disable"): plugin_data["instance"].on_disable() return True except Exception as e: logger.error(f"禁用插件失败 {plugin_name}: {e}") return False return True def register_plugin_routes(self, app: FastAPI, plugin_info: PluginInfo) -> bool: """注册插件路由""" try: plugin_path = Path(plugin_info.install_path) plugin_name = plugin_info.metadata.name api_router = self.create_api_router(plugin_path, plugin_info) if api_router: base_path = f"/plugins/{plugin_name}/api" app.include_router(api_router, prefix=base_path) api_base_path = f"/api/plugins/{plugin_name}" app.include_router(api_router, prefix=api_base_path) self._plugin_routers[plugin_name] = api_router logger.info(f"注册API路由: {base_path}") logger.info(f"注册API兼容路由: {api_base_path}") return True except Exception as e: logger.error(f"注册插件路由失败 {plugin_name}: {e}") return False def _import_plugin_module(self, plugin_path: Path, module_name: str) -> Optional[Any]: """动态导入插件模块""" try: module_file = plugin_path / f"{module_name}.py" if not module_file.exists(): return None full_module_name = f"plugins.{plugin_path.name}.{module_name}" # 如果模块已存在，返回已存在的模块（避免重复导入导致不同实例） if full_module_name in sys.modules: return sys.modules[full_module_name] spec = importlib.util.spec_from_file_location(full_module_name, module_file) if spec is None or spec.loader is None: return None module = importlib.util.module_from_spec(spec) sys.modules[full_module_name] = module spec.loader.exec_module(module) return module except Exception as e: logger.error(f"导入模块失败 {module_name}: {e}") return None def create_api_router(self, plugin_path: Path, plugin_info: PluginInfo) -> Optional[APIRouter]: """创建 API 路由器""" try: api_module = self._import_plugin_module(plugin_path, "api") if api_module and hasattr(api_module, "router"): router = getattr(api_module, "router") if isinstance(router, APIRouter): if hasattr(api_module, "set_plugin_instance"): plugin_instance = self.get_plugin_instance(plugin_info.metadata.name) if plugin_instance: api_module.set_plugin_instance(plugin_instance) return router return None except Exception as e: logger.error(f"创建 API 路由器失败 {plugin_info.metadata.name}: {e}") return None def get_plugin_instance(self, plugin_name: str) -> Optional[Any]: """获取插件实例""" return self._loaded_plugins.get(plugin_name, {}).get("instance") def get_loaded_plugins(self) -> List[str]: """获取已加载的插件列表""" return list(self._loaded_plugins.keys()) def is_plugin_loaded(self, plugin_name: str) -> bool: """检查插件是否已实例化""" return plugin_name in self._loaded_plugins @property def plugin_routers(self) -> Dict[str, APIRouter]: """公共访问器：获取已注册的路由器""" return self._plugin_routers @property def loaded_plugins(self) -> Dict[str, Any]: """公共访问器：获取已加载的插件""" return self._loaded_plugins plugin_loader = PluginLoader()