nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-03-20 03:57:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat：支持刮削详细设定

Browse Source
This commit is contained in:
jxxghp
2025-06-10 16:37:15 +08:00
parent 162ba9307d
commit f0d25a02a6
2 changed files with 242 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

358
app/chain/media.py
View File

@@ -10,9 +10,10 @@ from app.core.context import Context, MediaInfo
from app.core.event import eventmanager, Event
from app.core.meta import MetaBase
from app.core.metainfo import MetaInfo, MetaInfoPath
from app.db.systemconfig_oper import SystemConfigOper
from app.log import logger
from app.schemas import FileItem
from app.schemas.types import EventType, MediaType, ChainEventType
from app.schemas.types import EventType, MediaType, ChainEventType, SystemConfigKey
from app.utils.http import RequestUtils
from app.utils.string import StringUtils
@@ -26,6 +27,50 @@ class MediaChain(ChainBase):
媒体信息处理链，单例运行
"""
def __init__(self):
super().__init__()
self._systemconfig = SystemConfigOper()
def _get_scraping_switchs(self) -> dict:
"""
获取刮削开关配置
"""
switchs = self._systemconfig.get(SystemConfigKey.ScrapingSwitchs) or {}
# 默认配置
default_switchs = {
'movie_nfo': True, # 电影NFO
'movie_poster': True, # 电影海报
'movie_backdrop': True, # 电影背景图
'movie_logo': True, # 电影Logo
'movie_disc': True, # 电影光盘图
'movie_banner': True, # 电影横幅图
'movie_thumb': True, # 电影缩略图
'tv_nfo': True, # 电视剧NFO
'tv_poster': True, # 电视剧海报
'tv_backdrop': True, # 电视剧背景图
'tv_banner': True, # 电视剧横幅图
'tv_logo': True, # 电视剧Logo
'tv_thumb': True, # 电视剧缩略图
'season_poster': True, # 季海报
'season_banner': True, # 季横幅图
'season_thumb': True, # 季缩略图
'episode_nfo': True, # 集NFO
'episode_thumb': True # 集缩略图
}
# 合并用户配置和默认配置
for key, default_value in default_switchs.items():
if key not in switchs:
switchs[key] = default_value
return switchs
def set_scraping_switchs(self, switchs: dict) -> bool:
"""
设置刮削开关配置
:param switchs: 开关配置字典
:return: 是否设置成功
"""
return self._systemconfig.set(SystemConfigKey.ScrapingSwitchs, switchs)
def metadata_nfo(self, meta: MetaBase, mediainfo: MediaInfo,
season: Optional[int] = None, episode: Optional[int] = None) -> Optional[str]:
"""
@@ -404,37 +449,47 @@ class MediaChain(ChainBase):
if not mediainfo:
logger.warn(f"{filepath} 无法识别文件媒体信息！")
return
# 获取刮削开关配置
scraping_switchs = self._get_scraping_switchs()
logger.info(f"开始刮削：{filepath} ...")
if mediainfo.type == MediaType.MOVIE:
# 电影
if fileitem.type == "file":
# 是否已存在
nfo_path = filepath.with_suffix(".nfo")
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 电影文件
movie_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo)
if movie_nfo:
# 保存或上传nfo文件到上级目录
__save_file(_fileitem=parent, _path=nfo_path, _content=movie_nfo)
else:
logger.warn(f"{filepath.name} nfo文件生成失败！")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
else:
# 电影目录
if is_bluray_folder(fileitem):
# 原盘目录
nfo_path = filepath / (filepath.name + ".nfo")
# 检查电影NFO开关
if scraping_switchs.get('movie_nfo', True):
# 是否已存在
nfo_path = filepath.with_suffix(".nfo")
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 生成原盘nfo
# 电影文件
movie_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo)
if movie_nfo:
# 保存或上传nfo文件到当前目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=movie_nfo)
# 保存或上传nfo文件到上级目录
__save_file(_fileitem=parent, _path=nfo_path, _content=movie_nfo)
else:
logger.warn(f"{filepath.name} nfo文件生成失败！")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
else:
logger.info("电影NFO刮削已关闭，跳过")
else:
# 电影目录
if is_bluray_folder(fileitem):
# 原盘目录
if scraping_switchs.get('movie_nfo', True):
nfo_path = filepath / (filepath.name + ".nfo")
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 生成原盘nfo
movie_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo)
if movie_nfo:
# 保存或上传nfo文件到当前目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=movie_nfo)
else:
logger.warn(f"{filepath.name} nfo文件生成失败！")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
else:
logger.info("电影NFO刮削已关闭，跳过")
else:
# 处理目录内的文件
files = __list_files(_fileitem=fileitem)
@@ -449,16 +504,35 @@ class MediaChain(ChainBase):
image_dict = self.metadata_img(mediainfo=mediainfo)
if image_dict:
for image_name, image_url in image_dict.items():
image_path = filepath.with_name(image_name)
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 写入图片到当前目录
if content:
__save_file(_fileitem=fileitem, _path=image_path, _content=content)
# 根据图片类型检查开关
if 'poster' in image_name.lower():
should_scrape = scraping_switchs.get('movie_poster', True)
elif 'backdrop' in image_name.lower() or 'fanart' in image_name.lower():
should_scrape = scraping_switchs.get('movie_backdrop', True)
elif 'logo' in image_name.lower():
should_scrape = scraping_switchs.get('movie_logo', True)
elif 'disc' in image_name.lower() or 'cdart' in image_name.lower():
should_scrape = scraping_switchs.get('movie_disc', True)
elif 'banner' in image_name.lower():
should_scrape = scraping_switchs.get('movie_banner', True)
elif 'thumb' in image_name.lower():
should_scrape = scraping_switchs.get('movie_thumb', True)
else:
should_scrape = True # 未知类型默认刮削
if should_scrape:
image_path = filepath.with_name(image_name)
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 写入图片到当前目录
if content:
__save_file(_fileitem=fileitem, _path=image_path, _content=content)
else:
logger.info(f"已存在图片文件：{image_path}")
else:
logger.info(f"电影图片刮削已关闭，跳过：{image_name}")
else:
# 电视剧
if fileitem.type == "file":
@@ -472,38 +546,45 @@ class MediaChain(ChainBase):
if not file_mediainfo:
logger.warn(f"{filepath.name} 无法识别文件媒体信息！")
return
# 是否已存在
nfo_path = filepath.with_suffix(".nfo")
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 获取集的nfo文件
episode_nfo = self.metadata_nfo(meta=file_meta, mediainfo=file_mediainfo,
season=file_meta.begin_season,
episode=file_meta.begin_episode)
if episode_nfo:
# 保存或上传nfo文件到上级目录
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=nfo_path, _content=episode_nfo)
else:
logger.warn(f"{filepath.name} nfo文件生成失败！")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
# 获取集的图片
image_dict = self.metadata_img(mediainfo=file_mediainfo,
season=file_meta.begin_season, episode=file_meta.begin_episode)
if image_dict:
for episode, image_url in image_dict.items():
image_path = filepath.with_suffix(Path(image_url).suffix)
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
if content:
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=image_path, _content=content)
# 检查集NFO开关
if scraping_switchs.get('episode_nfo', True):
# 是否已存在
nfo_path = filepath.with_suffix(".nfo")
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 获取集的nfo文件
episode_nfo = self.metadata_nfo(meta=file_meta, mediainfo=file_mediainfo,
season=file_meta.begin_season,
episode=file_meta.begin_episode)
if episode_nfo:
# 保存或上传nfo文件到上级目录
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=nfo_path, _content=episode_nfo)
else:
logger.info(f"已存在图片文件：{image_path}")
logger.warn(f"{filepath.name} nfo文件生成失败！")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
else:
logger.info("集NFO刮削已关闭，跳过")
# 获取集的图片
if scraping_switchs.get('episode_thumb', True):
image_dict = self.metadata_img(mediainfo=file_mediainfo,
season=file_meta.begin_season, episode=file_meta.begin_episode)
if image_dict:
for episode, image_url in image_dict.items():
image_path = filepath.with_suffix(Path(image_url).suffix)
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
if content:
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=image_path, _content=content)
else:
logger.info(f"已存在图片文件：{image_path}")
else:
logger.info("集缩略图刮削已关闭，跳过")
else:
# 当前为目录，处理目录内的文件
files = __list_files(_fileitem=fileitem)
@@ -521,71 +602,95 @@ class MediaChain(ChainBase):
if filepath.name in settings.RENAME_FORMAT_S0_NAMES:
season_meta.begin_season = 0
if season_meta.begin_season is not None:
# 是否已存在
nfo_path = filepath / "season.nfo"
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 当前目录有季号，生成季nfo
season_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo,
season=season_meta.begin_season)
if season_nfo:
# 写入nfo到根目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=season_nfo)
else:
logger.warn(f"无法生成电视剧季nfo文件：{meta.name}")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
# TMDB季poster图片
image_dict = self.metadata_img(mediainfo=mediainfo, season=season_meta.begin_season)
if image_dict:
for image_name, image_url in image_dict.items():
image_path = filepath.with_name(image_name)
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到剧集目录
if content:
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=image_path, _content=content)
# 检查季NFO开关（暂时使用tv_nfo开关）
if scraping_switchs.get('tv_nfo', True):
# 是否已存在
nfo_path = filepath / "season.nfo"
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 当前目录有季号，生成季nfo
season_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo,
season=season_meta.begin_season)
if season_nfo:
# 写入nfo到根目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=season_nfo)
else:
logger.info(f"已存在图片文件：{image_path}")
# 额外fanart季图片：poster thumb banner
image_dict = self.metadata_img(mediainfo=mediainfo)
if image_dict:
for image_name, image_url in image_dict.items():
if image_name.startswith("season"):
logger.warn(f"无法生成电视剧季nfo文件：{meta.name}")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
else:
logger.info("季NFO刮削已关闭，跳过")
# TMDB季poster图片
if scraping_switchs.get('season_poster', True):
image_dict = self.metadata_img(mediainfo=mediainfo, season=season_meta.begin_season)
if image_dict:
for image_name, image_url in image_dict.items():
image_path = filepath.with_name(image_name)
# 只下载当前刮削季的图片
image_season = "00" if "specials" in image_name else image_name[6:8]
if image_season != str(season_meta.begin_season).rjust(2, '0'):
logger.info(f"当前刮削季为：{season_meta.begin_season}，跳过文件：{image_path}")
continue
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
# 保存图片文件到剧集目录
if content:
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=image_path, _content=content)
else:
logger.info(f"已存在图片文件：{image_path}")
else:
logger.info("季海报刮削已关闭，跳过")
# 额外fanart季图片：poster thumb banner
image_dict = self.metadata_img(mediainfo=mediainfo)
if image_dict:
for image_name, image_url in image_dict.items():
if image_name.startswith("season"):
# 根据季图片类型检查开关
if 'poster' in image_name.lower():
should_scrape = scraping_switchs.get('season_poster', True)
elif 'banner' in image_name.lower():
should_scrape = scraping_switchs.get('season_banner', True)
elif 'thumb' in image_name.lower():
should_scrape = scraping_switchs.get('season_thumb', True)
else:
should_scrape = True # 未知类型默认刮削
if should_scrape:
image_path = filepath.with_name(image_name)
# 只下载当前刮削季的图片
image_season = "00" if "specials" in image_name else image_name[6:8]
if image_season != str(season_meta.begin_season).rjust(2, '0'):
logger.info(f"当前刮削季为：{season_meta.begin_season}，跳过文件：{image_path}")
continue
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
if content:
if not parent:
parent = storagechain.get_parent_item(fileitem)
__save_file(_fileitem=parent, _path=image_path, _content=content)
else:
logger.info(f"已存在图片文件：{image_path}")
else:
logger.info(f"季图片刮削已关闭，跳过：{image_name}")
# 判断当前目录是不是剧集根目录
if not season_meta.season:
# 是否已存在
nfo_path = filepath / "tvshow.nfo"
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 当前目录有名称，生成tvshow nfo 和 tv图片
tv_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo)
if tv_nfo:
# 写入tvshow nfo到根目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=tv_nfo)
# 检查电视剧NFO开关
if scraping_switchs.get('tv_nfo', True):
# 是否已存在
nfo_path = filepath / "tvshow.nfo"
if overwrite or not storagechain.get_file_item(storage=fileitem.storage, path=nfo_path):
# 当前目录有名称，生成tvshow nfo 和 tv图片
tv_nfo = self.metadata_nfo(meta=meta, mediainfo=mediainfo)
if tv_nfo:
# 写入tvshow nfo到根目录
__save_file(_fileitem=fileitem, _path=nfo_path, _content=tv_nfo)
else:
logger.warn(f"无法生成电视剧nfo文件：{meta.name}")
else:
logger.warn(f"无法生成电视剧nfo文件：{meta.name}")
logger.info(f"已存在nfo文件：{nfo_path}")
else:
logger.info(f"已存在nfo文件：{nfo_path}")
logger.info("电视剧NFO刮削已关闭，跳过")
# 生成目录图片
image_dict = self.metadata_img(mediainfo=mediainfo)
if image_dict:
@@ -593,14 +698,31 @@ class MediaChain(ChainBase):
# 不下载季图片
if image_name.startswith("season"):
continue
image_path = filepath / image_name
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
if content:
__save_file(_fileitem=fileitem, _path=image_path, _content=content)
# 根据电视剧图片类型检查开关
if 'poster' in image_name.lower():
should_scrape = scraping_switchs.get('tv_poster', True)
elif 'backdrop' in image_name.lower() or 'fanart' in image_name.lower():
should_scrape = scraping_switchs.get('tv_backdrop', True)
elif 'banner' in image_name.lower():
should_scrape = scraping_switchs.get('tv_banner', True)
elif 'logo' in image_name.lower():
should_scrape = scraping_switchs.get('tv_logo', True)
elif 'thumb' in image_name.lower():
should_scrape = scraping_switchs.get('tv_thumb', True)
else:
logger.info(f"已存在图片文件：{image_path}")
should_scrape = True # 未知类型默认刮削
if should_scrape:
image_path = filepath / image_name
if overwrite or not storagechain.get_file_item(storage=fileitem.storage,
path=image_path):
# 下载图片
content = __download_image(image_url)
# 保存图片文件到当前目录
if content:
__save_file(_fileitem=fileitem, _path=image_path, _content=content)
else:
logger.info(f"已存在图片文件：{image_path}")
else:
logger.info(f"电视剧图片刮削已关闭，跳过：{image_name}")
logger.info(f"{filepath.name} 刮削完成")

2
app/schemas/types.py
View File

@@ -157,6 +157,8 @@ class SystemConfigKey(Enum):
NotificationSendTime = "NotificationSendTime"
# 通知消息格式模板
NotificationTemplates = "NotificationTemplates"
# 刮削开关设置
ScrapingSwitchs = "ScrapingSwitchs"
# 处理进度Key字典