From 8a08847005a85e21dc3efa0b6c279b95d5743843 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 29 Oct 2026 17:34:22 +0800 Subject: [PATCH] fix(transfer): handle replay jobs without parsed metadata (#7237) * fix(transfer): support replay jobs without parsed metadata * refactor(transfer): remove obsolete metadata narrowing helper --- app/application/transfer/jobs.py | 99 +++++---- app/application/transfer/projection.py | 17 -- tests/test_transfer_job_missing_meta.py | 284 ++++++++++++++++++++++++ 3 files changed, 339 insertions(+), 61 deletions(-) create mode 100644 tests/test_transfer_job_missing_meta.py diff --git a/app/application/transfer/jobs.py b/app/application/transfer/jobs.py index dd98c6fa5..fdd38be1b 100644 --- a/app/application/transfer/jobs.py +++ b/app/application/transfer/jobs.py @@ -13,9 +13,6 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Protocol, Tuple, Union, from app.application.transfer.projection import ( domain_to_dict as _domain_to_dict, ) -from app.application.transfer.projection import ( - transfer_task_meta as _transfer_task_meta, -) from app.domain.context import MediaInfo, MusicInfo from app.domain.media import normalize_music_type from app.domain.meta.metabase import MetaBase @@ -200,8 +197,9 @@ job_lock = threading.Lock() class JobManager: """ - 作业管理器 - task任务负责一个文件的整理，job作业负责一个媒体的整理 + 按媒体或解析名称聚合文件任务，缺少识别上下文时按源文件隔离。 + + 仅维护进程内视图与状态，不重建或改写持久恢复使用的元数据快照。 """ # 整理中的作业 @@ -224,15 +222,19 @@ class JobManager: self._task_state_changed_at = {} self._active_executions = set() - @staticmethod def __get_meta_id( - meta: Optional[MetaBase] = None, - season: Optional[int] = None, + self, + task: TransferTask, ) -> JobId: """ - 获取元数据ID + 按解析名称和季分组；尚无元数据的恢复任务按源文件隔离。 + + :param task: 可缺少元数据的整理任务 + :return: 元数据作业身份，或带存储和路径的源文件作业身份 """ - return cast(MetaBase, meta).name, season + if task.meta is not None: + return task.meta.name, task.meta.begin_season + return "file", self.__get_file_key(task.fileitem) @staticmethod def __get_media_id(media: Optional[Union[MediaInfo, MusicInfo]] = None, @@ -291,15 +293,14 @@ class JobManager: def __get_id(self, task: TransferTask) -> JobId: """ - 获取作业ID + 获取作业身份；无元数据时不推测季号，未识别媒体时按源文件分组。 """ - resolved_task = task - meta = _transfer_task_meta(resolved_task) - if resolved_task.mediainfo: + if task.mediainfo: return self.__get_media_id( - media=resolved_task.mediainfo, season=meta.begin_season + media=task.mediainfo, + season=task.meta.begin_season if task.meta else None, ) - return self.__get_meta_id(meta=meta, season=meta.begin_season) + return self.__get_meta_id(task) def get_job_id(self, task: TransferTask) -> JobId: """ @@ -319,7 +320,12 @@ class JobManager: return _SchemaMusicInfo(**_domain_to_dict(mediainfo)) return _SchemaMediaInfo(**_domain_to_dict(mediainfo)) else: - meta = _transfer_task_meta(task) + meta = task.meta + if meta is None: + return _SchemaMediaInfo( + title=task.fileitem.name, + title_year=task.fileitem.name, + ) if isinstance(meta, MetaMusic): # 未识别的音乐按已解析元数据兜底展示；音乐年份为 int， # 不能复用 MediaInfo（year 为 str），否则触发 pydantic 校验异常 @@ -342,20 +348,23 @@ class JobManager: ) @staticmethod - def __get_meta(task: TransferTask) -> Union[_SchemaMetaInfo, _SchemaMusicMeta]: + def __get_meta(task: TransferTask) -> Optional[Union[_SchemaMetaInfo, _SchemaMusicMeta]]: """ - 获取元数据 + 投影已解析元数据，恢复任务未携带时保留空值。 """ + if task.meta is None: + return None if isinstance(task.meta, MetaMusic): return _SchemaMusicMeta(**task.meta.to_dict()) - return _SchemaMetaInfo(**_domain_to_dict(_transfer_task_meta(task))) + return _SchemaMetaInfo(**_domain_to_dict(task.meta)) def add_task(self, task: TransferTask, state: Optional[str] = "waiting") -> bool: """ - 添加整理任务，自动分组到对应的作业中 + 添加整理任务，缺少元数据时仍按媒体或源文件身份登记。 + :return: True表示任务已添加，False表示任务无效或已存在（重复） """ - if not all([task, task.meta, task.fileitem]): + if not all([task, task.fileitem]): return False file_key = self.__get_file_key(task.fileitem) if not file_key: @@ -369,7 +378,7 @@ class JobManager: if __mediaid__ not in self._job_view: self._job_view[__mediaid__] = TransferJob( media=self.__get_media(task), - season=_transfer_task_meta(task).begin_season, + season=task.meta.begin_season if task.meta else None, tasks=[ TransferJobTask( fileitem=task.fileitem, @@ -392,29 +401,30 @@ class JobManager: ) ) self._task_state_changed_at[file_key] = monotonic() + episodes = task.meta.episode_list if task.meta else [] if self._season_episodes.get(__mediaid__): self._season_episodes[__mediaid__].extend( - _transfer_task_meta(task).episode_list + episodes ) self._season_episodes[__mediaid__] = list( set(self._season_episodes[__mediaid__]) ) else: - self._season_episodes[__mediaid__] = _transfer_task_meta(task).episode_list + self._season_episodes[__mediaid__] = episodes return True def migrate_task(self, task: TransferTask) -> bool: """ - 将任务从 meta 作业迁移到 media 作业 + 将元数据或源文件作业迁入媒体作业，保留原任务状态与执行标记。 """ curr_task, source_job_id = self.__remove_task_with_job_id( task.fileitem, preserve_execution=True ) if not self.add_task(task, state=curr_task.state if curr_task else "waiting"): return False - if curr_task and task.mediainfo: - meta = _transfer_task_meta(task) - metaid = self.__get_meta_id(meta=meta, season=meta.begin_season) + # 无元数据的源文件作业只含当前任务，迁走即为空，无需保留跨作业关联。 + if curr_task and task.mediainfo and task.meta is not None: + metaid = self.__get_meta_id(task) mediaid = self.__get_id(task) if source_job_id == metaid and mediaid != metaid: with job_lock: @@ -594,7 +604,7 @@ class JobManager: if __mediaid__ in self._season_episodes: self._season_episodes[__mediaid__] = list( set(self._season_episodes[__mediaid__]) - - set(_transfer_task_meta(task).episode_list) + - set(task.meta.episode_list if task.meta else []) ) def fail_unfinished_task(self, task: TransferTask) -> None: @@ -617,7 +627,7 @@ class JobManager: if mediaid in self._season_episodes: self._season_episodes[mediaid] = list( set(self._season_episodes[mediaid]) - - set(_transfer_task_meta(task).episode_list) + - set(getattr(job_task.meta, "episode_list", None) or []) ) return @@ -658,6 +668,7 @@ class JobManager: self._active_executions.discard(file_key) if not _job_tasks(job): self._job_view.pop(mediaid) + self._season_episodes.pop(mediaid, None) if mediaid in self._season_episodes: episodes = getattr(task.meta, "episode_list", None) or [] self._season_episodes[mediaid] = list( @@ -684,10 +695,10 @@ class JobManager: 尝试移除任务对应的作业（严格检查未完成作业，线程安全） """ with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __metaid__ = self.__get_meta_id(task) __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season + media=task.mediainfo, + season=task.meta.begin_season if task.meta else None, ) related_media_ids = set(self._meta_to_media_ids.get(__metaid__, set())) @@ -710,10 +721,10 @@ class JobManager: 检查任务对应的作业是否整理完成（不管成功还是失败） """ with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __metaid__ = self.__get_meta_id(task) __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season + media=task.mediainfo, + season=task.meta.begin_season if task.meta else None, ) if __metaid__ in self._job_view: meta_done = all( @@ -736,10 +747,10 @@ class JobManager: 检查任务对应的作业是否已完成且有成功的记录 """ with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __metaid__ = self.__get_meta_id(task) __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season + media=task.mediainfo, + season=task.meta.begin_season if task.meta else None, ) if __metaid__ in self._job_view: meta_finished = all( @@ -762,10 +773,10 @@ class JobManager: 检查任务对应的作业是否全部成功 """ with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __metaid__ = self.__get_meta_id(task) __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season + media=task.mediainfo, + season=task.meta.begin_season if task.meta else None, ) if __metaid__ in self._job_view: meta_success = all( @@ -837,7 +848,7 @@ class JobManager: if __mediaid__ in self._job_view: return True - __metaid__ = self.__get_meta_id(meta=meta, season=season) + __metaid__ = meta.name, season return ( __metaid__ in self._job_view and len(_job_tasks(self._job_view[__metaid__])) > 0 diff --git a/app/application/transfer/projection.py b/app/application/transfer/projection.py index e6f61d165..7a1728f46 100644 --- a/app/application/transfer/projection.py +++ b/app/application/transfer/projection.py @@ -27,16 +27,6 @@ class _DictionarySerializable(Protocol): """ -class _TransferTaskMetaSource(Protocol): - """ - 描述整理任务提供已解析领域元数据的最小形状。 - - :ivar meta: 整理任务中已解析的发布名元数据；未完成识别时为 None - """ - - meta: MetaBase | None - - def domain_to_dict(value: object) -> dict[str, Any]: """ 按领域对象既有 ``to_dict`` 合同生成字典投影。 @@ -44,13 +34,6 @@ def domain_to_dict(value: object) -> dict[str, Any]: return cast(_DictionarySerializable, value).to_dict() -def transfer_task_meta(task: _TransferTaskMetaSource) -> MetaBase: - """ - 声明进入作业与规划边界的整理任务已完成元数据解析。 - """ - return cast(MetaBase, task.meta) - - def _music_preview_status(meta: MetaMusic, info: Optional[MusicInfo], diagnostic: dict[str, Any], selected: bool) -> str: """ 只把实际来源核验称为在线确认；标签中的ID和历史未知来源不能据此升级状态。 diff --git a/tests/test_transfer_job_missing_meta.py b/tests/test_transfer_job_missing_meta.py new file mode 100644 index 000000000..2f20f123e --- /dev/null +++ b/tests/test_transfer_job_missing_meta.py @@ -0,0 +1,284 @@ +""" +缺少解析元数据的整理任务在作业视图与恢复入口中的回归。 +""" + +import queue +from dataclasses import replace +from types import SimpleNamespace +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.models import TransferTask +from app.domain.context import MediaInfo +from app.domain.meta.metavideo import MetaVideo +from app.schemas.file import FileItem +from app.schemas.system import TransferDirectoryConf +from app.schemas.transfer import TransferInfo +from app.schemas.types import MediaType +from tests.test_transfer_pending_replay import _admission, _build_chain +from tests.test_transfer_planning_checkpoint import _chain, _checkpoint + + +def _task(*, has_media: bool = False, path: str = "/downloads/Movie.mkv") -> TransferTask: + """ + 构造未解析文件名、可选已识别媒体的任务。 + """ + media = None + if has_media: + media = MediaInfo() + media.title = "Movie" + media.type = MediaType.MOVIE + media.tmdb_id = 123 + return TransferTask( + fileitem=FileItem( + storage="local", path=path, type="file", name="Movie.mkv", size=1024, + ), + mediainfo=media, + ) + + +@pytest.mark.parametrize("has_media", [False, True]) +def test_unregistered_task_without_meta_can_be_queried_and_cleaned(has_media: bool) -> None: + """ + 规划前失败或恢复尚未登记时，查询和重复清理不能再抛元数据空值异常。 + """ + manager = JobManager() + task = _task(has_media=has_media) + + manager.get_job_id(task) + manager.running_task(task) + manager.finish_task(task) + manager.fail_task(task) + manager.fail_unfinished_task(task) + assert manager.is_done(task) + assert manager.is_finished(task) + assert manager.is_success(task) + manager.try_remove_job(task) + manager.try_remove_job(task) + assert manager.remove_job(task) is None + assert manager.total() == 0 + + +@pytest.mark.parametrize("has_media", [False, True]) +@pytest.mark.parametrize("terminal", ["finish_task", "fail_task", "fail_unfinished_task"]) +def test_task_without_meta_is_registered_and_settled(has_media: bool, terminal: str) -> None: + """ + 无元数据任务仍应登记、去重并按真实结果进入终态，不得误报重复或残留视图。 + """ + manager = JobManager() + task = _task(has_media=has_media) + assert manager.add_task(task) + assert not manager.add_task(task.model_copy(deep=True)) + job = manager.list_jobs()[0] + assert job.media.title == ("Movie" if has_media else "Movie.mkv") + assert job.season is None + assert job.tasks[0].meta is None + assert not manager.is_done(task) + assert not manager.is_finished(task) + assert not manager.is_success(task) + + manager.start_execution(task) + manager.running_task(task) + assert job.tasks[0].state == "running" + manager.try_remove_job(task) + assert manager.total() == 1 + getattr(manager, terminal)(task) + assert manager.is_done(task) + assert manager.is_success(task) is (terminal == "finish_task") + if has_media: + assert manager.is_finished(task) is (terminal == "finish_task") + assert manager.season_episodes(task.mediainfo) == [] + manager.finish_execution(task) + manager.try_remove_job(task) + assert manager.total() == 0 + + +def test_unrecognized_tasks_without_meta_are_isolated_by_source() -> None: + """ + 无识别身份时按存储和路径隔离，同名文件的结束清理不能影响其它源任务。 + """ + manager = JobManager() + task = _task() + sibling = _task(path="/other/Movie.mkv") + remote = task.model_copy(deep=True) + remote.fileitem.storage = "alist" + for item in (task, sibling, remote): + assert manager.add_task(item) + assert len({manager.get_job_id(item) for item in (task, sibling, remote)}) == 3 + + manager.finish_task(task) + manager.try_remove_job(task) + + assert manager.total() == 2 + assert not manager.is_done(sibling) + assert not manager.is_done(remote) + + +def test_task_without_meta_can_migrate_after_recognition() -> None: + """ + 识别成功后从源文件作业迁入媒体作业，保留运行状态且正常结束清理。 + """ + manager = JobManager() + task = _task() + assert manager.add_task(task) + manager.running_task(task) + manager.start_execution(task) + original_id = manager.get_job_id(task) + task.mediainfo = _task(has_media=True).mediainfo + + assert manager.migrate_task(task) + assert manager.get_job_id(task) != original_id + assert manager.total() == 1 + assert manager.list_jobs()[0].tasks[0].state == "running" + manager.finish_task(task) + manager.finish_execution(task) + manager.try_remove_job(task) + assert manager.total() == 0 + + +def test_migrated_tasks_without_meta_keep_unfinished_siblings() -> None: + """ + 多个未知源任务识别为同一媒体后，保留未完成成员且不残留源文件关联。 + """ + manager = JobManager() + tasks = [_task(), _task(path="/other/Movie.mkv")] + for task in tasks: + assert manager.add_task(task) + task.mediainfo = _task(has_media=True).mediainfo + assert manager.migrate_task(task) + manager.finish_task(tasks[0]) + manager.try_remove_job(tasks[0]) + assert manager.total() == 2 + assert not manager.is_done(tasks[1]) + manager.fail_task(tasks[1]) + manager.try_remove_job(tasks[1]) + assert manager.total() == 0 + assert not manager._meta_to_media_ids + assert not manager._season_episodes + assert not manager._task_state_changed_at + + +def test_successful_recognition_without_meta_does_not_skip_transfer() -> None: + """ + 识别得到媒体的无元数据任务继续规划执行，不得被迁移失败误报为重复。 + """ + task = _task() + 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), + ) + 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) + return info.success, info.message + + assert chain._TransferChain__perform_transfer(task, settle) == (True, "完成") + chain._plan_checkpoint_and_execute.assert_called_once() + assert chain.jobview.total() == 0 + + +@pytest.mark.parametrize("season", [0, 1]) +def test_parsed_job_lookup_keeps_explicit_season(season: int) -> None: + """ + 无元数据兜底不改变已解析任务的按名称查找与特别篇季号。 + """ + manager = JobManager() + task = _task() + task.meta = MetaVideo(f"Show.S{season:02d}E01.mkv") + assert manager.add_task(task) + assert manager.has_tasks(task.meta, season=season) + assert not manager.has_tasks(task.meta, season=season + 1) + manager.fail_task(task) + manager.try_remove_job(task) + assert not manager.has_tasks(task.meta, season=season) + + +@pytest.mark.parametrize("planned", [False, True]) +def test_replay_without_meta_enters_real_jobview(planned: bool) -> None: + """ + accepted 媒体快照和无元数据的拒绝计划均可经真实作业登记恢复入队。 + """ + task = _task(has_media=not planned) + admission = _admission(task.fileitem.path) + planning_input = replace( + admission.planning_input, + mediainfo=task.mediainfo.to_dict() if task.mediainfo else None, + ) + checkpoint = replace( + _checkpoint(), planning_input=planning_input, items=(), + resolved_meta=None, resolved_mediainfo=None, + rejection_error="未识别到媒体信息", + ) if planned else None + admission = replace(admission, planning_input=planning_input, checkpoint=checkpoint) + chain = _build_chain(Mock()) + chain._queue = queue.Queue() + chain.jobview = JobManager() + chain._register_scrape_batch_task = Mock() + chain._TransferChain__register_claimed_admission(admission) + + enqueue = ( + chain._TransferChain__queue_planned_replay + if planned else chain._TransferChain__queue_accepted_replay + ) + assert enqueue(task.fileitem, admission) + recovered = chain._queue.get_nowait().task + assert recovered.meta is None + assert chain.jobview.total() == 1 + assert recovered.planning_context_restored + chain._transfer_admissions.release_claim.assert_not_called() + chain._finish_queue_item(recovered) + + +def test_rejected_plan_without_meta_preserves_original_failure_on_replay() -> None: + """ + 拒绝计划回放保留业务失败原因，作业清理不能掩盖结果或派生人工复核。 + """ + task = _task() + task.bind_admission_task_id("task-missing-meta") + chain = _chain() + chain.jobview = JobManager() + chain._finish_scrape_batch_task = Mock() + planning_input = replace( + _checkpoint().planning_input, meta=None, mediainfo=None, + ) + checkpoint = replace( + _checkpoint(), planning_input=planning_input, items=(), + resolved_meta=None, resolved_mediainfo=None, + rejection_error="未识别到媒体信息", + ) + task.bind_planning_input(planning_input) + task.bind_plan_checkpoint(checkpoint) + chain._TransferChain__claim_task_for_execution(task) + assert chain.jobview.add_task(task) + + def settle(recovered: TransferTask, info: TransferInfo) -> tuple[bool, str]: + """ + 在真实执行拒绝步骤后标记失败，保留原始错误供调用方结算。 + """ + assert recovered is task + assert info.success is False + chain.jobview.fail_task(recovered) + return info.success, info.message + + assert chain._TransferChain__handle_transfer(task, settle) == (False, "未识别到媒体信息") + assert task.meta is None + assert task.execution_checkpoint.payload["outcome"] == "failed" + assert [step.kind for step in chain._transfer_executions.steps.values()] == ["reject"] + assert chain.jobview.total() == 0 + assert chain._transfer_executions.state is TransferExecutionState.SETTLING + chain.execute_transfer_plan.assert_not_called()