diff --git a/app/application/transfer.py b/app/application/transfer.py index 534d28852..834837f9d 100644 --- a/app/application/transfer.py +++ b/app/application/transfer.py @@ -16,6 +16,7 @@ TransferJob / TransferJobTask,那两个用 app.schemas 的同名 DTO——一 import asyncio import threading from copy import deepcopy +from dataclasses import dataclass from pathlib import Path from time import monotonic from typing import Callable, Dict, List, Optional, Tuple, Union @@ -155,6 +156,97 @@ class TransferQueueService: return self._list_tasks() +@dataclass(frozen=True, slots=True) +class TransferFailureNotification: + """整理失败聚合器保存的单条通知快照。""" + + media_title: str + season_episode: str + reason: str + history_id: Optional[int] + image: Optional[str] + username: Optional[str] + manual_identity: bool = False + + +def build_transfer_failure_group_key(task: TransferTask) -> str: + """构造主程序和第三方整理路径可共同使用的失败通知分组键。""" + media_source, media_id = resolve_media_identity(media=task.mediainfo) + if not media_source or not media_id: + media_source, media_id = resolve_media_identity(media=task) + season = getattr(task.meta, "begin_season", None) if task.meta else None + username = task.username or "" + if media_source and media_id: + return f"media:{media_source}:{media_id}:season:{season}:user:{username}" + if task.download_hash: + return f"download:{task.download_hash}:user:{username}" + source_path = str(task.fileitem.path) if task.fileitem else "" + parent_path = str(Path(source_path).parent) if source_path else "" + return f"path:{parent_path or source_path}:user:{username}" + + +class TransferFailureNotificationAggregator: + """在短暂静默窗口内按媒体合并整理失败通知。""" + + NOTIFICATION_DEBOUNCE_SECONDS = 30 + + def __init__(self) -> None: + """初始化分组缓冲和定时器。""" + self._buffers: dict[str, list[TransferFailureNotification]] = {} + self._timers: dict[str, asyncio.TimerHandle] = {} + + def schedule( + self, + *, + group_key: str, + notification: TransferFailureNotification, + callback: Callable[[list[TransferFailureNotification]], None], + loop: asyncio.AbstractEventLoop, + ) -> None: + """从整理线程安全地把失败快照加入事件循环中的聚合缓冲。""" + loop.call_soon_threadsafe( + self._schedule_on_loop, + group_key, + notification, + callback, + loop, + ) + + def _schedule_on_loop( + self, + group_key: str, + notification: TransferFailureNotification, + callback: Callable[[list[TransferFailureNotification]], None], + loop: asyncio.AbstractEventLoop, + ) -> None: + """在所属事件循环中更新缓冲并重置静默窗口。""" + self._buffers.setdefault(group_key, []).append(notification) + timer = self._timers.pop(group_key, None) + if timer: + timer.cancel() + self._timers[group_key] = loop.call_later( + self.NOTIFICATION_DEBOUNCE_SECONDS, + self.flush, + group_key, + callback, + ) + + def flush( + self, + group_key: str, + callback: Callable[[list[TransferFailureNotification]], None], + ) -> None: + """发送一个分组内的聚合结果并释放缓冲。""" + notifications = self._buffers.pop(group_key, []) + self._timers.pop(group_key, None) + if not notifications: + return + try: + callback(notifications) + except Exception as err: + logger.error(f"发送整理失败聚合通知失败 (group={group_key}): {err}") + + # 作业锁:JobManager 与 TransferChain 共享,保护整理作业视图。 job_lock = threading.Lock() diff --git a/app/chain/transfer.py b/app/chain/transfer.py index e04bdc856..9ab605e45 100755 --- a/app/chain/transfer.py +++ b/app/chain/transfer.py @@ -4,6 +4,7 @@ import re import threading import traceback import uuid +from collections import Counter from copy import deepcopy from pathlib import Path from typing import List, Optional, Tuple, Union, Dict, Callable, Any @@ -58,9 +59,12 @@ from app.runtime.reload import ConfigReloadMixin from app.application.transfer import ( FailedRetryScheduler, JobManager, + TransferFailureNotification, + TransferFailureNotificationAggregator, TransferQueue, TransferQueueService, TransferTask, + build_transfer_failure_group_key, job_lock, ) from app.chain._transfer import (EpisodeFormatMixin, FailedRetryMixin, @@ -125,6 +129,8 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo self.jobview = JobManager() # Agent重试管理器 self.retry_scheduler = FailedRetryScheduler() + # 整理失败通知聚合器 + self.failure_notification_aggregator = TransferFailureNotificationAggregator() # 待整理文件落盘登记,用于进程重启后回放内存队列里未完成的任务 self._pendingoper = TransferPendingOper() # 转移成功的文件清单 @@ -323,28 +329,10 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo }, ) - # 发送失败消息 - self.post_message( - Message( - mtype=MessageType.Manual, - title=f"{task.mediainfo.title_year} {task.meta.season_episode} 入库失败!", - text="\n".join( - [ - f"原因:{transferinfo.message or '未知'}", - ( - f"如果按钮不可用,可回复:\n```\n/redo {history.id}\n```" - if history - else "" - ), - ] - ).strip(), - image=task.mediainfo.get_message_image(), - username=task.username, - link=settings.MP_DOMAIN("#/history"), - buttons=self.build_failed_transfer_buttons( - history.id if history else None - ), - ) + self.queue_failed_transfer_notification( + task=task, + transferinfo=transferinfo, + history_id=history.id if history else None, ) # 设置任务失败 @@ -359,11 +347,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo try: # 使用 download_hash 或源文件父目录作为分组键, # 同一批次(如同一个种子)的失败记录会被合并为一次agent调用 - group_key = ( - task.download_hash or str(task.fileitem.path).rsplit("/", 1)[0] - if task.fileitem - else "" - ) + group_key = build_transfer_failure_group_key(task) asyncio.run_coroutine_threadsafe( self.retry_scheduler.schedule_retry( history.id, group_key=group_key @@ -544,6 +528,109 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo return ret_status, ret_message + def queue_failed_transfer_notification( + self, + *, + task: TransferTask, + transferinfo: TransferInfo, + history_id: Optional[int], + manual_identity: bool = False, + ) -> None: + """按配置逐条发送或按媒体聚合整理失败通知,供第三方整理补丁复用。""" + notification = TransferFailureNotification( + media_title=( + task.mediainfo.title_year + if task.mediainfo + else task.fileitem.name if task.fileitem else "未知媒体" + ), + season_episode=getattr(task.meta, "season_episode", "") or "", + reason=transferinfo.message or "未知", + history_id=history_id, + image=( + task.mediainfo.get_message_image() + if task.mediainfo and hasattr(task.mediainfo, "get_message_image") + else None + ), + username=task.username, + manual_identity=manual_identity, + ) + if not settings.TRANSFER_FAILURE_NOTIFICATION_AGGREGATION: + self._send_transfer_failure_notifications([notification]) + return + try: + self.failure_notification_aggregator.schedule( + group_key=build_transfer_failure_group_key(task), + notification=notification, + callback=self._send_transfer_failure_notifications, + loop=global_vars.loop, + ) + except Exception as err: + logger.error(f"加入整理失败通知聚合缓冲失败,将立即发送:{err}") + self._send_transfer_failure_notifications([notification]) + + def _send_transfer_failure_notifications( + self, + notifications: List[TransferFailureNotification], + ) -> None: + """把一个媒体分组的失败快照渲染为单条消息。""" + if not notifications: + return + first = notifications[0] + history_ids = [item.history_id for item in notifications if item.history_id] + if len(notifications) == 1: + history_hint = ( + ( + "如果按钮不可用,可回复:\n" + f"```\n/redo {history_ids[0]}\n" + f"/redo {history_ids[0]} [media_source]|[media_id]|[类型]\n```\n" + "自动重试或手动识别整理。" + if first.manual_identity + else f"如果按钮不可用,可回复:\n```\n/redo {history_ids[0]}\n```" + ) + if history_ids + else "" + ) + text = "\n".join([f"原因:{first.reason}", history_hint]).strip() + buttons = self.build_failed_transfer_buttons( + history_ids[0] if history_ids else None + ) + title = ( + f"{first.media_title} 未识别到媒体信息,无法入库!" + if first.manual_identity + else f"{first.media_title} {first.season_episode} 入库失败!" + ) + else: + reason_counts = Counter(item.reason for item in notifications) + reason_lines = [ + f"- {reason} × {count}" + for reason, count in reason_counts.most_common() + ] + history_text = "、".join(f"#{history_id}" for history_id in history_ids) + text_parts = [ + f"失败文件:{len(notifications)} 个", + "原因统计:", + *reason_lines, + ] + if history_text: + text_parts.extend([f"整理记录:{history_text}", "可在整理历史中批量处理。"]) + text = "\n".join(text_parts) + buttons = [[{ + "text": "批量处理", + "url": settings.MP_DOMAIN("#/history"), + }]] + title = f"{first.media_title} 入库失败({len(notifications)} 个文件)" + self.post_message( + Message( + mtype=MessageType.Manual, + title=title, + text=text, + image=first.image, + username=first.username, + link=settings.MP_DOMAIN("#/history"), + buttons=buttons, + ) + ) + def __get_transfer_target_dir_path( self, transferinfo: Optional[TransferInfo] ) -> Optional[str]: @@ -1003,33 +1090,16 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo download_hash=task.download_hash, transfer_history_oper=transferhis, ) - self.post_message( - Message( - mtype=MessageType.Manual, - title=f"{task.fileitem.name} 未识别到媒体信息,无法入库!", - # 历史落库失败时 his 为 None(add_transfer_fail 末尾的 - # get_by_src 查不到即返回 None),此时 /redo 无 ID 可用, - # 只省去这段指引而不是让整条通知连同后续的作业清理、 - # 种子完成标记一起崩在 NoneType 上 - text="\n".join( - [ - "原因:未识别到媒体信息", - ( - "如果按钮不可用,可回复:\n" - f"```\n/redo {his.id}\n" - f"/redo {his.id} [media_source]|[media_id]|[类型]\n```\n" - "自动重试或手动识别整理。" - if his - else "" - ), - ] - ).strip(), - username=task.username, - link=settings.MP_DOMAIN("#/history"), - buttons=self.build_failed_transfer_buttons( - his.id if his else None - ), - ) + self.queue_failed_transfer_notification( + task=task, + transferinfo=TransferInfo( + success=False, + fileitem=task.fileitem, + message="未识别到媒体信息", + transfer_type=task.transfer_type, + ), + history_id=his.id if his else None, + manual_identity=True, ) # 任务失败,直接移除task self.jobview.remove_task(task.fileitem) @@ -1045,12 +1115,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo ): try: # 使用 download_hash 或源文件父目录作为分组键 - group_key = ( - task.download_hash - or str(task.fileitem.path).rsplit("/", 1)[0] - if task.fileitem - else "" - ) + group_key = build_transfer_failure_group_key(task) asyncio.run_coroutine_threadsafe( self.retry_scheduler.schedule_retry( his.id, group_key=group_key diff --git a/app/runtime/config.py b/app/runtime/config.py index fce144102..f9e6d0cbe 100644 --- a/app/runtime/config.py +++ b/app/runtime/config.py @@ -718,6 +718,8 @@ class ConfigModel(BaseModel): AI_AGENT_VERBOSE: bool = False # AI智能体自动重试整理失败记录开关 AI_AGENT_RETRY_TRANSFER: bool = False + # 是否按媒体聚合整理失败通知,关闭时保持逐条发送 + TRANSFER_FAILURE_NOTIFICATION_AGGREGATION: bool = True # 音频输入提供商:openai/openai_chat_audio/mimo/minimax AUDIO_INPUT_PROVIDER: str = "openai" diff --git a/tests/test_transfer_failure_notification_aggregation.py b/tests/test_transfer_failure_notification_aggregation.py new file mode 100644 index 000000000..0a0e80ca4 --- /dev/null +++ b/tests/test_transfer_failure_notification_aggregation.py @@ -0,0 +1,167 @@ +from unittest.mock import Mock + +from app.chain import transfer as transfer_module +from app.chain.transfer import TransferChain +from app.application.transfer import ( + TransferFailureNotification, + TransferFailureNotificationAggregator, + TransferTask, + build_transfer_failure_group_key, +) +from app.domain.context import MediaInfo +from app.domain.metainfo import MetaInfo +from app.runtime.config import ConfigModel +from app.schemas.file import FileItem +from app.schemas.transfer import TransferInfo +from app.schemas.types import MediaSource, MediaType + + +class _Timer: + """记录静默窗口是否因新失败到达而取消。""" + + def __init__(self, callback, args): + """保存定时回调与参数。""" + self.callback = callback + self.args = args + self.cancelled = False + + def cancel(self): + """标记当前定时器已取消。""" + self.cancelled = True + + +class _Loop: + """同步执行线程安全入队,并保留延迟回调供测试触发。""" + + def __init__(self): + """初始化定时器清单。""" + self.timers = [] + + def call_soon_threadsafe(self, callback, *args): + """同步执行本应投递到事件循环的回调。""" + callback(*args) + + def call_later(self, _delay, callback, *args): + """保存延迟回调并返回可取消句柄。""" + timer = _Timer(callback, args) + self.timers.append(timer) + return timer + + +def _task(*, episode: int, download_hash: str = "hash-1") -> TransferTask: + """构造同一媒体不同剧集的整理任务。""" + return TransferTask( + fileitem=FileItem( + storage="local", + path=f"/downloads/Show/Show.S01E{episode:02d}.mkv", + type="file", + name=f"Show.S01E{episode:02d}.mkv", + ), + meta=MetaInfo(f"Show S01E{episode:02d}"), + mediainfo=MediaInfo( + media_source=MediaSource.TMDB, + media_id="100", + tmdb_id=100, + title="测试剧", + type=MediaType.TV, + year="2026", + ), + download_hash=download_hash, + username="tester", + ) + + +def test_failure_group_key_prefers_media_identity_and_season(): + """同一媒体同一季应跨文件共享分组键。""" + first = build_transfer_failure_group_key(_task(episode=1, download_hash="hash-a")) + second = build_transfer_failure_group_key(_task(episode=2, download_hash="hash-b")) + + assert first == second + assert first == "media:themoviedb:100:season:1:user:tester" + + +def test_failure_notification_aggregation_defaults_on(): + """整理失败通知聚合默认开启。""" + field = ConfigModel.model_fields["TRANSFER_FAILURE_NOTIFICATION_AGGREGATION"] + + assert field.default is True + + +def test_aggregator_debounces_same_group_and_flushes_once(): + """同组失败应重置定时器并一次性回调全部快照。""" + loop = _Loop() + aggregator = TransferFailureNotificationAggregator() + callback = Mock() + notices = [ + TransferFailureNotification("测试剧 (2026)", "S01E01", "原因A", 1, None, "tester"), + TransferFailureNotification("测试剧 (2026)", "S01E02", "原因B", 2, None, "tester"), + ] + + for notice in notices: + aggregator.schedule( + group_key="media:test", + notification=notice, + callback=callback, + loop=loop, + ) + + assert loop.timers[0].cancelled is True + assert loop.timers[1].cancelled is False + loop.timers[1].callback(*loop.timers[1].args) + callback.assert_called_once_with(notices) + + +def test_aggregated_message_contains_count_reason_stats_and_batch_entry(): + """聚合消息应给出失败数、原因统计、历史 ID 和批量处理入口。""" + chain = object.__new__(TransferChain) + sent = [] + chain.post_message = sent.append + notices = [ + TransferFailureNotification("测试剧 (2026)", "S01E01", "未识别到媒体信息", 11, None, "tester"), + TransferFailureNotification("测试剧 (2026)", "S01E02", "目标已存在", 12, None, "tester"), + TransferFailureNotification("测试剧 (2026)", "S01E03", "目标已存在", 13, None, "tester"), + ] + + chain._send_transfer_failure_notifications(notices) + + assert len(sent) == 1 + message = sent[0] + assert message.title == "测试剧 (2026) 入库失败(3 个文件)" + assert "失败文件:3 个" in message.text + assert "- 目标已存在 × 2" in message.text + assert "整理记录:#11、#12、#13" in message.text + assert message.buttons == [[{ + "text": "批量处理", + "url": transfer_module.settings.MP_DOMAIN("#/history"), + }]] + + +def test_enabled_queue_uses_shared_group_key(monkeypatch): + """开启聚合后公开通知入口应投递到聚合器而不是立即发送。""" + chain = object.__new__(TransferChain) + chain.failure_notification_aggregator = Mock() + chain.post_message = Mock() + task = _task(episode=1) + transferinfo = TransferInfo( + success=False, + fileitem=task.fileitem, + message="整理失败", + transfer_type="copy", + ) + loop = transfer_module.global_vars.loop + monkeypatch.setattr( + transfer_module.settings, + "TRANSFER_FAILURE_NOTIFICATION_AGGREGATION", + True, + ) + chain.queue_failed_transfer_notification( + task=task, + transferinfo=transferinfo, + history_id=22, + ) + + chain.failure_notification_aggregator.schedule.assert_called_once() + kwargs = chain.failure_notification_aggregator.schedule.call_args.kwargs + assert kwargs["group_key"] == build_transfer_failure_group_key(task) + assert kwargs["loop"] is loop + chain.post_message.assert_not_called()