From 7ae81fd3958fd5a303cc9801360211a1922f48c1 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 17 Oct 2026 01:36:31 +0800 Subject: [PATCH] refactor(types): govern TVDB client response and query contracts --- app/modules/thetvdb/client.py | 197 ++++++++++++++++------------ tests/test_tvdb_client_contracts.py | 170 ++++++++++++++++++++++++ 2 files changed, 284 insertions(+), 83 deletions(-) create mode 100644 tests/test_tvdb_client_contracts.py diff --git a/app/modules/thetvdb/client.py b/app/modules/thetvdb/client.py index 2f8851c38..a97ca70b1 100644 --- a/app/modules/thetvdb/client.py +++ b/app/modules/thetvdb/client.py @@ -5,19 +5,43 @@ __version__ = "1.0.12" import json import urllib.parse +from collections.abc import Callable from http import HTTPStatus +from typing import Protocol, TypeAlias, TypeVar, Union, cast from app.adapters.network.http import RequestUtils from app.runtime.cache import cached from app.runtime.settings import get_runtime_setting +# 保留接口 data 原始 JSON 与 304 状态字典，不伪装成固定列表或对象。 +TVDBJson: TypeAlias = Union[dict[str, "TVDBJson"], list["TVDBJson"], str, int, float, bool, None] +_RequestFunction = TypeVar("_RequestFunction", bound=Callable[..., object]) + + +class _ClearableRequest(Protocol): + """请求缓存装饰器附加的同步清理入口。""" + + def cache_clear(self) -> None: + """清空当前请求函数共享的缓存。""" + ... + + +def _cache_request(function: _RequestFunction) -> _RequestFunction: + """缓存保持原始参数和返回值，类型桥接不改变缓存配置。""" + return cast(_RequestFunction, cached( + maxsize=get_runtime_setting('CONF').tmdb, + ttl=get_runtime_setting('CONF').meta, + skip_none=True, + )(function)) + class Auth: """ TVDB认证类 """ - def __init__(self, url: str, apikey: str, pin: str = "", proxy: dict = None, timeout: int = 15): + def __init__(self, url: str, apikey: str, pin: str = "", proxy: dict[str, str] | None = None, timeout: int = 15) -> None: + """登录 TVDB 并保留服务端返回的 token，失败沿用既有异常。""" login_info = {"apikey": apikey} if pin != "": login_info["pin"] = pin @@ -35,7 +59,7 @@ class Auth: if response and response.status_code == 200: result = response.json() - self.token = result["data"]["token"] + self.token: TVDBJson = result["data"]["token"] else: if response is not None: try: @@ -49,7 +73,7 @@ class Auth: except Exception as e: raise Exception(f"TVDB认证失败: {str(e)}") - def get_token(self): + def get_token(self) -> TVDBJson: """ 获取认证token """ @@ -61,14 +85,15 @@ class Request: 请求处理类 """ - def __init__(self, auth_token: str, proxy: dict = None, timeout: int = 15): + def __init__(self, auth_token: TVDBJson, proxy: dict[str, str] | None = None, timeout: int = 15) -> None: + """保存请求认证、代理与超时，分页 links 初始为空。""" self.auth_token = auth_token - self.links = None + self.links: TVDBJson = None self.proxy = proxy self.timeout = timeout - @cached(maxsize=get_runtime_setting('CONF').tmdb, ttl=get_runtime_setting('CONF').meta, skip_none=True) - def make_request(self, url: str, if_modified_since: bool = None): + @_cache_request + def make_request(self, url: str, if_modified_since: str | bool | None = None) -> TVDBJson: """ 向指定的 URL 发起请求并返回数据 """ @@ -92,7 +117,7 @@ class Request: if response.status_code == 200: result = response.json() - data = result.get("data", None) + data: TVDBJson = result.get("data", None) if data is not None and result.get("status", "failure") != "failure": self.links = result.get("links", None) return data @@ -119,11 +144,12 @@ class Url: URL构建类 """ - def __init__(self): + def __init__(self) -> None: + """初始化 TVDB v4 地址前缀。""" self.base_url = "https://api4.thetvdb.com/v4/" - def construct(self, url_sect: str, url_id: int = None, - url_subsect: str = None, url_lang: str = None, **kwargs): + def construct(self, url_sect: str, url_id: int | str | None = None, + url_subsect: str | None = None, url_lang: str | None = None, **kwargs: object) -> str: """ 构建API URL """ @@ -146,146 +172,147 @@ class TVDB: TVDB API主类 """ - def __init__(self, apikey: str, pin: str = "", proxy: dict = None, timeout: int = 15): + def __init__(self, apikey: str, pin: str = "", proxy: dict[str, str] | None = None, timeout: int = 15) -> None: + """组装地址、认证与请求对象，保留初始化时登录的行为。""" self.url = Url() login_url = self.url.construct("login") self.auth = Auth(login_url, apikey, pin, proxy, timeout) auth_token = self.auth.get_token() self.request = Request(auth_token, proxy, timeout) - def get_req_links(self) -> dict: + def get_req_links(self) -> TVDBJson: """ 获取上一次请求返回的链接信息（例如分页链接） """ return self.request.links - def get_artwork_statuses(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_artwork_statuses(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回艺术图状态列表 """ url = self.url.construct("artwork/statuses", meta=meta) return self.request.make_request(url, if_modified_since) - def get_artwork_types(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_artwork_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回艺术图类型列表 """ url = self.url.construct("artwork/types", meta=meta) return self.request.make_request(url, if_modified_since) - def get_artwork(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_artwork(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个艺术图信息的字典 """ url = self.url.construct("artwork", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_artwork_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_artwork_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个艺术图的扩展信息字典 """ url = self.url.construct("artwork", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_awards(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_awards(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回奖项列表 """ url = self.url.construct("awards", meta=meta) return self.request.make_request(url, if_modified_since) - def get_award(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_award(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个奖项信息的字典 """ url = self.url.construct("awards", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_award_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_award_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个奖项的扩展信息字典 """ url = self.url.construct("awards", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_award_categories(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_award_categories(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回奖项类别列表 """ url = self.url.construct("awards/categories", meta=meta) return self.request.make_request(url, if_modified_since) - def get_award_category(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_award_category(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个奖项类别信息的字典 """ url = self.url.construct("awards/categories", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_award_category_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_award_category_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个奖项类别的扩展信息字典 """ url = self.url.construct("awards/categories", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_content_ratings(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_content_ratings(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回内容分级列表 """ url = self.url.construct("content/ratings", meta=meta) return self.request.make_request(url, if_modified_since) - def get_countries(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_countries(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回国家列表 """ url = self.url.construct("countries", meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_companies(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_companies(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回公司列表 (可分页) """ url = self.url.construct("companies", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_company_types(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_company_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回公司类型列表 """ url = self.url.construct("companies/types", meta=meta) return self.request.make_request(url, if_modified_since) - def get_company(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_company(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个公司信息的字典 """ url = self.url.construct("companies", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_series(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_series(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回剧集列表 (可分页) """ url = self.url.construct("series", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_series(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_series(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个剧集信息的字典 """ url = self.url.construct("series", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_series_by_slug(self, slug: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_series_by_slug(self, slug: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 通过 slug (别名) 返回单个剧集信息的字典 """ url = self.url.construct("series/slug", slug, meta=meta) return self.request.make_request(url, if_modified_since) - def get_series_extended(self, id: int, meta=None, short=False, if_modified_since=None) -> dict: + def get_series_extended(self, id: int, meta: str | None = None, short: bool = False, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个剧集的扩展信息字典 """ @@ -293,7 +320,7 @@ class TVDB: return self.request.make_request(url, if_modified_since) def get_series_episodes(self, id: int, season_type: str = "default", page: int = 0, - lang: str = None, meta: str = None, if_modified_since: bool = None, **kwargs) -> dict: + lang: str | None = None, meta: str | None = None, if_modified_since: str | bool | None = None, **kwargs: object) -> TVDBJson: """ 返回指定剧集和季类型的各集信息字典 (可分页，可指定语言) """ @@ -302,119 +329,119 @@ class TVDB: ) return self.request.make_request(url, if_modified_since) - def get_series_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_series_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回剧集的指定语言翻译信息字典 """ url = self.url.construct("series", id, "translations", lang, meta=meta) return self.request.make_request(url, if_modified_since) - def get_series_artworks(self, id: int, lang: str, type=None, if_modified_since=None) -> dict: + def get_series_artworks(self, id: int, lang: str, type: int | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回包含艺术图数组的剧集记录 (可指定语言和类型) """ url = self.url.construct("series", id, "artworks", lang=lang, type=type) return self.request.make_request(url, if_modified_since) - def get_series_next_aired(self, id: int, if_modified_since=None) -> dict: + def get_series_next_aired(self, id: int, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回剧集的下一播出信息字典 """ url = self.url.construct("series", id, "nextAired") return self.request.make_request(url, if_modified_since) - def get_all_movies(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_movies(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回电影列表 (可分页) """ url = self.url.construct("movies", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_movie(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_movie(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个电影信息的字典 """ url = self.url.construct("movies", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_movie_by_slug(self, slug: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_movie_by_slug(self, slug: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 通过 slug (别名) 返回单个电影信息的字典 """ url = self.url.construct("movies/slug", slug, meta=meta) return self.request.make_request(url, if_modified_since) - def get_movie_extended(self, id: int, meta=None, short=False, if_modified_since=None) -> dict: + def get_movie_extended(self, id: int, meta: str | None = None, short: bool = False, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回电影的扩展信息字典 """ url = self.url.construct("movies", id, "extended", meta=meta, short=short) return self.request.make_request(url, if_modified_since) - def get_movie_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_movie_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回电影的指定语言翻译信息字典 """ url = self.url.construct("movies", id, "translations", lang, meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_seasons(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_seasons(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回季列表 (可分页) """ url = self.url.construct("seasons", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_season(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_season(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单季信息的字典 """ url = self.url.construct("seasons", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_season_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_season_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单季的扩展信息字典 """ url = self.url.construct("seasons", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_season_types(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_season_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回季类型列表 """ url = self.url.construct("seasons/types", meta=meta) return self.request.make_request(url, if_modified_since) - def get_season_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_season_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回季的指定语言翻译信息字典 """ url = self.url.construct("seasons", id, "translations", lang, meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_episodes(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_episodes(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回集列表 (可分页) """ url = self.url.construct("episodes", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_episode(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_episode(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单集信息的字典 """ url = self.url.construct("episodes", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_episode_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_episode_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单集的扩展信息字典 """ url = self.url.construct("episodes", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_episode_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_episode_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单集的指定语言翻译信息字典 """ @@ -424,70 +451,70 @@ class TVDB: # 兼容旧函数名。 get_episodes_translation = get_episode_translation - def get_all_genders(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_genders(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回性别列表 """ url = self.url.construct("genders", meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_genres(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_genres(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回类型（流派）列表 """ url = self.url.construct("genres", meta=meta) return self.request.make_request(url, if_modified_since) - def get_genre(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_genre(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个类型（流派）信息的字典 """ url = self.url.construct("genres", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_languages(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_languages(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回语言列表 """ url = self.url.construct("languages", meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_people(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_people(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回人物列表 (可分页) """ url = self.url.construct("people", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_person(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_person(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个人物信息的字典 """ url = self.url.construct("people", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_person_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_person_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个人物的扩展信息字典 """ url = self.url.construct("people", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_person_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_person_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回人物的指定语言翻译信息字典 """ url = self.url.construct("people", id, "translations", lang, meta=meta) return self.request.make_request(url, if_modified_since) - def get_character(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_character(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回角色信息的字典 """ url = self.url.construct("characters", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_people_types(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_people_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回人物类型列表 """ @@ -497,7 +524,7 @@ class TVDB: # 兼容旧函数名 get_all_people_types = get_people_types - def get_source_types(self, meta: str = None, if_modified_since: bool = None) -> list: + def get_source_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回来源类型列表 """ @@ -507,120 +534,124 @@ class TVDB: # 兼容旧函数名 get_all_sourcetypes = get_source_types - def get_updates(self, since: int, **kwargs) -> list: + def get_updates(self, since: int, *, url_id: int | str | None = None, + url_subsect: str | None = None, url_lang: str | None = None, + **kwargs: object) -> TVDBJson: """ - 返回更新列表 + 返回更新响应，兼容路径扩展参数与任意查询过滤键。 """ - url = self.url.construct("updates", since=since, **kwargs) + url = self.url.construct("updates", url_id, url_subsect, url_lang, since=since, **kwargs) return self.request.make_request(url) - def get_all_tag_options(self, page: int = None, meta: str = None, if_modified_since: bool = None) -> list: + def get_all_tag_options(self, page: int | None = None, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回标签选项列表 (可分页) """ url = self.url.construct("tags/options", page=page, meta=meta) return self.request.make_request(url, if_modified_since) - def get_tag_option(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_tag_option(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个标签选项信息的字典 """ url = self.url.construct("tags/options", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_all_lists(self, page: int = None, meta=None) -> dict: + def get_all_lists(self, page: int | None = None, meta: str | None = None) -> TVDBJson: """ 返回所有公开的列表信息 (可分页) """ url = self.url.construct("lists", page=page, meta=meta) return self.request.make_request(url) - def get_list(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_list(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个列表信息的字典 """ url = self.url.construct("lists", id, meta=meta) return self.request.make_request(url, if_modified_since) - def get_list_by_slug(self, slug: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_list_by_slug(self, slug: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 通过 slug (别名) 返回单个列表信息的字典 """ url = self.url.construct("lists/slug", slug, meta=meta) return self.request.make_request(url, if_modified_since) - def get_list_extended(self, id: int, meta: str = None, if_modified_since: bool = None) -> dict: + def get_list_extended(self, id: int, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回单个列表的扩展信息字典 """ url = self.url.construct("lists", id, "extended", meta=meta) return self.request.make_request(url, if_modified_since) - def get_list_translation(self, id: int, lang: str, meta: str = None, if_modified_since: bool = None) -> dict: + def get_list_translation(self, id: int, lang: str, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回列表的指定语言翻译信息字典 """ url = self.url.construct("lists", id, "translations", lang, meta=meta) return self.request.make_request(url, if_modified_since) - def get_inspiration_types(self, meta: str = None, if_modified_since: bool = None) -> dict: + def get_inspiration_types(self, meta: str | None = None, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回灵感类型列表 """ url = self.url.construct("inspiration/types", meta=meta) return self.request.make_request(url, if_modified_since) - def search(self, query: str, **kwargs) -> list: + def search(self, query: str, *, url_id: int | str | None = None, + url_subsect: str | None = None, url_lang: str | None = None, + **kwargs: object) -> TVDBJson: """ - 根据查询字符串进行搜索，并返回结果列表 + 按查询字符串搜索，兼容路径扩展参数并保留原始 JSON 响应。 """ - url = self.url.construct("search", query=query, **kwargs) + url = self.url.construct("search", url_id, url_subsect, url_lang, query=query, **kwargs) return self.request.make_request(url) - def search_by_remote_id(self, remoteid: str) -> list: + def search_by_remote_id(self, remoteid: str) -> TVDBJson: """ 通过外部 ID 精确匹配搜索，并返回结果列表 """ url = self.url.construct("search/remoteid", remoteid) return self.request.make_request(url) - def get_tags(self, slug: str, if_modified_since=None) -> dict: + def get_tags(self, slug: str, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回具有指定 slug 的标签实体信息字典 (此方法基于的 /entities/{slug} 端点非标准，请谨慎使用) """ url = self.url.construct("entities", url_subsect=slug) return self.request.make_request(url, if_modified_since) - def get_entities_types(self, if_modified_since=None) -> dict: + def get_entities_types(self, if_modified_since: str | bool | None = None) -> TVDBJson: """ 返回可用的实体类型列表 """ url = self.url.construct("entities") return self.request.make_request(url, if_modified_since) - def get_user_by_id(self, id: int) -> dict: + def get_user_by_id(self, id: int) -> TVDBJson: """ 通过用户 ID 返回用户信息字典 """ url = self.url.construct("user", id) return self.request.make_request(url) - def get_user(self) -> dict: + def get_user(self) -> TVDBJson: """ 返回当前认证的用户信息字典 """ url = self.url.construct("user") return self.request.make_request(url) - def get_user_favorites(self) -> dict: + def get_user_favorites(self) -> TVDBJson: """ 返回当前认证用户的收藏夹信息字典 """ url = self.url.construct('user/favorites') return self.request.make_request(url) - def clear_cache(self): + def clear_cache(self) -> None: """ 清除缓存 """ - self.request.make_request.cache_clear() + cast(_ClearableRequest, self.request.make_request).cache_clear() diff --git a/tests/test_tvdb_client_contracts.py b/tests/test_tvdb_client_contracts.py new file mode 100644 index 000000000..4173bb823 --- /dev/null +++ b/tests/test_tvdb_client_contracts.py @@ -0,0 +1,170 @@ +"""TVDB 类型声明必须保留 URL、原始响应、条件请求和缓存的实际行为。""" + +import inspect +import json +from unittest.mock import Mock, call + +import pytest + +from app.modules.thetvdb import client + + +@pytest.fixture(autouse=True) +def clear_request_cache(): + """隔离共享请求缓存，避免测试之间复用响应或遗留缓存状态。""" + client.Request.make_request.cache_clear() + yield + client.Request.make_request.cache_clear() + + +@pytest.fixture +def transport(monkeypatch): + """在统一 HTTP 边界注入内存响应，不建立真实连接。""" + request = Mock() + monkeypatch.setattr(client, "RequestUtils", Mock(return_value=request)) + return request + + +def response(payload, status=200): + """模拟保留原始 JSON 值的 HTTP 响应。""" + return Mock(status_code=status, json=Mock(return_value=payload)) + + +@pytest.mark.parametrize("data", [{"id": 42, "nested": [None, True]}, [1, "x"], {}, [], "text", 0, False]) +def test_request_preserves_raw_data_and_links(transport, data): + """合法 data 包括空容器与标量；分页链接按服务端原值保存。""" + links = {"next": "/series?page=2", "prev": None} + transport.get_res.return_value = response({"status": "success", "data": data, "links": links}) + request = client.Request("token") + + assert request.make_request("https://example.com/series") == data + assert request.links is links + transport.get_res.assert_called_once_with( + url="https://example.com/series", headers={"Authorization": "Bearer token"}, + ) + + +@pytest.mark.parametrize("modified", [None, False, True, "Wed, 21 Oct 2015 07:28:00 GMT"]) +def test_not_modified_preserves_response_and_previous_links(transport, modified): + """304 不解码 JSON、不覆盖分页状态，条件请求头保持原字符串化规则。""" + reply = response(None, status=304) + transport.get_res.return_value = reply + request = client.Request("token") + previous = {"next": "page-2"} + request.links = previous + + assert request.make_request("https://example.com/series", modified) == { + "code": 304, "message": "Not-Modified", + } + assert request.links is previous + reply.json.assert_not_called() + headers = {"Authorization": "Bearer token"} + if modified: + headers["If-Modified-Since"] = str(modified) + assert transport.get_res.call_args.kwargs["headers"] == headers + + +@pytest.mark.parametrize("reply, message", [ + (None, "网络连接失败"), + (response({"message": "denied"}, 401), "denied"), + (response({"status": "failure", "data": {"id": 1}, "message": "failed"}), "failed"), + (response({"status": "success", "data": None, "message": "empty"}), "empty"), + (response([]), "has no attribute 'get'"), +]) +def test_request_failures_keep_errors_and_are_not_cached(transport, reply, message): + """网络、业务和响应形状失败仍抛 ValueError；相同地址随后可以成功。""" + transport.get_res.side_effect = [reply, response({"status": "success", "data": {"id": 9}})] + request = client.Request("token") + with pytest.raises(ValueError, match=message): + request.make_request("https://example.com/retry") + assert request.make_request("https://example.com/retry") == {"id": 9} + assert transport.get_res.call_count == 2 + + +def test_request_keeps_json_value_error_identity(transport): + """原始 JSON ValueError 按既有分支重抛，不重新包装。""" + error = ValueError("invalid json") + transport.get_res.return_value = Mock(status_code=200, json=Mock(side_effect=error)) + with pytest.raises(ValueError) as raised: + client.Request("token").make_request("https://example.com/invalid") + assert raised.value is error + + +def test_cache_hit_and_clear_keep_original_state_transitions(transport): + """缓存命中不重放 links 副作用，清理后再次读取网络并更新分页状态。""" + transport.get_res.side_effect = [ + response({"status": "success", "data": {"id": 1}, "links": {"next": "a"}}), + response({"status": "success", "data": {"id": 2}, "links": {"next": "b"}}), + response({"status": "success", "data": {"id": 3}, "links": None}), + ] + api = object.__new__(client.TVDB) + api.request = client.Request("token") + assert api.request.make_request("https://example.com/a") == {"id": 1} + assert api.request.make_request("https://example.com/b") == {"id": 2} + assert api.request.make_request("https://example.com/a") == {"id": 1} + assert api.get_req_links() == {"next": "b"} + assert transport.get_res.call_count == 2 + api.clear_cache() + assert api.request.make_request("https://example.com/a") == {"id": 3} + assert api.get_req_links() is None + assert transport.get_res.call_count == 3 + + +@pytest.mark.parametrize("method, value, expected", [ + ("search", "中文 空格", "search/12/extended/zho?query=%E4%B8%AD%E6%96%87+%E7%A9%BA%E6%A0%BC&year=2026&flag=False"), + ("get_updates", 123, "updates/12/extended/zho?since=123&year=2026&flag=False"), +]) +def test_dynamic_queries_keep_path_extensions_and_filter_values(method, value, expected): + """原 kwargs 中的路径键仍构建路径，任意过滤键按原值编码且只去掉 None。""" + api = object.__new__(client.TVDB) + api.url = client.Url() + api.request = Mock() + getattr(api, method)(value, url_id=12, url_subsect="extended", url_lang="zho", + year=2026, absent=None, flag=False) + assert api.request.make_request.call_args == call(api.url.base_url + expected) + + +@pytest.mark.parametrize("method", ["search", "get_updates"]) +@pytest.mark.parametrize("options", [{"url_subsect": 1}, {"url_lang": 1}, {"url_sect": "duplicate"}]) +def test_invalid_dynamic_path_parameters_keep_type_error(method, options): + """无效路径类型与重复 section 参数仍在网络请求前失败。""" + api = object.__new__(client.TVDB) + api.url = client.Url() + api.request = Mock() + with pytest.raises(TypeError): + getattr(api, method)("value", **options) + api.request.make_request.assert_not_called() + + +API_METHODS = [ + name for name in dir(client.TVDB) + if (name.startswith("get_") or name.startswith("search")) and name != "get_req_links" +] + + +@pytest.mark.parametrize("method", API_METHODS) +def test_all_endpoint_wrappers_preserve_conditional_response(method): + """每个 API 封装均透传 304 字典，不能因名义上的列表返回类型改变响应。""" + api = object.__new__(client.TVDB) + api.url = client.Url() + result = {"code": 304, "message": "Not-Modified"} + api.request = Mock(make_request=Mock(return_value=result)) + function = getattr(api, method) + arguments = { + name: (42 if name in {"id", "since"} else "value") + for name, parameter in inspect.signature(function).parameters.items() + if parameter.default is inspect.Parameter.empty + and parameter.kind in {inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD} + } + assert function(**arguments) is result + api.request.make_request.assert_called_once() + + +@pytest.mark.parametrize("token", ["token", 42, None]) +def test_auth_keeps_token_and_login_body(transport, token): + """认证保留 JSON token 原值，登录参数不因类型声明发生转换。""" + transport.post_res.return_value = response({"data": {"token": token}}) + auth = client.Auth("https://example.com/login", "apikey", "pin") + assert auth.get_token() == token + assert json.loads(transport.post_res.call_args.kwargs["data"]) == {"apikey": "apikey", "pin": "pin"} + assert transport.post_res.call_args.kwargs["headers"] == {"Content-Type": "application/json"}