MoviePilot/app/adapters/system/plugin/dependency.py

"""
插件 Python 依赖聚合和安装适配器。
"""

from __future__ import annotations

import asyncio
import json
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from importlib.metadata import PackageNotFoundError, distribution, distributions
from pathlib import Path
from typing import Any, Optional, Protocol
from urllib.parse import parse_qsl, unquote, urlsplit

from packaging.markers import default_environment
from packaging.requirements import Requirement
from packaging.specifiers import InvalidSpecifier, SpecifierSet
from packaging.version import InvalidVersion, Version

from app.adapters.system.plugin.health import PluginRuntimeHealth
from app.adapters.system.plugin.manifest import (
    PluginDependencyManifestError,
    load_dependency_file,
    load_dependency_manifest,
)
from app.runtime.execution import run_in_threadpool_to_completion
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting


@dataclass
class _RequirementGroup:
    """
    聚合同一依赖包和同一来源的附加功能与版本约束。

    :param name: 按 PEP 503 规范化后的包名
    :param url: 显式依赖来源地址；None 表示从包索引安装
    :param extras: 各插件要求启用的附加功能名集合，默认空集合
    :param specifiers: 待求交集的版本约束集合，默认空集合
    """

    name: str
    url: Optional[str]
    extras: set[str] = field(default_factory=set)
    specifiers: set[str] = field(default_factory=set)


@dataclass(frozen=True, slots=True)
class _DependencyInstallRequest:
    """
    插件依赖安装所需的清单和本地 wheel 搜索目录。

    :param manifest_paths: 本次需要合并的依赖清单路径列表
    :param wheels_dirs: 安装时可搜索本地 wheel 的目录列表
    """

    manifest_paths: list[Path]
    wheels_dirs: list[Path]


class PluginDependencyPackagePort(Protocol):
    """
    声明依赖聚合器所需的唯一 Python 包安装边界。
    """

    def install_packages_with_fallback(
        self,
        dependency_files: list[Path],
        find_links_dirs: list[Path],
    ) -> tuple[bool, str]:
        """
        按安装策略同步安装已经聚合并通过准入检查的依赖。

        :param dependency_files: 本次安装使用的依赖清单路径列表
        :param find_links_dirs: 本地 wheel 候选目录列表；为空时不附加目录
        :return: 是否安装成功与结果说明；失败不承诺回滚安装器已经完成的环境修改
        """

    async def async_install_packages_with_fallback(
        self,
        dependency_files: list[Path],
        find_links_dirs: list[Path],
    ) -> tuple[bool, str]:
        """
        按安装策略异步等待已通过准入检查的依赖安装。

        :param dependency_files: 本次安装使用的依赖清单路径列表
        :param find_links_dirs: 本地 wheel 候选目录列表；为空时不附加目录
        :return: 安装结束后的成功标志与结果说明；失败不承诺回滚已经完成的环境修改
        """


class PluginDependencyInstaller:
    """
    独立负责插件依赖扫描、约束合并和安装。
    """

    def __init__(
        self,
        packages: Optional[PluginDependencyPackagePort] = None,
        *,
        installed_plugins_provider: Optional[Callable[[], list[str]]] = None,
        enabled_plugins_provider: Optional[Callable[[], set[str]]] = None,
        plugin_dir: Optional[Path] = None,
    ) -> None:
        """
        保存安装清单和启用源码读取器；旧调用方未提供启用位时保留安装集合。
        """
        if packages is None:
            packages = PluginRuntimeHealth()
        self._packages = packages
        self._installed_plugins_provider = installed_plugins_provider or (lambda: [])
        self._enabled_plugins_provider = enabled_plugins_provider
        self._plugin_dir = plugin_dir or (
            Path(get_runtime_setting('ROOT_PATH')) / "app" / "plugins"
        )

    @staticmethod
    def _standardize(name: str) -> str:
        """
        按 PEP 503 兼容规则标准化依赖包名。
        """
        return (name or "").lower().replace("-", "_").replace(".", "_")

    @classmethod
    def _installed_packages(cls) -> dict[str, Version]:
        """
        读取当前 Python 环境中可解析版本的已安装包。
        """
        installed: dict[str, Version] = {}
        try:
            for distribution in distributions():
                name = distribution.metadata.get("Name")
                version = distribution.metadata.get("Version") or getattr(
                    distribution,
                    "version",
                    None,
                )
                if not name or not version:
                    continue
                package_name = cls._standardize(name)
                try:
                    parsed = Version(version)
                except InvalidVersion:
                    logger.debug(
                        f"无法解析已安装包 '{package_name}' 的版本：{version}"
                    )
                    continue
                if package_name not in installed or parsed > installed[package_name]:
                    installed[package_name] = parsed
        except Exception as err:
            logger.error(f"获取已安装的包时发生错误：{err}")
        return installed

    @classmethod
    def _installed_distribution(cls, package_name: str) -> Any | None:
        """
        读取一个包的元数据，用于校验 extras 和 direct URL 来源。
        """
        try:
            return distribution(package_name)
        except PackageNotFoundError:
            return None

    def _requirement_satisfied(
        self,
        requirement: Requirement,
        installed: dict[str, Version],
        *,
        seen: Optional[set[tuple[str, tuple[str, ...], Optional[str]]]] = None,
    ) -> bool:
        """
        同时校验版本、extras 及 direct URL，不把同名包误认为同一制品。
        """
        package_name = self._standardize(requirement.name)
        installed_version = installed.get(package_name)
        try:
            if installed_version is None or not SpecifierSet(
                requirement.specifier
            ).contains(installed_version, prereleases=True):
                return False
        except InvalidSpecifier as err:
            logger.error(f"依赖 {package_name} 约束无效：{err}")
            return False

        installed_distribution = self._installed_distribution(package_name)
        if installed_distribution is None:
            return False if requirement.extras or requirement.url else True

        if requirement.url and not self._direct_url_matches(
            installed_distribution, requirement.url
        ):
            return False

        requested_extras = {
            self._standardize_extra(extra) for extra in requirement.extras
        }
        if requested_extras:
            provided_extras = {
                self._standardize_extra(extra)
                for extra in installed_distribution.metadata.get_all(
                    "Provides-Extra"
                )
                or []
            }
            if not requested_extras.issubset(provided_extras):
                return False

        marker_key = (package_name, tuple(sorted(requested_extras)), requirement.url)
        if seen is None:
            seen = set()
        if marker_key in seen:
            return True
        seen.add(marker_key)

        for raw_dependency in installed_distribution.metadata.get_all(
            "Requires-Dist"
        ) or []:
            try:
                extra_dependency = Requirement(raw_dependency)
            except Exception as err:
                logger.debug(
                    f"无法解析已安装包 {package_name} 的依赖项 '{raw_dependency}'：{err}"
                )
                continue
            if not self._marker_matches_for_extras(
                extra_dependency, requested_extras
            ):
                continue
            if not self._requirement_satisfied(
                extra_dependency, installed, seen=seen
            ):
                return False
        return True

    @classmethod
    def _marker_matches_for_extras(
        cls, requirement: Requirement, extras: set[str]
    ) -> bool:
        """
        判断已安装发行版声明的可选依赖是否属于当前请求的 extra。
        """
        if requirement.marker is None:
            return True
        environment = default_environment()
        if "extra" in str(requirement.marker):
            return any(
                requirement.marker.evaluate({**environment, "extra": extra})
                for extra in extras
            )
        return bool(requirement.marker.evaluate(environment))

    @staticmethod
    def _standardize_extra(name: str) -> str:
        """
        按 PEP 685 兼容规则标准化 extra 名称。
        """
        return (name or "").lower().replace("-", "_").replace(".", "_")

    @classmethod
    def _direct_url_matches(cls, installed_distribution: Any, required_url: str) -> bool:
        """
        校验安装发行版记录的 PEP 610 URL 与清单来源一致。
        """
        try:
            payload = installed_distribution.read_text("direct_url.json")
            if not payload:
                return False
            metadata = json.loads(payload)
            direct_url = metadata.get("url")
            if not isinstance(direct_url, str):
                return False
            required = urlsplit(required_url)
            if "+" in required.scheme:
                required = cls._vcs_source_url(required, metadata.get("vcs_info"))
                if required is None:
                    return False
            if cls._canonical_direct_url(required._replace(fragment="").geturl()) != cls._canonical_direct_url(
                urlsplit(direct_url)._replace(fragment="").geturl()
            ):
                return False
            return cls._direct_url_fragment_matches(required.fragment, metadata)
        except (AttributeError, json.JSONDecodeError, TypeError, ValueError):
            return False

    @staticmethod
    def _vcs_source_url(required: Any, vcs_info: Any) -> Any:
        """
        将VCS清单URL还原为PEP610仓库地址，并检查声明的版本或固定提交。

        :param required: 已解析的清单URL
        :param vcs_info: 安装元数据中的VCS身份
        :return: 去除VCS前缀和版本的URL；身份不匹配时为None
        """
        vcs, scheme = required.scheme.split("+", 1)
        if not isinstance(vcs_info, dict) or vcs_info.get("vcs") != vcs:
            return None
        path, separator, revision = required.path.rpartition("@")
        if not separator:
            return required._replace(scheme=scheme)
        revision = unquote(revision)
        commit = vcs_info.get("commit_id")
        if re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", revision):
            matched = isinstance(commit, str) and commit.casefold() == revision.casefold()
        else:
            matched = revision in {vcs_info.get("requested_revision"), commit}
        return required._replace(scheme=scheme, path=path) if matched else None

    @staticmethod
    def _direct_url_fragment_matches(fragment: str, metadata: dict[str, Any]) -> bool:
        """
        从PEP610独立字段核对子目录与归档摘要，不把URL片段静默丢弃。

        :param fragment: 清单URL的片段字段
        :param metadata: 安装来源元数据
        :return: 全部来源约束都有对应证据时为True
        """
        fields = dict(parse_qsl(fragment, keep_blank_values=True))
        if fields.pop("subdirectory", None) != metadata.get("subdirectory"):
            return False
        archive = metadata.get("archive_info") or {}
        hashes = dict(archive.get("hashes") or {})
        if archive.get("hash"):
            algorithm, _, digest = archive["hash"].partition("=")
            hashes.setdefault(algorithm, digest)
        return all(
            isinstance(hashes.get(algorithm), str)
            and hashes[algorithm].casefold() == digest.casefold()
            for algorithm, digest in fields.items()
        )

    @staticmethod
    def _canonical_direct_url(value: str) -> tuple[str, str, str, str, str]:
        """
        规范化来源 URL，同时保留 fragment 中可能存在的校验信息。
        """
        parsed = urlsplit(value)
        netloc = parsed.netloc.rsplit("@", 1)[-1].lower()
        return (
            parsed.scheme.lower(),
            netloc,
            parsed.path.rstrip("/"),
            parsed.query,
            parsed.fragment,
        )

    @classmethod
    def _merge(cls, dependencies: list[Requirement]) -> list[Requirement]:
        """
        按包和安装来源合并 extras 与约束，保留完整安装目标。
        """
        groups: dict[tuple[str, Optional[str]], _RequirementGroup] = {}
        for requirement in dependencies:
            package_name = cls._standardize(requirement.name)
            key = (package_name, requirement.url)
            group = groups.setdefault(
                key,
                _RequirementGroup(name=package_name, url=requirement.url),
            )
            group.extras.update(requirement.extras)
            group.specifiers.add(str(requirement.specifier))

        merged: list[Requirement] = []
        for group in groups.values():
            spec_set = SpecifierSet()
            for specifier in group.specifiers:
                if not specifier:
                    continue
                try:
                    spec_set &= SpecifierSet(specifier)
                except InvalidSpecifier as err:
                    logger.error(f"发生版本约束冲突：{err}")
            target = group.name
            if group.extras:
                target += f"[{','.join(sorted(group.extras))}]"
            if group.url:
                target += f" @ {group.url}"
            elif spec_set:
                target += str(spec_set)
            merged.append(Requirement(target))
        return merged

    def _dependency_plugin_ids(self) -> set[str]:
        """
        只为已安装且至少有一个启用实例的物理插件恢复依赖。
        """
        installed = {
            plugin_id.lower()
            for plugin_id in self._installed_plugins_provider() or []
        }
        if self._enabled_plugins_provider is None:
            return installed
        return installed & {
            plugin_id.lower() for plugin_id in self._enabled_plugins_provider()
        }

    def _plugin_manifests(self) -> list[Any]:
        """
        返回需要装载的已安装插件的原始依赖清单。
        """
        manifests = []
        installed_plugins = self._dependency_plugin_ids()
        try:
            plugin_dirs = list(self._plugin_dir.iterdir())
        except (FileNotFoundError, OSError):
            return []
        for plugin_dir in sorted(plugin_dirs, key=lambda item: item.name):
            if not plugin_dir.is_dir():
                continue
            if plugin_dir.name not in installed_plugins:
                logger.debug(f"忽略插件 {plugin_dir.name} 的依赖")
                continue
            manifest = load_dependency_manifest(plugin_dir)
            if manifest is None:
                continue
            manifests.append(manifest)
        return manifests

    def _plugin_dependencies(self) -> list[Requirement]:
        """
        扫描启用插件的生效依赖清单并合并版本约束。
        """
        dependencies: list[Requirement] = []
        for manifest in self._plugin_manifests():
            for requirement in manifest.dependencies:
                if requirement.marker and not requirement.marker.evaluate():
                    continue
                dependencies.append(requirement)
        return self._merge(dependencies)

    def find_missing(self) -> list[str]:
        """
        返回当前插件集合缺失或不满足约束的依赖项。
        """
        try:
            required = self._plugin_dependencies()
            installed = self._installed_packages()
            missing = []
            for requirement in required:
                if not self._requirement_satisfied(requirement, installed):
                    missing.append(str(requirement))
            return missing
        except PluginDependencyManifestError:
            raise
        except Exception as err:
            logger.error(f"收集所有需要安装或更新的依赖项时发生错误：{err}")
            return []

    def classify_plugins(self) -> tuple[list[str], list[str], list[str]]:
        """
        按源码和依赖状态划分已安装插件。
        """
        ready: list[str] = []
        missing_dependencies: list[str] = []
        missing_source: list[str] = []
        installed_packages = self._installed_packages()

        for plugin_id in self._installed_plugins_provider() or []:
            plugin_dir = self._plugin_dir / plugin_id.lower()
            if not plugin_dir.is_dir():
                missing_source.append(plugin_id)
                continue
            try:
                manifest = load_dependency_manifest(plugin_dir)
                requirements = [] if manifest is None else [
                    requirement
                    for requirement in manifest.dependencies
                    if not requirement.marker or requirement.marker.evaluate()
                ]
            except PluginDependencyManifestError as error:
                logger.error(f"插件 {plugin_id} 依赖清单无效：{error}")
                missing_dependencies.append(plugin_id)
                continue
            if all(
                self._requirement_satisfied(requirement, installed_packages)
                for requirement in requirements
            ):
                ready.append(plugin_id)
            else:
                missing_dependencies.append(plugin_id)

        return ready, missing_dependencies, missing_source

    def _wheels_dirs(self) -> list[Path]:
        """
        收集启用插件附带的本地 wheels 目录。
        """
        result = []
        for plugin_id in sorted(self._dependency_plugin_ids()):
            wheels_dir = self._plugin_dir / plugin_id / "wheels"
            if wheels_dir.is_dir():
                result.append(wheels_dir)
        return list(dict.fromkeys(result))

    def _prepare_install_request(
        self, dependencies: list[str]
    ) -> tuple[Optional[_DependencyInstallRequest], Optional[tuple[bool, str]]]:
        """
        统一校验依赖安装请求，并在进入包 I/O 前构造规范输入。
        """
        if not dependencies:
            return None, (False, "没有传入需要安装的依赖项")
        try:
            manifest_paths = [manifest.path for manifest in self._plugin_manifests()]
            if not manifest_paths:
                return None, (False, "没有找到已安装插件的依赖清单")
            return (
                _DependencyInstallRequest(
                    manifest_paths=manifest_paths,
                    wheels_dirs=self._wheels_dirs(),
                ),
                None,
            )
        # 统一映射为公开安装结果
        except Exception as error:  # noqa: BLE001
            return None, self._dependency_install_failure(error)

    @staticmethod
    def _dependency_install_failure(error: Exception) -> tuple[bool, str]:
        """
        统一记录同步与异步包安装异常，并保留既有错误文本合同。
        """
        message = f"安装依赖项时发生错误：{error}"
        logger.error(message)
        return False, message

    def _fallback_plan(
        self, request: _DependencyInstallRequest
    ) -> tuple[list[Path], list[Path]]:
        """
        批量失败后保留已满足清单的约束，逐个恢复其余插件。

        成功恢复的清单也会加入后续请求；互斥版本因而不能通过轮流覆盖环境假装成功。
        始终传递原始清单，保留 uv sources、索引、direct URL 和 extras 的语义。
        """
        installed = self._installed_packages()
        accepted: list[Path] = []
        pending: list[Path] = []
        for path in request.manifest_paths:
            manifest = load_dependency_file(path)
            satisfied = all(
                self._requirement_satisfied(requirement, installed)
                for requirement in manifest.dependencies
                if not requirement.marker or requirement.marker.evaluate()
            )
            (accepted if satisfied else pending).append(path)
        return accepted, pending

    @staticmethod
    def _record_group_result(
        path: Path,
        result: tuple[bool, str],
        accepted: list[Path],
        failures: list[str],
    ) -> None:
        """
        让成功清单保护后续安装，并记录失败插件而不中断其它组。
        """
        success, message = result
        if success:
            accepted.append(path)
            return
        failure = f"插件 {path.parent.name} 依赖恢复失败：{message}"
        logger.warning(failure)
        failures.append(failure)

    @staticmethod
    def _fallback_result(failures: list[str]) -> tuple[bool, str]:
        """
        部分恢复仍返回失败，使启动继续按实际环境分类和激活就绪插件。
        """
        if failures:
            return False, "\n".join(failures)
        return True, "插件依赖已分组恢复"

    def install(self, dependencies: list[str]) -> tuple[bool, str]:
        """
        先统一安装；失败后逐组恢复，并保护已就绪插件的约束。
        """
        request, error_result = self._prepare_install_request(dependencies)
        if error_result is not None:
            return error_result
        assert request is not None
        try:
            result = self._packages.install_packages_with_fallback(
                request.manifest_paths,
                request.wheels_dirs,
            )
            if result[0] or len(request.manifest_paths) < 2:
                return result
            logger.warning("插件依赖批量安装失败，开始逐组恢复并保留已就绪插件约束")
            accepted, pending = self._fallback_plan(request)
            if not pending:
                return result
            failures: list[str] = []
            for path in pending:
                try:
                    group_result = self._packages.install_packages_with_fallback(
                        [*accepted, path], request.wheels_dirs
                    )
                # 单组失败不得中断其余组
                except Exception as error:  # noqa: BLE001
                    group_result = self._dependency_install_failure(error)
                self._record_group_result(path, group_result, accepted, failures)
            return self._fallback_result(failures)
        # 统一映射为公开安装结果
        except Exception as error:  # noqa: BLE001
            return self._dependency_install_failure(error)

    async def async_find_missing(self) -> list[str]:
        """
        在线程池中扫描缺失依赖，避免阻塞事件循环。
        """
        return await asyncio.to_thread(self.find_missing)

    async def async_install(self, dependencies: list[str]) -> tuple[bool, str]:
        """
        使用可取消子进程执行统一安装和受约束的逐组恢复。
        """
        prepared: tuple[
            Optional[_DependencyInstallRequest], Optional[tuple[bool, str]]
        ] = await run_in_threadpool_to_completion(
            self._prepare_install_request, dependencies
        )
        request, error_result = prepared
        if error_result is not None:
            return error_result
        assert request is not None
        try:
            result = await self._packages.async_install_packages_with_fallback(
                request.manifest_paths,
                request.wheels_dirs,
            )
            if result[0] or len(request.manifest_paths) < 2:
                return result
            logger.warning("插件依赖批量安装失败，开始逐组恢复并保留已就绪插件约束")
            plan: tuple[list[Path], list[Path]] = await run_in_threadpool_to_completion(
                self._fallback_plan, request
            )
            accepted, pending = plan
            if not pending:
                return result
            failures: list[str] = []
            for path in pending:
                try:
                    group_result = await self._packages.async_install_packages_with_fallback(
                        [*accepted, path], request.wheels_dirs
                    )
                # CancelledError 必须向上传播
                except Exception as error:  # noqa: BLE001
                    group_result = self._dependency_install_failure(error)
                self._record_group_result(path, group_result, accepted, failures)
            return self._fallback_result(failures)
        # 统一映射为公开安装结果
        except Exception as error:  # noqa: BLE001
            return self._dependency_install_failure(error)