mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 20:05:42 +08:00
2352 lines
81 KiB
Python
2352 lines
81 KiB
Python
"""
|
||
封装 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)
|