GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-04 23:17:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

TODO 后台整理队列

Browse Source
This commit is contained in:
jxxghp
2024-12-23 21:46:59 +08:00
parent 10f8efc457
commit c4300332c9
4 changed files with 317 additions and 654 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/monitor.py
+15 -343
View File
@@ -1,6 +1,4 @@
import datetime
import platform
import re
import threading
import traceback
from pathlib import Path
@@ -17,19 +15,14 @@ from app.chain.storage import StorageChain
from app.chain.tmdb import TmdbChain
from app.chain.transfer import TransferChain
from app.core.config import settings
from app.core.context import MediaInfo
from app.core.event import EventManager
from app.core.metainfo import MetaInfoPath
from app.db.downloadhistory_oper import DownloadHistoryOper
from app.db.systemconfig_oper import SystemConfigOper
from app.db.transferhistory_oper import TransferHistoryOper
from app.helper.directory import DirectoryHelper
from app.helper.message import MessageHelper
from app.log import logger
from app.schemas import FileItem, TransferInfo, Notification, TransferTask
from app.schemas.types import SystemConfigKey, MediaType, NotificationType, EventType
from app.schemas import FileItem, TransferTask
from app.utils.singleton import Singleton
from app.utils.string import StringUtils
lock = Lock()
snapshot_lock = Lock()
@@ -76,12 +69,6 @@ class Monitor(metaclass=Singleton):
# 存储过照间隔（分钟）
_snapshot_interval = 5
# 消息汇总
_msg_medias = {}
# 消息汇总间隔（秒）
_msg_interval = 60
def __init__(self):
super().__init__()
self.chain = MonitorChain()
@@ -166,9 +153,6 @@ class Monitor(metaclass=Singleton):
'storage': mon_dir.storage,
'mon_path': mon_path
})
# 追加入库消息统一发送服务
self._scheduler.add_job(self.__send_msg, trigger='interval', seconds=15)
# 启动定时服务
if self._scheduler.get_jobs():
self._scheduler.print_jobs()
@@ -231,74 +215,8 @@ class Monitor(metaclass=Singleton):
"""
添加到整理队列
"""
def __callback(_task: TransferTask, _transferinfo: TransferInfo, /):
"""
整理完成后处理
"""
if not _transferinfo:
logger.error("文件转移模块运行失败")
return
if not _transferinfo.success:
# 转移失败
logger.warn(f"{_task.file_path.name} 入库失败：{_transferinfo.message}")
# 新增转移失败历史记录
self.transferhis.add_fail(
fileitem=_task.fileitem,
mode=_transferinfo.transfer_type if _transferinfo else '',
download_hash=_task.download_hash,
meta=_task.meta,
mediainfo=_task.mediainfo,
transferinfo=_transferinfo
)
# 发送失败消息
self.chain.post_message(Notification(
mtype=NotificationType.Manual,
title=f"{_task.mediainfo.title_year} {_task.meta.season_episode} 入库失败！",
text=f"原因：{_transferinfo.message or '未知'}",
image=_task.mediainfo.get_message_image(),
link=settings.MP_DOMAIN('#/history')
))
return
# 转移成功
logger.info(f"{_task.file_path.name} 入库成功：{_transferinfo.target_diritem.path}")
# 新增转移成功历史记录
self.transferhis.add_success(
fileitem=_task.fileitem,
mode=_transferinfo.transfer_type if _transferinfo else '',
download_hash=_task.download_hash,
meta=_task.meta,
mediainfo=_task.mediainfo,
transferinfo=_transferinfo
)
# 汇总刮削
if _transferinfo.need_scrape:
self.mediaChain.scrape_metadata(fileitem=_transferinfo.target_diritem,
meta=_task.meta,
mediainfo=_task.mediainfo)
# 广播事件
EventManager().send_event(EventType.TransferComplete, {
'fileitem': _task.fileitem,
'meta': _task.meta,
'mediainfo': _task.mediainfo,
'transferinfo': _transferinfo
})
# 发送消息汇总
if _transferinfo.need_notify:
self.__collect_msg_medias(mediainfo=_task.mediainfo, file_meta=_task.meta,
transferinfo=_transferinfo)
# 移动模式删除空目录
if _transferinfo.transfer_type in ["move"]:
self.storagechain.delete_media_file(_task.fileitem, delete_self=False)
# 加入整理队列
self.transferchain.put_to_queue(task=task, callback=__callback)
# 加入整理队列，使用默认的回调函数
self.transferchain.put_to_queue(task=task)
def __handle_file(self, storage: str, event_path: Path):
"""
@@ -306,271 +224,25 @@ class Monitor(metaclass=Singleton):
:param storage: 存储
:param event_path: 事件文件路径
"""
def __get_bluray_dir(_path: Path):
"""
获取BDMV目录的上级目录
"""
for p in _path.parents:
if p.name == "BDMV":
return p.parent
return None
# 全程加锁
with lock:
try:
# 回收站及隐藏的文件不处理
if str(event_path).find('/@Recycle/') != -1 \
or str(event_path).find('/#recycle/') != -1 \
or str(event_path).find('/.') != -1 \
or str(event_path).find('/@eaDir') != -1:
logger.debug(f"{event_path} 是回收站或隐藏的文件")
return
# 不是媒体文件不处理
if event_path.suffix.lower() not in self.all_exts:
logger.debug(f"{event_path} 不是媒体文件")
return
# 整理屏蔽词不处理
transfer_exclude_words = self.systemconfig.get(SystemConfigKey.TransferExcludeWords)
if transfer_exclude_words:
for keyword in transfer_exclude_words:
if not keyword:
continue
if keyword and re.search(r"%s" % keyword, str(event_path), re.IGNORECASE):
logger.info(f"{event_path} 命中整理屏蔽词 {keyword}，不处理")
return
# 判断是不是蓝光目录
bluray_flag = False
if re.search(r"BDMV[/\\]STREAM", str(event_path), re.IGNORECASE):
bluray_flag = True
# 截取BDMV前面的路径
event_path = __get_bluray_dir(event_path)
logger.info(f"{event_path} 是蓝光原盘目录，更正文件路径为：{event_path}")
# 查询历史记录，已转移的不处理
if self.transferhis.get_by_src(str(event_path), storage=storage):
logger.info(f"{event_path} 已经整理过了")
return
# 元数据
file_meta = MetaInfoPath(event_path)
if not file_meta.name:
logger.error(f"{event_path.name} 无法识别有效信息")
return
# 根据父路径获取下载历史
download_history = None
if bluray_flag:
# 蓝光原盘，按目录名查询
download_history = self.downloadhis.get_by_path(str(event_path))
else:
# 按文件全路径查询
download_file = self.downloadhis.get_file_by_fullpath(str(event_path))
if download_file:
download_history = self.downloadhis.get_by_hash(download_file.download_hash)
# 获取下载Hash
downloader, download_hash = None, None
if download_history:
downloader = download_history.downloader
download_hash = download_history.download_hash
# 识别媒体信息
if download_history and (download_history.tmdbid or download_history.doubanid):
# 下载记录中已存在识别信息
mediainfo: MediaInfo = self.mediaChain.recognize_media(mtype=MediaType(download_history.type),
tmdbid=download_history.tmdbid,
doubanid=download_history.doubanid)
if mediainfo:
# 更新自定义媒体类别
if download_history.media_category:
mediainfo.category = download_history.media_category
else:
mediainfo: MediaInfo = self.mediaChain.recognize_by_meta(file_meta)
if not mediainfo:
logger.warn(f'未识别到媒体信息，标题：{file_meta.name}')
# 新增转移失败历史记录
his = self.transferhis.add_fail(
fileitem=FileItem(
storage=storage,
type="file",
path=str(event_path),
name=event_path.name,
basename=event_path.stem,
extension=event_path.suffix[1:],
),
mode='',
meta=file_meta,
download_hash=download_hash
)
self.chain.post_message(Notification(
mtype=NotificationType.Manual,
title=f"{event_path.name} 未识别到媒体信息，无法入库！",
text=f"回复：```\n/redo {his.id} [tmdbid]|[类型]\n``` 手动识别转移。",
link=settings.MP_DOMAIN('#/history')
))
return
# 查询转移目的目录
dir_info = self.directoryhelper.get_dir(mediainfo, storage=storage, src_path=event_path)
if not dir_info:
logger.warn(f"{event_path.name} 未找到对应的目标目录")
return
# 查找这个文件项
file_item = self.storagechain.get_file_item(storage=storage, path=event_path)
if not file_item:
logger.warn(f"{event_path.name} 未找到对应的文件")
return
# 如果未开启新增已入库媒体是否跟随TMDB信息变化则根据tmdbid查询之前的title
if not settings.SCRAP_FOLLOW_TMDB:
transfer_history = self.transferhis.get_by_type_tmdbid(tmdbid=mediainfo.tmdb_id,
mtype=mediainfo.type.value)
if transfer_history:
mediainfo.title = transfer_history.title
logger.info(f"{event_path.name} 识别为：{mediainfo.type.value} {mediainfo.title_year}")
# 更新媒体图片
self.chain.obtain_images(mediainfo=mediainfo)
# 获取集数据
if mediainfo.type == MediaType.TV:
episodes_info = self.tmdbchain.tmdb_episodes(tmdbid=mediainfo.tmdb_id,
season=file_meta.begin_season or 1)
else:
episodes_info = None
# 进入队列
self.__transfer_queue(
TransferTask(
fileitem=file_item,
file_path=event_path,
meta=file_meta,
mediainfo=mediainfo,
target_directory=dir_info,
episodes_info=episodes_info,
downloader=downloader,
download_hash=download_hash
)
# 开始整理
# TODO 缺少文件大小
self.transferchain.do_transfer(
fileitem=FileItem(
storage=storage,
path=str(event_path),
type="file",
name=event_path.name,
basename=event_path.stem,
extension=event_path.suffix[1:]
),
src_match=True
)
except Exception as e:
logger.error("目录监控发生错误：%s - %s" % (str(e), traceback.format_exc()))
def __collect_msg_medias(self, mediainfo: MediaInfo, file_meta: MetaInfoPath, transferinfo: TransferInfo):
"""
收集媒体处理完的消息
"""
media_list = self._msg_medias.get(mediainfo.title_year + " " + file_meta.season) or {}
if media_list:
media_files = media_list.get("files") or []
if media_files:
file_exists = False
for file in media_files:
if str(transferinfo.fileitem.path) == file.get("path"):
file_exists = True
break
if not file_exists:
media_files.append({
"path": str(transferinfo.fileitem.path),
"mediainfo": mediainfo,
"file_meta": file_meta,
"transferinfo": transferinfo
})
else:
media_files = [
{
"path": str(transferinfo.fileitem.path),
"mediainfo": mediainfo,
"file_meta": file_meta,
"transferinfo": transferinfo
}
]
media_list = {
"files": media_files,
"time": datetime.datetime.now()
}
else:
media_list = {
"files": [
{
"path": str(transferinfo.fileitem.path),
"mediainfo": mediainfo,
"file_meta": file_meta,
"transferinfo": transferinfo
}
],
"time": datetime.datetime.now()
}
self._msg_medias[mediainfo.title_year + " " + file_meta.season] = media_list
def __send_msg(self):
"""
定时检查是否有媒体处理完，发送统一消息
"""
if not self._msg_medias or not self._msg_medias.keys():
return
# 遍历检查是否已刮削完，发送消息
for medis_title_year_season in list(self._msg_medias.keys()):
media_list = self._msg_medias.get(medis_title_year_season)
logger.info(f"开始处理媒体 {medis_title_year_season} 消息")
if not media_list:
continue
# 获取最后更新时间
last_update_time = media_list.get("time")
media_files = media_list.get("files")
if not last_update_time or not media_files:
continue
transferinfo = media_files[0].get("transferinfo")
file_meta = media_files[0].get("file_meta")
mediainfo = media_files[0].get("mediainfo")
# 判断剧集最后更新时间距现在是已超过10秒或者电影，发送消息
if (datetime.datetime.now() - last_update_time).total_seconds() > int(self._msg_interval) \
or mediainfo.type == MediaType.MOVIE:
# 汇总处理文件总大小
total_size = 0
file_count = 0
# 剧集汇总
episodes = []
for file in media_files:
transferinfo = file.get("transferinfo")
total_size += transferinfo.total_size
file_count += 1
file_meta = file.get("file_meta")
if file_meta and file_meta.begin_episode:
episodes.append(file_meta.begin_episode)
transferinfo.total_size = total_size
# 汇总处理文件数量
transferinfo.file_count = file_count
# 剧集季集信息 S01 E01-E04 || S01 E01、E02、E04
season_episode = None
# 处理文件多，说明是剧集，显示季入库消息
if mediainfo.type == MediaType.TV:
# 季集文本
season_episode = f"{file_meta.season} {StringUtils.format_ep(episodes)}"
# 发送消息
self.transferchain.send_transfer_message(meta=file_meta,
mediainfo=mediainfo,
transferinfo=transferinfo,
season_episode=season_episode)
# 发送完消息，移出key
del self._msg_medias[medis_title_year_season]
continue
def stop(self):
"""
退出插件