GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-27 11:11:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve media image scraping

Browse Source
This commit is contained in:
jxxghp
2026-05-19 10:48:36 +08:00
parent 9e9c398177
commit 29dab5a312
6 changed files with 375 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
app/chain/media.py
View File

@@ -363,6 +363,15 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
if season_image_name := season_image_name_map.get(metadata_type):
hint_ext = Path(filename_hint).suffix if filename_hint else ".jpg"
final_filename = f"{season_image_name}{hint_ext}"
elif item_type == ScrapingTarget.MOVIE and current_fileitem.type == "file":
# 电影文件的图片应与视频文件同级保存，避免把图片路径拼到文件名下面。
target_dir_item = parent_fileitem or self.storagechain.get_parent_item(
current_fileitem
)
if not target_dir_item:
logger.error(f"无法获取文件 {current_fileitem.path} 的父目录项。")
return current_fileitem, None
target_dir_path = Path(target_dir_item.path)
# 如果是 EPISODE 类型的图片（如thumb），通常也是放在文件同级目录，文件名与视频文件一致
elif (
metadata_type in [ScrapingMetadata.THUMB]
@@ -390,6 +399,52 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
target_full_path = target_dir_path / final_filename
return target_dir_item, target_full_path
def _get_target_fileitems_and_paths(
self,
current_fileitem: schemas.FileItem,
item_type: ScrapingTarget,
metadata_type: ScrapingMetadata,
filename_hint: Optional[str] = None,
parent_fileitem: Optional[schemas.FileItem] = None,
) -> List[Tuple[schemas.FileItem, Path]]:
"""
根据刮削上下文生成一个或多个保存目标。
季图片需要同时兼容根目录 seasonxx-poster 和季目录 poster 两种命名。
"""
target_item, target_path = self._get_target_fileitem_and_path(
current_fileitem=current_fileitem,
item_type=item_type,
metadata_type=metadata_type,
filename_hint=filename_hint,
parent_fileitem=parent_fileitem,
)
targets = [(target_item, target_path)] if target_path else []
if (
item_type != ScrapingTarget.SEASON
or not filename_hint
or not filename_hint.lower().startswith("season")
or metadata_type not in {
ScrapingMetadata.POSTER,
ScrapingMetadata.BANNER,
ScrapingMetadata.THUMB,
}
):
return targets
season_parent_item = parent_fileitem or self.storagechain.get_parent_item(
current_fileitem
)
if not season_parent_item:
logger.warn(f"无法获取季目录 {current_fileitem.path} 的父目录项，跳过根目录季图片")
return targets
season_root_path = Path(current_fileitem.path).with_name(filename_hint)
root_target = (season_parent_item, season_root_path)
if root_target not in targets:
targets.insert(0, root_target)
return targets
def metadata_nfo(
self,
meta: MetaBase,
@@ -773,7 +828,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
self.scrape_metadata(
fileitem=fileitem,
mediainfo=mediainfo,
init_folder=False,
init_folder=True,
parent=self.storagechain.get_parent_item(fileitem),
overwrite=overwrite,
)
@@ -985,8 +1040,8 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
)
continue
# 获取目标 FileItem (`base_item`) 和 Path (`image_path`)
base_item, image_path = self._get_target_fileitem_and_path(
# 获取目标 FileItem 和 Path，季图片会同时写根目录和季目录。
image_targets = self._get_target_fileitems_and_paths(
current_fileitem=current_fileitem,
item_type=item_type,
metadata_type=metadata_type,
@@ -994,19 +1049,20 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
parent_fileitem=parent_fileitem,
)
if not image_path:
continue
for base_item, image_path in image_targets:
if not image_path:
continue
# 文件存在检查
file_exists = self.storagechain.get_file_item(
storage=base_item.storage, path=image_path
)
# 刮削决策
if self._should_scrape(option, bool(file_exists), overwrite):
self._download_and_save_image(
fileitem=base_item, path=image_path, url=image_url
# 文件存在检查
file_exists = self.storagechain.get_file_item(
storage=base_item.storage, path=image_path
)
# 刮削决策
if self._should_scrape(option, bool(file_exists), overwrite):
self._download_and_save_image(
fileitem=base_item, path=image_path, url=image_url
)
else:
logger.debug(
f"未找到图片类型 {image_name} 对应的 ScrapingMetadata，跳过。"
@@ -1092,7 +1148,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
处理电影刮削
"""
if fileitem.type == "file":
# 电影文件：仅处理 NFO
# 电影文件始终处理 NFO，直接初始化文件时再补同级目录图片。
self._scrape_nfo_generic(
current_fileitem=fileitem,
meta=meta,
@@ -1101,6 +1157,14 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
parent_fileitem=parent,
overwrite=overwrite,
)
if init_folder:
self._scrape_images_generic(
current_fileitem=fileitem,
mediainfo=mediainfo,
item_type=ScrapingTarget.MOVIE,
parent_fileitem=parent,
overwrite=overwrite,
)
else:
# 电影目录：递归处理文件并初始化目录
self._handle_movie_directory(

57
app/modules/themoviedb/__init__.py
View File

@@ -969,7 +969,24 @@ class TheMovieDbModule(_ModuleBase):
return None
@staticmethod
def _process_tmdb_images(mediainfo: MediaInfo, images: dict) -> MediaInfo:
def _pick_best_tmdb_image(images: list) -> Optional[str]:
"""
从 TMDB 图片候选中选出评分最高的文件路径。
"""
if not images:
return None
images = sorted(
images,
key=lambda x: (
x.get("vote_average") or 0,
x.get("vote_count") or 0,
),
reverse=True,
)
return images[0].get("file_path")
@classmethod
def _process_tmdb_images(cls, mediainfo: MediaInfo, images: dict) -> MediaInfo:
"""
处理 TMDB 图片数据
:param mediainfo: 媒体信息
@@ -980,22 +997,16 @@ class TheMovieDbModule(_ModuleBase):
images = images[0]
# 背景图
if not mediainfo.backdrop_path:
backdrops = images.get("backdrops")
if backdrops:
backdrops = sorted(backdrops, key=lambda x: x.get("vote_average"), reverse=True)
mediainfo.backdrop_path = settings.TMDB_IMAGE_URL(backdrops[0].get("file_path"))
if image_path := cls._pick_best_tmdb_image(images.get("backdrops")):
mediainfo.backdrop_path = settings.TMDB_IMAGE_URL(image_path)
# 标志
if not mediainfo.logo_path:
logos = images.get("logos")
if logos:
logos = sorted(logos, key=lambda x: x.get("vote_average"), reverse=True)
mediainfo.logo_path = settings.TMDB_IMAGE_URL(logos[0].get("file_path"))
if image_path := cls._pick_best_tmdb_image(images.get("logos")):
mediainfo.logo_path = settings.TMDB_IMAGE_URL(image_path)
# 海报
if not mediainfo.poster_path:
posters = images.get("posters")
if posters:
posters = sorted(posters, key=lambda x: x.get("vote_average"), reverse=True)
mediainfo.poster_path = settings.TMDB_IMAGE_URL(posters[0].get("file_path"))
if image_path := cls._pick_best_tmdb_image(images.get("posters")):
mediainfo.poster_path = settings.TMDB_IMAGE_URL(image_path)
return mediainfo
def obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]:
@@ -1011,9 +1022,15 @@ class TheMovieDbModule(_ModuleBase):
# 调用TMDB图片接口
if mediainfo.type == MediaType.MOVIE:
images = self.tmdb.get_movie_images(mediainfo.tmdb_id)
images = self.tmdb.get_movie_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
else:
images = self.tmdb.get_tv_images(mediainfo.tmdb_id)
images = self.tmdb.get_tv_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
if not images:
return mediainfo
@@ -1033,9 +1050,15 @@ class TheMovieDbModule(_ModuleBase):
# 调用TMDB图片接口
if mediainfo.type == MediaType.MOVIE:
images = await self.tmdb.async_get_movie_images(mediainfo.tmdb_id)
images = await self.tmdb.async_get_movie_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
else:
images = await self.tmdb.async_get_tv_images(mediainfo.tmdb_id)
images = await self.tmdb.async_get_tv_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
if not images:
return mediainfo

47
app/modules/themoviedb/scraper.py
View File

@@ -127,6 +127,7 @@ class TmdbScraper:
if poster_name and poster_url:
images[poster_name] = poster_url
else:
self.__ensure_main_images(mediainfo)
# 获取媒体信息中原有图片
for attr_name, attr_value in vars(mediainfo).items():
if (
@@ -155,6 +156,52 @@ class TmdbScraper:
images[image_name] = image_url
return images
def __ensure_main_images(self, mediainfo: MediaInfo) -> None:
"""
主媒体图片缺失时从 TMDB images 接口回填，避免当前语言没有图时只生成 NFO。
"""
if not mediainfo or not mediainfo.tmdb_id:
return
if mediainfo.poster_path and mediainfo.backdrop_path:
return
if mediainfo.type == MediaType.MOVIE:
image_info = self.default_tmdb.get_movie_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
else:
image_info = self.default_tmdb.get_tv_images(
mediainfo.tmdb_id,
original_language=mediainfo.original_language,
)
if not image_info:
return
if not mediainfo.poster_path:
poster_path = self.__pick_best_image_path(image_info.get("posters"))
if poster_path:
mediainfo.poster_path = settings.TMDB_IMAGE_URL(poster_path)
if not mediainfo.backdrop_path:
backdrop_path = self.__pick_best_image_path(image_info.get("backdrops"))
if backdrop_path:
mediainfo.backdrop_path = settings.TMDB_IMAGE_URL(backdrop_path)
@staticmethod
def __pick_best_image_path(images: list) -> Optional[str]:
"""
从 TMDB 图片列表中选择评分和投票数最高的一张。
"""
if not images:
return None
images = sorted(
images,
key=lambda item: (
item.get("vote_average") or 0,
item.get("vote_count") or 0,
),
reverse=True,
)
return images[0].get("file_path")
@staticmethod
def get_season_poster(seasoninfo: dict, season: int) -> Tuple[str, str]:
"""

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

@@ -4,6 +4,7 @@ from typing import Optional, List
import zhconv
from app.core.config import settings
from app.log import logger
from app.schemas.types import MediaType
from app.utils.string import StringUtils
@@ -1232,7 +1233,24 @@ class TmdbApi:
logger.error(str(e))
return []
def get_movie_images(self, tmdbid: int) -> dict:
@staticmethod
def _build_include_image_language(original_language: Optional[str] = None) -> str:
"""
构造图片接口语言回退列表，避免当前语言没有图片时返回空列表。
"""
languages = []
for language in (
getattr(settings, "TMDB_LOCALE", None),
"en",
None,
original_language,
):
language = "null" if language is None else str(language).strip()
if language and language not in languages:
languages.append(language)
return ",".join(languages)
def get_movie_images(self, tmdbid: int, original_language: Optional[str] = None) -> dict:
"""
获取电影的图片
"""
@@ -1240,12 +1258,17 @@ class TmdbApi:
return {}
try:
logger.debug(f"正在获取电影图片：{tmdbid}...")
return self.movie.images(movie_id=tmdbid) or {}
return self.movie.images(
movie_id=tmdbid,
include_image_language=self._build_include_image_language(
original_language
),
) or {}
except Exception as e:
logger.error(str(e))
return {}
def get_tv_images(self, tmdbid: int) -> dict:
def get_tv_images(self, tmdbid: int, original_language: Optional[str] = None) -> dict:
"""
获取电视剧的图片
"""
@@ -1253,7 +1276,12 @@ class TmdbApi:
return {}
try:
logger.debug(f"正在获取电视剧图片：{tmdbid}...")
return self.tv.images(tv_id=tmdbid) or {}
return self.tv.images(
tv_id=tmdbid,
include_image_language=self._build_include_image_language(
original_language
),
) or {}
except Exception as e:
logger.error(str(e))
return {}
@@ -1965,7 +1993,9 @@ class TmdbApi:
logger.error(str(e))
return []
async def async_get_movie_images(self, tmdbid: int) -> dict:
async def async_get_movie_images(
self, tmdbid: int, original_language: Optional[str] = None
) -> dict:
"""
获取电影的图片（异步版本）
"""
@@ -1973,12 +2003,19 @@ class TmdbApi:
return {}
try:
logger.debug(f"正在获取电影图片：{tmdbid}...")
return await self.movie.async_images(movie_id=tmdbid) or {}
return await self.movie.async_images(
movie_id=tmdbid,
include_image_language=self._build_include_image_language(
original_language
),
) or {}
except Exception as e:
logger.error(str(e))
return {}
async def async_get_tv_images(self, tmdbid: int) -> dict:
async def async_get_tv_images(
self, tmdbid: int, original_language: Optional[str] = None
) -> dict:
"""
获取电视剧的图片（异步版本）
"""
@@ -1986,7 +2023,12 @@ class TmdbApi:
return {}
try:
logger.debug(f"正在获取电视剧图片：{tmdbid}...")
return await self.tv.async_images(tv_id=tmdbid) or {}
return await self.tv.async_images(
tv_id=tmdbid,
include_image_language=self._build_include_image_language(
original_language
),
) or {}
except Exception as e:
logger.error(str(e))
return {}