From 586a4f595ee1e1f2bd963382bf49116694771ff3 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 30 Oct 2026 14:34:07 +0800 Subject: [PATCH] fix(transfer): settle tasks with missing metadata safely (#7245) --- app/application/history/__init__.py | 7 +- app/application/transfer/history.py | 28 +-- app/chain/transfer/execution.py | 19 +- app/chain/transfer/settlement.py | 4 +- docs/architecture-overview.md | 2 +- .../architecture/complexity-baseline.json | 2 +- .../architecture/dependency-baseline.json | 5 +- tests/test_transfer_execution_metadata.py | 214 ++++++++++++++++++ tests/test_transfer_job_missing_meta.py | 26 ++- 9 files changed, 267 insertions(+), 40 deletions(-) create mode 100644 tests/test_transfer_execution_metadata.py diff --git a/app/application/history/__init__.py b/app/application/history/__init__.py index 1be5c7bf8..6eddb1b8e 100644 --- a/app/application/history/__init__.py +++ b/app/application/history/__init__.py @@ -709,7 +709,7 @@ def add_transfer_success( def add_transfer_fail( fileitem: FileItem, mode: str, - meta: MetaBase, + meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, transferinfo: Optional[TransferInfo] = None, downloader: Optional[str] = None, @@ -722,10 +722,11 @@ def add_transfer_fail( 新增转移失败历史记录。 识别结果与整理结果齐备时按完整字段落库；缺任一项则走「未识别到媒体信息」分支， - 此时只有文件名解析出的元数据可用，不写目标路径。 + 此时仅保存已有文件名元数据，不写目标路径；缺少元数据也必须保留源文件与失败原因。 + :param fileitem: 源文件项 :param mode: 整理方式 - :param meta: 文件名识别结果 + :param meta: 文件名识别结果，未解析时为 None :param mediainfo: 媒体识别结果，未识别时为 None :param transferinfo: 整理结果，未进入整理时为 None :param downloader: 下载器 diff --git a/app/application/transfer/history.py b/app/application/transfer/history.py index ac834a3d1..356d585c1 100644 --- a/app/application/transfer/history.py +++ b/app/application/transfer/history.py @@ -31,17 +31,17 @@ def classification_fields( def history_title( - meta: MetaBase, + meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, ) -> Optional[str]: """ - 音乐文件优先记录曲目标题，其它媒体保持识别标题。 + 音乐文件优先记录曲目标题，其它媒体保持识别标题；均未识别时不推测标题。 """ if isinstance(meta, MetaMusic) and meta.title: return str(meta.title) if mediainfo and mediainfo.title: return str(mediainfo.title) - return str(meta.name) if meta.name else None + return str(meta.name) if meta and meta.name else None def history_source_path(fileitem: FileItem) -> str: @@ -118,7 +118,7 @@ def success_fields( def failure_fields( fileitem: FileItem, mode: str, - meta: MetaBase, + meta: Optional[MetaBase], mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None, transferinfo: Optional[TransferInfo] = None, downloader: Optional[str] = None, @@ -130,10 +130,12 @@ def failure_fields( 投影转移失败记录，不执行数据库写入。 识别结果与整理结果齐备时按完整字段落库；缺任一项则走「未识别到媒体信息」分支， - 此时只有文件名解析出的元数据可用，不写目标路径。 + 此时仅保存已有的文件名元数据，不写目标路径。元数据为空时仍保留源文件与失败原因， + 不推测年份或季集。 + :param fileitem: 源文件项 :param mode: 整理方式 - :param meta: 文件名识别结果 + :param meta: 文件名识别结果，未解析时为 None :param mediainfo: 媒体识别结果，未识别时为 None :param transferinfo: 整理结果，未进入整理时为 None :param downloader: 下载器 @@ -165,7 +167,7 @@ def failure_fields( type=(mediainfo.type.value if mediainfo.type else ""), **classification_fields(mediainfo), title=history_title(meta, mediainfo), - year=history_year(mediainfo.year or meta.year), + year=history_year(mediainfo.year or (meta.year if meta else None)), media_source=media_source, media_id=media_id, music_type=getattr(mediainfo, "music_type", None), @@ -175,8 +177,8 @@ def failure_fields( bit_depth=getattr(meta, "bit_depth", None), sample_rate=getattr(meta, "sample_rate", None), bitrate=getattr(meta, "bitrate", None), - seasons=meta.season, - episodes=meta.episode, + seasons=meta.season if meta else None, + episodes=meta.episode if meta else None, image=mediainfo.get_poster_image(), downloader=downloader, download_hash=download_hash, @@ -194,9 +196,9 @@ def failure_fields( else: media_source, media_id = resolve_media_identity(media=meta) history = dict( - type=meta.type.value if meta.type else None, + type=meta.type.value if meta and meta.type else None, title=history_title(meta), - year=history_year(meta.year), + year=history_year(meta.year if meta else None), media_source=media_source, media_id=media_id, music_type=MUSIC_ENTITY_RECORDING if isinstance(meta, MetaMusic) else None, @@ -209,8 +211,8 @@ def failure_fields( src_storage=fileitem.storage, src_fileitem=fileitem.model_dump(), mode=mode, - seasons=meta.season, - episodes=meta.episode, + seasons=meta.season if meta else None, + episodes=meta.episode if meta else None, downloader=downloader, download_hash=download_hash, status=False, diff --git a/app/chain/transfer/execution.py b/app/chain/transfer/execution.py index 621fc3a87..45139d2df 100644 --- a/app/chain/transfer/execution.py +++ b/app/chain/transfer/execution.py @@ -39,7 +39,6 @@ from app.chain.transfer.music import defer_music_recognition, refresh_music_retr from app.chain.transfer.records import apply_download_history_classification from app.domain.context import MediaInfo, MusicInfo from app.domain.media import is_music_media_source -from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic from app.runtime.log import logger from app.schemas.transfer import TransferInfo @@ -519,7 +518,9 @@ class TransferExecutionOwner(_TransferOwnerBase): callback: Optional[Callable[[TransferTask, TransferInfo], Tuple[bool, str]]] = None, ) -> Optional[Tuple[bool, str]]: """ - 处理整理任务 + 执行冻结计划，或在元数据完整时识别并规划新任务。 + + 未规划任务缺少文件名元数据时走可结算的规划拒绝，不猜测季集或执行文件操作。 """ try: if task.plan_checkpoint is not None: @@ -528,6 +529,14 @@ class TransferExecutionOwner(_TransferOwnerBase): self._TransferChain__handle_planned_transfer(task, callback), ) refresh_music_retry_context(self, task) + if task.meta is None: + transferinfo = self._TransferChain__checkpoint_planning_rejection( + task, + "整理规划输入缺少文件名元数据，请重新预览并确认季集信息后重新整理", + ) + return callback(task, transferinfo) if callback else ( + transferinfo.success, transferinfo.message or "" + ) transferhis = self.transfer_history_repository # 显式标注联合：下面既会赋回音乐识别结果（MusicInfo），也会赋回影视识别 # 结果（MediaInfo），不标注时会被推断成其中一种，另一种就成了假错误 @@ -637,11 +646,9 @@ class TransferExecutionOwner(_TransferOwnerBase): ): # 第零季表示特别篇，不能按空值回退为第一季。 season_num = cast(MediaInfo, mediainfo).season - if season_num is None and cast(MetaBase, task.meta).season_seq: - if cast(MetaBase, task.meta).season_seq.isdigit(): - season_num = int(cast(MetaBase, task.meta).season_seq) if season_num is None: - season_num = 1 + season_seq = task.meta.season_seq + season_num = int(season_seq) if season_seq.isdigit() else 1 task.episodes_info = TmdbChain().tmdb_episodes( tmdbid=mediainfo.tmdb_id, season=season_num, diff --git a/app/chain/transfer/settlement.py b/app/chain/transfer/settlement.py index 37892737e..c55ea07e1 100644 --- a/app/chain/transfer/settlement.py +++ b/app/chain/transfer/settlement.py @@ -606,8 +606,8 @@ class TransferSettlementOwner(_TransferOwnerBase): self._record_scrape_target(task, transferinfo) self.notify_successful_transfer(task, transferinfo) - # 全部整理完成且有成功的任务时，发送消息和事件 - if self.jobview.is_finished(task): + # 成功汇总需要季集元数据；缺少元数据的规划拒绝只走失败反馈。 + if task.meta is not None and self.jobview.is_finished(task): with job_lock: transferinfo.file_list_new = list(dict.fromkeys( self._success_target_files.pop(job_id, []) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 743d1cc19..d21b0fb60 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1142 | -| 内部导入边 | 9,872 | +| 内部导入边 | 9,871 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 4aee6fbb0..067fc9a65 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -57,7 +57,7 @@ "app/chain/subscribe/query.py:SubscribeQueryOwner.subscribe_files_info": 24, "app/chain/subscribe/refresh.py:SubscribeRefreshOwner.backfill_existing_episodes": 16, "app/chain/subscribe/search.py:_SubscribeSearchQueueCoordinator._execute_inline_search": 16, - "app/chain/transfer/execution.py:TransferExecutionOwner._TransferChain__perform_transfer": 31, + "app/chain/transfer/execution.py:TransferExecutionOwner._TransferChain__perform_transfer": 30, "app/chain/transfer/filter.py:FileFilterMixin._restore_music_download_context": 24, "app/chain/transfer/music.py:_release_groups": 16, "app/chain/transfer/music.py:prepare_music_batch_context": 18, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 4bbac8f27..56e1de248 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9872, - "edge_sha256": "579e3f11be9b9ee458fb9913ffeec67db47b98d431141d9664201135aa87b83d", + "edge_count": 9871, + "edge_sha256": "3ef0223109f42e8ed72880732b7fc09839d18d29d91ed23ba731eeb40218ddc6", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -5936,7 +5936,6 @@ "app.chain.transfer.execution -> app.domain.context", "app.chain.transfer.execution -> app.domain.media", "app.chain.transfer.execution -> app.domain.meta", - "app.chain.transfer.execution -> app.domain.meta.metabase", "app.chain.transfer.execution -> app.domain.meta.metamusic", "app.chain.transfer.execution -> app.runtime", "app.chain.transfer.execution -> app.runtime.log", diff --git a/tests/test_transfer_execution_metadata.py b/tests/test_transfer_execution_metadata.py new file mode 100644 index 000000000..0b5038320 --- /dev/null +++ b/tests/test_transfer_execution_metadata.py @@ -0,0 +1,214 @@ +""" +整理执行的空元数据拒绝、季集选择与失败历史回归。 +""" + +from pathlib import Path +from types import SimpleNamespace +from typing import Optional +from unittest.mock import Mock + +import pytest + +from app.application.transfer.execution import TransferExecutionState +from app.application.transfer.jobs import JobManager +from app.application.transfer.workflow import TransferTask +from app.db.adapters.chain import TransactionalChainDurableEventWriter +from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository +from app.db.adapters.transfer.execution import TransactionalTransferExecutionRepository +from app.db.models.transferexecutionstep import TransferExecutionStep +from app.db.models.transferhistory import TransferHistory +from app.db.models.transferpending import TransferPending +from app.db.models.transfersettlementreceipt import TransferSettlementReceipt +from app.db.session import SessionFactory +from app.domain.context import MediaInfo +from app.domain.metainfo import MetaInfoPath +from app.schemas.file import FileItem +from app.schemas.system import TransferDirectoryConf +from app.schemas.transfer import TransferInfo +from app.schemas.types import MediaSource, MediaType +from tests.test_transfer_history_write_path import _repository +from tests.test_transfer_planning_checkpoint import _chain, _planned_admission + + +def _task(*, preview: bool = False, has_media: bool = True) -> TransferTask: + """ + 构造有 TMDB 身份、缺少文件名元数据的第二季单集任务。 + """ + name = "Tantei.wa.mou.Shindeiru.S02E04.2026.1080p.Baha.WEB-DL.x264.AAC-ADWeb.mp4" + return TransferTask( + fileitem=FileItem( + storage="local", path=f"/downloads/{name}", type="file", name=name, + basename=Path(name).stem, extension="mp4", size=1024, + ), + mediainfo=MediaInfo( + type=MediaType.TV, title="侦探已死", tmdb_id=117061, + media_source=MediaSource.TMDB, media_id="117061", + ) if has_media else None, + target_directory=TransferDirectoryConf(library_storage="local", library_path="/library"), + transfer_type="copy", + preview=preview, + ) + + +def _execution_chain(monkeypatch, task: TransferTask): + """ + 隔离在线识别与文件执行，保留真实规划拒绝和持久步骤编排。 + """ + chain = _chain() + chain.jobview = JobManager() + chain.jobview.add_task(task) + chain.runtime_config = SimpleNamespace(scrape_follow_tmdb=True) + chain._success_target_files = {} + task.bind_admission_task_id("task-execution-metadata") + chain._finish_scrape_batch_task = Mock() + chain._transfer_admissions.checkpoint_plan.side_effect = ( + lambda **kwargs: _planned_admission(task, kwargs["checkpoint"]) + ) + media = Mock() + media.recognize_by_meta.return_value = MediaInfo(type=MediaType.TV, tmdb_id=117061) + media.supplement_tmdb_info.side_effect = lambda info, _meta: info + tmdb = Mock() + tmdb.tmdb_episodes.return_value = [] + monkeypatch.setattr("app.chain.transfer.execution.MediaChain", lambda: media) + monkeypatch.setattr("app.chain.transfer.execution.TmdbChain", lambda: tmdb) + return chain, media, tmdb + + +@pytest.mark.parametrize("preview", [False, True]) +@pytest.mark.parametrize("has_media", [False, True]) +@pytest.mark.parametrize("with_callback", [False, True]) +def test_missing_meta_is_rejected_before_recognition_and_file_operations( + monkeypatch, preview: bool, has_media: bool, with_callback: bool, +) -> None: + """ + 空元数据须返回可操作失败；正式任务提交拒绝检查点，预览保持零写入。 + """ + task = _task(preview=preview, has_media=has_media) + chain, media, tmdb = _execution_chain(monkeypatch, task) + callback = Mock(side_effect=lambda _task, info: (info.success, info.message)) if with_callback else None + + state, message = chain._TransferChain__handle_transfer(task, callback) + + assert state is False + assert "缺少文件名元数据" in message + assert "重新预览" in message + assert "季集" in message + assert task.meta is None + media.recognize_by_meta.assert_not_called() + media.supplement_tmdb_info.assert_not_called() + tmdb.tmdb_episodes.assert_not_called() + chain.plan_transfer.assert_not_called() + chain.execute_transfer_plan.assert_not_called() + if with_callback: + callback.assert_called_once() + if preview: + assert task.plan_checkpoint is None + assert task.execution_checkpoint is None + chain._transfer_admissions.checkpoint_plan.assert_not_called() + else: + assert task.plan_checkpoint.rejection_error == message + assert task.plan_checkpoint.items == () + assert task.execution_checkpoint.payload["outcome"] == "failed" + assert [step.kind for step in chain._transfer_executions.steps.values()] == ["reject"] + + +@pytest.mark.parametrize( + ("filename", "media_season", "expected_season"), + [ + ("Tantei.wa.mou.Shindeiru.S02E04.2026.mp4", None, 2), + ("Test.Show.S00E04.mp4", None, 0), + ("Test.Show.S02E04.mp4", 0, 0), + ("Test.Show.S02E04.mp4", 3, 3), + ("Test.Show.E04.mp4", None, 1), + ], +) +def test_tv_episode_lookup_preserves_resolved_season( + monkeypatch, filename: str, media_season: Optional[int], expected_season: int, +) -> None: + """ + 剧集详情优先使用媒体季，其次文件名季；保留特别篇和有效元数据的第一季约定。 + """ + task = _task(preview=True) + task.meta = MetaInfoPath(Path("/downloads") / filename) + task.mediainfo.season = media_season + chain, _media, tmdb = _execution_chain(monkeypatch, task) + chain._plan_checkpoint_and_execute = Mock(return_value=TransferInfo(success=True)) + + assert chain._TransferChain__handle_transfer(task)[0] is True + + tmdb.tmdb_episodes.assert_called_once_with( + tmdbid=117061, season=expected_season, episode_group=None, + ) + assert task.meta.begin_episode == 4 + + +@pytest.mark.parametrize("restored", [False, True]) +@pytest.mark.parametrize("has_media", [False, True]) +def test_missing_meta_failure_is_atomically_written_to_history( + db, tmp_path, monkeypatch, restored: bool, has_media: bool, +) -> None: + """ + accepted 恢复与普通执行均写入失败历史并结束任务，源文件不变且不进入人工复核。 + """ + db.watermark(TransferExecutionStep, TransferHistory, TransferPending, TransferSettlementReceipt) + task = _task(has_media=has_media) + source = tmp_path / task.fileitem.name + source.write_bytes(b"unchanged episode") + task.fileitem.path = source.as_posix() + chain, _media, tmdb = _execution_chain(monkeypatch, task) + chain._transfer_admissions = TransactionalTransferAdmissionRepository(SessionFactory) + chain._transfer_executions = TransactionalTransferExecutionRepository(SessionFactory) + chain.durable_event_writer = TransactionalChainDurableEventWriter(SessionFactory) + chain.transfer_history_repository = _repository() + chain._is_overwrite_declined = Mock(return_value=False) + chain._durable_transfer_event = Mock(return_value=None) + chain.queue_failed_transfer_notification = Mock() + chain._TransferChain__mark_torrent_completed_if_done = Mock() + chain._maybe_take_over_failed_transfer = Mock() + chain.notify_successful_transfer = Mock() + chain._send_metadata_scrape_event = Mock() + chain._TransferChain__admit_transfer(task) + if restored: + chain.jobview.remove_job(task) + chain._transfer_admissions.release_claim( + task_id=task.admission_task_id, lease_token=task.lease_token, + ) + admission = chain._transfer_admissions.claim_task( + task_id=task.admission_task_id, owner_id=chain._worker_owner_id, lease_seconds=120, + ) + queued = [] + chain.put_to_queue = Mock(side_effect=lambda replay: queued.append(replay) or True) + assert chain._TransferChain__queue_accepted_replay(task.fileitem, admission) + task = queued[0] + assert task.meta is None + assert task.planning_context_restored is has_media + assert chain.jobview.add_task(task) + + state, message = chain._TransferChain__handle_transfer( + task, chain._TransferChain__default_callback, + ) + + history = chain.transfer_history_repository.get_by_src(source.as_posix(), "local") + snapshot = chain._transfer_executions.get_snapshot(task_id=task.admission_task_id) + assert state is False + assert history is not None + assert history.status is False + assert history.errmsg == message + assert "缺少文件名元数据" in message + assert history.failure_stage == "planning" + assert "重新生成计划" in history.recovery_action + assert history.transfer_task_id == task.admission_task_id + assert history.seasons is None + assert history.episodes is None + assert history.year is None + assert history.media_id == ("117061" if has_media else None) + assert task.terminal_settled + assert snapshot.state is TransferExecutionState.FAILED + assert snapshot.terminal_history_id == history.id + assert [step.kind for step in snapshot.steps] == ["reject"] + assert source.read_bytes() == b"unchanged episode" + assert chain.jobview.total() == 0 + tmdb.tmdb_episodes.assert_not_called() + chain.queue_failed_transfer_notification.assert_called_once() + chain.notify_successful_transfer.assert_not_called() + chain._send_metadata_scrape_event.assert_not_called() diff --git a/tests/test_transfer_job_missing_meta.py b/tests/test_transfer_job_missing_meta.py index 2f20f123e..55da33cb1 100644 --- a/tests/test_transfer_job_missing_meta.py +++ b/tests/test_transfer_job_missing_meta.py @@ -161,34 +161,38 @@ def test_migrated_tasks_without_meta_keep_unfinished_siblings() -> None: assert not manager._task_state_changed_at -def test_successful_recognition_without_meta_does_not_skip_transfer() -> None: +def test_missing_meta_rejection_preserves_media_and_cleans_job() -> None: """ - 识别得到媒体的无元数据任务继续规划执行，不得被迁移失败误报为重复。 + 已识别媒体不能替代季集元数据，缺失时应明确拒绝并清理作业而非误报重复。 """ - task = _task() + task = _task(has_media=True) task.preview = True task.target_directory = TransferDirectoryConf(library_path="/library") task.mark_planning_context_restored() chain = _chain() chain.jobview = JobManager() chain.runtime_config = SimpleNamespace(scrape_follow_tmdb=True) - chain._recognize_task_meta = Mock(return_value=_task(has_media=True).mediainfo) - chain._plan_checkpoint_and_execute = Mock( - return_value=TransferInfo(success=True, message="完成", fileitem=task.fileitem), - ) + media = task.mediainfo + chain._recognize_task_meta = Mock() + chain._plan_checkpoint_and_execute = Mock() chain._finish_scrape_batch_task = Mock() assert chain.jobview.add_task(task) def settle(recovered: TransferTask, info: TransferInfo) -> tuple[bool, str]: """ - 回读真实迁移后的任务并结束作业，避免保留未完成测试状态。 + 保留已有媒体身份与明确失败原因，并结束无元数据作业。 """ assert recovered.meta is None - chain.jobview.finish_task(recovered) + assert recovered.mediainfo is media + assert info.success is False + chain.jobview.fail_task(recovered) return info.success, info.message - assert chain._TransferChain__perform_transfer(task, settle) == (True, "完成") - chain._plan_checkpoint_and_execute.assert_called_once() + state, message = chain._TransferChain__perform_transfer(task, settle) + assert state is False + assert "缺少文件名元数据" in message + chain._recognize_task_meta.assert_not_called() + chain._plan_checkpoint_and_execute.assert_not_called() assert chain.jobview.total() == 0