MoviePilot/app/runtime/extensions/plugin_manager.py

import asyncio
import posixpath
import threading
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Dict, List, Optional, Type, Union, Callable, Tuple

from watchfiles import watch

from app.schemas.plugin import Plugin as _SchemaPlugin
from app.schemas.plugin import PluginDashboard as _SchemaPluginDashboard
from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
from app.foundation.crypto import RSAUtils
from app.foundation.singleton import Singleton
from app.foundation.version import compare_version
from app.runtime.log import logger
from app.runtime.observability import observe_compat_facade
from app.runtime.settings import RuntimeSettingsCompat

settings = RuntimeSettingsCompat()
from app.runtime.events import EventHandlerBinding, eventmanager
from app.runtime.reload import ConfigReloadMixin
from app.runtime.extensions.plugin.loader import PluginLoader
from app.runtime.extensions.plugin.lifecycle import PluginLifecycle
from app.runtime.extensions.plugin.metadata import PluginMetadataMapper
from app.runtime.extensions.plugin.monitor import (
    PluginChangeMonitor,
    PluginMonitorController,
)
from app.runtime.extensions.plugin.projection import PluginProjection
from app.runtime.extensions.plugin.registry import PluginRegistry
from app.runtime.extensions.plugin.storage import get_plugin_storage
from app.runtime.extensions.plugin.system import get_plugin_system
from app.runtime.extensions.plugin.tools import PluginToolCatalog
from app.runtime.extensions.plugin.sync import (
    LocalPluginSyncService,
    PluginSyncService,
)
from app.runtime.extensions.plugin.clone import PluginCloneService
from app.runtime.extensions.plugin.access import PluginAccessPolicy
from app.runtime.extensions.plugin.catalog import PluginCatalogFacade
from app.runtime.extensions.plugin.paths import PluginPathResolver
from app.runtime.extensions.plugin.dependency import (
    PluginDependencyClassification,
    PluginDependencyInstallResult,
    PluginDependencyService,
)
from app.runtime.extensions.plugin.storage import PluginConfigStore, PluginInstanceStore
from app.schemas.types import EventType, SystemConfigKey

LegacyDiagnosticsConfigurator = Callable[..., None]
LegacyImportScanner = Callable[..., None]
LegacyPluginImportPreparer = Callable[..., None]
PluginInstallReporter = Callable[..., None]
SiteAuthLevelProvider = Callable[[], int]
PluginCatalogFactory = Callable[["PluginManager"], Any]


def _ignore_legacy_diagnostics(**_kwargs) -> None:
    """在启动组合根尚未注入兼容服务时保持插件加载可用。"""


def _ignore_plugin_resource_imports(**_kwargs) -> None:
    """未进入应用启动组合时不主动创建进程级宿主资源。"""


def _unavailable_site_auth_level() -> int:
    """站点能力尚未装配时返回未认证等级。"""
    return 0


def _unavailable_plugin_catalog_factory(_manager: "PluginManager") -> Any:
    """在启动组合根尚未装配目录用例时拒绝隐式跨层构造。"""
    raise RuntimeError("插件目录应用服务尚未由启动组合根装配")


_legacy_diagnostics_configurator: LegacyDiagnosticsConfigurator = (
    _ignore_legacy_diagnostics
)
_legacy_import_scanner: LegacyImportScanner = _ignore_legacy_diagnostics
_legacy_plugin_import_preparer: LegacyPluginImportPreparer = (
    _ignore_plugin_resource_imports
)
_plugin_install_reporter: PluginInstallReporter = _ignore_legacy_diagnostics
_site_auth_level_provider: SiteAuthLevelProvider = _unavailable_site_auth_level
_plugin_catalog_factory: PluginCatalogFactory = _unavailable_plugin_catalog_factory


def configure_plugin_legacy_import_services(
    *,
    diagnostics_configurator: LegacyDiagnosticsConfigurator,
    import_scanner: LegacyImportScanner,
) -> None:
    """由启动组合根注入插件旧导入诊断服务，避免扩展层反向依赖兼容层。"""
    global _legacy_diagnostics_configurator, _legacy_import_scanner
    _legacy_diagnostics_configurator = diagnostics_configurator
    _legacy_import_scanner = import_scanner


def configure_plugin_resource_import_preparer(
    preparer: LegacyPluginImportPreparer,
) -> None:
    """注入旧插件导入前的宿主资源准备器。"""
    global _legacy_plugin_import_preparer
    _legacy_plugin_import_preparer = preparer


def configure_plugin_install_reporter(reporter: PluginInstallReporter) -> None:
    """由启动组合根注入插件安装上报器，避免扩展层依赖远程服务。"""
    global _plugin_install_reporter
    _plugin_install_reporter = reporter


def configure_site_auth_level_provider(provider: SiteAuthLevelProvider) -> None:
    """由启动组合根注入站点认证等级，避免扩展运行时依赖应用服务。"""
    global _site_auth_level_provider
    _site_auth_level_provider = provider


def configure_plugin_catalog_factory(factory: PluginCatalogFactory) -> None:
    """由启动组合根注入插件目录应用服务工厂，消除 Runtime 反向依赖。"""
    global _plugin_catalog_factory
    _plugin_catalog_factory = factory


@observe_compat_facade("PluginManager")
class PluginManager(ConfigReloadMixin, metaclass=Singleton):
    """插件管理器"""
    CONFIG_WATCH = {"DEV", "PLUGIN_AUTO_RELOAD", "PLUGIN_LOCAL_REPO_PATHS"}
    AGENT_TOOLS_BUILD_MAX_ATTEMPTS = 3

    def __init__(self):
        """初始化插件注册表、缓存和开发模式监控状态。"""
        self._plugin_registry = PluginRegistry()
        # 旧属性继续引用注册表拥有的可变字典，保持插件和测试的访问身份。
        self._plugins = self._plugin_registry.classes
        self._running_plugins = self._plugin_registry.running
        self._plugin_instance_store = PluginInstanceStore(
            storage=lambda: get_plugin_storage(),
        )
        # 配置Key
        self._config_key: str = "plugin.%s"
        self._plugin_config_store = PluginConfigStore(
            storage=lambda: get_plugin_storage(),
            plugin_exists=lambda pid: bool(self._plugins.get(pid)),
            key_prefix=self._config_key,
        )
        self._plugin_access = PluginAccessPolicy(
            auth_level=lambda: _site_auth_level_provider(),
            verify_keys=RSAUtils.verify_rsa_keys,
            log=logger,
        )
        self._plugin_catalog_view = PluginCatalogFacade(
            classes=lambda: self._plugins,
            running=lambda: self._running_plugins,
            storage=lambda: get_plugin_storage(),
            system=get_plugin_system,
            market_catalog=lambda: self._plugin_catalog(),
            market_loader=lambda market, package_version=None, force=False: (
                self.get_plugins_from_market(market, package_version, force)
            ),
            async_market_loader=lambda market, package_version=None, force=False: (
                self.async_get_plugins_from_market(market, package_version, force)
            ),
            map_plugin=lambda **kwargs: self._process_plugin_info(**kwargs),
            auth_checker=lambda **kwargs: self.__set_and_check_auth_level(**kwargs),
            plugin_attr=lambda pid, attr: self.get_plugin_attr(pid, attr),
            plugin_instance=self.get_plugin_instance,
            plugin_instances=self.get_plugin_instances,
            runtime_status=self._plugin_registry.runtime_status,
            log=logger,
        )
        # 本地插件同步写入运行目录后的短时忽略窗口
        self._recent_local_sync: Dict[str, float] = {}
        self._monitor_suppression_lock = threading.Lock()
        self._suppressed_monitor_plugins: Dict[str, int] = {}
        self._plugin_paths = PluginPathResolver(
            runtime_root=settings.ROOT_PATH / "app" / "plugins",
            running=lambda: self._running_plugins,
            system=get_plugin_system,
            strict_system_version=lambda: not settings.DEV,
            log=logger,
        )
        self._local_plugin_sync = LocalPluginSyncService(
            installed_plugins=lambda: get_plugin_storage().read(
                SystemConfigKey.UserInstalledPlugins
            ) or [],
            candidate=lambda pid: get_plugin_system().local_candidate(pid),
            system=get_plugin_system,
            recent_sync=self._recent_local_sync,
            log=logger,
        )
        self._plugin_monitor = PluginMonitorController(
            runner=self._run_file_watcher,
            log=logger,
        )
        self._plugin_dependencies = PluginDependencyService(
            system=get_plugin_system,
            log=logger,
        )
        self._plugin_loader = PluginLoader(
            plugins_root=settings.ROOT_PATH / "app" / "plugins",
            import_preparer=lambda **kwargs: _legacy_plugin_import_preparer(**kwargs),
            import_scanner=lambda **kwargs: _legacy_import_scanner(**kwargs),
            log=logger,
        )
        self._plugin_tool_catalog = PluginToolCatalog(
            max_attempts=self.AGENT_TOOLS_BUILD_MAX_ATTEMPTS
        )
        self._plugin_lifecycle = PluginLifecycle(
            classes=self._plugins,
            running=self._running_plugins,
            load_plugins=lambda pid, installed, check: self._load_runtime_plugins(
                pid,
                installed,
                check,
            ),
            installed_plugins=lambda: get_plugin_storage().read(
                SystemConfigKey.UserInstalledPlugins
            ) or [],
            plugin_config=self.get_plugin_config,
            auth_checker=lambda plugin: self.__set_and_check_auth_level(plugin=plugin),
            clear_modules=lambda pid: self._clear_plugin_modules(pid),
            clear_tools=self.clear_plugin_agent_tools_cache,
            enable_events=eventmanager.enable_event_handler,
            disable_events=eventmanager.disable_event_handler,
            runtime_status_writer=self._plugin_registry.set_runtime_status,
            log=logger,
            event_sender=eventmanager.send_event,
        )
        self._plugin_metadata = PluginMetadataMapper(
            plugin_instance=self._plugin_registry.instance,
            plugin_class=self._plugin_registry.plugin_class,
            annotate_system_version=lambda info: get_plugin_system().annotate_system_version(
                info
            ),
            is_package_compatible=lambda info, version: get_plugin_system().is_package_compatible(
                info,
                version,
            ),
            auth_checker=lambda plugin, source: self.__set_and_check_auth_level(
                plugin,
                source,
            ),
            version_compare=compare_version,
            log=logger,
        )
        self._plugin_sync = PluginSyncService(
            frozen=lambda: get_plugin_system().is_frozen(),
            installed_plugins=lambda: get_plugin_storage().read(
                SystemConfigKey.UserInstalledPlugins
            ) or [],
            online_plugins=lambda: self.get_online_plugins(),
            local_plugins=lambda: self.get_local_repo_plugins(),
            merge_plugins=lambda higher, base, _markets: self.process_plugins_list(
                higher,
                base,
            ),
            plugin_exists=lambda plugin_id, version: self.is_plugin_exists(
                plugin_id,
                version,
            ),
            install=lambda plugin_id, repo_url, force: get_plugin_system().package.install(
                plugin_id=plugin_id,
                repo_url=repo_url,
                force_install=force,
            ),
            report=lambda **kwargs: _plugin_install_reporter(**kwargs),
            log=logger,
        )
        self._plugin_clone = PluginCloneService(
            plugin_class=self._plugin_registry.plugin_class,
            plugin_exists=lambda plugin_id: self.is_plugin_exists(plugin_id),
            source_plugin_id=self.get_plugin_source_id,
            save_instance=self._plugin_instance_store.save,
            delete_instance=self._plugin_instance_store.delete,
            read_config=self.get_plugin_config,
            save_config=lambda plugin_id, config: self.save_plugin_config(
                plugin_id,
                config,
                force=True,
            ),
            delete_config=lambda plugin_id: self.delete_plugin_config(
                plugin_id,
                force=True,
            ),
            reload_plugin=self.reload_plugin,
            remove_plugin=self.remove_plugin,
            log=logger,
        )
        # 事件总线只通过通用解析器访问运行中的插件实例。
        eventmanager.register_handler_instance_resolver(
            "plugins",
            self.resolve_event_handler_instance,
        )
    def resolve_event_handler_instance(
            self,
            owner_class: Type[Any],
    ) -> Optional[EventHandlerBinding]:
        """为插件声明的事件方法解析当前运行实例。"""
        plugin_id = owner_class.__name__
        # 旧测试与部分扩展会替换私有映射来构造隔离运行态，解析器继续尊重该接缝。
        if plugin_id not in self._plugins:
            return None
        plugin = self._running_plugins.get(plugin_id)
        owner_name = plugin_id
        if plugin and callable(getattr(plugin, "get_name", None)):
            owner_name = plugin.get_name()
        return EventHandlerBinding(
            instance=plugin,
            owner_name=owner_name,
            run_sync_in_threadpool=True,
        )

    def init_config(self):
        """按最新系统配置完整重启插件。"""
        # 停止已有插件
        self.stop()
        classification = self.classify_plugins()
        self.apply_plugin_dependency_classification(classification)
        for plugin_id in classification.ready:
            self.start(plugin_id)

    def start(self, pid: Optional[str] = None) -> Dict[str, PluginRuntimeStatus]:
        """
        启动加载插件
        :param pid: 插件ID，为空加载所有插件
        """

        _legacy_diagnostics_configurator(enabled=settings.DEBUG, emitter=logger.warning)
        return self._plugin_lifecycle.start(pid)

    def init_plugin(self, plugin_id: str, conf: dict):
        """
        初始化插件
        :param plugin_id: 插件ID
        :param conf: 插件配置
        """
        self._plugin_lifecycle.initialize(plugin_id, conf)

    def clear_plugin_agent_tools_cache(self) -> None:
        """
        清空插件智能体工具注册表缓存。
        """
        self._plugin_tool_catalog.clear()

    def get_plugin_agent_tools_revision(self) -> int:
        """
        获取插件智能体工具注册表版本号。
        """
        return self._plugin_tool_catalog.revision

    @property
    def _plugin_agent_tools_revision(self) -> int:
        """兼容读取旧私有字段，实际版本由独立工具目录持有。"""
        return self._plugin_tool_catalog.revision

    def stop(self, pid: Optional[str] = None):
        """
        停止插件服务
        :param pid: 插件ID，为空停止所有插件
        """
        self._plugin_lifecycle.stop(pid)

    @staticmethod
    def _load_selective_plugins(pid: Optional[str], installed_plugins: List[str],
                                check_module_func: Callable) -> List[Any]:
        """
        选择性加载插件，只import符合条件的插件
        :param pid: 指定插件ID，为空则加载所有已安装插件
        :param installed_plugins: 已安装插件列表
        :param check_module_func: 模块检查函数
        :return: 插件类列表
        """
        return PluginLoader(
            plugins_root=settings.ROOT_PATH / "app" / "plugins",
            import_preparer=lambda **kwargs: _legacy_plugin_import_preparer(**kwargs),
            import_scanner=lambda **kwargs: _legacy_import_scanner(**kwargs),
            log=logger,
        ).load(pid, installed_plugins, check_module_func)

    def _load_runtime_plugins(
        self,
        pid: Optional[str],
        installed_plugins: List[str],
        check_module_func: Callable[[Any], bool],
    ) -> List[Any]:
        """加载物理插件或虚拟实例，并保留旧选择性加载方法的合同。"""
        if pid:
            instance = self._plugin_instance_store.get(pid)
            if instance:
                return self._plugin_loader.load_instance(instance, check_module_func)
            return self._plugin_loader.load(pid, installed_plugins, check_module_func)

        plugins = self._plugin_loader.load(
            None,
            installed_plugins,
            check_module_func,
        )
        for instance in self._plugin_instance_store.all().values():
            plugins.extend(
                self._plugin_loader.load_instance(instance, check_module_func)
            )
        return plugins

    @property
    def running_plugins(self) -> Dict[str, Any]:
        """
        获取运行态插件列表
        :return: 运行态插件列表
        """
        return self._plugin_registry.running

    @property
    def plugins(self) -> Dict[str, Any]:
        """
        获取插件列表
        :return: 插件列表
        """
        return self._plugin_registry.classes

    def on_config_changed(self):
        """在插件监控配置变化后重建文件监控。"""
        self.reload_monitor()

    def get_reload_name(self) -> str:
        """返回配置重载日志使用的功能名称。"""
        return "插件文件修改监测"

    def start_monitor(self):
        """按当前配置启动插件文件修改监测。"""
        if (
            not self.is_plugin_settling()
            and (settings.DEV or settings.PLUGIN_AUTO_RELOAD)
        ):
            self._plugin_monitor.start()

    def reload_monitor(self):
        """
        重新加载插件文件修改监测
        """
        self._plugin_monitor.reload(
            enabled=(
                not self.is_plugin_settling()
                and (settings.DEV or settings.PLUGIN_AUTO_RELOAD)
            )
        )

    def stop_monitor(self):
        """
        停止监测插件文件修改监测
        """
        self._plugin_monitor.stop()

    def _run_file_watcher(self):
        """
        运行 watchfiles 监视器的主循环。
        """
        PluginChangeMonitor(
            runtime_root=settings.ROOT_PATH / "app" / "plugins",
            local_roots=get_plugin_system().local_repo_paths,
            stop_event=self._plugin_monitor.stop_event,
            recent_sync=self._recent_local_sync,
            federated_change=self._get_federated_plugin_change,
            runtime_plugin=self._get_plugin_id_from_path,
            monitor_suppressed=self.is_plugin_monitor_suppressed,
            local_candidate=self._get_local_plugin_candidate_from_path,
            sync_local=self._sync_local_plugin_if_installed,
            reload_plugin=self.reload_plugin_tree,
            dependency_manifest_status=(
                get_plugin_system().dependency_manifest_status
            ),
            watch=watch,
            log=logger,
        ).run()

    def _get_federated_plugin_change(
        self,
        event_path: Path,
    ) -> Optional[Tuple[str, Optional[dict], bool]]:
        """
        识别运行态 Vue 插件声明目录内的构建产物变化。

        :return: 插件 ID、本地仓库候选和联邦入口是否完整；非联邦目录变化返回 None。
        """
        return self._plugin_paths.federated_change(event_path)

    def _get_plugin_id_from_path(self, event_path: Path) -> Optional[str]:
        """
        根据文件路径解析出插件的ID。
        :param event_path: 被修改文件的 Path 对象。
        :return: 插件ID字符串，如果不是有效插件文件则返回 None。
        """
        return self._plugin_paths.runtime_plugin(event_path)

    def _get_local_plugin_candidate_from_path(self, event_path: Path) -> Optional[dict]:
        """
        根据本地插件仓库路径解析具体插件候选，保留 plugins/plugins.v2 来源差异
        """
        return self._plugin_paths.local_candidate(event_path)

    def _sync_local_plugin_if_installed(self, pid: str, candidate: Optional[dict] = None) -> bool:
        """
        已安装本地插件源码变化时，同步到运行目录
        """
        return self._local_plugin_sync.sync(pid, candidate)

    @contextmanager
    def suppress_plugin_monitor(self, plugin_id: str):
        """在插件目录原子更新期间阻止文件监控抢先重载半成品。"""
        normalized_id = plugin_id.lower()
        with self._monitor_suppression_lock:
            self._suppressed_monitor_plugins[normalized_id] = (
                self._suppressed_monitor_plugins.get(normalized_id, 0) + 1
            )
        try:
            yield
        finally:
            with self._monitor_suppression_lock:
                count = self._suppressed_monitor_plugins.get(normalized_id, 0)
                if count <= 1:
                    self._suppressed_monitor_plugins.pop(normalized_id, None)
                else:
                    self._suppressed_monitor_plugins[normalized_id] = count - 1

    def is_plugin_monitor_suppressed(self, plugin_id: str) -> bool:
        """判断指定插件是否处于安装或替换写入阶段。"""
        with self._monitor_suppression_lock:
            return self._suppressed_monitor_plugins.get(plugin_id.lower(), 0) > 0

    def remove_plugin(self, plugin_id: str):
        """
        从内存中移除一个插件
        :param plugin_id: 插件ID
        """
        self._plugin_lifecycle.stop(plugin_id)
        self._plugin_registry.remove(plugin_id)

    def reload_plugin(self, plugin_id: str) -> PluginRuntimeStatus:
        """
        将一个插件重新加载到内存
        :param plugin_id: 插件ID
        """
        return self._plugin_lifecycle.reload(plugin_id, EventType.PluginReload)

    def reload_plugin_tree(self, plugin_id: str) -> PluginRuntimeStatus:
        """重载源码插件，并同步刷新所有引用该源码的虚拟实例。"""
        source_plugin_id = self.get_plugin_source_id(plugin_id)
        status = self.reload_plugin(source_plugin_id)
        for instance in self._plugin_instance_store.for_source(source_plugin_id):
            self.reload_plugin(instance.instance_id)
        return status

    def get_plugin_reload_targets(self, plugin_id: str) -> List[str]:
        """返回源码更新后需要刷新注册信息的源插件及其实例 ID。"""
        source_plugin_id = self.get_plugin_source_id(plugin_id)
        return [
            source_plugin_id,
            *(
                instance.instance_id
                for instance in self._plugin_instance_store.for_source(
                    source_plugin_id
                )
            ),
        ]

    @staticmethod
    def _clear_plugin_modules(plugin_id: Optional[str] = None):
        """
        清除插件及其所有子模块的缓存
        :param plugin_id: 插件ID
        """

        return PluginLoader(
            plugins_root=settings.ROOT_PATH / "app" / "plugins",
            import_preparer=lambda **kwargs: _legacy_plugin_import_preparer(**kwargs),
            import_scanner=lambda **kwargs: _legacy_import_scanner(**kwargs),
            log=logger,
        ).clear_modules(plugin_id)

    def sync(self) -> List[str]:
        """
        安装本地不存在或需要更新的插件
        """

        return self._plugin_sync.sync()

    @staticmethod
    def install_plugin_missing_dependencies() -> List[str]:
        """
        安装插件中缺失或不兼容的依赖项
        """
        return PluginDependencyService(
            system=get_plugin_system,
            log=logger,
        ).install_missing()

    @staticmethod
    def install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult:
        """安装插件缺失依赖并返回缺失项及安装成功状态。"""
        return PluginDependencyService(
            system=get_plugin_system,
            log=logger,
        ).install_missing_with_status()

    @staticmethod
    async def async_install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult:
        """在异步启动链中恢复插件依赖并保留取消语义。"""
        return await PluginDependencyService(
            system=get_plugin_system,
            log=logger,
        ).async_install_missing_with_status()

    def classify_plugins(self) -> PluginDependencyClassification:
        """按源码依赖状态分类物理插件，并把结果映射到虚拟实例。"""
        source_classification = PluginDependencyService(
            system=get_plugin_system,
            log=logger,
        ).classify_plugins()
        ready = list(source_classification.ready)
        missing_dependencies = list(source_classification.missing_dependencies)
        missing_source = list(source_classification.missing_source)
        source_ready = set(source_classification.ready)
        source_pending = set(source_classification.missing_dependencies)
        for instance in self._plugin_instance_store.all().values():
            if instance.source_plugin_id in source_ready:
                ready.append(instance.instance_id)
            elif instance.source_plugin_id in source_pending:
                missing_dependencies.append(instance.instance_id)
            else:
                missing_source.append(instance.instance_id)
        return PluginDependencyClassification(
            ready=tuple(ready),
            missing_dependencies=tuple(missing_dependencies),
            missing_source=tuple(missing_source),
        )

    def apply_plugin_dependency_classification(
        self,
        classification: PluginDependencyClassification,
    ) -> None:
        """把源码和依赖分类写入运行状态，已激活插件保持当前结果。"""
        running_ids = set(self._plugin_registry.running_ids())
        for plugin_id in classification.missing_source:
            self._plugin_registry.set_runtime_status(
                plugin_id,
                PluginRuntimeStatus.SOURCE_MISSING,
            )
        for plugin_id in classification.missing_dependencies:
            self._plugin_registry.set_runtime_status(
                plugin_id,
                PluginRuntimeStatus.DEPENDENCY_PENDING,
            )
        for plugin_id in classification.ready:
            current_status = self._plugin_registry.runtime_status(plugin_id)
            if (
                plugin_id in running_ids
                and current_status is not PluginRuntimeStatus.DEPENDENCY_PENDING
            ):
                continue
            self._plugin_registry.set_runtime_status(
                plugin_id,
                PluginRuntimeStatus.READY,
            )

    def set_plugin_settling(self, settling: bool) -> None:
        """更新启动后的插件恢复任务状态。"""
        self._plugin_registry.set_settling(settling)

    def get_plugin_runtime_statuses(self) -> Dict[str, PluginRuntimeStatus]:
        """返回插件运行状态快照。"""
        return self._plugin_registry.runtime_status_snapshot()

    def get_plugin_runtime_generation(self) -> int:
        """返回插件状态变化代次。"""
        return self._plugin_registry.generation

    def is_plugin_settling(self) -> bool:
        """返回插件源码和依赖是否仍在后台恢复。"""
        return self._plugin_registry.settling

    def get_plugin_config(self, pid: str) -> dict:
        """
        获取插件配置
        :param pid: 插件ID
        """
        return self._plugin_config_store.read(pid)

    def get_plugin_instances(self) -> dict[str, PluginInstance]:
        """返回全部有效虚拟插件实例描述。"""
        return self._plugin_instance_store.all()

    def get_plugin_instance(self, plugin_id: str) -> Optional[PluginInstance]:
        """返回指定虚拟插件实例描述，物理插件返回空。"""
        return self._plugin_instance_store.get(plugin_id)

    def get_plugin_source_id(self, plugin_id: str) -> str:
        """解析插件运行身份对应的源码身份，普通插件保持原值。"""
        instance = self._plugin_instance_store.get(plugin_id)
        return instance.source_plugin_id if instance else plugin_id

    def get_plugin_source_instances(self, plugin_id: str) -> List[PluginInstance]:
        """返回直接引用指定源码插件的虚拟实例。"""
        return self._plugin_instance_store.for_source(plugin_id)

    def delete_plugin_instance(self, plugin_id: str) -> bool:
        """删除虚拟实例描述；调用方仍负责停止实例和清理业务数据。"""
        return self._plugin_instance_store.delete(plugin_id)

    def save_plugin_config(self, pid: str, conf: dict, force: bool = False) -> bool:
        """
        保存插件配置
        :param pid: 插件ID
        :param conf: 配置
        :param force: 强制保存
        """
        return self._plugin_config_store.write(pid, conf, force)

    async def async_save_plugin_config(
        self, pid: str, conf: dict, force: bool = False
    ) -> bool:
        """
        异步保存插件配置。
        :param pid: 插件ID
        :param conf: 配置
        :param force: 强制保存
        """
        return await self._plugin_config_store.async_write(pid, conf, force)

    def delete_plugin_config(self, pid: str, force: bool = False) -> bool:
        """
        删除插件配置
        :param pid: 插件ID
        :param force: 插件停止后仍允许按插件 ID 删除持久化配置
        """
        return self._plugin_config_store.delete(pid, force)

    def delete_plugin_data(self, pid: str, force: bool = False) -> bool:
        """
        删除插件数据
        :param pid: 插件ID
        :param force: 插件停止后仍允许按插件 ID 删除持久化数据
        """
        return self._plugin_config_store.delete_data(pid, force)

    def get_plugin_state(self, pid: str) -> bool:
        """
        获取插件状态
        :param pid: 插件ID
        """
        plugin = self._plugin_registry.instance(pid)
        return plugin.get_state() if plugin else False

    def _plugin_projection(self) -> PluginProjection:
        """构造绑定当前运行态插件注册表的能力投影服务。"""
        return PluginProjection(
            self._plugin_registry.running,
            logger,
            self.get_plugin_remote_entry,
        )

    def _plugin_catalog(self) -> Any:
        """构造绑定当前市场客户端和插件 DTO 映射器的目录应用服务。"""
        return _plugin_catalog_factory(self)

    def get_plugin_commands(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件命令
        [{
            "cmd": "/xx",
            "event": EventType.xx,
            "desc": "xxxx",
            "data": {},
            "pid": "",
        }]
        """
        return self._plugin_projection().commands(pid)

    def get_plugin_apis(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件API
        [{
            "path": "/xx",
            "endpoint": self.xxx,
            "methods": ["GET", "POST"],
            "summary": "API名称",
            "description": "API说明",
            "allow_anonymous": false
        }]
        """
        return self._plugin_projection().apis(pid)

    def get_plugin_services(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件服务
        [{
            "id": "服务ID",
            "name": "服务名称",
            "trigger": "触发器：cron、interval、date、CronTrigger.from_crontab()",
            "func": self.xxx,
            "kwargs": {} # 定时器参数,
            "func_kwargs": {} # 方法参数
        }]
        """
        return self._plugin_projection().services(pid)

    def get_plugin_modules(self, pid: Optional[str] = None) -> Dict[tuple, Dict[str, Any]]:
        """
        获取插件模块
        {
            plugin_id: {
                method: function
            }
        }
        """
        return self._plugin_projection().modules(pid)

    def get_media_sources(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """获取运行中插件声明的媒体数据源。"""
        return self._plugin_projection().media_sources(pid)

    def get_plugin_actions(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件动作
        [{
            "id": "动作ID",
            "name": "动作名称",
            "func": self.xxx,
            "kwargs": {} # 需要附加传递的参数
        }]
        """
        return self._plugin_projection().actions(pid)

    @staticmethod
    def _copy_plugin_agent_tools(
        tools_info: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """
        复制插件智能体工具注册信息，避免调用方修改缓存内容。
        """
        return PluginToolCatalog.copy(tools_info)

    def get_plugin_agent_tools(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件智能体工具
        [{
            "plugin_id": "插件ID",
            "plugin_name": "插件名称",
            "tools": [ToolClass1, ToolClass2, ...]
        }]
        """
        return self._plugin_tool_catalog.get(
            self._running_plugins,
            plugin_id=pid,
            log=logger,
        )

    @staticmethod
    def get_plugin_remote_entry(plugin_id: str, dist_path: str) -> str:
        """
        获取插件的远程入口地址
        :param plugin_id: 插件 ID
        :param dist_path: 插件的分发路径
        :return: 远程入口地址
        """
        dist_path = dist_path.strip("/")
        path = posixpath.join(
            "plugin",
            "file",
            plugin_id.lower(),
            dist_path,
            "remoteEntry.js",
        )
        if not path.startswith("/"):
            path = "/" + path
        return path

    def get_plugin_remotes(self, pid: Optional[str] = None) -> List[Dict[str, Any]]:
        """
        获取插件联邦组件列表
        """
        return self._plugin_projection().remotes(pid)

    def get_plugin_auth_providers(self) -> List[Dict[str, Any]]:
        """
        聚合插件声明的登录认证提供方。

        :return: 插件认证入口列表
        """
        return self._plugin_projection().auth_providers()

    def get_plugin_sidebar_nav(self) -> List[Dict[str, Any]]:
        """
        聚合所有已启用 Vue 插件的侧栏导航项（get_sidebar_nav）。
        """
        return self._plugin_projection().sidebar()

    def get_plugin_dashboard_meta(self) -> List[Dict[str, str]]:
        """
        获取所有插件仪表盘元信息
        """
        return self._plugin_projection().dashboard_metadata()

    def get_plugin_dashboard(self, pid: str, key: str, user_agent: str = None) -> Optional[_SchemaPluginDashboard]:
        """
        获取插件仪表盘
        """
        return self._plugin_projection().dashboard(pid, key, user_agent)

    def get_plugin_attr(self, pid: str, attr: str) -> Any:
        """
        获取插件属性
        :param pid: 插件ID
        :param attr: 属性名
        """
        plugin = self._plugin_registry.instance(pid)
        if not plugin:
            return None
        if not hasattr(plugin, attr):
            return None
        return getattr(plugin, attr)

    def run_plugin_method(self, pid: str, method: str, *args, **kwargs) -> Any:
        """
        运行插件方法
        :param pid: 插件ID
        :param method: 方法名
        :param args: 参数
        :param kwargs: 关键字参数
        """
        plugin = self._plugin_registry.instance(pid)
        if not plugin:
            return None
        if not hasattr(plugin, method):
            return None
        return getattr(plugin, method)(*args, **kwargs)

    async def async_run_plugin_method(self, pid: str, method: str, *args, **kwargs) -> Any:
        """
        异步运行插件方法
        :param pid: 插件ID
        :param method: 方法名
        :param args: 参数
        :param kwargs: 关键字参数
        """
        plugin = self._plugin_registry.instance(pid)
        if not plugin:
            return None
        if not hasattr(plugin, method):
            return None
        method_func = getattr(plugin, method)
        if asyncio.iscoroutinefunction(method_func):
            return await method_func(*args, **kwargs)
        else:
            return method_func(*args, **kwargs)

    def get_plugin_ids(self) -> List[str]:
        """
        获取所有插件ID
        """
        return self._plugin_registry.plugin_ids()

    def get_running_plugin_ids(self) -> List[str]:
        """
        获取所有运行态插件ID
        """
        return self._plugin_registry.running_ids()

    def get_online_plugins(self, force: bool = False) -> List[_SchemaPlugin]:
        """
        获取所有在线插件信息
        """
        return self._plugin_catalog_view.online(force)

    def get_local_plugins(self) -> List[_SchemaPlugin]:
        """
        获取所有本地已下载的插件信息
        """
        return self._plugin_catalog_view.local()

    def get_installed_plugins(self) -> List[_SchemaPlugin]:
        """按安装清单返回插件，即使运行时尚未加载也保留卡片。"""
        return self._plugin_catalog_view.installed()

    def get_local_plugin_version(self, pid: str) -> Optional[str]:
        """
        获取指定已安装插件的本地版本，不触发全部插件的状态、页面和权限计算。

        插件类由运行期动态加载，旧插件可能未声明版本属性，因此缺失时返回 None。
        """
        return self._plugin_catalog_view.local_version(pid)

    def get_local_repo_plugins(self) -> List[_SchemaPlugin]:
        """
        获取本地插件仓库目录中的插件信息
        """
        return self._plugin_catalog_view.local_repository()

    def is_plugin_exists(self, pid: str, version: str = None) -> bool:
        """
        判断插件是否存在，并满足版本要求(有传入version时)
        :param pid: 插件ID
        :param version: 插件版本
        """
        return self._plugin_catalog_view.exists(pid, version)

    def get_plugins_from_market(self, market: str,
                                package_version: Optional[str] = None,
                                force: bool = False) -> Optional[List[_SchemaPlugin]]:
        """
        从指定的市场获取插件信息
        :param market: 市场的 URL 或标识
        :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本
        :param force: 是否强制刷新（忽略缓存）
        :return: 返回插件的列表，若获取失败返回 []
        """
        return self._plugin_catalog_view.get_from_market(
            market,
            package_version,
            force,
        )

    def process_plugins_list(self, higher_version_plugins: List[_SchemaPlugin],
                             base_version_plugins: List[_SchemaPlugin]) -> List[_SchemaPlugin]:
        """
        处理插件列表：合并、去重、排序、保留最高版本
        :param higher_version_plugins: 高版本插件列表
        :param base_version_plugins: 基础版本插件列表
        :return: 处理后的插件列表
        """
        return self._plugin_catalog_view.merge(
            higher_version_plugins,
            base_version_plugins,
        )

    def _process_plugin_info(self, pid: str, plugin_info: dict, market: str,
                             installed_apps: List[str], add_time: int,
                             package_version: Optional[str] = None) -> Optional[_SchemaPlugin]:
        """
        处理单个插件信息，创建 schemas.Plugin 对象
        :param pid: 插件ID
        :param plugin_info: 插件信息字典
        :param market: 市场URL
        :param installed_apps: 已安装插件列表
        :param add_time: 添加顺序
        :param package_version: 包版本
        :return: 创建的插件对象，如果验证失败返回None
        """
        return self._plugin_metadata.map(
            plugin_id=pid,
            plugin_info=plugin_info,
            market=market,
            installed_plugins=installed_apps,
            add_time=add_time,
            package_version=package_version,
        )

    @staticmethod
    def _normalize_plugin_label(labels: Any) -> Optional[str]:
        """
        规整插件市场标签字段，兼容旧字符串和新列表格式。

        :param labels: 插件市场 package 中的 labels 字段
        :return: 用空格拼接后的标签字符串，无法识别或为空时返回 None
        """
        return PluginMetadataMapper.normalize_label(labels)

    async def async_get_online_plugins(
            self,
            force: bool = False,
            progress_callback: Optional[Callable[..., None]] = None,
    ) -> List[_SchemaPlugin]:
        """
        异步获取所有在线插件信息
        :param force: 是否强制刷新（忽略缓存）
        :param progress_callback: 定时服务进度更新回调
        """
        return await self._plugin_catalog_view.async_online(
            force,
            progress_callback,
        )

    async def async_get_plugins_from_market(self, market: str,
                                            package_version: Optional[str] = None,
                                            force: bool = False) -> Optional[List[_SchemaPlugin]]:
        """
        异步从指定的市场获取插件信息
        :param market: 市场的 URL 或标识
        :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本
        :param force: 是否强制刷新（忽略缓存）
        :return: 返回插件的列表，若获取失败返回 []
        """
        return await self._plugin_catalog_view.async_get_from_market(
            market,
            package_version,
            force,
        )

    def __set_and_check_auth_level(
        self,
        plugin: Union[_SchemaPlugin, Type[Any]],
        source: Optional[Union[dict, Type[Any]]] = None,
    ) -> bool:
        """
        设置并检查插件的认证级别
        :param plugin: 插件对象或包含 auth_level 属性的对象
        :param source: 可选的字典对象或类对象，可能包含 "level" 或 "auth_level" 键
        :return: 如果插件的认证级别有效且当前环境的认证级别满足要求，返回 True，否则返回 False
        """
        return self._plugin_access.check(plugin, source)

    @staticmethod
    def __get_plugin_private_key(plugin_id: str) -> Optional[str]:
        """
        根据插件标识获取对应的私钥
        :param plugin_id: 插件标识
        :return: 对应的插件私钥，如果未找到则返回 None
        """
        return PluginAccessPolicy.private_key(plugin_id)

    def clone_plugin(self, plugin_id: str, suffix: str, name: str, description: str,
                     version: str = None, icon: str = None) -> Tuple[bool, str]:
        """
        创建插件分身
        :param plugin_id: 原插件ID
        :param suffix: 分身后缀
        :param name: 分身名称
        :param description: 分身描述
        :param version: 自定义版本号
        :param icon: 自定义图标URL
        :return: (是否成功, 错误信息)
        """
        return self._plugin_clone.clone(
            plugin_id=plugin_id,
            suffix=suffix,
            name=name,
            description=description,
            version=version,
            icon=icon,
        )

    def _modify_plugin_files(self, plugin_dir: Path, original_id: str, suffix: str,
                             name: str, description: str, version: str = None,
                             icon: str = None) -> Tuple[bool, str]:
        """
        兼容旧内部调用，将分身文件改写委托给包适配器。
        :param plugin_dir: 插件目录
        :param original_id: 原插件ID
        :param suffix: 分身后缀
        :param name: 分身名称
        :param description: 分身描述
        :param version: 自定义版本号
        :param icon: 自定义图标URL
        :return: (是否成功, 错误信息)
        """
        original_plugin_class = self._plugins.get(original_id)
        if not original_plugin_class:
            return False, f"无法获取原插件类 {original_id}"
        return get_plugin_system().package._modify_plugin_files(
            plugin_dir=plugin_dir,
            original_class_name=original_plugin_class.__name__,
            suffix=suffix,
            name=name,
            description=description,
            version=version,
            icon=icon,
        )

    @staticmethod
    def _modify_python_file(file_path: Path, original_class_name: str,
                            clone_class_name: str, name: str, description: str,
                            version: str = None, icon: str = None) -> Tuple[bool, str]:
        """
        兼容旧内部调用，将 Python 文件改写委托给包适配器。
        """
        return get_plugin_system().package._modify_python_file(
            file_path=file_path,
            original_class_name=original_class_name,
            clone_class_name=clone_class_name,
            name=name,
            description=description,
            version=version,
            icon=icon,
        )

    def _modify_federation_files(self, dist_dir: Path, original_class_name: str,
                                 clone_class_name: str) -> Tuple[bool, str]:
        """
        兼容旧内部调用，将联邦文件改写委托给包适配器。
        """
        return get_plugin_system().package._modify_federation_files(
            dist_dir=dist_dir,
            original_class_name=original_class_name,
            clone_class_name=clone_class_name,
        )

    @staticmethod
    def _rename_federation_assets(dist_dir: Path, original_class_name: str, clone_class_name: str):
        """
        兼容旧内部调用，将资源重命名委托给包适配器。
        """
        get_plugin_system().package._rename_federation_assets(
            dist_dir,
            original_class_name,
            clone_class_name,
        )