MoviePilot/app/adapters/web/plugin/routes.py

"""
FastAPI 动态插件路由适配器。
"""

import asyncio
from concurrent.futures import Future
from concurrent.futures import TimeoutError as FutureTimeoutError
from threading import Lock
from typing import Any, Callable, Optional

from fastapi import Depends, FastAPI
from fastapi.dependencies import models as fastapi_dependency_models
from fastapi.routing import APIRoute


def _clear_fastapi_callable_caches() -> None:
    """
    清空 FastAPI 按可调用对象身份缓存的分类结果。

    FastAPI 在模块级 ``lru_cache`` 中以强引用保存每个端点与依赖函数（容量 4096），
    插件路由移除后旧端点函数仍被缓存持有，进而经 ``__globals__`` 留住整个旧插件模块，
    每次重载都多留一份。缓存只记录“是否生成器/协程”这类分类结果，清空后下次请求按需
    重算。缓存属于 FastAPI 私有实现，按属性探测，版本不再提供时自然跳过。
    """
    for value in vars(fastapi_dependency_models).values():
        cache_clear = getattr(value, "cache_clear", None)
        if callable(cache_clear):
            cache_clear()


class FastAPIDynamicRouteRegistry:
    """
    在 FastAPI 上注册插件自由响应路由，并维护 OpenAPI 缓存。
    """

    _dispatch_admission_timeout = 5.0

    def __init__(
        self,
        app: FastAPI,
        plugin_ids: Callable[[], list[str]],
        plugin_apis: Callable[[str], list[dict[str, Any]]],
        verify_token: Callable[..., Any],
        verify_apikey: Callable[..., Any],
        prefix: str,
        log: Any,
        event_loop: Callable[[], asyncio.AbstractEventLoop | None] | None = None,
    ) -> None:
        """
        注入应用、插件查询入口、认证依赖和日志端口。

        :param app: 持有动态路由表的 FastAPI 应用
        :param plugin_ids: 返回当前插件标识列表的回调
        :param plugin_apis: 按插件标识返回路由声明的回调
        :param verify_token: 路由选择 bear 认证时使用的令牌校验依赖
        :param verify_apikey: 默认 API 密钥校验依赖
        :param prefix: 插件路由的共同路径前缀
        :param log: 记录路由注册与移除结果的日志对象
        :param event_loop: 返回主事件循环的回调；省略时直接在调用线程更新路由
        """
        self._app = app
        self._plugin_ids = plugin_ids
        self._plugin_apis = plugin_apis
        self._verify_token = verify_token
        self._verify_apikey = verify_apikey
        self._prefix = prefix
        self._logger = log
        self._event_loop = event_loop

    def update(self, plugin_id: Optional[str], action: str) -> None:
        """
        按插件生命周期新增或移除动态路由，并使 OpenAPI 缓存失效。

        配置事件循环后在该循环执行更新；跨线程调用会阻塞到更新结束。
        等待准入超过 5 秒且更新尚未开始时放弃请求；一旦开始则等待确定结果。

        :param plugin_id: 要更新的插件标识；为空时处理当前全部插件
        :param action: add 先移除旧路由再注册当前声明，remove 只移除旧路由
        :raises ValueError: 动作不是 add 或 remove
        :raises RuntimeError: 指定的事件循环不可用，或未在准入期限内接收更新
        """
        if self._event_loop is None:
            self._update(plugin_id, action)
            return
        target_loop = self._event_loop()
        if (
            target_loop is None
            or not target_loop.is_running()
            or target_loop.is_closed()
        ):
            raise RuntimeError("主事件循环未运行，无法更新插件动态路由")
        try:
            current_loop = asyncio.get_running_loop()
        except RuntimeError:
            current_loop = None
        if current_loop is target_loop:
            self._update(plugin_id, action)
            return

        completed: Future[None] = Future()
        dispatch_lock = Lock()
        dispatch_started = False
        dispatch_abandoned = False

        def apply_update() -> None:
            """
            在目标事件循环的单个回调中完成路由表与 OpenAPI 缓存更新。
            """
            nonlocal dispatch_started
            with dispatch_lock:
                if dispatch_abandoned:
                    return
                dispatch_started = True
            try:
                self._update(plugin_id, action)
            except BaseException as error:
                completed.set_exception(error)
            else:
                completed.set_result(None)

        target_loop.call_soon_threadsafe(apply_update)
        try:
            completed.result(timeout=self._dispatch_admission_timeout)
        except FutureTimeoutError as error:
            with dispatch_lock:
                if not dispatch_started:
                    dispatch_abandoned = True
                    raise RuntimeError(
                        "主事件循环未及时接收插件动态路由更新"
                    ) from error
            # 回调一旦开始便不可撤销，等待确定终态以免失败回滚后发生迟到写入。
            completed.result()

    def _update(self, plugin_id: Optional[str], action: str) -> None:
        """
        执行不可中断的路由表与 OpenAPI 投影更新。
        """
        if action not in {"add", "remove"}:
            raise ValueError("Action must be 'add' or 'remove'")

        modified = False
        removed = False
        plugin_ids = [plugin_id] if plugin_id else self._plugin_ids()
        for current_id in plugin_ids:
            if self.remove(current_id):
                modified = True
                removed = True
            if action != "add":
                continue
            for source_api in self._plugin_apis(current_id):
                api = dict(source_api)
                api["dependencies"] = list(source_api.get("dependencies") or ())
                api_path = f"{self._prefix}{api.get('path', '')}"
                try:
                    api["path"] = api_path
                    allow_anonymous = api.pop("allow_anonymous", False)
                    auth_mode = api.pop("auth", "apikey")
                    dependencies = api.setdefault("dependencies", [])
                    if not allow_anonymous:
                        authentication = Depends(
                            self._verify_token if auth_mode == "bear" else self._verify_apikey
                        )
                        if authentication not in dependencies:
                            dependencies.append(authentication)
                    # 插件 API 自行决定响应结构，不使用宿主统一 Response 路由。
                    api.setdefault("route_class_override", APIRoute)
                    self._app.router.add_api_route(**api, tags=["plugin"])
                    modified = True
                    self._logger.debug(f"Added plugin route: {api_path}")
                except Exception as error:
                    self._logger.error(
                        f"Error adding plugin route {api_path}: {str(error)}"
                    )
        if modified:
            # 文档路由由 app.adapters.web.docs 一次性注册，这里只让已缓存的文档按新路由表重建。
            self._app.openapi_schema = None
        if removed:
            # 旧路由已不在路由表中，此后不会再有请求把旧端点写回缓存
            _clear_fastapi_callable_caches()

    def remove(self, plugin_id: str) -> bool:
        """
        移除指定插件前缀下的全部动态路由。

        直接修改路由表，不清理 OpenAPI 或函数分类缓存；完整生命周期更新应调用 update。

        :param plugin_id: 要移除的插件标识；为空时不操作
        :return: 至少成功移除一条路由时返回 True；无匹配或全部移除失败时返回 False
        """
        if not plugin_id:
            return False
        root_path = f"{self._prefix}/{plugin_id}"
        routes = [
            route for route in self._app.routes
            if (path := self._route_path(route)) is not None
            and (path == root_path or path.startswith(root_path + "/"))
        ]
        removed = False
        for route in routes:
            try:
                self._app.routes.remove(route)
                removed = True
                self._logger.debug(f"Removed plugin route: {self._route_path(route)}")
            except Exception as error:
                self._logger.error(
                    f"Error removing plugin route {self._route_path(route)}: {str(error)}"
                )
        return removed

    @staticmethod
    def _route_path(route: Any) -> Optional[str]:
        """
        返回公开路由路径，跳过 FastAPI 内部的无路径 include 包装器。
        """
        path = getattr(route, "path", None)
        return path if isinstance(path, str) else None