MoviePilot/app/application/server/report.py

"""
中心服务存量上报用例。
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from typing import Any, Optional

from app.application.server.payload import build_subscribe_payload


class ServerReportService:
    """
    协调本地订阅、插件清单和中心服务统计上报。
    """

    SUBSCRIBE_FIELDS = frozenset({
        "name", "year", "type", "media_source", "media_id", "music_type",
        "total_tracks", "genre_ids", "season", "poster", "backdrop", "vote",
        "description",
    })

    def __init__(
            self,
            *,
            config_reader: Callable[[Any], Any],
            config_writer: Callable[[Any, Any], Any],
            installed_plugins_provider: Callable[[], list[str]],
            subscribes_provider: Callable[[], list[Any]],
            async_subscribes_provider: Callable[[], Awaitable[list[Any]]] | None = None,
            plugin_report_sender: Callable[[list[dict[str, object]]], Any],
            async_plugin_report_sender: Callable[[list[dict[str, object]]], Awaitable[Any]],
            subscribe_report_sender: Callable[[list[dict[str, object]]], Any],
            repo_url_sanitizer: Callable[[Optional[str]], Optional[str]],
            async_subscribe_report_sender: Callable[[list[dict[str, object]]], Awaitable[Any]] | None = None,
            async_config_writer: Callable[[Any, Any], Awaitable[Any]] | None = None,
    ) -> None:
        """
        保存本地读取端口和只负责 I/O 的中心服务发送端口。
        """
        self._config_reader = config_reader
        self._config_writer = config_writer
        self._installed_plugins_provider = installed_plugins_provider
        self._subscribes_provider = subscribes_provider
        self._async_subscribes_provider = async_subscribes_provider
        self._plugin_report_sender = plugin_report_sender
        self._async_plugin_report_sender = async_plugin_report_sender
        self._subscribe_report_sender = subscribe_report_sender
        self._async_subscribe_report_sender = async_subscribe_report_sender
        self._async_config_writer = async_config_writer
        self._repo_url_sanitizer = repo_url_sanitizer

    def init_report(
            self,
            *,
            enabled: bool,
            state_key: Any,
            reporter: Callable[[], bool],
    ) -> None:
        """
        首次成功上报后写入对应的完成标记。
        """
        if self._should_initialize_report(enabled=enabled, state_key=state_key) and reporter():
            self._config_writer(state_key, "1")

    async def async_init_report(
        self,
        *,
        enabled: bool,
        state_key: Any,
        reporter: Callable[[], Awaitable[bool]],
    ) -> None:
        """
        异步完成首次上报，并通过异步配置端口持久化完成标记。
        """
        if not self._should_initialize_report(enabled=enabled, state_key=state_key):
            return
        if not await reporter():
            return
        if self._async_config_writer is None:
            raise RuntimeError("中心服务上报未配置异步配置写入端口")
        await self._async_config_writer(state_key, "1")

    def _should_initialize_report(self, *, enabled: bool, state_key: Any) -> bool:
        """
        统一判断首次上报是否仍需执行。
        """
        return enabled and not self._config_reader(state_key)

    def build_subscribe_payload(self, item: Optional[dict[str, object]]) -> Optional[dict[str, object]]:
        """
        构造中心服务订阅统计载荷并移除本地运行字段。
        """
        payload: dict[str, object] | None = build_subscribe_payload(item, self.SUBSCRIBE_FIELDS)
        return payload

    def build_plugin_payload(
            self,
            items: Optional[list[tuple[str, Optional[str]]]] = None,
    ) -> list[dict[str, object]]:
        """
        构造插件安装统计载荷并脱敏本地仓库路径。
        """
        if items:
            return [
                {
                    "plugin_id": plugin_id,
                    "repo_url": self._repo_url_sanitizer(repo_url),
                }
                for plugin_id, repo_url in items
                if plugin_id
            ]
        return [
            {"plugin_id": plugin_id, "repo_url": None}
            for plugin_id in self._installed_plugins_provider()
            if plugin_id
        ]

    def _prepare_subscribe_report(
        self,
        subscribes: list[Any],
    ) -> tuple[bool, Optional[list[dict[str, object]]]]:
        """
        把订阅读取结果投影为无需发送的终态或待发送载荷。
        """
        if not subscribes:
            return True, None
        payloads = [
            payload
            for subscribe in subscribes
            if (payload := self.build_subscribe_payload(subscribe.to_dict()))
        ]
        if not payloads:
            return True, None
        return False, payloads

    def _prepare_plugin_report(
        self,
        *,
        enabled: bool,
        items: Optional[list[tuple[str, Optional[str]]]],
    ) -> Optional[list[dict[str, object]]]:
        """
        统一执行插件上报准入并构造脱敏载荷。
        """
        if not enabled:
            return None
        payload = self.build_plugin_payload(items)
        return payload or None

    @staticmethod
    def _report_succeeded(response: Any) -> bool:
        """
        统一判定中心服务是否确认接收上报。
        """
        return bool(response is not None and response.status_code == 200)

    def report_subscribes(self, *, enabled: bool) -> bool:
        """
        上报当前全部有效订阅的公开统计字段。
        """
        if not enabled:
            return False
        subscribes = self._subscribes_provider()
        completed, payloads = self._prepare_subscribe_report(subscribes)
        if completed:
            return True
        assert payloads is not None
        response = self._subscribe_report_sender(payloads)
        return self._report_succeeded(response)

    def report_plugins(
            self,
            *,
            enabled: bool,
            items: Optional[list[tuple[str, Optional[str]]]] = None,
    ) -> bool:
        """
        同步上报当前插件安装清单。
        """
        payload = self._prepare_plugin_report(enabled=enabled, items=items)
        if not payload:
            return False
        response = self._plugin_report_sender(payload)
        return self._report_succeeded(response)

    async def async_report_plugins(
            self,
            *,
            enabled: bool,
            items: Optional[list[tuple[str, Optional[str]]]] = None,
    ) -> bool:
        """
        异步上报当前插件安装清单。
        """
        payload = self._prepare_plugin_report(enabled=enabled, items=items)
        if not payload:
            return False
        response = await self._async_plugin_report_sender(payload)
        return self._report_succeeded(response)

    async def async_report_subscribes(self, *, enabled: bool) -> bool:
        """
        异步上报当前全部有效订阅的公开统计字段。
        """
        if not enabled:
            return False
        if self._async_subscribe_report_sender is None:
            raise RuntimeError("中心服务上报未配置异步订阅发送端口")
        if self._async_subscribes_provider is None:
            raise RuntimeError("中心服务未配置异步订阅读取端口")
        subscribes = await self._async_subscribes_provider()
        completed, payloads = self._prepare_subscribe_report(subscribes)
        if completed:
            return True
        assert payloads is not None
        response = await self._async_subscribe_report_sender(payloads)
        return self._report_succeeded(response)