From 8c59a89d57fb9cbfc756cf4f6de9eab86166084e Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 16 Oct 2026 13:24:31 +0800 Subject: [PATCH] fix(transfer): debounce monitored file success notifications (#7026) --- app/application/transfer/notifications.py | 34 ++++- app/chain/transfer/scrape.py | 9 +- app/chain/transfer/workflow.py | 9 ++ docs/architecture-overview.md | 4 +- docs/mcp-api.md | 2 +- docs/refactor/optimization-checklist.md | 2 +- docs/transfer-notifications.md | 7 +- .../architecture/dependency-baseline.json | 6 +- tests/test_transfer_movie_collection.py | 4 +- ...ansfer_success_notification_aggregation.py | 137 +++++++++++++++++- 10 files changed, 193 insertions(+), 21 deletions(-) diff --git a/app/application/transfer/notifications.py b/app/application/transfer/notifications.py index 33edefd3f..4f47029c2 100644 --- a/app/application/transfer/notifications.py +++ b/app/application/transfer/notifications.py @@ -72,6 +72,7 @@ class TransferNotificationAggregator[Notification]: self._generations: dict[str, int] = {} self._next_generation = 0 self._waiting_batches: dict[str, set[str]] = {} + self._debounce_loops: dict[str, asyncio.AbstractEventLoop] = {} self._lock = threading.Lock() self._closed = False @@ -84,7 +85,7 @@ class TransferNotificationAggregator[Notification]: loop: Optional[asyncio.AbstractEventLoop] = None, defer_until: Optional[str] = None, ) -> None: - """先登记快照；明确批次由完成信号刷新，其余通知使用事件循环静默窗口。""" + """先登记快照；明确批次等待完成信号，其余通知使用事件循环静默窗口。""" # 先在调用线程登记快照，关闭流程才能覆盖已接收但尚未进入事件循环的通知。 with self._lock: if self._closed: @@ -97,8 +98,20 @@ class TransferNotificationAggregator[Notification]: self._generations[group_key] = generation if defer_until: self._waiting_batches.setdefault(group_key, set()).add(defer_until) + if loop is not None: + self._debounce_loops[group_key] = loop if defer_until: return + self._schedule_flush(group_key, generation, callback, loop) + + def _schedule_flush( + self, + group_key: str, + generation: int, + callback: Callable[[list[Notification]], None], + loop: Optional[asyncio.AbstractEventLoop], + ) -> None: + """统一投递静默窗口，事件循环不可用时回退为立即发送。""" if loop is None: self.flush(group_key, generation, callback) return @@ -170,22 +183,28 @@ class TransferNotificationAggregator[Notification]: timer = self._timers.pop(group_key, None) self._generations.pop(group_key, None) self._waiting_batches.pop(group_key, None) + self._debounce_loops.pop(group_key, None) if timer: timer.cancel() if not notifications: return self._deliver(group_key, notifications, callback) - def release_batch(self, batch_id: str) -> None: - """批次全部终结后刷新相关分组；单文件批次同样即时发送且不等待静默窗。""" + def release_batch(self, batch_id: str, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None: + """批次终结后释放等待；传入循环时合并后续文件，同组仍须等待其他批次。""" with self._lock: ready = [] for group_key, batches in self._waiting_batches.items(): - batches.discard(batch_id) + if batch_id not in batches: + continue + batches.remove(batch_id) + if loop is not None: + self._debounce_loops[group_key] = loop if not batches: - ready.append((group_key, self._generations[group_key], self._callbacks[group_key])) - for group_key, generation, callback in ready: - self.flush(group_key, generation, callback) + ready.append((group_key, self._generations[group_key], self._callbacks[group_key], + self._debounce_loops.get(group_key))) + for group_key, generation, callback, debounce_loop in ready: + self._schedule_flush(group_key, generation, callback, debounce_loop) @staticmethod def _deliver( @@ -219,6 +238,7 @@ class TransferNotificationAggregator[Notification]: self._buffers.clear() self._callbacks.clear() self._waiting_batches.clear() + self._debounce_loops.clear() for timer in timers: timer.cancel() diff --git a/app/chain/transfer/scrape.py b/app/chain/transfer/scrape.py index 679cdc951..d56613ddc 100644 --- a/app/chain/transfer/scrape.py +++ b/app/chain/transfer/scrape.py @@ -6,6 +6,7 @@ from app.application.transfer.workflow import TransferTask, job_lock from app.chain._contracts import TransferMixinHost from app.chain.transfer.contract import _TransferOwnerBase from app.domain.context import MusicInfo +from app.runtime.loop import main_loop_registry from app.schemas.transfer import TransferInfo from app.schemas.types import ( EventType, @@ -176,7 +177,7 @@ class ScrapeBatchMixin(_TransferOwnerBase): def _flush_scrape_batch_if_ready(self, batch_id: Optional[str]): """ - 批次任务全部结束后发送聚合后的刮削事件。 + 批次任务全部结束后释放成功通知等待，并发送聚合后的刮削事件。 """ if not batch_id: return @@ -190,11 +191,15 @@ class ScrapeBatchMixin(_TransferOwnerBase): ): return targets = list(batch.get("targets", {}).values()) + notification_debounce = batch.get("notification_debounce", False) self._scrape_batches.pop(batch_id, None) aggregator = getattr(self, "success_notification_aggregator", None) if aggregator is not None: - aggregator.release_batch(batch_id) + aggregator.release_batch( + batch_id, + loop=main_loop_registry.current if notification_debounce else None, + ) for target in targets: fileitem = target.get("fileitem") if not fileitem: diff --git a/app/chain/transfer/workflow.py b/app/chain/transfer/workflow.py index 4d1766c57..8fe1eecc7 100644 --- a/app/chain/transfer/workflow.py +++ b/app/chain/transfer/workflow.py @@ -22,6 +22,7 @@ from app.application.transfer.workflow import ( TransferAdmission, TransferAdmissionConflictError, TransferTask, + job_lock, ) from app.chain.media import MediaChain from app.chain.storage import StorageChain @@ -653,6 +654,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): downloader=downloader, download_hash=download_hash, transfer_batch_id=transfer_batch_id, + notification_debounce=fileitem.type == "file" and not manual, manual=bool(manual), background=bool(background), preview=bool(preview), @@ -720,6 +722,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): downloader: Optional[str], download_hash: Optional[str], transfer_batch_id: str, + notification_debounce: bool, manual: bool, background: bool, preview: bool, @@ -745,6 +748,12 @@ class TransferWorkflowOwner(_TransferOwnerBase): and sum(self._is_audio_file(item) for item, _ in file_items) > 1 ) music_batch_context = self._prepare_music_batch_context(file_items, batch_mtype) + # 监控逐文件调用没有完整批次边界，完成后仍需静默窗口等待同组后续文件。 + with job_lock: + self._scrape_batches[transfer_batch_id] = { + "pending": set(), "targets": {}, "closed": False, + "notification_debounce": notification_debounce, + } try: for file_item, bluray_dir in self._music_transfer_items(file_items, music_batch_context): if runtime_stop_state.is_system_stopped: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 6c3ac6fbd..2af6e7dad 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -763,7 +763,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1121 | -| 内部导入边 | 9,559 | +| 内部导入边 | 9,561 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 49（债务已清零，49 条精确 containment） | @@ -773,7 +773,7 @@ flowchart LR | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | -成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,559；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 +目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,561；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 43e0839f5..432f8f726 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -500,7 +500,7 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返 影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，未绑定身份的音乐标题及目录识别按 `SEARCH_SOURCE` 中内置音乐来源的配置顺序回退（MusicBrainz、TheAudioDB、豆瓣音乐），未选择音乐来源时兼容默认 MusicBrainz。显式 `media_source` / 主身份或 MusicBrainz 发行标签固定所属来源；来源之间不转换或拼接 ID。`MediaInfo` 响应仍可能包含 `tmdb_id`、`douban_id`、`bangumi_id`、`anilist_id` 等跨源映射辅助字段，但这些字段不是通用请求入口。明确归属 `/tmdb`、`/douban`、`/bangumi`、`/anilist` 的接口，以及固定使用 TMDB 的剧集组和排期接口，仍可按其单数据源契约接收原生 ID。 -整理目录的 `notify` 保持布尔类型；成功通知聚合统一由全局开关控制。开启 `TRANSFER_SUCCESS_NOTIFICATION_AGGREGATION`，明确批次完成后一次发送，外部无批次任务使用三十秒静默窗口。详见 [整理成功通知粒度](transfer-notifications.md)。 +整理目录的 `notify` 保持布尔类型；成功通知聚合统一由全局开关控制。开启 `TRANSFER_SUCCESS_NOTIFICATION_AGGREGATION`，整目录或手动整理的明确批次完成后一次发送；目录监控等自动单文件入口在批次结束后使用三十秒静默窗口，外部无批次任务同样按静默窗口合并。详见 [整理成功通知粒度](transfer-notifications.md)。 | 方法 | 路径 | 说明 | | :--- | :--- | :--- | diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 1bca494c9..cbfed3878 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -100,7 +100,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1121 / 9,559 | `dependency-baseline.json` 当前快照；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC | +| 宿主 Python 模块 / 内部依赖边 | 1121 / 9,561 | `dependency-baseline.json` 当前快照；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/transfer-notifications.md b/docs/transfer-notifications.md index 595db0630..12288959c 100644 --- a/docs/transfer-notifications.md +++ b/docs/transfer-notifications.md @@ -8,14 +8,17 @@ 总大小（`total_size`）、音乐专辑及已知总曲目数（`album`、`total_tracks`）、 电视剧季集范围（`season_episode`）等上下文；最终文案和排版由模板决定。 目标目录可从 `__transferinfo__.target_diritem.path` 读取。 -多文件通知保留“查看整理记录”入口；单文件明确批次结束即发，不等待静默窗口。 +多文件通知保留“查看整理记录”入口；整目录或手动整理的明确批次结束即发，即使只有一个文件。 电影、电视剧和音乐使用同一个全局开关。关闭时保持原有按媒体作业发送的行为。 目录设置继续使用原来的“发送通知”开关（`notify: true/false`），不增加通知粒度设置。 批次聚合只影响展示消息，逐文件整理历史、完成/失败事件、刮削和种子状态回写保持原流程。 失败通知仍由独立的失败聚合开关控制。 -没有明确批次的外部任务采用三十秒静默窗口；连续到达会重置窗口。 +目录监控等自动单文件入口在各自批次结束后采用三十秒静默窗口， +同组后续文件会重置窗口，将连续入库的同季剧集合并为一条通知并汇总集数、文件数和大小。 +没有明确批次的外部任务同样采用三十秒静默窗口。 +窗口内若还有同组批次未结束，会继续等待；不同来源、目标目录、季和用户仍分别通知。 明确批次等待登记关闭且所有任务终结，较慢的曲目不会打断汇总，失败任务不会计入成功文件数。 前台手动整理仍不发送成功通知。关闭进程时在整理 worker 收敛后刷新剩余缓冲。 通知缓冲是内存展示状态，异常退出不保证重发此前已完成任务的成功通知；逐文件历史仍可查询。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 1ddff2386..66737a486 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -941,8 +941,8 @@ "runtime_only": true } }, - "edge_count": 9559, - "edge_sha256": "4509a0f286f9a1314a125872c4bbe8218ac3afb6b480e4e9e4303f8ea519efba", + "edge_count": 9561, + "edge_sha256": "33be79e78a558a10f4add55117460d34d0e763269d71711bc27b387fe2c5552d", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -6009,6 +6009,8 @@ "app.chain.transfer.scrape -> app.chain.transfer.contract", "app.chain.transfer.scrape -> app.domain", "app.chain.transfer.scrape -> app.domain.context", + "app.chain.transfer.scrape -> app.runtime", + "app.chain.transfer.scrape -> app.runtime.loop", "app.chain.transfer.scrape -> app.schemas", "app.chain.transfer.scrape -> app.schemas.transfer", "app.chain.transfer.scrape -> app.schemas.types", diff --git a/tests/test_transfer_movie_collection.py b/tests/test_transfer_movie_collection.py index 654041881..3272fb373 100644 --- a/tests/test_transfer_movie_collection.py +++ b/tests/test_transfer_movie_collection.py @@ -5,7 +5,7 @@ import pytest from app.application.history import DownloadHistorySnapshot from app.application.transfer.workflow import TransferTask -from app.chain.transfer import TransferChain +from app.chain.transfer.facade import TransferChain from app.domain.context import MediaInfo from app.domain.meta.metabase import MetaBase from app.runtime.config import settings @@ -16,6 +16,7 @@ from app.schemas.types import MediaSource, MediaType def _make_chain() -> TransferChain: """构造不启动后台线程的整理链测试实例。""" chain = object.__new__(TransferChain) + chain._scrape_batches = {} chain._media_exts = settings.RMT_MEDIAEXT chain._subtitle_exts = settings.RMT_SUBEXT chain._audio_exts = settings.RMT_AUDIOEXT @@ -46,6 +47,7 @@ class _FileMeta(MetaBase): name = None def __init__(self, year: str = "2013"): + """初始化用于检验合集年份冲突的单部电影元数据。""" super().__init__(title="The Hunger Games Catching Fire") self.name = "The Hunger Games Catching Fire" self.year = year diff --git a/tests/test_transfer_success_notification_aggregation.py b/tests/test_transfer_success_notification_aggregation.py index 5e4cb7e1f..680b8adbd 100644 --- a/tests/test_transfer_success_notification_aggregation.py +++ b/tests/test_transfer_success_notification_aggregation.py @@ -25,6 +25,8 @@ from app.runtime.loop import main_loop_registry from app.schemas.file import FileItem from app.schemas.transfer import TransferInfo from app.schemas.types import ContentType, EventType, MediaType, MessageType, SystemConfigKey +from tests.test_manual_transfer_history import _patch_transfer_planning +from tests.test_transfer_sync_extra_files import make_transfer_chain class _Loop: @@ -33,17 +35,25 @@ class _Loop: def __init__(self): """初始化可手动触发的定时回调。""" self.timers = [] + self.now = 0 def call_soon_threadsafe(self, callback, *args): """同步登记事件循环回调。""" callback(*args) - def call_later(self, _delay, callback, *args): + def call_later(self, delay, callback, *args): """保存定时器并提供可取消的句柄。""" - timer = SimpleNamespace(callback=callback, args=args, cancel=Mock()) + timer = SimpleNamespace(callback=callback, args=args, cancel=Mock(), deadline=self.now + delay) self.timers.append(timer) return timer + def advance(self, seconds): + """推进虚拟时间，只触发已到期且未取消的定时器。""" + self.now += seconds + for timer in list(self.timers): + if timer.deadline <= self.now and not timer.cancel.called: + timer.callback(*timer.args) + def flush(self): """按登记顺序触发定时器，包含已取消回调以验证代际保护。""" for timer in self.timers: @@ -170,6 +180,120 @@ def test_unbatched_events_use_quiet_window_and_deduplicate_paths(chain, monkeypa assert chain.post_message.call_args.kwargs["transferinfo"].total_size == 200 +@pytest.mark.parametrize("manual,source_type", [(False, "file"), (True, "file"), (False, "dir")]) +def test_single_file_workflow_notification_policy(chain, monkeypatch, manual, source_type): + """真实工作流仅为自动文件入口启用静默窗，目录与手动单文件仍在批次结束即发。""" + owner = make_transfer_chain() + owner.runtime_config = chain.runtime_config + owner.success_notification_aggregator = chain.success_notification_aggregator + owner.post_message = chain.post_message + task = _task() + fileitem = FileItem(storage="local", path="/downloads/Show/S01E01.mkv", name="S01E01.mkv", + extension="mkv", type="file") + _patch_transfer_planning(monkeypatch, owner, fileitem, None, [], []) + owner._register_scrape_batch_task = TransferChain._register_scrape_batch_task.__get__(owner) + owner._close_scrape_batch = TransferChain._close_scrape_batch.__get__(owner) + queued = [] + + def enqueue(task): + """保留真实批次登记，只隔离后台线程与文件执行。""" + owner._register_scrape_batch_task(task) + queued.append(task) + return True + + owner.put_to_queue = enqueue + loop = _Loop() + monkeypatch.setattr(type(main_loop_registry), "current", property(lambda _self: loop)) + source = fileitem if source_type == "file" else FileItem(storage="local", path="/downloads/Album/", type="dir") + + success, _ = owner.do_transfer(fileitem=source, manual=manual, background=True) + + assert success and len(queued) == 1 + queued[0].meta, queued[0].mediainfo = task.meta, task.mediainfo + owner.notify_successful_transfer(queued[0], _result()) + owner._finish_scrape_batch_task(queued[0]) + if not manual and source_type == "file": + owner.post_message.assert_not_called() + loop.advance(30) + owner.post_message.assert_called_once() + assert not owner._scrape_batches + + +def test_monitor_episode_batches_merge_after_last_completion(chain, monkeypatch): + """13 个间隔十秒的单文件批次只在末集静默三十秒后发一条季集汇总。""" + loop = _Loop() + monkeypatch.setattr(type(main_loop_registry), "current", property(lambda _self: loop)) + batch_ids = [] + for number in range(1, 14): + loop.advance(10) + batch_id = f"monitor-{number}" + batch_ids.append(batch_id) + task = _task(number, batch=batch_id, download_hash=None) + task.meta = MetaInfo(f"Show S01E{number:02d}") + task.mediainfo = MediaInfo(title="Show", year="2026", type=MediaType.TV) + chain._register_scrape_batch_task(task) + chain._scrape_batches[batch_id]["notification_debounce"] = True + chain._close_scrape_batch(batch_id) + chain.notify_successful_transfer(task, _result(number)) + chain._finish_scrape_batch_task(task) + chain.post_message.assert_not_called() + assert len(set(batch_ids)) == 13 + loop.advance(29) + chain.post_message.assert_not_called() + loop.advance(1) + chain.post_message.assert_called_once() + context = chain.post_message.call_args.kwargs + assert context["season_episode"] == "S01 E01-E13" + assert context["transferinfo"].file_count == 13 + assert context["transferinfo"].total_size == 1300 + assert chain.post_message.call_args.args[0].title is None + assert not chain._scrape_batches + + +@pytest.mark.parametrize("monitor_first", [False, True]) +def test_monitor_window_survives_overlapping_explicit_batch(monitor_first): + """重叠批次两种完成顺序都保留静默窗，旧定时器不得越过新批次等待。""" + loop = _Loop() + aggregator = TransferNotificationAggregator[str]() + callback = Mock() + aggregator.schedule(group_key="group", notification="monitor", callback=callback, defer_until="monitor") + aggregator.release_batch("monitor", loop=loop) + old_timer = loop.timers[-1] + aggregator.schedule(group_key="group", notification="batch", callback=callback, defer_until="batch") + aggregator.schedule(group_key="group", notification="monitor2", callback=callback, defer_until="monitor2") + old_timer.callback(*old_timer.args) + if monitor_first: + aggregator.release_batch("monitor2", loop=loop) + else: + aggregator.release_batch("batch") + loop.advance(60) + callback.assert_not_called() + if monitor_first: + aggregator.release_batch("batch") + else: + aggregator.release_batch("monitor2", loop=loop) + loop.advance(29) + aggregator.release_batch("unrelated") + aggregator.release_batch("monitor2", loop=loop) + callback.assert_not_called() + loop.advance(1) + callback.assert_called_once_with(["monitor", "batch", "monitor2"]) + assert not aggregator._debounce_loops + + +def test_monitor_timer_failure_falls_back_without_losing_notification(): + """批次释放时循环停止要立即发送，关闭时不再重复发送。""" + loop = Mock() + loop.call_soon_threadsafe.side_effect = RuntimeError("loop closed") + aggregator = TransferNotificationAggregator[str]() + callback = Mock() + aggregator.schedule(group_key="group", notification="monitor", callback=callback, defer_until="monitor") + aggregator.release_batch("monitor", loop=loop) + aggregator.close() + callback.assert_called_once_with(["monitor"]) + assert not aggregator._debounce_loops + + def test_overlapping_batches_wait_for_both_and_shutdown_flushes_once(chain): """同组重叠批次不会互相提前刷新；关闭发送剩余快照且重复关闭幂等。""" for number, batch in [(1, "a"), (2, "b")]: @@ -304,7 +428,8 @@ def test_old_callback_cannot_flush_recreated_group(): @pytest.mark.parametrize("media_type", [MediaType.MOVIE, MediaType.TV, MediaType.MUSIC]) @pytest.mark.parametrize("file_count", [1, 3]) -def test_success_notifications_render_configured_template(chain, monkeypatch, media_type, file_count): +@pytest.mark.parametrize("debounce", [False, True]) +def test_success_notifications_render_configured_template(chain, monkeypatch, media_type, file_count, debounce): """单文件与汇总通知经真实入口和模板引擎渲染，隔离配置、历史和渠道边界。""" template = """{ 'title': '自定义 {{ title_year }}{% if track_number %} #{{ track_number }}{% endif %}' @@ -319,6 +444,8 @@ def test_success_notifications_render_configured_template(chain, monkeypatch, me chain.post_message = TransferChain.post_message.__get__(chain) chain.messageoper = Mock() chain._dispatch_notification_steps = Mock() + loop = _Loop() + monkeypatch.setattr(type(main_loop_registry), "current", property(lambda _self: loop)) for number in range(1, file_count + 1): task = _task(number) @@ -326,10 +453,14 @@ def test_success_notifications_render_configured_template(chain, monkeypatch, me task.meta = MetaInfo("Movie 2026" if media_type == MediaType.MOVIE else f"Show S01E{number:02d}") task.mediainfo = MediaInfo(title="Movie" if media_type == MediaType.MOVIE else "Show", year="2026", type=media_type) chain._register_scrape_batch_task(task) + chain._scrape_batches["batch"]["notification_debounce"] = debounce chain.notify_successful_transfer(task, _result(number)) chain._finish_scrape_batch_task(task) chain._dispatch_notification_steps.assert_not_called() chain._close_scrape_batch("batch") + if debounce: + chain._dispatch_notification_steps.assert_not_called() + loop.advance(30) chain._dispatch_notification_steps.assert_called_once() message = chain._dispatch_notification_steps.call_args.args[0]