refactor(types): govern TVDB client response and query contracts

This commit is contained in:
jxxghp
2026-10-17 01:36:31 +08:00
parent 9b19f3d143
commit 7ae81fd395
2 changed files with 284 additions and 83 deletions

View File

@@ -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()

View File

@@ -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"}