mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-04-05 03:38:36 +08:00
Merge pull request #4365 from Aqr-K/fix-modules/thetvdb
This commit is contained in:
jxxghp
@@ -1,3 +1,4 @@
|
||||
from threading import Lock
|
||||
from typing import Optional, Tuple, Union
|
||||
|
||||
from app.core.config import settings
|
||||
@@ -8,41 +9,83 @@ from app.schemas.types import ModuleType, MediaRecognizeType
|
||||
|
||||
|
||||
class TheTvDbModule(_ModuleBase):
|
||||
tvdb: tvdb_v4_official.TVDB = None
|
||||
"""
|
||||
TVDB媒体信息匹配
|
||||
"""
|
||||
__timeout: int = 15
|
||||
tvdb: Optional[tvdb_v4_official.TVDB] = None
|
||||
__auth_lock = Lock()
|
||||
|
||||
def init_module(self) -> None:
|
||||
self._initialize_tvdb_session()
|
||||
pass
|
||||
|
||||
def _initialize_tvdb_session(self) -> None:
|
||||
def _initialize_tvdb_session(self, is_retry: bool = False) -> None:
|
||||
"""
|
||||
创建或刷新 TVDB 登录会话
|
||||
创建或刷新 TVDB 登录会话。
|
||||
:param is_retry: 是否是由于token失效后的重试登录
|
||||
"""
|
||||
action = "刷新" if is_retry else "创建"
|
||||
logger.info(f"开始{action}TVDB登录会话...")
|
||||
try:
|
||||
if not settings.TVDB_V4_API_KEY:
|
||||
raise ConnectionError("TVDB API Key 未配置，无法初始化会话。")
|
||||
self.tvdb = tvdb_v4_official.TVDB(apikey=settings.TVDB_V4_API_KEY,
|
||||
pin=settings.TVDB_V4_API_PIN,
|
||||
proxy=settings.PROXY)
|
||||
proxy=settings.PROXY,
|
||||
timeout=self.__timeout)
|
||||
if self.tvdb:
|
||||
logger.info(f"TVDB登录会话{action}成功。")
|
||||
else:
|
||||
raise ValueError(f"TVDB登录会话{action}后实例仍为None。")
|
||||
except Exception as e:
|
||||
logger.error(f"TVDB 登录失败: {str(e)}")
|
||||
self.tvdb = None
|
||||
raise ConnectionError(f"TVDB登录会话{action}失败: {str(e)}") from e
|
||||
|
||||
def _handle_tvdb_call(self, func, *args, **kwargs):
|
||||
def _ensure_tvdb_session(self, is_retry: bool = False) -> None:
|
||||
"""
|
||||
确保TVDB会话存在。如果不存在或需要强制重新初始化，则进行初始化。
|
||||
:param is_retry: 是否重新初始化（例如token失效时）
|
||||
"""
|
||||
# 第一次检查 (无锁)，提高性能，避免不必要锁竞争
|
||||
if not self.tvdb or is_retry:
|
||||
with self.__auth_lock:
|
||||
# 第二次检查 (有锁)，防止多个线程都通过第一次检查后重复初始化
|
||||
if not self.tvdb or is_retry:
|
||||
self._initialize_tvdb_session(is_retry=is_retry)
|
||||
|
||||
def _handle_tvdb_call(self, method_name: str, *args, **kwargs):
|
||||
"""
|
||||
包裹 TVDB 调用，处理 token 失效情况并尝试重新初始化
|
||||
:param method_name: 要在 self.tvdb 实例上调用的方法的名称 (字符串)
|
||||
"""
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
self._ensure_tvdb_session()
|
||||
actual_method = getattr(self.tvdb, method_name)
|
||||
return actual_method(*args, **kwargs)
|
||||
except ValueError as e:
|
||||
# 检查错误信息中是否包含 token 失效相关描述
|
||||
if "Unauthorized" in str(e):
|
||||
logger.warning("TVDB Token 可能已失效，正在尝试重新登录...")
|
||||
self._initialize_tvdb_session()
|
||||
return func(*args, **kwargs)
|
||||
elif "NotFoundException" in str(e):
|
||||
logger.warning("TVDB 剧集不存在")
|
||||
try:
|
||||
self._ensure_tvdb_session(is_retry=True)
|
||||
actual_method = getattr(self.tvdb, method_name)
|
||||
return actual_method(*args, **kwargs)
|
||||
except ConnectionError as conn_err:
|
||||
logger.error(f"TVDB Token失效后重新登录失败: {conn_err}")
|
||||
raise
|
||||
elif "NotFoundException" in str(e) or "ID not found" in str(e):
|
||||
logger.warning(f"TVDB 资源未找到 (调用 {method_name}): {e}")
|
||||
return None
|
||||
else:
|
||||
logger.error(f"TVDB 调用 ({method_name}) 时发生未处理的 ValueError: {str(e)}")
|
||||
raise
|
||||
except ConnectionError as e:
|
||||
logger.error(f"TVDB 连接会话错误: {str(e)}")
|
||||
raise
|
||||
except AttributeError as e:
|
||||
logger.error(f"TVDB 实例上没有方法 '{method_name}': {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"TVDB 调用出错: {str(e)}")
|
||||
logger.error(f"TVDB 调用时发生未知错误: {str(e)}", exc_info=True)
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
@@ -71,16 +114,16 @@ class TheTvDbModule(_ModuleBase):
|
||||
return 4
|
||||
|
||||
def stop(self):
|
||||
pass
|
||||
logger.info("TheTvDbModule 停止。正在清除 TVDB 会话。")
|
||||
with self.__auth_lock:
|
||||
self.tvdb = None
|
||||
|
||||
def test(self) -> Tuple[bool, str]:
|
||||
"""
|
||||
测试模块连接性
|
||||
"""
|
||||
if not self.tvdb:
|
||||
return False, "TheTVDB 连接失败"
|
||||
try:
|
||||
self._handle_tvdb_call(self.tvdb.get_series, 81189)
|
||||
self._handle_tvdb_call("get_series", 81189)
|
||||
return True, ""
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
@@ -96,7 +139,7 @@ class TheTvDbModule(_ModuleBase):
|
||||
"""
|
||||
try:
|
||||
logger.info(f"开始获取TVDB信息: {tvdbid} ...")
|
||||
return self._handle_tvdb_call(self.tvdb.get_series_extended, tvdbid)
|
||||
return self._handle_tvdb_call("get_series_extended", tvdbid)
|
||||
except Exception as err:
|
||||
logger.error(f"获取TVDB信息失败: {str(err)}")
|
||||
return None
|
||||
@@ -109,8 +152,13 @@ class TheTvDbModule(_ModuleBase):
|
||||
"""
|
||||
try:
|
||||
logger.info(f"开始用标题搜索TVDB剧集: {title} ...")
|
||||
res = self._handle_tvdb_call(self.tvdb.search, title)
|
||||
return [item for item in res if item.get("type") == "series"]
|
||||
res = self._handle_tvdb_call("search", title)
|
||||
if res is None:
|
||||
return []
|
||||
if not isinstance(res, list):
|
||||
logger.warning(f"TVDB 搜索 '{title}' 未返回列表：{type(res)}")
|
||||
return []
|
||||
return [item for item in res if isinstance(item, dict) and item.get("type") == "series"]
|
||||
except Exception as err:
|
||||
logger.error(f"用标题搜索TVDB剧集失败: {str(err)}")
|
||||
logger.error(f"用标题搜索TVDB剧集失败 ({title}): {str(err)}")
|
||||
return []
|
||||
|
||||
@@ -11,9 +11,11 @@ from app.utils.http import RequestUtils
|
||||
|
||||
|
||||
class Auth:
|
||||
"""TVDB认证类"""
|
||||
"""
|
||||
TVDB认证类
|
||||
"""
|
||||
|
||||
def __init__(self, url, apikey, pin="", proxy=None):
|
||||
def __init__(self, url, apikey, pin="", proxy=None, timeout: int = 15):
|
||||
login_info = {"apikey": apikey}
|
||||
if pin != "":
|
||||
login_info["pin"] = pin
|
||||
@@ -22,7 +24,7 @@ class Auth:
|
||||
|
||||
try:
|
||||
# 使用项目统一的RequestUtils类
|
||||
req_utils = RequestUtils(proxies=proxy, timeout=30)
|
||||
req_utils = RequestUtils(proxies=proxy, timeout=timeout)
|
||||
response = req_utils.post_res(
|
||||
url=url,
|
||||
data=login_info_bytes,
|
||||
@@ -45,31 +47,38 @@ class Auth:
|
||||
raise Exception(f"TVDB认证失败: {str(e)}")
|
||||
|
||||
def get_token(self):
|
||||
"""获取认证token"""
|
||||
"""
|
||||
获取认证token
|
||||
"""
|
||||
return self.token
|
||||
|
||||
|
||||
class Request:
|
||||
"""请求处理类"""
|
||||
"""
|
||||
请求处理类
|
||||
"""
|
||||
|
||||
def __init__(self, auth_token, proxy=None):
|
||||
def __init__(self, auth_token, proxy=None, timeout=15):
|
||||
self.auth_token = auth_token
|
||||
self.links = None
|
||||
self.proxy = proxy
|
||||
self.timeout = timeout
|
||||
|
||||
def make_request(self, url, if_modified_since=None):
|
||||
"""Makes a request to the given URL and returns the data"""
|
||||
"""
|
||||
向指定的 URL 发起请求并返回数据
|
||||
"""
|
||||
headers = {"Authorization": f"Bearer {self.auth_token}"}
|
||||
if if_modified_since:
|
||||
headers["If-Modified-Since"] = str(if_modified_since)
|
||||
|
||||
try:
|
||||
# 使用项目统一的RequestUtils类
|
||||
req_utils = RequestUtils(proxies=self.proxy, timeout=30)
|
||||
req_utils = RequestUtils(proxies=self.proxy, timeout=self.timeout)
|
||||
response = req_utils.get_res(url=url, headers=headers)
|
||||
|
||||
if response is None:
|
||||
raise ValueError(f"failed to get {url}\n 网络连接失败")
|
||||
raise ValueError(f"获取 {url} 失败\n 网络连接失败")
|
||||
|
||||
if response.status_code == HTTPStatus.NOT_MODIFIED:
|
||||
return {
|
||||
@@ -84,8 +93,8 @@ class Request:
|
||||
self.links = result.get("links", None)
|
||||
return data
|
||||
|
||||
msg = result.get("message", "UNKNOWN FAILURE")
|
||||
raise ValueError(f"failed to get {url}\n {str(msg)}")
|
||||
msg = result.get("message", "未知错误")
|
||||
raise ValueError(f"获取 {url} 失败\n {str(msg)}")
|
||||
else:
|
||||
# 处理其他HTTP错误状态码
|
||||
try:
|
||||
@@ -93,22 +102,26 @@ class Request:
|
||||
msg = error_data.get("message", f"HTTP {response.status_code}")
|
||||
except Exception as err:
|
||||
msg = f"HTTP {response.status_code} {err}"
|
||||
raise ValueError(f"failed to get {url}\n {str(msg)}")
|
||||
raise ValueError(f"获取 {url} 失败\n {str(msg)}")
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, ValueError):
|
||||
raise
|
||||
raise ValueError(f"failed to get {url}\n {str(e)}")
|
||||
raise ValueError(f"获取 {url} 失败\n {str(e)}")
|
||||
|
||||
|
||||
class Url:
|
||||
"""URL构建类"""
|
||||
"""
|
||||
URL构建类
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.base_url = "https://api4.thetvdb.com/v4/"
|
||||
|
||||
def construct(self, url_sect, url_id=None, url_subsect=None, url_lang=None, **kwargs):
|
||||
"""构建API URL"""
|
||||
"""
|
||||
构建API URL
|
||||
"""
|
||||
url = self.base_url + url_sect
|
||||
if url_id:
|
||||
url += "/" + str(url_id)
|
||||
@@ -124,352 +137,479 @@ class Url:
|
||||
|
||||
|
||||
class TVDB:
|
||||
"""TVDB API主类"""
|
||||
"""
|
||||
TVDB API主类
|
||||
"""
|
||||
|
||||
def __init__(self, apikey: str, pin="", proxy=None):
|
||||
def __init__(self, apikey: str, pin="", proxy=None, timeout: int = 15):
|
||||
self.url = Url()
|
||||
login_url = self.url.construct("login")
|
||||
self.auth = Auth(login_url, apikey, pin, proxy)
|
||||
self.auth = Auth(login_url, apikey, pin, proxy, timeout)
|
||||
auth_token = self.auth.get_token()
|
||||
self.request = Request(auth_token, proxy)
|
||||
self.request = Request(auth_token, proxy, timeout)
|
||||
|
||||
def get_req_links(self) -> dict:
|
||||
"""获取请求链接"""
|
||||
"""
|
||||
获取上一次请求返回的链接信息（例如分页链接）
|
||||
"""
|
||||
return self.request.links
|
||||
|
||||
def get_artwork_statuses(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of artwork statuses"""
|
||||
"""
|
||||
返回艺术图状态列表
|
||||
"""
|
||||
url = self.url.construct("artwork/statuses", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_artwork_types(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of artwork types"""
|
||||
"""
|
||||
返回艺术图类型列表
|
||||
"""
|
||||
url = self.url.construct("artwork/types", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_artwork(self, id: int, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns an artwork dictionary"""
|
||||
"""
|
||||
返回单个艺术图信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an artwork extended dictionary"""
|
||||
"""
|
||||
返回单个艺术图的扩展信息字典
|
||||
"""
|
||||
url = self.url.construct("artwork", id, "extended", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_awards(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of awards"""
|
||||
"""
|
||||
返回奖项列表
|
||||
"""
|
||||
url = self.url.construct("awards", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_award(self, id: int, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns an award dictionary"""
|
||||
"""
|
||||
返回单个奖项信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an award extended dictionary"""
|
||||
"""
|
||||
返回单个奖项的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of award categories"""
|
||||
"""
|
||||
返回奖项类别列表
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an award category dictionary"""
|
||||
"""
|
||||
返回单个奖项类别信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an award category extended dictionary"""
|
||||
"""
|
||||
返回单个奖项类别的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of content ratings"""
|
||||
"""
|
||||
返回内容分级列表
|
||||
"""
|
||||
url = self.url.construct("content/ratings", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_countries(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of countries"""
|
||||
"""
|
||||
返回国家列表
|
||||
"""
|
||||
url = self.url.construct("countries", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_companies(self, page=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of companies"""
|
||||
"""
|
||||
返回公司列表 (可分页)
|
||||
"""
|
||||
url = self.url.construct("companies", page=page, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_company_types(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of company types"""
|
||||
"""
|
||||
返回公司类型列表
|
||||
"""
|
||||
url = self.url.construct("companies/types", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_company(self, id: int, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns a company dictionary"""
|
||||
"""
|
||||
返回单个公司信息的字典
|
||||
"""
|
||||
url = self.url.construct("companies", id, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_series(self, page=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of series"""
|
||||
"""
|
||||
返回剧集列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a series dictionary"""
|
||||
"""
|
||||
返回单个剧集信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a series dictionary"""
|
||||
"""
|
||||
通过 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:
|
||||
"""Returns a series extended dictionary"""
|
||||
"""
|
||||
返回单个剧集的扩展信息字典
|
||||
"""
|
||||
url = self.url.construct("series", id, "extended", meta=meta, short=short)
|
||||
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=None, if_modified_since=None, **kwargs) -> dict:
|
||||
"""Returns a series episodes dictionary"""
|
||||
"""
|
||||
返回指定剧集和季类型的各集信息字典 (可分页，可指定语言)
|
||||
"""
|
||||
url = self.url.construct(
|
||||
"series", id, "episodes/" + season_type, lang, page=page, meta=meta, **kwargs
|
||||
)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_series_translation(self, id: int, lang: str, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns a series translation dictionary"""
|
||||
"""
|
||||
返回剧集的指定语言翻译信息字典
|
||||
"""
|
||||
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:
|
||||
"""Returns a series record with an artwork array"""
|
||||
"""
|
||||
返回包含艺术图数组的剧集记录 (可指定语言和类型)
|
||||
"""
|
||||
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:
|
||||
"""Returns a series extended dictionary"""
|
||||
"""
|
||||
返回剧集的下一播出信息字典
|
||||
"""
|
||||
url = self.url.construct("series", id, "nextAired")
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_movies(self, page=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of movies"""
|
||||
"""
|
||||
返回电影列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a movie dictionary"""
|
||||
"""
|
||||
返回单个电影信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a movie dictionary"""
|
||||
"""
|
||||
通过 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:
|
||||
"""Returns a movie extended dictionary"""
|
||||
"""
|
||||
返回电影的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a movie translation dictionary"""
|
||||
"""
|
||||
返回电影的指定语言翻译信息字典
|
||||
"""
|
||||
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=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of seasons"""
|
||||
"""
|
||||
返回季列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a season dictionary"""
|
||||
"""
|
||||
返回单季信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a season extended dictionary"""
|
||||
"""
|
||||
返回单季的扩展信息字典
|
||||
"""
|
||||
url = self.url.construct("seasons", id, "extended", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_season_types(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of season types"""
|
||||
"""
|
||||
返回季类型列表
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a seasons translation dictionary"""
|
||||
"""
|
||||
返回季的指定语言翻译信息字典
|
||||
"""
|
||||
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=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of episodes"""
|
||||
"""
|
||||
返回集列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an episode dictionary"""
|
||||
"""
|
||||
返回单集信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an episode extended dictionary"""
|
||||
"""
|
||||
返回单集的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an episode translation dictionary"""
|
||||
"""
|
||||
返回单集的指定语言翻译信息字典
|
||||
"""
|
||||
url = self.url.construct("episodes", id, "translations", lang, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
# Support the old name of the function.
|
||||
# 兼容旧函数名。
|
||||
get_episodes_translation = get_episode_translation
|
||||
|
||||
def get_all_genders(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of genders"""
|
||||
"""
|
||||
返回性别列表
|
||||
"""
|
||||
url = self.url.construct("genders", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_genres(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of genres"""
|
||||
"""
|
||||
返回类型（流派）列表
|
||||
"""
|
||||
url = self.url.construct("genres", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_genre(self, id: int, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns a genres dictionary"""
|
||||
"""
|
||||
返回单个类型（流派）信息的字典
|
||||
"""
|
||||
url = self.url.construct("genres", id, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_languages(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of languages"""
|
||||
"""
|
||||
返回语言列表
|
||||
"""
|
||||
url = self.url.construct("languages", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_people(self, page=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of people"""
|
||||
"""
|
||||
返回人物列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a people dictionary"""
|
||||
"""
|
||||
返回单个人物信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a people extended dictionary"""
|
||||
"""
|
||||
返回单个人物的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a people translation dictionary"""
|
||||
"""
|
||||
返回人物的指定语言翻译信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a character dictionary"""
|
||||
"""
|
||||
返回角色信息的字典
|
||||
"""
|
||||
url = self.url.construct("characters", id, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_people_types(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of people types"""
|
||||
"""
|
||||
返回人物类型列表
|
||||
"""
|
||||
url = self.url.construct("people/types", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
# Support the old function name
|
||||
# 兼容旧函数名
|
||||
get_all_people_types = get_people_types
|
||||
|
||||
def get_source_types(self, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of source types"""
|
||||
"""
|
||||
返回来源类型列表
|
||||
"""
|
||||
url = self.url.construct("sources/types", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
# Support the old function name
|
||||
# 兼容旧函数名
|
||||
get_all_sourcetypes = get_source_types
|
||||
|
||||
def get_updates(self, since: int, **kwargs) -> list:
|
||||
"""Returns a list of updates
|
||||
|
||||
Args:
|
||||
since: Timestamp since when to get updates
|
||||
**kwargs: Additional parameters such as page=2, action='update', type='artwork'
|
||||
"""
|
||||
返回更新列表
|
||||
"""
|
||||
url = self.url.construct("updates", since=since, **kwargs)
|
||||
return self.request.make_request(url)
|
||||
|
||||
def get_all_tag_options(self, page=None, meta=None, if_modified_since=None) -> list:
|
||||
"""Returns a list of tag options"""
|
||||
"""
|
||||
返回标签选项列表 (可分页)
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a tag option dictionary"""
|
||||
"""
|
||||
返回单个标签选项信息的字典
|
||||
"""
|
||||
url = self.url.construct("tags/options", id, meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_all_lists(self, page=None, meta=None) -> dict:
|
||||
"""Returns all lists"""
|
||||
"""
|
||||
返回所有公开的列表信息 (可分页)
|
||||
"""
|
||||
url = self.url.construct("lists", page=page, meta=meta)
|
||||
return self.request.make_request(url)
|
||||
|
||||
def get_list(self, id: int, meta=None, if_modified_since=None) -> dict:
|
||||
"""Returns a list dictionary"""
|
||||
"""
|
||||
返回单个列表信息的字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a list dictionary by slug"""
|
||||
"""
|
||||
通过 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=None, if_modified_since=None) -> dict:
|
||||
"""Returns an extended list dictionary"""
|
||||
"""
|
||||
返回单个列表的扩展信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns a list translation dictionary"""
|
||||
"""
|
||||
返回列表的指定语言翻译信息字典
|
||||
"""
|
||||
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=None, if_modified_since=None) -> dict:
|
||||
"""Returns inspiration types"""
|
||||
"""
|
||||
返回灵感类型列表
|
||||
"""
|
||||
url = self.url.construct("inspiration/types", meta=meta)
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def search(self, query, **kwargs) -> list:
|
||||
"""Returns a list of search results"""
|
||||
def search(self, query: str, **kwargs) -> list:
|
||||
"""
|
||||
根据查询字符串进行搜索，并返回结果列表
|
||||
"""
|
||||
url = self.url.construct("search", query=query, **kwargs)
|
||||
return self.request.make_request(url)
|
||||
|
||||
def search_by_remote_id(self, remoteid: str) -> list:
|
||||
"""Returns a list of search results by remote id exact match"""
|
||||
"""
|
||||
通过外部 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:
|
||||
"""Returns a tag option dictionary"""
|
||||
"""
|
||||
返回具有指定 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:
|
||||
"""Returns an entities types dictionary"""
|
||||
"""
|
||||
返回可用的实体类型列表
|
||||
"""
|
||||
url = self.url.construct("entities")
|
||||
return self.request.make_request(url, if_modified_since)
|
||||
|
||||
def get_user_by_id(self, id: int) -> dict:
|
||||
"""Returns a user info dictionary"""
|
||||
"""
|
||||
通过用户 ID 返回用户信息字典
|
||||
"""
|
||||
url = self.url.construct("user", id)
|
||||
return self.request.make_request(url)
|
||||
|
||||
def get_user(self) -> dict:
|
||||
"""Returns a user info dictionary"""
|
||||
"""
|
||||
返回当前认证的用户信息字典
|
||||
"""
|
||||
url = self.url.construct("user")
|
||||
return self.request.make_request(url)
|
||||
|
||||
def get_user_favorites(self) -> dict:
|
||||
"""Returns a user favorites dictionary"""
|
||||
"""
|
||||
返回当前认证用户的收藏夹信息字典
|
||||
"""
|
||||
url = self.url.construct('user/favorites')
|
||||
return self.request.make_request(url)
|
||||
|
||||
Reference in New Issue
Block a user