""" 插件依赖检查器。 产品主流程只检查依赖是否满足，不执行安装操作。 """ import logging from datetime import datetime from importlib import metadata from pathlib import Path from typing import List, Optional from packaging import version from packaging.requirements import InvalidRequirement, Requirement from packaging.specifiers import SpecifierSet from app.plugins.models import ( DependencyCheckResult, DependencyCheckStatus, DependencyItem, ) logger = logging.getLogger(__name__) class DependencyManager: """依赖检查器。""" def check_plugin_dependencies( self, plugin_path: Path, requirements: Optional[List[str]] = None, requirements_file: str = "requirements.txt", ) -> DependencyCheckResult: """检查插件依赖是否满足。""" plugin_name = plugin_path.name items: List[DependencyItem] = [] try: all_requirements = list(requirements or []) req_file = plugin_path / requirements_file if req_file.exists(): try: with open(req_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): all_requirements.append(line) except Exception as e: logger.warning(f"读取 requirements.txt 失败: {e}") all_requirements = list(set(all_requirements)) if not all_requirements: return DependencyCheckResult( plugin_name=plugin_name, overall_status=DependencyCheckStatus.SATISFIED, items=[], missing_count=0, mismatch_count=0, ) missing_count = 0 mismatch_count = 0 check_failed = False for req in all_requirements: try: item = self._check_single_dependency(req) items.append(item) if item.status == DependencyCheckStatus.MISSING: missing_count += 1 elif item.status == DependencyCheckStatus.VERSION_MISMATCH: mismatch_count += 1 elif item.status == DependencyCheckStatus.CHECK_FAILED: check_failed = True except Exception as e: logger.error(f"检查依赖 {req} 失败: {e}") items.append( DependencyItem( name=self._parse_requirement(req)[0], required=req, status=DependencyCheckStatus.CHECK_FAILED, message=f"检查异常: {str(e)}", ) ) check_failed = True if check_failed: overall_status = DependencyCheckStatus.CHECK_FAILED elif missing_count > 0: overall_status = DependencyCheckStatus.MISSING elif mismatch_count > 0: overall_status = DependencyCheckStatus.VERSION_MISMATCH else: overall_status = DependencyCheckStatus.SATISFIED return DependencyCheckResult( plugin_name=plugin_name, overall_status=overall_status, items=items, missing_count=missing_count, mismatch_count=mismatch_count, ) except Exception as e: logger.error(f"检查插件 {plugin_name} 依赖失败: {e}") return DependencyCheckResult( plugin_name=plugin_name, overall_status=DependencyCheckStatus.CHECK_FAILED, items=[], missing_count=0, mismatch_count=0, ) def _check_single_dependency(self, dependency: str) -> DependencyItem: """检查单个依赖是否满足。""" name, required_version, operator = self._parse_requirement(dependency) try: # 使用标准库读取发行包版本，避免依赖当前环境必须安装 pip 模块。 installed_version = metadata.version(name) except metadata.PackageNotFoundError: return DependencyItem( name=name, required=dependency, installed=None, status=DependencyCheckStatus.MISSING, message="包未安装", ) except Exception as e: logger.error(f"检查依赖异常 {dependency}: {e}") return DependencyItem( name=name, required=dependency, status=DependencyCheckStatus.CHECK_FAILED, message=f"检查异常: {str(e)}", ) if not required_version: return DependencyItem( name=name, required=dependency, installed=installed_version, status=DependencyCheckStatus.SATISFIED, message="已安装", ) if self._check_version(installed_version, required_version, operator): return DependencyItem( name=name, required=dependency, installed=installed_version, status=DependencyCheckStatus.SATISFIED, message="版本满足要求", ) return DependencyItem( name=name, required=dependency, installed=installed_version, status=DependencyCheckStatus.VERSION_MISMATCH, message=f"版本不满足要求: 需要 {operator}{required_version}，已安装 {installed_version}", ) def _parse_requirement(self, requirement: str) -> tuple[str, Optional[str], Optional[str]]: """解析依赖要求字符串，返回包名、版本约束和值。""" requirement = requirement.strip() try: parsed = Requirement(requirement) specifiers = list(parsed.specifier) if len(specifiers) == 1: spec = specifiers[0] return parsed.name, spec.version, spec.operator if specifiers: return parsed.name, str(parsed.specifier), "specifier" return parsed.name, None, None except InvalidRequirement: logger.warning(f"依赖声明格式不标准，使用兼容解析: {requirement}") for operator in (">=", "<=", "==", ">", "<"): if operator in requirement: name, version_str = requirement.split(operator, 1) return name.strip(), version_str.strip(), operator return requirement, None, None def _check_version(self, installed: str, required: str, operator: str) -> bool: """检查版本是否满足要求。""" try: if operator == "specifier": return version.parse(installed) in SpecifierSet(required) installed_ver = version.parse(installed) required_ver = version.parse(required) if operator == ">=": return installed_ver >= required_ver if operator == "<=": return installed_ver <= required_ver if operator == "==": return installed_ver == required_ver if operator == ">": return installed_ver > required_ver if operator == "<": return installed_ver < required_ver return True except Exception as e: logger.error(f"版本比较失败: {e}") return False def _is_dependency_satisfied(self, dependency: str) -> bool: """检查依赖是否已满足。""" result = self._check_single_dependency(dependency) return result.status == DependencyCheckStatus.SATISFIED def check_dependency_conflicts(self, dependencies: List[str]) -> tuple: """检查依赖冲突。""" try: conflicts = [] for dep in dependencies: item = self._check_single_dependency(dep) if item.status == DependencyCheckStatus.SATISFIED: conflicts.append(f"{item.name} 已安装版本 {item.installed}") elif item.status == DependencyCheckStatus.VERSION_MISMATCH: conflicts.append( f"{item.name} 版本不匹配: 需要 {item.required}，已安装 {item.installed}" ) return len(conflicts) == 0, conflicts except Exception as e: logger.error(f"检查依赖冲突失败: {e}") return False, [f"检查冲突异常: {str(e)}"] dependency_manager = DependencyManager()