| """ |
| 插件依赖检查器。 |
| |
| 产品主流程只检查依赖是否满足，不执行安装操作。 |
| """ |
|
|
| 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: |
| |
| 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() |
|
|
