MoviePilot/app/runtime/extensions/plugin/registry.py

"""
插件类与运行实例注册表。
"""

from typing import Any, Dict, Optional

from app.schemas.plugin import PluginRuntimeStatus


class PluginRegistry:
    """
    集中持有插件类和运行实例，并为读取方提供稳定快照。
    """

    def __init__(self) -> None:
        """
        创建彼此独立但生命周期一致的类表和实例表。
        """
        self._classes: Dict[str, Any] = {}
        self._running: Dict[str, Any] = {}
        self._runtime_statuses: Dict[str, PluginRuntimeStatus] = {}
        self._restart_required_plugins: Dict[str, tuple[str, ...]] = {}
        # 加载时使 free-threaded 进程回退到 GIL 的插件，只在当前进程内有效，重启后重新观察
        self._gil_fallback_plugins: Dict[str, bool] = {}
        self._settling = False
        self._generation = 0

    @property
    def classes(self) -> Dict[str, Any]:
        """
        返回兼容旧调用方可变访问语义的插件类表。
        """
        return self._classes

    @property
    def running(self) -> Dict[str, Any]:
        """
        返回兼容旧调用方可变访问语义的运行实例表。
        """
        return self._running

    def has_class(self, plugin_id: str) -> bool:
        """
        判断插件类是否已经登记。
        """
        return plugin_id in self._classes

    def plugin_class(self, plugin_id: str) -> Optional[Any]:
        """
        读取指定插件类，未登记时返回空。
        """
        return self._classes.get(plugin_id)

    def instance(self, plugin_id: str) -> Optional[Any]:
        """
        读取指定运行实例，未运行时返回空。
        """
        return self._running.get(plugin_id)

    def plugin_ids(self) -> list[str]:
        """
        返回保持登记顺序的插件类 ID 快照。
        """
        return list(self.classes_snapshot())

    def running_ids(self) -> list[str]:
        """
        返回保持登记顺序的运行实例 ID 快照。
        """
        return list(self.running_snapshot())

    def classes_snapshot(self) -> Dict[str, Any]:
        """
        复制插件类表，避免插件重载期间迭代失效。
        """
        return dict(self._classes)

    def running_snapshot(self) -> Dict[str, Any]:
        """
        复制运行实例表，避免插件重载期间迭代失效。
        """
        return dict(self._running)

    def set_runtime_status(
        self,
        plugin_id: str,
        status: PluginRuntimeStatus,
    ) -> None:
        """
        记录插件当前状态，并在实际变化时推进前端刷新代次。
        """
        if self._runtime_statuses.get(plugin_id) == status:
            return
        self._runtime_statuses[plugin_id] = status
        self._generation += 1

    def runtime_status(self, plugin_id: str) -> Optional[PluginRuntimeStatus]:
        """
        读取指定插件状态。
        """
        return self._runtime_statuses.get(plugin_id)

    def runtime_status_snapshot(self) -> Dict[str, PluginRuntimeStatus]:
        """
        复制插件状态表，避免后台加载期间迭代失效。
        """
        return dict(self._runtime_statuses)

    def mark_restart_required(
        self,
        plugin_id: str,
        distributions: tuple[str, ...],
    ) -> None:
        """
        记录当前进程仍持有旧原生载荷的插件和发行包。
        """
        normalized = tuple(sorted(set(distributions)))
        previous = self._restart_required_plugins.get(plugin_id, ())
        merged = tuple(sorted(set(previous).union(normalized)))
        if previous == merged:
            return
        self._restart_required_plugins[plugin_id] = merged
        self._generation += 1

    def restart_required_snapshot(self) -> Dict[str, tuple[str, ...]]:
        """
        返回重启后才能完整激活的插件及原生发行包。
        """
        return dict(self._restart_required_plugins)

    def mark_gil_fallback(self, plugin_id: str) -> None:
        """
        记录插件加载使 free-threaded 运行时回退到 GIL。
        """
        if plugin_id in self._gil_fallback_plugins:
            return
        self._gil_fallback_plugins[plugin_id] = True
        self._generation += 1

    def mark_enabled_state_changed(self, plugin_id: str) -> None:
        """
        记录插件配置切换了启用状态，推进前端刷新代次。

        启停只改实例自身的启用位，不经过运行状态表，但侧栏、仪表盘和助手形象等
        前端投影都按启用位过滤，读取方需要据此重新拉取。
        """
        del plugin_id
        self._generation += 1

    def gil_fallback_snapshot(self) -> list[str]:
        """
        返回当前进程内导致 GIL 回退的插件 ID，保持记录顺序。
        """
        return list(self._gil_fallback_plugins)

    def set_settling(self, settling: bool) -> None:
        """
        标记启动后的插件源码与依赖收敛任务是否仍在执行。
        """
        if self._settling == settling:
            return
        self._settling = settling
        self._generation += 1

    @property
    def settling(self) -> bool:
        """
        返回插件后台收敛任务是否仍在执行。
        """
        return self._settling

    @property
    def generation(self) -> int:
        """
        返回状态变化代次，供读取方识别刷新边界。
        """
        return self._generation

    def remove(self, plugin_id: str) -> None:
        """
        同时移除指定插件类、运行实例和状态。
        """
        self._classes.pop(plugin_id, None)
        self._running.pop(plugin_id, None)
        status_removed = self._runtime_statuses.pop(plugin_id, None) is not None
        restart_requirement_removed = (
            self._restart_required_plugins.pop(plugin_id, None) is not None
        )
        gil_fallback_removed = (
            self._gil_fallback_plugins.pop(plugin_id, None) is not None
        )
        if status_removed or restart_requirement_removed or gil_fallback_removed:
            self._generation += 1

    def clear(self) -> None:
        """
        原地清空注册表，保持外部持有的兼容字典引用有效。
        """
        self._classes.clear()
        self._running.clear()
        had_runtime_state = bool(
            self._runtime_statuses
            or self._restart_required_plugins
            or self._gil_fallback_plugins
        )
        self._runtime_statuses.clear()
        self._restart_required_plugins.clear()
        self._gil_fallback_plugins.clear()
        if had_runtime_state:
            self._generation += 1