import asyncio
from collections import deque
import importlib
import io
import json
import re
import shutil
import site
import stat
import sys
import tempfile
import threading
import time
import traceback
import uuid
import zipfile
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable, Sequence
from urllib.parse import parse_qs, quote, unquote, urlparse, urlsplit

import aiofiles
import aioshutil
import httpx2
from anyio import Path as AsyncPath
from packaging.markers import default_environment
from packaging.requirements import Requirement
from packaging.specifiers import SpecifierSet, InvalidSpecifier
from packaging.version import Version, InvalidVersion
from importlib.metadata import distributions
from requests import Response

from app.runtime.cache import cached, is_fresh
from app.runtime.settings import RuntimeSettingsCompat
from app.adapters.system.package import (
    PackageInstallRequest,
    build_package_install_strategies,
    build_project_sync_strategies,
    find_uv,
)
from app.adapters.system.plugin.manifest import (
    PluginDependencyManifestError,
    load_dependency_file,
    load_dependency_manifest,
)
from app.runtime.log import logger
from app.runtime.observability import observe_compat_facade
from app.runtime.execution import (
    await_task_to_terminal,
    run_in_threadpool_to_completion as _await_thread_operation,
)
from app.runtime.tasks import get_task_registry
from app.adapters.network.http import RequestUtils, AsyncRequestUtils
from app.foundation.singleton import WeakSingleton

from app.foundation.version import compare_version
from app.adapters.system.host import SystemUtils
from app.foundation.url import UrlUtils
from version import APP_VERSION

# 保留模块级可替换入口，代理默认读取组合根的最新 runtime 配置。
settings = RuntimeSettingsCompat()
PLUGIN_DIR = Path(settings.ROOT_PATH) / "app" / "plugins"
LOCAL_REPO_PREFIX = "local://"
PLUGIN_SYSTEM_VERSION_FIELD = "system_version"
PLUGIN_MARKET_WIKI_START = "<!-- plugin-market-repos:start -->"
PLUGIN_MARKET_WIKI_END = "<!-- plugin-market-repos:end -->"
PLUGIN_MARKET_WIKI_URL = (
    "https://raw.githubusercontent.com/jxxghp/MoviePilot-Wiki/main/plugin.md"
)
PLUGIN_MARKET_REPO_PATTERN = re.compile(
    r"https?://github\.com/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+(?:\.git)?/?",
    re.IGNORECASE,
)
# 主程序重大版本可扫描的旧索引；V3 临时默认兼容 V2，条目可用 v3:false 排除。
VERSION_BACKWARD_COMPATIBLE_FLAGS: Dict[str, List[str]] = {
    "v3": ["v2"],
}

InstalledPluginsProvider = Callable[[], List[str]]


def _empty_installed_plugins() -> List[str]:
    """组合根尚未注入配置读取器时返回空安装清单。"""
    return []


_installed_plugins_provider: InstalledPluginsProvider = _empty_installed_plugins

def configure_installed_plugins_provider(
    provider: InstalledPluginsProvider,
) -> None:
    """由启动组合层注入已安装插件读取器，避免市场适配器访问数据库。"""
    global _installed_plugins_provider
    _installed_plugins_provider = provider


def normalize_plugin_market_repo_url(repo_url: str) -> Optional[str]:
    """规范化插件仓库地址，便于跨来源合并去重。"""
    repo_url = (repo_url or "").strip().rstrip("/")
    if not repo_url:
        return None
    repo_url = repo_url.removesuffix(".git")
    parsed_url = urlparse(repo_url)
    if parsed_url.scheme not in {"http", "https"}:
        return None
    if (parsed_url.hostname or "").lower() != "github.com":
        return None
    paths = [item for item in parsed_url.path.split("/") if item]
    if len(paths) < 2:
        return None
    return f"https://github.com/{paths[0]}/{paths[1]}"


def split_plugin_market_repo_urls(value: Optional[str]) -> list[str]:
    """拆分插件市场仓库配置并保持原有顺序去重。"""
    repos: list[str] = []
    seen_repos = set()
    for item in re.split(r"[\n,，]+", value or ""):
        normalized_repo = normalize_plugin_market_repo_url(item)
        if not normalized_repo or normalized_repo.lower() in seen_repos:
            continue
        repos.append(normalized_repo)
        seen_repos.add(normalized_repo.lower())
    return repos


def extract_plugin_market_repos_from_wiki(
    markdown: str, require_markers: bool = False
) -> list[str]:
    """
    从 Wiki 插件文档中提取插件仓库地址。

    :param markdown: Wiki 插件文档 Markdown 内容
    :param require_markers: 是否要求文档包含唯一且有序的清单边界标记
    :return: 规范化并按文档顺序去重的插件仓库地址
    """
    content = markdown or ""
    start_count = content.count(PLUGIN_MARKET_WIKI_START)
    end_count = content.count(PLUGIN_MARKET_WIKI_END)
    start_index = content.find(PLUGIN_MARKET_WIKI_START)
    end_index = content.find(PLUGIN_MARKET_WIKI_END)
    if start_count == 1 and end_count == 1 and start_index < end_index:
        content = content[
            start_index + len(PLUGIN_MARKET_WIKI_START):end_index
        ]
    elif require_markers:
        raise ValueError("Wiki 插件仓库清单必须包含唯一且有序的开始和结束标记")

    repos: list[str] = []
    seen_repos = set()
    for item in PLUGIN_MARKET_REPO_PATTERN.findall(content):
        normalized_repo = normalize_plugin_market_repo_url(item)
        if not normalized_repo or normalized_repo.lower() in seen_repos:
            continue
        repos.append(normalized_repo)
        seen_repos.add(normalized_repo.lower())
    return repos


def merge_plugin_market_repos(
    local_repos: list[str], wiki_repos: list[str]
) -> list[str]:
    """合并本地与 Wiki 插件仓库地址，并保持来源中的既有顺序。"""
    merged_repos: list[str] = []
    seen_repos = set()
    for repo in local_repos + wiki_repos:
        normalized_repo = normalize_plugin_market_repo_url(repo)
        if not normalized_repo or normalized_repo.lower() in seen_repos:
            continue
        merged_repos.append(normalized_repo)
        seen_repos.add(normalized_repo.lower())
    return merged_repos


@observe_compat_facade("PluginHelper")
class PluginHelper(metaclass=WeakSingleton):
    """
    插件市场管理，下载安装插件到本地
    """

    _base_url = "https://raw.githubusercontent.com/{user}/{repo}/main/"
    # 串行化运行期依赖安装，避免多个包安装子进程和导入缓存刷新互相踩踏。
    _package_install_lock = threading.Lock()
    PLUGIN_DEPENDENCY_INSTALL_TIMEOUT = 300
    # 同仓库的并发 Release 请求共享任务；事件循环参与键控，避免热重载或测试循环切换后复用失效任务。
    _release_task_lock = threading.Lock()
    _release_tasks: Dict[Tuple[asyncio.AbstractEventLoop, str, bool], asyncio.Task] = {}
    # 这些包一旦被插件覆盖，最容易直接拖垮主程序启动，因此冲突提示需要单独高亮。
    _protected_runtime_packages = frozenset({
        "alembic",
        "fastapi",
        "pydantic",
        "pydantic_core",
        "pydantic_settings",
        "sqlalchemy",
        "starlette",
        "uvicorn",
    })
    _runtime_import_probe = (
        "import alembic, fastapi, pydantic, pydantic_core, pydantic_settings, "
        "sqlalchemy, starlette, uvicorn; from pydantic import BaseModel, Field"
    )

    @staticmethod
    def is_local_repo_url(repo_url: Optional[str]) -> bool:
        """
        判断是否为本地插件来源标识
        """
        return bool(repo_url and repo_url.startswith(LOCAL_REPO_PREFIX))

    @staticmethod
    def make_local_repo_url(pid: str, repo_path: Optional[Path] = None,
                            package_version: Optional[str] = None) -> str:
        """
        生成本地插件安装来源标识
        """
        repo_url = f"{LOCAL_REPO_PREFIX}{quote(pid, safe='')}"
        params = []
        if repo_path:
            params.append(f"path={quote(str(repo_path), safe='/:~')}")
        if package_version:
            params.append(f"version={quote(package_version, safe='')}")
        if params:
            repo_url = f"{repo_url}?{'&'.join(params)}"
        return repo_url

    @staticmethod
    def parse_local_repo_url(repo_url: str) -> Optional[str]:
        """
        从本地插件来源标识中解析插件ID
        """
        if not PluginHelper.is_local_repo_url(repo_url):
            return None
        try:
            parts = urlsplit(repo_url)
            pid = unquote(parts.netloc or parts.path.strip("/"))
        except Exception:
            pid = repo_url[len(LOCAL_REPO_PREFIX):].split("?", 1)[0].strip("/")
        return pid or None

    @staticmethod
    def parse_local_repo_path(repo_url: str) -> Optional[Path]:
        """
        从本地插件来源标识中解析仓库路径
        """
        if not PluginHelper.is_local_repo_url(repo_url):
            return None
        try:
            values = parse_qs(urlsplit(repo_url).query).get("path")
            if not values:
                return None
            path = Path(values[0]).expanduser()
            if not path.is_absolute():
                path = settings.ROOT_PATH / path
            return path.resolve()
        except Exception:
            return None

    @staticmethod
    def parse_local_repo_package_version(repo_url: str) -> Optional[str]:
        """
        从本地插件来源标识中解析 package 版本
        """
        if not PluginHelper.is_local_repo_url(repo_url):
            return None
        try:
            values = parse_qs(urlsplit(repo_url).query).get("version")
            if not values:
                return None
            return values[0]
        except Exception:
            return None

    @staticmethod
    def get_current_system_version() -> Optional[Version]:
        """
        解析当前主程序版本，供插件 package 中的系统版本范围匹配使用。
        """
        try:
            return Version(str(APP_VERSION))
        except InvalidVersion:
            logger.error(f"当前主程序版本号无法解析：{APP_VERSION}")
            return None

    @classmethod
    def get_compatible_version_flags(cls) -> List[str]:
        """
        返回当前主程序版本可兼容的全部版本标识，包含自身及向后兼容的低版本，按优先级降序。
        未启用 VERSION_FLAG（v1）时返回空列表，表示仅使用 package.json 基础索引。
        """
        flags: List[str] = []
        if settings.VERSION_FLAG:
            flags.append(settings.VERSION_FLAG)
            flags.extend(VERSION_BACKWARD_COMPATIBLE_FLAGS.get(settings.VERSION_FLAG, []))
        return flags

    @classmethod
    def is_plugin_info_compatible(cls, plugin_info: Optional[dict]) -> bool:
        """
        判断 package.json 中的插件元数据是否兼容当前主程序版本。

        默认索引需要声明当前版本；V3 临时兼容已声明 V2 的共享实现，
        但显式 ``v3: false`` 始终优先拒绝。
        """
        if not isinstance(plugin_info, dict):
            return False
        if not settings.VERSION_FLAG:
            return True
        current_flag = settings.VERSION_FLAG
        if plugin_info.get(current_flag) is False:
            return False
        if plugin_info.get(current_flag) is True:
            return True
        return any(
            plugin_info.get(flag) is True
            for flag in VERSION_BACKWARD_COMPATIBLE_FLAGS.get(
                current_flag, []
            )
        )

    @classmethod
    def is_package_plugin_compatible(
            cls,
            plugin_info: Optional[dict],
            package_version: Optional[str],
    ) -> bool:
        """
        判断指定索引中的插件条目能否在当前主程序版本使用。

        当前代专用索引直接兼容。V3 临时默认兼容 V2 专用索引，
        除非条目显式声明 ``v3: false``；默认索引仍需先声明 ``v2: true``。
        """
        if not isinstance(plugin_info, dict):
            return False
        current_flag = settings.VERSION_FLAG
        if not current_flag:
            return not package_version
        if package_version == current_flag:
            return True
        if package_version in VERSION_BACKWARD_COMPATIBLE_FLAGS.get(
                current_flag, []
        ):
            return plugin_info.get(current_flag) is not False
        if not package_version:
            return cls.is_plugin_info_compatible(plugin_info)
        return False

    @classmethod
    def check_plugin_system_version(cls, plugin_info: Optional[dict]) -> Tuple[bool, str]:
        """
        检查插件 package 元数据中的主系统版本范围是否满足当前 MoviePilot 版本。
        """
        if not isinstance(plugin_info, dict):
            return True, ""

        raw_specifier = plugin_info.get(PLUGIN_SYSTEM_VERSION_FIELD)
        if raw_specifier is None or raw_specifier == "":
            return True, ""
        if not isinstance(raw_specifier, str):
            return False, (
                f"插件限定的系统版本范围 {PLUGIN_SYSTEM_VERSION_FIELD} 必须是字符串，"
                f"请使用 PEP 440 版本范围格式，例如 >=2.12.0,<3"
            )

        system_version = cls.get_current_system_version()
        if system_version is None:
            return False, f"当前 MoviePilot 版本 {APP_VERSION} 无法解析，已拒绝安装带版本限制的插件"

        try:
            specifier_set = SpecifierSet(raw_specifier)
        except InvalidSpecifier:
            return False, (
                f"插件限定的系统版本范围格式不正确：{raw_specifier}，"
                f"请使用 PEP 440 版本范围格式，例如 >=2.12.0,<3"
            )

        if specifier_set.contains(system_version, prereleases=True):
            return True, ""

        return False, (
            f"插件要求 MoviePilot 版本 {raw_specifier}，当前版本 {APP_VERSION} 不满足，已拒绝安装"
        )

    @classmethod
    def annotate_plugin_system_version(cls, plugin_info: dict) -> dict:
        """
        为插件 package 元数据补充系统版本兼容状态，便于市场展示和安装流程复用。
        """
        if not isinstance(plugin_info, dict):
            return plugin_info

        compatible, message = cls.check_plugin_system_version(plugin_info)
        plugin_info["system_version_compatible"] = compatible
        plugin_info["system_version_message"] = message
        return plugin_info

    @staticmethod
    def get_local_repo_paths() -> List[Path]:
        """
        获取本地插件仓库目录列表
        """
        if not settings.PLUGIN_LOCAL_REPO_PATHS:
            return []
        paths = []
        for item in settings.PLUGIN_LOCAL_REPO_PATHS.split(","):
            local_repo_path = item.strip()
            if not local_repo_path:
                continue
            path = Path(local_repo_path).expanduser()
            if not path.is_absolute():
                path = settings.ROOT_PATH / path
            paths.append(path.resolve())
        return paths

    @staticmethod
    def __get_local_package(repo_path: Path, package_version: Optional[str] = None) -> Optional[Dict[str, dict]]:
        """
        从本地插件仓库读取 package.json 或 package.{version}.json
        """
        package_file = repo_path / (
            f"package.{package_version}.json" if package_version else "package.json"
        )
        if not package_file.exists():
            return {}
        try:
            content = package_file.read_text(encoding="utf-8", errors="replace")
            payload = json.loads(content)
        except Exception as e:
            logger.warn(f"读取本地插件包 {package_file} 失败：{e}")
            return None
        if not isinstance(payload, dict):
            logger.warn(f"本地插件包 {package_file} 格式不正确")
            return None
        return payload

    @staticmethod
    def __get_local_plugin_dir(repo_path: Path, pid: str, package_version: Optional[str]) -> Path:
        """按插件包版本计算本地插件源码目录。"""
        plugin_root = f"plugins.{package_version}" if package_version else "plugins"
        return repo_path / plugin_root / pid.lower()

    def get_local_plugin_candidates(self) -> Dict[str, dict]:
        """
        扫描本地插件仓库，按插件ID保留版本号最高的候选
        """
        candidates: Dict[str, dict] = {}
        for repo_order, repo_path in enumerate(self.get_local_repo_paths()):
            if not repo_path.exists() or not repo_path.is_dir():
                logger.warn(f"本地插件仓库目录不存在或不可读：{repo_path}")
                continue

            package_candidates = []
            if settings.VERSION_FLAG:
                package_candidates.append((settings.VERSION_FLAG, self.__get_local_package(repo_path,
                                                                                           settings.VERSION_FLAG)))
                # 向后兼容：补充扫描更低版本的 package 文件，便于本地仓库复用历史版本插件。
                for backward_flag in VERSION_BACKWARD_COMPATIBLE_FLAGS.get(settings.VERSION_FLAG, []):
                    package_candidates.append((backward_flag, self.__get_local_package(repo_path, backward_flag)))
            package_candidates.append(("", self.__get_local_package(repo_path)))

            for package_version, local_plugins in package_candidates:
                if local_plugins is None:
                    continue
                for pid, plugin_info in local_plugins.items():
                    if not isinstance(plugin_info, dict):
                        continue
                    if not self.is_package_plugin_compatible(
                            plugin_info, package_version
                    ):
                        continue

                    plugin_dir = self.__get_local_plugin_dir(repo_path, pid, package_version)
                    if not plugin_dir.is_dir():
                        logger.debug(f"跳过本地插件 {pid}：插件目录不存在 {plugin_dir}")
                        continue

                    candidate = plugin_info.copy()
                    candidate["id"] = pid
                    candidate["package_version"] = package_version
                    candidate["repo_order"] = repo_order
                    candidate["repo_path"] = repo_path
                    candidate["path"] = plugin_dir
                    self.annotate_plugin_system_version(candidate)
                    candidate_version = str(candidate.get("version") or "0")

                    existing = candidates.get(pid)
                    if not existing:
                        candidates[pid] = candidate
                        continue

                    existing_version = str(existing.get("version") or "0")
                    if compare_version(candidate_version, ">", existing_version):
                        candidates[pid] = candidate
                    elif (
                        candidate_version == existing_version
                        and repo_order < int(existing.get("repo_order", repo_order))
                    ):
                        logger.info(f"本地插件 {pid} 存在同版本来源，使用靠前目录：{repo_path}")
                        candidates[pid] = candidate

        return candidates

    def get_local_plugin_candidate(self, pid: str, package_version: Optional[str] = None,
                                   repo_path: Optional[Path] = None,
                                   strict_compat: bool = True,
                                   strict_system_version: bool = True) -> Optional[dict]:
        """
        获取指定插件ID的本地插件候选
        :param strict_system_version: 是否将主系统版本范围不匹配视为不可用候选
        """
        if not pid:
            return None
        if package_version is not None or repo_path is not None:
            repo_paths = [repo_path.resolve()] if repo_path else self.get_local_repo_paths()
            package_versions = [package_version] if package_version is not None else []
            if package_version is None:
                if settings.VERSION_FLAG:
                    package_versions.append(settings.VERSION_FLAG)
                    package_versions.extend(VERSION_BACKWARD_COMPATIBLE_FLAGS.get(settings.VERSION_FLAG, []))
                package_versions.append("")
            selected_candidate = None
            for repo_order, local_repo_path in enumerate(self.get_local_repo_paths()):
                if local_repo_path not in repo_paths:
                    continue
                for current_package_version in package_versions:
                    local_plugins = self.__get_local_package(local_repo_path, current_package_version or "")
                    if not local_plugins:
                        continue
                    for candidate_pid, plugin_info in local_plugins.items():
                        if candidate_pid.lower() != pid.lower() or not isinstance(plugin_info, dict):
                            continue
                        is_compatible = self.is_package_plugin_compatible(
                            plugin_info,
                            current_package_version or "",
                        )
                        if not is_compatible and strict_compat:
                            continue
                        plugin_dir = self.__get_local_plugin_dir(local_repo_path, candidate_pid,
                                                                 current_package_version or "")
                        if not plugin_dir.is_dir():
                            continue
                        candidate = plugin_info.copy()
                        candidate["id"] = candidate_pid
                        candidate["package_version"] = current_package_version or ""
                        candidate["repo_order"] = repo_order
                        candidate["repo_path"] = local_repo_path
                        candidate["path"] = plugin_dir
                        if not is_compatible:
                            candidate["compatible"] = False
                            candidate["skip_reason"] = (
                                f"插件索引条目不兼容 {settings.VERSION_FLAG}"
                            )
                        self.annotate_plugin_system_version(candidate)
                        if strict_system_version and candidate.get("system_version_compatible") is False:
                            candidate["compatible"] = False
                            candidate["skip_reason"] = candidate.get("system_version_message")
                        if package_version is not None:
                            return candidate
                        if not selected_candidate:
                            selected_candidate = candidate
                            continue
                        selected_version = str(selected_candidate.get("version") or "0")
                        candidate_version = str(candidate.get("version") or "0")
                        if compare_version(candidate_version, ">", selected_version):
                            selected_candidate = candidate
            return selected_candidate

        candidates = self.get_local_plugin_candidates()
        for candidate_pid, candidate in candidates.items():
            if candidate_pid.lower() == pid.lower():
                if strict_system_version and candidate.get("system_version_compatible") is False:
                    candidate = candidate.copy()
                    candidate["compatible"] = False
                    candidate["skip_reason"] = candidate.get("system_version_message")
                return candidate
        return None

    @staticmethod
    def __append_cache_buster(url: str) -> str:
        """
        强制刷新插件库索引时追加时间戳，绕过 GitHub 镜像或中间代理的缓存。
        """
        if not is_fresh():
            return url

        parts = urlsplit(url)
        refresh_param = f"_refresh={time.time_ns()}"
        query = f"{parts.query}&{refresh_param}" if parts.query else refresh_param
        return parts._replace(query=query).geturl()

    @staticmethod
    def __parse_plugin_index_response(content: str) -> Optional[Dict[str, dict]]:
        """
        解析插件索引响应，仅缓存成功解析出的字典结果。
        """
        try:
            payload = json.loads(content)
        except json.JSONDecodeError:
            if "404: Not Found" not in content:
                logger.warn(f"插件包数据解析失败：{content}")
            return None

        if not isinstance(payload, dict):
            logger.warn(f"插件包数据格式不正确，期望 dict，实际为 {type(payload).__name__}")
            return None

        return payload

    @staticmethod
    def __build_plugin_release_item(pid: str, release_info: dict) -> Optional[dict]:
        """
        从 GitHub release 响应中提取可安装版本，仅接受规范 tag 与同名 zip 资产。
        """
        if not isinstance(release_info, dict):
            return None

        tag_name = release_info.get("tag_name")
        if not isinstance(tag_name, str):
            return None

        tag_prefix = f"{pid}_v"
        if not tag_name.startswith(tag_prefix):
            return None

        version = tag_name[len(tag_prefix):]
        if not version:
            return None

        asset_name = f"{tag_name.lower()}.zip"
        assets = release_info.get("assets") or []
        if not any(isinstance(asset, dict) and asset.get("name") == asset_name for asset in assets):
            return None

        return {
            "version": version,
            "tag_name": tag_name,
            "name": release_info.get("name") or tag_name,
            "published_at": release_info.get("published_at"),
            "body": release_info.get("body") or "",
            "asset_name": asset_name,
        }

    @staticmethod
    def __parse_plugin_release_response(pid: str, payload) -> List[dict]:
        """
        解析 GitHub release 列表，过滤出当前插件可直接安装的 release 资产。
        """
        if not isinstance(payload, list):
            return []

        releases = []
        for release_info in payload:
            item = PluginHelper.__build_plugin_release_item(pid, release_info)
            if item:
                releases.append(item)
        return releases

    @staticmethod
    def __normalize_plugin_release_response(payload) -> List[dict]:
        """仅保留版本展示和资产匹配所需字段，控制仓库级缓存体积。"""
        if not isinstance(payload, list):
            return []
        return [
            {
                "tag_name": release_info.get("tag_name"),
                "name": release_info.get("name"),
                "published_at": release_info.get("published_at"),
                "body": release_info.get("body"),
                "assets": [
                    {"name": asset.get("name")}
                    for asset in release_info.get("assets") or []
                    if isinstance(asset, dict)
                ],
            }
            for release_info in payload
            if isinstance(release_info, dict)
        ]

    @cached(maxsize=128, ttl=1800)
    def get_plugins(self, repo_url: str,
                    package_version: Optional[str] = None) -> Optional[Dict[str, dict]]:
        """
        获取Github所有最新插件列表
        :param repo_url: Github仓库地址
        :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本
        """
        if not repo_url:
            return None

        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return None

        raw_url = self._base_url.format(user=user, repo=repo)
        package_url = f"{raw_url}package.{package_version}.json" if package_version else f"{raw_url}package.json"
        package_url = self.__append_cache_buster(package_url)

        res = self.__request_with_fallback(package_url, headers=settings.REPO_GITHUB_HEADERS(repo=f"{user}/{repo}"))
        if res is None:
            return None
        if res.status_code == 404:
            return {}
        if res.status_code != 200:
            return None
        return self.__parse_plugin_index_response(res.text)

    @cached(maxsize=32, ttl=1800, shared_key="get_plugin_repo_releases")
    def _get_plugin_repo_releases(self, repo_url: str) -> Optional[List[dict]]:
        """
        按仓库获取 GitHub Release 原始分页数据，供仓库内所有插件共享。
        """
        if not repo_url:
            return []

        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return []

        user_repo = f"{user}/{repo}"
        releases = []
        for page in range(1, 11):
            release_api = f"https://api.github.com/repos/{user_repo}/releases?per_page=100&page={page}"
            release_api = self.__append_cache_buster(release_api)
            res = self.__request_with_fallback(
                release_api,
                headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
                timeout=30,
                is_api=True,
            )
            if res is None or res.status_code != 200:
                return None

            try:
                payload = res.json()
                if not payload:
                    break
                if not isinstance(payload, list):
                    return None
                releases.extend(self.__normalize_plugin_release_response(payload))
                if len(payload) < 100:
                    break
            except Exception as e:
                logger.error(f"解析插件仓库 {repo_url} Release 列表失败：{e}")
                return None
        return releases

    def get_plugin_release_versions(self, pid: str, repo_url: str) -> List[dict]:
        """
        获取插件可安装的 GitHub Release 版本列表。

        GitHub 分页结果按仓库缓存，插件 ID 只参与本地过滤，避免同仓库重复分页。
        """
        if not pid or not repo_url:
            return []
        return self.__parse_plugin_release_response(pid, self._get_plugin_repo_releases(repo_url.rstrip("/")))

    @staticmethod
    def __has_installable_release_version(release_items: List[dict], release_version: str) -> bool:
        """
        指定版本必须来自已解析出的可安装 Release 列表，避免直接拼接任意 tag。
        """
        return any(item.get("version") == release_version for item in release_items)

    def get_plugin_package_version(self, pid: str, repo_url: str,
                                   package_version: Optional[str] = None) -> Optional[str]:
        """
        检查并获取指定插件的可用版本，支持多版本优先级加载和版本兼容性检测
        1. 如果未指定版本，则使用系统配置的默认版本（通过 settings.VERSION_FLAG 设置）
        2. 优先检查指定版本的插件（如 `package.v2.json`）
        3. 检查更低版本的 package 文件，并应用版本兼容标志
        4. 检查 `package.json` 文件，并应用共享实现兼容标志
        5. 如果插件不存在或不兼容指定版本，返回 `None`
        :param pid: 插件 ID，用于在插件列表中查找
        :param repo_url: 插件仓库的 URL，指定用于获取插件信息的 GitHub 仓库地址
        :param package_version: 首选插件版本 (如 "v2", "v3")，如不指定则默认使用系统配置的版本
        :return: 返回可用的插件版本号 (如 "v2"，如果指定版本不可用则返回空字符串表示 v1)，如果插件不可用则返回 None
        """
        # 如果没有指定版本，则使用当前系统配置的版本（如 "v3"）
        if not package_version:
            package_version = settings.VERSION_FLAG

        # 优先检查指定索引；即使显式指定 V2，也必须尊重 v3:false 排除标志。
        plugin = (self.get_plugins(repo_url, package_version) or {}).get(pid)
        if plugin and self.is_package_plugin_compatible(
                plugin, package_version
        ):
            return package_version

        # V3 临时默认接纳 V2 专用索引，v3:false 的 V3 专用副本旧条目除外。
        for backward_flag in VERSION_BACKWARD_COMPATIBLE_FLAGS.get(package_version, []):
            plugin = (self.get_plugins(repo_url, backward_flag) or {}).get(pid)
            if plugin and self.is_package_plugin_compatible(
                    plugin, backward_flag
            ):
                return backward_flag

        # 默认索引只接纳声明 v2:true 或当前版本兼容的共享实现。
        plugin = (self.get_plugins(repo_url) or {}).get(pid, None)
        if plugin and self.is_package_plugin_compatible(plugin, ""):
            return ""

        # 如果所有版本都不存在或插件不兼容，返回 None，表示插件不可用
        return None

    @staticmethod
    def get_repo_info(repo_url: str) -> Tuple[Optional[str], Optional[str]]:
        """
        获取GitHub仓库信息
        """
        if not repo_url:
            return None, None
        if not repo_url.endswith("/"):
            repo_url += "/"
        if repo_url.count("/") < 6:
            repo_url = f"{repo_url}main/"
        try:
            user, repo = repo_url.split("/")[-4:-2]
        except Exception as e:
            logger.error(f"解析GitHub仓库地址失败：{str(e)} - {traceback.format_exc()}")
            return None, None
        return user, repo

    def install(self, pid: str, repo_url: str, package_version: Optional[str] = None,
                release_version: Optional[str] = None, force_install: bool = False) \
            -> Tuple[bool, str]:
        """
        安装插件，包括版本检查、内容准备、生效清单依赖安装和失败恢复。
        :param pid: 插件 ID
        :param repo_url: 插件仓库地址
        :param package_version: 首选插件版本 (如 "v2", "v3")，如不指定则默认使用系统配置的版本
        :param release_version: 指定安装的 release 资产版本；未指定时安装当前索引版本
        :param force_install: 是否强制安装插件，默认不启用，启用时不进行备份和恢复操作
        :return: (是否成功, 错误信息)
        """
        if self.is_local_repo_url(repo_url):
            return self.install_local(pid=pid, repo_url=repo_url, force_install=force_install)

        if SystemUtils.is_frozen():
            return False, "可执行文件模式下，只能安装本地插件"

        # 验证参数
        if not pid or not repo_url:
            return False, "参数错误"

        # 从 GitHub 的 repo_url 获取用户和项目名
        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return False, "不支持的插件仓库地址格式"

        user_repo = f"{user}/{repo}"

        if not package_version:
            package_version = settings.VERSION_FLAG

        # 1. 优先检查指定版本的插件
        package_version = self.get_plugin_package_version(pid, repo_url, package_version)
        # 如果 package_version 为None，说明没有找到匹配的插件
        if package_version is None:
            msg = f"{pid} 没有找到适用于当前版本的插件"
            logger.debug(msg)
            return False, msg
        # package_version 为空，表示从 package.json 中找到插件
        elif package_version == "":
            logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件")
        else:
            logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件")

        # 2. 决定安装方式（release 或文件列表）并执行统一安装流程。
        meta = self.__get_plugin_meta(pid, repo_url, package_version)
        # 是否使用 Release 打包。Release 缺失或资产不可用时仍保留文件列表兜底，
        # 避免索引先发布、Actions 打包滞后导致插件短时间无法安装。
        is_release = meta.get("release")
        # 插件版本号
        plugin_version = meta.get("version")
        if release_version:
            if not is_release:
                return False, f"{pid} 未声明 Release 安装，无法安装指定版本"
            if not self.__has_installable_release_version(
                    self.get_plugin_release_versions(pid, repo_url), release_version
            ):
                return False, f"{pid} 未找到可安装的 Release 版本：{release_version}"
            if release_version == plugin_version:
                compatible, message = self.check_plugin_system_version(meta)
                if not compatible:
                    logger.debug(f"{pid} 插件系统版本兼容性检查失败：{message}")
                    return False, message
            release_tag = f"{pid}_v{release_version}"

            def prepare_selected_release() -> Tuple[bool, str]:
                return self.__install_from_release(pid, user_repo, release_tag)

            return self.__install_flow_sync(pid, force_install, prepare_selected_release, repo_url)

        compatible, message = self.check_plugin_system_version(meta)
        if not compatible:
            logger.debug(f"{pid} 插件系统版本兼容性检查失败：{message}")
            return False, message
        if is_release:
            # 使用 插件ID_插件版本号 作为 Release tag
            if not plugin_version:
                return False, f"未在插件清单中找到 {pid} 的版本号，无法进行 Release 安装"
            # 拼接 release_tag
            release_tag = f"{pid}_v{plugin_version}"

            # 使用 release 进行安装
            def prepare_release() -> Tuple[bool, str]:
                ok, msg = self.__install_from_release(pid, user_repo, release_tag)
                if ok:
                    return True, msg
                logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}")
                self.__remove_old_plugin(pid)
                return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version)

            return self.__install_flow_sync(pid, force_install, prepare_release, repo_url)
        else:
            # 未声明 release 打包的插件继续使用文件列表方式安装。
            def prepare_filelist() -> Tuple[bool, str]:
                return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version)

            return self.__install_flow_sync(pid, force_install, prepare_filelist, repo_url)

    def install_local(self, pid: str, repo_url: str = "", force_install: bool = False) -> Tuple[bool, str]:
        """
        从本地插件仓库目录安装插件
        """
        local_pid = self.parse_local_repo_url(repo_url) if repo_url else pid
        if not local_pid or local_pid.lower() != pid.lower():
            return False, "本地插件来源与插件ID不匹配"

        repo_path = self.parse_local_repo_path(repo_url) if repo_url else None
        package_version = self.parse_local_repo_package_version(repo_url) if repo_url else None
        candidate = self.get_local_plugin_candidate(
            pid,
            package_version=package_version,
            repo_path=repo_path
        )
        if not candidate:
            return False, f"未找到本地插件：{pid}"
        compatible, message = self.check_plugin_system_version(candidate)
        if not compatible:
            logger.debug(f"{pid} 本地插件系统版本兼容性检查失败：{message}")
            return False, message

        source_dir = Path(candidate.get("path"))
        dest_dir = PLUGIN_DIR / pid.lower()
        try:
            if source_dir.resolve() == dest_dir.resolve():
                return False, "本地插件来源不能与运行目录相同"
        except Exception:
            return False, "本地插件来源路径无效"

        def prepare_local() -> Tuple[bool, str]:
            try:
                shutil.copytree(
                    source_dir,
                    dest_dir,
                    dirs_exist_ok=True,
                    ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store", "node_modules")
                )
                return True, ""
            except Exception as e:
                logger.error(f"复制本地插件 {pid} 失败：{e}")
                return False, f"复制本地插件失败：{e}"

        return self.__install_flow_sync(
            pid=pid,
            force_install=force_install,
            prepare_content=prepare_local,
            repo_url=repo_url or self.make_local_repo_url(
                pid,
                candidate.get("repo_path"),
                candidate.get("package_version")
            )
        )

    def __get_file_list(self, pid: str, user_repo: str, package_version: Optional[str] = None) -> \
            Tuple[Optional[list], Optional[str]]:
        """
        获取插件的文件列表
        :param pid: 插件 ID
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (文件列表, 错误信息)
        """
        file_api = f"https://api.github.com/repos/{user_repo}/contents/plugins"
        # 如果 package_version 存在（如 "v2"），则加上版本号
        if package_version:
            file_api += f".{package_version}"
        file_api += f"/{pid.lower()}"

        res = self.__request_with_fallback(file_api,
                                           headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
                                           is_api=True,
                                           timeout=30)
        if res is None:
            return None, "连接仓库失败"
        elif res.status_code != 200:
            return None, f"连接仓库失败：{res.status_code} - " \
                         f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.reason}"

        try:
            ret = res.json()
            if isinstance(ret, list) and len(ret) > 0 and "message" not in ret[0]:
                return ret, ""
            else:
                return None, "插件在仓库中不存在或返回数据格式不正确"
        except Exception as e:
            logger.error(f"插件数据解析失败：{e}")
            return None, "插件数据解析失败"

    def __download_files(self, pid: str, file_list: List[dict], user_repo: str,
                         package_version: Optional[str] = None) -> Tuple[bool, str]:
        """
        下载插件文件
        :param pid: 插件 ID
        :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接）
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (是否成功, 错误信息)
        """
        if not file_list:
            return False, "文件列表为空"

        # 使用栈结构来替代递归调用，避免递归深度过大问题
        stack = [(pid, file_list)]

        while stack:
            current_pid, current_file_list = stack.pop()

            for item in current_file_list:
                if item.get("download_url"):
                    logger.debug(f"正在下载文件：{item.get('path')}")
                    res = self.__request_with_fallback(item.get('download_url'),
                                                       headers=settings.REPO_GITHUB_HEADERS(repo=user_repo))
                    if not res:
                        return False, f"文件 {item.get('path')} 下载失败！"
                    elif res.status_code != 200:
                        return False, f"下载文件 {item.get('path')} 失败：{res.status_code}"

                    # 确保文件路径不包含版本号（如 v2、v3），如果有 package_version，移除路径中的版本号
                    relative_path = item.get("path")
                    if package_version:
                        relative_path = relative_path.replace(f"plugins.{package_version}", "plugins", 1)

                    # 创建插件文件夹并写入文件
                    file_path = Path(settings.ROOT_PATH) / "app" / relative_path
                    file_path.parent.mkdir(parents=True, exist_ok=True)
                    with open(file_path, "w", encoding="utf-8") as f:
                        f.write(res.text)
                    logger.debug(f"文件 {item.get('path')} 下载成功，保存路径：{file_path}")
                else:
                    # 如果是子目录，则将子目录内容加入栈中继续处理
                    sub_list, msg = self.__get_file_list(f"{current_pid}/{item.get('name')}", user_repo,
                                                         package_version)
                    if not sub_list:
                        return False, msg
                    stack.append((f"{current_pid}/{item.get('name')}", sub_list))

        return True, ""

    def __install_dependencies_if_required(self, pid: str) -> Tuple[bool, bool, str]:
        """
        安装插件依赖。
        :param pid: 插件 ID
        :return: (是否存在依赖，安装是否成功, 错误信息)
        """
        plugin_dir = PLUGIN_DIR / pid.lower()
        try:
            manifest = load_dependency_manifest(plugin_dir)
        except PluginDependencyManifestError as error:
            logger.error(f"{pid} 依赖清单无效：{error}")
            return True, False, str(error)
        if manifest is not None:
            logger.info(f"{pid} 存在依赖，开始尝试安装依赖")
            success, error_message = self.install_packages_with_fallback(manifest.path)
            return True, success, "" if success else error_message

        return False, False, "不存在依赖"

    @staticmethod
    def __backup_plugin(pid: str) -> str:
        """
        备份旧插件目录
        :param pid: 插件 ID
        :return: 备份目录路径
        """
        plugin_dir = PLUGIN_DIR / pid.lower()
        backup_dir = Path(settings.TEMP_PATH) / "plugin_backup" / pid.lower()

        if plugin_dir.exists():
            # 备份时清理已有的备份目录，防止残留文件影响
            if backup_dir.exists():
                shutil.rmtree(backup_dir, ignore_errors=True)
                logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}")

            shutil.copytree(plugin_dir, backup_dir, dirs_exist_ok=True)
            logger.debug(f"{pid} 插件已备份到 {backup_dir}")

        return str(backup_dir) if backup_dir.exists() else None

    @staticmethod
    def __restore_plugin(pid: str, backup_dir: str):
        """
        还原旧插件目录
        :param pid: 插件 ID
        :param backup_dir: 备份目录路径
        """
        plugin_dir = PLUGIN_DIR / pid.lower()
        if plugin_dir.exists():
            shutil.rmtree(plugin_dir, ignore_errors=True)
            logger.debug(f"{pid} 已清理插件目录 {plugin_dir}")

        if Path(backup_dir).exists():
            shutil.copytree(backup_dir, plugin_dir, dirs_exist_ok=True)
            logger.debug(f"{pid} 已还原插件目录 {plugin_dir}")
            shutil.rmtree(backup_dir, ignore_errors=True)
            logger.debug(f"{pid} 已删除备份目录 {backup_dir}")

    @staticmethod
    def __remove_old_plugin(pid: str):
        """
        删除旧插件
        :param pid: 插件 ID
        """
        plugin_dir = PLUGIN_DIR / pid.lower()
        if plugin_dir.exists():
            shutil.rmtree(plugin_dir, ignore_errors=True)

    @staticmethod
    def refresh_persistent_plugin_backup(pid: str) -> bool:
        """
        刷新插件持久化备份目录，供 docker 重置后恢复使用
        """
        if not SystemUtils.is_docker():
            return True

        plugin_dir = PLUGIN_DIR / pid.lower()
        if not plugin_dir.exists():
            logger.warn(f"{pid} 插件目录不存在，跳过刷新插件备份")
            return False

        backup_root = settings.CONFIG_PATH / "plugins_backup"
        backup_dir = backup_root / pid.lower()
        staging_dir = backup_root / f".{pid.lower()}.tmp-{uuid.uuid4().hex}"
        previous_dir = backup_root / f".{pid.lower()}.old-{uuid.uuid4().hex}"
        try:
            backup_root.mkdir(parents=True, exist_ok=True)
            shutil.copytree(
                plugin_dir,
                staging_dir,
                ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store")
            )
            if backup_dir.exists():
                backup_dir.replace(previous_dir)
            staging_dir.replace(backup_dir)
            if previous_dir.exists():
                shutil.rmtree(previous_dir, ignore_errors=True)
            logger.info(f"已刷新插件备份: {pid}")
            return True
        except Exception as e:
            if not backup_dir.exists() and previous_dir.exists():
                try:
                    previous_dir.replace(backup_dir)
                except Exception as rollback_error:
                    logger.error(
                        f"恢复插件旧备份失败，已保留恢复材料 {previous_dir}: "
                        f"{rollback_error}"
                    )
            logger.error(f"刷新插件备份失败: {pid} - {e}")
            return False
        finally:
            if staging_dir.exists():
                shutil.rmtree(staging_dir, ignore_errors=True)
            if backup_dir.exists() and previous_dir.exists():
                shutil.rmtree(previous_dir, ignore_errors=True)

    def __collect_plugin_wheels_dirs(self) -> List[Path]:
        """
        收集已安装插件目录下可用的 wheels 目录，供批量依赖安装时复用。
        """
        wheels_dirs = []
        try:
            install_plugins = {
                plugin_id.lower()
                for plugin_id in _installed_plugins_provider() or []
            }
            for plugin_id in install_plugins:
                wheels_dir = PLUGIN_DIR / plugin_id / "wheels"
                if wheels_dir.is_dir():
                    wheels_dirs.append(wheels_dir)
        except Exception as e:
            logger.error(f"收集插件 wheels 目录时发生错误：{e}")
            return []

        # 去重并保持稳定顺序，避免重复传递相同目录
        return list(dict.fromkeys(wheels_dirs))

    @staticmethod
    def __build_runtime_uv_command(*args: str) -> List[str]:
        """构造绑定当前解释器环境的 uv pip 命令。"""
        uv_bin = find_uv(Path(sys.executable))
        if not uv_bin:
            return []
        return [str(uv_bin), "pip", *args, "--python", sys.executable]

    @staticmethod
    def __format_package_name(name: str) -> str:
        """将内部包名转换为依赖清单常用的连字符形式。"""
        return name.replace("_", "-")

    @staticmethod
    def __marker_matches(marker, extra: str = "") -> bool:
        """
        使用当前运行环境和可选 extra 上下文判断 marker 是否生效。
        """
        if not marker:
            return True
        try:
            env = default_environment()
            env["extra"] = extra
            return marker.evaluate(env)
        except Exception as err:
            logger.debug(f"依赖 marker 计算失败，按不匹配处理：{err}")
            return False

    @classmethod
    def __parse_project_requirement_roots(
            cls,
            project_file: Path,
    ) -> Dict[str, Set[str]]:
        """解析主项目 pyproject，收集当前平台生效的根依赖和 extras。"""
        roots = {}
        if not project_file.exists():
            logger.warning(f"主项目依赖文件不存在：{project_file}")
            return roots

        try:
            manifest = load_dependency_file(project_file)
            for requirement in manifest.dependencies:
                if not cls.__marker_matches(requirement.marker):
                    continue
                package_name = cls.__standardize_pkg_name(requirement.name)
                roots.setdefault(package_name, set()).update(
                    extra.lower() for extra in requirement.extras
                )
            return roots
        except Exception as e:
            logger.error(f"解析主项目依赖文件失败：{project_file} - {e}")
            return {}

    @classmethod
    def __get_installed_distribution_requirements(cls) -> Dict[str, Tuple[Version, List[Requirement]]]:
        """
        获取当前环境中每个已安装包的依赖声明，用于展开主程序依赖图。
        """
        requirement_graph = {}
        try:
            for dist in distributions():
                name = dist.metadata.get("Name")
                if not name:
                    continue

                package_name = cls.__standardize_pkg_name(name)
                version_str = dist.metadata.get("Version") or getattr(dist, "version", None)
                if not version_str:
                    continue

                try:
                    version = Version(version_str)
                except InvalidVersion:
                    logger.debug(f"无法解析已安装包 '{package_name}' 的版本：{version_str}")
                    continue

                requirements = []
                for raw_requirement in dist.requires or []:
                    try:
                        requirements.append(Requirement(raw_requirement))
                    except Exception as err:
                        logger.debug(f"无法解析已安装包 '{package_name}' 的依赖项 '{raw_requirement}'：{err}")

                if package_name not in requirement_graph or version > requirement_graph[package_name][0]:
                    requirement_graph[package_name] = (version, requirements)
            return requirement_graph
        except Exception as e:
            logger.error(f"收集已安装包依赖图时发生错误：{e}")
            return {}

    @classmethod
    def __get_protected_runtime_packages(
            cls,
            installed_packages: Optional[Dict[str, Version]] = None
    ) -> Dict[str, Version]:
        """
        仅收集主程序依赖图中的已安装包版本。

        主项目 pyproject 中声明的根依赖及其当前已安装的传递依赖都会被冻结，
        未被主程序依赖图引用的插件自带包允许后续插件按需升级或降级。
        """
        if installed_packages is None:
            installed_packages = cls.__get_installed_packages()
        protected_packages = {
            package_name: version
            for package_name, version in installed_packages.items()
            if package_name in cls._protected_runtime_packages
        }

        project_file = settings.ROOT_PATH / "pyproject.toml"
        root_requirements = cls.__parse_project_requirement_roots(project_file)
        if not root_requirements:
            return protected_packages

        requirement_graph = cls.__get_installed_distribution_requirements()
        active_extras = {
            package_name: set(extras)
            for package_name, extras in root_requirements.items()
        }
        pending_packages = deque(active_extras.keys())
        processed_extras: Dict[str, Set[str]] = {}

        while pending_packages:
            package_name = pending_packages.popleft()
            selected_extras = active_extras.get(package_name, set())
            previous_extras = processed_extras.get(package_name)
            if previous_extras is not None and selected_extras.issubset(previous_extras):
                continue

            processed_extras[package_name] = set(selected_extras)
            if package_name in installed_packages:
                protected_packages[package_name] = installed_packages[package_name]

            _, requirements = requirement_graph.get(package_name, (None, []))
            if not requirements:
                continue

            active_extra_values = [""] + sorted(selected_extras)
            for requirement in requirements:
                if requirement.marker and not any(
                        cls.__marker_matches(requirement.marker, extra)
                        for extra in active_extra_values
                ):
                    continue

                dep_name = cls.__standardize_pkg_name(requirement.name)
                known_extras = active_extras.setdefault(dep_name, set())
                before_len = len(known_extras)
                known_extras.update(extra.lower() for extra in requirement.extras)
                if dep_name not in processed_extras or len(known_extras) != before_len:
                    pending_packages.append(dep_name)

        return protected_packages

    @staticmethod
    def __is_upgrade_only_conflict(specifier_set: SpecifierSet, installed_version: Version) -> bool:
        """
        判断版本冲突是否只能通过升级来解决（specifier 允许的所有版本都严格高于已安装版本）。
        返回 True 表示纯升级冲突；返回 False 表示可能需要降级或无法确定方向。
        """
        has_lower_bound = False
        for spec in specifier_set:
            op = spec.operator
            ver_str = spec.version.rstrip("*").rstrip(".") or "0"
            try:
                ver = Version(ver_str)
            except InvalidVersion:
                return False

            if op in ("<", "<="):
                upper = ver if op == "<" else Version(f"{ver}.post0")
                if upper <= installed_version:
                    return False
            elif op == "==":
                if ver <= installed_version:
                    return False
            elif op == "~=":
                # ~=X.Y.Z 等价于 >=X.Y.Z, <X.(Y+1)；若 X.Y.Z <= 已安装版本说明需降级
                if ver <= installed_version:
                    return False
                has_lower_bound = True
            elif op in (">=", ">"):
                has_lower_bound = True
            # != 操作符：单独出现时可能允许低版本，需结合其他约束判断

        # 若没有任何明确的下限约束（仅 != 等），保守地视为不确定 → 返回 False
        return has_lower_bound

    @classmethod
    def __validate_runtime_dependency_conflicts(
            cls,
            dependency_file: Path,
            protected_packages: Dict[str, Version]
    ) -> Tuple[bool, str]:
        """
        在真正执行安装前，先拦截插件对主程序依赖的显式覆盖请求。

        共享 venv 场景下，仅冻结主程序依赖；插件新增依赖、以及插件之间共享的额外依赖，
        允许后续安装继续调整版本。
        """
        conflicts = []
        try:
            manifest = load_dependency_file(dependency_file)
            for requirement in manifest.dependencies:
                if not cls.__marker_matches(requirement.marker):
                    continue

                package_name = cls.__standardize_pkg_name(requirement.name)
                installed_version = protected_packages.get(package_name)
                if installed_version is None:
                    continue

                if requirement.url:
                    conflicts.append((
                        package_name,
                        str(installed_version),
                        f"来自 {requirement.url} 的同名包",
                        package_name in cls._protected_runtime_packages,
                    ))
                    continue

                if requirement.specifier and not requirement.specifier.contains(
                        installed_version,
                        prereleases=True
                ):
                    is_core = package_name in cls._protected_runtime_packages
                    # 非核心包的纯升级冲突允许放行，由安装约束控制实际版本。
                    if is_core or not cls.__is_upgrade_only_conflict(
                            requirement.specifier, installed_version):
                        conflicts.append((
                            package_name,
                            str(installed_version),
                            str(requirement.specifier),
                            is_core,
                        ))
        except Exception as e:
            logger.error(f"执行运行环境依赖冲突预检时发生错误：{e}")
            return False, f"插件依赖预检失败：{e}"

        if not conflicts:
            return True, ""

        def sort_key(item: Tuple[str, str, str, bool]) -> Tuple[int, str]:
            return 0 if item[3] else 1, item[0]

        details = []
        for package_name, installed_version, expected, _is_protected in sorted(conflicts, key=sort_key)[:5]:
            details.append(
                f"{cls.__format_package_name(package_name)} 当前为 {installed_version}，"
                f"插件要求 {expected}"
            )
        if len(conflicts) > 5:
            details.append(f"其余 {len(conflicts) - 5} 项冲突已省略")

        scope = "主程序核心依赖" if any(item[3] for item in conflicts) else "主程序依赖"
        return False, (
            f"插件依赖与当前运行环境的{scope}冲突：{'；'.join(details)}。"
            f"为避免共享运行环境被污染，已拒绝安装。"
        )

    @classmethod
    def __create_runtime_constraints_file(cls, protected_packages: Dict[str, Version]) -> Path:
        """
        以主程序依赖的当前已安装版本生成临时约束文件，确保插件安装不会改写主程序依赖。
        """
        temp_dir = Path(settings.TEMP_PATH) / "plugin_dependencies"
        temp_dir.mkdir(parents=True, exist_ok=True)
        with tempfile.NamedTemporaryFile(
                mode="w",
                encoding="utf-8",
                dir=temp_dir,
                prefix="runtime-constraints-",
                suffix=".txt",
                delete=False
        ) as temp_file:
            for package_name, version in sorted(protected_packages.items()):
                if package_name in cls._protected_runtime_packages:
                    # 核心包严格锁定，插件不得改写
                    temp_file.write(f"{cls.__format_package_name(package_name)}=={version}\n")
                else:
                    # 非核心主程序依赖：允许升级，但禁止降级
                    temp_file.write(f"{cls.__format_package_name(package_name)}>={version}\n")
        return Path(temp_file.name)

    @classmethod
    async def __async_create_runtime_constraints_file(
            cls,
            protected_packages: Dict[str, Version],
    ) -> Path:
        """创建临时约束文件，取消时等待创建收口并删除已产生的文件。"""
        create_task = asyncio.create_task(
            asyncio.to_thread(
                cls.__create_runtime_constraints_file,
                protected_packages,
            )
        )
        try:
            return await asyncio.shield(create_task)
        except asyncio.CancelledError:
            async def cleanup_created_file() -> None:
                try:
                    created_file = await create_task
                except BaseException:
                    return
                await asyncio.to_thread(created_file.unlink, missing_ok=True)

            cleanup_task = asyncio.create_task(cleanup_created_file())
            try:
                await await_task_to_terminal(cleanup_task)
            except Exception as err:
                logger.warning(f"[UV] 取消后清理运行环境约束文件失败：{err}")
            raise

    @staticmethod
    def __refresh_import_system():
        """
        依赖安装或修复后刷新当前解释器的导入缓存，保证后续动态导入能看到新状态。
        """
        importlib.reload(site)
        importlib.invalidate_caches()

    @classmethod
    def __build_package_install_request(
            cls,
            dependency_files: Path | Sequence[Path],
            find_links_dirs: Optional[List[Path]] = None,
            constraints_file: Optional[Path] = None,
            purpose: str = "plugin",
    ) -> PackageInstallRequest:
        """
        将 MoviePilot 运行配置转换为 uv 安装请求，统一缓存、镜像和代理语义。
        """
        if isinstance(dependency_files, Path):
            resolved_dependency_files = (dependency_files,)
        else:
            resolved_dependency_files = tuple(Path(item) for item in dependency_files)
        return PackageInstallRequest(
            dependency_files=resolved_dependency_files,
            python_bin=Path(sys.executable),
            find_links_dirs=find_links_dirs or [],
            constraints_file=constraints_file,
            config_dir=settings.CONFIG_PATH,
            package_cache_root=settings.PACKAGE_CACHE_PATH,
            package_index_url=settings.PIP_PROXY or None,
            proxy_url=settings.PROXY_HOST or None,
            purpose=purpose,
        )

    @classmethod
    def __repair_if_runtime_broken(
            cls,
            snapshot_file: Optional[Path] = None,
            baseline_health: Optional[Dict[str, Tuple[bool, str]]] = None
    ) -> Tuple[bool, str]:
        """
        安装失败后检查主运行环境；若相对安装前新增异常，先恢复主程序依赖再返回。
        """
        current_health = cls.__run_runtime_healthcheck()
        health_message = cls.__runtime_health_regression_message(
            baseline_health or {},
            current_health
        )
        if not health_message:
            return True, ""
        repair_ok, repair_message = cls.__repair_main_runtime_dependencies(snapshot_file)
        if not repair_ok:
            return False, f"插件依赖安装失败后主运行环境异常，且恢复失败：{health_message}; {repair_message}"
        restored_health = cls.__run_runtime_healthcheck()
        restored_message = cls.__runtime_health_regression_message(
            baseline_health or {},
            restored_health
        )
        if restored_message:
            return False, f"插件依赖安装失败后主运行环境异常，恢复后仍异常：{restored_message}"
        return True, "主运行环境已恢复"

    @classmethod
    def __run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]:
        """
        执行全部运行环境自检并返回逐项结果，避免前一项失败遮蔽后续异常。
        """
        health_snapshot = {}
        uv_check = cls.__build_runtime_uv_command("check")
        if uv_check:
            checks = [("uv check", uv_check)]
        else:
            health_snapshot["uv check"] = (False, "未找到 uv 可执行文件")
            checks = []
        checks.append(("核心依赖导入检查", [sys.executable, "-c", cls._runtime_import_probe]))
        for check_name, command in checks:
            success, message = SystemUtils.execute_with_subprocess(command)
            health_snapshot[check_name] = (success, message)
        return health_snapshot

    @staticmethod
    def __runtime_health_regression_message(
            baseline_health: Dict[str, Tuple[bool, str]],
            current_health: Dict[str, Tuple[bool, str]]
    ) -> str:
        """
        汇总相对基线从正常变为异常的检查项，不解析第三方工具的错误文本。
        """
        regressions = []
        for check_name, (success, message) in current_health.items():
            baseline_success = baseline_health.get(check_name, (True, ""))[0]
            if baseline_success and not success:
                regressions.append(f"{check_name}失败：{message}")
        return "；".join(regressions)

    @classmethod
    def __repair_main_runtime_dependencies(cls, snapshot_file: Optional[Path] = None) -> Tuple[bool, str]:
        """
        依赖安装后如果发现主运行环境已异常，优先恢复主程序依赖快照；
        若快照不可用，再按主项目锁定依赖恢复运行环境。
        """
        repair_target = snapshot_file
        repair_desc = "主程序依赖快照"
        if repair_target and not repair_target.exists():
            repair_target = None
        if repair_target is None:
            repair_target = settings.ROOT_PATH / "pyproject.toml"
            repair_desc = "主程序 uv.lock"
        if not repair_target.exists():
            return False, f"恢复依赖文件不存在：{repair_target}"
        if snapshot_file is None and not (settings.ROOT_PATH / "uv.lock").exists():
            return False, f"恢复依赖文件不存在：{settings.ROOT_PATH / 'uv.lock'}"

        last_error = ""
        request = cls.__build_package_install_request(repair_target, purpose="runtime-repair")
        strategies = (
            build_package_install_strategies(request)
            if snapshot_file is not None
            else build_project_sync_strategies(request)
        )
        for strategy in strategies:
            logger.warning(f"[UV] 运行环境异常，尝试使用策略：{strategy.strategy_name} 恢复{repair_desc}")
            success, message = SystemUtils.execute_with_subprocess(
                strategy.command,
                env=strategy.env,
                safe_command=strategy.safe_log_command,
            )
            if success:
                cls.__refresh_import_system()
                return True, message
            last_error = message
            logger.error(f"[UV] 使用策略：{strategy.strategy_name} 恢复{repair_desc}失败：{message}")
        return False, last_error or f"恢复{repair_desc}失败"

    @classmethod
    def install_packages_with_fallback(cls,
                                       dependency_files: Path | Sequence[Path],
                                       find_links_dirs: Optional[List[Path]] = None) -> Tuple[bool, str]:
        """
        使用自动降级策略安装依赖，并确保新安装的包可被动态导入
        :param dependency_files: 一个或多个插件依赖清单路径
        :param find_links_dirs: 额外的本地 wheels 目录列表
        :return: (是否成功, 错误信息)
        """
        if isinstance(dependency_files, Path):
            resolved_dependency_files = (dependency_files,)
        else:
            resolved_dependency_files = tuple(Path(item) for item in dependency_files)
        if not resolved_dependency_files:
            return False, "没有传入插件依赖清单"

        candidate_dirs = []
        for dependency_file in resolved_dependency_files:
            wheels_dir = dependency_file.parent / "wheels"
            if wheels_dir.is_dir():
                candidate_dirs.append(wheels_dir)
        if find_links_dirs:
            candidate_dirs.extend(find_links_dirs)

        # 去重并保持传入顺序
        resolved_dirs = []
        seen_dirs = set()
        for candidate_dir in candidate_dirs:
            candidate_path = Path(candidate_dir)
            if not candidate_path.is_dir():
                continue
            candidate_key = str(candidate_path.resolve())
            if candidate_key in seen_dirs:
                continue
            seen_dirs.add(candidate_key)
            resolved_dirs.append(candidate_path)

        if resolved_dirs:
            for local_wheels_dir in resolved_dirs:
                logger.debug(f"[UV] 发现可用的 wheels 目录: {local_wheels_dir}，将优先从本地安装。")
        else:
            logger.debug("[UV] 未发现可用的 wheels 目录，将仅使用在线源。")

        installed_packages = cls.__get_installed_packages()
        protected_packages = cls.__get_protected_runtime_packages(installed_packages)
        for dependency_file in resolved_dependency_files:
            check_ok, check_message = cls.__validate_runtime_dependency_conflicts(
                dependency_file,
                protected_packages,
            )
            if not check_ok:
                logger.error(f"[UV] 运行环境冲突预检失败：{check_message}")
                return False, check_message

        constraints_file = None
        if protected_packages:
            try:
                constraints_file = cls.__create_runtime_constraints_file(protected_packages)
            except Exception as e:
                logger.error(f"[UV] 创建运行环境约束文件失败：{e}")
                return False, f"创建运行环境约束文件失败：{e}"

        request = cls.__build_package_install_request(
            resolved_dependency_files,
            find_links_dirs=resolved_dirs,
            constraints_file=constraints_file,
            purpose="plugin",
        )
        strategies = build_package_install_strategies(request)

        try:
            # 安装器会修改当前解释器的 site-packages，安装与缓存刷新必须串行。
            with cls._package_install_lock:
                loaded_modules_before_install = set(sys.modules.keys())
                baseline_health = cls.__run_runtime_healthcheck()
                baseline_health_message = cls.__runtime_health_regression_message({}, baseline_health)
                if baseline_health_message:
                    logger.warning(
                        f"[UV] 安装前运行环境已存在异常，本次安装仅拦截新增异常：{baseline_health_message}"
                    )
                # 遍历策略进行安装
                last_error = ""
                for strategy in strategies:
                    logger.debug(
                        f"[UV] 尝试使用策略：{strategy.strategy_name} 安装依赖，"
                        f"命令：{' '.join(strategy.safe_log_command)}"
                    )
                    success, message = SystemUtils.execute_with_subprocess(
                        strategy.command,
                        env=strategy.env,
                        safe_command=strategy.safe_log_command,
                    )
                    if success:
                        logger.debug(f"[UV] 策略：{strategy.strategy_name} 安装依赖成功，输出：{message}")
                        current_health = cls.__run_runtime_healthcheck()
                        health_message = cls.__runtime_health_regression_message(
                            baseline_health,
                            current_health
                        )
                        if health_message:
                            logger.error(f"[UV] 依赖安装后运行环境自检失败：{health_message}")
                            repair_ok, repair_message = cls.__repair_main_runtime_dependencies(
                                constraints_file if protected_packages else None
                            )
                            if repair_ok:
                                restored_health = cls.__run_runtime_healthcheck()
                                restored_message = cls.__runtime_health_regression_message(
                                    baseline_health,
                                    restored_health
                                )
                                if not restored_message:
                                    cls.__refresh_import_system()
                                    return False, (
                                        f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：{health_message}"
                                    )
                                logger.error(
                                    f"[UV] 主程序依赖恢复后仍未通过健康检查：{restored_message}"
                                )
                                return False, (
                                    f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常："
                                    f"{restored_message}"
                                )
                            return False, (
                                f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败："
                                f"{repair_message}"
                            )

                        remaining_health_message = cls.__runtime_health_regression_message({}, current_health)
                        if remaining_health_message:
                            logger.warning(
                                f"[UV] 依赖安装成功，安装前已有的运行环境异常仍然存在："
                                f"{remaining_health_message}"
                            )

                        cls.__refresh_import_system()
                        loaded_modules_after_install = set(sys.modules.keys())
                        loaded_modules_during_install = loaded_modules_after_install - loaded_modules_before_install
                        logger.debug(f"[UV] 已刷新导入系统，新加载的模块: {loaded_modules_during_install}")
                        return True, message

                    last_error = message
                    repair_ok, repair_message = cls.__repair_if_runtime_broken(
                        constraints_file if protected_packages else None,
                        baseline_health
                    )
                    logger.error(f"[UV] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}")
                    if not repair_ok or repair_message:
                        return False, (
                            f"策略 {strategy.strategy_name} 安装依赖失败：{message}；"
                            f"{repair_message}"
                        )
        finally:
            if constraints_file:
                constraints_file.unlink(missing_ok=True)

        if last_error:
            return False, f"[UV] 所有策略均安装依赖失败：{last_error}"
        return False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束"

    @staticmethod
    def __request_with_fallback(url: str,
                                headers: Optional[dict] = None,
                                timeout: Optional[int] = 60,
                                is_api: bool = False) -> Optional[Response]:
        """
        使用自动降级策略，请求资源，优先级依次为镜像站、代理、直连
        :param url: 目标URL
        :param headers: 请求头信息
        :param timeout: 请求超时时间
        :param is_api: 是否为GitHub API请求，API请求不走镜像站
        :return: 请求成功则返回 Response，失败返回 None
        """
        strategies = []

        # 1. 尝试使用镜像站，镜像站一般不支持API请求，因此API请求直接跳过镜像站
        if not is_api and settings.GITHUB_PROXY:
            proxy_url = f"{UrlUtils.standardize_base_url(settings.GITHUB_PROXY)}{url}"
            strategies.append(("镜像站", proxy_url, {"headers": headers, "timeout": timeout}))

        # 2. 尝试使用代理
        if settings.PROXY_HOST:
            strategies.append(("代理", url, {"headers": headers, "proxies": settings.PROXY, "timeout": timeout}))

        # 3. 最后尝试直连
        strategies.append(("直连", url, {"headers": headers, "timeout": timeout}))

        # 遍历策略并尝试请求
        for strategy_name, target_url, request_params in strategies:
            logger.debug(f"[GitHub] 尝试使用策略：{strategy_name} 请求 URL：{target_url}")

            try:
                res = RequestUtils(**request_params).get_res(url=target_url, raise_exception=True)
                logger.debug(f"[GitHub] 请求成功，策略：{strategy_name}, URL: {target_url}")
                return res
            except Exception as e:
                logger.error(f"[GitHub] 请求失败，策略：{strategy_name}, URL: {target_url}，错误：{str(e)}")

        logger.error(f"[GitHub] 所有策略均请求失败，URL: {url}，请检查网络连接或 GitHub 配置")
        return None

    def __get_plugin_meta(self, pid: str, repo_url: str,
                          package_version: Optional[str]) -> dict:
        """读取远端插件元数据，并把异常收敛为空映射。"""
        try:
            plugins = (
                          self.get_plugins(repo_url) if not package_version
                          else self.get_plugins(repo_url, package_version)
                      ) or {}
            meta = plugins.get(pid)
            return meta if isinstance(meta, dict) else {}
        except Exception as e:
            logger.error(f"获取插件 {pid} 元数据失败：{e}")
            return {}

    def get_plugin_system_version_check_message(self, pid: str, repo_url: str) -> Optional[str]:
        """
        获取指定插件来源的主系统版本兼容错误；兼容或无法定位元数据时返回 None。
        """
        if not pid or not repo_url:
            return None

        if self.is_local_repo_url(repo_url):
            candidate = self.get_local_plugin_candidate(
                pid=pid,
                package_version=self.parse_local_repo_package_version(repo_url),
                repo_path=self.parse_local_repo_path(repo_url),
                strict_compat=False
            )
            if not candidate:
                return None
            compatible, message = self.check_plugin_system_version(candidate)
            return None if compatible else message

        package_version = self.get_plugin_package_version(pid, repo_url, settings.VERSION_FLAG)
        if package_version is None:
            return None
        meta = self.__get_plugin_meta(pid, repo_url, package_version)
        compatible, message = self.check_plugin_system_version(meta)
        return None if compatible else message

    async def async_get_plugin_system_version_check_message(self, pid: str, repo_url: str) -> Optional[str]:
        """
        异步获取指定插件来源的主系统版本兼容错误；兼容或无法定位元数据时返回 None。
        """
        if not pid or not repo_url:
            return None

        if self.is_local_repo_url(repo_url):
            return await asyncio.to_thread(self.get_plugin_system_version_check_message, pid, repo_url)

        package_version = await self.async_get_plugin_package_version(pid, repo_url, settings.VERSION_FLAG)
        if package_version is None:
            return None
        meta = await self.__async_get_plugin_meta(pid, repo_url, package_version)
        compatible, message = self.check_plugin_system_version(meta)
        return None if compatible else message

    def __install_flow_sync(self, pid: str, force_install: bool,
                            prepare_content: Callable[[], Tuple[bool, str]],
                            repo_url: Optional[str] = None) -> Tuple[bool, str]:
        """
        同步安装统一流程：备份→清理→准备内容→安装依赖→上报
        prepare_content 负责把插件文件放到 app/plugins/{pid}
        """
        backup_dir = None
        if not force_install:
            backup_dir = self.__backup_plugin(pid)

        self.__remove_old_plugin(pid)

        success, message = prepare_content()
        if not success:
            logger.error(f"{pid} 准备插件内容失败：{message}")
            if backup_dir:
                self.__restore_plugin(pid, backup_dir)
                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
            else:
                self.__remove_old_plugin(pid)
                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
            return False, message

        dependencies_exist, dep_ok, dep_msg = self.__install_dependencies_if_required(pid)
        if dependencies_exist and not dep_ok:
            logger.error(f"{pid} 依赖安装失败：{dep_msg}")
            if backup_dir:
                self.__restore_plugin(pid, backup_dir)
                logger.warn(f"{pid} 插件安装失败，已还原备份插件")
            else:
                self.__remove_old_plugin(pid)
                logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装")
            return False, dep_msg

        self.refresh_persistent_plugin_backup(pid)
        return True, ""

    @staticmethod
    def __validate_release_zip_name(name: str) -> None:
        """
        校验 release zip 成员名在 POSIX 与 Windows 语义下都只能表示相对路径。
        """
        if not name:
            raise ValueError("非法 Release 压缩包成员：成员名为空")
        if "\x00" in name:
            raise ValueError(f"非法 Release 压缩包成员：{name}")
        if "\\" in name:
            raise ValueError(f"非法 Release 压缩包成员：{name}")

        posix_path = PurePosixPath(name)
        windows_path = PureWindowsPath(name)
        if (
            name.startswith("//")
            or posix_path.is_absolute()
            or windows_path.is_absolute()
            or windows_path.drive
        ):
            raise ValueError(f"非法 Release 压缩包成员：{name}")

        parts = [part for part in posix_path.parts if part not in ("", ".")]
        if not parts:
            raise ValueError(f"非法 Release 压缩包成员：{name}")
        if ".." in parts:
            raise ValueError(f"非法 Release 压缩包成员：{name}")

    @staticmethod
    def __validate_release_zip_type(info: zipfile.ZipInfo) -> None:
        """
        release zip 只接受普通文件和目录，避免归档内的符号链接或设备文件影响安装边界。
        """
        mode = info.external_attr >> 16
        file_type = stat.S_IFMT(mode)
        if not file_type:
            return
        if stat.S_ISREG(mode) or stat.S_ISDIR(mode):
            return
        raise ValueError(f"非法 Release 压缩包成员：{info.filename}")

    @staticmethod
    def __get_release_zip_base_prefix(infos: List[zipfile.ZipInfo]) -> str:
        """
        识别 release zip 的单一顶层目录，用于保持插件包根目录剥离行为。
        """
        names = [info.filename for info in infos]
        names_with_slash = [name for name in names if "/" in name]
        if names_with_slash and len(names_with_slash) == len(names):
            first_seg = names_with_slash[0].split("/", 1)[0]
            if first_seg and all(name.startswith(first_seg + "/") for name in names):
                return first_seg + "/"
        return ""

    @classmethod
    def __iter_release_zip_targets(
        cls, zf: zipfile.ZipFile, dest_base: Path
    ) -> List[Tuple[zipfile.ZipInfo, Path, bool]]:
        """
        将 release zip 成员解析为安装目标路径，并保证目标路径不会逃逸插件目录。
        """
        infos = zf.infolist()
        for info in infos:
            cls.__validate_release_zip_type(info)
            cls.__validate_release_zip_name(info.filename)

        base_prefix = cls.__get_release_zip_base_prefix(infos)
        dest_root = dest_base.resolve()
        targets = []
        for info in infos:
            raw_name = info.filename
            rel_name = raw_name[len(base_prefix):] if base_prefix else raw_name
            if not rel_name:
                if base_prefix and raw_name == base_prefix:
                    continue
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}")

            cls.__validate_release_zip_name(rel_name)
            rel_parts = [part for part in PurePosixPath(rel_name).parts if part not in ("", ".")]
            if not rel_parts:
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}")
            dest_path = (dest_root / Path(*rel_parts)).resolve()
            try:
                dest_path.relative_to(dest_root)
            except ValueError as exc:
                raise ValueError(f"非法 Release 压缩包成员：{raw_name}") from exc
            targets.append((info, dest_path, info.is_dir()))
        return targets

    def __install_from_release(self, pid: str, user_repo: str, release_tag: str) -> Tuple[bool, str]:
        """
        通过 GitHub Release 资产文件安装插件。
        规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件；
        将其全部解压到 app/plugins/{pid}
        """
        # 拼接资产文件名
        asset_name = f"{release_tag.lower()}.zip"

        release_api = f"https://api.github.com/repos/{user_repo}/releases/tags/{release_tag}"
        rel_res = self.__request_with_fallback(
            release_api,
            headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
            timeout=30,
            is_api=True,
        )
        if rel_res is None or rel_res.status_code != 200:
            return False, f"获取 Release 信息失败：{rel_res.status_code if rel_res else '连接失败'}"

        try:
            rel_json = rel_res.json()
            assets = rel_json.get("assets") or []
            asset = next((a for a in assets if a.get("name") == asset_name), None)
            if not asset:
                return False, f"未找到资产文件：{asset_name}"
            asset_id = asset.get("id")
            if not asset_id:
                return False, "资产缺少ID信息"
            # 构建资产的API下载URL
            download_url = f"https://api.github.com/repos/{user_repo}/releases/assets/{asset_id}"
        except Exception as e:
            logger.error(f"解析 Release 信息失败：{e}")
            return False, f"解析 Release 信息失败：{e}"

        # 使用资产的API端点下载，需要设置Accept头为application/octet-stream
        headers = settings.REPO_GITHUB_HEADERS(repo=user_repo).copy()
        headers["Accept"] = "application/octet-stream"
        res = self.__request_with_fallback(download_url, headers=headers, is_api=True)
        if res is None or res.status_code != 200:
            return False, f"下载资产失败：{res.status_code if res else '连接失败'}"

        try:
            with zipfile.ZipFile(io.BytesIO(res.content)) as zf:
                infos = zf.infolist()
                if not infos:
                    return False, "压缩包内容为空"
                dest_base = Path(settings.ROOT_PATH) / "app" / "plugins" / pid.lower()
                targets = self.__iter_release_zip_targets(zf, dest_base)
                wrote_any = False
                for info, dest_path, is_dir in targets:
                    if is_dir:
                        dest_path.mkdir(parents=True, exist_ok=True)
                        continue
                    dest_path.parent.mkdir(parents=True, exist_ok=True)
                    with zf.open(info, 'r') as src, open(dest_path, 'wb') as dst:
                        dst.write(src.read())
                    wrote_any = True
                if not wrote_any:
                    return False, "压缩包中无可写入文件"
            return True, ""
        except Exception as e:
            logger.error(f"解压 Release 压缩包失败：{e}")
            return False, f"解压 Release 压缩包失败：{e}"

    def find_missing_dependencies(self) -> List[str]:
        """兼容旧市场入口，转发到独立依赖适配器。"""
        installer = importlib.import_module(
            "app.adapters.system.plugin.dependency"
        ).PluginDependencyInstaller
        return installer(
            self,
            installed_plugins_provider=_installed_plugins_provider,
            plugin_dir=PLUGIN_DIR,
        ).find_missing()

    def install_dependencies(self, dependencies: List[str]) -> Tuple[bool, str]:
        """兼容旧市场入口，转发到独立依赖适配器。"""
        installer = importlib.import_module(
            "app.adapters.system.plugin.dependency"
        ).PluginDependencyInstaller
        return installer(
            self,
            installed_plugins_provider=_installed_plugins_provider,
            plugin_dir=PLUGIN_DIR,
        ).install(dependencies)

    @classmethod
    def __get_installed_packages(cls) -> Dict[str, Version]:
        """
        获取已安装的包及其版本
        使用 importlib.metadata 获取当前环境中已安装的包，标准化包名并转换版本信息
        对于无法解析的版本，记录警告日志并跳过
        :return: 已安装包的字典，格式为 {package_name: Version}
        """
        installed_packages = {}
        try:
            for dist in distributions():
                name = dist.metadata.get("Name")
                if not name:
                    continue
                pkg_name = cls.__standardize_pkg_name(name)
                version_str = dist.metadata.get("Version") or getattr(dist, "version", None)
                if not version_str:
                    continue
                try:
                    v = Version(version_str)
                    if pkg_name not in installed_packages or v > installed_packages[pkg_name]:
                        installed_packages[pkg_name] = v
                except InvalidVersion:
                    logger.debug(f"无法解析已安装包 '{pkg_name}' 的版本：{version_str}")
                    continue
            return installed_packages
        except Exception as e:
            logger.error(f"获取已安装的包时发生错误：{e}")
            return {}

    @staticmethod
    def __standardize_pkg_name(name: str) -> str:
        """
        标准化包名，将包名转换为小写，连字符与点替换为下划线（与 PEP 503 归一化风格一致）

        :param name: 原始包名
        :return: 标准化后的包名
        """
        if not name:
            return name
        return name.lower().replace("-", "_").replace(".", "_")

    async def async_get_plugin_package_version(self, pid: str, repo_url: str,
                                               package_version: Optional[str] = None) -> Optional[str]:
        """
        异步版本的获取插件版本方法，功能同 get_plugin_package_version
        """
        if not package_version:
            package_version = settings.VERSION_FLAG

        plugin = (
            await self.async_get_plugins(repo_url, package_version) or {}
        ).get(pid)
        if plugin and self.is_package_plugin_compatible(
                plugin, package_version
        ):
            return package_version

        # 异步安装链路与同步链路使用相同的 V2 默认兼容规则。
        for backward_flag in VERSION_BACKWARD_COMPATIBLE_FLAGS.get(package_version, []):
            plugin = (
                await self.async_get_plugins(repo_url, backward_flag) or {}
            ).get(pid)
            if plugin and self.is_package_plugin_compatible(
                    plugin, backward_flag
            ):
                return backward_flag

        plugin = (await self.async_get_plugins(repo_url) or {}).get(pid, None)
        if plugin and self.is_package_plugin_compatible(plugin, ""):
            return ""

        return None

    @staticmethod
    async def __async_request_with_fallback(url: str,
                                            headers: Optional[dict] = None,
                                            timeout: Optional[int] = 60,
                                            is_api: bool = False) -> Optional[httpx2.Response]:
        """
        使用自动降级策略，异步请求资源，优先级依次为镜像站、代理、直连
        :param url: 目标URL
        :param headers: 请求头信息
        :param timeout: 请求超时时间
        :param is_api: 是否为GitHub API请求，API请求不走镜像站
        :return: 请求成功则返回 Response，失败返回 None
        """
        strategies = []

        # 1. 尝试使用镜像站，镜像站一般不支持API请求，因此API请求直接跳过镜像站
        if not is_api and settings.GITHUB_PROXY:
            proxy_url = f"{UrlUtils.standardize_base_url(settings.GITHUB_PROXY)}{url}"
            strategies.append(("镜像站", proxy_url, {"headers": headers, "timeout": timeout}))

        # 2. 尝试使用代理
        if settings.PROXY_HOST:
            strategies.append(("代理", url, {"headers": headers, "proxies": settings.PROXY, "timeout": timeout}))

        # 3. 最后尝试直连
        strategies.append(("直连", url, {"headers": headers, "timeout": timeout}))

        # 遍历策略并尝试请求
        for strategy_name, target_url, request_params in strategies:
            logger.debug(f"[GitHub] 尝试使用策略：{strategy_name} 请求 URL：{target_url}")

            try:
                res = await AsyncRequestUtils(**request_params).get_res(url=target_url, raise_exception=True)
                logger.debug(f"[GitHub] 请求成功，策略：{strategy_name}, URL: {target_url}")
                return res
            except Exception as e:
                logger.error(f"[GitHub] 请求失败，策略：{strategy_name}, URL: {target_url}，错误：{str(e)}")

        logger.error(f"[GitHub] 所有策略均请求失败，URL: {url}，请检查网络连接或 GitHub 配置")
        return None

    @cached(maxsize=128, ttl=1800)
    async def async_get_plugins(self, repo_url: str,
                                package_version: Optional[str] = None) -> Optional[Dict[str, dict]]:
        """
        异步获取Github所有最新插件列表
        :param repo_url: Github仓库地址
        :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本
        """
        if not repo_url:
            return None

        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return None

        raw_url = self._base_url.format(user=user, repo=repo)
        package_url = f"{raw_url}package.{package_version}.json" if package_version else f"{raw_url}package.json"
        package_url = self.__append_cache_buster(package_url)

        res = await self.__async_request_with_fallback(package_url,
                                                       headers=settings.REPO_GITHUB_HEADERS(repo=f"{user}/{repo}"))
        if res is None:
            return None
        if res.status_code == 404:
            return {}
        if res.status_code != 200:
            return None
        return self.__parse_plugin_index_response(res.text)

    @cached(maxsize=32, ttl=1800, shared_key="get_plugin_repo_releases")
    async def _async_get_plugin_repo_releases(self, repo_url: str) -> Optional[List[dict]]:
        """
        异步按仓库获取 GitHub Release 原始分页数据。
        """
        if not repo_url:
            return []

        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return []

        user_repo = f"{user}/{repo}"
        releases = []
        for page in range(1, 11):
            release_api = f"https://api.github.com/repos/{user_repo}/releases?per_page=100&page={page}"
            release_api = self.__append_cache_buster(release_api)
            res = await self.__async_request_with_fallback(
                release_api,
                headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
                timeout=30,
                is_api=True,
            )
            if res is None or res.status_code != 200:
                return None

            try:
                payload = res.json()
                if not payload:
                    break
                if not isinstance(payload, list):
                    return None
                releases.extend(self.__normalize_plugin_release_response(payload))
                if len(payload) < 100:
                    break
            except Exception as e:
                logger.error(f"解析插件仓库 {repo_url} Release 列表失败：{e}")
                return None
        return releases

    async def async_get_plugin_release_versions(self, pid: str, repo_url: str) -> List[dict]:
        """
        异步获取插件可安装的 GitHub Release 版本列表。

        同一事件循环内，同仓库的并发读取和强制刷新共享一个请求任务。
        """
        if not pid or not repo_url:
            return []

        loop = asyncio.get_running_loop()
        normalized_repo_url = repo_url.rstrip("/")
        normal_task_key = (loop, normalized_repo_url, False)
        force_task_key = (loop, normalized_repo_url, True)
        with self._release_task_lock:
            if is_fresh():
                force_task = self._release_tasks.get(force_task_key)
                if force_task and not force_task.done():
                    task_key = force_task_key
                    task = force_task
                else:
                    pending_normal_task = self._release_tasks.get(normal_task_key)
                    if pending_normal_task and pending_normal_task.done():
                        pending_normal_task = None
                    task_key = force_task_key
                    task = get_task_registry().create(
                        self._async_refresh_plugin_repo_releases(
                            normalized_repo_url,
                            pending_normal_task,
                        ),
                        owner="plugin.market.release_refresh",
                    )
                    self._release_tasks[task_key] = task
                    task.add_done_callback(
                        lambda completed_task: self._remove_release_task(task_key, completed_task)
                    )
            else:
                task_key = normal_task_key
                pending_normal_task = self._release_tasks.get(normal_task_key)
                if pending_normal_task is None or pending_normal_task.done():
                    task = get_task_registry().create(
                        self._async_get_plugin_repo_releases(normalized_repo_url),
                        owner="plugin.market.release_read",
                    )
                    self._release_tasks[task_key] = task
                    task.add_done_callback(
                        lambda completed_task: self._remove_release_task(task_key, completed_task)
                    )
                else:
                    task = pending_normal_task

        payload = await asyncio.shield(task)
        return self.__parse_plugin_release_response(pid, payload)

    async def async_has_plugin_release_cache(self, repo_url: str) -> bool:
        """
        判断指定仓库的 Release 列表缓存是否已经存在。
        """
        if not repo_url:
            return False
        return await self._async_get_plugin_repo_releases.cache_exists(
            self, repo_url.rstrip("/")
        )

    async def _async_refresh_plugin_repo_releases(
        self,
        repo_url: str,
        pending_normal_task: Optional[asyncio.Task],
    ) -> Optional[List[dict]]:
        """等待在途普通读取落盘后执行强刷，确保旧结果不会覆盖强刷缓存。"""
        if pending_normal_task:
            try:
                await asyncio.shield(pending_normal_task)
            except (Exception, asyncio.CancelledError):
                pass
        return await self._async_get_plugin_repo_releases(repo_url)

    @classmethod
    def _remove_release_task(cls, task_key: Tuple[asyncio.AbstractEventLoop, str, bool], task: asyncio.Task) -> None:
        """请求任务完成后释放事件循环和仓库引用。"""
        with cls._release_task_lock:
            if cls._release_tasks.get(task_key) is task:
                cls._release_tasks.pop(task_key, None)

    async def __async_get_file_list(self, pid: str, user_repo: str, package_version: Optional[str] = None) -> \
            Tuple[Optional[list], Optional[str]]:
        """
        异步获取插件的文件列表
        :param pid: 插件 ID
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (文件列表, 错误信息)
        """
        file_api = f"https://api.github.com/repos/{user_repo}/contents/plugins"
        # 如果 package_version 存在（如 "v2"），则加上版本号
        if package_version:
            file_api += f".{package_version}"
        file_api += f"/{pid.lower()}"

        res = await self.__async_request_with_fallback(file_api,
                                                       headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
                                                       is_api=True,
                                                       timeout=30)
        if res is None:
            return None, "连接仓库失败"
        elif res.status_code != 200:
            return None, f"连接仓库失败：{res.status_code} - " \
                         f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.text}"

        try:
            ret = res.json()
            if isinstance(ret, list) and len(ret) > 0 and "message" not in ret[0]:
                return ret, ""
            else:
                return None, "插件在仓库中不存在或返回数据格式不正确"
        except Exception as e:
            logger.error(f"插件数据解析失败：{e}")
            return None, "插件数据解析失败"

    async def __async_download_files(self, pid: str, file_list: List[dict], user_repo: str,
                                     package_version: Optional[str] = None) -> Tuple[bool, str]:
        """
        异步下载插件文件
        :param pid: 插件 ID
        :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接）
        :param user_repo: GitHub 仓库的 user/repo 路径
        :return: (是否成功, 错误信息)
        """
        if not file_list:
            return False, "文件列表为空"

        # 使用栈结构来替代递归调用，避免递归深度过大问题
        stack = [(pid, file_list)]

        while stack:
            current_pid, current_file_list = stack.pop()

            for item in current_file_list:
                if item.get("download_url"):
                    logger.debug(f"正在下载文件：{item.get('path')}")
                    res = await self.__async_request_with_fallback(item.get('download_url'),
                                                                   headers=settings.REPO_GITHUB_HEADERS(repo=user_repo))
                    if not res:
                        return False, f"文件 {item.get('path')} 下载失败！"
                    elif res.status_code != 200:
                        return False, f"下载文件 {item.get('path')} 失败：{res.status_code}"

                    # 确保文件路径不包含版本号（如 v2、v3），如果有 package_version，移除路径中的版本号
                    relative_path = item.get("path")
                    if package_version:
                        relative_path = relative_path.replace(f"plugins.{package_version}", "plugins", 1)

                    # 创建插件文件夹并写入文件
                    file_path = AsyncPath(settings.ROOT_PATH) / "app" / relative_path
                    await file_path.parent.mkdir(parents=True, exist_ok=True)
                    async with aiofiles.open(file_path, "w", encoding="utf-8") as f:
                        await f.write(res.text)
                    logger.debug(f"文件 {item.get('path')} 下载成功，保存路径：{file_path}")
                else:
                    # 如果是子目录，则将子目录内容加入栈中继续处理
                    sub_list, msg = await self.__async_get_file_list(f"{current_pid}/{item.get('name')}", user_repo,
                                                                     package_version)
                    if not sub_list:
                        return False, msg
                    stack.append((f"{current_pid}/{item.get('name')}", sub_list))

        return True, ""

    @classmethod
    async def __async_run_runtime_healthcheck(cls) -> Dict[str, Tuple[bool, str]]:
        """异步执行插件安装后的运行环境检查。"""
        health_snapshot: Dict[str, Tuple[bool, str]] = {}
        uv_check = cls.__build_runtime_uv_command("check")
        if uv_check:
            checks = [("uv check", uv_check)]
        else:
            health_snapshot["uv check"] = (False, "未找到 uv 可执行文件")
            checks = []
        checks.append(("核心依赖导入检查", [
            sys.executable,
            "-c",
            cls._runtime_import_probe,
        ]))
        for check_name, command in checks:
            health_snapshot[check_name] = (
                await SystemUtils.execute_with_subprocess_async(
                    command,
                    timeout=30,
                )
            )
        return health_snapshot

    @classmethod
    async def __async_repair_main_runtime_dependencies(
            cls,
            snapshot_file: Optional[Path] = None,
    ) -> Tuple[bool, str]:
        """异步恢复主程序运行依赖，避免修复命令绕过可取消进程边界。"""
        repair_target = snapshot_file
        repair_desc = "主程序依赖快照"
        if repair_target and not await _await_thread_operation(repair_target.exists):
            repair_target = None
        if repair_target is None:
            repair_target = settings.ROOT_PATH / "pyproject.toml"
            repair_desc = "主程序 uv.lock"
        if not await _await_thread_operation(repair_target.exists):
            return False, f"恢复依赖文件不存在：{repair_target}"
        lock_file = settings.ROOT_PATH / "uv.lock"
        if snapshot_file is None and not await _await_thread_operation(lock_file.exists):
            return False, f"恢复依赖文件不存在：{settings.ROOT_PATH / 'uv.lock'}"

        request = cls.__build_package_install_request(
            repair_target,
            purpose="runtime-repair",
        )
        strategies = (
            build_package_install_strategies(request)
            if snapshot_file is not None
            else build_project_sync_strategies(request)
        )
        last_error = ""
        for strategy in strategies:
            logger.warning(
                f"[UV] 运行环境异常，尝试使用策略：{strategy.strategy_name} 恢复{repair_desc}"
            )
            success, message = await SystemUtils.execute_with_subprocess_async(
                strategy.command,
                env=strategy.env,
                safe_command=strategy.safe_log_command,
                timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT,
            )
            if success:
                cls.__refresh_import_system()
                return True, message
            last_error = message
            logger.error(
                f"[UV] 使用策略：{strategy.strategy_name} 恢复{repair_desc}失败：{message}"
            )
        return False, last_error or f"恢复{repair_desc}失败"

    @classmethod
    async def __async_repair_if_runtime_broken(
            cls,
            snapshot_file: Optional[Path],
            baseline_health: Dict[str, Tuple[bool, str]],
    ) -> Tuple[bool, str]:
        """异步检查并修复安装过程中新增的主程序环境异常。"""
        current_health = await cls.__async_run_runtime_healthcheck()
        health_message = cls.__runtime_health_regression_message(
            baseline_health,
            current_health,
        )
        if not health_message:
            return True, ""
        repair_ok, repair_message = (
            await cls.__async_repair_main_runtime_dependencies(snapshot_file)
        )
        if not repair_ok:
            return False, (
                f"插件依赖安装失败后主运行环境异常，且恢复失败："
                f"{health_message}; {repair_message}"
            )
        restored_health = await cls.__async_run_runtime_healthcheck()
        restored_message = cls.__runtime_health_regression_message(
            baseline_health,
            restored_health,
        )
        if restored_message:
            return False, (
                f"插件依赖安装失败后主运行环境异常，恢复后仍异常："
                f"{restored_message}"
            )
        return True, "主运行环境已恢复"

    async def async_install_packages_with_fallback(
            self,
            dependency_files: Path | Sequence[Path],
            find_links_dirs: Optional[List[Path]] = None,
    ) -> Tuple[bool, str]:
        """通过可取消子进程异步安装一组插件依赖清单。"""
        return await self.__async_install_packages_with_fallback(
            dependency_files,
            find_links_dirs,
        )

    @classmethod
    async def __async_install_packages_with_fallback(
            cls,
            dependency_files: Path | Sequence[Path],
            find_links_dirs: Optional[List[Path]] = None,
    ) -> Tuple[bool, str]:
        """异步安装插件依赖，并让取消能够终止 uv 子进程。"""
        if isinstance(dependency_files, Path):
            resolved_dependency_files = (dependency_files,)
        else:
            resolved_dependency_files = tuple(Path(item) for item in dependency_files)
        if not resolved_dependency_files:
            return False, "没有传入插件依赖清单"

        candidate_dirs = []
        for dependency_file in resolved_dependency_files:
            wheels_dir = dependency_file.parent / "wheels"
            if await _await_thread_operation(wheels_dir.is_dir):
                candidate_dirs.append(wheels_dir)
        if find_links_dirs:
            candidate_dirs.extend(find_links_dirs)

        resolved_dirs = []
        seen_dirs = set()
        for candidate_dir in candidate_dirs:
            candidate_path = Path(candidate_dir)
            if not await _await_thread_operation(candidate_path.is_dir):
                continue
            candidate_key = str(
                await _await_thread_operation(candidate_path.resolve)
            )
            if candidate_key in seen_dirs:
                continue
            seen_dirs.add(candidate_key)
            resolved_dirs.append(candidate_path)

        installed_packages = await _await_thread_operation(
            cls.__get_installed_packages,
        )
        protected_packages = await _await_thread_operation(
            cls.__get_protected_runtime_packages,
            installed_packages,
        )
        for dependency_file in resolved_dependency_files:
            check_ok, check_message = await _await_thread_operation(
                cls.__validate_runtime_dependency_conflicts,
                dependency_file,
                protected_packages,
            )
            if not check_ok:
                logger.error(f"[UV] 运行环境冲突预检失败：{check_message}")
                return False, check_message

        constraints_file = None
        if protected_packages:
            try:
                constraints_file = await cls.__async_create_runtime_constraints_file(
                    protected_packages,
                )
            except Exception as err:
                logger.error(f"[UV] 创建运行环境约束文件失败：{err}")
                return False, f"创建运行环境约束文件失败：{err}"

        request = cls.__build_package_install_request(
            resolved_dependency_files,
            find_links_dirs=resolved_dirs,
            constraints_file=constraints_file,
            purpose="plugin",
        )
        strategies = build_package_install_strategies(request)
        acquired = False
        try:
            while not cls._package_install_lock.acquire(blocking=False):
                await asyncio.sleep(0.01)
            acquired = True
            baseline_health = await cls.__async_run_runtime_healthcheck()
            baseline_health_message = cls.__runtime_health_regression_message(
                {},
                baseline_health,
            )
            if baseline_health_message:
                logger.warning(
                    f"[UV] 安装前运行环境已存在异常，本次安装仅拦截新增异常："
                    f"{baseline_health_message}"
                )

            last_error = ""
            for strategy in strategies:
                logger.debug(
                    f"[UV] 尝试使用策略：{strategy.strategy_name} 安装依赖，"
                    f"命令：{' '.join(strategy.safe_log_command)}"
                )
                success, message = await SystemUtils.execute_with_subprocess_async(
                    strategy.command,
                    env=strategy.env,
                    safe_command=strategy.safe_log_command,
                    timeout=cls.PLUGIN_DEPENDENCY_INSTALL_TIMEOUT,
                )
                if success:
                    current_health = await cls.__async_run_runtime_healthcheck()
                    health_message = cls.__runtime_health_regression_message(
                        baseline_health,
                        current_health,
                    )
                    if health_message:
                        logger.error(f"[UV] 依赖安装后运行环境自检失败：{health_message}")
                        repair_ok, repair_message = (
                            await cls.__async_repair_main_runtime_dependencies(
                                constraints_file if protected_packages else None
                            )
                        )
                        if repair_ok:
                            restored_health = await cls.__async_run_runtime_healthcheck()
                            restored_message = cls.__runtime_health_regression_message(
                                baseline_health,
                                restored_health,
                            )
                            if not restored_message:
                                cls.__refresh_import_system()
                                return False, (
                                    f"依赖安装后运行环境自检失败，已自动恢复主程序依赖："
                                    f"{health_message}"
                                )
                            return False, (
                                f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常："
                                f"{restored_message}"
                            )
                        return False, (
                            f"依赖安装后运行环境自检失败，且自动恢复主程序依赖失败："
                            f"{repair_message}"
                        )

                    cls.__refresh_import_system()
                    return True, message

                last_error = message
                repair_ok, repair_message = await cls.__async_repair_if_runtime_broken(
                    constraints_file if protected_packages else None,
                    baseline_health,
                )
                logger.error(
                    f"[UV] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}"
                )
                if not repair_ok or repair_message:
                    return False, (
                        f"策略 {strategy.strategy_name} 安装依赖失败：{message}；"
                        f"{repair_message}"
                    )
            return False, (
                f"[UV] 所有策略均安装依赖失败：{last_error}"
                if last_error
                else "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束"
            )
        finally:
            if acquired:
                cls._package_install_lock.release()
            if constraints_file:
                await _await_thread_operation(
                    constraints_file.unlink,
                    missing_ok=True,
                )

    async def __async_backup_plugin(self, pid: str) -> str:
        """
        异步备份旧插件目录
        :param pid: 插件 ID
        :return: 备份目录路径
        """
        plugin_dir = AsyncPath(PLUGIN_DIR) / pid.lower()
        backup_dir = AsyncPath(settings.TEMP_PATH) / "plugin_backup" / pid.lower()

        if await plugin_dir.exists():
            try:
                if await backup_dir.exists():
                    await aioshutil.rmtree(backup_dir, ignore_errors=True)
                    logger.debug(f"{pid} 旧的备份目录已清理 {backup_dir}")

                await self._async_copytree(plugin_dir, backup_dir)
                logger.debug(f"{pid} 插件已备份到 {backup_dir}")
            except asyncio.CancelledError:
                await aioshutil.rmtree(backup_dir, ignore_errors=True)
                raise

        return str(backup_dir) if await backup_dir.exists() else None

    async def __async_restore_plugin(self, pid: str, backup_dir: str):
        """
        异步还原旧插件目录
        :param pid: 插件 ID
        :param backup_dir: 备份目录路径
        """
        plugin_dir = AsyncPath(PLUGIN_DIR) / pid.lower()
        if await plugin_dir.exists():
            await aioshutil.rmtree(plugin_dir, ignore_errors=True)
            logger.debug(f"{pid} 已清理插件目录 {plugin_dir}")

        backup_path = AsyncPath(backup_dir)
        if await backup_path.exists():
            await self._async_copytree(src=backup_path, dst=plugin_dir)
            logger.debug(f"{pid} 已还原插件目录 {plugin_dir}")
            await aioshutil.rmtree(backup_path, ignore_errors=True)
            logger.debug(f"{pid} 已删除备份目录 {backup_dir}")

    @staticmethod
    async def __async_remove_old_plugin(pid: str):
        """
        异步删除旧插件
        :param pid: 插件 ID
        """
        plugin_dir = AsyncPath(PLUGIN_DIR) / pid.lower()
        if await plugin_dir.exists():
            await aioshutil.rmtree(plugin_dir, ignore_errors=True)

    async def _async_copytree(self, src: AsyncPath, dst: AsyncPath):
        """
        异步递归复制目录
        :param src: 源目录
        :param dst: 目标目录
        """
        if not await src.exists():
            return

        await dst.mkdir(parents=True, exist_ok=True)

        async for item in src.iterdir():
            dst_item = dst / item.name
            if await item.is_dir():
                await self._async_copytree(item, dst_item)
            else:
                async with aiofiles.open(item, 'rb') as src_file:
                    content = await src_file.read()
                async with aiofiles.open(dst_item, 'wb') as dst_file:
                    await dst_file.write(content)

    async def __async_install_dependencies_if_required(self, pid: str) -> Tuple[bool, bool, str]:
        """
        异步安装插件依赖。
        :param pid: 插件 ID
        :return: (是否存在依赖，安装是否成功, 错误信息)
        """
        plugin_dir = PLUGIN_DIR / pid.lower()
        try:
            manifest = load_dependency_manifest(plugin_dir)
        except PluginDependencyManifestError as error:
            logger.error(f"{pid} 依赖清单无效：{error}")
            return True, False, str(error)
        if manifest is not None:
            logger.info(f"{pid} 存在依赖，开始尝试安装依赖")
            success, error_message = await self.__async_install_packages_with_fallback(manifest.path)
            return True, success, "" if success else error_message

        return False, False, "不存在依赖"

    async def async_install_dependencies(self, dependencies: List[str]) -> Tuple[bool, str]:
        """兼容旧异步市场入口，转发到独立依赖适配器。"""
        installer = importlib.import_module(
            "app.adapters.system.plugin.dependency"
        ).PluginDependencyInstaller
        return await installer(
            self,
            installed_plugins_provider=_installed_plugins_provider,
            plugin_dir=PLUGIN_DIR,
        ).async_install(dependencies)

    async def async_find_missing_dependencies(self) -> List[str]:
        """兼容旧异步市场入口，转发到独立依赖适配器。"""
        installer = importlib.import_module(
            "app.adapters.system.plugin.dependency"
        ).PluginDependencyInstaller
        return await installer(
            self,
            installed_plugins_provider=_installed_plugins_provider,
            plugin_dir=PLUGIN_DIR,
        ).async_find_missing()

    async def async_install(self, pid: str, repo_url: str, package_version: Optional[str] = None,
                            release_version: Optional[str] = None,
                            force_install: bool = False) -> Tuple[bool, str]:
        """
        异步安装插件，包括版本检查、内容准备、生效清单依赖安装和失败恢复。
        :param pid: 插件 ID
        :param repo_url: 插件仓库地址
        :param package_version: 首选插件版本 (如 "v2", "v3")，如不指定则默认使用系统配置的版本
        :param release_version: 指定安装的 release 资产版本；未指定时安装当前索引版本
        :param force_install: 是否强制安装插件，默认不启用，启用时不进行备份和恢复操作
        :return: (是否成功, 错误信息)
        """
        if self.is_local_repo_url(repo_url):
            return await _await_thread_operation(
                self.install_local,
                pid,
                repo_url,
                force_install,
            )

        if SystemUtils.is_frozen():
            return False, "可执行文件模式下，只能安装本地插件"

        # 验证参数
        if not pid or not repo_url:
            return False, "参数错误"

        # 从 GitHub 的 repo_url 获取用户和项目名
        user, repo = self.get_repo_info(repo_url)
        if not user or not repo:
            return False, "不支持的插件仓库地址格式"

        user_repo = f"{user}/{repo}"

        if not package_version:
            package_version = settings.VERSION_FLAG

        # 1. 优先检查指定版本的插件
        package_version = await self.async_get_plugin_package_version(pid, repo_url, package_version)
        # 如果 package_version 为None，说明没有找到匹配的插件
        if package_version is None:
            msg = f"{pid} 没有找到适用于当前版本的插件"
            logger.debug(msg)
            return False, msg
        # package_version 为空，表示从 package.json 中找到插件
        elif package_version == "":
            logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件")
        else:
            logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件")

        # 2. 统一异步安装流程（release 或文件列表）。
        meta = await self.__async_get_plugin_meta(pid, repo_url, package_version)
        # 是否使用 Release 打包；失败时兜底文件列表，保持同步/异步安装语义一致。
        is_release = meta.get("release")
        # 插件版本号
        plugin_version = meta.get("version")
        if release_version:
            if not is_release:
                return False, f"{pid} 未声明 Release 安装，无法安装指定版本"
            release_items = await self.async_get_plugin_release_versions(pid, repo_url)
            if not self.__has_installable_release_version(release_items, release_version):
                return False, f"{pid} 未找到可安装的 Release 版本：{release_version}"
            if release_version == plugin_version:
                compatible, message = self.check_plugin_system_version(meta)
                if not compatible:
                    logger.debug(f"{pid} 插件系统版本兼容性检查失败：{message}")
                    return False, message
            release_tag = f"{pid}_v{release_version}"

            async def prepare_selected_release() -> Tuple[bool, str]:
                return await self.__async_install_from_release(pid, user_repo, release_tag)

            return await self.__install_flow_async(pid, force_install, prepare_selected_release, repo_url)

        compatible, message = self.check_plugin_system_version(meta)
        if not compatible:
            logger.debug(f"{pid} 插件系统版本兼容性检查失败：{message}")
            return False, message
        if is_release:
            # 使用 插件ID_插件版本号 作为 Release tag
            if not plugin_version:
                return False, f"未在插件清单中找到 {pid} 的版本号，无法进行 Release 安装"
            # 拼接 release_tag
            release_tag = f"{pid}_v{plugin_version}"

            # 使用 release 进行安装
            async def prepare_release() -> Tuple[bool, str]:
                ok, msg = await self.__async_install_from_release(pid, user_repo, release_tag)
                if ok:
                    return True, msg
                logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}")
                await self.__async_remove_old_plugin(pid)
                return await self.__prepare_content_via_filelist_async(pid.lower(), user_repo, package_version)

            return await self.__install_flow_async(pid, force_install, prepare_release, repo_url)
        else:
            # 未声明 release 打包的插件继续使用文件列表方式安装。
            async def prepare_filelist() -> Tuple[bool, str]:
                return await self.__prepare_content_via_filelist_async(pid.lower(), user_repo, package_version)

            return await self.__install_flow_async(pid, force_install, prepare_filelist, repo_url)

    async def __async_get_plugin_meta(self, pid: str, repo_url: str,
                                      package_version: Optional[str]) -> dict:
        """异步读取远端插件元数据，并把异常收敛为空映射。"""
        try:
            plugins = (
                          await self.async_get_plugins(repo_url) if not package_version
                          else await self.async_get_plugins(repo_url, package_version)
                      ) or {}
            meta = plugins.get(pid)
            return meta if isinstance(meta, dict) else {}
        except Exception as e:
            logger.warn(f"获取插件 {pid} 元数据失败：{e}")
            return {}

    async def __install_flow_async(self, pid: str, force_install: bool,
                                   prepare_content: Callable[[], Awaitable[Tuple[bool, str]]],
                                   repo_url: Optional[str] = None) -> Tuple[bool, str]:
        """
        异步安装流程，处理插件内容准备、依赖安装和注册
        """
        backup_dir = None
        try:
            if not force_install:
                backup_dir = await self.__async_backup_plugin(pid)

            await self.__async_remove_old_plugin(pid)

            success, message = await prepare_content()
            if not success:
                logger.error(f"{pid} 准备插件内容失败：{message}")
                if backup_dir:
                    await self.__async_restore_plugin(pid, backup_dir)
                    logger.warning(f"{pid} 插件安装失败，已还原备份插件")
                else:
                    await self.__async_remove_old_plugin(pid)
                    logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装")
                return False, message

            dependencies_exist, dep_ok, dep_msg = (
                await self.__async_install_dependencies_if_required(pid)
            )
            if dependencies_exist and not dep_ok:
                logger.error(f"{pid} 依赖安装失败：{dep_msg}")
                if backup_dir:
                    await self.__async_restore_plugin(pid, backup_dir)
                    logger.warning(f"{pid} 插件安装失败，已还原备份插件")
                else:
                    await self.__async_remove_old_plugin(pid)
                    logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装")
                return False, dep_msg

            await _await_thread_operation(self.refresh_persistent_plugin_backup, pid)
            return True, ""
        except asyncio.CancelledError:
            logger.warning(
                f"{pid} 插件安装被取消，Python 依赖环境可能已经改变"
            )
            raise
        finally:
            if backup_dir:
                await aioshutil.rmtree(backup_dir, ignore_errors=True)

    def __prepare_content_via_filelist_sync(self, pid: str, user_repo: str,
                                            package_version: Optional[str]) -> Tuple[bool, str]:
        """
        同步准备插件内容，通过文件列表获取插件文件和依赖
        """
        file_list, msg = self.__get_file_list(pid, user_repo, package_version)
        if not file_list:
            return False, msg
        ok, m = self.__download_files(pid, file_list, user_repo, package_version)
        if not ok:
            return False, m
        return True, ""

    async def __prepare_content_via_filelist_async(self, pid: str, user_repo: str,
                                                   package_version: Optional[str]) -> Tuple[bool, str]:
        """
        异步准备插件内容，通过文件列表获取插件文件和依赖
        """
        file_list, msg = await self.__async_get_file_list(pid, user_repo, package_version)
        if not file_list:
            return False, msg
        ok, m = await self.__async_download_files(pid, file_list, user_repo, package_version)
        if not ok:
            return False, m
        return True, ""

    async def __async_install_from_release(self, pid: str, user_repo: str, release_tag: str) -> Tuple[bool, str]:
        """
        通过 GitHub Release 资产文件安装插件（异步）。
        规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件；
        将其全部解压到 app/plugins/{pid}
        """
        # 拼接资产文件名
        asset_name = f"{release_tag.lower()}.zip"

        release_api = f"https://api.github.com/repos/{user_repo}/releases/tags/{release_tag}"
        rel_res = await self.__async_request_with_fallback(
            release_api,
            headers=settings.REPO_GITHUB_HEADERS(repo=user_repo),
            timeout=30,
            is_api=True,
        )
        if rel_res is None or rel_res.status_code != 200:
            return False, f"获取 Release 信息失败：{rel_res.status_code if rel_res else '连接失败'}"

        try:
            rel_json = rel_res.json()
            assets = rel_json.get("assets") or []
            asset = next((a for a in assets if a.get("name") == asset_name), None)
            if not asset:
                return False, f"未找到资产文件：{asset_name}"
            asset_id = asset.get("id")
            if not asset_id:
                return False, "资产缺少ID信息"
            # 构建资产的API下载URL
            download_url = f"https://api.github.com/repos/{user_repo}/releases/assets/{asset_id}"
        except Exception as e:
            logger.error(f"解析 Release 信息失败：{e}")
            return False, f"解析 Release 信息失败：{e}"

        # 使用资产的API端点下载，需要设置Accept头为application/octet-stream
        headers = settings.REPO_GITHUB_HEADERS(repo=user_repo).copy()
        headers["Accept"] = "application/octet-stream"
        res = await self.__async_request_with_fallback(download_url,
                                                       headers=headers,
                                                       is_api=True)
        if res is None or res.status_code != 200:
            return False, f"下载资产失败：{res.status_code if res else '连接失败'}"

        try:
            with zipfile.ZipFile(io.BytesIO(res.content)) as zf:
                infos = zf.infolist()
                if not infos:
                    return False, "压缩包内容为空"
                dest_base = Path(settings.ROOT_PATH) / "app" / "plugins" / pid.lower()
                targets = self.__iter_release_zip_targets(zf, dest_base)
                wrote_any = False
                for info, dest_path, is_dir in targets:
                    async_dest_path = AsyncPath(dest_path)
                    if is_dir:
                        await async_dest_path.mkdir(parents=True, exist_ok=True)
                        continue
                    await async_dest_path.parent.mkdir(parents=True, exist_ok=True)
                    data = await asyncio.to_thread(
                        self.__read_release_zip_member,
                        zf,
                        info,
                    )
                    async with aiofiles.open(dest_path, 'wb') as dst:
                        await dst.write(data)
                    wrote_any = True
                if not wrote_any:
                    return False, "压缩包中无可写入文件"
            return True, ""
        except Exception as e:
            logger.error(f"解压 Release 压缩包失败：{e}")
            return False, f"解压 Release 压缩包失败：{e}"

    @staticmethod
    def __read_release_zip_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
        """在线程池读取并解压单个 Release 文件，避免阻塞事件循环。"""
        with zf.open(info, "r") as source:
            return source.read()


# 公开 Release 查询的缓存管理统一指向仓库级分页缓存。
PluginHelper.get_plugin_release_versions.cache_clear = PluginHelper._get_plugin_repo_releases.cache_clear
PluginHelper.get_plugin_release_versions.cache_region = PluginHelper._get_plugin_repo_releases.cache_region
PluginHelper.async_get_plugin_release_versions.cache_clear = PluginHelper._async_get_plugin_repo_releases.cache_clear
PluginHelper.async_get_plugin_release_versions.cache_region = PluginHelper._async_get_plugin_repo_releases.cache_region
