refactor(server): unify sync async decisions

This commit is contained in:
jxxghp
2026-08-29 22:07:53 +08:00
parent 4b1d4dd5b3
commit ac772646f7
2 changed files with 752 additions and 160 deletions
+281 -160
View File
@@ -167,15 +167,10 @@ class MoviePilotServerHelper:
"""
获取当前 GitHub 用户名。
"""
if cls._github_user is None and get_runtime_setting('GITHUB_HEADERS'):
res = RequestUtils(
headers=get_runtime_setting('GITHUB_HEADERS'),
proxies=get_runtime_setting('PROXY'),
timeout=15,
).get_res("https://api.github.com/user")
if res:
cls._github_user = res.json().get("login")
logger.info(f"当前Github用户: {cls._github_user}")
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
@@ -183,17 +178,31 @@ class MoviePilotServerHelper:
"""
异步获取当前 GitHub 用户名。
"""
if cls._github_user is None and get_runtime_setting('GITHUB_HEADERS'):
res = await AsyncRequestUtils(
headers=get_runtime_setting('GITHUB_HEADERS'),
proxies=get_runtime_setting('PROXY'),
timeout=15,
).get_res("https://api.github.com/user")
if res:
cls._github_user = res.json().get("login")
logger.info(f"当前Github用户: {cls._github_user}")
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: Any) -> 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):
"""
@@ -227,9 +236,7 @@ class MoviePilotServerHelper:
if not github_user:
return {}
try:
res = cls.user_permissions(github_user)
if res is not None and res.status_code == 200:
return res.json()
return cls._handle_mapping_response(cls.user_permissions(github_user))
except Exception as err:
logger.debug(f"获取服务端用户权限失败:{str(err)}")
return {}
@@ -243,11 +250,11 @@ class MoviePilotServerHelper:
if not github_user:
return {}
try:
res = await cls.async_user_permissions(github_user)
if res is not None and res.status_code == 200:
return res.json()
return cls._handle_mapping_response(
await cls.async_user_permissions(github_user)
)
except Exception as err:
logger.debug(f"异步获取服务端用户权限失败:{str(err)}")
logger.debug(f"获取服务端用户权限失败:{str(err)}")
return {}
@classmethod
@@ -255,19 +262,20 @@ class MoviePilotServerHelper:
"""
判断当前用户是否为共享管理用户。
"""
permissions = cls.get_user_permissions()
return bool(
permissions.get("is_admin")
or permissions.get("subscribe_share_manage")
or permissions.get("workflow_share_manage")
)
return cls._permissions_allow_sharing(cls.get_user_permissions())
@classmethod
async def async_is_admin_user(cls) -> bool:
"""
异步判断当前用户是否为共享管理用户。
"""
permissions = await cls.async_get_user_permissions()
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")
@@ -293,14 +301,11 @@ class MoviePilotServerHelper:
"""
上报当前安装实例的版本统计。
"""
if not get_runtime_setting('USAGE_STATISTIC_SHARE'):
return False
payload = cls.build_usage_payload()
if not payload.get("user_uid"):
payload = cls._build_usage_report_plan()
if not payload:
return False
try:
res = cls.usage_report(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(cls.usage_report(payload))
except Exception as err:
logger.debug(f"上报安装版本统计失败:{str(err)}")
return False
@@ -310,18 +315,23 @@ class MoviePilotServerHelper:
"""
异步上报当前安装实例的版本统计。
"""
if not get_runtime_setting('USAGE_STATISTIC_SHARE'):
return False
payload = cls.build_usage_payload()
if not payload.get("user_uid"):
payload = cls._build_usage_report_plan()
if not payload:
return False
try:
res = await cls.async_usage_report(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(await cls.async_usage_report(payload))
except Exception as err:
logger.debug(f"异步上报安装版本统计失败:{str(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]:
"""
@@ -330,9 +340,7 @@ class MoviePilotServerHelper:
if not get_runtime_setting('USAGE_STATISTIC_SHARE'):
return {}
try:
res = await cls.async_usage_statistic()
if res is not None and res.status_code == 200:
return res.json()
return cls._handle_mapping_response(await cls.async_usage_statistic())
except Exception as err:
logger.debug(f"异步获取安装版本统计报表失败:{str(err)}")
return {}
@@ -386,6 +394,18 @@ class MoviePilotServerHelper:
return res.json()
return []
@staticmethod
def _handle_mapping_response(res: Any) -> Dict[str, Any]:
"""处理服务端返回的对象响应,非成功状态统一映射为空对象。"""
if res is not None and res.status_code == 200:
return res.json()
return {}
@staticmethod
def _response_succeeded(res: Any) -> bool:
"""把同步与异步传输响应统一映射为服务端成功状态。"""
return bool(res is not None and res.status_code == 200)
@staticmethod
def _handle_response(res, clear_cache=None) -> Tuple[bool, str]:
"""
@@ -521,10 +541,7 @@ class MoviePilotServerHelper:
"""
if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'):
return {}
res = cls.plugin_statistic()
if res is not None and res.status_code == 200:
return res.json()
return {}
return cls._handle_mapping_response(cls.plugin_statistic())
@classmethod
async def async_get_plugin_statistic(cls) -> Dict:
@@ -533,10 +550,7 @@ class MoviePilotServerHelper:
"""
if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'):
return {}
res = await cls.async_plugin_statistic()
if res is not None and res.status_code == 200:
return res.json()
return {}
return cls._handle_mapping_response(await cls.async_plugin_statistic())
@classmethod
async def async_get_plugin_ratings(
@@ -595,30 +609,36 @@ class MoviePilotServerHelper:
"""
上报单个插件安装统计。
"""
if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'):
payload = cls._build_plugin_install_plan(plugin_id, repo_url)
if not payload:
return False
if not plugin_id:
return False
res = cls.plugin_install(plugin_id, {
"plugin_id": plugin_id,
"repo_url": cls.sanitize_plugin_repo_url(repo_url),
})
return bool(res is not None and res.status_code == 200)
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:
"""
异步上报单个插件安装统计。
"""
if not get_runtime_setting('PLUGIN_STATISTIC_SHARE'):
payload = cls._build_plugin_install_plan(plugin_id, repo_url)
if not payload:
return False
if not plugin_id:
return False
res = await cls.async_plugin_install(plugin_id, {
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),
})
return bool(res is not None and res.status_code == 200)
}
@classmethod
def install_plugin_report(cls, items: Optional[List[Tuple[str, Optional[str]]]] = None) -> bool:
@@ -810,6 +830,36 @@ class MoviePilotServerHelper:
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(region="subscribe_share", maxsize=32, ttl=1800, skip_empty=True)
def get_subscribe_statistic(
@@ -825,9 +875,7 @@ class MoviePilotServerHelper:
"""
获取订阅统计数据。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return []
params = cls._build_subscribe_query_params(
params = cls._build_subscribe_query_plan(
page=page,
count=count,
genre_id=genre_id,
@@ -836,6 +884,8 @@ class MoviePilotServerHelper:
sort_type=sort_type,
stype=stype,
)
if params is None:
return []
return cls._handle_list_response(cls.subscribe_statistic(params))
@classmethod
@@ -853,9 +903,7 @@ class MoviePilotServerHelper:
"""
异步获取订阅统计数据。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return []
params = cls._build_subscribe_query_params(
params = cls._build_subscribe_query_plan(
page=page,
count=count,
genre_id=genre_id,
@@ -864,6 +912,8 @@ class MoviePilotServerHelper:
sort_type=sort_type,
stype=stype,
)
if params is None:
return []
return cls._handle_list_response(await cls.async_subscribe_statistic(params))
@classmethod
@@ -871,26 +921,20 @@ class MoviePilotServerHelper:
"""
新增订阅统计。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False
payload = cls._build_subscribe_statistic_payload(sub)
payload = cls._build_subscribe_statistic_plan(sub)
if not payload:
return False
res = cls.subscribe_add(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(cls.subscribe_add(payload))
@classmethod
async def async_sub_reg(cls, sub: dict) -> bool:
"""
异步新增订阅统计。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False
payload = cls._build_subscribe_statistic_payload(sub)
payload = cls._build_subscribe_statistic_plan(sub)
if not payload:
return False
res = await cls.async_subscribe_add(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(await cls.async_subscribe_add(payload))
@classmethod
def sub_reg_durable(cls, sub: dict) -> bool:
@@ -911,24 +955,18 @@ class MoviePilotServerHelper:
"""
完成订阅统计。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False
payload = cls._build_subscribe_statistic_payload(sub)
payload = cls._build_subscribe_statistic_plan(sub)
if not payload:
return False
res = cls.subscribe_done(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(cls.subscribe_done(payload))
@classmethod
async def async_sub_done(cls, sub: dict) -> bool:
"""异步完成订阅统计,并仅在服务端确认成功时返回 True。"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False
payload = cls._build_subscribe_statistic_payload(sub)
payload = cls._build_subscribe_statistic_plan(sub)
if not payload:
return False
res = await cls.async_subscribe_done(payload)
return bool(res is not None and res.status_code == 200)
return cls._response_succeeded(await cls.async_subscribe_done(payload))
@classmethod
def sub_done_durable(cls, sub: dict) -> bool:
@@ -1061,8 +1099,12 @@ class MoviePilotServerHelper:
"""
删除订阅分享。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False, "当前没有开启订阅数据共享功能"
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,
@@ -1073,8 +1115,12 @@ class MoviePilotServerHelper:
"""
异步删除订阅分享。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False, "当前没有开启订阅数据共享功能"
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,
@@ -1085,8 +1131,12 @@ class MoviePilotServerHelper:
"""
复用订阅分享。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False, "当前没有开启订阅数据共享功能"
disabled = cls._feature_disabled_result(
'SUBSCRIBE_STATISTIC_SHARE',
"当前没有开启订阅数据共享功能",
)
if disabled:
return disabled
return cls._handle_response(cls.subscribe_fork(share_id))
@classmethod
@@ -1094,8 +1144,12 @@ class MoviePilotServerHelper:
"""
异步复用订阅分享。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return False, "当前没有开启订阅数据共享功能"
disabled = cls._feature_disabled_result(
'SUBSCRIBE_STATISTIC_SHARE',
"当前没有开启订阅数据共享功能",
)
if disabled:
return disabled
return cls._handle_response(await cls.async_subscribe_fork(share_id))
@classmethod
@@ -1113,9 +1167,7 @@ class MoviePilotServerHelper:
"""
获取订阅分享数据。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return []
params = cls._build_subscribe_query_params(
params = cls._build_subscribe_query_plan(
page=page,
count=count,
genre_id=genre_id,
@@ -1124,6 +1176,8 @@ class MoviePilotServerHelper:
sort_type=sort_type,
name=name,
)
if params is None:
return []
return cls._handle_list_response(cls.subscribe_shares(params))
@classmethod
@@ -1141,9 +1195,7 @@ class MoviePilotServerHelper:
"""
异步获取订阅分享数据。
"""
if not get_runtime_setting('SUBSCRIBE_STATISTIC_SHARE'):
return []
params = cls._build_subscribe_query_params(
params = cls._build_subscribe_query_plan(
page=page,
count=count,
genre_id=genre_id,
@@ -1152,6 +1204,8 @@ class MoviePilotServerHelper:
sort_type=sort_type,
name=name,
)
if params is None:
return []
return cls._handle_list_response(await cls.async_subscribe_shares(params))
@classmethod
@@ -1245,6 +1299,17 @@ class MoviePilotServerHelper:
"""
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,
@@ -1288,8 +1353,12 @@ class MoviePilotServerHelper:
"""
删除工作流分享。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
return False, "当前没有开启工作流数据共享功能"
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,
@@ -1300,8 +1369,12 @@ class MoviePilotServerHelper:
"""
异步删除工作流分享。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
return False, "当前没有开启工作流数据共享功能"
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,
@@ -1312,8 +1385,12 @@ class MoviePilotServerHelper:
"""
复用工作流分享。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
return False, "当前没有开启工作流数据共享功能"
disabled = cls._feature_disabled_result(
'WORKFLOW_STATISTIC_SHARE',
"当前没有开启工作流数据共享功能",
)
if disabled:
return disabled
return cls._handle_response(cls.workflow_fork(share_id))
@classmethod
@@ -1321,8 +1398,12 @@ class MoviePilotServerHelper:
"""
异步复用工作流分享。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
return False, "当前没有开启工作流数据共享功能"
disabled = cls._feature_disabled_result(
'WORKFLOW_STATISTIC_SHARE',
"当前没有开启工作流数据共享功能",
)
if disabled:
return disabled
return cls._handle_response(await cls.async_workflow_fork(share_id))
@classmethod
@@ -1336,13 +1417,10 @@ class MoviePilotServerHelper:
"""
获取工作流分享数据。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
params = cls._build_workflow_query_plan(name, page, count)
if params is None:
return []
return cls._handle_list_response(cls.workflow_shares({
"name": name,
"page": page,
"count": count,
}))
return cls._handle_list_response(cls.workflow_shares(params))
@classmethod
@cached(region="workflow_share", maxsize=1, skip_empty=True)
@@ -1355,13 +1433,10 @@ class MoviePilotServerHelper:
"""
异步获取工作流分享数据。
"""
if not get_runtime_setting('WORKFLOW_STATISTIC_SHARE'):
params = cls._build_workflow_query_plan(name, page, count)
if params is None:
return []
return cls._handle_list_response(await cls.async_workflow_shares({
"name": name,
"page": page,
"count": count,
}))
return cls._handle_list_response(await cls.async_workflow_shares(params))
@classmethod
def _validate_workflow(cls, workflow) -> Tuple[bool, str]:
@@ -1383,45 +1458,97 @@ class MoviePilotServerHelper:
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]):
"""
查询共享识别结果。
"""
api_url = cls.recognize_share_url()
if not api_url:
plan = cls._build_recognize_transport_plan(params)
if not plan:
return None
return cls._get(api_url, params=params, timeout=5)
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]):
"""
异步查询共享识别结果。
"""
api_url = cls.recognize_share_url()
if not api_url:
plan = cls._build_recognize_transport_plan(params)
if not plan:
return None
return await cls._async_get(api_url, params=params, timeout=5)
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]):
"""
上报共享识别结果。
"""
api_url = cls.recognize_share_url()
if not api_url:
plan = cls._build_recognize_transport_plan(payload)
if not plan:
return None
return cls._post_json(api_url, payload, timeout=5)
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]):
"""
异步上报共享识别结果。
"""
api_url = cls.recognize_share_url()
if not api_url:
plan = cls._build_recognize_transport_plan(payload)
if not plan:
return None
return await cls._async_post_json(api_url, payload, timeout=5)
api_url, request_data = plan
return await cls._async_post_json(api_url, request_data, timeout=5)
@classmethod
def query_recognize_share(
@@ -1434,18 +1561,17 @@ class MoviePilotServerHelper:
"""
查询共享识别结果。
"""
if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'):
return None
params = cls._build_recognize_query_params(
plan = cls._build_recognize_query_plan(
meta=meta,
mtype=mtype,
keyword_meta=keyword_meta,
music_type=music_type,
)
if not params:
if not plan:
return None
params, keyword = plan
response = cls.recognize_query(params)
return cls._parse_recognize_response(response, params.get("keyword"))
return cls._parse_recognize_response(response, keyword)
@classmethod
async def async_query_recognize_share(
@@ -1458,18 +1584,17 @@ class MoviePilotServerHelper:
"""
异步查询共享识别结果。
"""
if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'):
return None
params = cls._build_recognize_query_params(
plan = cls._build_recognize_query_plan(
meta=meta,
mtype=mtype,
keyword_meta=keyword_meta,
music_type=music_type,
)
if not params:
if not plan:
return None
params, keyword = plan
response = await cls.async_recognize_query(params)
return cls._parse_recognize_response(response, params.get("keyword"))
return cls._parse_recognize_response(response, keyword)
@classmethod
def report_recognize_share(
@@ -1481,14 +1606,12 @@ class MoviePilotServerHelper:
"""
上报共享识别结果,电影、电视剧、音乐共用。
"""
if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'):
return False
payload = cls._build_recognize_report_payload(
payload = cls._build_recognize_report_plan(
meta=meta,
mediainfo=mediainfo,
keyword_meta=keyword_meta,
)
if not payload:
if payload is None:
return False
response = cls.recognize_report(payload)
return cls._parse_recognize_report_response(response)
@@ -1503,14 +1626,12 @@ class MoviePilotServerHelper:
"""
异步上报共享识别结果,电影、电视剧、音乐共用。
"""
if not get_runtime_setting('MEDIA_RECOGNIZE_SHARE'):
return False
payload = cls._build_recognize_report_payload(
payload = cls._build_recognize_report_plan(
meta=meta,
mediainfo=mediainfo,
keyword_meta=keyword_meta,
)
if not payload:
if payload is None:
return False
response = await cls.async_recognize_report(payload)
return cls._parse_recognize_report_response(response)
+471
View File
@@ -0,0 +1,471 @@
"""MoviePilot 中心服务适配器同步异步一致性测试。"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
from app.adapters.external import server as server_module
from app.adapters.external.server import MoviePilotServerHelper
@pytest.fixture(autouse=True)
def _reset_server_identity_cache():
"""隔离中心服务用户名缓存,避免同步入口短路异步入口。"""
previous_user = MoviePilotServerHelper._github_user
MoviePilotServerHelper._github_user = None
yield
MoviePilotServerHelper._github_user = previous_user
def _response(status_code: int, payload: object) -> SimpleNamespace:
"""构造不会触发真实网络的最小 HTTP 响应。"""
return SimpleNamespace(status_code=status_code, json=Mock(return_value=payload))
def _settings(monkeypatch, **values) -> None:
"""只为当前用例注入中心服务配置。"""
monkeypatch.setattr(
server_module,
"get_runtime_setting",
lambda key: values.get(key),
)
@pytest.mark.asyncio
async def test_github_user_sync_async_use_same_request_plan(monkeypatch) -> None:
"""GitHub 用户同步异步入口必须使用同一凭据、代理和响应映射。"""
response = _response(200, {"login": "moviepilot"})
sync_client = Mock()
sync_client.get_res.return_value = response
async_client = Mock()
async_client.get_res = AsyncMock(return_value=response)
sync_factory = Mock(return_value=sync_client)
async_factory = Mock(return_value=async_client)
_settings(
monkeypatch,
GITHUB_HEADERS={"Authorization": "token"},
PROXY={"https": "http://proxy"},
)
monkeypatch.setattr(server_module, "RequestUtils", sync_factory)
monkeypatch.setattr(server_module, "AsyncRequestUtils", async_factory)
sync_result = MoviePilotServerHelper.get_github_user()
MoviePilotServerHelper._github_user = None
async_result = await MoviePilotServerHelper.async_get_github_user()
assert sync_result == async_result == "moviepilot"
assert sync_factory.call_args.kwargs == async_factory.call_args.kwargs == {
"headers": {"Authorization": "token"},
"proxies": {"https": "http://proxy"},
"timeout": 15,
}
sync_client.get_res.assert_called_once_with("https://api.github.com/user")
async_client.get_res.assert_awaited_once_with("https://api.github.com/user")
@pytest.mark.asyncio
@pytest.mark.parametrize(
("status_code", "payload", "expected"),
[
(200, {"is_admin": True}, {"is_admin": True}),
(403, {"message": "forbidden"}, {}),
],
)
async def test_user_permissions_sync_async_classify_response_identically(
monkeypatch,
status_code: int,
payload: dict,
expected: dict,
) -> None:
"""用户权限同步异步入口必须共享 HTTP 状态和对象响应分类。"""
response = _response(status_code, payload)
monkeypatch.setattr(
MoviePilotServerHelper,
"get_github_user",
Mock(return_value="moviepilot"),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_get_github_user",
AsyncMock(return_value="moviepilot"),
)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "user_permissions", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_user_permissions",
async_request,
)
sync_result = MoviePilotServerHelper.get_user_permissions()
async_result = await MoviePilotServerHelper.async_get_user_permissions()
assert sync_result == async_result == expected
sync_request.assert_called_once_with("moviepilot")
async_request.assert_awaited_once_with("moviepilot")
@pytest.mark.asyncio
async def test_user_permissions_sync_async_map_transport_failure_identically(
monkeypatch,
) -> None:
"""用户权限同步异步传输异常都必须回退为空权限。"""
monkeypatch.setattr(
MoviePilotServerHelper,
"get_github_user",
Mock(return_value="moviepilot"),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_get_github_user",
AsyncMock(return_value="moviepilot"),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"user_permissions",
Mock(side_effect=RuntimeError("offline")),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_user_permissions",
AsyncMock(side_effect=RuntimeError("offline")),
)
assert MoviePilotServerHelper.get_user_permissions() == {}
assert await MoviePilotServerHelper.async_get_user_permissions() == {}
@pytest.mark.asyncio
@pytest.mark.parametrize(
("enabled", "user_uid", "status_code", "expected"),
[
(False, "uid-1", 200, False),
(True, "", 200, False),
(True, "uid-1", 500, False),
(True, "uid-1", 200, True),
],
)
async def test_usage_report_sync_async_share_preflight_and_result_mapping(
monkeypatch,
enabled: bool,
user_uid: str,
status_code: int,
expected: bool,
) -> None:
"""版本统计开关、实例身份和成功状态必须由同一计划解释。"""
payload = {"user_uid": user_uid, "backend_version": "3.0.0"}
response = _response(status_code, {})
_settings(monkeypatch, USAGE_STATISTIC_SHARE=enabled)
monkeypatch.setattr(
MoviePilotServerHelper,
"build_usage_payload",
Mock(return_value=payload),
)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "usage_report", sync_request)
monkeypatch.setattr(MoviePilotServerHelper, "async_usage_report", async_request)
sync_result = MoviePilotServerHelper.report_usage()
async_result = await MoviePilotServerHelper.async_report_usage()
assert sync_result == async_result == expected
if enabled and user_uid:
sync_request.assert_called_once_with(payload)
async_request.assert_awaited_once_with(payload)
else:
sync_request.assert_not_called()
async_request.assert_not_awaited()
@pytest.mark.asyncio
async def test_usage_report_sync_async_map_transport_failure_identically(
monkeypatch,
) -> None:
"""版本统计同步异步传输异常都必须映射为上报失败。"""
_settings(monkeypatch, USAGE_STATISTIC_SHARE=True)
monkeypatch.setattr(
MoviePilotServerHelper,
"build_usage_payload",
Mock(return_value={"user_uid": "uid-1"}),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"usage_report",
Mock(side_effect=RuntimeError("offline")),
)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_usage_report",
AsyncMock(side_effect=RuntimeError("offline")),
)
assert MoviePilotServerHelper.report_usage() is False
assert await MoviePilotServerHelper.async_report_usage() is False
@pytest.mark.asyncio
async def test_plugin_install_sync_async_share_payload_and_result(monkeypatch) -> None:
"""插件安装统计同步异步入口必须使用同一脱敏载荷和成功分类。"""
response = _response(200, {})
_settings(monkeypatch, PLUGIN_STATISTIC_SHARE=True)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "plugin_install", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_plugin_install",
async_request,
)
sync_result = MoviePilotServerHelper.install_plugin_reg(
"DemoPlugin",
"local://DemoPlugin?version=v3",
)
async_result = await MoviePilotServerHelper.async_install_plugin_reg(
"DemoPlugin",
"local://DemoPlugin?version=v3",
)
expected_payload = {
"plugin_id": "DemoPlugin",
"repo_url": "local://DemoPlugin?version=v3",
}
assert sync_result == async_result is True
sync_request.assert_called_once_with("DemoPlugin", expected_payload)
async_request.assert_awaited_once_with("DemoPlugin", expected_payload)
@pytest.mark.asyncio
async def test_plugin_statistic_sync_async_share_response_classification(
monkeypatch,
) -> None:
"""插件统计同步异步入口必须共享功能开关和对象响应分类。"""
response = _response(200, {"DemoPlugin": 3})
_settings(monkeypatch, PLUGIN_STATISTIC_SHARE=True)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "plugin_statistic", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_plugin_statistic",
async_request,
)
MoviePilotServerHelper.get_plugin_statistic.cache_clear()
sync_result = MoviePilotServerHelper.get_plugin_statistic()
async_result = await MoviePilotServerHelper.async_get_plugin_statistic()
assert sync_result == async_result == {"DemoPlugin": 3}
sync_request.assert_called_once_with()
async_request.assert_awaited_once_with()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("sync_name", "async_name", "sync_transport", "async_transport"),
[
("sub_reg", "async_sub_reg", "subscribe_add", "async_subscribe_add"),
("sub_done", "async_sub_done", "subscribe_done", "async_subscribe_done"),
],
)
async def test_subscribe_statistic_sync_async_share_payload_and_result(
monkeypatch,
sync_name: str,
async_name: str,
sync_transport: str,
async_transport: str,
) -> None:
"""订阅新增与完成统计必须共享功能开关、载荷生成和成功分类。"""
response = _response(200, {})
payload = {"media_source": "tmdb", "media_id": "1"}
_settings(monkeypatch, SUBSCRIBE_STATISTIC_SHARE=True)
payload_builder = Mock(return_value=payload)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(
MoviePilotServerHelper,
"_build_subscribe_statistic_payload",
payload_builder,
)
monkeypatch.setattr(MoviePilotServerHelper, sync_transport, sync_request)
monkeypatch.setattr(MoviePilotServerHelper, async_transport, async_request)
sync_result = getattr(MoviePilotServerHelper, sync_name)({"id": 1})
async_result = await getattr(MoviePilotServerHelper, async_name)({"id": 1})
assert sync_result == async_result is True
assert payload_builder.call_count == 2
sync_request.assert_called_once_with(payload)
async_request.assert_awaited_once_with(payload)
@pytest.mark.asyncio
async def test_subscribe_list_sync_async_share_query_and_result(monkeypatch) -> None:
"""订阅分享列表同步异步入口必须共享筛选参数和列表响应分类。"""
response = _response(200, [{"id": 1}])
_settings(monkeypatch, SUBSCRIBE_STATISTIC_SHARE=True)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "subscribe_shares", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_subscribe_shares",
async_request,
)
MoviePilotServerHelper.get_subscribe_shares.cache_clear()
await MoviePilotServerHelper.async_get_subscribe_shares.cache_clear()
sync_result = MoviePilotServerHelper.get_subscribe_shares(
name="Demo",
page=2,
count=10,
genre_id=16,
min_rating=7.0,
sort_type="rating",
)
await MoviePilotServerHelper.async_get_subscribe_shares.cache_clear()
async_result = await MoviePilotServerHelper.async_get_subscribe_shares(
name="Demo",
page=2,
count=10,
genre_id=16,
min_rating=7.0,
sort_type="rating",
)
expected_params = {
"page": 2,
"count": 10,
"name": "Demo",
"genre_id": 16,
"min_rating": 7.0,
"sort_type": "rating",
}
assert sync_result == async_result == [{"id": 1}]
sync_request.assert_called_once_with(expected_params)
async_request.assert_awaited_once_with(expected_params)
@pytest.mark.asyncio
async def test_workflow_list_sync_async_share_query_and_result(monkeypatch) -> None:
"""工作流分享列表同步异步入口必须共享开关、分页计划和结果分类。"""
response = _response(200, [{"id": 2}])
_settings(monkeypatch, WORKFLOW_STATISTIC_SHARE=True)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(MoviePilotServerHelper, "workflow_shares", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_workflow_shares",
async_request,
)
MoviePilotServerHelper.get_workflow_shares.cache_clear()
await MoviePilotServerHelper.async_get_workflow_shares.cache_clear()
sync_result = MoviePilotServerHelper.get_workflow_shares("Demo", 3, 20)
await MoviePilotServerHelper.async_get_workflow_shares.cache_clear()
async_result = await MoviePilotServerHelper.async_get_workflow_shares(
"Demo", 3, 20
)
expected_params = {"name": "Demo", "page": 3, "count": 20}
assert sync_result == async_result == [{"id": 2}]
sync_request.assert_called_once_with(expected_params)
async_request.assert_awaited_once_with(expected_params)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("setting_key", "sync_name", "async_name", "message"),
[
(
"SUBSCRIBE_STATISTIC_SHARE",
"share_delete",
"async_share_delete",
"当前没有开启订阅数据共享功能",
),
(
"WORKFLOW_STATISTIC_SHARE",
"workflow_share_delete_by_id",
"async_workflow_share_delete_by_id",
"当前没有开启工作流数据共享功能",
),
],
)
async def test_share_delete_sync_async_use_same_disabled_result(
monkeypatch,
setting_key: str,
sync_name: str,
async_name: str,
message: str,
) -> None:
"""订阅与工作流删除入口必须共享禁用状态和失败文案。"""
_settings(monkeypatch, **{setting_key: False})
sync_result = getattr(MoviePilotServerHelper, sync_name)(9)
async_result = await getattr(MoviePilotServerHelper, async_name)(9)
assert sync_result == async_result == (False, message)
@pytest.mark.asyncio
async def test_recognize_query_sync_async_share_plan_and_result(monkeypatch) -> None:
"""共享识别查询必须共用启用判断、查询计划和业务响应解析。"""
response = _response(200, {"code": 0, "data": {"item": {"media_id": "1"}}})
params = {"keyword": "Demo", "type": "movie"}
_settings(monkeypatch, MEDIA_RECOGNIZE_SHARE=True)
builder = Mock(return_value=params)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(
MoviePilotServerHelper,
"_build_recognize_query_params",
builder,
)
monkeypatch.setattr(MoviePilotServerHelper, "recognize_query", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_recognize_query",
async_request,
)
sync_result = MoviePilotServerHelper.query_recognize_share(None)
async_result = await MoviePilotServerHelper.async_query_recognize_share(None)
assert sync_result == async_result == {"media_id": "1"}
assert builder.call_count == 2
sync_request.assert_called_once_with(params)
async_request.assert_awaited_once_with(params)
@pytest.mark.asyncio
async def test_recognize_report_sync_async_share_plan_and_result(monkeypatch) -> None:
"""共享识别上报必须共用启用判断、载荷计划和业务成功分类。"""
response = _response(200, {"code": 0})
payload = {"keyword": "Demo", "media_source": "tmdb", "media_id": "1"}
_settings(monkeypatch, MEDIA_RECOGNIZE_SHARE=True)
builder = Mock(return_value=payload)
sync_request = Mock(return_value=response)
async_request = AsyncMock(return_value=response)
monkeypatch.setattr(
MoviePilotServerHelper,
"_build_recognize_report_payload",
builder,
)
monkeypatch.setattr(MoviePilotServerHelper, "recognize_report", sync_request)
monkeypatch.setattr(
MoviePilotServerHelper,
"async_recognize_report",
async_request,
)
sync_result = MoviePilotServerHelper.report_recognize_share(None, None)
async_result = await MoviePilotServerHelper.async_report_recognize_share(None, None)
assert sync_result == async_result is True
assert builder.call_count == 2
sync_request.assert_called_once_with(payload)
async_request.assert_awaited_once_with(payload)