diff --git a/app/application/transfer/notifications.py b/app/application/transfer/notifications.py new file mode 100644 index 000000000..268ebbd5e --- /dev/null +++ b/app/application/transfer/notifications.py @@ -0,0 +1,180 @@ +"""整理失败通知的分组键和短暂聚合。""" + +from __future__ import annotations + +import asyncio +import threading +from pathlib import Path +from typing import TYPE_CHECKING, Callable + +from app.application.transfer.feedback import TransferFailureNotification +from app.runtime.log import logger +from app.schemas.media import resolve_media_identity + +if TYPE_CHECKING: + from app.application.transfer.workflow import TransferTask + + +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._callbacks: dict[ + str, + Callable[[list[TransferFailureNotification]], None], + ] = {} + self._timers: dict[str, asyncio.TimerHandle] = {} + self._generations: dict[str, int] = {} + self._lock = threading.Lock() + self._closed = False + + def schedule( + self, + *, + group_key: str, + notification: TransferFailureNotification, + callback: Callable[[list[TransferFailureNotification]], None], + loop: asyncio.AbstractEventLoop, + ) -> None: + """从整理线程安全地把失败快照加入事件循环中的聚合缓冲。""" + # 先在调用线程登记快照，关闭流程才能覆盖已接收但尚未进入事件循环的通知。 + with self._lock: + if self._closed: + raise RuntimeError("整理失败通知聚合器正在关闭，不能再接收通知") + self._buffers.setdefault(group_key, []).append(notification) + self._callbacks[group_key] = callback + generation = self._generations.get(group_key, 0) + 1 + self._generations[group_key] = generation + try: + loop.call_soon_threadsafe( + self._schedule_on_loop, + group_key, + generation, + callback, + loop, + ) + except Exception as err: + logger.error( + f"创建整理失败通知聚合定时器失败，将立即发送 " + f"(group={group_key}): {err}" + ) + self.flush(group_key, generation, callback) + + def _schedule_on_loop( + self, + group_key: str, + generation: int, + callback: Callable[[list[TransferFailureNotification]], None], + loop: asyncio.AbstractEventLoop, + ) -> None: + """在所属事件循环中为已登记缓冲重置静默窗口。""" + schedule_error: Exception | None = None + with self._lock: + if ( + self._closed + or group_key not in self._buffers + or self._generations.get(group_key) != generation + ): + return + timer = self._timers.pop(group_key, None) + if timer: + timer.cancel() + try: + self._timers[group_key] = loop.call_later( + self.NOTIFICATION_DEBOUNCE_SECONDS, + self.flush, + group_key, + generation, + callback, + ) + except Exception as err: + schedule_error = err + if schedule_error is not None: + logger.error( + f"创建整理失败通知聚合定时器失败，将立即发送 " + f"(group={group_key}): {schedule_error}" + ) + self.flush(group_key, generation, callback) + + def flush( + self, + group_key: str, + generation: int, + callback: Callable[[list[TransferFailureNotification]], None], + ) -> None: + """发送一个分组内的聚合结果并释放缓冲。""" + with self._lock: + # 新通知已登记但 timer 重置回调尚未执行时，旧代不得提前发送新批次。 + if self._generations.get(group_key) != generation: + return + notifications = self._buffers.pop(group_key, []) + self._callbacks.pop(group_key, None) + timer = self._timers.pop(group_key, None) + self._generations.pop(group_key, None) + if timer: + timer.cancel() + if not notifications: + return + self._deliver(group_key, notifications, callback) + + @staticmethod + def _deliver( + group_key: str, + notifications: list[TransferFailureNotification], + callback: Callable[[list[TransferFailureNotification]], None], + ) -> None: + """调用聚合通知回调，并统一观察发送异常。""" + try: + callback(notifications) + except Exception as err: + logger.error(f"发送整理失败聚合通知失败 (group={group_key}): {err}") + + def close(self) -> None: + """停止接收新通知，取消定时器并同步发送全部已缓冲通知。""" + with self._lock: + if self._closed and not self._buffers and not self._timers: + return + self._closed = True + timers = list(self._timers.values()) + pending = [] + orphaned = [] + for group_key, notifications in self._buffers.items(): + callback = self._callbacks.get(group_key) + if callback is None: + orphaned.append((group_key, len(notifications))) + continue + pending.append((group_key, notifications, callback)) + self._timers.clear() + self._generations.clear() + self._buffers.clear() + self._callbacks.clear() + + for timer in timers: + timer.cancel() + for group_key, notification_count in orphaned: + logger.error( + f"整理失败通知聚合缓冲缺少发送回调，无法刷新 " + f"(group={group_key}, count={notification_count})" + ) + for group_key, notifications, callback in pending: + self._deliver(group_key, notifications, callback) diff --git a/app/application/transfer/workflow.py b/app/application/transfer/workflow.py index 0ded75a6b..a39ea5258 100644 --- a/app/application/transfer/workflow.py +++ b/app/application/transfer/workflow.py @@ -13,7 +13,6 @@ app.schemas -> app.schemas.transfer -> app.domain.* -> app.schemas.types -> app. TransferJob / TransferJobTask，那两个用 app.schemas 的同名 DTO——一个是工作项，一个是 视图，分开表达之后两边都不必再迁就对方。 """ -import asyncio import hashlib import threading from copy import deepcopy @@ -39,7 +38,11 @@ from pydantic import BaseModel, ConfigDict, PrivateAttr from app.application.history import DownloadHistorySnapshot from app.application.transfer import checkpoint as checkpoint_codec from app.application.transfer.execution import TransferExecutionCheckpoint -from app.application.transfer.feedback import TransferFailureNotification +from app.application.transfer.notifications import ( + TransferFailureNotification, + TransferFailureNotificationAggregator, + build_transfer_failure_group_key, +) from app.domain.context import MediaInfo, MusicInfo from app.domain.media import normalize_music_type from app.domain.meta.metabase import MetaBase @@ -1022,171 +1025,6 @@ class TransferQueueService: return self._list_tasks() -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._callbacks: dict[ - str, - Callable[[list[TransferFailureNotification]], None], - ] = {} - self._timers: dict[str, asyncio.TimerHandle] = {} - self._generations: dict[str, int] = {} - self._lock = threading.Lock() - self._closed = False - - def schedule( - self, - *, - group_key: str, - notification: TransferFailureNotification, - callback: Callable[[list[TransferFailureNotification]], None], - loop: asyncio.AbstractEventLoop, - ) -> None: - """从整理线程安全地把失败快照加入事件循环中的聚合缓冲。""" - # 先在调用线程登记快照，关闭流程才能覆盖已接收但尚未进入事件循环的通知。 - with self._lock: - if self._closed: - raise RuntimeError("整理失败通知聚合器正在关闭，不能再接收通知") - self._buffers.setdefault(group_key, []).append(notification) - self._callbacks[group_key] = callback - generation = self._generations.get(group_key, 0) + 1 - self._generations[group_key] = generation - try: - loop.call_soon_threadsafe( - self._schedule_on_loop, - group_key, - generation, - callback, - loop, - ) - except Exception as err: - logger.error( - f"创建整理失败通知聚合定时器失败，将立即发送 " - f"(group={group_key}): {err}" - ) - self.flush(group_key, generation, callback) - - def _schedule_on_loop( - self, - group_key: str, - generation: int, - callback: Callable[[list[TransferFailureNotification]], None], - loop: asyncio.AbstractEventLoop, - ) -> None: - """在所属事件循环中为已登记缓冲重置静默窗口。""" - schedule_error: Optional[Exception] = None - with self._lock: - if ( - self._closed - or group_key not in self._buffers - or self._generations.get(group_key) != generation - ): - return - timer = self._timers.pop(group_key, None) - if timer: - timer.cancel() - try: - self._timers[group_key] = loop.call_later( - self.NOTIFICATION_DEBOUNCE_SECONDS, - self.flush, - group_key, - generation, - callback, - ) - except Exception as err: - schedule_error = err - if schedule_error is not None: - logger.error( - f"创建整理失败通知聚合定时器失败，将立即发送 " - f"(group={group_key}): {schedule_error}" - ) - self.flush(group_key, generation, callback) - - def flush( - self, - group_key: str, - generation: int, - callback: Callable[[list[TransferFailureNotification]], None], - ) -> None: - """发送一个分组内的聚合结果并释放缓冲。""" - with self._lock: - # 新通知已登记但 timer 重置回调尚未执行时，旧代不得提前发送新批次。 - if self._generations.get(group_key) != generation: - return - notifications = self._buffers.pop(group_key, []) - self._callbacks.pop(group_key, None) - timer = self._timers.pop(group_key, None) - self._generations.pop(group_key, None) - if timer: - timer.cancel() - if not notifications: - return - self._deliver(group_key, notifications, callback) - - @staticmethod - def _deliver( - group_key: str, - notifications: list[TransferFailureNotification], - callback: Callable[[list[TransferFailureNotification]], None], - ) -> None: - """调用聚合通知回调，并统一观察发送异常。""" - try: - callback(notifications) - except Exception as err: - logger.error(f"发送整理失败聚合通知失败 (group={group_key}): {err}") - - def close(self) -> None: - """停止接收新通知，取消定时器并同步发送全部已缓冲通知。""" - with self._lock: - if self._closed and not self._buffers and not self._timers: - return - self._closed = True - timers = list(self._timers.values()) - pending = [] - orphaned = [] - for group_key, notifications in self._buffers.items(): - callback = self._callbacks.get(group_key) - if callback is None: - orphaned.append((group_key, len(notifications))) - continue - pending.append((group_key, notifications, callback)) - self._timers.clear() - self._generations.clear() - self._buffers.clear() - self._callbacks.clear() - - for timer in timers: - timer.cancel() - for group_key, notification_count in orphaned: - logger.error( - f"整理失败通知聚合缓冲缺少发送回调，无法刷新 " - f"(group={group_key}, count={notification_count})" - ) - for group_key, notifications, callback in pending: - self._deliver(group_key, notifications, callback) - - # 作业锁：JobManager 与 TransferChain 共享，保护整理作业视图。 job_lock = threading.Lock()