fix douban apis

This commit is contained in:
jxxghp 2024-04-27 22:29:17 +08:00
parent 25e5f7a9f6
commit e56cfd6ad4
4 changed files with 145 additions and 140 deletions

View File

@ -34,21 +34,7 @@ def douban_person(person_id: int,
"""
根据人物ID查询人物详情
"""
personinfo = DoubanChain().person_detail(person_id=person_id)
if not personinfo:
return schemas.MediaPerson(source='douban')
else:
also_known_as = []
infos = personinfo.get("extra", {}).get("info")
if infos:
also_known_as = ["".join(info) for info in infos]
return schemas.MediaPerson(source='douban', **{
"id": personinfo.get("id"),
"name": personinfo.get("title"),
"avatar": personinfo.get("cover_img", {}).get("url"),
"biography": personinfo.get("extra", {}).get("short_info"),
"also_known_as": also_known_as,
})
return DoubanChain().person_detail(person_id=person_id)
@router.get("/person/credits/{person_id}", summary="人物参演作品", response_model=List[schemas.MediaInfo])
@ -58,11 +44,10 @@ def douban_person_credits(person_id: int,
"""
根据人物ID查询人物参演作品
"""
works = DoubanChain().person_credits(person_id=person_id, page=page)
if not works:
return []
else:
return [MediaInfo(douban_info=work.get("subject")).to_dict() for work in works]
medias = DoubanChain().person_credits(person_id=person_id, page=page)
if medias:
return [media.to_dict() for media in medias]
return []
@router.get("/showing", summary="豆瓣正在热映", response_model=List[schemas.MediaInfo])
@ -73,10 +58,9 @@ def movie_showing(page: int = 1,
浏览豆瓣正在热映
"""
movies = DoubanChain().movie_showing(page=page, count=count)
if not movies:
return []
medias = [MediaInfo(douban_info=movie) for movie in movies]
return [media.to_dict() for media in medias]
if movies:
return [media.to_dict() for media in movies]
return []
@router.get("/movies", summary="豆瓣电影", response_model=List[schemas.MediaInfo])
@ -90,13 +74,9 @@ def douban_movies(sort: str = "R",
"""
movies = DoubanChain().douban_discover(mtype=MediaType.MOVIE,
sort=sort, tags=tags, page=page, count=count)
if not movies:
return []
medias = [MediaInfo(douban_info=movie) for movie in movies]
return [media.to_dict() for media in medias
if media.poster_path
and "movie_large.jpg" not in media.poster_path
and "tv_normal.png" not in media.poster_path]
if movies:
return [media.to_dict() for media in movies]
return []
@router.get("/tvs", summary="豆瓣剧集", response_model=List[schemas.MediaInfo])
@ -110,14 +90,9 @@ def douban_tvs(sort: str = "R",
"""
tvs = DoubanChain().douban_discover(mtype=MediaType.TV,
sort=sort, tags=tags, page=page, count=count)
if not tvs:
return []
medias = [MediaInfo(douban_info=tv) for tv in tvs]
return [media.to_dict() for media in medias
if media.poster_path
and "movie_large.jpg" not in media.poster_path
and "tv_normal.jpg" not in media.poster_path
and "tv_large.jpg" not in media.poster_path]
if tvs:
return [media.to_dict() for media in tvs]
return []
@router.get("/movie_top250", summary="豆瓣电影TOP250", response_model=List[schemas.MediaInfo])
@ -127,8 +102,10 @@ def movie_top250(page: int = 1,
"""
浏览豆瓣剧集信息
"""
movies = DoubanChain().movie_top250(page=page, count=count) or []
return [MediaInfo(douban_info=movie).to_dict() for movie in movies]
movies = DoubanChain().movie_top250(page=page, count=count)
if movies:
return [media.to_dict() for media in movies]
return []
@router.get("/tv_weekly_chinese", summary="豆瓣国产剧集周榜", response_model=List[schemas.MediaInfo])
@ -138,8 +115,10 @@ def tv_weekly_chinese(page: int = 1,
"""
中国每周剧集口碑榜
"""
tvs = DoubanChain().tv_weekly_chinese(page=page, count=count) or []
return [MediaInfo(douban_info=tv).to_dict() for tv in tvs]
tvs = DoubanChain().tv_weekly_chinese(page=page, count=count)
if tvs:
return [media.to_dict() for media in tvs]
return []
@router.get("/tv_weekly_global", summary="豆瓣全球剧集周榜", response_model=List[schemas.MediaInfo])
@ -149,8 +128,10 @@ def tv_weekly_global(page: int = 1,
"""
全球每周剧集口碑榜
"""
tvs = DoubanChain().tv_weekly_global(page=page, count=count) or []
return [MediaInfo(douban_info=tv).to_dict() for tv in tvs]
tvs = DoubanChain().tv_weekly_global(page=page, count=count)
if tvs:
return [media.to_dict() for media in tvs]
return []
@router.get("/tv_animation", summary="豆瓣动画剧集", response_model=List[schemas.MediaInfo])
@ -160,8 +141,10 @@ def tv_animation(page: int = 1,
"""
热门动画剧集
"""
tvs = DoubanChain().tv_animation(page=page, count=count) or []
return [MediaInfo(douban_info=tv).to_dict() for tv in tvs]
tvs = DoubanChain().tv_animation(page=page, count=count)
if tvs:
return [media.to_dict() for media in tvs]
return []
@router.get("/movie_hot", summary="豆瓣热门电影", response_model=List[schemas.MediaInfo])
@ -171,8 +154,10 @@ def movie_hot(page: int = 1,
"""
热门电影
"""
movies = DoubanChain().movie_hot(page=page, count=count) or []
return [MediaInfo(douban_info=movie).to_dict() for movie in movies]
movies = DoubanChain().movie_hot(page=page, count=count)
if movies:
return [media.to_dict() for media in movies]
return []
@router.get("/tv_hot", summary="豆瓣热门电视剧", response_model=List[schemas.MediaInfo])
@ -182,8 +167,10 @@ def tv_hot(page: int = 1,
"""
热门电视剧
"""
tvs = DoubanChain().tv_hot(page=page, count=count) or []
return [MediaInfo(douban_info=tv).to_dict() for tv in tvs]
tvs = DoubanChain().tv_hot(page=page, count=count)
if tvs:
return [media.to_dict() for media in tvs]
return []
@router.get("/credits/{doubanid}/{type_name}", summary="豆瓣演员阵容", response_model=List[schemas.MediaPerson])
@ -196,18 +183,10 @@ def douban_credits(doubanid: str,
"""
mediatype = MediaType(type_name)
if mediatype == MediaType.MOVIE:
doubaninfos = DoubanChain().movie_credits(doubanid=doubanid, page=page)
return DoubanChain().movie_credits(doubanid=doubanid)
elif mediatype == MediaType.TV:
doubaninfos = DoubanChain().tv_credits(doubanid=doubanid, page=page)
else:
return []
if not doubaninfos:
return []
else:
# 更新豆瓣演员信息中的ID从URI中提取'douban://douban.com/celebrity/1316132?subject_id=27503705' subject_id
for doubaninfo in doubaninfos:
doubaninfo['id'] = doubaninfo.get('uri', '').split('?subject_id=')[-1]
return [schemas.MediaPerson(source='douban', **doubaninfo) for doubaninfo in doubaninfos]
return DoubanChain().tv_credits(doubanid=doubanid)
return []
@router.get("/recommend/{doubanid}/{type_name}", summary="豆瓣推荐电影/电视剧", response_model=List[schemas.MediaInfo])
@ -219,15 +198,14 @@ def douban_recommend(doubanid: str,
"""
mediatype = MediaType(type_name)
if mediatype == MediaType.MOVIE:
doubaninfos = DoubanChain().movie_recommend(doubanid=doubanid)
medias = DoubanChain().movie_recommend(doubanid=doubanid)
elif mediatype == MediaType.TV:
doubaninfos = DoubanChain().tv_recommend(doubanid=doubanid)
medias = DoubanChain().tv_recommend(doubanid=doubanid)
else:
return []
if not doubaninfos:
return []
else:
return [MediaInfo(douban_info=doubaninfo).to_dict() for doubaninfo in doubaninfos]
if medias:
return [media.to_dict() for media in medias]
return []
@router.get("/{doubanid}", summary="查询豆瓣详情", response_model=schemas.MediaInfo)

View File

@ -1,7 +1,9 @@
from typing import Optional, List
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
@ -11,14 +13,14 @@ class DoubanChain(ChainBase, metaclass=Singleton):
豆瓣处理链单例运行
"""
def person_detail(self, person_id: int) -> dict:
def person_detail(self, person_id: int) -> Optional[schemas.MediaPerson]:
"""
根据人物ID查询豆瓣人物详情
:param person_id: 人物ID
"""
return self.run_module("douban_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) -> List[MediaInfo]:
"""
根据人物ID查询人物参演作品
:param person_id: 人物ID
@ -26,7 +28,7 @@ class DoubanChain(ChainBase, metaclass=Singleton):
"""
return self.run_module("douban_person_credits", person_id=person_id, page=page)
def movie_top250(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def movie_top250(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取豆瓣电影TOP250
:param page: 页码
@ -34,26 +36,26 @@ class DoubanChain(ChainBase, metaclass=Singleton):
"""
return self.run_module("movie_top250", page=page, count=count)
def movie_showing(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def movie_showing(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取正在上映的电影
"""
return self.run_module("movie_showing", page=page, count=count)
def tv_weekly_chinese(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def tv_weekly_chinese(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取本周中国剧集榜
"""
return self.run_module("tv_weekly_chinese", page=page, count=count)
def tv_weekly_global(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def tv_weekly_global(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取本周全球剧集榜
"""
return self.run_module("tv_weekly_global", page=page, count=count)
def douban_discover(self, mtype: MediaType, sort: str, tags: str,
page: int = 0, count: int = 30) -> Optional[List[dict]]:
page: int = 0, count: int = 30) -> Optional[List[MediaInfo]]:
"""
发现豆瓣电影剧集
:param mtype: 媒体类型
@ -66,13 +68,13 @@ class DoubanChain(ChainBase, metaclass=Singleton):
return self.run_module("douban_discover", mtype=mtype, sort=sort, tags=tags,
page=page, count=count)
def tv_animation(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def tv_animation(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取动画剧集
"""
return self.run_module("tv_animation", page=page, count=count)
def movie_hot(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def movie_hot(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取热门电影
"""
@ -80,7 +82,7 @@ class DoubanChain(ChainBase, metaclass=Singleton):
return None
return self.run_module("movie_hot", page=page, count=count)
def tv_hot(self, page: int = 1, count: int = 30) -> Optional[List[dict]]:
def tv_hot(self, page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
获取热门剧集
"""
@ -88,30 +90,28 @@ class DoubanChain(ChainBase, metaclass=Singleton):
return None
return self.run_module("tv_hot", page=page, count=count)
def movie_credits(self, doubanid: str, page: int = 1) -> List[dict]:
def movie_credits(self, doubanid: str) -> Optional[List[schemas.MediaPerson]]:
"""
根据TMDBID查询电影演职人员
:param doubanid: 豆瓣ID
:param page: 页码
"""
return self.run_module("douban_movie_credits", doubanid=doubanid, page=page)
return self.run_module("douban_movie_credits", doubanid=doubanid)
def tv_credits(self, doubanid: str, page: int = 1) -> List[dict]:
def tv_credits(self, doubanid: str) -> Optional[List[schemas.MediaPerson]]:
"""
根据TMDBID查询电视剧演职人员
:param doubanid: 豆瓣ID
:param page: 页码
"""
return self.run_module("douban_tv_credits", doubanid=doubanid, page=page)
return self.run_module("douban_tv_credits", doubanid=doubanid)
def movie_recommend(self, doubanid: str) -> List[dict]:
def movie_recommend(self, doubanid: str) -> List[MediaInfo]:
"""
根据豆瓣ID查询推荐电影
:param doubanid: 豆瓣ID
"""
return self.run_module("douban_movie_recommend", doubanid=doubanid)
def tv_recommend(self, doubanid: str) -> List[dict]:
def tv_recommend(self, doubanid: str) -> List[MediaInfo]:
"""
根据豆瓣ID查询推荐电视剧
:param doubanid: 豆瓣ID

View File

@ -127,6 +127,6 @@ class TmdbChain(ChainBase, metaclass=Singleton):
# 随机一个电影
while True:
info = random.choice(infos)
if info and info.get("backdrop_path"):
return f"https://{settings.TMDB_IMAGE_DOMAIN}/t/p/original{info.get('backdrop_path')}"
if info and info.backdrop_path:
return f"https://{settings.TMDB_IMAGE_DOMAIN}/t/p/original{info.backdrop_path}"
return None

View File

@ -4,6 +4,7 @@ from typing import List, Optional, Tuple, Union
import cn2an
from app import schemas
from app.core.config import settings
from app.core.context import MediaInfo
from app.core.meta import MetaBase
@ -450,7 +451,7 @@ class DoubanModule(_ModuleBase):
return __douban_movie() or __douban_tv()
def douban_discover(self, mtype: MediaType, sort: str, tags: str,
page: int = 1, count: int = 30) -> Optional[List[dict]]:
page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
"""
发现豆瓣电影剧集
:param mtype: 媒体类型
@ -467,69 +468,75 @@ class DoubanModule(_ModuleBase):
else:
infos = self.doubanapi.tv_recommend(start=(page - 1) * count, count=count,
sort=sort, tags=tags)
if not infos:
return []
return infos.get("items") or []
if infos:
medias = [MediaInfo(douban_info=info) for info in infos.get("items")]
return [media for media in medias if media.poster_path
and "movie_large.jpg" not in media.poster_path
and "tv_normal.png" not in media.poster_path
and "movie_large.jpg" not in media.poster_path
and "tv_normal.jpg" not in media.poster_path
and "tv_large.jpg" not in media.poster_path]
return []
def movie_showing(self, page: int = 1, count: int = 30) -> List[dict]:
def movie_showing(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取正在上映的电影
"""
infos = self.doubanapi.movie_showing(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def tv_weekly_chinese(self, page: int = 1, count: int = 30) -> List[dict]:
def tv_weekly_chinese(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣本周口碑国产剧
"""
infos = self.doubanapi.tv_chinese_best_weekly(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def tv_weekly_global(self, page: int = 1, count: int = 30) -> List[dict]:
def tv_weekly_global(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣本周口碑外国剧
"""
infos = self.doubanapi.tv_global_best_weekly(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def tv_animation(self, page: int = 1, count: int = 30) -> List[dict]:
def tv_animation(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣动画剧
"""
infos = self.doubanapi.tv_animation(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def movie_hot(self, page: int = 1, count: int = 30) -> List[dict]:
def movie_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣热门电影
"""
infos = self.doubanapi.movie_hot_gaia(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def tv_hot(self, page: int = 1, count: int = 30) -> List[dict]:
def tv_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣热门剧集
"""
infos = self.doubanapi.tv_hot(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def search_medias(self, meta: MetaBase) -> Optional[List[MediaInfo]]:
"""
@ -632,15 +639,15 @@ class DoubanModule(_ModuleBase):
return item
return {}
def movie_top250(self, page: int = 1, count: int = 30) -> List[dict]:
def movie_top250(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
"""
获取豆瓣电影TOP250
"""
infos = self.doubanapi.movie_top250(start=(page - 1) * count,
count=count)
if not infos:
return []
return infos.get("subject_collection_items")
if infos:
return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
return []
def scrape_metadata(self, path: Path, mediainfo: MediaInfo, transfer_type: str,
metainfo: MetaBase = None, force_nfo: bool = False, force_img: bool = False) -> None:
@ -781,60 +788,79 @@ class DoubanModule(_ModuleBase):
self.cache.clear()
logger.info("豆瓣缓存清除完成")
def douban_movie_credits(self, doubanid: str, page: int = 1, count: int = 20) -> List[dict]:
def douban_movie_credits(self, doubanid: str) -> List[schemas.MediaPerson]:
"""
根据TMDBID查询电影演职员表
:param doubanid: 豆瓣ID
:param page: 页码
:param count: 数量
"""
result = self.doubanapi.movie_celebrities(subject_id=doubanid)
if not result:
return []
ret_list = result.get("actors") or []
if ret_list:
return ret_list[(page - 1) * count: page * count]
else:
return []
# 更新豆瓣演员信息中的ID从URI中提取'douban://douban.com/celebrity/1316132?subject_id=27503705' subject_id
for doubaninfo in ret_list:
doubaninfo['id'] = doubaninfo.get('uri', '').split('?subject_id=')[-1]
return [schemas.MediaPerson(source='douban', **doubaninfo) for doubaninfo in ret_list]
return []
def douban_tv_credits(self, doubanid: str, page: int = 1, count: int = 20) -> List[dict]:
def douban_tv_credits(self, doubanid: str) -> List[schemas.MediaPerson]:
"""
根据TMDBID查询电视剧演职员表
:param doubanid: 豆瓣ID
:param page: 页码
:param count: 数量
"""
result = self.doubanapi.tv_celebrities(subject_id=doubanid)
if not result:
return []
ret_list = result.get("actors") or []
if ret_list:
return ret_list[(page - 1) * count: page * count]
else:
return []
# 更新豆瓣演员信息中的ID从URI中提取'douban://douban.com/celebrity/1316132?subject_id=27503705' subject_id
for doubaninfo in ret_list:
doubaninfo['id'] = doubaninfo.get('uri', '').split('?subject_id=')[-1]
return [schemas.MediaPerson(source='douban', **doubaninfo) for doubaninfo in ret_list]
return []
def douban_movie_recommend(self, doubanid: str) -> List[dict]:
def douban_movie_recommend(self, doubanid: str) -> List[MediaInfo]:
"""
根据豆瓣ID查询推荐电影
:param doubanid: 豆瓣ID
"""
return self.doubanapi.movie_recommendations(subject_id=doubanid)
recommend = self.doubanapi.movie_recommendations(subject_id=doubanid)
if recommend:
return [MediaInfo(douban_info=info) for info in recommend]
return []
def douban_tv_recommend(self, doubanid: str) -> List[dict]:
def douban_tv_recommend(self, doubanid: str) -> List[MediaInfo]:
"""
根据豆瓣ID查询推荐电视剧
:param doubanid: 豆瓣ID
"""
return self.doubanapi.tv_recommendations(subject_id=doubanid)
recommend = self.doubanapi.tv_recommendations(subject_id=doubanid)
if recommend:
return [MediaInfo(douban_info=info) for info in recommend]
return []
def douban_person_detail(self, person_id: int) -> dict:
def douban_person_detail(self, person_id: int) -> schemas.MediaPerson:
"""
获取人物详细信息
:param person_id: 豆瓣人物ID
"""
return self.doubanapi.person_detail(person_id)
detail = self.doubanapi.person_detail(person_id)
if detail:
also_known_as = []
infos = detail.get("extra", {}).get("info")
if infos:
also_known_as = ["".join(info) for info in infos]
return schemas.MediaPerson(source='douban', **{
"id": detail.get("id"),
"name": detail.get("title"),
"avatar": detail.get("cover_img", {}).get("url"),
"biography": detail.get("extra", {}).get("short_info"),
"also_known_as": also_known_as,
})
return schemas.MediaPerson(source='douban')
def douban_person_credits(self, person_id: int, page: int = 1) -> List[dict]:
def douban_person_credits(self, person_id: int, page: int = 1) -> List[MediaInfo]:
"""
根据TMDBID查询人物参演作品
:param person_id: 人物ID
@ -852,5 +878,6 @@ class DoubanModule(_ModuleBase):
if collection_id:
collections = self.doubanapi.person_work(subject_id=collection_id, start=(page - 1) * 20, count=20)
if collections:
return collections.get("works")
works = collections.get("works")
return [MediaInfo(douban_info=work.get("subject")) for work in works]
return []