mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-05 23:47:41 +08:00
feat：媒体查询协程处理
This commit is contained in:
jxxghp
+419
-131
@@ -83,6 +83,279 @@ class TheMovieDbModule(_ModuleBase):
|
||||
def init_setting(self) -> Tuple[str, Union[str, bool]]:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _validate_recognize_params(meta: MetaBase, tmdbid: Optional[int]) -> bool:
|
||||
"""
|
||||
验证识别参数
|
||||
"""
|
||||
if not tmdbid and not meta:
|
||||
return False
|
||||
|
||||
if meta and not tmdbid and settings.RECOGNIZE_SOURCE != "themoviedb":
|
||||
return False
|
||||
|
||||
if meta and not meta.name:
|
||||
logger.warn("识别媒体信息时未提供元数据名称")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _prepare_search_names(meta: MetaBase) -> List[str]:
|
||||
"""
|
||||
准备搜索名称列表
|
||||
"""
|
||||
# 简体名称
|
||||
zh_name = zhconv.convert(meta.cn_name, "zh-hans") if meta.cn_name else None
|
||||
# 使用中英文名分别识别，去重去空，但要保持顺序
|
||||
return list(dict.fromkeys([k for k in [meta.cn_name, zh_name, meta.en_name] if k]))
|
||||
|
||||
def _search_by_name(self, name: str, meta: MetaBase, group_seasons: List[dict]) -> dict:
|
||||
"""
|
||||
根据名称搜索媒体信息
|
||||
"""
|
||||
if meta.begin_season:
|
||||
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
|
||||
else:
|
||||
logger.info(f"正在识别 {name} ...")
|
||||
|
||||
if meta.type == MediaType.UNKNOWN and not meta.year:
|
||||
return self.tmdb.match_multi(name)
|
||||
else:
|
||||
if meta.type == MediaType.TV:
|
||||
# 确定是电视
|
||||
info = self.tmdb.match(name=name,
|
||||
year=meta.year,
|
||||
mtype=meta.type,
|
||||
season_year=meta.year,
|
||||
season_number=meta.begin_season,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份再查一次
|
||||
info = self.tmdb.match(name=name, mtype=meta.type)
|
||||
return info
|
||||
else:
|
||||
# 有年份先按电影查
|
||||
info = self.tmdb.match(name=name, year=meta.year, mtype=MediaType.MOVIE)
|
||||
# 没有再按电视剧查
|
||||
if not info:
|
||||
info = self.tmdb.match(name=name, year=meta.year, mtype=MediaType.TV,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份和类型再查一次
|
||||
info = self.tmdb.match_multi(name=name)
|
||||
return info
|
||||
|
||||
async def _async_search_by_name(self, name: str, meta: MetaBase, group_seasons: List[dict]) -> dict:
|
||||
"""
|
||||
根据名称搜索媒体信息（异步版本）
|
||||
"""
|
||||
if meta.begin_season:
|
||||
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
|
||||
else:
|
||||
logger.info(f"正在识别 {name} ...")
|
||||
|
||||
if meta.type == MediaType.UNKNOWN and not meta.year:
|
||||
return await self.tmdb.async_match_multi(name)
|
||||
else:
|
||||
if meta.type == MediaType.TV:
|
||||
# 确定是电视
|
||||
info = await self.tmdb.async_match(name=name,
|
||||
year=meta.year,
|
||||
mtype=meta.type,
|
||||
season_year=meta.year,
|
||||
season_number=meta.begin_season,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份再查一次
|
||||
info = await self.tmdb.async_match(name=name, mtype=meta.type)
|
||||
return info
|
||||
else:
|
||||
# 有年份先按电影查
|
||||
info = await self.tmdb.async_match(name=name, year=meta.year, mtype=MediaType.MOVIE)
|
||||
# 没有再按电视剧查
|
||||
if not info:
|
||||
info = await self.tmdb.async_match(name=name, year=meta.year, mtype=MediaType.TV,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份和类型再查一次
|
||||
info = await self.tmdb.async_match_multi(name=name)
|
||||
return info
|
||||
|
||||
def _process_episode_groups(self, mediainfo: MediaInfo, episode_group: Optional[str],
|
||||
group_seasons: List[dict]) -> MediaInfo:
|
||||
"""
|
||||
处理剧集组信息
|
||||
"""
|
||||
if mediainfo.type == MediaType.TV and mediainfo.episode_groups:
|
||||
if group_seasons:
|
||||
# 指定剧集组时
|
||||
seasons = {}
|
||||
season_info = []
|
||||
season_years = {}
|
||||
for group_season in group_seasons:
|
||||
# 季
|
||||
season = group_season.get("order")
|
||||
# 集列表
|
||||
episodes = group_season.get("episodes")
|
||||
if not episodes:
|
||||
continue
|
||||
seasons[season] = [ep.get("episode_number") for ep in episodes]
|
||||
season_info.append(group_season)
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
if re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
# 每季集清单
|
||||
if seasons:
|
||||
mediainfo.seasons = seasons
|
||||
mediainfo.number_of_seasons = len(seasons)
|
||||
# 每季集详情
|
||||
if season_info:
|
||||
mediainfo.season_info = season_info
|
||||
# 每季年份
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
# 所有剧集组
|
||||
mediainfo.episode_group = episode_group
|
||||
mediainfo.episode_groups = group_seasons
|
||||
else:
|
||||
# 每季年份
|
||||
season_years = {}
|
||||
for group in mediainfo.episode_groups:
|
||||
if group.get('type') != 6:
|
||||
# 只处理剧集部分
|
||||
continue
|
||||
group_episodes = self.tmdb.get_tv_group_seasons(group.get('id'))
|
||||
if not group_episodes:
|
||||
continue
|
||||
for group_episode in group_episodes:
|
||||
season = group_episode.get('order')
|
||||
episodes = group_episode.get('episodes')
|
||||
if not episodes:
|
||||
continue
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
# 判断是不是日期格式
|
||||
if first_date and re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
return mediainfo
|
||||
|
||||
async def _async_process_episode_groups(self, mediainfo: MediaInfo, episode_group: Optional[str],
|
||||
group_seasons: List[dict]) -> MediaInfo:
|
||||
"""
|
||||
处理剧集组信息（异步版本）
|
||||
"""
|
||||
if mediainfo.type == MediaType.TV and mediainfo.episode_groups:
|
||||
if group_seasons:
|
||||
# 指定剧集组时
|
||||
seasons = {}
|
||||
season_info = []
|
||||
season_years = {}
|
||||
for group_season in group_seasons:
|
||||
# 季
|
||||
season = group_season.get("order")
|
||||
# 集列表
|
||||
episodes = group_season.get("episodes")
|
||||
if not episodes:
|
||||
continue
|
||||
seasons[season] = [ep.get("episode_number") for ep in episodes]
|
||||
season_info.append(group_season)
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
if re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
# 每季集清单
|
||||
if seasons:
|
||||
mediainfo.seasons = seasons
|
||||
mediainfo.number_of_seasons = len(seasons)
|
||||
# 每季集详情
|
||||
if season_info:
|
||||
mediainfo.season_info = season_info
|
||||
# 每季年份
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
# 所有剧集组
|
||||
mediainfo.episode_group = episode_group
|
||||
mediainfo.episode_groups = group_seasons
|
||||
else:
|
||||
# 每季年份
|
||||
season_years = {}
|
||||
for group in mediainfo.episode_groups:
|
||||
if group.get('type') != 6:
|
||||
# 只处理剧集部分
|
||||
continue
|
||||
group_episodes = await self.tmdb.async_get_tv_group_seasons(group.get('id'))
|
||||
if not group_episodes:
|
||||
continue
|
||||
for group_episode in group_episodes:
|
||||
season = group_episode.get('order')
|
||||
episodes = group_episode.get('episodes')
|
||||
if not episodes:
|
||||
continue
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
# 判断是不是日期格式
|
||||
if first_date and re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
return mediainfo
|
||||
|
||||
def _build_media_info_result(self, info: dict, meta: MetaBase, tmdbid: Optional[int],
|
||||
episode_group: Optional[str], group_seasons: List[dict]) -> MediaInfo:
|
||||
"""
|
||||
构建MediaInfo结果
|
||||
"""
|
||||
# 确定二级分类
|
||||
if info.get('media_type') == MediaType.TV:
|
||||
cat = self.category.get_tv_category(info)
|
||||
else:
|
||||
cat = self.category.get_movie_category(info)
|
||||
|
||||
# 赋值TMDB信息并返回
|
||||
mediainfo = MediaInfo(tmdb_info=info)
|
||||
mediainfo.set_category(cat)
|
||||
|
||||
if meta:
|
||||
logger.info(f"{meta.name} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year} "
|
||||
f"{mediainfo.tmdb_id}")
|
||||
else:
|
||||
logger.info(f"{tmdbid} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year}")
|
||||
|
||||
# 处理剧集组信息
|
||||
return self._process_episode_groups(mediainfo, episode_group, group_seasons)
|
||||
|
||||
async def _async_build_media_info_result(self, info: dict, meta: MetaBase, tmdbid: Optional[int],
|
||||
episode_group: Optional[str], group_seasons: List[dict]) -> MediaInfo:
|
||||
"""
|
||||
构建MediaInfo结果（异步版本）
|
||||
"""
|
||||
# 确定二级分类
|
||||
if info.get('media_type') == MediaType.TV:
|
||||
cat = self.category.get_tv_category(info)
|
||||
else:
|
||||
cat = self.category.get_movie_category(info)
|
||||
|
||||
# 赋值TMDB信息并返回
|
||||
mediainfo = MediaInfo(tmdb_info=info)
|
||||
mediainfo.set_category(cat)
|
||||
|
||||
if meta:
|
||||
logger.info(f"{meta.name} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year} "
|
||||
f"{mediainfo.tmdb_id}")
|
||||
else:
|
||||
logger.info(f"{tmdbid} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year}")
|
||||
|
||||
# 处理剧集组信息
|
||||
return await self._async_process_episode_groups(mediainfo, episode_group, group_seasons)
|
||||
|
||||
def recognize_media(self, meta: MetaBase = None,
|
||||
mtype: MediaType = None,
|
||||
tmdbid: Optional[int] = None,
|
||||
@@ -98,19 +371,13 @@ class TheMovieDbModule(_ModuleBase):
|
||||
:param cache: 是否使用缓存
|
||||
:return: 识别的媒体信息，包括剧集信息
|
||||
"""
|
||||
if not tmdbid and not meta:
|
||||
return None
|
||||
|
||||
if meta and not tmdbid \
|
||||
and settings.RECOGNIZE_SOURCE != "themoviedb":
|
||||
# 验证参数
|
||||
if not self._validate_recognize_params(meta, tmdbid):
|
||||
return None
|
||||
|
||||
if not meta:
|
||||
# 未提供元数据时，直接使用tmdbid查询，不使用缓存
|
||||
cache_info = {}
|
||||
elif not meta.name:
|
||||
logger.warn("识别媒体信息时未提供元数据名称")
|
||||
return None
|
||||
else:
|
||||
# 读取缓存
|
||||
if mtype:
|
||||
@@ -132,50 +399,13 @@ class TheMovieDbModule(_ModuleBase):
|
||||
# 直接查询详情
|
||||
info = self.tmdb.get_info(mtype=mtype, tmdbid=tmdbid)
|
||||
if not info and meta:
|
||||
info = {}
|
||||
# 简体名称
|
||||
zh_name = zhconv.convert(meta.cn_name, "zh-hans") if meta.cn_name else None
|
||||
# 使用中英文名分别识别，去重去空，但要保持顺序
|
||||
names = list(dict.fromkeys([k for k in [meta.cn_name, zh_name, meta.en_name] if k]))
|
||||
# 准备搜索名称
|
||||
names = self._prepare_search_names(meta)
|
||||
for name in names:
|
||||
if meta.begin_season:
|
||||
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
|
||||
else:
|
||||
logger.info(f"正在识别 {name} ...")
|
||||
if meta.type == MediaType.UNKNOWN and not meta.year:
|
||||
info = self.tmdb.match_multi(name)
|
||||
else:
|
||||
if meta.type == MediaType.TV:
|
||||
# 确定是电视
|
||||
info = self.tmdb.match(name=name,
|
||||
year=meta.year,
|
||||
mtype=meta.type,
|
||||
season_year=meta.year,
|
||||
season_number=meta.begin_season,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份再查一次
|
||||
info = self.tmdb.match(name=name,
|
||||
mtype=meta.type)
|
||||
else:
|
||||
# 有年份先按电影查
|
||||
info = self.tmdb.match(name=name,
|
||||
year=meta.year,
|
||||
mtype=MediaType.MOVIE)
|
||||
# 没有再按电视剧查
|
||||
if not info:
|
||||
info = self.tmdb.match(name=name,
|
||||
year=meta.year,
|
||||
mtype=MediaType.TV,
|
||||
group_seasons=group_seasons)
|
||||
if not info:
|
||||
# 去掉年份和类型再查一次
|
||||
info = self.tmdb.match_multi(name=name)
|
||||
|
||||
info = self._search_by_name(name, meta, group_seasons)
|
||||
if not info:
|
||||
# 从网站查询
|
||||
info = self.tmdb.match_web(name=name,
|
||||
mtype=meta.type)
|
||||
info = self.tmdb.match_web(name=name, mtype=meta.type)
|
||||
if info:
|
||||
# 查到就退出
|
||||
break
|
||||
@@ -201,78 +431,88 @@ class TheMovieDbModule(_ModuleBase):
|
||||
info = None
|
||||
|
||||
if info:
|
||||
# 确定二级分类
|
||||
if info.get('media_type') == MediaType.TV:
|
||||
cat = self.category.get_tv_category(info)
|
||||
else:
|
||||
cat = self.category.get_movie_category(info)
|
||||
# 赋值TMDB信息并返回
|
||||
mediainfo = MediaInfo(tmdb_info=info)
|
||||
mediainfo.set_category(cat)
|
||||
if meta:
|
||||
logger.info(f"{meta.name} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year} "
|
||||
f"{mediainfo.tmdb_id}")
|
||||
else:
|
||||
logger.info(f"{tmdbid} TMDB识别结果：{mediainfo.type.value} "
|
||||
f"{mediainfo.title_year}")
|
||||
return self._build_media_info_result(info, meta, tmdbid, episode_group, group_seasons)
|
||||
else:
|
||||
logger.info(f"{meta.name if meta else tmdbid} 未匹配到TMDB媒体信息")
|
||||
|
||||
# 使用剧集组的集信息和年份
|
||||
if mediainfo.type == MediaType.TV and mediainfo.episode_groups:
|
||||
if group_seasons:
|
||||
# 指定剧集组时
|
||||
seasons = {}
|
||||
season_info = []
|
||||
season_years = {}
|
||||
for group_season in group_seasons:
|
||||
# 季
|
||||
season = group_season.get("order")
|
||||
# 集列表
|
||||
episodes = group_season.get("episodes")
|
||||
if not episodes:
|
||||
continue
|
||||
seasons[season] = [ep.get("episode_number") for ep in episodes]
|
||||
season_info.append(group_season)
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
if re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
# 每季集清单
|
||||
if seasons:
|
||||
mediainfo.seasons = seasons
|
||||
mediainfo.number_of_seasons = len(seasons)
|
||||
# 每季集详情
|
||||
if season_info:
|
||||
mediainfo.season_info = season_info
|
||||
# 每季年份
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
# 所有剧集组
|
||||
mediainfo.episode_group = episode_group
|
||||
mediainfo.episode_groups = group_seasons
|
||||
else:
|
||||
# 每季年份
|
||||
season_years = {}
|
||||
for group in mediainfo.episode_groups:
|
||||
if group.get('type') != 6:
|
||||
# 只处理剧集部分
|
||||
continue
|
||||
group_episodes = self.tmdb.get_tv_group_seasons(group.get('id'))
|
||||
if not group_episodes:
|
||||
continue
|
||||
for group_episode in group_episodes:
|
||||
season = group_episode.get('order')
|
||||
episodes = group_episode.get('episodes')
|
||||
if not episodes:
|
||||
continue
|
||||
# 当前季第一季时间
|
||||
first_date = episodes[0].get("air_date")
|
||||
# 判断是不是日期格式
|
||||
if first_date and re.match(r"^\d{4}-\d{2}-\d{2}$", first_date):
|
||||
season_years[season] = str(first_date).split("-")[0]
|
||||
if season_years:
|
||||
mediainfo.season_years = season_years
|
||||
return mediainfo
|
||||
return None
|
||||
|
||||
async def async_recognize_media(self, meta: MetaBase = None,
|
||||
mtype: MediaType = None,
|
||||
tmdbid: Optional[int] = None,
|
||||
episode_group: Optional[str] = None,
|
||||
cache: Optional[bool] = True,
|
||||
**kwargs) -> Optional[MediaInfo]:
|
||||
"""
|
||||
识别媒体信息（异步版本）
|
||||
:param meta: 识别的元数据
|
||||
:param mtype: 识别的媒体类型，与tmdbid配套
|
||||
:param tmdbid: tmdbid
|
||||
:param episode_group: 剧集组
|
||||
:param cache: 是否使用缓存
|
||||
:return: 识别的媒体信息，包括剧集信息
|
||||
"""
|
||||
# 验证参数
|
||||
if not self._validate_recognize_params(meta, tmdbid):
|
||||
return None
|
||||
|
||||
if not meta:
|
||||
# 未提供元数据时，直接使用tmdbid查询，不使用缓存
|
||||
cache_info = {}
|
||||
else:
|
||||
# 读取缓存
|
||||
if mtype:
|
||||
meta.type = mtype
|
||||
if tmdbid:
|
||||
meta.tmdbid = tmdbid
|
||||
cache_info = self.cache.get(meta)
|
||||
|
||||
# 查询剧集组
|
||||
group_seasons = []
|
||||
if episode_group:
|
||||
group_seasons = await self.tmdb.async_get_tv_group_seasons(episode_group)
|
||||
|
||||
# 识别匹配
|
||||
if not cache_info or not cache:
|
||||
info = None
|
||||
# 缓存没有或者强制不使用缓存
|
||||
if tmdbid:
|
||||
# 直接查询详情
|
||||
info = await self.tmdb.async_get_info(mtype=mtype, tmdbid=tmdbid)
|
||||
if not info and meta:
|
||||
# 准备搜索名称
|
||||
names = self._prepare_search_names(meta)
|
||||
for name in names:
|
||||
info = await self._async_search_by_name(name, meta, group_seasons)
|
||||
if not info:
|
||||
# 从网站查询
|
||||
info = await self.tmdb.async_match_web(name=name, mtype=meta.type)
|
||||
if info:
|
||||
# 查到就退出
|
||||
break
|
||||
# 补充全量信息
|
||||
if info and not info.get("genres"):
|
||||
info = await self.tmdb.async_get_info(mtype=info.get("media_type"),
|
||||
tmdbid=info.get("id"))
|
||||
elif not info:
|
||||
logger.error("识别媒体信息时未提供元数据或唯一且有效的tmdbid")
|
||||
return None
|
||||
|
||||
# 保存到缓存
|
||||
if meta:
|
||||
self.cache.update(meta, info)
|
||||
else:
|
||||
# 使用缓存信息
|
||||
if cache_info.get("title"):
|
||||
logger.info(f"{meta.name} 使用TMDB识别缓存：{cache_info.get('title')}")
|
||||
info = await self.tmdb.async_get_info(mtype=cache_info.get("type"),
|
||||
tmdbid=cache_info.get("id"))
|
||||
else:
|
||||
logger.info(f"{meta.name} 使用TMDB识别缓存：无法识别")
|
||||
info = None
|
||||
|
||||
if info:
|
||||
return await self._async_build_media_info_result(info, meta, tmdbid, episode_group, group_seasons)
|
||||
else:
|
||||
logger.info(f"{meta.name if meta else tmdbid} 未匹配到TMDB媒体信息")
|
||||
|
||||
@@ -528,11 +768,12 @@ class TheMovieDbModule(_ModuleBase):
|
||||
"""
|
||||
self.cache.save()
|
||||
|
||||
def obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]:
|
||||
@staticmethod
|
||||
def _validate_obtain_images_params(mediainfo: MediaInfo) -> Optional[MediaInfo]:
|
||||
"""
|
||||
补充抓取媒体信息图片
|
||||
:param mediainfo: 识别的媒体信息
|
||||
:return: 更新后的媒体信息
|
||||
验证 obtain_images 参数
|
||||
:param mediainfo: 媒体信息
|
||||
:return: None 表示不处理，MediaInfo 表示继续处理
|
||||
"""
|
||||
if settings.RECOGNIZE_SOURCE != "themoviedb":
|
||||
return None
|
||||
@@ -543,13 +784,16 @@ class TheMovieDbModule(_ModuleBase):
|
||||
and mediainfo.backdrop_path:
|
||||
# 没有图片缺失
|
||||
return mediainfo
|
||||
# 调用TMDB图片接口
|
||||
if mediainfo.type == MediaType.MOVIE:
|
||||
images = self.tmdb.get_movie_images(mediainfo.tmdb_id)
|
||||
else:
|
||||
images = self.tmdb.get_tv_images(mediainfo.tmdb_id)
|
||||
if not images:
|
||||
return mediainfo
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _process_tmdb_images(mediainfo: MediaInfo, images: dict) -> MediaInfo:
|
||||
"""
|
||||
处理 TMDB 图片数据
|
||||
:param mediainfo: 媒体信息
|
||||
:param images: 图片数据
|
||||
:return: 更新后的媒体信息
|
||||
"""
|
||||
if isinstance(images, list):
|
||||
images = images[0]
|
||||
# 背景图
|
||||
@@ -572,6 +816,50 @@ class TheMovieDbModule(_ModuleBase):
|
||||
mediainfo.poster_path = posters[0].get("file_path")
|
||||
return mediainfo
|
||||
|
||||
def obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]:
|
||||
"""
|
||||
补充抓取媒体信息图片
|
||||
:param mediainfo: 识别的媒体信息
|
||||
:return: 更新后的媒体信息
|
||||
"""
|
||||
# 验证参数
|
||||
result = self._validate_obtain_images_params(mediainfo)
|
||||
if result is not None:
|
||||
return result
|
||||
|
||||
# 调用TMDB图片接口
|
||||
if mediainfo.type == MediaType.MOVIE:
|
||||
images = self.tmdb.get_movie_images(mediainfo.tmdb_id)
|
||||
else:
|
||||
images = self.tmdb.get_tv_images(mediainfo.tmdb_id)
|
||||
if not images:
|
||||
return mediainfo
|
||||
|
||||
# 处理图片数据
|
||||
return self._process_tmdb_images(mediainfo, images)
|
||||
|
||||
async def async_obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]:
|
||||
"""
|
||||
补充抓取媒体信息图片（异步版本）
|
||||
:param mediainfo: 识别的媒体信息
|
||||
:return: 更新后的媒体信息
|
||||
"""
|
||||
# 验证参数
|
||||
result = self._validate_obtain_images_params(mediainfo)
|
||||
if result is not None:
|
||||
return result
|
||||
|
||||
# 调用TMDB图片接口
|
||||
if mediainfo.type == MediaType.MOVIE:
|
||||
images = await self.tmdb.async_get_movie_images(mediainfo.tmdb_id)
|
||||
else:
|
||||
images = await self.tmdb.async_get_tv_images(mediainfo.tmdb_id)
|
||||
if not images:
|
||||
return mediainfo
|
||||
|
||||
# 处理图片数据
|
||||
return self._process_tmdb_images(mediainfo, images)
|
||||
|
||||
def obtain_specific_image(self, mediaid: Union[str, int], mtype: MediaType,
|
||||
image_type: MediaImageType, image_prefix: Optional[str] = "w500",
|
||||
season: Optional[int] = None, episode: Optional[int] = None) -> Optional[str]:
|
||||
|
||||
+347
-185
@@ -11,7 +11,7 @@ from app.core.config import settings
|
||||
from app.log import logger
|
||||
from app.schemas import APIRateLimitException
|
||||
from app.schemas.types import MediaType
|
||||
from app.utils.http import RequestUtils
|
||||
from app.utils.http import RequestUtils, AsyncRequestUtils
|
||||
from app.utils.limit import rate_limit_exponential
|
||||
from app.utils.string import StringUtils
|
||||
from .tmdbv3api import TMDb, Search, Movie, TV, Season, Episode, Discover, Trending, Person, Collection
|
||||
@@ -138,6 +138,250 @@ class TmdbApi:
|
||||
return True
|
||||
return False
|
||||
|
||||
# 公共方法
|
||||
@staticmethod
|
||||
def _validate_match_params(name: str, search_obj) -> bool:
|
||||
"""
|
||||
验证匹配方法的基本参数
|
||||
"""
|
||||
if not search_obj:
|
||||
return False
|
||||
if not name:
|
||||
return False
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _generate_year_range(year: Optional[str]) -> List[Optional[str]]:
|
||||
"""
|
||||
生成年份范围用于匹配
|
||||
"""
|
||||
year_range = [year]
|
||||
if year:
|
||||
year_range.append(str(int(year) + 1))
|
||||
year_range.append(str(int(year) - 1))
|
||||
return year_range
|
||||
|
||||
@staticmethod
|
||||
def _log_match_debug(mtype: MediaType, name: str, year: Optional[str] = None,
|
||||
season_number: Optional[int] = None, season_year: Optional[str] = None):
|
||||
"""
|
||||
记录匹配调试日志
|
||||
"""
|
||||
if season_number and season_year:
|
||||
logger.debug(f"正在识别{mtype.value}：{name}, 季集={season_number}, 季集年份={season_year} ...")
|
||||
else:
|
||||
logger.debug(f"正在识别{mtype.value}：{name}, 年份={year} ...")
|
||||
|
||||
@staticmethod
|
||||
def _set_media_type(info: dict, mtype: MediaType) -> dict:
|
||||
"""
|
||||
设置媒体类型
|
||||
"""
|
||||
if info:
|
||||
info['media_type'] = mtype
|
||||
return info
|
||||
|
||||
@staticmethod
|
||||
def _sort_multi_results(multis: List[dict]) -> List[dict]:
|
||||
"""
|
||||
按年份降序排列搜索结果，电影在前面
|
||||
"""
|
||||
return sorted(
|
||||
multis,
|
||||
key=lambda x: ("1"
|
||||
if x.get("media_type") == "movie"
|
||||
else "0") + (x.get('release_date')
|
||||
or x.get('first_air_date')
|
||||
or '0000-00-00'),
|
||||
reverse=True
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _convert_media_type(ret_info: dict) -> dict:
|
||||
"""
|
||||
转换媒体类型为MediaType枚举
|
||||
"""
|
||||
if (ret_info
|
||||
and not isinstance(ret_info.get("media_type"), MediaType)):
|
||||
ret_info['media_type'] = MediaType.MOVIE if ret_info.get("media_type") == "movie" else MediaType.TV
|
||||
return ret_info
|
||||
|
||||
def _match_multi_item(self, name: str, multi: dict, get_info_func) -> Optional[dict]:
|
||||
"""
|
||||
匹配单个多媒体搜索结果项
|
||||
:param name: 查询名称
|
||||
:param multi: 搜索结果项
|
||||
:param get_info_func: 获取详细信息的函数（同步或异步）
|
||||
:return: 匹配的结果或None
|
||||
"""
|
||||
if multi.get("media_type") == "movie":
|
||||
if self.__compare_names(name, multi.get('title')) \
|
||||
or self.__compare_names(name, multi.get('original_title')):
|
||||
return multi
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = get_info_func(mtype=MediaType.MOVIE, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
return multi
|
||||
elif multi.get("media_type") == "tv":
|
||||
if self.__compare_names(name, multi.get('name')) \
|
||||
or self.__compare_names(name, multi.get('original_name')):
|
||||
return multi
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = get_info_func(mtype=MediaType.TV, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
return multi
|
||||
return None
|
||||
|
||||
async def _async_match_multi_item(self, name: str, multi: dict) -> Optional[dict]:
|
||||
"""
|
||||
匹配单个多媒体搜索结果项（异步版本）
|
||||
:param name: 查询名称
|
||||
:param multi: 搜索结果项
|
||||
:return: 匹配的结果或None
|
||||
"""
|
||||
if multi.get("media_type") == "movie":
|
||||
if self.__compare_names(name, multi.get('title')) \
|
||||
or self.__compare_names(name, multi.get('original_title')):
|
||||
return multi
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = await self.async_get_info(mtype=MediaType.MOVIE, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
return multi
|
||||
elif multi.get("media_type") == "tv":
|
||||
if self.__compare_names(name, multi.get('name')) \
|
||||
or self.__compare_names(name, multi.get('original_name')):
|
||||
return multi
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = await self.async_get_info(mtype=MediaType.TV, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
return multi
|
||||
return None
|
||||
|
||||
# match_web 公共方法
|
||||
@staticmethod
|
||||
def _validate_web_params(name: str) -> Optional[dict]:
|
||||
"""
|
||||
验证网站搜索参数
|
||||
:return: None表示继续，dict表示直接返回结果
|
||||
"""
|
||||
if not name:
|
||||
return None
|
||||
if StringUtils.is_chinese(name):
|
||||
return {}
|
||||
return None # 继续执行
|
||||
|
||||
@staticmethod
|
||||
def _build_tmdb_search_url(name: str) -> str:
|
||||
"""
|
||||
构建TMDB搜索URL
|
||||
"""
|
||||
return "https://www.themoviedb.org/search?query=%s" % quote(name)
|
||||
|
||||
@staticmethod
|
||||
def _validate_response(res) -> Optional[dict]:
|
||||
"""
|
||||
验证HTTP响应
|
||||
:return: None表示继续，dict表示直接返回结果，Exception表示抛出异常
|
||||
"""
|
||||
if res is None:
|
||||
return None
|
||||
if res.status_code == 429:
|
||||
raise APIRateLimitException("触发TheDbMovie网站限流，获取媒体信息失败")
|
||||
if res.status_code != 200:
|
||||
return {}
|
||||
return None # 继续执行
|
||||
|
||||
@staticmethod
|
||||
def _extract_tmdb_links(html_text: str, mtype: MediaType) -> List[str]:
|
||||
"""
|
||||
从HTML文本中提取TMDB链接
|
||||
"""
|
||||
if not html_text:
|
||||
return []
|
||||
|
||||
html = None
|
||||
try:
|
||||
tmdb_links = []
|
||||
html = etree.HTML(html_text)
|
||||
if mtype == MediaType.TV:
|
||||
links = html.xpath("//a[@data-id and @data-media-type='tv']/@href")
|
||||
else:
|
||||
links = html.xpath("//a[@data-id]/@href")
|
||||
for link in links:
|
||||
if not link or (not link.startswith("/tv") and not link.startswith("/movie")):
|
||||
continue
|
||||
if link not in tmdb_links:
|
||||
tmdb_links.append(link)
|
||||
return tmdb_links
|
||||
except Exception as err:
|
||||
logger.error(f"解析TMDB网站HTML出错：{str(err)}")
|
||||
return []
|
||||
finally:
|
||||
if html is not None:
|
||||
del html
|
||||
|
||||
@staticmethod
|
||||
def _log_web_search_result(name: str, tmdbinfo: dict):
|
||||
"""
|
||||
记录网站搜索结果日志
|
||||
"""
|
||||
if tmdbinfo.get('media_type') == MediaType.MOVIE:
|
||||
logger.info("%s 从WEB识别到 电影：TMDBID=%s, 名称=%s, 上映日期=%s" % (
|
||||
name,
|
||||
tmdbinfo.get('id'),
|
||||
tmdbinfo.get('title'),
|
||||
tmdbinfo.get('release_date')))
|
||||
else:
|
||||
logger.info("%s 从WEB识别到 电视剧：TMDBID=%s, 名称=%s, 首播日期=%s" % (
|
||||
name,
|
||||
tmdbinfo.get('id'),
|
||||
tmdbinfo.get('name'),
|
||||
tmdbinfo.get('first_air_date')))
|
||||
|
||||
def _process_web_search_links(self, name: str, mtype: MediaType,
|
||||
tmdb_links: List[str], get_info_func) -> Optional[dict]:
|
||||
"""
|
||||
处理网站搜索得到的链接
|
||||
"""
|
||||
if len(tmdb_links) == 1:
|
||||
tmdbinfo = get_info_func(
|
||||
mtype=MediaType.TV if tmdb_links[0].startswith("/tv") else MediaType.MOVIE,
|
||||
tmdbid=tmdb_links[0].split("/")[-1])
|
||||
if tmdbinfo:
|
||||
if mtype == MediaType.TV and tmdbinfo.get('media_type') != MediaType.TV:
|
||||
return {}
|
||||
self._log_web_search_result(name, tmdbinfo)
|
||||
return tmdbinfo
|
||||
elif len(tmdb_links) > 1:
|
||||
logger.info("%s TMDB网站返回数据过多：%s" % (name, len(tmdb_links)))
|
||||
else:
|
||||
logger.info("%s TMDB网站未查询到媒体信息！" % name)
|
||||
return {}
|
||||
|
||||
async def _async_process_web_search_links(self, name: str,
|
||||
mtype: MediaType, tmdb_links: List[str]) -> Optional[dict]:
|
||||
"""
|
||||
处理网站搜索得到的链接（异步版本）
|
||||
"""
|
||||
if len(tmdb_links) == 1:
|
||||
tmdbinfo = await self.async_get_info(
|
||||
mtype=MediaType.TV if tmdb_links[0].startswith("/tv") else MediaType.MOVIE,
|
||||
tmdbid=int(tmdb_links[0].split("/")[-1]))
|
||||
if tmdbinfo:
|
||||
if mtype == MediaType.TV and tmdbinfo.get('media_type') != MediaType.TV:
|
||||
return {}
|
||||
self._log_web_search_result(name, tmdbinfo)
|
||||
return tmdbinfo
|
||||
elif len(tmdb_links) > 1:
|
||||
logger.info("%s TMDB网站返回数据过多：%s" % (name, len(tmdb_links)))
|
||||
else:
|
||||
logger.info("%s TMDB网站未查询到媒体信息！" % name)
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def __get_names(tmdb_info: dict) -> List[str]:
|
||||
"""
|
||||
@@ -188,47 +432,36 @@ class TmdbApi:
|
||||
:param group_seasons: 集数组信息
|
||||
:return: TMDB的INFO，同时会将mtype赋值到media_type中
|
||||
"""
|
||||
if not self.search:
|
||||
return None
|
||||
if not name:
|
||||
# 基本参数验证
|
||||
if not self._validate_match_params(name, self.search):
|
||||
return None
|
||||
|
||||
# TMDB搜索
|
||||
info = {}
|
||||
if mtype != MediaType.TV:
|
||||
year_range = [year]
|
||||
if year:
|
||||
year_range.append(str(int(year) + 1))
|
||||
year_range.append(str(int(year) - 1))
|
||||
for year in year_range:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 年份={year} ...")
|
||||
info = self.__search_movie_by_name(name, year)
|
||||
year_range = self._generate_year_range(year)
|
||||
for search_year in year_range:
|
||||
self._log_match_debug(mtype, name, search_year)
|
||||
info = self.__search_movie_by_name(name, search_year)
|
||||
if info:
|
||||
info['media_type'] = MediaType.MOVIE
|
||||
break
|
||||
info = self._set_media_type(info, MediaType.MOVIE)
|
||||
else:
|
||||
# 有当前季和当前季集年份，使用精确匹配
|
||||
if season_year and season_number:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 季集={season_number}, 季集年份={season_year} ...")
|
||||
self._log_match_debug(mtype, name, season_year, season_number, season_year)
|
||||
info = self.__search_tv_by_season(name,
|
||||
season_year,
|
||||
season_number,
|
||||
group_seasons)
|
||||
if not info:
|
||||
year_range = [year]
|
||||
if year:
|
||||
year_range.append(str(int(year) + 1))
|
||||
year_range.append(str(int(year) - 1))
|
||||
for year in year_range:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 年份={year} ...")
|
||||
info = self.__search_tv_by_name(name, year)
|
||||
year_range = self._generate_year_range(year)
|
||||
for search_year in year_range:
|
||||
self._log_match_debug(mtype, name, search_year)
|
||||
info = self.__search_tv_by_name(name, search_year)
|
||||
if info:
|
||||
break
|
||||
if info:
|
||||
info['media_type'] = MediaType.TV
|
||||
# 返回
|
||||
info = self._set_media_type(info, MediaType.TV)
|
||||
return info
|
||||
|
||||
def __search_movie_by_name(self, name: str, year: str) -> Optional[dict]:
|
||||
@@ -454,51 +687,24 @@ class TmdbApi:
|
||||
print(traceback.format_exc())
|
||||
return None
|
||||
logger.debug(f"API返回：{str(self.search.total_results)}")
|
||||
|
||||
# 返回结果
|
||||
ret_info = {}
|
||||
if (multis is None) or (len(multis) == 0):
|
||||
logger.debug(f"{name} 未找到相关媒体息!")
|
||||
return {}
|
||||
else:
|
||||
# 按年份降序排列，电影在前面
|
||||
multis = sorted(
|
||||
multis,
|
||||
key=lambda x: ("1"
|
||||
if x.get("media_type") == "movie"
|
||||
else "0") + (x.get('release_date')
|
||||
or x.get('first_air_date')
|
||||
or '0000-00-00'),
|
||||
reverse=True
|
||||
)
|
||||
for multi in multis:
|
||||
if multi.get("media_type") == "movie":
|
||||
if self.__compare_names(name, multi.get('title')) \
|
||||
or self.__compare_names(name, multi.get('original_title')):
|
||||
ret_info = multi
|
||||
break
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = self.get_info(mtype=MediaType.MOVIE, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
ret_info = multi
|
||||
break
|
||||
elif multi.get("media_type") == "tv":
|
||||
if self.__compare_names(name, multi.get('name')) \
|
||||
or self.__compare_names(name, multi.get('original_name')):
|
||||
ret_info = multi
|
||||
break
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = self.get_info(mtype=MediaType.TV, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
ret_info = multi
|
||||
break
|
||||
# 类型变更
|
||||
if (ret_info
|
||||
and not isinstance(ret_info.get("media_type"), MediaType)):
|
||||
ret_info['media_type'] = MediaType.MOVIE if ret_info.get("media_type") == "movie" else MediaType.TV
|
||||
|
||||
return ret_info
|
||||
# 按年份降序排列，电影在前面
|
||||
multis = self._sort_multi_results(multis)
|
||||
|
||||
ret_info = {}
|
||||
for multi in multis:
|
||||
matched = self._match_multi_item(name, multi, self.get_info)
|
||||
if matched:
|
||||
ret_info = matched
|
||||
break
|
||||
|
||||
# 类型变更
|
||||
return self._convert_media_type(ret_info)
|
||||
|
||||
@cached(maxsize=settings.CONF.tmdb, ttl=settings.CONF.meta)
|
||||
@rate_limit_exponential(source="match_tmdb_web", base_wait=5, max_wait=1800, enable_logging=True)
|
||||
@@ -508,66 +714,28 @@ class TmdbApi:
|
||||
:param name: 名称
|
||||
:param mtype: 媒体类型
|
||||
"""
|
||||
if not name:
|
||||
return None
|
||||
if StringUtils.is_chinese(name):
|
||||
return {}
|
||||
# 参数验证
|
||||
validation_result = self._validate_web_params(name)
|
||||
if validation_result is not None:
|
||||
return validation_result
|
||||
|
||||
logger.info("正在从TheDbMovie网站查询：%s ..." % name)
|
||||
tmdb_url = "https://www.themoviedb.org/search?query=%s" % quote(name)
|
||||
tmdb_url = self._build_tmdb_search_url(name)
|
||||
res = RequestUtils(timeout=5, ua=settings.NORMAL_USER_AGENT, proxies=settings.PROXY).get_res(url=tmdb_url)
|
||||
if res is None:
|
||||
return None
|
||||
if res.status_code == 429:
|
||||
raise APIRateLimitException("触发TheDbMovie网站限流，获取媒体信息失败")
|
||||
if res.status_code != 200:
|
||||
return {}
|
||||
html = None
|
||||
html_text = res.text
|
||||
if not html_text:
|
||||
return {}
|
||||
|
||||
# 响应验证
|
||||
response_result = self._validate_response(res)
|
||||
if response_result is not None:
|
||||
return response_result
|
||||
|
||||
try:
|
||||
tmdb_links = []
|
||||
html = etree.HTML(html_text)
|
||||
if mtype == MediaType.TV:
|
||||
links = html.xpath("//a[@data-id and @data-media-type='tv']/@href")
|
||||
else:
|
||||
links = html.xpath("//a[@data-id]/@href")
|
||||
for link in links:
|
||||
if not link or (not link.startswith("/tv") and not link.startswith("/movie")):
|
||||
continue
|
||||
if link not in tmdb_links:
|
||||
tmdb_links.append(link)
|
||||
if len(tmdb_links) == 1:
|
||||
tmdbinfo = self.get_info(
|
||||
mtype=MediaType.TV if tmdb_links[0].startswith("/tv") else MediaType.MOVIE,
|
||||
tmdbid=tmdb_links[0].split("/")[-1])
|
||||
if tmdbinfo:
|
||||
if mtype == MediaType.TV and tmdbinfo.get('media_type') != MediaType.TV:
|
||||
return {}
|
||||
if tmdbinfo.get('media_type') == MediaType.MOVIE:
|
||||
logger.info("%s 从WEB识别到 电影：TMDBID=%s, 名称=%s, 上映日期=%s" % (
|
||||
name,
|
||||
tmdbinfo.get('id'),
|
||||
tmdbinfo.get('title'),
|
||||
tmdbinfo.get('release_date')))
|
||||
else:
|
||||
logger.info("%s 从WEB识别到 电视剧：TMDBID=%s, 名称=%s, 首播日期=%s" % (
|
||||
name,
|
||||
tmdbinfo.get('id'),
|
||||
tmdbinfo.get('name'),
|
||||
tmdbinfo.get('first_air_date')))
|
||||
return tmdbinfo
|
||||
elif len(tmdb_links) > 1:
|
||||
logger.info("%s TMDB网站返回数据过多：%s" % (name, len(tmdb_links)))
|
||||
else:
|
||||
logger.info("%s TMDB网站未查询到媒体信息！" % name)
|
||||
return {}
|
||||
# 提取链接
|
||||
tmdb_links = self._extract_tmdb_links(res.text, mtype)
|
||||
# 处理结果
|
||||
return self._process_web_search_links(name, mtype, tmdb_links, self.get_info)
|
||||
except Exception as err:
|
||||
logger.error(f"从TheDbMovie网站查询出错：{str(err)}")
|
||||
return {}
|
||||
finally:
|
||||
if html is not None:
|
||||
del html
|
||||
|
||||
def get_info(self,
|
||||
mtype: MediaType,
|
||||
@@ -1648,6 +1816,38 @@ class TmdbApi:
|
||||
return None
|
||||
|
||||
# 公共异步方法
|
||||
@cached(maxsize=settings.CONF.tmdb, ttl=settings.CONF.meta)
|
||||
@rate_limit_exponential(source="match_tmdb_web", base_wait=5, max_wait=1800, enable_logging=True)
|
||||
async def async_match_web(self, name: str, mtype: MediaType) -> Optional[dict]:
|
||||
"""
|
||||
搜索TMDB网站，直接抓取结果，结果只有一条时才返回（异步版本）
|
||||
:param name: 名称
|
||||
:param mtype: 媒体类型
|
||||
"""
|
||||
# 参数验证
|
||||
validation_result = self._validate_web_params(name)
|
||||
if validation_result is not None:
|
||||
return validation_result
|
||||
|
||||
logger.info("正在从TheDbMovie网站查询：%s ..." % name)
|
||||
tmdb_url = self._build_tmdb_search_url(name)
|
||||
res = await AsyncRequestUtils(timeout=5, ua=settings.NORMAL_USER_AGENT, proxies=settings.PROXY).get_res(
|
||||
url=tmdb_url)
|
||||
|
||||
# 响应验证
|
||||
response_result = self._validate_response(res)
|
||||
if response_result is not None:
|
||||
return response_result
|
||||
|
||||
try:
|
||||
# 提取链接
|
||||
tmdb_links = self._extract_tmdb_links(res.text, mtype)
|
||||
# 处理结果
|
||||
return await self._async_process_web_search_links(name, mtype, tmdb_links)
|
||||
except Exception as err:
|
||||
logger.error(f"从TheDbMovie网站查询出错：{str(err)}")
|
||||
return {}
|
||||
|
||||
async def async_search_multiis(self, title: str) -> List[dict]:
|
||||
"""
|
||||
同时查询模糊匹配的电影、电视剧TMDB信息（异步版本）
|
||||
@@ -1776,47 +1976,36 @@ class TmdbApi:
|
||||
:param group_seasons: 集数组信息
|
||||
:return: TMDB的INFO，同时会将mtype赋值到media_type中
|
||||
"""
|
||||
if not self.search:
|
||||
return None
|
||||
if not name:
|
||||
# 基本参数验证
|
||||
if not self._validate_match_params(name, self.search):
|
||||
return None
|
||||
|
||||
# TMDB搜索
|
||||
info = {}
|
||||
if mtype != MediaType.TV:
|
||||
year_range = [year]
|
||||
if year:
|
||||
year_range.append(str(int(year) + 1))
|
||||
year_range.append(str(int(year) - 1))
|
||||
for year in year_range:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 年份={year} ...")
|
||||
info = await self.__async_search_movie_by_name(name, year)
|
||||
year_range = self._generate_year_range(year)
|
||||
for search_year in year_range:
|
||||
self._log_match_debug(mtype, name, search_year)
|
||||
info = await self.__async_search_movie_by_name(name, search_year)
|
||||
if info:
|
||||
info['media_type'] = MediaType.MOVIE
|
||||
break
|
||||
info = self._set_media_type(info, MediaType.MOVIE)
|
||||
else:
|
||||
# 有当前季和当前季集年份，使用精确匹配
|
||||
if season_year and season_number:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 季集={season_number}, 季集年份={season_year} ...")
|
||||
self._log_match_debug(mtype, name, season_year, season_number, season_year)
|
||||
info = await self.__async_search_tv_by_season(name,
|
||||
season_year,
|
||||
season_number,
|
||||
group_seasons)
|
||||
if not info:
|
||||
year_range = [year]
|
||||
if year:
|
||||
year_range.append(str(int(year) + 1))
|
||||
year_range.append(str(int(year) - 1))
|
||||
for year in year_range:
|
||||
logger.debug(
|
||||
f"正在识别{mtype.value}：{name}, 年份={year} ...")
|
||||
info = await self.__async_search_tv_by_name(name, year)
|
||||
year_range = self._generate_year_range(year)
|
||||
for search_year in year_range:
|
||||
self._log_match_debug(mtype, name, search_year)
|
||||
info = await self.__async_search_tv_by_name(name, search_year)
|
||||
if info:
|
||||
break
|
||||
if info:
|
||||
info['media_type'] = MediaType.TV
|
||||
# 返回
|
||||
info = self._set_media_type(info, MediaType.TV)
|
||||
return info
|
||||
|
||||
async def async_match_multi(self, name: str) -> Optional[dict]:
|
||||
@@ -1835,51 +2024,24 @@ class TmdbApi:
|
||||
print(traceback.format_exc())
|
||||
return None
|
||||
logger.debug(f"API返回：{str(self.search.total_results)}")
|
||||
|
||||
# 返回结果
|
||||
ret_info = {}
|
||||
if (multis is None) or (len(multis) == 0):
|
||||
logger.debug(f"{name} 未找到相关媒体息!")
|
||||
return {}
|
||||
else:
|
||||
# 按年份降序排列，电影在前面
|
||||
multis = sorted(
|
||||
multis,
|
||||
key=lambda x: ("1"
|
||||
if x.get("media_type") == "movie"
|
||||
else "0") + (x.get('release_date')
|
||||
or x.get('first_air_date')
|
||||
or '0000-00-00'),
|
||||
reverse=True
|
||||
)
|
||||
for multi in multis:
|
||||
if multi.get("media_type") == "movie":
|
||||
if self.__compare_names(name, multi.get('title')) \
|
||||
or self.__compare_names(name, multi.get('original_title')):
|
||||
ret_info = multi
|
||||
break
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = await self.async_get_info(mtype=MediaType.MOVIE, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
ret_info = multi
|
||||
break
|
||||
elif multi.get("media_type") == "tv":
|
||||
if self.__compare_names(name, multi.get('name')) \
|
||||
or self.__compare_names(name, multi.get('original_name')):
|
||||
ret_info = multi
|
||||
break
|
||||
# 匹配别名、译名
|
||||
if not multi.get("names"):
|
||||
multi = await self.async_get_info(mtype=MediaType.TV, tmdbid=multi.get("id"))
|
||||
if multi and self.__compare_names(name, multi.get("names")):
|
||||
ret_info = multi
|
||||
break
|
||||
# 类型变更
|
||||
if (ret_info
|
||||
and not isinstance(ret_info.get("media_type"), MediaType)):
|
||||
ret_info['media_type'] = MediaType.MOVIE if ret_info.get("media_type") == "movie" else MediaType.TV
|
||||
|
||||
return ret_info
|
||||
# 按年份降序排列，电影在前面
|
||||
multis = self._sort_multi_results(multis)
|
||||
|
||||
ret_info = {}
|
||||
for multi in multis:
|
||||
matched = await self._async_match_multi_item(name, multi)
|
||||
if matched:
|
||||
ret_info = matched
|
||||
break
|
||||
|
||||
# 类型变更
|
||||
return self._convert_media_type(ret_info)
|
||||
|
||||
async def async_get_info(self,
|
||||
mtype: MediaType,
|
||||
|
||||
Reference in New Issue
Block a user