nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-31 12:24:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(types): make ZSpace response contracts explicit

Browse Source
This commit is contained in:
jxxghp
2026-10-17 20:05:48 +08:00
parent 0bde181702
commit 74489ca32f
5 changed files with 161 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
app/modules/zspace/zspace.py
View File

@@ -6,7 +6,7 @@ import json
import re
import traceback
from pathlib import Path
from typing import Any, Dict, Generator, List, Optional, Tuple, Union
from typing import Any, Dict, Generator, List, Mapping, Optional, Tuple, Union
from requests import Response
@@ -27,6 +27,24 @@ from app.schemas.types import MediaSource, MediaType
DEFAULT_ITEMS_PAGE_SIZE = 100
def _optional_text(value: object) -> Optional[str]:
"""
读取服务端可空文本字段，不将非法类型隐式转成文本。
"""
if value is None or isinstance(value, str):
return value
raise TypeError("极影视文本字段格式错误")
def _user_id(value: object) -> Optional[Union[str, int]]:
"""
读取兼容字符串和整数的用户标识，保留空值。
"""
if value is None or isinstance(value, (str, int)):
return value
raise TypeError("极影视用户标识格式错误")
class ZSpace:
"""
通过极影视的 Emby 兼容接口提供媒体检索、登录和播放信息。
@@ -43,7 +61,7 @@ class ZSpace:
def __init__(self, host: Optional[str] = None, username: Optional[str] = None,
password: Optional[str] = None, play_host: Optional[str] = None,
sync_libraries: list = None, **kwargs):
sync_libraries: Optional[list[str]] = None, **kwargs: object) -> None:
"""
保存连接配置并登录，随后缓存媒体库路径和服务器标识。
@@ -69,8 +87,8 @@ class ZSpace:
self._password = password
self._sync_libraries = sync_libraries or []
self.user = None
self.folders = []
self.serverid = None
self.folders: List[dict[str, Any]] = []
self.serverid: Optional[str] = None
if not self.reconnect():
logger.error(f"请检查极影视服务端地址 {host}")
@@ -98,7 +116,7 @@ class ZSpace:
def __request_utils(self,
timeout: Optional[int] = None,
include_token: bool = True,
headers: Optional[dict] = None) -> RequestUtils:
headers: Optional[dict[str, str]] = None) -> RequestUtils:
"""
统一构造极影视请求客户端。
@@ -156,7 +174,7 @@ class ZSpace:
self.serverid = self.get_server_id()
return True
def get_library_folders(self) -> List[dict]:
def get_library_folders(self) -> List[dict[str, Any]]:
"""
获取极影视媒体库路径列表。
@@ -172,13 +190,16 @@ class ZSpace:
try:
res = self.__request_utils().get_res(url)
if res:
return res.json()
folders: object = res.json()
if not isinstance(folders, list) or not all(isinstance(folder, dict) for folder in folders):
raise TypeError("极影视媒体库目录响应格式错误")
return folders
logger.debug("Library/SelectableMediaFolders 未获取到返回数据，回退到 Users/{uid}/Views")
except Exception as e:
logger.debug(f"连接Library/SelectableMediaFolders 出错：{e}，回退到 Users/{{uid}}/Views")
return self.__get_library_views() or []
def get_virtual_folders(self) -> List[dict]:
def get_virtual_folders(self) -> List[dict[str, Any]]:
"""
获取极影视媒体库所有路径列表（包含共享路径）。
@@ -230,7 +251,7 @@ class ZSpace:
})
return libraries
def __get_library_views(self, username: Optional[str] = None) -> Optional[List[dict]]:
def __get_library_views(self, username: Optional[str] = None) -> Optional[List[dict[str, Any]]]:
"""
获取极影视媒体库列表。
"""
@@ -322,13 +343,13 @@ class ZSpace:
if user_name:
for user in users:
if user.get("Name") == user_name:
return user.get("Id")
return _user_id(user.get("Id"))
for user in users:
if user.get("Policy", {}).get("IsAdministrator"):
return user.get("Id")
return _user_id(user.get("Id"))
for user in users:
if user.get("Id"):
return user.get("Id")
return _user_id(user.get("Id"))
else:
logger.debug("Users 返回数据格式错误，回退到当前登录用户")
else:
@@ -363,7 +384,7 @@ class ZSpace:
try:
res = self.__request_utils().get_res(url)
if res:
return res.json().get("Id")
return _optional_text(res.json().get("Id"))
else:
logger.error("System/Info 未获取到返回数据")
except Exception as e:
@@ -387,6 +408,8 @@ class ZSpace:
if res:
count = res.json().get("TotalRecordCount")
if count:
if not isinstance(count, int):
raise TypeError("极影视用户数量格式错误")
return count
else:
# 极影视未实现该端点会走到这里，降为 debug 避免主日志误报
@@ -497,7 +520,7 @@ class ZSpace:
return total, "tvshows"
return total, None
def __get_series_id_by_name(self, name: str, year: str) -> Optional[str]:
def __get_series_id_by_name(self, name: Optional[str], year: Optional[Union[str, int]]) -> Optional[str]:
"""
在最多 10 条电视剧搜索结果中匹配标题和年份。
@@ -525,7 +548,7 @@ class ZSpace:
for res_item in res_items:
if res_item.get('Name') == name and (
not year or str(res_item.get('ProductionYear')) == str(year)):
return res_item.get('Id')
return _optional_text(res_item.get('Id'))
except Exception as e:
logger.error(f"连接Items出错：{e}")
return None
@@ -533,7 +556,7 @@ class ZSpace:
def get_movies(self,
title: str,
year: Optional[str] = None,
year: Optional[Union[str, int]] = None,
media_source: Optional[MediaSource] = None,
media_id: Optional[str] = None) -> Optional[List[_SchemaMediaServerItem]]:
"""
@@ -619,7 +642,7 @@ class ZSpace:
def get_tv_episodes(self,
item_id: Optional[str] = None,
title: Optional[str] = None,
year: Optional[str] = None,
year: Optional[Union[str, int]] = None,
media_source: Optional[MediaSource] = None,
media_id: Optional[str] = None,
season: Optional[int] = None
@@ -670,7 +693,7 @@ class ZSpace:
if res_json:
tv_item = res_json.json()
res_items = tv_item.get("Items")
season_episodes = {}
season_episodes: Dict[int, List[int]] = {}
for res_item in res_items or []:
season_index = res_item.get("ParentIndexNumber")
if season_index is None:
@@ -707,7 +730,7 @@ class ZSpace:
if images:
for image in images:
if image.get("ProviderName") == "TheMovieDb" and image.get("Type") == image_type:
return image.get("Url")
return _optional_text(image.get("Url"))
logger.info("Items/RemoteImages 未获取到返回数据，采用本地图片")
return self.generate_external_image_link(item_id, image_type)
except Exception as e:
@@ -731,7 +754,7 @@ class ZSpace:
res = self.__request_utils().get_res(url)
if res and res.status_code != 404:
logger.info(f"影片图片链接:{res.url}")
return res.url
return _optional_text(res.url)
else:
logger.info(f"Items/Id/Images 未获取到返回数据或无该影片{image_type}图片")
return None
@@ -834,7 +857,7 @@ class ZSpace:
try:
subfolder_path = Path(subfolder.get("Path"))
if item_path.is_relative_to(subfolder_path):
return folder.get("Id")
return _optional_text(folder.get("Id"))
except Exception as err:
logger.debug(f"匹配子目录出错：{err} - {traceback.format_exc()}")
if item.category:
@@ -843,11 +866,11 @@ class ZSpace:
if subfolder.get("Path") and re.search(
r"[/\\]%s" % re.escape(item.category), subfolder.get("Path")
):
return folder.get("Id")
return _optional_text(folder.get("Id"))
return "/"
@staticmethod
def __format_item_info(item) -> Optional[_SchemaMediaServerItem]:
def __format_item_info(item: Mapping[str, object]) -> Optional[_SchemaMediaServerItem]:
"""
通过统一 Emby 系转换契约生成极影视条目。
"""
@@ -874,7 +897,7 @@ class ZSpace:
logger.error(f"连接/Users/{self.user}/Items/{itemid}出错：{e}")
return None
def get_items_count(self, parent: Union[str, int]) -> Optional[int]:
def get_items_count(self, parent: Optional[Union[str, int]]) -> Optional[int]:
"""
获取指定媒体库的媒体条目总数。
@@ -901,7 +924,7 @@ class ZSpace:
return None
def get_items(self, parent: Union[str, int], start_index: Optional[int] = 0,
limit: Optional[int] = -1) -> Generator[MediaServerItem | None | Any, Any, None]:
limit: Optional[int] = -1) -> Generator[MediaServerItem | None, None, None]:
"""
分页遍历媒体库，展开合集并补查缺少身份或路径的条目。
@@ -917,7 +940,7 @@ class ZSpace:
return None
url = f"{self._host}emby/Users/{self.user}/Items"
fetch_all = limit is None or limit == -1
page_size = DEFAULT_ITEMS_PAGE_SIZE if fetch_all else limit
page_size = DEFAULT_ITEMS_PAGE_SIZE if limit is None or limit == -1 else limit
current_start_index = max(start_index or 0, 0)
while True:
params = {
@@ -976,7 +999,7 @@ class ZSpace:
break
return None
def get_webhook_message(self, form: Any, args: dict) -> Optional[_SchemaWebhookEventInfo]:
def get_webhook_message(self, form: Any, args: Mapping[str, object]) -> Optional[_SchemaWebhookEventInfo]:
"""
解析兼容 Emby 的极影视通知，并补充媒体身份和图片。
@@ -1085,16 +1108,19 @@ class ZSpace:
"""
if not self._host or not self._apikey:
return None
user = self.user or ''
if not isinstance(user, str):
raise TypeError(f"replace() argument 2 must be str, not {type(user).__name__}")
url = url.replace("[HOST]", self._host or '') \
.replace("[APIKEY]", self._apikey or '') \
.replace("[USER]", self.user or '')
.replace("[USER]", user)
try:
return self.__request_utils(headers={"Content-Type": "application/json"}).get_res(url=url)
except Exception as e:
logger.error(f"连接极影视出错：{e}")
return None
def post_data(self, url: str, data: Optional[str] = None, headers: dict = None) -> Optional[Response]:
def post_data(self, url: str, data: Optional[str] = None, headers: Optional[dict[str, str]] = None) -> Optional[Response]:
"""
替换连接占位符并发送自定义 POST 请求。
@@ -1107,9 +1133,12 @@ class ZSpace:
"""
if not self._host or not self._apikey:
return None
user = self.user or ''
if not isinstance(user, str):
raise TypeError(f"replace() argument 2 must be str, not {type(user).__name__}")
url = url.replace("[HOST]", self._host or '') \
.replace("[APIKEY]", self._apikey or '') \
.replace("[USER]", self.user or '')
.replace("[USER]", user)
try:
return self.__request_utils(headers=headers).post_res(url=url, data=data)
except Exception as e:
@@ -1146,7 +1175,7 @@ class ZSpace:
return f"{host_url}emby/Items/{item_id}/" \
f"Images/Backdrop?tag={image_tag}&api_key={self._apikey}"
def __get_local_image_by_id(self, item_id: str) -> str:
def __get_local_image_by_id(self, item_id: object) -> str:
"""
根据ItemId从媒体服务器查询本地图片地址。
@@ -1186,7 +1215,7 @@ class ZSpace:
res = self.__request_utils().get_res(url, params=params)
if res:
result = res.json().get("Items") or []
ret_resume = []
ret_resume: List[_SchemaMediaServerPlayItem] = []
library_folders = self.get_user_library_folders()
for item in result:
if len(ret_resume) == num:
@@ -1271,7 +1300,7 @@ class ZSpace:
result = payload.get("Items") or []
else:
result = payload or []
ret_latest = []
ret_latest: List[_SchemaMediaServerPlayItem] = []
library_folders = self.get_user_library_folders()
for item in result:
if len(ret_latest) == num:
@@ -1302,7 +1331,7 @@ class ZSpace:
logger.error(f"连接 Users/Items（DateCreated 排序）出错：{e}")
return None
def get_user_library_folders(self):
def get_user_library_folders(self) -> List[str]:
"""
汇总同步列表中明确选中的媒体库路径。
@@ -1316,7 +1345,10 @@ class ZSpace:
for library in self.get_virtual_folders() or []:
if self._sync_libraries and library.get("Id") not in self._sync_libraries:
continue
library_folders += [folder for folder in library.get("Path")]
paths = library.get("Path")
if paths is None:
raise TypeError("'NoneType' object is not iterable")
library_folders += [folder for folder in paths]
return library_folders
def __login(self, username: Optional[str], password: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
@@ -1352,7 +1384,7 @@ class ZSpace:
logger.error(f"连接Users/AuthenticateByName出错：{e}")
return None, None
def __get_current_user(self) -> Optional[dict]:
def __get_current_user(self) -> Optional[dict[str, Any]]:
"""
获取当前登录用户信息。
@@ -1374,13 +1406,13 @@ class ZSpace:
logger.error(f"连接 {url} 出错：{e}")
return None
def __get_current_user_id(self) -> Optional[str]:
def __get_current_user_id(self) -> Optional[Union[str, int]]:
"""
获取当前登录用户ID。
"""
current_user = self.__get_current_user()
if current_user:
current_user_id = current_user.get("Id")
current_user_id = _user_id(current_user.get("Id"))
if current_user_id:
self.user = current_user_id
return current_user_id

2
docs/refactor/optimization-checklist.md
View File

@@ -108,7 +108,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
| Python 源码量 | 373,625 行 | 本次重新扫描物理行数，排除 `app/plugins/**`；74 个文件超过 1,000 行，12 个超过 2,000 行；源码尺寸仅作观察 |
| 长方法 | 309 个超过 80 行 | 本次 AST 扫描排除 `app/plugins/**`；61 个超过 150 行，17 个超过 250 行；新增函数仍受 C901/嵌套深度门禁约束 |
| 全量 mypy 历史债务 | 4,348 / 250 文件 | 本分支为 Pydantic 字段补充描述并保留默认值，减少 3 项 assignment 错误；本次 NFO 返回类型与契约对齐，减少 2 项历史类型错误；本轮在消息、站点、诊断、备份、用户配置、订阅、AcoustID、TMDB、仪表板和命令生命周期边界补齐类型，历史债务再减少 53 项；TMDB v3 API 显式导出并保留集合 parts 的列表合同，逐文件逐错误码无增长；此前工作流筛选、动态插件路由和 stdio 流补齐边界类型，历史债务减少 3 项；Bitpt 与 CookieCloud 补齐边界类型，历史债务减少 16 项；知行站点解析器补齐数值属性与基类合同，历史债务减少 18 项；通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项；Transmission 进度回退值归一化再减少 2 项；本批清理 Servarr、认证、存储、订阅策略和搜索/规则/通知 Schema 的 18 项类型错误，并修复系统配置中的 Python 2 异常语法；本批再清理订阅元数据、订阅写入、交互宿主、整理路径、删除统计和工作流动作的 14 项类型错误，逐文件逐错误码无增长；本批补齐 OCR 请求参数、工具 Schema 摘要和宿主模块配置快照的 5 项边界类型，逐文件逐错误码无增长；本批为 Rousi 用户消息分页解析补齐返回类型与数据结构兜底，减少 2 项类型错误；本批优先处理 TMDB v3 电影与电视剧对象模型，补齐 98 个同步/异步方法注解并减少 122 项历史类型错误；随后补齐剧集、季度、人物、搜索和趋势对象的 142 个方法注解，再减少 142 项类型错误；绿联 API 响应边界补齐具体泛型与运行时字典收窄，减少 31 项类型错误；运行时限流器补齐状态方法、装饰器边界和时间队列类型，减少 31 项类型错误；QQ Bot API 响应边界补齐具体泛型与 JSON 形状校验，减少 17 项类型错误；rTorrent 客户端清理 182 项错误，调用端同步明确字典合同，累计再减少 190 项错误；豆瓣 API、中心服务接口及订阅消费端补齐返回边界和参数收窄，结合 Context 方法签名与缓存代理类型清理，累计再减少 455 项错误；共享 HTTP 客户端清理 110 项错误，TMDB API 清理 84 项错误，索引器爬虫清理 87 项错误，并补充 U115、AliPan、OpenList 对可空流响应的防守；U115 其余存储基类契约债务暂留待测试完备后处理；缓存同步/异步继承冲突保留为设计债务，未纳入严格门禁 |
| 全量 mypy 历史债务 | 4,305 / 249 文件 | 本分支为 Pydantic 字段补充描述并保留默认值，减少 3 项 assignment 错误；本次 NFO 返回类型与契约对齐，减少 2 项历史类型错误；本轮在消息、站点、诊断、备份、用户配置、订阅、AcoustID、TMDB、仪表板和命令生命周期边界补齐类型，历史债务再减少 53 项；TMDB v3 API 显式导出并保留集合 parts 的列表合同，逐文件逐错误码无增长；此前工作流筛选、动态插件路由和 stdio 流补齐边界类型，历史债务减少 3 项；Bitpt 与 CookieCloud 补齐边界类型，历史债务减少 16 项；知行站点解析器补齐数值属性与基类合同，历史债务减少 18 项；通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项；Transmission 进度回退值归一化再减少 2 项；本批清理 Servarr、认证、存储、订阅策略和搜索/规则/通知 Schema 的 18 项类型错误，并修复系统配置中的 Python 2 异常语法；本批再清理订阅元数据、订阅写入、交互宿主、整理路径、删除统计和工作流动作的 14 项类型错误，逐文件逐错误码无增长；本批补齐 OCR 请求参数、工具 Schema 摘要和宿主模块配置快照的 5 项边界类型，逐文件逐错误码无增长；本批为 Rousi 用户消息分页解析补齐返回类型与数据结构兜底，减少 2 项类型错误；本批优先处理 TMDB v3 电影与电视剧对象模型，补齐 98 个同步/异步方法注解并减少 122 项历史类型错误；随后补齐剧集、季度、人物、搜索和趋势对象的 142 个方法注解，再减少 142 项类型错误；绿联 API 响应边界补齐具体泛型与运行时字典收窄，减少 31 项类型错误；运行时限流器补齐状态方法、装饰器边界和时间队列类型，减少 31 项类型错误；QQ Bot API 响应边界补齐具体泛型与 JSON 形状校验，减少 17 项类型错误；rTorrent 客户端清理 182 项错误，调用端同步明确字典合同，累计再减少 190 项错误；豆瓣 API、中心服务接口及订阅消费端补齐返回边界和参数收窄，结合 Context 方法签名与缓存代理类型清理，累计再减少 455 项错误；共享 HTTP 客户端清理 110 项错误，TMDB API 清理 84 项错误，索引器爬虫清理 87 项错误，并补充 U115、AliPan、OpenList 对可空流响应的防守；U115 其余存储基类契约债务暂留待测试完备后处理；缓存同步/异步继承冲突保留为设计债务，未纳入严格门禁 |
| Ruff 历史诊断 | 73 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
| 覆盖率逐文件低水位 | 全 app（排除 app/plugins）当前聚合 77.68% | 存量文件比例不降；语句减少且未覆盖行数不增时允许比例变化，新文件至少 80% |

1
mypy.ini
View File

@@ -7,6 +7,7 @@ follow_imports = skip
show_error_codes = True
pretty = True
files =
app/modules/zspace/zspace.py,
app/modules/jellyfin/jellyfin.py,
app/modules/ugreen/ugreen.py,
app/adapters/external/ocr.py,

11
tests/fixtures/architecture/mypy-baseline.json vendored
View File

@@ -1305,17 +1305,6 @@
"no-untyped-def": 2,
"type-arg": 1
},
"app/modules/zspace/zspace.py": {
"arg-type": 9,
"assignment": 3,
"no-any-return": 12,
"no-untyped-call": 2,
"no-untyped-def": 3,
"operator": 1,
"type-arg": 8,
"union-attr": 1,
"var-annotated": 4
},
"app/monitor/monitor.py": {
"arg-type": 8,
"assignment": 1,

89
tests/test_zspace_type_boundaries.py Normal file
View File

@@ -0,0 +1,89 @@
"""
极影视外部响应边界及自定义请求失败语义回归。
"""
from unittest.mock import Mock, patch
import pytest
from app.modules.zspace.zspace import ZSpace
def _client() -> ZSpace:
"""
构造不访问网络的已配置客户端。
"""
client = ZSpace.__new__(ZSpace)
client._host = "http://zspace.local/"
client._apikey = "token"
client._sync_libraries = []
client.user = "user"
return client
@pytest.mark.parametrize("payload", [{"bad": True}, ["bad"]])
def test_invalid_selectable_folders_uses_existing_views_fallback(payload):
"""
目录响应形状错误时继续使用现有视图回退，保留视图的扩展字段。
"""
client = _client()
views = [{"Id": "library", "Extra": {"enabled": True}}]
response = Mock()
response.json.return_value = payload
with patch("app.modules.zspace.zspace.RequestUtils") as transport, patch.object(
client, "_ZSpace__get_library_views", return_value=views
) as get_views:
transport.return_value.get_res.return_value = response
assert client.get_library_folders() == views
get_views.assert_called_once_with()
@pytest.mark.parametrize("method", ["get_data", "post_data"])
@pytest.mark.parametrize("user", [5, 0, None, "user"])
def test_custom_requests_preserve_user_placeholder_type_error(method, user):
"""
非零整数用户仍在请求前抛出 TypeError，空用户仍替换为空文本。
"""
client = _client()
client.user = user
with patch("app.modules.zspace.zspace.RequestUtils") as transport:
if user == 5:
with pytest.raises(TypeError, match="replace"):
getattr(client, method)("[HOST][USER]")
transport.assert_not_called()
else:
getattr(client, method)("[HOST][USER]")
request = transport.return_value.get_res if method == "get_data" else transport.return_value.post_res
assert request.call_args.kwargs["url"] == f"http://zspace.local/{user or ''}"
def test_missing_library_path_still_raises_outside_request_handling():
"""
缺失路径仍抛出原有 TypeError，不静默跳过错误的库信息。
"""
client = _client()
with patch.object(client, "get_virtual_folders", return_value=[{"Id": "library"}]):
with pytest.raises(TypeError, match="NoneType"):
client.get_user_library_folders()
@pytest.mark.parametrize("identifier", ["user", 42, None])
def test_current_user_identity_preserves_integer_and_null(identifier):
"""
当前用户查询保留整数、文本和空标识，不转换为字符串。
"""
client = _client()
with patch.object(client, "_ZSpace__get_current_user", return_value={"Id": identifier}):
assert client._ZSpace__get_current_user_id() == identifier
def test_invalid_server_identifier_uses_existing_error_result():
"""
非文本服务器标识走现有异常处理结果，不产生虚假的文本标识。
"""
client = _client()
response = Mock()
response.json.return_value = {"Id": ["invalid"]}
with patch("app.modules.zspace.zspace.RequestUtils") as transport:
transport.return_value.get_res.return_value = response
assert client.get_server_id() is None