fix tmdb apis

This commit is contained in:
jxxghp 2024-04-27 21:56:27 +08:00
parent 6d69ac42e5
commit 25e5f7a9f6
3 changed files with 94 additions and 79 deletions

View File

@ -4,7 +4,6 @@ from fastapi import APIRouter, Depends
from app import schemas
from app.chain.tmdb import TmdbChain
from app.core.context import MediaInfo
from app.core.security import verify_token
from app.schemas.types import MediaType
@ -17,10 +16,9 @@ def tmdb_seasons(tmdbid: int, _: schemas.TokenPayload = Depends(verify_token)) -
根据TMDBID查询themoviedb所有季信息
"""
seasons_info = TmdbChain().tmdb_seasons(tmdbid=tmdbid)
if not seasons_info:
return []
else:
if seasons_info:
return seasons_info
return []
@router.get("/similar/{tmdbid}/{type_name}", summary="类似电影/电视剧", response_model=List[schemas.MediaInfo])
@ -32,15 +30,14 @@ def tmdb_similar(tmdbid: int,
"""
mediatype = MediaType(type_name)
if mediatype == MediaType.MOVIE:
tmdbinfos = TmdbChain().movie_similar(tmdbid=tmdbid)
medias = TmdbChain().movie_similar(tmdbid=tmdbid)
elif mediatype == MediaType.TV:
tmdbinfos = TmdbChain().tv_similar(tmdbid=tmdbid)
medias = TmdbChain().tv_similar(tmdbid=tmdbid)
else:
return []
if not tmdbinfos:
if medias:
return [media.to_dict() for media in medias]
return []
else:
return [MediaInfo(tmdb_info=tmdbinfo).to_dict() for tmdbinfo in tmdbinfos]
@router.get("/recommend/{tmdbid}/{type_name}", summary="推荐电影/电视剧", response_model=List[schemas.MediaInfo])
@ -52,15 +49,14 @@ def tmdb_recommend(tmdbid: int,
"""
mediatype = MediaType(type_name)
if mediatype == MediaType.MOVIE:
tmdbinfos = TmdbChain().movie_recommend(tmdbid=tmdbid)
medias = TmdbChain().movie_recommend(tmdbid=tmdbid)
elif mediatype == MediaType.TV:
tmdbinfos = TmdbChain().tv_recommend(tmdbid=tmdbid)
medias = TmdbChain().tv_recommend(tmdbid=tmdbid)
else:
return []
if not tmdbinfos:
if medias:
return [media.to_dict() for media in medias]
return []
else:
return [MediaInfo(tmdb_info=tmdbinfo).to_dict() for tmdbinfo in tmdbinfos]
@router.get("/credits/{tmdbid}/{type_name}", summary="演员阵容", response_model=List[schemas.MediaPerson])
@ -73,15 +69,12 @@ def tmdb_credits(tmdbid: int,
"""
mediatype = MediaType(type_name)
if mediatype == MediaType.MOVIE:
tmdbinfos = TmdbChain().movie_credits(tmdbid=tmdbid, page=page)
persons = TmdbChain().movie_credits(tmdbid=tmdbid, page=page)
elif mediatype == MediaType.TV:
tmdbinfos = TmdbChain().tv_credits(tmdbid=tmdbid, page=page)
persons = TmdbChain().tv_credits(tmdbid=tmdbid, page=page)
else:
return []
if not tmdbinfos:
return []
else:
return [schemas.MediaPerson(source='themoviedb', **tmdbinfo) for tmdbinfo in tmdbinfos]
return persons or []
@router.get("/person/{person_id}", summary="人物详情", response_model=schemas.MediaPerson)
@ -90,11 +83,7 @@ def tmdb_person(person_id: int,
"""
根据人物ID查询人物详情
"""
tmdbinfo = TmdbChain().person_detail(person_id=person_id)
if not tmdbinfo:
return schemas.MediaPerson(source='themoviedb')
else:
return schemas.MediaPerson(source='themoviedb', **tmdbinfo)
return TmdbChain().person_detail(person_id=person_id)
@router.get("/person/credits/{person_id}", summary="人物参演作品", response_model=List[schemas.MediaInfo])
@ -104,11 +93,10 @@ def tmdb_person_credits(person_id: int,
"""
根据人物ID查询人物参演作品
"""
tmdbinfo = TmdbChain().person_credits(person_id=person_id, page=page)
if not tmdbinfo:
medias = TmdbChain().person_credits(person_id=person_id, page=page)
if medias:
return [media.to_dict() for media in medias]
return []
else:
return [MediaInfo(tmdb_info=tmdbinfo).to_dict() for tmdbinfo in tmdbinfo]
@router.get("/movies", summary="TMDB电影", response_model=List[schemas.MediaInfo])
@ -127,7 +115,7 @@ def tmdb_movies(sort_by: str = "popularity.desc",
page=page)
if not movies:
return []
return [MediaInfo(tmdb_info=movie).to_dict() for movie in movies]
return [movie.to_dict() for movie in movies]
@router.get("/tvs", summary="TMDB剧集", response_model=List[schemas.MediaInfo])
@ -146,7 +134,7 @@ def tmdb_tvs(sort_by: str = "popularity.desc",
page=page)
if not tvs:
return []
return [MediaInfo(tmdb_info=tv).to_dict() for tv in tvs]
return [tv.to_dict() for tv in tvs]
@router.get("/trending", summary="TMDB流行趋势", response_model=List[schemas.MediaInfo])
@ -158,7 +146,7 @@ def tmdb_trending(page: int = 1,
infos = TmdbChain().tmdb_trending(page=page)
if not infos:
return []
return [MediaInfo(tmdb_info=info).to_dict() for info in infos]
return [info.to_dict() for info in infos]
@router.get("/{tmdbid}/{season}", summary="TMDB季所有集", response_model=List[schemas.TmdbEpisode])
@ -167,8 +155,4 @@ def tmdb_season_episodes(tmdbid: int, season: int,
"""
根据TMDBID查询某季的所有信信息
"""
episodes_info = TmdbChain().tmdb_episodes(tmdbid=tmdbid, season=season)
if not episodes_info:
return []
else:
return episodes_info
return TmdbChain().tmdb_episodes(tmdbid=tmdbid, season=season)

View File

@ -6,6 +6,7 @@ from cachetools import cached, TTLCache
from app import schemas
from app.chain import ChainBase
from app.core.config import settings
from app.core.context import MediaInfo
from app.schemas import MediaType
from app.utils.singleton import Singleton
@ -16,7 +17,7 @@ class TmdbChain(ChainBase, metaclass=Singleton):
"""
def tmdb_discover(self, mtype: MediaType, sort_by: str, with_genres: str,
with_original_language: str, page: int = 1) -> Optional[List[dict]]:
with_original_language: str, page: int = 1) -> Optional[List[MediaInfo]]:
"""
:param mtype: 媒体类型
:param sort_by: 排序方式
@ -32,7 +33,7 @@ class TmdbChain(ChainBase, metaclass=Singleton):
with_original_language=with_original_language,
page=page)
def tmdb_trending(self, page: int = 1) -> Optional[List[dict]]:
def tmdb_trending(self, page: int = 1) -> Optional[List[MediaInfo]]:
"""
TMDB流行趋势
:param page: 第几页
@ -57,35 +58,35 @@ class TmdbChain(ChainBase, metaclass=Singleton):
"""
return self.run_module("tmdb_episodes", tmdbid=tmdbid, season=season)
def movie_similar(self, tmdbid: int) -> List[dict]:
def movie_similar(self, tmdbid: int) -> Optional[List[MediaInfo]]:
"""
根据TMDBID查询类似电影
:param tmdbid: TMDBID
"""
return self.run_module("tmdb_movie_similar", tmdbid=tmdbid)
def tv_similar(self, tmdbid: int) -> List[dict]:
def tv_similar(self, tmdbid: int) -> Optional[List[MediaInfo]]:
"""
根据TMDBID查询类似电视剧
:param tmdbid: TMDBID
"""
return self.run_module("tmdb_tv_similar", tmdbid=tmdbid)
def movie_recommend(self, tmdbid: int) -> List[dict]:
def movie_recommend(self, tmdbid: int) -> Optional[List[MediaInfo]]:
"""
根据TMDBID查询推荐电影
:param tmdbid: TMDBID
"""
return self.run_module("tmdb_movie_recommend", tmdbid=tmdbid)
def tv_recommend(self, tmdbid: int) -> List[dict]:
def tv_recommend(self, tmdbid: int) -> Optional[List[MediaInfo]]:
"""
根据TMDBID查询推荐电视剧
:param tmdbid: TMDBID
"""
return self.run_module("tmdb_tv_recommend", tmdbid=tmdbid)
def movie_credits(self, tmdbid: int, page: int = 1) -> List[dict]:
def movie_credits(self, tmdbid: int, page: int = 1) -> Optional[List[schemas.MediaPerson]]:
"""
根据TMDBID查询电影演职人员
:param tmdbid: TMDBID
@ -93,7 +94,7 @@ class TmdbChain(ChainBase, metaclass=Singleton):
"""
return self.run_module("tmdb_movie_credits", tmdbid=tmdbid, page=page)
def tv_credits(self, tmdbid: int, page: int = 1) -> List[dict]:
def tv_credits(self, tmdbid: int, page: int = 1) -> Optional[List[schemas.MediaPerson]]:
"""
根据TMDBID查询电视剧演职人员
:param tmdbid: TMDBID
@ -101,14 +102,14 @@ class TmdbChain(ChainBase, metaclass=Singleton):
"""
return self.run_module("tmdb_tv_credits", tmdbid=tmdbid, page=page)
def person_detail(self, person_id: int) -> dict:
def person_detail(self, person_id: int) -> Optional[schemas.MediaPerson]:
"""
根据TMDBID查询演职员详情
:param person_id: 人物ID
"""
return self.run_module("tmdb_person_detail", person_id=person_id)
def person_credits(self, person_id: int, page: int = 1) -> List[dict]:
def person_credits(self, person_id: int, page: int = 1) -> Optional[List[MediaInfo]]:
"""
根据人物ID查询人物参演作品
:param person_id: 人物ID
@ -117,7 +118,7 @@ class TmdbChain(ChainBase, metaclass=Singleton):
return self.run_module("tmdb_person_credits", person_id=person_id, page=page)
@cached(cache=TTLCache(maxsize=1, ttl=3600))
def get_random_wallpager(self):
def get_random_wallpager(self) -> Optional[str]:
"""
获取随机壁纸缓存1个小时
"""

View File

@ -317,7 +317,7 @@ class TheMovieDbModule(_ModuleBase):
logger.info(f"{path} 刮削完成")
def tmdb_discover(self, mtype: MediaType, sort_by: str, with_genres: str, with_original_language: str,
page: int = 1) -> Optional[List[dict]]:
page: int = 1) -> Optional[List[MediaInfo]]:
"""
:param mtype: 媒体类型
:param sort_by: 排序方式
@ -327,25 +327,31 @@ class TheMovieDbModule(_ModuleBase):
:return: 媒体信息列表
"""
if mtype == MediaType.MOVIE:
return self.tmdb.discover_movies(sort_by=sort_by,
infos = self.tmdb.discover_movies(sort_by=sort_by,
with_genres=with_genres,
with_original_language=with_original_language,
page=page)
elif mtype == MediaType.TV:
return self.tmdb.discover_tvs(sort_by=sort_by,
infos = self.tmdb.discover_tvs(sort_by=sort_by,
with_genres=with_genres,
with_original_language=with_original_language,
page=page)
else:
return None
return []
if infos:
return [MediaInfo(tmdb_info=info) for info in infos]
return []
def tmdb_trending(self, page: int = 1) -> List[dict]:
def tmdb_trending(self, page: int = 1) -> List[MediaInfo]:
"""
TMDB流行趋势
:param page: 第几页
:return: TMDB信息列表
"""
return self.tmdb.trending.all_week(page=page)
trending = self.tmdb.trending.all_week(page=page)
if trending:
return [MediaInfo(tmdb_info=info) for info in trending]
return []
def tmdb_seasons(self, tmdbid: int) -> List[schemas.TmdbSeason]:
"""
@ -453,64 +459,88 @@ class TheMovieDbModule(_ModuleBase):
return f"https://{settings.TMDB_IMAGE_DOMAIN}/t/p/{image_prefix}{image_path}"
return None
def tmdb_movie_similar(self, tmdbid: int) -> List[dict]:
def tmdb_movie_similar(self, tmdbid: int) -> List[MediaInfo]:
"""
根据TMDBID查询类似电影
:param tmdbid: TMDBID
"""
return self.tmdb.get_movie_similar(tmdbid=tmdbid)
similar = self.tmdb.get_movie_similar(tmdbid=tmdbid)
if similar:
return [MediaInfo(tmdb_info=info) for info in similar]
return []
def tmdb_tv_similar(self, tmdbid: int) -> List[dict]:
def tmdb_tv_similar(self, tmdbid: int) -> List[MediaInfo]:
"""
根据TMDBID查询类似电视剧
:param tmdbid: TMDBID
"""
return self.tmdb.get_tv_similar(tmdbid=tmdbid)
similar = self.tmdb.get_tv_similar(tmdbid=tmdbid)
if similar:
return [MediaInfo(tmdb_info=info) for info in similar]
return []
def tmdb_movie_recommend(self, tmdbid: int) -> List[dict]:
def tmdb_movie_recommend(self, tmdbid: int) -> List[MediaInfo]:
"""
根据TMDBID查询推荐电影
:param tmdbid: TMDBID
"""
return self.tmdb.get_movie_recommend(tmdbid=tmdbid)
recommend = self.tmdb.get_movie_recommend(tmdbid=tmdbid)
if recommend:
return [MediaInfo(tmdb_info=info) for info in recommend]
return []
def tmdb_tv_recommend(self, tmdbid: int) -> List[dict]:
def tmdb_tv_recommend(self, tmdbid: int) -> List[MediaInfo]:
"""
根据TMDBID查询推荐电视剧
:param tmdbid: TMDBID
"""
return self.tmdb.get_tv_recommend(tmdbid=tmdbid)
recommend = self.tmdb.get_tv_recommend(tmdbid=tmdbid)
if recommend:
return [MediaInfo(tmdb_info=info) for info in recommend]
return []
def tmdb_movie_credits(self, tmdbid: int, page: int = 1) -> List[dict]:
def tmdb_movie_credits(self, tmdbid: int, page: int = 1) -> List[schemas.MediaPerson]:
"""
根据TMDBID查询电影演职员表
:param tmdbid: TMDBID
:param page: 页码
"""
return self.tmdb.get_movie_credits(tmdbid=tmdbid, page=page)
credit_infos = self.tmdb.get_movie_credits(tmdbid=tmdbid, page=page)
if credit_infos:
return [schemas.MediaPerson(source="themoviedb", **info) for info in credit_infos]
return []
def tmdb_tv_credits(self, tmdbid: int, page: int = 1) -> List[dict]:
def tmdb_tv_credits(self, tmdbid: int, page: int = 1) -> List[schemas.MediaPerson]:
"""
根据TMDBID查询电视剧演职员表
:param tmdbid: TMDBID
:param page: 页码
"""
return self.tmdb.get_tv_credits(tmdbid=tmdbid, page=page)
credit_infos = self.tmdb.get_tv_credits(tmdbid=tmdbid, page=page)
if credit_infos:
return [schemas.MediaPerson(source="themoviedb", **info) for info in credit_infos]
return []
def tmdb_person_detail(self, person_id: int) -> dict:
def tmdb_person_detail(self, person_id: int) -> schemas.MediaPerson:
"""
根据TMDBID查询人物详情
:param person_id: 人物ID
"""
return self.tmdb.get_person_detail(person_id=person_id)
detail = self.tmdb.get_person_detail(person_id=person_id)
if detail:
return schemas.MediaPerson(source="themoviedb", **detail)
return schemas.MediaPerson
def tmdb_person_credits(self, person_id: int, page: int = 1) -> List[dict]:
def tmdb_person_credits(self, person_id: int, page: int = 1) -> List[MediaInfo]:
"""
根据TMDBID查询人物参演作品
:param person_id: 人物ID
:param page: 页码
"""
return self.tmdb.get_person_credits(person_id=person_id, page=page)
infos = self.tmdb.get_person_credits(person_id=person_id, page=page)
if infos:
return [MediaInfo(tmdb_info=tmdbinfo) for tmdbinfo in infos]
return []
def clear_cache(self):
"""