Refactor plugin dependency cache and check (#1140)

backend/plugin/requirements.py

@@ -1,5 +1,3 @@
-import hashlib
-import os
 import site
 import subprocess
 import sys
@@ -8,15 +6,15 @@ from importlib import invalidate_caches
 from importlib.metadata import PackageNotFoundError, distribution
 from pathlib import Path
 
+from packaging.markers import default_environment
 from packaging.requirements import Requirement
+from packaging.utils import canonicalize_name
 from starlette.concurrency import run_in_threadpool
 
 from backend.core.conf import settings
 from backend.core.path_conf import PLUGIN_DIR
-from backend.database.redis import RedisCli
 from backend.plugin.core import get_plugins
 from backend.plugin.errors import PluginInstallError
-from backend.utils.async_helper import run_await
 
 
 def _is_in_virtualenv() -> bool:
@@ -24,6 +22,55 @@ def _is_in_virtualenv() -> bool:
     return hasattr(sys, 'real_prefix') or (hasattr(sys, 'base_prefix') and sys.base_prefix != sys.prefix)
 
 
+def _requirements_installed(requirements_file: Path) -> bool:  # noqa: C901
+    """检查 requirements 及其 extras 子依赖是否已安装"""
+    requirements = []
+    for line in requirements_file.read_text(encoding='utf-8').splitlines():
+        line = line.strip()
+        if not line or line.startswith('#'):
+            continue
+        try:
+            requirements.append(Requirement(line))
+        except Exception as e:
+            raise PluginInstallError(f'依赖 {line} 格式错误: {e!s}') from e
+
+    environment = default_environment()
+    visited = set()
+
+    def requirement_satisfied(requirement: Requirement, active_extras: frozenset[str] = frozenset({''})) -> bool:
+        if requirement.marker and not any(
+            requirement.marker.evaluate(environment={**environment, 'extra': extra}) for extra in active_extras
+        ):
+            return True
+
+        try:
+            dist = distribution(requirement.name)
+        except PackageNotFoundError:
+            return False
+
+        if requirement.specifier and not requirement.specifier.contains(dist.version, prereleases=True):
+            return False
+
+        requested_extras = tuple(sorted(requirement.extras))
+        state = (canonicalize_name(requirement.name), requested_extras)
+        if state in visited:
+            return True
+        visited.add(state)
+
+        child_active_extras = frozenset(requirement.extras) or frozenset({''})
+        for dependency_line in dist.requires or []:
+            try:
+                dependency = Requirement(dependency_line)
+            except Exception as e:
+                raise PluginInstallError(f'依赖元数据 {dependency_line} 格式错误: {e!s}') from e
+            if not requirement_satisfied(dependency, child_active_extras):
+                return False
+
+        return True
+
+    return all(requirement_satisfied(requirement) for requirement in requirements)
+
+
 def install_requirements(plugin: str | None) -> None:  # noqa: C901
     """
     安装插件依赖
@@ -33,73 +80,35 @@ def install_requirements(plugin: str | None) -> None:  # noqa: C901
     """
     plugins = [plugin] if plugin else get_plugins()
 
-    # 使用独立连接
-    current_redis_client = RedisCli()
-    run_await(current_redis_client.init)()
+    for plugin in plugins:
+        requirements_file = PLUGIN_DIR / plugin / 'requirements.txt'
+        if not requirements_file.exists() or _requirements_installed(requirements_file):
+            continue
 
-    try:
-        for plugin in plugins:
-            requirements_file = PLUGIN_DIR / plugin / 'requirements.txt'
-            hash_key = f'{settings.PLUGIN_REDIS_PREFIX}:requirements_hash:{plugin}'
-            cached_hash = run_await(current_redis_client.get)(hash_key)
+        pip_install = ['uv', 'pip', 'install', '-r', requirements_file]
+        if not _is_in_virtualenv():
+            pip_install.append('--system')
+        if settings.PLUGIN_PIP_CHINA:
+            pip_install.extend(['-i', settings.PLUGIN_PIP_INDEX_URL])
 
-            if not os.path.exists(requirements_file):
-                run_await(current_redis_client.delete)(hash_key)
+        max_retries = settings.PLUGIN_PIP_MAX_RETRY
+        for attempt in range(max_retries):
+            try:
+                subprocess.check_call(pip_install)
+                invalidate_caches()
+                for site_dir in site.getsitepackages():
+                    if site_dir.endswith('site-packages'):
+                        site.addsitedir(site_dir)
+                break
+            except subprocess.TimeoutExpired:
+                if attempt == max_retries - 1:
+                    raise PluginInstallError(f'插件 {plugin} 依赖安装超时')
                 continue
-
-            missing_dependencies = False
-            for line in Path(requirements_file).read_text().splitlines():
-                line = line.strip()
-                if not line or line.startswith('#'):
-                    continue
-                try:
-                    req = Requirement(line)
-                    dependency = req.name.lower()
-                except Exception as e:
-                    raise PluginInstallError(f'插件 {plugin} 依赖 {line} 格式错误: {e!s}') from e
-
-                try:
-                    dist = distribution(dependency)
-                except PackageNotFoundError:
-                    missing_dependencies = True
-                    break
-
-                if req.specifier and not req.specifier.contains(dist.version, prereleases=True):
-                    missing_dependencies = True
-                    break
-
-            current_hash = hashlib.sha256(Path(requirements_file).read_bytes()).hexdigest()
-            if cached_hash == current_hash and not missing_dependencies:
+            except subprocess.CalledProcessError as e:
+                if attempt == max_retries - 1:
+                    raise PluginInstallError(f'插件 {plugin} 依赖安装失败：{e}') from e
                 continue
 
-            pip_install = ['uv', 'pip', 'install', '-r', requirements_file]
-            if not _is_in_virtualenv():
-                pip_install.append('--system')
-            if settings.PLUGIN_PIP_CHINA:
-                pip_install.extend(['-i', settings.PLUGIN_PIP_INDEX_URL])
-
-            max_retries = settings.PLUGIN_PIP_MAX_RETRY
-            for attempt in range(max_retries):
-                try:
-                    subprocess.check_call(pip_install)
-                    # 刷新依赖包缓存
-                    invalidate_caches()
-                    for site_dir in site.getsitepackages():
-                        if site_dir.endswith('site-packages'):
-                            site.addsitedir(site_dir)
-                    run_await(current_redis_client.set)(hash_key, current_hash)
-                    break
-                except subprocess.TimeoutExpired:
-                    if attempt == max_retries - 1:
-                        raise PluginInstallError(f'插件 {plugin} 依赖安装超时')
-                    continue
-                except subprocess.CalledProcessError as e:
-                    if attempt == max_retries - 1:
-                        raise PluginInstallError(f'插件 {plugin} 依赖安装失败：{e}') from e
-                    continue
-    finally:
-        run_await(current_redis_client.aclose)()
-
 
 def uninstall_requirements(plugin: str) -> None:
     """
@@ -108,23 +117,17 @@ def uninstall_requirements(plugin: str) -> None:
     :param plugin: 插件名称
     :return:
     """
-    # 使用独立连接
-    current_redis_client = RedisCli()
-    run_await(current_redis_client.init)()
+    requirements_file = PLUGIN_DIR / plugin / 'requirements.txt'
+    if not requirements_file.exists():
+        return
 
     try:
-        run_await(current_redis_client.delete)(f'{settings.PLUGIN_REDIS_PREFIX}:requirements_hash:{plugin}')
-        requirements_file = PLUGIN_DIR / plugin / 'requirements.txt'
-        if os.path.exists(requirements_file):
-            try:
-                pip_uninstall = ['uv', 'pip', 'uninstall', '-r', str(requirements_file)]
-                if not _is_in_virtualenv():
-                    pip_uninstall.append('--system')
-                subprocess.check_call(pip_uninstall, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-            except subprocess.CalledProcessError as e:
-                raise PluginInstallError(f'插件 {plugin} 依赖卸载失败：{e}') from e
-    finally:
-        run_await(current_redis_client.aclose)()
+        pip_uninstall = ['uv', 'pip', 'uninstall', '-r', str(requirements_file)]
+        if not _is_in_virtualenv():
+            pip_uninstall.append('--system')
+        subprocess.check_call(pip_uninstall, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    except subprocess.CalledProcessError as e:
+        raise PluginInstallError(f'插件 {plugin} 依赖卸载失败：{e}') from e
 
 
 async def install_requirements_async(plugin: str | None = None) -> None: