nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-04 14:24:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(types): type TMDB API response boundaries

Browse Source
This commit is contained in:
jxxghp
2026-10-17 17:59:13 +08:00
parent 49f7d585b8
commit 653180c8f0
1 changed files with 66 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
app/modules/themoviedb/tmdbapi.py
View File

@@ -95,7 +95,7 @@ class TmdbApi:
tvs = self.search.tv_shows(**self._search_kwargs(title, year, MediaType.TV)) or []
return self._project_search_results(title, tvs, MediaType.TV)
def search_persons(self, name: str) -> List[dict]:
def search_persons(self, name: str) -> List[_MediaRecord]:
"""
查询模糊匹配的所有人物TMDB信息
"""
@@ -112,7 +112,7 @@ class TmdbApi:
collections = self.search.collections(term=name) or []
return self._project_collections(collections)
def get_collection(self, collection_id: int) -> List[dict]:
def get_collection(self, collection_id: int) -> List[_MediaRecord]:
"""
根据合集ID查询合集详情
"""
@@ -896,7 +896,7 @@ class TmdbApi:
return self._project_info(tmdb_info)
@staticmethod
def __get_content_rating(tmdb_info: dict) -> Optional[str]:
def __get_content_rating(tmdb_info: _MediaRecord) -> Optional[str]:
"""
获得tmdb中的内容评级
:param tmdb_info: TMDB信息
@@ -958,12 +958,12 @@ class TmdbApi:
return ratings.get("CN") or ratings.get("US")
@staticmethod
def __update_tmdbinfo_cn_title(tmdb_info: dict):
def __update_tmdbinfo_cn_title(tmdb_info: _MediaRecord) -> None:
"""
更新TMDB信息中的中文名称
"""
def __get_tmdb_chinese_title(tmdbinfo) -> Optional[str]:
def __get_tmdb_chinese_title(tmdbinfo: _MediaRecord) -> Optional[str]:
"""
从别名中获取中文标题
"""
@@ -977,17 +977,18 @@ class TmdbApi:
iso_3166_1 = alternative_title.get("iso_3166_1")
if iso_3166_1 == "CN":
title = alternative_title.get("title")
if title and text_tools.contains_chinese(title) \
if isinstance(title, str) and text_tools.contains_chinese(title) \
and zhconv_convert(title, "zh-hans") == title:
return title
return tmdbinfo.get("title") if tmdbinfo.get("media_type") == MediaType.MOVIE else tmdbinfo.get("name")
title = tmdbinfo.get("title") if tmdbinfo.get("media_type") == MediaType.MOVIE else tmdbinfo.get("name")
return title if isinstance(title, str) else None
# 原标题
org_title = tmdb_info.get("title") \
if tmdb_info.get("media_type") == MediaType.MOVIE \
else tmdb_info.get("name")
# 查找中文名
if not text_tools.contains_chinese(org_title):
if not text_tools.contains_chinese(org_title if isinstance(org_title, str) else ""):
cn_title = __get_tmdb_chinese_title(tmdb_info)
if cn_title and cn_title != org_title:
# 使用中文别名
@@ -1005,12 +1006,12 @@ class TmdbApi:
tmdb_info['name'] = sg_title
@staticmethod
def __update_tmdbinfo_extra_title(tmdb_info: dict):
def __update_tmdbinfo_extra_title(tmdb_info: _MediaRecord) -> None:
"""
更新TMDB信息中的其它语种名称
"""
def __get_tmdb_lang_title(tmdbinfo: dict, lang: Optional[str] = "US") -> Optional[str]:
def __get_tmdb_lang_title(tmdbinfo: _MediaRecord, lang: Optional[str] = "US") -> Optional[str]:
"""
从译名中获取其它语种标题
"""
@@ -1019,8 +1020,9 @@ class TmdbApi:
translations = tmdb_info.get("translations", {}).get("translations", [])
for translation in translations:
if translation.get("iso_3166_1") == lang:
return translation.get("data", {}).get("title") if tmdbinfo.get("media_type") == MediaType.MOVIE \
title = translation.get("data", {}).get("title") if tmdbinfo.get("media_type") == MediaType.MOVIE \
else translation.get("data", {}).get("name")
return title if isinstance(title, str) else None
return None
# 原标题
@@ -1051,7 +1053,7 @@ class TmdbApi:
"translations,"
"release_dates,"
"external_ids",
raise_on_connection_error: bool = False) -> Optional[dict]:
raise_on_connection_error: bool = False) -> Optional[_MediaRecord]:
"""
获取电影的详情
:param tmdbid: TMDB ID
@@ -1173,7 +1175,7 @@ class TmdbApi:
"content_ratings,"
"external_ids,"
"episode_groups",
raise_on_connection_error: bool = False) -> Optional[dict]:
raise_on_connection_error: bool = False) -> Optional[_MediaRecord]:
"""
获取电视剧的详情
:param tmdbid: TMDB ID
@@ -1357,7 +1359,7 @@ class TmdbApi:
logger.error(str(e))
return None
def get_tv_season_detail(self, tmdbid: int, season: int):
def get_tv_season_detail(self, tmdbid: int, season: int) -> _MediaRecord:
"""
获取电视剧季的详情
:param tmdbid: TMDB ID
@@ -1431,7 +1433,7 @@ class TmdbApi:
logger.error(str(e))
return {}
def get_tv_episode_detail(self, tmdbid: int, season: int, episode: int) -> dict:
def get_tv_episode_detail(self, tmdbid: int, season: int, episode: int) -> _MediaRecord:
"""
获取电视剧集的详情
:param tmdbid: TMDB ID
@@ -1448,7 +1450,7 @@ class TmdbApi:
logger.error(str(e))
return {}
def discover_movies(self, params: dict) -> List[dict]:
def discover_movies(self, params: _MediaRecord) -> List[_MediaRecord]:
"""
发现电影
:param params: 参数
@@ -1467,7 +1469,7 @@ class TmdbApi:
logger.error(str(e))
return []
def discover_tvs(self, params: dict) -> List[dict]:
def discover_tvs(self, params: _MediaRecord) -> List[_MediaRecord]:
"""
发现电视剧
:param params: 参数
@@ -1487,7 +1489,7 @@ class TmdbApi:
return []
@staticmethod
def _normalize_trending_infos(infos: Optional[List[dict]]) -> List[dict]:
def _normalize_trending_infos(infos: Optional[List[_MediaRecord]]) -> List[_MediaRecord]:
"""
过滤流行趋势中的人物等非媒体项，并统一电影、电视剧的媒体类型。
"""
@@ -1506,7 +1508,7 @@ class TmdbApi:
ret_infos.append(info)
return ret_infos
def discover_trending(self, page: Optional[int] = 1) -> List[dict]:
def discover_trending(self, page: Optional[int] = 1) -> List[_MediaRecord]:
"""
流行趋势
"""
@@ -1537,7 +1539,7 @@ class TmdbApi:
languages.append(language)
return ",".join(languages)
def get_movie_images(self, tmdbid: int, original_language: Optional[str] = None) -> dict:
def get_movie_images(self, tmdbid: int, original_language: Optional[str] = None) -> _MediaRecord:
"""
获取电影的图片
"""
@@ -1555,7 +1557,7 @@ class TmdbApi:
logger.error(str(e))
return {}
def get_tv_images(self, tmdbid: int, original_language: Optional[str] = None) -> dict:
def get_tv_images(self, tmdbid: int, original_language: Optional[str] = None) -> _MediaRecord:
"""
获取电视剧的图片
"""
@@ -1573,7 +1575,7 @@ class TmdbApi:
logger.error(str(e))
return {}
def get_movie_similar(self, tmdbid: int) -> List[dict]:
def get_movie_similar(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电影的相似电影
"""
@@ -1586,7 +1588,7 @@ class TmdbApi:
logger.error(str(e))
return []
def get_tv_similar(self, tmdbid: int) -> List[dict]:
def get_tv_similar(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电视剧的相似电视剧
"""
@@ -1599,7 +1601,7 @@ class TmdbApi:
logger.error(str(e))
return []
def get_movie_recommend(self, tmdbid: int) -> List[dict]:
def get_movie_recommend(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电影的推荐电影
"""
@@ -1612,7 +1614,7 @@ class TmdbApi:
logger.error(str(e))
return []
def get_tv_recommend(self, tmdbid: int) -> List[dict]:
def get_tv_recommend(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电视剧的推荐电视剧
"""
@@ -1625,7 +1627,7 @@ class TmdbApi:
logger.error(str(e))
return []
def get_movie_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
def get_movie_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取电影的演职员列表
"""
@@ -1636,13 +1638,13 @@ class TmdbApi:
info = self.movie.credits(movie_id=tmdbid) or {}
cast = info.get('cast') or []
if cast:
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
def get_tv_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
def get_tv_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取电视剧的演职员列表
"""
@@ -1653,13 +1655,13 @@ class TmdbApi:
info = self.tv.credits(tv_id=tmdbid) or {}
cast = info.get('cast') or []
if cast:
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
def get_tv_group_seasons(self, group_id: str) -> List[dict]:
def get_tv_group_seasons(self, group_id: str) -> List[_MediaRecord]:
"""
获取电视剧剧集组季集列表
"""
@@ -1683,7 +1685,7 @@ class TmdbApi:
logger.error(str(e))
return []
def get_tv_group_detail(self, group_id: str, season: int) -> dict:
def get_tv_group_detail(self, group_id: str, season: int) -> _MediaRecord:
"""
获取剧集组某个季的信息
"""
@@ -1695,7 +1697,7 @@ class TmdbApi:
return group_season
return {}
def get_person_detail(self, person_id: int) -> dict:
def get_person_detail(self, person_id: int) -> _MediaRecord:
"""
获取人物详情
{
@@ -1728,7 +1730,7 @@ class TmdbApi:
logger.error(str(e))
return {}
def get_person_credits(self, person_id: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
def get_person_credits(self, person_id: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取人物参演作品
"""
@@ -1743,18 +1745,18 @@ class TmdbApi:
# 按年份降序排列
cast = sorted(cast, key=lambda x: x.get('release_date') or x.get('first_air_date') or '1900-01-01',
reverse=True)
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
def clear_cache(self):
def clear_cache(self) -> None:
"""
清除缓存
"""
self.discover.discover_movies.cache_clear()
self.discover.discover_tv_shows.cache_clear()
getattr(self.discover.discover_movies, "cache_clear", lambda: None)()
getattr(self.discover.discover_tv_shows, "cache_clear", lambda: None)()
self.tmdb.cache_clear()
# 私有异步方法
@@ -1849,7 +1851,7 @@ class TmdbApi:
"translations,"
"release_dates,"
"external_ids",
raise_on_connection_error: bool = False) -> Optional[dict]:
raise_on_connection_error: bool = False) -> Optional[_MediaRecord]:
"""
获取电影的详情（异步版本）
:param tmdbid: TMDB ID
@@ -1883,7 +1885,7 @@ class TmdbApi:
"content_ratings,"
"external_ids,"
"episode_groups",
raise_on_connection_error: bool = False) -> Optional[dict]:
raise_on_connection_error: bool = False) -> Optional[_MediaRecord]:
"""
获取电视剧的详情（异步版本）
:param tmdbid: TMDB ID
@@ -1936,8 +1938,8 @@ class TmdbApi:
return self._project_search_results(title, tvs, MediaType.TV)
async def async_discover_movies(
self, params: dict, raise_exception: bool = False
) -> List[dict]:
self, params: _MediaRecord, raise_exception: bool = False
) -> List[_MediaRecord]:
"""
发现电影（异步版本）
"""
@@ -1955,8 +1957,8 @@ class TmdbApi:
return []
async def async_discover_tvs(
self, params: dict, raise_exception: bool = False
) -> List[dict]:
self, params: _MediaRecord, raise_exception: bool = False
) -> List[_MediaRecord]:
"""
发现电视剧（异步版本）
"""
@@ -1973,7 +1975,7 @@ class TmdbApi:
raise
return []
async def async_search_persons(self, name: str) -> List[dict]:
async def async_search_persons(self, name: str) -> List[_MediaRecord]:
"""
查询模糊匹配的所有人物TMDB信息（异步版本）
"""
@@ -1990,7 +1992,7 @@ class TmdbApi:
collections = await self.search.async_collections(term=name) or []
return self._project_collections(collections)
async def async_get_collection(self, collection_id: int) -> List[dict]:
async def async_get_collection(self, collection_id: int) -> List[_MediaRecord]:
"""
根据合集ID查询合集详情（异步版本）
"""
@@ -2076,7 +2078,7 @@ class TmdbApi:
)
return self._project_info(tmdb_info)
async def async_get_tv_season_detail(self, tmdbid: int, season: int):
async def async_get_tv_season_detail(self, tmdbid: int, season: int) -> _MediaRecord:
"""
获取电视剧季的详情（异步版本）
:param tmdbid: TMDB ID
@@ -2093,7 +2095,7 @@ class TmdbApi:
logger.error(str(e))
return {}
async def async_get_tv_episode_detail(self, tmdbid: int, season: int, episode: int) -> dict:
async def async_get_tv_episode_detail(self, tmdbid: int, season: int, episode: int) -> _MediaRecord:
"""
获取电视剧集的详情（异步版本）
:param tmdbid: TMDB ID
@@ -2112,7 +2114,7 @@ class TmdbApi:
async def async_discover_trending(
self, page: Optional[int] = 1, raise_exception: bool = False
) -> List[dict]:
) -> List[_MediaRecord]:
"""
流行趋势（异步版本）
"""
@@ -2130,7 +2132,7 @@ class TmdbApi:
async def async_get_movie_images(
self, tmdbid: int, original_language: Optional[str] = None
) -> dict:
) -> _MediaRecord:
"""
获取电影的图片（异步版本）
"""
@@ -2150,7 +2152,7 @@ class TmdbApi:
async def async_get_tv_images(
self, tmdbid: int, original_language: Optional[str] = None
) -> dict:
) -> _MediaRecord:
"""
获取电视剧的图片（异步版本）
"""
@@ -2168,7 +2170,7 @@ class TmdbApi:
logger.error(str(e))
return {}
async def async_get_movie_similar(self, tmdbid: int) -> List[dict]:
async def async_get_movie_similar(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电影的相似电影（异步版本）
"""
@@ -2181,7 +2183,7 @@ class TmdbApi:
logger.error(str(e))
return []
async def async_get_tv_similar(self, tmdbid: int) -> List[dict]:
async def async_get_tv_similar(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电视剧的相似电视剧（异步版本）
"""
@@ -2194,7 +2196,7 @@ class TmdbApi:
logger.error(str(e))
return []
async def async_get_movie_recommend(self, tmdbid: int) -> List[dict]:
async def async_get_movie_recommend(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电影的推荐电影（异步版本）
"""
@@ -2207,7 +2209,7 @@ class TmdbApi:
logger.error(str(e))
return []
async def async_get_tv_recommend(self, tmdbid: int) -> List[dict]:
async def async_get_tv_recommend(self, tmdbid: int) -> List[_MediaRecord]:
"""
获取电视剧的推荐电视剧（异步版本）
"""
@@ -2221,7 +2223,7 @@ class TmdbApi:
return []
async def async_get_movie_credits(self, tmdbid: int,
page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取电影的演职员列表（异步版本）
"""
@@ -2232,13 +2234,13 @@ class TmdbApi:
info = await self.movie.async_credits(movie_id=tmdbid) or {}
cast = info.get('cast') or []
if cast:
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
async def async_get_tv_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
async def async_get_tv_credits(self, tmdbid: int, page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取电视剧的演职员列表（异步版本）
"""
@@ -2249,13 +2251,13 @@ class TmdbApi:
info = await self.tv.async_credits(tv_id=tmdbid) or {}
cast = info.get('cast') or []
if cast:
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
async def async_get_tv_group_seasons(self, group_id: str) -> List[dict]:
async def async_get_tv_group_seasons(self, group_id: str) -> List[_MediaRecord]:
"""
获取电视剧剧集组季集列表（异步版本）
"""
@@ -2279,7 +2281,7 @@ class TmdbApi:
logger.error(str(e))
return []
async def async_get_tv_group_detail(self, group_id: str, season: int) -> dict:
async def async_get_tv_group_detail(self, group_id: str, season: int) -> _MediaRecord:
"""
获取剧集组某个季的信息（异步版本）
"""
@@ -2291,7 +2293,7 @@ class TmdbApi:
return group_season
return {}
async def async_get_person_detail(self, person_id: int) -> dict:
async def async_get_person_detail(self, person_id: int) -> _MediaRecord:
"""
获取人物详情（异步版本）
"""
@@ -2305,7 +2307,7 @@ class TmdbApi:
return {}
async def async_get_person_credits(self, person_id: int,
page: Optional[int] = 1, count: Optional[int] = 24) -> List[dict]:
page: Optional[int] = 1, count: Optional[int] = 24) -> List[_MediaRecord]:
"""
获取人物参演作品（异步版本）
"""
@@ -2320,13 +2322,13 @@ class TmdbApi:
# 按年份降序排列
cast = sorted(cast, key=lambda x: x.get('release_date') or x.get('first_air_date') or '1900-01-01',
reverse=True)
return cast[(page - 1) * count: page * count]
return cast[((1 if page is None else page) - 1) * (24 if count is None else count): (1 if page is None else page) * (24 if count is None else count)]
return []
except Exception as e:
logger.error(str(e))
return []
def close(self):
def close(self) -> None:
"""
关闭连接
"""