fix:优化下载器整理控制逻辑

This commit is contained in:
jxxghp
2026-01-21 22:21:17 +08:00
parent 0ba8d51b2a
commit 46466624ae
+68 -21
View File
@@ -151,7 +151,7 @@ class JobManager:
def running_task(self, task: TransferTask): def running_task(self, task: TransferTask):
""" """
任务运行中 设置任务运行中
""" """
with job_lock: with job_lock:
__mediaid__ = self.__get_id(task) __mediaid__ = self.__get_id(task)
@@ -165,7 +165,7 @@ class JobManager:
def finish_task(self, task: TransferTask): def finish_task(self, task: TransferTask):
""" """
任务完成 设置任务完成
""" """
with job_lock: with job_lock:
__mediaid__ = self.__get_id(task) __mediaid__ = self.__get_id(task)
@@ -179,7 +179,7 @@ class JobManager:
def fail_task(self, task: TransferTask): def fail_task(self, task: TransferTask):
""" """
任务失败 设置任务失败
""" """
with job_lock: with job_lock:
__mediaid__ = self.__get_id(task) __mediaid__ = self.__get_id(task)
@@ -198,7 +198,7 @@ class JobManager:
def remove_task(self, fileitem: FileItem) -> Optional[TransferJobTask]: def remove_task(self, fileitem: FileItem) -> Optional[TransferJobTask]:
""" """
移除所有作业中的整理任务 移除任务
""" """
with job_lock: with job_lock:
for mediaid in list(self._job_view): for mediaid in list(self._job_view):
@@ -233,7 +233,7 @@ class JobManager:
def is_done(self, task: TransferTask) -> bool: def is_done(self, task: TransferTask) -> bool:
""" """
检查某项作业是否整理完成(不管成功还是失败) 检查任务对应的作业是否整理完成(不管成功还是失败)
""" """
__metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season) __metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season)
__mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season) __mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season)
@@ -256,7 +256,7 @@ class JobManager:
def is_finished(self, task: TransferTask) -> bool: def is_finished(self, task: TransferTask) -> bool:
""" """
检查某项作业是否已完成且有成功的记录 检查任务对应的作业是否已完成且有成功的记录
""" """
__metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season) __metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season)
__mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season) __mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season)
@@ -282,7 +282,7 @@ class JobManager:
def is_success(self, task: TransferTask) -> bool: def is_success(self, task: TransferTask) -> bool:
""" """
检查某项作业是否全部成功 检查任务对应的作业是否全部成功
""" """
__metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season) __metaid__ = self.__get_meta_id(meta=task.meta, season=task.meta.begin_season)
__mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season) __mediaid__ = self.__get_media_id(media=task.mediainfo, season=task.meta.begin_season)
@@ -303,6 +303,18 @@ class JobManager:
media_success = True media_success = True
return meta_success and media_success return meta_success and media_success
def has_tasks(self, meta: MetaBase, mediainfo: Optional[MediaInfo] = None, season: Optional[int] = None) -> bool:
"""
判断是否有任务正在处理
"""
if mediainfo:
__mediaid__ = self.__get_media_id(media=meta, season=season)
if __mediaid__ in self._job_view:
return True
__metaid__ = self.__get_meta_id(meta=meta, season=season)
return __metaid__ in self._job_view
def success_tasks(self, media: MediaInfo, season: Optional[int] = None) -> List[TransferJobTask]: def success_tasks(self, media: MediaInfo, season: Optional[int] = None) -> List[TransferJobTask]:
""" """
获取某项任务成功的任务 获取某项任务成功的任务
@@ -313,6 +325,16 @@ class JobManager:
return [] return []
return [task for task in self._job_view[__mediaid__].tasks if task.state == "completed"] return [task for task in self._job_view[__mediaid__].tasks if task.state == "completed"]
def all_tasks(self, media: MediaInfo, season: Optional[int] = None) -> List[TransferJobTask]:
"""
获取全部任务
"""
__mediaid__ = self.__get_media_id(media=media, season=season)
with job_lock:
if __mediaid__ not in self._job_view:
return []
return self._job_view[__mediaid__].tasks
def count(self, media: MediaInfo, season: Optional[int] = None) -> int: def count(self, media: MediaInfo, season: Optional[int] = None) -> int:
""" """
获取某项任务成功总数 获取某项任务成功总数
@@ -478,7 +500,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
# 更新文件清单 # 更新文件清单
with job_lock: with job_lock:
transferinfo.file_list_new = self._success_target_files.pop(transferinfo.target_diritem.path, []) transferinfo.file_list_new = self._success_target_files.pop(transferinfo.target_diritem.path, [])
# 发送通知,实时手动整理时不发 # 发送通知,实时手动整理时不发
if transferinfo.need_notify and (task.background or not task.manual): if transferinfo.need_notify and (task.background or not task.manual):
se_str = None se_str = None
@@ -493,7 +515,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
transferinfo=transferinfo, transferinfo=transferinfo,
season_episode=se_str, season_episode=se_str,
username=task.username) username=task.username)
# 刮削事件 # 刮削事件
if transferinfo.need_scrape and self.__is_media_file(task.fileitem): if transferinfo.need_scrape and self.__is_media_file(task.fileitem):
self.eventmanager.send_event(EventType.MetadataScrape, { self.eventmanager.send_event(EventType.MetadataScrape, {
@@ -503,7 +525,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
'file_list': transferinfo.file_list_new, 'file_list': transferinfo.file_list_new,
'overwrite': False 'overwrite': False
}) })
# 移除已完成的任务 # 移除已完成的任务
with job_lock: with job_lock:
self.jobview.remove_job(task) self.jobview.remove_job(task)
@@ -544,22 +566,22 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
username=task.username, username=task.username,
link=settings.MP_DOMAIN('#/history') link=settings.MP_DOMAIN('#/history')
)) ))
# 整理失败 # 整理失败
with job_lock: with job_lock:
self.jobview.fail_task(task) self.jobview.fail_task(task)
# 全部整理完成且有成功的任务时 # 全部整理完成且有成功的任务时
if self.jobview.is_finished(task): if self.jobview.is_finished(task):
# 发送消息、刮削事件、移除任务 # 发送消息、刮削事件、移除任务
__all_finished() __all_finished()
return False, transferinfo.message return False, transferinfo.message
# task转移成功 # task转移成功
with job_lock: with job_lock:
self.jobview.finish_task(task) self.jobview.finish_task(task)
logger.info(f"{task.fileitem.name} 入库成功:{transferinfo.target_diritem.path}") logger.info(f"{task.fileitem.name} 入库成功:{transferinfo.target_diritem.path}")
# 新增task转移成功历史记录 # 新增task转移成功历史记录
@@ -583,10 +605,6 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
'downloader': task.downloader, 'downloader': task.downloader,
'download_hash': task.download_hash, 'download_hash': task.download_hash,
}) })
# 下载器整理任务更新任务标签
if task.download_hash:
self.transfer_completed(hashs=task.download_hash, downloader=task.downloader)
# task登记转移成功文件清单 # task登记转移成功文件清单
target_dir_path = transferinfo.target_diritem.path target_dir_path = transferinfo.target_diritem.path
@@ -618,6 +636,15 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
# 发送消息、刮削事件、移除任务 # 发送消息、刮削事件、移除任务
__all_finished() __all_finished()
# 全部整理完成不管成功还是失败
if self.jobview.is_done(task):
# 所有任务
tasks = self.jobview.all_tasks()
for t in tasks:
if t.download_hash:
# 设置种子状态为已整理
self.transfer_completed(hashs=t.download_hash, downloader=t.downloader)
return True, "" return True, ""
def put_to_queue(self, task: TransferTask): def put_to_queue(self, task: TransferTask):
@@ -915,15 +942,16 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
""" """
获取下载器中的种子列表,并执行整理 获取下载器中的种子列表,并执行整理
""" """
# 全局锁,避免定时服务重复
# 全局锁,避免重复处理
with downloader_lock: with downloader_lock:
# 获取下载器监控目录 # 获取下载器监控目录
download_dirs = DirectoryHelper().get_download_dirs() download_dirs = DirectoryHelper().get_download_dirs()
# 如果没有下载器监控的目录则不处理 # 如果没有下载器监控的目录则不处理
if not any(dir_info.monitor_type == "downloader" and dir_info.storage == "local" if not any(dir_info.monitor_type == "downloader" and dir_info.storage == "local"
for dir_info in download_dirs): for dir_info in download_dirs):
return True return True
logger.info("开始整理下载器中已经完成下载的文件 ...") logger.info("开始整理下载器中已经完成下载的文件 ...")
# 从下载器获取种子列表 # 从下载器获取种子列表
@@ -938,11 +966,13 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
for torrent in torrents: for torrent in torrents:
if global_vars.is_system_stopped: if global_vars.is_system_stopped:
break break
# 文件路径 # 文件路径
file_path = torrent.path file_path = torrent.path
if not file_path.exists(): if not file_path.exists():
logger.warn(f"文件不存在:{file_path}") logger.warn(f"文件不存在:{file_path}")
continue continue
# 检查是否为下载器监控目录中的文件 # 检查是否为下载器监控目录中的文件
is_downloader_monitor = False is_downloader_monitor = False
for dir_info in download_dirs: for dir_info in download_dirs:
@@ -956,15 +986,25 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
if not is_downloader_monitor: if not is_downloader_monitor:
logger.debug(f"文件 {file_path} 不在下载器监控目录中,不通过下载器进行整理") logger.debug(f"文件 {file_path} 不在下载器监控目录中,不通过下载器进行整理")
continue continue
# 查询下载记录识别情况 # 查询下载记录识别情况
downloadhis: DownloadHistory = DownloadHistoryOper().get_by_hash(torrent.hash) downloadhis: DownloadHistory = DownloadHistoryOper().get_by_hash(torrent.hash)
if downloadhis: if downloadhis:
# 获取自定义识别词
custom_words_list = None
if downloadhis.custom_words:
custom_words_list = downloadhis.custom_words.split('\n')
# 类型 # 类型
try: try:
mtype = MediaType(downloadhis.type) mtype = MediaType(downloadhis.type)
except ValueError: except ValueError:
mtype = MediaType.TV mtype = MediaType.TV
# 按TMDBID识别
# 识别元数据
metainfo = MetaInfoPath(file_path, custom_words=custom_words_list)
# 识别媒体信息
mediainfo = self.recognize_media(mtype=mtype, mediainfo = self.recognize_media(mtype=mtype,
tmdbid=downloadhis.tmdbid, tmdbid=downloadhis.tmdbid,
doubanid=downloadhis.doubanid, doubanid=downloadhis.doubanid,
@@ -975,10 +1015,17 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
# 更新自定义媒体类别 # 更新自定义媒体类别
if downloadhis.media_category: if downloadhis.media_category:
mediainfo.category = downloadhis.media_category mediainfo.category = downloadhis.media_category
else: else:
# 非MoviePilot下载的任务,按文件识别 # 非MoviePilot下载的任务,按文件识别
metainfo = MetaInfoPath(file_path)
mediainfo = None mediainfo = None
# 检查是否已经有任务处理中,如有则跳过本次整理
if self.jobview.has_tasks(meta=metainfo, mediainfo=mediainfo):
logger.info(f"有任务正在整理中,跳过本次整理 ...")
return False
# 执行异步整理,匹配源目录 # 执行异步整理,匹配源目录
self.do_transfer( self.do_transfer(
fileitem=FileItem( fileitem=FileItem(