""" 插件管理器 插件运行时状态事实源，负责启动扫描、状态持久化、详情聚合和启停约束。 """ import json import logging from typing import Dict, List, Optional, Any from pathlib import Path from .models import ( PluginStatus, PluginMetadata, PluginOperationResult, PluginListItem, PluginDetail, PluginErrorInfo, PluginErrorStage, PluginGovernanceInfo, PluginCategory, PluginToolSummary, PluginMcpSummary, PluginUIEntry, PluginUIType, UISchemaDefinition, UISubmitContract, UIFieldDefinition, ) from .loader import plugin_loader from .dependency_manager import dependency_manager from .state_store import get_state_store_manager from app.config.settings import settings logger = logging.getLogger(__name__) # 全局应用引用（由 main.py 在启动时设置） _app_instance = None def set_app_instance(app): """设置全局应用实例""" global _app_instance _app_instance = app class PluginManager: """插件管理器 插件运行时状态事实源，负责： - 启动扫描和结构校验 - 状态持久化（data/plugin_state.json） - 详情聚合（依赖、治理、Tool、MCP、UI） - 启停约束（依赖检查、加载状态） """ def __init__(self): self.plugin_directory = settings.PLUGINS_DIR self.state_store = get_state_store_manager( settings.DATA_DIR / "plugin_state.json" ) # 插件运行时信息（包含元数据、状态、错误等） self._plugins: Dict[str, Dict[str, Any]] = {} # 启动扫描 self._scan_plugins() def _scan_plugins(self): """启动扫描插件 分别处理结构校验、依赖检查、加载摘要和 UI entry 检测。 单个插件失败不阻断其他插件。 """ logger.info("扫描插件目录...") if not self.plugin_directory.exists(): logger.warning(f"插件目录不存在: {self.plugin_directory}") return for plugin_dir in self.plugin_directory.iterdir(): if not plugin_dir.is_dir(): continue try: self._scan_single_plugin(plugin_dir) except Exception as e: # 单个插件失败不阻断其他插件 plugin_name = plugin_dir.name logger.error(f"扫描插件 {plugin_name} 失败: {e}") self._plugins[plugin_name] = { "metadata": None, "status": PluginStatus.ERROR, "install_path": str(plugin_dir), "errors": [ PluginErrorInfo( error_code="SCAN_FAILED", message=f"扫描失败: {str(e)}", stage=PluginErrorStage.STRUCTURE, detail=str(e), ) ], "dependency_check": None, "governance": None, "tool_summary": PluginToolSummary(), "mcp_summary": PluginMcpSummary(), "ui_entry": None, } logger.info(f"扫描完成，共 {len(self._plugins)} 个插件") def _scan_single_plugin(self, plugin_dir: Path): """扫描单个插件 Args: plugin_dir: 插件目录 """ plugin_name = plugin_dir.name errors: List[PluginErrorInfo] = [] # 1. 结构校验 - 读取 plugin.json metadata, metadata_dict, struct_error = self._validate_structure(plugin_dir) if struct_error: errors.append(struct_error) self._plugins[plugin_name] = { "metadata": None, "status": PluginStatus.ERROR, "install_path": str(plugin_dir), "errors": errors, "dependency_check": None, "governance": None, "tool_summary": PluginToolSummary(), "mcp_summary": PluginMcpSummary(), "ui_entry": None, } return # 2. 依赖检查 dep_result = dependency_manager.check_plugin_dependencies( plugin_path=plugin_dir, requirements=metadata_dict.get("requirements", []), ) # 3. 从状态文件获取用户启停状态 stored_enabled = self.state_store.is_enabled(plugin_name) if stored_enabled is None: # 首次扫描，从 plugin.json 初始化 default_enabled = metadata_dict.get("enabled", False) self.state_store.initialize_from_metadata(plugin_name, default_enabled) stored_enabled = default_enabled # 4. 确定初始状态 if dep_result.overall_status.value in ("missing", "version_mismatch", "check_failed"): initial_status = PluginStatus.DEPENDENCY_ERROR elif not metadata: initial_status = PluginStatus.INCOMPLETE else: initial_status = PluginStatus.ENABLED if stored_enabled else PluginStatus.DISABLED # 5. 检测 UI 入口 ui_entry = self._detect_ui_entry(plugin_dir, metadata_dict) # 6. 构建治理信息 governance = self._build_governance(metadata_dict, plugin_dir) # 7. 尝试加载插件（如果状态允许） load_error = None if initial_status == PluginStatus.ENABLED: try: if not plugin_loader.load_plugin_from_dir(plugin_dir): load_error = PluginErrorInfo( error_code="LOAD_FAILED", message="插件加载失败", stage=PluginErrorStage.LOAD, ) errors.append(load_error) initial_status = PluginStatus.ERROR except Exception as e: load_error = PluginErrorInfo( error_code="LOAD_EXCEPTION", message=f"插件加载异常: {str(e)}", stage=PluginErrorStage.LOAD, detail=str(e), ) errors.append(load_error) initial_status = PluginStatus.ERROR # 8. 存储插件信息 self._plugins[plugin_name] = { "metadata": metadata, "metadata_dict": metadata_dict, "status": initial_status, "install_path": str(plugin_dir), "errors": errors, "dependency_check": dep_result, "governance": governance, "tool_summary": PluginToolSummary(), # T-006 填充 "mcp_summary": PluginMcpSummary(), # T-007 填充 "ui_entry": ui_entry, } logger.info(f"扫描插件: {plugin_name} (状态: {initial_status.value})") def _validate_structure( self, plugin_dir: Path ) -> tuple: """校验插件结构 Returns: (metadata, metadata_dict, error) 元组 """ plugin_json_path = plugin_dir / "plugin.json" if not plugin_json_path.exists(): return None, None, PluginErrorInfo( error_code="MISSING_PLUGIN_JSON", message="缺少 plugin.json 文件", stage=PluginErrorStage.STRUCTURE, ) try: with open(plugin_json_path, "r", encoding="utf-8") as f: content = f.read().strip() if not content: return None, None, PluginErrorInfo( error_code="EMPTY_PLUGIN_JSON", message="plugin.json 文件为空", stage=PluginErrorStage.STRUCTURE, ) metadata_dict = json.loads(content) except json.JSONDecodeError as e: return None, None, PluginErrorInfo( error_code="INVALID_PLUGIN_JSON", message=f"plugin.json 格式错误: {str(e)}", stage=PluginErrorStage.STRUCTURE, detail=str(e), ) except Exception as e: return None, None, PluginErrorInfo( error_code="READ_PLUGIN_JSON_FAILED", message=f"读取 plugin.json 失败: {str(e)}", stage=PluginErrorStage.STRUCTURE, detail=str(e), ) # 验证元数据（允许部分字段缺失，标记 incomplete） try: metadata = PluginMetadata(**metadata_dict) except Exception as e: # 元数据不完整，但不阻断扫描 logger.warning(f"插件 {plugin_dir.name} 元数据不完整: {e}") metadata = None return metadata, metadata_dict, None def _detect_ui_entry( self, plugin_dir: Path, metadata_dict: dict ) -> Optional[PluginUIEntry]: """检测插件 UI 入口 Args: plugin_dir: 插件目录 metadata_dict: plugin.json 内容 Returns: PluginUIEntry 或 None """ # 检查是否有静态 UI 文件 frontend_dir = plugin_dir / "frontend" has_static_ui = frontend_dir.exists() and (frontend_dir / "index.html").exists() if has_static_ui: return PluginUIEntry( type=PluginUIType.STATIC, entry_path=f"/plugins/{plugin_dir.name}/ui/", ) # 检查是否有 schema UI 声明 ui_config = metadata_dict.get("ui", {}) if isinstance(ui_config, dict) and ui_config.get("type") == "schema": # 解析 schema UI 配置 try: schema_def = self._parse_schema_ui_config(ui_config, plugin_dir.name) submit_def = self._parse_submit_contract(ui_config, plugin_dir.name) return PluginUIEntry( type=PluginUIType.SCHEMA, schema=schema_def, submit=submit_def, ) except Exception as e: logger.warning(f"解析 schema UI 配置失败 {plugin_dir.name}: {e}") return PluginUIEntry( type=PluginUIType.SCHEMA, unavailable_reason=f"Schema UI 配置解析失败: {str(e)}", ) # 无 UI return PluginUIEntry(type=PluginUIType.NONE) def _parse_schema_ui_config(self, ui_config: dict, plugin_name: str) -> UISchemaDefinition: """解析 schema UI 配置 Args: ui_config: plugin.json 中的 ui 配置 plugin_name: 插件名称 Returns: UISchemaDefinition """ # 解析字段列表 fields = [] for field_config in ui_config.get("fields", []): if isinstance(field_config, dict): fields.append(UIFieldDefinition( name=field_config.get("name", ""), label=field_config.get("label", field_config.get("name", "")), type=field_config.get("type", "string"), required=field_config.get("required", False), default=field_config.get("default"), placeholder=field_config.get("placeholder"), help_text=field_config.get("help_text"), validation=field_config.get("validation"), )) return UISchemaDefinition( title=ui_config.get("title", plugin_name), description=ui_config.get("description"), fields=fields, result_schema=ui_config.get("result_schema"), ) def _parse_submit_contract(self, ui_config: dict, plugin_name: str) -> UISubmitContract: """解析提交契约 Args: ui_config: plugin.json 中的 ui 配置 plugin_name: 插件名称 Returns: UISubmitContract """ submit_config = ui_config.get("submit", {}) if isinstance(submit_config, dict): return UISubmitContract( method=submit_config.get("method", "POST"), path=submit_config.get("path", f"/plugins/{plugin_name}/api/run"), content_type=submit_config.get("content_type", "application/json"), success_path=submit_config.get("success_path"), error_path=submit_config.get("error_path"), ) # 默认提交契约 return UISubmitContract( method="POST", path=f"/plugins/{plugin_name}/api/run", content_type="application/json", ) def _build_governance( self, metadata_dict: dict, plugin_dir: Path ) -> PluginGovernanceInfo: """构建插件治理信息 Args: metadata_dict: plugin.json 内容 plugin_dir: 插件目录 Returns: PluginGovernanceInfo """ # 解析分类 category_str = metadata_dict.get("category", "other") try: category = PluginCategory(category_str) except ValueError: category = PluginCategory.OTHER # 解析能力标签 capabilities = metadata_dict.get("capabilities", []) # 检查是否有 MCP 工具 has_mcp = (plugin_dir / "mcp.py").exists() # 检查是否有 API 路由 has_api = (plugin_dir / "api.py").exists() # 检查是否有 UI frontend_dir = plugin_dir / "frontend" has_ui = frontend_dir.exists() and (frontend_dir / "index.html").exists() return PluginGovernanceInfo( category=category, capabilities=capabilities, has_mcp_tools=has_mcp, has_api_routes=has_api, has_ui=has_ui, ) def get_plugin_list(self) -> List[PluginListItem]: """获取插件列表 Returns: PluginListItem 列表 """ result = [] for plugin_name, info in self._plugins.items(): try: item = PluginListItem( metadata=info.get("metadata") or PluginMetadata( name=plugin_name, version="unknown", description="元数据不完整", author="unknown", ), status=info["status"], dependency_summary=info.get("dependency_check"), governance_summary=info.get("governance"), tool_summary=info.get("tool_summary", PluginToolSummary()), mcp_summary=info.get("mcp_summary", PluginMcpSummary()), ui_entry=info.get("ui_entry"), errors=info.get("errors", []), ) result.append(item) except Exception as e: logger.error(f"构建插件 {plugin_name} 列表项失败: {e}") # 单个插件失败不阻断列表 return result def get_enabled_plugins(self) -> List[PluginListItem]: """获取已启用的插件列表 Returns: 状态为 ENABLED 的 PluginListItem 列表 """ all_plugins = self.get_plugin_list() return [p for p in all_plugins if p.status == PluginStatus.ENABLED] def get_plugin_detail(self, plugin_name: str) -> Optional[PluginDetail]: """获取插件详情 Args: plugin_name: 插件名 Returns: PluginDetail 或 None """ info = self._plugins.get(plugin_name) if info is None: return None try: return PluginDetail( metadata=info.get("metadata") or PluginMetadata( name=plugin_name, version="unknown", description="元数据不完整", author="unknown", ), status=info["status"], install_path=info["install_path"], dependency_check=info.get("dependency_check"), governance=info.get("governance"), tools=[], # T-006 填充 mcp_tools=[], # T-007 填充 ui_entry=info.get("ui_entry"), errors=info.get("errors", []), ) except Exception as e: logger.error(f"构建插件 {plugin_name} 详情失败: {e}") return None def enable_plugin(self, plugin_name: str) -> PluginOperationResult: """启用插件 启用前检查结构、依赖和加载状态。 """ info = self._plugins.get(plugin_name) if info is None: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件不存在" ) current_status = info["status"] # 检查是否可以启用 if current_status == PluginStatus.ERROR: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件有结构或加载错误，无法启用" ) if current_status == PluginStatus.DEPENDENCY_ERROR: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件依赖缺失或版本不匹配，无法启用" ) if current_status == PluginStatus.ENABLED: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件已启用" ) # 尝试加载插件 plugin_dir = Path(info["install_path"]) try: if not plugin_loader.load_plugin_from_dir(plugin_dir): info["status"] = PluginStatus.ERROR info["errors"].append( PluginErrorInfo( error_code="LOAD_FAILED", message="插件加载失败", stage=PluginErrorStage.LOAD, ) ) return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件加载失败" ) except Exception as e: info["status"] = PluginStatus.ERROR info["errors"].append( PluginErrorInfo( error_code="LOAD_EXCEPTION", message=f"插件加载异常: {str(e)}", stage=PluginErrorStage.LOAD, detail=str(e), ) ) return PluginOperationResult.error_result( plugin_name=plugin_name, error=f"插件加载异常: {str(e)}" ) # 更新状态 info["status"] = PluginStatus.ENABLED self.state_store.set_enabled(plugin_name, True, updated_by="user") logger.info(f"插件 {plugin_name} 已启用") return PluginOperationResult.success_result( plugin_name=plugin_name, message=f"插件 {plugin_name} 已启用" ) def disable_plugin(self, plugin_name: str) -> PluginOperationResult: """禁用插件""" info = self._plugins.get(plugin_name) if info is None: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件不存在" ) current_status = info["status"] if current_status == PluginStatus.DISABLED: return PluginOperationResult.error_result( plugin_name=plugin_name, error="插件已禁用" ) # 更新状态 info["status"] = PluginStatus.DISABLED self.state_store.set_enabled(plugin_name, False, updated_by="user") logger.info(f"插件 {plugin_name} 已禁用") return PluginOperationResult.success_result( plugin_name=plugin_name, message=f"插件 {plugin_name} 已禁用" ) def get_plugin_status(self, plugin_name: str) -> Optional[PluginStatus]: """获取插件状态。 Phase 2.10：可选 readiness hook。 若插件实例实现了 check_readiness() 且返回 ok=False， 当基础状态为 ENABLED 时降级为 DEPENDENCY_ERROR。 hook 抛任何异常都忽略，绝不阻断其他插件。 """ info = self._plugins.get(plugin_name) if info is None: return None base_status = info["status"] # readiness hook 严格隔离 if base_status == PluginStatus.ENABLED: try: from app.plugins.loader import plugin_loader instance = plugin_loader.get_plugin_instance(plugin_name) if instance is not None and hasattr(instance, "check_readiness"): readiness = instance.check_readiness() if isinstance(readiness, dict) and readiness.get("ok") is False: return PluginStatus.DEPENDENCY_ERROR except Exception: # hook 异常绝不能影响插件状态查询 pass return base_status def is_plugin_available(self, plugin_name: str) -> bool: """检查插件是否可用（用于 MCP 门控）""" status = self.get_plugin_status(plugin_name) return status == PluginStatus.ENABLED def get_unavailable_reason(self, plugin_name: str) -> Optional[str]: """获取插件不可用原因（用于 MCP 错误反馈）""" info = self._plugins.get(plugin_name) if info is None: return "插件不存在" status = info["status"] if status == PluginStatus.ENABLED: return None if status == PluginStatus.DISABLED: return "插件已禁用" if status == PluginStatus.DEPENDENCY_ERROR: dep_check = info.get("dependency_check") if dep_check: return f"依赖问题: {dep_check.overall_status.value}" return "依赖检查失败" if status == PluginStatus.ERROR: errors = info.get("errors", []) if errors: return f"错误: {errors[0].message}" return "插件错误" if status == PluginStatus.INCOMPLETE: return "插件元数据不完整" return f"插件状态: {status.value}" def get_system_plugin_summary(self) -> Dict[str, Any]: """获取系统页插件摘要 Returns: 包含插件统计和状态的字典 """ total = len(self._plugins) enabled = sum( 1 for info in self._plugins.values() if info["status"] == PluginStatus.ENABLED ) disabled = sum( 1 for info in self._plugins.values() if info["status"] == PluginStatus.DISABLED ) error = sum( 1 for info in self._plugins.values() if info["status"] in ( PluginStatus.ERROR, PluginStatus.DEPENDENCY_ERROR, PluginStatus.INCOMPLETE, ) ) return { "total": total, "enabled": enabled, "disabled": disabled, "error": error, "plugin_directory": str(self.plugin_directory), } # 创建全局插件管理器实例（延迟初始化） plugin_manager: Optional[PluginManager] = None def get_plugin_manager() -> PluginManager: """获取插件管理器实例""" global plugin_manager if plugin_manager is None: plugin_manager = PluginManager() return plugin_manager