From cc5fc2a89bcdd0c7c408b363985fce2304113e68 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 16 Oct 2026 23:46:16 +0800 Subject: [PATCH] refactor(types): define TMDb request and JSON response contracts --- app/modules/themoviedb/tmdbv3api/tmdb.py | 225 +++++++++++++++-------- mypy.ini | 1 + 2 files changed, 148 insertions(+), 78 deletions(-) diff --git a/app/modules/themoviedb/tmdbv3api/tmdb.py b/app/modules/themoviedb/tmdbv3api/tmdb.py index 5eb94bf39..4542eb60b 100644 --- a/app/modules/themoviedb/tmdbv3api/tmdb.py +++ b/app/modules/themoviedb/tmdbv3api/tmdb.py @@ -3,8 +3,10 @@ import asyncio import logging import time +from collections.abc import Callable, Mapping from copy import deepcopy from datetime import datetime +from typing import Protocol, TypeAlias, TypedDict, TypeVar, Union, cast import requests import requests.exceptions @@ -15,6 +17,38 @@ from app.runtime.settings import get_runtime_setting from .exceptions import TMDbConnectionError, TMDbException +TMDbJson: TypeAlias = Union[dict[str, "TMDbJson"], list["TMDbJson"], str, int, float, bool, None] +_RequestFunction = TypeVar("_RequestFunction", bound=Callable[..., object]) + + +class _HttpResponse(Protocol): + """requests 与 httpx 响应在 TMDb 解析层使用的最小接口。""" + + @property + def headers(self) -> Mapping[str, str]: + """读取保持大小写兼容语义的响应头。""" + ... + + def json(self) -> TMDbJson: + """解析 JSON；无效内容按客户端原有规则抛出异常。""" + ... + + +class _ResponseSnapshot(TypedDict): + """可序列化响应快照的固定字段，另带运行期快照标记。""" + + headers: dict[str, str] + json: TMDbJson + + +class _ClearableRequest(Protocol): + """缓存装饰器为同步请求附加的清理入口。""" + + def cache_clear(self) -> None: + """清空当前请求函数绑定的缓存区。""" + ... + + logger = logging.getLogger(__name__) # 单次重试前的退避等待时间（秒）。NAS+FUSE网盘等环境下TMDB连接的失败大多是数秒内 @@ -24,7 +58,7 @@ logger = logging.getLogger(__name__) RETRY_BACKOFF_SECONDS = 2 -def _is_business_failure_snapshot(snapshot) -> bool: +def _is_business_failure_snapshot(snapshot: object) -> bool: """ 判断响应快照是否为TMDB业务失败（success=false，如404/限流/服务端错误的合法JSON）。 @@ -38,7 +72,7 @@ def _is_business_failure_snapshot(snapshot) -> bool: return isinstance(json_data, dict) and json_data.get("success") is False -def _is_empty_result_snapshot(snapshot) -> bool: +def _is_empty_result_snapshot(snapshot: object) -> bool: """ 判断响应快照是否为空结果（列表/搜索类接口的 results 为空列表）。 @@ -55,22 +89,35 @@ def _is_empty_result_snapshot(snapshot) -> bool: return isinstance(results, list) and not results +def _cache_request(function: _RequestFunction) -> _RequestFunction: + """保留请求的调用类型；cached 仅附加缓存管理入口而不改变返回值。""" + return cast(_RequestFunction, cached( + maxsize=get_runtime_setting('CONF').tmdb, + ttl=get_runtime_setting('CONF').meta, + skip_none=True, + skip_if=_is_business_failure_snapshot, + empty_ttl=get_runtime_setting('EMPTY_RESULT_CACHE_TTL'), + empty_if=_is_empty_result_snapshot, + )(function)) + + class TMDb(object): """封装 TMDB 请求、响应缓存及有界的连接和限流重试。""" _RESPONSE_SNAPSHOT_MARKER = "__mp_tmdb_response_snapshot__" - def __init__(self, session=None, language=None): - self._api_key = get_runtime_setting('TMDB_API_KEY') - self._language = language or get_runtime_setting('TMDB_LOCALE') or "en-US" - self._session_id = None + def __init__(self, session: requests.Session | None = None, language: str | None = None) -> None: + """保存配置快照，并按调用方会话所有权初始化双入口客户端。""" + self._api_key: str | None = get_runtime_setting('TMDB_API_KEY') + self._language: str | None = language or get_runtime_setting('TMDB_LOCALE') or "en-US" + self._session_id: str | None = None self._session = session self._wait_on_rate_limit = True - self._proxies = get_runtime_setting('PROXY') - self._domain = get_runtime_setting('TMDB_API_DOMAIN') - self._page = None - self._total_results = None - self._total_pages = None + self._proxies: dict[str, str] | None = get_runtime_setting('PROXY') + self._domain: str | None = get_runtime_setting('TMDB_API_DOMAIN') + self._page: TMDbJson = None + self._total_results: TMDbJson = None + self._total_pages: TMDbJson = None self._owns_session = session is None self._init_session(session=session) @@ -83,10 +130,10 @@ class TMDb(object): ) self._remaining = 40 - self._reset = None + self._reset: int | None = None self._timeout = 15 - def _init_session(self, session=None): + def _init_session(self, session: requests.Session | None = None) -> None: """ 初始化同步请求会话。 """ @@ -97,7 +144,7 @@ class TMDb(object): proxies=self.proxies, ) - def _reset_owned_session(self): + def _reset_owned_session(self) -> None: """ 重建当前实例自持有的同步请求会话。 """ @@ -111,75 +158,90 @@ class TMDb(object): self._init_session() @property - def page(self): + def page(self) -> TMDbJson: + """返回响应携带的页码，不转换上游字段。""" return self._page @property - def total_results(self): + def total_results(self) -> TMDbJson: + """返回响应携带的结果总数。""" return self._total_results @property - def total_pages(self): + def total_pages(self) -> TMDbJson: + """返回响应携带的总页数。""" return self._total_pages @property - def api_key(self): + def api_key(self) -> str | None: + """读取当前 API Key。""" return self._api_key - @property - def domain(self): - return self._domain - - @property - def proxies(self): - return self._proxies - - @proxies.setter - def proxies(self, proxies): - self._proxies = proxies - @api_key.setter - def api_key(self, api_key): + def api_key(self, api_key: object) -> None: + """沿用旧入口的字符串转换规则保存 API Key。""" self._api_key = str(api_key) + @property + def domain(self) -> str | None: + """读取当前 API 域名。""" + return self._domain + @domain.setter - def domain(self, domain): + def domain(self, domain: object) -> None: + """沿用旧入口的字符串转换规则保存 API 域名。""" self._domain = str(domain) @property - def language(self): + def proxies(self) -> dict[str, str] | None: + """读取当前 HTTP 代理映射。""" + return self._proxies + + @proxies.setter + def proxies(self, proxies: dict[str, str] | None) -> None: + """保留调用方传入的代理映射。""" + self._proxies = proxies + + @property + def language(self) -> str | None: + """读取请求使用的语言。""" return self._language @language.setter - def language(self, language): + def language(self, language: str | None) -> None: + """保存调用方指定的语言。""" self._language = language @property - def has_session(self): + def has_session(self) -> bool: + """判断是否已保存认证会话标识。""" return True if self._session_id else False @property - def session_id(self): + def session_id(self) -> str: + """读取认证会话；未认证时保持原异常。""" if not self._session_id: raise TMDbException("Must Authenticate to create a session run Authentication(username, password)") return self._session_id @session_id.setter - def session_id(self, session_id): + def session_id(self, session_id: str | None) -> None: + """保存或清空认证会话标识。""" self._session_id = session_id @property - def wait_on_rate_limit(self): + def wait_on_rate_limit(self) -> bool: + """读取是否等待服务端限流重置。""" return self._wait_on_rate_limit @wait_on_rate_limit.setter - def wait_on_rate_limit(self, wait_on_rate_limit): + def wait_on_rate_limit(self, wait_on_rate_limit: object) -> None: + """按原有真值规则设置限流等待开关。""" self._wait_on_rate_limit = bool(wait_on_rate_limit) - @cached(maxsize=get_runtime_setting('CONF').tmdb, ttl=get_runtime_setting('CONF').meta, skip_none=True, - skip_if=_is_business_failure_snapshot, - empty_ttl=get_runtime_setting('EMPTY_RESULT_CACHE_TTL'), empty_if=_is_empty_result_snapshot) - def request(self, method, url, data, json, **kwargs): + @_cache_request + def request(self, method: str, url: str, data: dict[str, TMDbJson] | None, json: dict[str, TMDbJson] | None, **kwargs: object) -> _ResponseSnapshot: + """发送同步请求并缓存隔离快照，仅在自持有会话下重试连接失败。""" req = self._request_once(method, url, data, json) if req is None and method == "GET" and self._owns_session: logger.debug(f"TMDB同步请求失败，等待{RETRY_BACKOFF_SECONDS}秒后重建会话重试一次") @@ -194,18 +256,17 @@ class TMDb(object): raise TMDbConnectionError("无法连接TheMovieDb，请检查网络连接！") return self._snapshot_response(req) - def _request_once(self, method, url, data, json): + def _request_once(self, method: str, url: str, data: dict[str, TMDbJson] | None, json: dict[str, TMDbJson] | None) -> _HttpResponse | None: """ 执行一次TMDB同步请求，调用方负责决定是否重建会话后重试。 """ if method == "GET": - return self._req.get_res(url, params=data, json=json) - return self._req.post_res(url, data=data, json=json) + return cast(_HttpResponse | None, self._req.get_res(url, params=data, json=json)) + return cast(_HttpResponse | None, self._req.post_res(url, data=data, json=json)) - @cached(maxsize=get_runtime_setting('CONF').tmdb, ttl=get_runtime_setting('CONF').meta, skip_none=True, - skip_if=_is_business_failure_snapshot, - empty_ttl=get_runtime_setting('EMPTY_RESULT_CACHE_TTL'), empty_if=_is_empty_result_snapshot) - async def async_request(self, method, url, data, json, **kwargs): + @_cache_request + async def async_request(self, method: str, url: str, data: dict[str, TMDbJson] | None, json: dict[str, TMDbJson] | None, **kwargs: object) -> _ResponseSnapshot: + """发送异步请求，连接失败时最多退避重试一次。""" req = await self._async_request_once(method, url, data, json) if req is None: logger.debug(f"TMDB异步请求失败，等待{RETRY_BACKOFF_SECONDS}秒后重试一次") @@ -217,43 +278,45 @@ class TMDb(object): raise TMDbConnectionError("无法连接TheMovieDb，请检查网络连接！") return self._snapshot_response(req) - async def _async_request_once(self, method, url, data, json): + async def _async_request_once(self, method: str, url: str, data: dict[str, TMDbJson] | None, json: dict[str, TMDbJson] | None) -> _HttpResponse | None: """ 执行一次TMDB异步请求，调用方负责决定是否重试。 """ if method == "GET": - return await self._async_req.get_res(url, params=data, json=json) - return await self._async_req.post_res(url, data=data, json=json) + return cast(_HttpResponse | None, await self._async_req.get_res(url, params=data, json=json)) + return cast(_HttpResponse | None, await self._async_req.post_res(url, data=data, json=json)) @classmethod - def _snapshot_response(cls, response): + def _snapshot_response(cls, response: _HttpResponse) -> _ResponseSnapshot: """ 生成可缓存的响应快照，并在入缓存前拦截明显异常的TMDB响应结构。 """ json_data = cls._decode_response_json(response) cls._validate_json_response(json_data) # Redis 不能稳定序列化 requests/httpx 响应对象，缓存里只保留当前流程会用到的数据。 - return { + return cast(_ResponseSnapshot, { cls._RESPONSE_SNAPSHOT_MARKER: True, "headers": dict(response.headers.items()), "json": json_data, - } + }) @classmethod - def _get_response_headers(cls, response): + def _get_response_headers(cls, response: _HttpResponse | _ResponseSnapshot) -> Mapping[str, str]: + """兼容缓存快照与原始 HTTP 响应的响应头。""" if isinstance(response, dict) and response.get(cls._RESPONSE_SNAPSHOT_MARKER): return response.get("headers") or {} - return response.headers + return cast(_HttpResponse, response).headers @classmethod - def _get_response_json(cls, response): + def _get_response_json(cls, response: _HttpResponse | _ResponseSnapshot) -> TMDbJson: + """读取原始 JSON，并隔离缓存命中的可变数据。""" if isinstance(response, dict) and response.get(cls._RESPONSE_SNAPSHOT_MARKER): # 调用方会补充 media_type 等字段，缓存快照必须隔离这些原地修改。 return deepcopy(response.get("json")) - return cls._decode_response_json(response) + return cls._decode_response_json(cast(_HttpResponse, response)) @classmethod - def _decode_response_json(cls, response): + def _decode_response_json(cls, response: _HttpResponse) -> TMDbJson: """ 解析TMDB响应JSON，并把空响应、代理错误页或错误编码的响应统一转换为TMDB异常。 """ @@ -263,7 +326,7 @@ class TMDb(object): raise TMDbException(cls._build_invalid_json_message(response, err)) from err @staticmethod - def _get_header_value(headers, name): + def _get_header_value(headers: Mapping[str, str], name: str) -> str | None: """ 从不同响应头对象中按大小写兼容读取指定响应头。 """ @@ -284,7 +347,7 @@ class TMDb(object): return None @classmethod - def _build_invalid_json_message(cls, response, parse_error: Exception = None): + def _build_invalid_json_message(cls, response: object, parse_error: Exception | None = None) -> str: """ 生成非JSON响应的诊断信息，避免日志只保留JSONDecodeError文本。 """ @@ -323,7 +386,7 @@ class TMDb(object): return "，".join(message_parts) @staticmethod - def _validate_json_response(json_data): + def _validate_json_response(json_data: object) -> None: """ 校验TMDB响应JSON顶层结构，避免代理错误页等标量值继续按字典解析。 """ @@ -339,7 +402,7 @@ class TMDb(object): ) @staticmethod - def _get_json_key(json_data, key): + def _get_json_key(json_data: TMDbJson, key: str) -> TMDbJson: """ 从TMDB对象响应中读取指定字段，避免异常顶层结构触发AttributeError。 """ @@ -350,14 +413,17 @@ class TMDb(object): ) return json_data.get(key) - def cache_clear(self): - return self.request.cache_clear() + def cache_clear(self) -> None: + """通过缓存装饰器附加的入口清理同步请求缓存。""" + return cast(_ClearableRequest, self.request).cache_clear() - def _validate_api_key(self): + def _validate_api_key(self) -> None: + """在发送请求前拒绝缺失 API Key。""" if self.api_key is None or self.api_key == "": raise TMDbException("TheMovieDb API Key 未设置！") - def _build_url(self, action, params=""): + def _build_url(self, action: str, params: str = "") -> str: + """按旧格式拼接 API 路径、认证参数和语言。""" return "https://%s/3%s?api_key=%s&%s&language=%s" % ( self.domain, action, @@ -366,7 +432,7 @@ class TMDb(object): self.language, ) - def _handle_headers(self, headers): + def _handle_headers(self, headers: Mapping[str, str] | None) -> None: """只记录当前响应的限流状态，缺失字段不能沿用上次耗尽的配额。""" normalized_headers = { str(key).lower(): value for key, value in dict(headers or {}).items() @@ -376,7 +442,7 @@ class TMDb(object): reset = normalized_headers.get("x-ratelimit-reset") self._reset = int(reset) if reset is not None else None - def _handle_rate_limit(self): + def _handle_rate_limit(self) -> int: """计算尚未到期的限流等待；缺少重置时间时直接报告限流。""" if self._remaining < 1: if not self.wait_on_rate_limit or self._reset is None: @@ -384,7 +450,7 @@ class TMDb(object): return max(0, self._reset - int(time.time())) return 0 - def _process_json_response(self, json_data, is_async=False): + def _process_json_response(self, json_data: TMDbJson, is_async: bool = False) -> None: """ 从TMDB对象响应中记录分页信息；数组响应没有分页字段，直接跳过。 """ @@ -401,7 +467,7 @@ class TMDb(object): self._total_pages = json_data["total_pages"] @staticmethod - def _handle_errors(json_data): + def _handle_errors(json_data: TMDbJson) -> None: """ 将TMDB标准错误字段转换为统一异常，非对象响应由结构校验提前处理。 """ @@ -414,8 +480,9 @@ class TMDb(object): if "success" in json_data and json_data["success"] is False: raise TMDbException(json_data["status_message"]) - def _request_obj(self, action, params="", call_cached=True, - method="GET", data=None, json=None, key=None): + def _request_obj(self, action: str, params: str = "", call_cached: bool = True, + method: str = "GET", data: dict[str, TMDbJson] | None = None, json: dict[str, TMDbJson] | None = None, + key: str | None = None) -> TMDbJson: """同步解析响应；限流最多等待重试一次，避免递归耗尽请求调用栈。""" self._validate_api_key() url = self._build_url(action, params) @@ -450,8 +517,9 @@ class TMDb(object): return self._get_json_key(json_data, key) return json_data - async def _async_request_obj(self, action, params="", call_cached=True, - method="GET", data=None, json=None, key=None): + async def _async_request_obj(self, action: str, params: str = "", call_cached: bool = True, + method: str = "GET", data: dict[str, TMDbJson] | None = None, json: dict[str, TMDbJson] | None = None, + key: str | None = None) -> TMDbJson: """异步解析响应，沿用同步入口的单次限流重试上限。""" self._validate_api_key() url = self._build_url(action, params) @@ -486,6 +554,7 @@ class TMDb(object): return self._get_json_key(json_data, key) return json_data - def close(self): + def close(self) -> None: + """关闭当前同步会话，保持旧调用方所有权约定。""" if self._session: self._session.close() diff --git a/mypy.ini b/mypy.ini index efa110b58..cbb3b503b 100644 --- a/mypy.ini +++ b/mypy.ini @@ -172,4 +172,5 @@ files = app/db/adapters/workflow.py, app/api/context.py, app/api/dependencies/subscription.py, + app/modules/themoviedb/tmdbv3api/tmdb.py, scripts/architecture/async_blocking.py