""" 封装 MoviePilot 中心服务请求、响应解析，以及供旧调用方使用的上报和分享入口。 """ import asyncio import json import platform from collections.abc import Coroutine from typing import ( TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional, ParamSpec, Protocol, Tuple, TypeVar, Union, cast, ) from urllib.parse import parse_qs, quote, urlparse, urlsplit, urlunsplit from httpx2 import Response as AsyncResponse from requests import Response from app.adapters.network.http import AsyncRequestUtils, RequestUtils from app.adapters.system.host import SystemUtils from app.domain.context import MediaInfo, MusicInfo from app.domain.media import normalize_music_type from app.domain.meta.metabase import MetaBase from app.runtime.cache import cached from app.runtime.log import logger from app.runtime.loop import main_loop_registry from app.runtime.observability import observe_compat_facade from app.runtime.settings import get_runtime_setting from app.runtime.tasks import get_task_registry from app.runtime.version import get_app_version, get_frontend_version from app.schemas.media import resolve_media_identity from app.schemas.types import ( MUSIC_ENTITY_RECORDING, MediaType, SystemConfigKey, media_type_to_agent, ) if TYPE_CHECKING: from app.application.server.report import ServerReportService from app.application.server.share import ServerSharingService from app.application.workflow import WorkflowSnapshot # 中心服务适配器只通过 runtime 读取端口消费组合根的最新配置。 _LookupParams = ParamSpec("_LookupParams") _LookupResult = TypeVar("_LookupResult") class _CacheControl(Protocol): """ 缓存装饰器为包装函数提供的失效接口。 """ def cache_clear(self) -> None: """ 清空此包装函数对应的缓存区域；不发起服务端请求。 """ ... def _cached_server_result( *, region: Optional[str] = None, maxsize: int = 1024, ttl: Optional[int] = None, skip_empty: bool = False, ) -> Callable[[Callable[_LookupParams, _LookupResult]], Callable[_LookupParams, _LookupResult]]: """ 保留统计和分享查询的参数、结果类型，沿用底层缓存策略。 :param region: 原有缓存区域名称；未指定时按被包装函数生成 :param maxsize: 缓存条目上限 :param ttl: 条目有效秒数；未指定时沿用底层默认值 :param skip_empty: 是否跳过空结果 :return: 不改变调用签名的缓存装饰器 """ def decorator(function: Callable[_LookupParams, _LookupResult]) -> Callable[_LookupParams, _LookupResult]: """ 恢复底层包装器保持不变的原始调用签名。 """ return cached(region=region, maxsize=maxsize, ttl=ttl, skip_empty=skip_empty)(function) return decorator def _clear_server_cache(function: Callable[..., object]) -> None: """ 清空由缓存装饰器附加的区域，不改变原有失效顺序。 """ cast(_CacheControl, function).cache_clear() def _response_mapping(response: Response | AsyncResponse) -> dict[str, Any]: """ 在外部 JSON 边界确认对象形状，保留协议允许的任意字段内容。 :param response: 已取得的同步或异步 HTTP 响应 :return: 原始 JSON 对象，不复制或丢弃字段 :raises ValueError: JSON 解析失败或顶层不是对象 """ payload: object = response.json() if not isinstance(payload, dict): raise ValueError("中心服务响应不是 JSON 对象") return payload def _cached_server_lookup( maxsize: int, ) -> Callable[ [Callable[_LookupParams, Awaitable[Dict[str, Any]]]], Callable[_LookupParams, Awaitable[Dict[str, Any]]], ]: """ 为中心服务开关与权限查询提供保留签名的短时进程内缓存。 成功结果缓存 60 秒，使服务端开关与名单变更在一分钟内生效；不可达或非 200 得到的 空结果只缓存 15 秒，故障恢复后尽快重新判定。结果随本实例身份变化，不共享到 Redis。 """ def decorator( function: Callable[_LookupParams, Awaitable[Dict[str, Any]]], ) -> Callable[_LookupParams, Awaitable[Dict[str, Any]]]: """ 包装单个查询函数，缓存层本身未标注类型，这里显式恢复原签名。 """ return cached(maxsize=maxsize, ttl=60, empty_ttl=15, local_only=True)(function) return decorator _server_report_service: ServerReportService | None = None _server_sharing_service: ServerSharingService | None = None def configure_server_application_services( *, report_service: ServerReportService, sharing_service: ServerSharingService, ) -> None: """ 由启动组合根注入分享和存量上报应用服务。 """ global _server_report_service, _server_sharing_service _server_report_service = report_service _server_sharing_service = sharing_service def reset_server_application_services() -> None: """ 清除当前应用生命周期的中心服务上报与分享用例。 """ global _server_report_service, _server_sharing_service _server_report_service = None _server_sharing_service = None @observe_compat_facade("MoviePilotServerHelper") class MoviePilotServerHelper: """ MoviePilot 服务端请求辅助工具。 中心服务请求统一使用 HTTP/1.1，用于对照验证间歇性请求失败是否与 HTTP/2 链路有关。 """ USER_UID_HEADER = "X-MoviePilot-User-Uid" _USAGE_REPORT_PATH = "/usage/report" _USAGE_STATISTIC_PATH = "/usage/statistic" _PLUGIN_INSTALL_PATH = "/plugin/install" _PLUGIN_RATING_PATH = "/plugin/rating" _PLUGIN_STATISTIC_PATH = "/plugin/statistic" _SUBSCRIBE_ADD_PATH = "/subscribe/add" _SUBSCRIBE_DONE_PATH = "/subscribe/done" _SUBSCRIBE_REPORT_PATH = "/subscribe/report" _SUBSCRIBE_STATISTIC_PATH = "/subscribe/statistic" _SUBSCRIBE_SHARE_PATH = "/subscribe/share" _SUBSCRIBE_SHARES_PATH = "/subscribe/shares" _SUBSCRIBE_SHARE_STATISTICS_PATH = "/subscribe/share/statistics" _SUBSCRIBE_FORK_PATH = "/subscribe/fork" _WORKFLOW_SHARE_PATH = "/workflow/share" _WORKFLOW_SHARES_PATH = "/workflow/shares" _WORKFLOW_FORK_PATH = "/workflow/fork" _RECOGNIZE_SHARE_PATH = "/recognize/share" _USER_PERMISSIONS_PATH = "/user/permissions" _USER_FEATURES_PATH = "/user/features" _LOCAL_REPO_PREFIX = "local://" _user_uid: Optional[str] = None _github_user: Optional[str] = None @staticmethod def normalize_relay_url(server: str) -> str: """ 校验中转地址并统一持久凭据的作用域，不允许凭据、查询参数或片段。 :param server: 管理员配置的 HTTPS 服务根地址，回环调试允许 HTTP :return: 规范化且去除尾部斜线的地址 :raises ValueError: 地址不能安全接收实例凭据 """ target = urlsplit(server) local = target.hostname in {"localhost", "127.0.0.1", "::1"} if (target.scheme != "https" and not (local and target.scheme == "http")) or not target.hostname or target.username or target.password or target.query or target.fragment: raise ValueError("中转地址必须使用 HTTPS，本机回环调试除外") return urlunsplit((target.scheme, target.netloc.lower(), target.path.rstrip("/"), "", "")) @staticmethod def relay_push(server: str, key: str, method: str, path: str, data: dict[str, Any]) -> dict[str, Any]: """ 向显式配置的私有 Server 转发通知，发送密钥只放请求头且不跟随重定向。 :param server: 管理员配置的 HTTPS 中转地址；本机联调允许回环 HTTP :param key: 该实例专用的中转发送密钥 :param method: 内部限定的 GET、POST 或 DELETE 方法 :param path: 内部限定的 push 接口路径 :param data: 已绑定当前用户的注册或消息数据 :return: Server 返回的对象；失败抛出不含凭据或原始响应的 ValueError """ server = MoviePilotServerHelper.normalize_relay_url(server) allowed = {("GET", "devices"), ("POST", "devices"), ("DELETE", "devices"), ("POST", "pairings"), ("POST", "messages"), ("POST", "instances"), ("PUT", "instances/credentials"), ("DELETE", "instances")} if (method, path) not in allowed: raise ValueError("不支持的通知接口") if len(key) < 32: raise ValueError("请配置至少 32 位的中转发送密钥") # 首次登记或恢复轮换之后还要执行原操作，两段请求均需落在 App 的65秒截止时间内。 response = RequestUtils(timeout=20, headers={"Authorization": f"Bearer {key}"}).request( method, f"{server.rstrip('/')}/push/{path}", allow_redirects=False, verify=True, params=data if method == "GET" else None, json=data if method != "GET" else None, ) if response is None or response.status_code != 200: code = response.status_code if response is not None else 0 if code == 403 and path == "instances": raise ValueError("消息中心尚未开放自动接入，请联系中心运营者") if code == 401: raise ValueError("通知接入凭据无效或已停用，请联系中心运营者") if code == 429: raise ValueError("消息中心请求过于频繁，请稍后重试") raise ValueError(f"私有通知服务请求失败（{code}）") try: return _response_mapping(response) except ValueError as error: raise ValueError("私有通知服务响应无效") from error @classmethod def _report_service(cls) -> ServerReportService: """ 返回启动组合根注入的存量上报应用服务。 """ if _server_report_service is None: raise RuntimeError("中心服务上报用例尚未由启动组合根装配") return _server_report_service @classmethod def _sharing_service(cls) -> ServerSharingService: """ 返回启动组合根注入的订阅和工作流分享应用服务。 """ if _server_sharing_service is None: raise RuntimeError("中心服务分享用例尚未由启动组合根装配") return _server_sharing_service @classmethod def get_user_uid(cls) -> Optional[str]: """ 获取当前安装实例用于服务端统计识别的稳定用户 ID。 """ if cls._user_uid is None: # 底层兼容工具尚未标注签名；实现始终返回哈希字符串或 None。 uid_provider: Callable[[], Optional[str]] = SystemUtils.generate_user_unique_id cls._user_uid = uid_provider() return cls._user_uid @staticmethod def is_server_url(url: str) -> bool: """ 判断请求地址是否指向配置中的 MoviePilot 服务端。 """ server_host = (get_runtime_setting('MP_SERVER_HOST') or "").strip().rstrip("/") if not server_host or not url: return False try: target = urlparse(str(url).strip()) server = urlparse(server_host) except Exception: return False return bool( target.scheme and target.netloc and server.scheme and server.netloc and target.scheme == server.scheme and target.netloc == server.netloc ) @classmethod def build_headers( cls, url: str, headers: Optional[dict[str, Any]] = None, content_type: Optional[str] = None, ) -> dict[str, Any]: """ 构建访问 MoviePilot 服务端需要的请求头。 """ request_headers = { key: value for key, value in (headers or {}).items() if value is not None } if content_type and not cls._has_header(request_headers, "Content-Type"): request_headers["Content-Type"] = content_type if not cls.is_server_url(url) or cls._has_header(request_headers, cls.USER_UID_HEADER): return request_headers user_uid = cls.get_user_uid() if user_uid: request_headers[cls.USER_UID_HEADER] = user_uid request_headers["User-Agent"] = get_runtime_setting('USER_AGENT') return request_headers @classmethod def get_user_uuid(cls) -> str: """ 获取当前用户 UUID。 """ user_uid = cls.get_user_uid() if user_uid: logger.info(f"当前用户UUID: {user_uid}") return user_uid or "" @classmethod def get_github_user(cls) -> str: """ 获取当前 GitHub 用户名。 """ plan = cls._build_github_user_request() if plan: res = RequestUtils(**plan).get_res("https://api.github.com/user") cls._remember_github_user(res) return cls._github_user or "" @classmethod async def async_get_github_user(cls) -> str: """ 异步获取当前 GitHub 用户名。 """ plan = cls._build_github_user_request() if plan: res = await AsyncRequestUtils(**plan).get_res("https://api.github.com/user") cls._remember_github_user(res) return cls._github_user or "" @classmethod def _build_github_user_request(cls) -> Optional[Dict[str, Any]]: """ 生成 GitHub 当前用户查询计划，已有缓存或未配置凭据时不发请求。 """ headers = get_runtime_setting('GITHUB_HEADERS') if cls._github_user is not None or not headers: return None return { "headers": headers, "proxies": get_runtime_setting('PROXY'), "timeout": 15, } @classmethod def _remember_github_user(cls, response: Response | AsyncResponse | None) -> None: """ 统一解释 GitHub 用户响应并更新同步异步共享的用户名缓存。 """ if response: cls._github_user = response.json().get("login") logger.info(f"当前Github用户: {cls._github_user}") @classmethod def user_permissions(cls, github_user: str) -> Optional[Response]: """ 查询服务端用户权限。 """ return cls._get( cls._server_url(cls._USER_PERMISSIONS_PATH), params={"github_user": github_user}, include_user_uid=False, timeout=5, ) @classmethod async def async_user_permissions(cls, github_user: str) -> Optional[AsyncResponse]: """ 异步查询服务端用户权限。 """ return await cls._async_get( cls._server_url(cls._USER_PERMISSIONS_PATH), params={"github_user": github_user}, include_user_uid=False, timeout=5, ) @classmethod async def async_user_features(cls) -> Any: """ 异步查询服务端对所有实例相同的全局功能开关。 """ return await cls._async_get( cls._server_url(cls._USER_FEATURES_PATH), include_user_uid=False, timeout=5, ) @classmethod @_cached_server_lookup(maxsize=1) async def async_get_features(cls) -> Dict[str, Any]: """ 获取服务端全局功能开关（如 ``community_medals``），失败时返回空对象。 结果与实例身份无关，用单一键缓存：成功 60 秒，使开关变更在一分钟内生效；服务端 不可达或非 200 得到的空结果只缓存 15 秒。仅缓存于进程内存。 """ try: return cls._handle_mapping_response(await cls.async_user_features()) except Exception as err: logger.debug(f"获取服务端全局功能开关失败：{str(err)}") return {} @classmethod def get_user_permissions(cls) -> Dict[str, Any]: """ 获取当前用户在服务端配置中的权限。 """ github_user = cls.get_github_user() if not github_user: return {} try: return cls._handle_mapping_response(cls.user_permissions(github_user)) except Exception as err: logger.debug(f"获取服务端用户权限失败：{str(err)}") return {} @classmethod async def async_get_user_permissions(cls) -> Dict[str, Any]: """ 异步获取当前用户在服务端配置中的权限；未登录 GitHub 时不发请求并返回空权限。 """ github_user = await cls.async_get_github_user() if not github_user: return {} return await cls._async_fetch_user_permissions(github_user) @classmethod @_cached_server_lookup(maxsize=4) async def _async_fetch_user_permissions(cls, github_user: str) -> Dict[str, Any]: """ 按 GitHub 用户名查询并短时缓存服务端权限。 前端每次加载会同时请求 ``/system/global/user`` 与 ``/github/community``，两者都依赖 此结果（共享管理权限与社区勋章预览），短时缓存避免每次页面加载向服务端重复往返。 缓存键为用户名，换账号后不沿用旧结论。成功结果缓存 60 秒，使服务端开关与管理员名单变更在一分钟内生效；服务端 不可达或非 200 得到的空权限只缓存 15 秒，故障恢复后尽快重新判定。仅缓存于进程内存， 权限随本实例 GitHub 身份变化，不共享到 Redis。 """ try: return cls._handle_mapping_response( await cls.async_user_permissions(github_user) ) except Exception as err: logger.debug(f"获取服务端用户权限失败：{str(err)}") return {} @classmethod def is_admin_user(cls) -> bool: """ 判断当前用户是否为共享管理用户。 """ return cls._permissions_allow_sharing(cls.get_user_permissions()) @classmethod async def async_is_admin_user(cls) -> bool: """ 异步判断当前用户是否为共享管理用户。 """ return cls._permissions_allow_sharing( await cls.async_get_user_permissions() ) @staticmethod def _permissions_allow_sharing(permissions: Dict[str, Any]) -> bool: """ 按统一权限字段判断当前用户是否可管理共享内容。 """ return bool( permissions.get("is_admin") or permissions.get("subscribe_share_manage") or permissions.get("workflow_share_manage") ) @classmethod def build_usage_payload(cls) -> Dict[str, Any]: """ 构建安装版本统计上报载荷。 """ return { "user_uid": cls.get_user_uid(), "backend_version": get_app_version(), "frontend_version": get_frontend_version(), "version_flag": get_runtime_setting('VERSION_FLAG'), "platform": f"{platform.system()} {platform.release()}".strip(), "arch": SystemUtils.cpu_arch(), } @classmethod def report_usage(cls) -> bool: """ 上报当前安装实例的版本统计。 """ payload = cls._build_usage_report_plan() if not payload: return False try: return cls._response_succeeded(cls.usage_report(payload)) except Exception as err: logger.debug(f"上报安装版本统计失败：{str(err)}") return False @classmethod async def async_report_usage(cls) -> bool: """ 异步上报当前安装实例的版本统计。 """ payload = cls._build_usage_report_plan() if not payload: return False try: return cls._response_succeeded(await cls.async_usage_report(payload)) except Exception as err: logger.debug(f"上报安装版本统计失败：{str(err)}") return False @classmethod def _build_usage_report_plan(cls) -> Optional[Dict[str, Any]]: """ 在统计已启用且实例身份有效时生成版本统计载荷。 """ if not get_runtime_setting('USAGE_STATISTIC_SHARE'): return None payload = cls.build_usage_payload() return payload if payload.get("user_uid") else None @classmethod async def async_get_usage_statistic(cls) -> Dict[str, Any]: """ 异步获取安装版本统计报表。 """ if not get_runtime_setting('USAGE_STATISTIC_SHARE'): return {} try: return cls._handle_mapping_response(await cls.async_usage_statistic()) except Exception as err: logger.debug(f"异步获取安装版本统计报表失败：{str(err)}") return {} @classmethod def init_subscribe_report(cls) -> None: """ 初始化订阅统计上报状态。 """ cls._report_service().init_report( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), state_key=SystemConfigKey.SubscribeReport, reporter=cls.sub_report, ) @classmethod async def async_init_subscribe_report(cls) -> None: """ 异步初始化订阅统计标记。 """ await cls._report_service().async_init_report( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), state_key=SystemConfigKey.SubscribeReport, reporter=cls.async_sub_report, ) @classmethod def init_plugin_report(cls) -> None: """ 初始化插件安装统计上报状态。 """ cls._report_service().init_report( enabled=get_runtime_setting('PLUGIN_STATISTIC_SHARE'), state_key=SystemConfigKey.PluginInstallReport, reporter=cls.install_plugin_report, ) @classmethod async def async_init_plugin_report(cls) -> None: """ 异步初始化插件统计标记。 """ await cls._report_service().async_init_report( enabled=get_runtime_setting('PLUGIN_STATISTIC_SHARE'), state_key=SystemConfigKey.PluginInstallReport, reporter=cls.async_install_plugin_report, ) @staticmethod def _handle_list_response(res: Response | AsyncResponse | None) -> List[dict[str, Any]]: """ 处理服务端返回的列表响应。 """ if res is not None and res.status_code == 200: payload: object = res.json() if not isinstance(payload, list) or not all(isinstance(item, dict) for item in payload): raise ValueError("中心服务响应不是 JSON 对象列表") return payload return [] @staticmethod def _handle_mapping_response(res: Response | AsyncResponse | None) -> Dict[str, Any]: """ 处理服务端返回的对象响应，非成功状态统一映射为空对象。 """ if res is not None and res.status_code == 200: return _response_mapping(res) return {} @staticmethod def _response_succeeded(res: Response | AsyncResponse | None) -> bool: """ 把同步与异步传输响应统一映射为服务端成功状态。 """ return bool(res is not None and res.status_code == 200) @staticmethod def _handle_response( res: Response | AsyncResponse | None, clear_cache: Optional[Callable[[], None]] = None, ) -> Tuple[bool, str]: """ 处理服务端写入类接口响应。 """ if res is None: return False, "连接MoviePilot服务器失败" if res.status_code == 200: if clear_cache: clear_cache() return True, "" try: return False, res.json().get("message", "未知错误") except (json.JSONDecodeError, ValueError, AttributeError): return False, f"响应解析失败: {getattr(res, 'text', '')[:100]}..." @classmethod def usage_report(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 上报安装版本统计。 """ return cls._post_json(cls._server_url(cls._USAGE_REPORT_PATH), payload, timeout=5) @classmethod async def async_usage_report(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步上报安装版本统计。 """ return await cls._async_post_json(cls._server_url(cls._USAGE_REPORT_PATH), payload, timeout=5) @classmethod def usage_statistic(cls) -> Optional[Response]: """ 获取安装版本统计报表。 """ return cls._get(cls._server_url(cls._USAGE_STATISTIC_PATH), timeout=10) @classmethod async def async_usage_statistic(cls) -> Optional[AsyncResponse]: """ 异步获取安装版本统计报表。 """ return await cls._async_get(cls._server_url(cls._USAGE_STATISTIC_PATH), timeout=10) @classmethod def plugin_statistic(cls) -> Optional[Response]: """ 获取插件安装统计。 """ return cls._get(cls._server_url(cls._PLUGIN_STATISTIC_PATH), timeout=10) @classmethod async def async_plugin_statistic(cls) -> Optional[AsyncResponse]: """ 异步获取插件安装统计。 """ return await cls._async_get(cls._server_url(cls._PLUGIN_STATISTIC_PATH), timeout=10) @classmethod async def async_plugin_ratings(cls, plugin_ids: Optional[List[str]] = None) -> Optional[AsyncResponse]: """ 异步批量查询中心端插件评分。 """ params = {"plugin_ids": ",".join(plugin_ids)} if plugin_ids is not None else None return await cls._async_get( cls._server_url(cls._PLUGIN_RATING_PATH), params=params, timeout=10, ) @classmethod async def async_plugin_rating(cls, plugin_id: str) -> Optional[AsyncResponse]: """ 异步查询中心端单个插件评分。 """ return await cls._async_get( f"{cls._server_url(cls._PLUGIN_RATING_PATH)}/{quote(plugin_id, safe='')}", timeout=10, ) @classmethod async def async_rate_plugin(cls, plugin_id: str, rating: float) -> Optional[AsyncResponse]: """ 异步提交当前安装实例的插件评分。 """ return await cls._async_post_json( f"{cls._server_url(cls._PLUGIN_RATING_PATH)}/{quote(plugin_id, safe='')}", {"rating": rating}, timeout=10, ) @classmethod def plugin_install(cls, plugin_id: str, payload: Dict[str, Any]) -> Optional[Response]: """ 上报单个插件安装统计。 """ return cls._post_json(f"{cls._server_url(cls._PLUGIN_INSTALL_PATH)}/{plugin_id}", payload, timeout=5) @classmethod async def async_plugin_install(cls, plugin_id: str, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步上报单个插件安装统计。 """ return await cls._async_post_json( f"{cls._server_url(cls._PLUGIN_INSTALL_PATH)}/{plugin_id}", payload, timeout=5, ) @classmethod def plugin_install_report(cls, plugins: List[Dict[str, Any]]) -> Optional[Response]: """ 批量上报插件安装统计。 """ return cls._post_json(cls._server_url(cls._PLUGIN_INSTALL_PATH), {"plugins": plugins}, timeout=5) @classmethod async def async_plugin_install_report(cls, plugins: List[Dict[str, Any]]) -> Optional[AsyncResponse]: """ 异步批量上报插件安装统计。 """ return await cls._async_post_json( cls._server_url(cls._PLUGIN_INSTALL_PATH), {"plugins": plugins}, timeout=5, ) @classmethod @_cached_server_result(maxsize=1, ttl=1800) def get_plugin_statistic(cls) -> Dict[str, Any]: """ 获取插件安装统计。 """ if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'): return {} return cls._handle_mapping_response(cls.plugin_statistic()) @classmethod async def async_get_plugin_statistic(cls) -> Dict[str, Any]: """ 异步获取插件安装统计。 """ if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'): return {} return cls._handle_mapping_response(await cls.async_plugin_statistic()) @classmethod async def async_get_plugin_ratings( cls, plugin_ids: Optional[List[str]] = None, ) -> Dict[str, Dict[str, Any]]: """ 批量获取插件评分，中心端不可用时返回空结果。 """ try: res = await cls.async_plugin_ratings(plugin_ids) if res is not None and res.status_code == 200: payload = _response_mapping(res) if not all(isinstance(value, dict) for value in payload.values()): raise ValueError("中心服务评分响应包含非对象记录") return payload except Exception as err: logger.debug(f"批量获取插件评分失败：{str(err)}") return {} @classmethod async def async_get_plugin_rating(cls, plugin_id: str) -> Dict[str, Any]: """ 获取单个插件评分，中心端不可用时返回零评分。 """ empty_rating = { "plugin_id": plugin_id, "average_rating": 0.0, "rating_count": 0, "user_rating": None, } try: res = await cls.async_plugin_rating(plugin_id) if res is not None and res.status_code == 200: return _response_mapping(res) except Exception as err: logger.debug(f"获取插件 {plugin_id} 评分失败：{str(err)}") return empty_rating @classmethod async def async_submit_plugin_rating( cls, plugin_id: str, rating: float, ) -> Optional[Dict[str, Any]]: """ 提交插件评分，成功时返回最新评分结果。 """ try: res = await cls.async_rate_plugin(plugin_id, rating) if res is not None and res.status_code == 200: return _response_mapping(res) except Exception as err: logger.debug(f"提交插件 {plugin_id} 评分失败：{str(err)}") return None @classmethod def install_plugin_reg(cls, plugin_id: str, repo_url: Optional[str] = None) -> bool: """ 上报单个插件安装统计。 """ payload = cls._build_plugin_install_plan(plugin_id, repo_url) if not payload: return False return cls._response_succeeded(cls.plugin_install(plugin_id, payload)) @classmethod async def async_install_plugin_reg(cls, plugin_id: str, repo_url: Optional[str] = None) -> bool: """ 异步上报单个插件安装统计。 """ payload = cls._build_plugin_install_plan(plugin_id, repo_url) if not payload: return False return cls._response_succeeded( await cls.async_plugin_install(plugin_id, payload) ) @classmethod def _build_plugin_install_plan( cls, plugin_id: str, repo_url: Optional[str], ) -> Optional[Dict[str, Any]]: """ 在插件统计已启用且插件身份有效时生成单次安装上报载荷。 """ if not get_runtime_setting('PLUGIN_STATISTIC_SHARE') or not plugin_id: return None return { "plugin_id": plugin_id, "repo_url": cls.sanitize_plugin_repo_url(repo_url), } @classmethod def install_plugin_report(cls, items: Optional[List[Tuple[str, Optional[str]]]] = None) -> bool: """ 批量上报存量插件安装统计。 """ return cls._report_service().report_plugins( enabled=get_runtime_setting('PLUGIN_STATISTIC_SHARE'), items=items, ) @classmethod async def async_install_plugin_report(cls, items: Optional[List[Tuple[str, Optional[str]]]] = None) -> bool: """ 异步批量上报存量插件安装统计。 """ return await cls._report_service().async_report_plugins( enabled=get_runtime_setting('PLUGIN_STATISTIC_SHARE'), items=items, ) @classmethod def subscribe_statistic(cls, params: Dict[str, Any]) -> Optional[Response]: """ 获取订阅统计数据。 """ return cls._get(cls._server_url(cls._SUBSCRIBE_STATISTIC_PATH), params=params, timeout=15) @classmethod async def async_subscribe_statistic(cls, params: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步获取订阅统计数据。 """ return await cls._async_get(cls._server_url(cls._SUBSCRIBE_STATISTIC_PATH), params=params, timeout=15) @classmethod def subscribe_add(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 新增订阅统计。 """ return cls._post_json(cls._server_url(cls._SUBSCRIBE_ADD_PATH), payload, timeout=5) @classmethod async def async_subscribe_add(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步新增订阅统计。 """ return await cls._async_post_json(cls._server_url(cls._SUBSCRIBE_ADD_PATH), payload, timeout=5) @classmethod def subscribe_done(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 完成订阅统计。 """ return cls._post_json(cls._server_url(cls._SUBSCRIBE_DONE_PATH), payload, timeout=5) @classmethod async def async_subscribe_done(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步完成订阅统计，并返回可检查 HTTP 状态的响应对象。 """ return await cls._async_post_json( cls._server_url(cls._SUBSCRIBE_DONE_PATH), payload, timeout=5, ) @classmethod async def async_subscribe_report(cls, subscribes: List[Dict[str, Any]]) -> Optional[AsyncResponse]: """ 异步批量上报存量订阅统计。 """ return await cls._async_post_json( cls._server_url(cls._SUBSCRIBE_REPORT_PATH), {"subscribes": subscribes}, timeout=10, ) @classmethod def subscribe_report(cls, subscribes: List[Dict[str, Any]]) -> Optional[Response]: """ 批量上报存量订阅统计。 """ return cls._post_json( cls._server_url(cls._SUBSCRIBE_REPORT_PATH), {"subscribes": subscribes}, timeout=10, ) @classmethod def subscribe_share(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 分享订阅数据。 """ return cls._post_json(cls._server_url(cls._SUBSCRIBE_SHARE_PATH), payload, timeout=10) @classmethod async def async_subscribe_share(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步分享订阅数据。 """ return await cls._async_post_json(cls._server_url(cls._SUBSCRIBE_SHARE_PATH), payload, timeout=10) @classmethod def subscribe_share_delete(cls, share_id: int, share_uid: str) -> Optional[Response]: """ 删除订阅分享数据。 """ return cls._delete( f"{cls._server_url(cls._SUBSCRIBE_SHARE_PATH)}/{share_id}", params={"share_uid": share_uid}, timeout=5, ) @classmethod async def async_subscribe_share_delete(cls, share_id: int, share_uid: str) -> Optional[AsyncResponse]: """ 异步删除订阅分享数据。 """ return await cls._async_delete( f"{cls._server_url(cls._SUBSCRIBE_SHARE_PATH)}/{share_id}", params={"share_uid": share_uid}, timeout=5, ) @classmethod def subscribe_fork(cls, share_id: int) -> Optional[Response]: """ 复用订阅分享数据。 """ return cls._get(f"{cls._server_url(cls._SUBSCRIBE_FORK_PATH)}/{share_id}", timeout=5) @classmethod async def async_subscribe_fork(cls, share_id: int) -> Optional[AsyncResponse]: """ 异步复用订阅分享数据。 """ return await cls._async_get(f"{cls._server_url(cls._SUBSCRIBE_FORK_PATH)}/{share_id}", timeout=5) @classmethod def subscribe_shares(cls, params: Dict[str, Any]) -> Optional[Response]: """ 获取订阅分享数据。 """ return cls._get(cls._server_url(cls._SUBSCRIBE_SHARES_PATH), params=params, timeout=15) @classmethod async def async_subscribe_shares(cls, params: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步获取订阅分享数据。 """ return await cls._async_get(cls._server_url(cls._SUBSCRIBE_SHARES_PATH), params=params, timeout=15) @classmethod def subscribe_share_statistics(cls) -> Optional[Response]: """ 获取订阅分享统计数据。 """ return cls._get(cls._server_url(cls._SUBSCRIBE_SHARE_STATISTICS_PATH), timeout=15) @classmethod async def async_subscribe_share_statistics(cls) -> Optional[AsyncResponse]: """ 异步获取订阅分享统计数据。 """ return await cls._async_get(cls._server_url(cls._SUBSCRIBE_SHARE_STATISTICS_PATH), timeout=15) @staticmethod def _build_subscribe_query_params( page: Optional[int] = 1, count: Optional[int] = 30, genre_id: Optional[int] = None, min_rating: Optional[float] = None, max_rating: Optional[float] = None, sort_type: Optional[str] = None, **extra: object, ) -> Dict[str, Any]: """ 构建订阅统计与分享查询参数。 """ params = { "page": page, "count": count, **extra, } if genre_id is not None: params["genre_id"] = genre_id if min_rating is not None: params["min_rating"] = min_rating if max_rating is not None: params["max_rating"] = max_rating if sort_type is not None: params["sort_type"] = sort_type return params @classmethod def _build_subscribe_query_plan( cls, **kwargs: Any, ) -> Optional[Dict[str, Any]]: """ 在订阅统计已启用时生成统计或分享列表查询计划。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return None return cls._build_subscribe_query_params(**kwargs) @classmethod def _build_subscribe_statistic_plan( cls, item: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """ 在订阅统计已启用且载荷有效时生成新增或完成上报计划。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return None return cls._build_subscribe_statistic_payload(item) @staticmethod def _feature_disabled_result( setting_key: str, message: str, ) -> Optional[Tuple[bool, str]]: """ 将分享功能开关统一映射为可直接返回的禁用结果。 """ if not get_runtime_setting(setting_key): return False, message return None @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) def get_subscribe_statistic( cls, stype: str, page: Optional[int] = 1, count: Optional[int] = 30, genre_id: Optional[int] = None, min_rating: Optional[float] = None, max_rating: Optional[float] = None, sort_type: Optional[str] = None, ) -> List[dict[str, Any]]: """ 获取订阅统计数据。 """ params = cls._build_subscribe_query_plan( page=page, count=count, genre_id=genre_id, min_rating=min_rating, max_rating=max_rating, sort_type=sort_type, stype=stype, ) if params is None: return [] return cls._handle_list_response(cls.subscribe_statistic(params)) @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) async def async_get_subscribe_statistic( cls, stype: str, page: Optional[int] = 1, count: Optional[int] = 30, genre_id: Optional[int] = None, min_rating: Optional[float] = None, max_rating: Optional[float] = None, sort_type: Optional[str] = None, ) -> List[dict[str, Any]]: """ 异步获取订阅统计数据。 """ params = cls._build_subscribe_query_plan( page=page, count=count, genre_id=genre_id, min_rating=min_rating, max_rating=max_rating, sort_type=sort_type, stype=stype, ) if params is None: return [] return cls._handle_list_response(await cls.async_subscribe_statistic(params)) @classmethod def sub_reg(cls, sub: dict[str, Any]) -> bool: """ 新增订阅统计。 """ payload = cls._build_subscribe_statistic_plan(sub) if not payload: return False return cls._response_succeeded(cls.subscribe_add(payload)) @classmethod async def async_sub_reg(cls, sub: dict[str, Any]) -> bool: """ 异步新增订阅统计。 """ payload = cls._build_subscribe_statistic_plan(sub) if not payload: return False return cls._response_succeeded(await cls.async_subscribe_add(payload)) @classmethod def sub_reg_durable(cls, sub: dict[str, Any]) -> bool: """ 同步上报新增统计；明确禁用时视为无需投递。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return True return cls.sub_reg(sub) @classmethod async def async_sub_reg_durable(cls, sub: dict[str, Any]) -> bool: """ 异步上报新增统计；明确禁用时视为无需投递。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return True return await cls.async_sub_reg(sub) @classmethod def sub_done(cls, sub: dict[str, Any]) -> bool: """ 完成订阅统计。 """ payload = cls._build_subscribe_statistic_plan(sub) if not payload: return False return cls._response_succeeded(cls.subscribe_done(payload)) @classmethod async def async_sub_done(cls, sub: dict[str, Any]) -> bool: """ 异步完成订阅统计，并仅在服务端确认成功时返回 True。 """ payload = cls._build_subscribe_statistic_plan(sub) if not payload: return False return cls._response_succeeded(await cls.async_subscribe_done(payload)) @classmethod def sub_done_durable(cls, sub: dict[str, Any]) -> bool: """ 同步上报完成统计；明确禁用时视为无需投递。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return True return cls.sub_done(sub) @classmethod async def async_sub_done_durable(cls, sub: dict[str, Any]) -> bool: """ 异步上报完成统计；明确禁用时视为无需投递。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return True return await cls.async_sub_done(sub) @staticmethod def _submit_statistic_report( coroutine: Coroutine[Any, Any, Any], *, submit: Callable[[Coroutine[Any, Any, Any]], object], action: str, ) -> bool: """ 提交兼容统计上报，并在宿主不再接收任务时关闭未执行协程。 """ try: submit(coroutine) return True except Exception as err: coroutine.close() logger.warning(f"调度{action}失败：{err}") return False @classmethod def sub_reg_async(cls, sub: dict[str, Any]) -> bool: """ 兼容旧同步入口，在宿主任务登记器中提交新增订阅统计。 """ return cls._submit_statistic_report( asyncio.to_thread(cls.sub_reg, sub), submit=lambda coroutine: get_task_registry().submit_threadsafe( coroutine, loop=main_loop_registry.require(), owner="compat.server.subscribe_added_report", cancel_on_shutdown=False, ), action="新增订阅统计上报", ) @classmethod def sub_done_async(cls, sub: dict[str, Any]) -> bool: """ 兼容旧同步入口，在宿主任务登记器中提交订阅完成统计。 """ return cls._submit_statistic_report( asyncio.to_thread(cls.sub_done, sub), submit=lambda coroutine: get_task_registry().submit_threadsafe( coroutine, loop=main_loop_registry.require(), owner="compat.server.subscribe_done_report", cancel_on_shutdown=False, ), action="订阅完成统计上报", ) @classmethod def sub_report(cls) -> bool: """ 上报存量订阅统计。 """ return cls._report_service().report_subscribes( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), ) @classmethod async def async_sub_report(cls) -> bool: """ 异步上报存量订阅统计。 """ return await cls._report_service().async_report_subscribes( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), ) @classmethod def sub_share( cls, subscribe_id: int, share_title: str, share_comment: str, share_user: str, ) -> Tuple[bool, str]: """ 分享订阅。 """ return cls._sharing_service().share_subscribe( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), subscribe_id=subscribe_id, share_title=share_title, share_comment=share_comment, share_user=share_user, ) @classmethod async def async_sub_share( cls, subscribe_id: int, share_title: str, share_comment: str, share_user: str, ) -> Tuple[bool, str]: """ 异步分享订阅。 """ return await cls._sharing_service().async_share_subscribe( enabled=get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'), subscribe_id=subscribe_id, share_title=share_title, share_comment=share_comment, share_user=share_user, ) @classmethod def _build_subscribe_statistic_payload( cls, item: Optional[dict[str, Any]] ) -> Optional[dict[str, Any]]: """ 构造中心服务订阅统计载荷，只保留统一身份和公开统计字段。 """ return cls._report_service().build_subscribe_payload(item) @classmethod def _build_subscribe_share_payload( cls, item: Optional[dict[str, Any]] ) -> Optional[dict[str, Any]]: """ 构造中心服务订阅分享载荷，隔离本地运行字段和旧专用 ID。 """ return cls._sharing_service().build_subscribe_payload(item) @classmethod def share_delete(cls, share_id: int) -> Tuple[bool, str]: """ 删除订阅分享。 """ disabled = cls._feature_disabled_result( 'SUBSCRIBE_STATISTIC_SHARE', "当前没有开启订阅数据共享功能", ) if disabled: return disabled return cls._handle_response( cls.subscribe_share_delete(share_id, cls.get_user_uuid()), cls._clear_subscribe_share_cache, ) @classmethod async def async_share_delete(cls, share_id: int) -> Tuple[bool, str]: """ 异步删除订阅分享。 """ disabled = cls._feature_disabled_result( 'SUBSCRIBE_STATISTIC_SHARE', "当前没有开启订阅数据共享功能", ) if disabled: return disabled return cls._handle_response( await cls.async_subscribe_share_delete(share_id, cls.get_user_uuid()), cls._clear_subscribe_share_cache, ) @classmethod def sub_fork(cls, share_id: int) -> Tuple[bool, str]: """ 复用订阅分享。 """ disabled = cls._feature_disabled_result( 'SUBSCRIBE_STATISTIC_SHARE', "当前没有开启订阅数据共享功能", ) if disabled: return disabled return cls._handle_response(cls.subscribe_fork(share_id)) @classmethod async def async_sub_fork(cls, share_id: int) -> Tuple[bool, str]: """ 异步复用订阅分享。 """ disabled = cls._feature_disabled_result( 'SUBSCRIBE_STATISTIC_SHARE', "当前没有开启订阅数据共享功能", ) if disabled: return disabled return cls._handle_response(await cls.async_subscribe_fork(share_id)) @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) def get_subscribe_shares( cls, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30, genre_id: Optional[int] = None, min_rating: Optional[float] = None, max_rating: Optional[float] = None, sort_type: Optional[str] = None, ) -> List[dict[str, Any]]: """ 获取订阅分享数据。 """ params = cls._build_subscribe_query_plan( page=page, count=count, genre_id=genre_id, min_rating=min_rating, max_rating=max_rating, sort_type=sort_type, name=name, ) if params is None: return [] return cls._handle_list_response(cls.subscribe_shares(params)) @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) async def async_get_subscribe_shares( cls, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30, genre_id: Optional[int] = None, min_rating: Optional[float] = None, max_rating: Optional[float] = None, sort_type: Optional[str] = None, ) -> List[dict[str, Any]]: """ 异步获取订阅分享数据。 """ params = cls._build_subscribe_query_plan( page=page, count=count, genre_id=genre_id, min_rating=min_rating, max_rating=max_rating, sort_type=sort_type, name=name, ) if params is None: return [] return cls._handle_list_response(await cls.async_subscribe_shares(params)) @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) def get_subscribe_share_statistics(cls) -> List[dict[str, Any]]: """ 获取订阅分享统计数据。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return [] return cls._handle_list_response(cls.subscribe_share_statistics()) @classmethod @_cached_server_result(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True) async def async_get_subscribe_share_statistics(cls) -> List[dict[str, Any]]: """ 异步获取订阅分享统计数据。 """ if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'): return [] return cls._handle_list_response(await cls.async_subscribe_share_statistics()) @classmethod def workflow_share(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 分享工作流数据。 """ return cls._post_json(cls._server_url(cls._WORKFLOW_SHARE_PATH), payload, timeout=10) @classmethod async def async_workflow_share(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步分享工作流数据。 """ return await cls._async_post_json(cls._server_url(cls._WORKFLOW_SHARE_PATH), payload, timeout=10) @classmethod def workflow_share_delete(cls, share_id: int, share_uid: str) -> Optional[Response]: """ 删除工作流分享数据。 """ return cls._delete( f"{cls._server_url(cls._WORKFLOW_SHARE_PATH)}/{share_id}", params={"share_uid": share_uid}, timeout=5, ) @classmethod async def async_workflow_share_delete(cls, share_id: int, share_uid: str) -> Optional[AsyncResponse]: """ 异步删除工作流分享数据。 """ return await cls._async_delete( f"{cls._server_url(cls._WORKFLOW_SHARE_PATH)}/{share_id}", params={"share_uid": share_uid}, timeout=5, ) @classmethod def workflow_fork(cls, share_id: int) -> Optional[Response]: """ 复用工作流分享数据。 """ return cls._get(f"{cls._server_url(cls._WORKFLOW_FORK_PATH)}/{share_id}", timeout=5) @classmethod async def async_workflow_fork(cls, share_id: int) -> Optional[AsyncResponse]: """ 异步复用工作流分享数据。 """ return await cls._async_get(f"{cls._server_url(cls._WORKFLOW_FORK_PATH)}/{share_id}", timeout=5) @classmethod def workflow_shares(cls, params: Dict[str, Any]) -> Optional[Response]: """ 获取工作流分享数据。 """ return cls._get(cls._server_url(cls._WORKFLOW_SHARES_PATH), params=params, timeout=15) @classmethod async def async_workflow_shares(cls, params: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步获取工作流分享数据。 """ return await cls._async_get(cls._server_url(cls._WORKFLOW_SHARES_PATH), params=params, timeout=15) @classmethod def _prepare_workflow_data(cls, workflow: WorkflowSnapshot) -> dict[str, Any]: """ 准备工作流分享数据。 """ return cls._sharing_service().prepare_workflow(workflow) @staticmethod def _build_workflow_query_plan( name: Optional[str], page: Optional[int], count: Optional[int], ) -> Optional[Dict[str, Any]]: """ 在工作流分享已启用时生成列表查询计划。 """ if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'): return None return {"name": name, "page": page, "count": count} @classmethod def workflow_share_by_id( cls, workflow_id: int, share_title: str, share_comment: str, share_user: str, ) -> Tuple[bool, str]: """ 分享工作流。 """ return cls._sharing_service().share_workflow( enabled=get_runtime_setting('WORKFLOW_STATISTIC_SHARE'), workflow_id=workflow_id, share_title=share_title, share_comment=share_comment, share_user=share_user, ) @classmethod async def async_workflow_share_by_id( cls, workflow_id: int, share_title: str, share_comment: str, share_user: str, ) -> Tuple[bool, str]: """ 异步分享工作流。 """ return await cls._sharing_service().async_share_workflow( enabled=get_runtime_setting('WORKFLOW_STATISTIC_SHARE'), workflow_id=workflow_id, share_title=share_title, share_comment=share_comment, share_user=share_user, ) @classmethod def workflow_share_delete_by_id(cls, share_id: int) -> Tuple[bool, str]: """ 删除工作流分享。 """ disabled = cls._feature_disabled_result( 'WORKFLOW_STATISTIC_SHARE', "当前没有开启工作流数据共享功能", ) if disabled: return disabled return cls._handle_response( cls.workflow_share_delete(share_id, cls.get_user_uuid()), cls._clear_workflow_share_cache, ) @classmethod async def async_workflow_share_delete_by_id(cls, share_id: int) -> Tuple[bool, str]: """ 异步删除工作流分享。 """ disabled = cls._feature_disabled_result( 'WORKFLOW_STATISTIC_SHARE', "当前没有开启工作流数据共享功能", ) if disabled: return disabled return cls._handle_response( await cls.async_workflow_share_delete(share_id, cls.get_user_uuid()), cls._clear_workflow_share_cache, ) @classmethod def workflow_fork_by_id(cls, share_id: int) -> Tuple[bool, str]: """ 复用工作流分享。 """ disabled = cls._feature_disabled_result( 'WORKFLOW_STATISTIC_SHARE', "当前没有开启工作流数据共享功能", ) if disabled: return disabled return cls._handle_response(cls.workflow_fork(share_id)) @classmethod async def async_workflow_fork_by_id(cls, share_id: int) -> Tuple[bool, str]: """ 异步复用工作流分享。 """ disabled = cls._feature_disabled_result( 'WORKFLOW_STATISTIC_SHARE', "当前没有开启工作流数据共享功能", ) if disabled: return disabled return cls._handle_response(await cls.async_workflow_fork(share_id)) @classmethod @_cached_server_result(region="workflow_share", maxsize=1, skip_empty=True) def get_workflow_shares( cls, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30, ) -> List[dict[str, Any]]: """ 获取工作流分享数据。 """ params = cls._build_workflow_query_plan(name, page, count) if params is None: return [] return cls._handle_list_response(cls.workflow_shares(params)) @classmethod @_cached_server_result(region="workflow_share", maxsize=1, skip_empty=True) async def async_get_workflow_shares( cls, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30, ) -> List[dict[str, Any]]: """ 异步获取工作流分享数据。 """ params = cls._build_workflow_query_plan(name, page, count) if params is None: return [] return cls._handle_list_response(await cls.async_workflow_shares(params)) @classmethod def _validate_workflow(cls, workflow: Optional[WorkflowSnapshot]) -> Tuple[bool, str]: """ 验证工作流是否可以分享。 """ return cls._sharing_service().validate_workflow(workflow) @classmethod def recognize_share_url(cls) -> Optional[str]: """ 获取共享识别服务端地址。 """ custom_api = (get_runtime_setting('MEDIA_RECOGNIZE_SHARE_API') or "").strip() if custom_api: return custom_api.rstrip("/") server_host = (get_runtime_setting('MP_SERVER_HOST') or "").strip().rstrip("/") if not server_host: return None return f"{server_host}{cls._RECOGNIZE_SHARE_PATH}" @classmethod def _build_recognize_transport_plan( cls, data: Dict[str, Any], ) -> Optional[Tuple[str, Dict[str, Any]]]: """ 把共享识别地址与已验证请求数据组合成纯传输计划。 """ api_url = cls.recognize_share_url() if not api_url: return None return api_url, data @classmethod def _build_recognize_query_plan( cls, meta: Optional[MetaBase], mtype: Optional[MediaType] = None, keyword_meta: Optional[MetaBase] = None, music_type: Optional[str] = None, ) -> Optional[Tuple[Dict[str, Any], Optional[str]]]: """ 在共享识别已启用且关键字有效时生成查询计划。 """ if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'): return None params = cls._build_recognize_query_params( meta=meta, mtype=mtype, keyword_meta=keyword_meta, music_type=music_type, ) if not params: return None return params, params.get("keyword") @classmethod def _build_recognize_report_plan( cls, meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]], keyword_meta: Optional[MetaBase] = None, ) -> Optional[Dict[str, Any]]: """ 在共享识别已启用且统一媒体身份完整时生成上报载荷。 """ if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'): return None return cls._build_recognize_report_payload( meta=meta, mediainfo=mediainfo, keyword_meta=keyword_meta, ) @classmethod def recognize_query(cls, params: Dict[str, Any]) -> Optional[Response]: """ 查询共享识别结果。 """ plan = cls._build_recognize_transport_plan(params) if not plan: return None api_url, request_data = plan return cls._get(api_url, params=request_data, timeout=5) @classmethod async def async_recognize_query(cls, params: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步查询共享识别结果。 """ plan = cls._build_recognize_transport_plan(params) if not plan: return None api_url, request_data = plan return await cls._async_get(api_url, params=request_data, timeout=5) @classmethod def recognize_report(cls, payload: Dict[str, Any]) -> Optional[Response]: """ 上报共享识别结果。 """ plan = cls._build_recognize_transport_plan(payload) if not plan: return None api_url, request_data = plan return cls._post_json(api_url, request_data, timeout=5) @classmethod async def async_recognize_report(cls, payload: Dict[str, Any]) -> Optional[AsyncResponse]: """ 异步上报共享识别结果。 """ plan = cls._build_recognize_transport_plan(payload) if not plan: return None api_url, request_data = plan return await cls._async_post_json(api_url, request_data, timeout=5) @classmethod def query_recognize_share( cls, meta: Optional[MetaBase], mtype: Optional[MediaType] = None, keyword_meta: Optional[MetaBase] = None, music_type: Optional[str] = None, ) -> Optional[dict[str, Any]]: """ 查询共享识别结果。 """ plan = cls._build_recognize_query_plan( meta=meta, mtype=mtype, keyword_meta=keyword_meta, music_type=music_type, ) if not plan: return None params, keyword = plan response = cls.recognize_query(params) return cls._parse_recognize_response(response, keyword) @classmethod async def async_query_recognize_share( cls, meta: Optional[MetaBase], mtype: Optional[MediaType] = None, keyword_meta: Optional[MetaBase] = None, music_type: Optional[str] = None, ) -> Optional[dict[str, Any]]: """ 异步查询共享识别结果。 """ plan = cls._build_recognize_query_plan( meta=meta, mtype=mtype, keyword_meta=keyword_meta, music_type=music_type, ) if not plan: return None params, keyword = plan response = await cls.async_recognize_query(params) return cls._parse_recognize_response(response, keyword) @classmethod def report_recognize_share( cls, meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]], keyword_meta: Optional[MetaBase] = None, ) -> bool: """ 上报共享识别结果，电影、电视剧、音乐共用。 """ payload = cls._build_recognize_report_plan( meta=meta, mediainfo=mediainfo, keyword_meta=keyword_meta, ) if payload is None: return False response = cls.recognize_report(payload) return cls._parse_recognize_report_response(response) @classmethod async def async_report_recognize_share( cls, meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]], keyword_meta: Optional[MetaBase] = None, ) -> bool: """ 异步上报共享识别结果，电影、电视剧、音乐共用。 """ payload = cls._build_recognize_report_plan( meta=meta, mediainfo=mediainfo, keyword_meta=keyword_meta, ) if payload is None: return False response = await cls.async_recognize_report(payload) return cls._parse_recognize_report_response(response) @classmethod def to_recognize_params(cls, item: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]: """ 将服务端返回的共享识别结果转成本地统一识别参数。 """ if not isinstance(item, dict): return None media_type = cls._normalize_media_type(item.get("type")) mtype = MediaType.from_agent(media_type) if media_type else None media_source, media_id = resolve_media_identity(media=item) if not media_source or not media_id: return None return { "mtype": mtype, "media_source": media_source, "media_id": media_id, "music_type": normalize_music_type(item.get("music_type")) or (MUSIC_ENTITY_RECORDING if mtype == MediaType.MUSIC else None), "season": item.get("season"), } @classmethod def sanitize_plugin_repo_url(cls, repo_url: Optional[str]) -> Optional[str]: """ 统计上报前脱敏插件仓库地址。 """ if not repo_url: return repo_url if not repo_url.startswith(cls._LOCAL_REPO_PREFIX): return repo_url plugin_id = cls._parse_local_repo_plugin_id(repo_url) if not plugin_id: return cls._LOCAL_REPO_PREFIX.rstrip("/") return cls._make_local_repo_url( plugin_id=plugin_id, package_version=cls._parse_local_repo_package_version(repo_url), ) @classmethod def _normalize_media_type(cls, media_type: Optional[object]) -> Optional[str]: """ 统一媒体类型，兼容枚举、中文值和 agent 风格字符串。 """ normalized = media_type_to_agent(media_type) if normalized in {"movie", "tv", "music"}: return normalized if isinstance(media_type, str): if media_type == MediaType.MOVIE.value: return "movie" if media_type == MediaType.TV.value: return "tv" if media_type == MediaType.MUSIC.value: return "music" return None @staticmethod def _extract_keyword(meta: Optional[MetaBase]) -> Optional[str]: """ 提取识别关键字。 """ if not meta: return None keyword = meta.original_name or meta.name if keyword: keyword = str(keyword).strip() return keyword or None @classmethod def _extract_media_type( cls, meta: Optional[MetaBase] = None, mtype: Optional[MediaType] = None, mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, ) -> Optional[str]: """ 提取媒体类型。 """ media_type = cls._normalize_media_type(mtype) if media_type: return media_type shareable_types = {MediaType.MOVIE, MediaType.TV, MediaType.MUSIC} if mediainfo and mediainfo.type in shareable_types: return mediainfo.type.to_agent() if meta and meta.type in shareable_types: return meta.type.to_agent() if meta and (meta.begin_season is not None or meta.begin_episode is not None): return "tv" return None @classmethod def _extract_season( cls, media_type: Optional[str], meta: Optional[MetaBase] = None, mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, ) -> Optional[int]: """ 提取季信息，仅电视剧使用。 """ if media_type != "tv": return None season = meta.begin_season if meta else None if season is None and mediainfo: season = mediainfo.season try: return int(season) if season is not None else None except (TypeError, ValueError): return None @staticmethod def _extract_year( meta: Optional[MetaBase] = None, mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, ) -> Optional[str]: """ 提取年份。 """ year = (meta.year if meta else None) or (mediainfo.year if mediainfo else None) if year is None: return None year_text = str(year).strip() return year_text or None @classmethod def _build_recognize_query_params( cls, meta: Optional[MetaBase], mtype: Optional[MediaType] = None, keyword_meta: Optional[MetaBase] = None, music_type: Optional[str] = None, ) -> Optional[dict[str, Any]]: """ 组装共享识别查询参数。 """ keyword = cls._extract_keyword(keyword_meta or meta) if not keyword: return None media_type = cls._extract_media_type(meta=meta, mtype=mtype) params: dict[str, str | int] = {"keyword": keyword} if media_type: params["type"] = media_type if media_type == "music": params["music_type"] = ( normalize_music_type(music_type) or MUSIC_ENTITY_RECORDING ) if year := cls._extract_year(meta=meta): params["year"] = year season = cls._extract_season(media_type=media_type, meta=meta) if season is not None: params["season"] = season return params @classmethod def _build_recognize_report_payload( cls, meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]], keyword_meta: Optional[MetaBase] = None, ) -> Optional[dict[str, Any]]: """ 组装共享识别上报载荷，电影、电视剧、音乐共用同一结构。 """ if not meta or not mediainfo: return None keyword = cls._extract_keyword(keyword_meta or meta) media_type = cls._extract_media_type(meta=meta, mediainfo=mediainfo) if not keyword or not media_type: return None media_source, media_id = resolve_media_identity(media=mediainfo) if not media_id: return None return { "keyword": keyword, "type": media_type, "title": mediainfo.title or keyword, "year": cls._extract_year(meta=meta, mediainfo=mediainfo), "season": cls._extract_season( media_type=media_type, meta=meta, mediainfo=mediainfo, ), "media_source": str(media_source), "media_id": media_id, "music_type": ( normalize_music_type(getattr(mediainfo, "music_type", None)) if media_type == "music" else None ), } @classmethod def _parse_recognize_response(cls, response: Response | AsyncResponse | None, keyword: Optional[str]) -> Optional[dict[str, Any]]: """ 解析共享识别查询响应。 """ if not response or response.status_code != 200: if response is not None: logger.warn( f"查询共享媒体识别失败：status={response.status_code} " f"message={cls._response_message(response)}" ) return None try: payload = response.json() except (json.JSONDecodeError, ValueError) as err: logger.warn(f"解析共享媒体识别响应失败：{err}") return None if payload.get("code") != 0: return None item = cls._parse_response_item(payload) if item: logger.info(f"共享媒体识别命中：{keyword} - {item}") return item @classmethod def _parse_recognize_report_response(cls, response: Response | AsyncResponse | None) -> bool: """ 解析共享识别上报响应。 """ if not response or response.status_code != 200: if response is not None: logger.warn( f"上报共享媒体识别失败：status={response.status_code} " f"message={cls._response_message(response)}" ) return False try: result = _response_mapping(response) except (json.JSONDecodeError, ValueError) as err: logger.warn(f"解析共享媒体识别上报响应失败：{err}") return False return bool(result.get("code") == 0) @staticmethod def _parse_response_item(data: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]: """ 解析服务端返回的共享识别数据。 """ if not isinstance(data, dict): return None item = (data.get("data") or {}).get("item") if not isinstance(item, dict): return None return item @staticmethod def _response_message(response: Response | AsyncResponse) -> str: """ 获取响应消息，兼容非 JSON 响应。 """ try: payload = response.json() return str(payload.get("message") or "") except (json.JSONDecodeError, ValueError, AttributeError): return "" @classmethod def _build_plugin_report_payload( cls, items: Optional[List[Tuple[str, Optional[str]]]] = None, ) -> List[Dict[str, Any]]: """ 构建批量插件安装统计载荷。 """ return cls._report_service().build_plugin_payload(items) @classmethod def _parse_local_repo_plugin_id(cls, repo_url: str) -> Optional[str]: """ 从本地插件来源标识中解析插件 ID。 """ try: parts = urlsplit(repo_url) plugin_id = parts.netloc or parts.path.strip("/") except Exception: plugin_id = repo_url[len(cls._LOCAL_REPO_PREFIX):].split("?", 1)[0].strip("/") return plugin_id or None @staticmethod def _parse_local_repo_package_version(repo_url: str) -> Optional[str]: """ 从本地插件来源标识中解析 package 版本。 """ try: values = parse_qs(urlsplit(repo_url).query).get("version") if not values: return None return values[0] except Exception: return None @classmethod def _make_local_repo_url( cls, plugin_id: str, package_version: Optional[str] = None, ) -> str: """ 生成脱敏后的本地插件来源标识。 """ repo_url = f"{cls._LOCAL_REPO_PREFIX}{quote(plugin_id, safe='')}" if package_version: repo_url = f"{repo_url}?version={quote(package_version, safe='')}" return repo_url @classmethod def _clear_subscribe_share_cache(cls) -> None: """ 清理订阅共享相关缓存。 """ _clear_server_cache(cls.get_subscribe_shares) _clear_server_cache(cls.get_subscribe_statistic) _clear_server_cache(cls.get_subscribe_share_statistics) @classmethod def _clear_workflow_share_cache(cls) -> None: """ 清理工作流共享相关缓存。 """ _clear_server_cache(cls.get_workflow_shares) @classmethod def _has_header(cls, headers: dict[str, Any], name: str) -> bool: """ 按 HTTP 头大小写不敏感规则判断请求头是否存在。 """ header_name = name.lower() return any(str(key).lower() == header_name for key in headers) @staticmethod def _server_url(path: str) -> str: """ 根据服务端基础地址和路径生成完整 URL。 """ return f"{get_runtime_setting('MP_SERVER_HOST').rstrip('/')}{path}" @classmethod def _get( cls, url: str, params: Optional[dict[str, Any]] = None, timeout: int = 10, include_user_uid: bool = True, ) -> Optional[Response]: """ 发送服务端 GET 请求，默认携带安装用户 ID。 """ return RequestUtils( proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url) if include_user_uid else {}, ).get_res(url, params=params) @classmethod async def _async_get( cls, url: str, params: Optional[dict[str, Any]] = None, timeout: int = 10, include_user_uid: bool = True, ) -> Optional[AsyncResponse]: """ 异步发送服务端 GET 请求，默认携带安装用户 ID。 """ return await AsyncRequestUtils( http2=False, proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url) if include_user_uid else {}, ).get_res(url, params=params) @classmethod def _post_json(cls, url: str, payload: Dict[str, Any], timeout: int = 10) -> Optional[Response]: """ 发送携带安装用户 ID 的服务端 JSON POST 请求。 """ return RequestUtils( proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url, content_type="application/json"), ).post(url, json=payload) @classmethod async def _async_post_json(cls, url: str, payload: Dict[str, Any], timeout: int = 10) -> Optional[AsyncResponse]: """ 异步发送携带安装用户 ID 的服务端 JSON POST 请求。 """ return await AsyncRequestUtils( http2=False, proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url, content_type="application/json"), ).post(url, json=payload) @classmethod def _delete(cls, url: str, params: Optional[dict[str, Any]] = None, timeout: int = 10) -> Optional[Response]: """ 发送携带安装用户 ID 的服务端 DELETE 请求。 """ client = RequestUtils( proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url), ) if params is None: return client.delete_res(url) return client.delete_res(url, params=params) @classmethod async def _async_delete(cls, url: str, params: Optional[dict[str, Any]] = None, timeout: int = 10) -> Optional[AsyncResponse]: """ 异步发送携带安装用户 ID 的服务端 DELETE 请求。 """ client = AsyncRequestUtils( http2=False, proxies=get_runtime_setting('PROXY'), timeout=timeout, headers=cls.build_headers(url), ) if params is None: return await client.delete_res(url) return await client.delete_res(url, params=params)