mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-04 14:24:58 +08:00
fix(transfer): debounce monitored file success notifications (#7026)
This commit is contained in:
jxxghp
@@ -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()
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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` 再执行快照测试及一次
|
||||
|
||||
@@ -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)。
|
||||
|
||||
| 方法 | 路径 | 说明 |
|
||||
| :--- | :--- | :--- |
|
||||
|
||||
@@ -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 等基线均为零 |
|
||||
|
||||
@@ -8,14 +8,17 @@
|
||||
总大小（`total_size`）、音乐专辑及已知总曲目数（`album`、`total_tracks`）、
|
||||
电视剧季集范围（`season_episode`）等上下文；最终文案和排版由模板决定。
|
||||
目标目录可从 `__transferinfo__.target_diritem.path` 读取。
|
||||
多文件通知保留“查看整理记录”入口；单文件明确批次结束即发，不等待静默窗口。
|
||||
多文件通知保留“查看整理记录”入口；整目录或手动整理的明确批次结束即发，即使只有一个文件。
|
||||
|
||||
电影、电视剧和音乐使用同一个全局开关。关闭时保持原有按媒体作业发送的行为。
|
||||
目录设置继续使用原来的“发送通知”开关（`notify: true/false`），不增加通知粒度设置。
|
||||
批次聚合只影响展示消息，逐文件整理历史、完成/失败事件、刮削和种子状态回写保持原流程。
|
||||
失败通知仍由独立的失败聚合开关控制。
|
||||
|
||||
没有明确批次的外部任务采用三十秒静默窗口；连续到达会重置窗口。
|
||||
目录监控等自动单文件入口在各自批次结束后采用三十秒静默窗口，
|
||||
同组后续文件会重置窗口，将连续入库的同季剧集合并为一条通知并汇总集数、文件数和大小。
|
||||
没有明确批次的外部任务同样采用三十秒静默窗口。
|
||||
窗口内若还有同组批次未结束，会继续等待；不同来源、目标目录、季和用户仍分别通知。
|
||||
明确批次等待登记关闭且所有任务终结，较慢的曲目不会打断汇总，失败任务不会计入成功文件数。
|
||||
前台手动整理仍不发送成功通知。关闭进程时在整理 worker 收敛后刷新剩余缓冲。
|
||||
通知缓冲是内存展示状态，异常退出不保证重发此前已完成任务的成功通知；逐文件历史仍可查询。
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user