From 11303a8e88f663feed8e0f7bd3f25a2a8b67a047 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 8 Oct 2026 13:42:08 +0800 Subject: [PATCH] fix(transfer): suppress completion events without target items --- app/chain/transfer/settlement.py | 9 +- .../adr/0007-background-action-reliability.md | 4 + tests/test_transfer_completed_event.py | 97 +++++++++++++++++++ 3 files changed, 108 insertions(+), 2 deletions(-) create mode 100644 tests/test_transfer_completed_event.py diff --git a/app/chain/transfer/settlement.py b/app/chain/transfer/settlement.py index 6d8def70b..1c06ffab5 100644 --- a/app/chain/transfer/settlement.py +++ b/app/chain/transfer/settlement.py @@ -345,7 +345,7 @@ class TransferSettlementOwner(_TransferOwnerBase): self, task: TransferTask, transferinfo: TransferInfo, / ) -> Tuple[bool, str]: """ - 整理完成后处理 + 原子结算整理结果；成功但无目标项的静默跳过不发布完成事件。 """ # 状态 ret_status = True @@ -504,7 +504,12 @@ class TransferSettlementOwner(_TransferOwnerBase): # 转移成功 logger.info(f"{task.fileitem.name} 入库成功：{target_dir_path or ''}") - durable_event = self._durable_transfer_event(task, success=True) + # 静默跳过沿用成功终态，但没有目标文件可供完成事件的消费者处理。 + durable_event = ( + self._durable_transfer_event(task, success=True) + if transferinfo.target_item is not None + else None + ) topic = durable_event[0] if durable_event else None event_type = durable_event[1] if durable_event else None event_payload = self._transfer_result_payload(task, transferinfo) diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index 9c025b810..28b99c585 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -58,6 +58,10 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 `AudioTransferComplete`、`AudioTransferFailed`：整理步骤本身属于 E3， 但历史行提交后向事件消费者发布结果属于 E2，使用同一 JSON 快照与恢复 handler。 +宿主仅在整理成功且 `transferinfo.target_item` 非空时登记媒体、字幕或音频的完成事件。 +静默跳过即使返回 `success=True`，也不产生完成事件的 outbox intent 或广播；任务终态与历史 +仍按原成功路径结算。失败结果继续登记对应失败事件，无需目标项。 + ## 非 Event 后台机制映射 ### FastAPI BackgroundTasks diff --git a/tests/test_transfer_completed_event.py b/tests/test_transfer_completed_event.py new file mode 100644 index 000000000..b2f539177 --- /dev/null +++ b/tests/test_transfer_completed_event.py @@ -0,0 +1,97 @@ +"""整理完成事件只对应实际目标文件，静默跳过仍正常结算。""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from app.schemas.transfer import TransferInfo +from app.schemas.types import EventType +from tests.test_transfer_job_manager import ( + FakeMedia, + bind_terminal_checkpoint, + make_fileitem, + make_task, + make_transfer_chain, +) + + +@pytest.mark.parametrize( + ("extension", "completed_topic", "completed_event", "failed_topic", "failed_event"), + [ + ("mkv", "transfer.completed", EventType.TransferComplete, + "transfer.failed", EventType.TransferFailed), + ("srt", "transfer.subtitle.completed", EventType.SubtitleTransferComplete, + "transfer.subtitle.failed", EventType.SubtitleTransferFailed), + ("flac", "transfer.audio.completed", EventType.AudioTransferComplete, + "transfer.audio.failed", EventType.AudioTransferFailed), + ], +) +@pytest.mark.parametrize("outcome", ["skipped", "without_target", "completed", "failed"]) +def test_transfer_result_events_require_completed_target( + extension: str, + completed_topic: str, + completed_event: EventType, + failed_topic: str, + failed_event: EventType, + outcome: str, +) -> None: + """视频、字幕和音频跳过不发完成事件，正常完成和失败保留原事件与终态。""" + chain = make_transfer_chain() + chain.eventmanager = MagicMock() + chain.queue_failed_transfer_notification = MagicMock() + chain._TransferChain__mark_torrent_completed_if_done = MagicMock() + chain._record_scrape_target = MagicMock() + chain._send_metadata_scrape_event = MagicMock() + task = make_task(1) + task.fileitem = make_fileitem(f"/downloads/Test.Show.S01E01.{extension}") + task.mediainfo = FakeMedia() + assert chain._TransferChain__put_to_jobview(task) + transferinfo = TransferInfo( + success=outcome != "failed", + message="综艺非正片已全部跳过（1 个文件）" if outcome == "skipped" else None, + fileitem=task.fileitem, + target_item=( + make_fileitem(f"/library/Test.Show.S01E01.{extension}") + if outcome == "completed" + else None + ), + transfer_type="copy", + need_notify=False, + ) + bind_terminal_checkpoint(task, transferinfo) + with patch( + "app.chain.transfer.settlement.add_transfer_success", + return_value=SimpleNamespace(id=1), + ) as add_success, patch( + "app.chain.transfer.settlement.add_transfer_fail", + return_value=SimpleNamespace(id=1), + ) as add_fail, patch( + "app.chain.transfer.settlement.clear_transfer_failures", + ), patch( + "app.chain.transfer.settlement.record_transfer_failure", + ): + state, _message = chain._TransferChain__default_callback(task, transferinfo) + + assert state is (outcome != "failed") + chain.durable_event_writer.transfer_result.assert_called_once() + call = chain.durable_event_writer.transfer_result.call_args.kwargs + assert call["settlement"].outcome == ("failed" if outcome == "failed" else "succeeded") + assert task.terminal_settled is True + if outcome == "failed": + add_fail.assert_called_once() + add_success.assert_not_called() + assert call["topic"] == failed_topic + assert chain.eventmanager.send_event.call_args.args[0] is failed_event + else: + add_success.assert_called_once() + add_fail.assert_not_called() + chain.queue_failed_transfer_notification.assert_not_called() + if outcome in ("skipped", "without_target"): + assert call["topic"] is None + chain.eventmanager.send_event.assert_not_called() + else: + assert call["topic"] == completed_topic + event_type, payload = chain.eventmanager.send_event.call_args.args + assert event_type is completed_event + assert payload["transferinfo"].target_item == transferinfo.target_item