feat(transfer): aggregate failure notifications by media

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()
 

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

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"

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()