nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-03 13:55:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(transfer): settle tasks with missing metadata safely (#7245)

Browse Source
This commit is contained in:
jxxghp
2026-10-30 14:34:07 +08:00
committed by GitHub
parent a9c992d9b9
commit 586a4f595e
9 changed files with 267 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/application/history/__init__.py
View File

@@ -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: 下载器

28
app/application/transfer/history.py
View File

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

19
app/chain/transfer/execution.py
View File

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

4
app/chain/transfer/settlement.py
View File

@@ -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, [])

2
docs/architecture-overview.md
View File

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

2
tests/fixtures/architecture/complexity-baseline.json vendored
View File

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

5
tests/fixtures/architecture/dependency-baseline.json vendored
View File

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

214
tests/test_transfer_execution_metadata.py Normal file
View File

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

26
tests/test_transfer_job_missing_meta.py
View File

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