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

@@ -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

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和历史未知来源不能据此升级状态。

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()