From 653180c8f09b1fa7f19f216192f7da6d829b6cab Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 17 Oct 2026 17:59:13 +0800 Subject: [PATCH] fix(types): type TMDB API response boundaries --- app/modules/themoviedb/tmdbapi.py | 130 +++++++++++++++--------------- 1 file changed, 66 insertions(+), 64 deletions(-) diff --git a/app/modules/themoviedb/tmdbapi.py b/app/modules/themoviedb/tmdbapi.py index 8b73f6603..d71e4136e 100644 --- a/app/modules/themoviedb/tmdbapi.py +++ b/app/modules/themoviedb/tmdbapi.py @@ -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: """ 关闭连接 """