mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 04:47:03 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
750 lines
32 KiB
Python
750 lines
32 KiB
Python
"""
|
||
整理请求的候选路径解析与批次归属规划。
|
||
"""
|
||
|
||
from copy import deepcopy
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
|
||
|
||
from app.application.formatting import FormatParser
|
||
from app.application.history import (
|
||
DownloadHistoryQueryPort,
|
||
DownloadHistorySnapshot,
|
||
)
|
||
from app.application.transfer.projection import music_transfer_preview
|
||
from app.application.transfer.workflow import TransferTask
|
||
from app.chain.media import MediaChain
|
||
from app.chain.storage import StorageChain
|
||
from app.chain.transfer.music import append_cue_companions
|
||
from app.domain.context import MediaInfo, MusicInfo
|
||
from app.domain.meta.metabase import MetaBase
|
||
from app.domain.metainfo import MetaInfoPath
|
||
from app.schemas.exception import OperationInterrupted
|
||
from app.schemas.file import FileItem
|
||
from app.schemas.transfer import TransferInfo
|
||
from app.schemas.types import (
|
||
MUSIC_ENTITY_RECORDING,
|
||
MediaSource,
|
||
MediaType,
|
||
)
|
||
|
||
|
||
def preview_media_title(
|
||
mediainfo: Optional[Union[MediaInfo, MusicInfo]],
|
||
) -> Optional[str]:
|
||
"""
|
||
音乐预览以专辑为批次标题，影视继续使用原有标题。
|
||
"""
|
||
if isinstance(mediainfo, MusicInfo) and mediainfo.album:
|
||
return (
|
||
f"{mediainfo.album} ({mediainfo.year})"
|
||
if mediainfo.year
|
||
else mediainfo.album
|
||
)
|
||
return mediainfo.title_year if mediainfo else None
|
||
|
||
|
||
def _should_discard_batch_recording_identity(
|
||
*,
|
||
multi_track_music_batch: bool,
|
||
manual: bool,
|
||
media_source: Optional[MediaSource],
|
||
media_id: Optional[str],
|
||
mediainfo: Optional[MediaInfo | MusicInfo],
|
||
history_music_type: Optional[str],
|
||
) -> bool:
|
||
"""
|
||
判断自动整专是否误带了共享单曲身份。
|
||
"""
|
||
if not multi_track_music_batch or (manual and media_source and media_id):
|
||
return False
|
||
batch_music_type = getattr(mediainfo, "music_type", None)
|
||
return (
|
||
batch_music_type == MUSIC_ENTITY_RECORDING
|
||
or (not batch_music_type and history_music_type == MUSIC_ENTITY_RECORDING)
|
||
)
|
||
|
||
|
||
def _should_discard_batch_music_identity(
|
||
*,
|
||
batch_mtype: Optional[MediaType],
|
||
manual: bool,
|
||
multi_track_music_batch: bool,
|
||
media_source: Optional[MediaSource],
|
||
media_id: Optional[str],
|
||
mediainfo: Optional[MediaInfo | MusicInfo],
|
||
history_music_type: Optional[str],
|
||
) -> bool:
|
||
"""
|
||
仅音乐批次在未显式指定媒体时丢弃共享身份并重新识别内部作品。
|
||
"""
|
||
if batch_mtype != MediaType.MUSIC:
|
||
return False
|
||
if manual and not (media_source and media_id):
|
||
return True
|
||
return _should_discard_batch_recording_identity(
|
||
multi_track_music_batch=multi_track_music_batch,
|
||
manual=manual,
|
||
media_source=media_source,
|
||
media_id=media_id,
|
||
mediainfo=mediainfo,
|
||
history_music_type=history_music_type,
|
||
)
|
||
|
||
|
||
class _TransferCandidatePlanner:
|
||
"""
|
||
持有一次整理请求的只读候选规划上下文。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
chain: Any,
|
||
*,
|
||
meta: Optional[MetaBase],
|
||
season: Optional[int],
|
||
formater: Optional[FormatParser],
|
||
batch_mtype: Optional[MediaType],
|
||
mediainfo: Optional[Union[MediaInfo, MusicInfo]],
|
||
continue_callback: Optional[Callable[[], bool]],
|
||
has_episode_format_template: bool,
|
||
transfer_exclude_words: Optional[list[str]],
|
||
download_hash: Optional[str],
|
||
sync_extra_files: bool,
|
||
fileitem: FileItem,
|
||
sync_music_covers: bool = False,
|
||
) -> None:
|
||
"""
|
||
冻结候选规划依赖，避免请求阶段再读取可变配置。
|
||
|
||
:param sync_music_covers: 自动整理音乐封面，不扩展影视附加文件的同步范围
|
||
"""
|
||
self._chain = chain
|
||
self._meta = meta
|
||
self._season = season
|
||
self._formaterHandler = formater
|
||
self._batch_mtype = batch_mtype
|
||
self._mediainfo = mediainfo
|
||
self._continue_callback = continue_callback
|
||
self._has_episode_format_template = has_episode_format_template
|
||
self._transfer_exclude_words = transfer_exclude_words
|
||
self._download_hash = download_hash
|
||
self._sync_extra_files = sync_extra_files
|
||
self._sync_music_covers = sync_extra_files or sync_music_covers
|
||
self._fileitem = fileitem
|
||
|
||
def _build_file_meta(
|
||
self,
|
||
fileitem: FileItem,
|
||
custom_word_list: Optional[List[str]] = None,
|
||
) -> Optional[MetaBase]:
|
||
"""
|
||
构建整理任务使用的文件元数据，并应用手动季集/自定义格式覆盖。
|
||
"""
|
||
if not fileitem.path:
|
||
return None
|
||
source_path = Path(fileitem.path)
|
||
built_meta = deepcopy(self._meta) if self._meta else self._build_path_meta(
|
||
source_path, custom_word_list=custom_word_list, storage=fileitem.storage
|
||
)
|
||
if not built_meta:
|
||
return None
|
||
if not self._meta:
|
||
# _build_path_meta 已经应用过手动季集/自定义格式覆盖；
|
||
# 这里避免再次偏移集数，导致手动整理的集数偏移翻倍。
|
||
return built_meta
|
||
return self._apply_meta_overrides(built_meta, source_path)
|
||
|
||
def _build_file_context(
|
||
self,
|
||
fileitem: FileItem,
|
||
history: Optional[DownloadHistorySnapshot],
|
||
inherited: Optional[MetaBase],
|
||
discard_saved_identity: bool,
|
||
) -> tuple[Optional[MetaBase], Optional[MusicInfo]]:
|
||
"""
|
||
按显式输入、下载证据及附加文件归属构建上下文，读取始终绑定具体存储。
|
||
"""
|
||
if not fileitem.path:
|
||
return None, None
|
||
history_meta, history_info = self._chain._restore_music_download_context(
|
||
download_history=history, file_path=Path(fileitem.path),
|
||
discard_saved_identity=discard_saved_identity,
|
||
storage=fileitem.storage, batch_mtype=self._batch_mtype,
|
||
)
|
||
if self._meta:
|
||
return self._build_file_meta(fileitem), history_info
|
||
if history_meta:
|
||
return history_meta, history_info
|
||
if inherited:
|
||
return deepcopy(inherited), history_info
|
||
return self._build_file_meta(fileitem, self._chain._get_subscribe_custom_words(history)), history_info
|
||
|
||
def _has_reliable_video_source(self) -> bool:
|
||
"""
|
||
是否存在可靠的影视类型来源；存在时音频按附加音轨解析，
|
||
避免影视场景的音频文件误入音乐识别。
|
||
"""
|
||
if self._batch_mtype is not None:
|
||
return self._batch_mtype != MediaType.MUSIC
|
||
# 预载媒体信息为非音乐时，整批整理视为影视上下文
|
||
return self._mediainfo is not None and not isinstance(self._mediainfo, MusicInfo)
|
||
|
||
def _build_path_meta(
|
||
self,
|
||
source_path: Path,
|
||
custom_word_list: Optional[List[str]] = None,
|
||
force_video: Optional[bool] = False,
|
||
*,
|
||
storage: Optional[str] = "local",
|
||
) -> Optional[MetaBase]:
|
||
"""
|
||
从文件路径识别媒体信息，用于判断附加文件是否属于当前主视频。
|
||
|
||
:param force_video: 强制按视频解析，附加文件归属匹配专用，避免音乐判定干扰归属比较
|
||
"""
|
||
# 音频后缀且无可靠影视类型来源时按音乐解析，走 MusicBrainz 识别链
|
||
if (
|
||
not force_video
|
||
and source_path.suffix.lower() in self._chain._audio_exts
|
||
and not self._has_reliable_video_source()
|
||
):
|
||
path_meta = MediaChain.read_path_meta(source_path, storage=storage)
|
||
else:
|
||
# 影视场景附加音轨（如评论音轨）强制按视频解析，保留季集归属
|
||
path_meta = MetaInfoPath(
|
||
source_path, custom_words=custom_word_list, force_video=True
|
||
)
|
||
if not path_meta:
|
||
return None
|
||
return self._apply_meta_overrides(path_meta, source_path)
|
||
|
||
def _apply_meta_overrides(
|
||
self,
|
||
current_meta: MetaBase, source_path: Path
|
||
) -> Optional[MetaBase]:
|
||
"""
|
||
应用手动传入的季集覆盖和自定义识别格式。
|
||
"""
|
||
if self._season is not None:
|
||
current_meta.begin_season = self._season
|
||
|
||
if self._formaterHandler:
|
||
begin_ep, end_ep, part = self._formaterHandler.split_episode(
|
||
file_name=source_path.name, file_meta=current_meta
|
||
)
|
||
if begin_ep is not None:
|
||
current_meta.begin_episode = begin_ep
|
||
if part is not None:
|
||
current_meta.part = part
|
||
if end_ep is not None:
|
||
current_meta.end_episode = end_ep
|
||
|
||
return current_meta
|
||
|
||
def _is_allowed_transfer_item(self, item: FileItem, _is_bluray_dir: bool) -> bool:
|
||
"""
|
||
筛选单文件模式额外读取的字幕/音频，保持模板和屏蔽词语义。
|
||
"""
|
||
if self._continue_callback and not self._continue_callback():
|
||
raise OperationInterrupted()
|
||
item_path = item.path
|
||
if not item_path:
|
||
return False
|
||
item_name = item.name or Path(item_path).name
|
||
if self._has_episode_format_template and self._formaterHandler and not self._formaterHandler.match(item_name):
|
||
return False
|
||
if any(
|
||
marker in item_path
|
||
for marker in ("/@Recycle/", "/#recycle/", "/.", "/@eaDir")
|
||
):
|
||
return False
|
||
return not self._chain._is_blocked_by_exclude_words(item_path, self._transfer_exclude_words)
|
||
|
||
def _build_main_meta(
|
||
self,
|
||
main_fileitem: FileItem,
|
||
main_bluray_dir: bool,
|
||
download_history_repository: DownloadHistoryQueryPort,
|
||
) -> Optional[MetaBase]:
|
||
"""
|
||
构建主视频元数据。
|
||
"""
|
||
main_path = Path(main_fileitem.path or "")
|
||
main_download_history = self._chain._resolve_download_history(
|
||
repository=download_history_repository,
|
||
file_path=main_path,
|
||
bluray_dir=main_bluray_dir,
|
||
download_hash=self._download_hash,
|
||
)
|
||
return self._build_file_meta(
|
||
main_fileitem,
|
||
custom_word_list=self._chain._get_subscribe_custom_words(main_download_history),
|
||
)
|
||
|
||
def _append_item(
|
||
self,
|
||
planned_items: List[Tuple[FileItem, bool]],
|
||
seen_file_keys: set[Tuple[str, str]],
|
||
item: FileItem,
|
||
is_bluray_dir: bool,
|
||
) -> bool:
|
||
"""
|
||
添加待整理文件项并去重。
|
||
"""
|
||
file_key = self._chain._get_file_key(item)
|
||
if file_key in seen_file_keys:
|
||
return False
|
||
planned_items.append((item, is_bluray_dir))
|
||
seen_file_keys.add(file_key)
|
||
return True
|
||
|
||
def _build_directory_index(
|
||
self,
|
||
items: List[Tuple[FileItem, bool]]
|
||
) -> Tuple[
|
||
Dict[Tuple[str, str], List[FileItem]],
|
||
Dict[Tuple[str, str], List[Tuple[FileItem, bool]]],
|
||
]:
|
||
"""
|
||
基于已遍历结果构建同目录主视频和附加文件索引。
|
||
"""
|
||
main_items_by_dir: Dict[Tuple[str, str], List[FileItem]] = {}
|
||
extra_items_by_dir: Dict[Tuple[str, str], List[Tuple[FileItem, bool]]] = {}
|
||
for item, is_bluray_dir in items:
|
||
if not item or item.type != "file":
|
||
continue
|
||
dir_key = self._chain._get_file_parent_key(item)
|
||
if not is_bluray_dir and self._chain._is_media_file(item, self._batch_mtype):
|
||
main_items_by_dir.setdefault(dir_key, []).append(item)
|
||
elif (
|
||
self._chain._is_subtitle_file(item)
|
||
or self._chain._is_audio_file(item)
|
||
or self._chain._is_music_lyrics_file(item)
|
||
or self._is_music_cover_candidate(item)
|
||
):
|
||
extra_items_by_dir.setdefault(dir_key, []).append((item, is_bluray_dir))
|
||
return main_items_by_dir, extra_items_by_dir
|
||
|
||
def _get_single_file_sibling_items(
|
||
self,
|
||
current_fileitem: FileItem,
|
||
) -> Tuple[List[FileItem], List[Tuple[FileItem, bool]]]:
|
||
"""
|
||
单文件整理时只额外读取一次父目录，收集同目录主视频和附加文件。
|
||
"""
|
||
storagechain = StorageChain()
|
||
if not hasattr(storagechain, "get_parent_item") or not hasattr(
|
||
storagechain, "list_files"
|
||
):
|
||
return [], []
|
||
parent_item = storagechain.get_parent_item(current_fileitem)
|
||
if not parent_item:
|
||
return [], []
|
||
main_fileitems: List[FileItem] = []
|
||
extra_items: List[Tuple[FileItem, bool]] = []
|
||
for item in storagechain.list_files(parent_item, recursion=False) or []:
|
||
if not item or item.type != "file":
|
||
continue
|
||
if self._chain._is_media_file(item, self._batch_mtype):
|
||
main_fileitems.append(item)
|
||
continue
|
||
if not (
|
||
self._chain._is_subtitle_file(item)
|
||
or self._chain._is_audio_file(item)
|
||
or self._chain._is_music_lyrics_file(item)
|
||
or self._is_music_cover_candidate(item)
|
||
):
|
||
continue
|
||
if not self._is_allowed_transfer_item(item, False):
|
||
continue
|
||
extra_items.append((item, False))
|
||
return main_fileitems, extra_items
|
||
|
||
def _is_music_cover_candidate(self, item: FileItem) -> bool:
|
||
"""
|
||
封面随音乐主文件同步；自动请求可单独启用，影视批次仍不接纳图片。
|
||
"""
|
||
return (self._sync_music_covers and not self._has_reliable_video_source()
|
||
and self._chain._is_music_cover_file(item))
|
||
|
||
def _infer_music_batch_type(self, items: List[Tuple[FileItem, bool]]) -> None:
|
||
"""
|
||
未指定类型的纯音频请求仍建立音乐主文件索引，混合影视目录保持原类型。
|
||
"""
|
||
if (self._batch_mtype is None and not self._has_reliable_video_source()
|
||
and any(self._chain._is_audio_file(item) for item, _ in items)
|
||
and not any(bluray or self._chain._is_media_file(item, None) for item, bluray in items)):
|
||
self._batch_mtype = MediaType.MUSIC
|
||
|
||
def _append_remaining_items(
|
||
self, items: List[Tuple[FileItem, bool]], planned: List[Tuple[FileItem, bool]],
|
||
seen: set[Tuple[str, str]], inherited: Dict[Tuple[str, str], MetaBase],
|
||
) -> None:
|
||
"""
|
||
保留未匹配的普通附件，音乐歌词和封面必须已获得主文件归属。
|
||
"""
|
||
for item, is_bluray_dir in items:
|
||
if self._chain._is_music_cover_file(item) and self._chain._get_file_key(item) not in inherited:
|
||
continue
|
||
if (self._batch_mtype == MediaType.MUSIC and self._chain._is_music_lyrics_file(item)
|
||
and self._chain._get_file_key(item) not in inherited):
|
||
continue
|
||
self._append_item(planned, seen, item, is_bluray_dir)
|
||
|
||
def _plan_file_items(
|
||
self,
|
||
items: List[Tuple[FileItem, bool]]
|
||
) -> Tuple[List[Tuple[FileItem, bool]], Dict[Tuple[str, str], MetaBase]]:
|
||
"""
|
||
生成最终整理顺序：主视频优先，同名附加文件跟随，剩余附加文件最后处理。
|
||
"""
|
||
if not items:
|
||
return [], {}
|
||
|
||
self._infer_music_batch_type(items)
|
||
|
||
download_history_repository = self._chain.download_history_repository
|
||
inherited_map: Dict[Tuple[str, str], MetaBase] = {}
|
||
main_items_by_dir, extra_items_by_dir = self._build_directory_index(items)
|
||
main_items = [
|
||
(item, is_bluray_dir)
|
||
for item, is_bluray_dir in items
|
||
if item
|
||
and (
|
||
is_bluray_dir
|
||
or (
|
||
item.type == "file"
|
||
and self._chain._is_media_file(item, self._batch_mtype)
|
||
)
|
||
)
|
||
]
|
||
|
||
single_file_mode = len(items) == 1 and self._fileitem.type == "file"
|
||
if single_file_mode:
|
||
current_item, current_bluray_dir = items[0]
|
||
if current_item.type == "file":
|
||
sibling_main_items, sibling_extra_items = self._get_single_file_sibling_items(
|
||
current_item
|
||
)
|
||
current_dir_key = self._chain._get_file_parent_key(current_item)
|
||
if not current_bluray_dir and self._chain._is_media_file(
|
||
current_item, self._batch_mtype
|
||
):
|
||
main_items = [(current_item, current_bluray_dir)]
|
||
main_items_by_dir[current_dir_key] = [current_item]
|
||
extra_items_by_dir[current_dir_key] = sibling_extra_items
|
||
elif (
|
||
self._chain._is_subtitle_file(current_item)
|
||
or self._chain._is_audio_file(current_item)
|
||
or self._chain._is_music_lyrics_file(current_item)
|
||
):
|
||
related_main_file_key = self._chain._get_related_main_file_key(
|
||
extra_fileitem=current_item,
|
||
main_fileitems=sibling_main_items,
|
||
)
|
||
related_main_fileitem = next(
|
||
(
|
||
main_item
|
||
for main_item in sibling_main_items
|
||
if self._chain._get_file_key(main_item) == related_main_file_key
|
||
),
|
||
None,
|
||
)
|
||
if related_main_fileitem:
|
||
main_meta = self._build_main_meta(
|
||
related_main_fileitem,
|
||
False,
|
||
download_history_repository,
|
||
)
|
||
if main_meta:
|
||
inherited_map[self._chain._get_file_key(current_item)] = deepcopy(main_meta)
|
||
return append_cue_companions(self._chain, list(items), inherited_map, self._has_reliable_video_source(), self._transfer_exclude_words)
|
||
|
||
if not main_items:
|
||
remaining = [
|
||
item
|
||
for item in items
|
||
if not self._chain._is_music_cover_file(item[0]) and not (
|
||
self._batch_mtype == MediaType.MUSIC
|
||
and self._chain._is_music_lyrics_file(item[0])
|
||
)
|
||
]
|
||
return append_cue_companions(self._chain, remaining, inherited_map, self._has_reliable_video_source(), self._transfer_exclude_words)
|
||
|
||
planned_items: List[Tuple[FileItem, bool]] = []
|
||
seen_file_keys: set[Tuple[str, str]] = set()
|
||
extra_meta_cache: Dict[Tuple[str, Tuple[str, ...]], Optional[MetaBase]] = {}
|
||
|
||
def _get_cached_extra_meta(
|
||
extra_path: Path,
|
||
custom_word_list: Optional[List[str]],
|
||
) -> Optional[MetaBase]:
|
||
"""
|
||
同一组识别词下的附加文件只解析一次。
|
||
"""
|
||
custom_words_key = tuple(custom_word_list or [])
|
||
cache_key = (extra_path.as_posix(), custom_words_key)
|
||
if cache_key not in extra_meta_cache:
|
||
# 归属匹配专用视频解析：此处目的是判断附加文件是否跟随主视频，
|
||
# 若按音乐解析会导致影视目录内的音频无法与主视频比较归属
|
||
extra_meta_cache[cache_key] = self._build_path_meta(
|
||
extra_path,
|
||
custom_word_list=list(custom_words_key) or None,
|
||
force_video=True,
|
||
)
|
||
return extra_meta_cache[cache_key]
|
||
|
||
for main_item, main_bluray_dir in main_items:
|
||
self._append_item(planned_items, seen_file_keys, main_item, main_bluray_dir)
|
||
if main_bluray_dir or not self._chain._is_media_file(
|
||
main_item, self._batch_mtype
|
||
):
|
||
continue
|
||
|
||
main_path = Path(main_item.path or "")
|
||
main_download_history = self._chain._resolve_download_history(
|
||
repository=download_history_repository,
|
||
file_path=main_path,
|
||
bluray_dir=main_bluray_dir,
|
||
download_hash=self._download_hash,
|
||
)
|
||
subscribe_custom_words = self._chain._get_subscribe_custom_words(
|
||
main_download_history
|
||
)
|
||
main_meta = self._build_file_meta(
|
||
main_item,
|
||
custom_word_list=subscribe_custom_words,
|
||
)
|
||
if not main_meta:
|
||
continue
|
||
|
||
dir_key = self._chain._get_file_parent_key(main_item)
|
||
main_fileitems = main_items_by_dir.get(dir_key) or [main_item]
|
||
main_file_key = self._chain._get_file_key(main_item)
|
||
for extra_item, extra_bluray_dir in extra_items_by_dir.get(dir_key, []):
|
||
if self._chain._get_file_key(extra_item) in seen_file_keys:
|
||
continue
|
||
if self._is_music_cover_candidate(extra_item):
|
||
# 图片没有曲名，先继承主音轨标签；批次分组再确认唯一发行归属。
|
||
if self._chain._is_audio_file(main_item) and self._append_item(
|
||
planned_items, seen_file_keys, extra_item, extra_bluray_dir):
|
||
inherited_map[self._chain._get_file_key(extra_item)] = deepcopy(main_meta)
|
||
continue
|
||
related_main_file_key = self._chain._get_related_main_file_key(
|
||
extra_fileitem=extra_item,
|
||
main_fileitems=main_fileitems,
|
||
)
|
||
if related_main_file_key:
|
||
if related_main_file_key == main_file_key:
|
||
if self._append_item(
|
||
planned_items,
|
||
seen_file_keys,
|
||
extra_item,
|
||
extra_bluray_dir,
|
||
):
|
||
inherited_map[self._chain._get_file_key(extra_item)] = deepcopy(main_meta)
|
||
continue
|
||
|
||
if single_file_mode or not self._sync_extra_files:
|
||
continue
|
||
|
||
extra_meta = _get_cached_extra_meta(
|
||
Path(extra_item.path or ""),
|
||
subscribe_custom_words,
|
||
)
|
||
if extra_meta is None or not self._chain._is_same_media_meta(main_meta, extra_meta):
|
||
continue
|
||
if self._append_item(
|
||
planned_items,
|
||
seen_file_keys,
|
||
extra_item,
|
||
extra_bluray_dir,
|
||
):
|
||
inherited_map[self._chain._get_file_key(extra_item)] = deepcopy(extra_meta)
|
||
|
||
self._append_remaining_items(items, planned_items, seen_file_keys, inherited_map)
|
||
|
||
return append_cue_companions(self._chain, planned_items, inherited_map, self._has_reliable_video_source(), self._transfer_exclude_words)
|
||
|
||
|
||
def build_transfer_preview_item(task: TransferTask, transferinfo: TransferInfo) -> dict[str, Any]:
|
||
"""
|
||
按实际整理结果投影预览项，失败阶段与恢复动作保留原始裁决。
|
||
"""
|
||
from app.application.transfer.feedback import classify_transfer_failure
|
||
|
||
item_meta = task.meta
|
||
item_media = task.mediainfo
|
||
feedback = classify_transfer_failure(
|
||
transferinfo.message,
|
||
overwrite_skipped=bool(transferinfo.overwrite_skipped),
|
||
)
|
||
return (
|
||
{
|
||
"source": task.fileitem.path,
|
||
"source_storage": task.fileitem.storage,
|
||
"source_item": task.fileitem.model_dump(exclude={"children", "url", "thumbnail"}),
|
||
"music": music_transfer_preview(item_meta, item_media, task.music_preview_context,
|
||
selected=bool(task.manual and task.media_source and task.media_id)),
|
||
"target": transferinfo.target_item.path if transferinfo.target_item else None,
|
||
"target_dir": transferinfo.target_diritem.path if transferinfo.target_diritem else None,
|
||
"success": transferinfo.success,
|
||
"message": transferinfo.message,
|
||
"failure_stage": (
|
||
transferinfo.failure_stage or feedback.stage.value
|
||
if not transferinfo.success
|
||
else None
|
||
),
|
||
"recovery_action": (
|
||
transferinfo.recovery_action or feedback.action
|
||
if not transferinfo.success
|
||
else None
|
||
),
|
||
"overwrite_skipped": bool(transferinfo.overwrite_skipped),
|
||
"type": item_media.type.value if item_media and item_media.type else None,
|
||
"title": preview_media_title(item_media),
|
||
"season": item_meta.begin_season if item_meta else None,
|
||
"episode": item_meta.begin_episode if item_meta else None,
|
||
"episode_end": item_meta.end_episode if item_meta else None,
|
||
"part": item_meta.part if item_meta else None,
|
||
"org_string": item_meta.org_string if item_meta else None,
|
||
"apply_words": (item_meta.apply_words or []) if item_meta else [],
|
||
"resource_team": item_meta.resource_team if item_meta else None,
|
||
"customization": item_meta.customization if item_meta else None,
|
||
}
|
||
)
|
||
|
||
|
||
class _TransferSubmissionCollector:
|
||
"""
|
||
按当前请求收集真实阶段回执，不改变任务、重试或结算的所有权。
|
||
"""
|
||
|
||
def __init__(self, enabled: bool = False, source: Optional[FileItem] = None) -> None:
|
||
"""
|
||
普通整理调用默认关闭回执，维持原有字符串返回契约。
|
||
"""
|
||
self.enabled = enabled
|
||
self.source = source
|
||
self.items: list[dict[str, Any]] = []
|
||
self._skipped_history_count = 0
|
||
self._results: dict[int, TransferInfo] = {}
|
||
self._pending: dict[tuple[Optional[str], Optional[str]], FileItem] = {}
|
||
self._preview_rejections: list[dict[str, Any]] = []
|
||
|
||
def record_music_package_preview(self, fileitem: FileItem, message: str, *, preview: bool) -> None:
|
||
"""
|
||
保留尚未生成任务的音乐包拒绝项，混合目录也能逐文件显示能力边界。
|
||
"""
|
||
if not preview:
|
||
return
|
||
self._preview_rejections.append({
|
||
"source": fileitem.path, "source_storage": fileitem.storage,
|
||
"source_item": fileitem.model_dump(exclude={"children", "url", "thumbnail"}),
|
||
"success": False, "message": message, "type": MediaType.MUSIC.value,
|
||
"failure_stage": "recognition", "recovery_action": message,
|
||
"music": {"status": "unsupported", "online_confirmed": False},
|
||
})
|
||
|
||
def expect(self, fileitems: list[tuple[FileItem, bool]]) -> None:
|
||
"""
|
||
保留候选快照，取消发生后仍能说明哪些文件没有被执行。
|
||
"""
|
||
if self.enabled:
|
||
self._pending.update({(item.storage, item.path): item for item, _bluray in fileitems})
|
||
|
||
def record(
|
||
self, fileitem: FileItem,
|
||
state: Literal["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"],
|
||
message: Optional[str] = None,
|
||
*, target_dir: Optional[Path] = None,
|
||
) -> None:
|
||
"""
|
||
在准入、历史跳过或重试决策发生时记录回执。
|
||
"""
|
||
if not self.enabled:
|
||
return
|
||
self._pending.pop((fileitem.storage, fileitem.path), None)
|
||
self.items.append({
|
||
"source": fileitem.path, "state": state,
|
||
"success": state in {"accepted", "completed", "retry_wait"},
|
||
"message": message,
|
||
"target_dir": target_dir.as_posix() if target_dir else None,
|
||
})
|
||
|
||
def record_history_skip(self, fileitem: FileItem) -> None:
|
||
"""
|
||
逐项保留跳过回执，总提示只累计数量，避免大目录产生冗长的文件名列表。
|
||
"""
|
||
self._skipped_history_count += 1
|
||
self.record(fileitem, "skipped", f"已跳过成功整理记录：{fileitem.name}")
|
||
|
||
def capture(self, task: TransferTask, transferinfo: TransferInfo) -> None:
|
||
"""
|
||
暂存执行结果，等终态原子结算确认后再公布 completed。
|
||
"""
|
||
if self.enabled:
|
||
self._results[id(task)] = transferinfo
|
||
|
||
def execution(
|
||
self, task: TransferTask, success: bool, message: str,
|
||
*, durable_state: Optional[str],
|
||
) -> None:
|
||
"""
|
||
只有已确认的终态可报告完成，未确认终态保留后台重试状态。
|
||
"""
|
||
if not self.enabled:
|
||
return
|
||
info = self._results.pop(id(task), None)
|
||
state: Literal["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"] = "failed"
|
||
if task.terminal_settled:
|
||
state = "completed" if (info.success if info else success) else "failed"
|
||
if info and info.overwrite_skipped:
|
||
state = "skipped"
|
||
elif durable_state == "manual_review":
|
||
state = "manual_review"
|
||
elif durable_state in {"not_started", "running", "retry_wait", "settling"}:
|
||
state = "retry_wait"
|
||
elif durable_state == "unknown":
|
||
state = "accepted"
|
||
message = f"{message}；任务状态暂未确认，请刷新整理历史"
|
||
elif success:
|
||
state = "accepted"
|
||
if state == "completed" and not success:
|
||
message = "文件已完成整理，但后续处理失败，请检查整理日志，无需重复整理"
|
||
self.record(task.fileitem, state, message, target_dir=task.target_path)
|
||
if info:
|
||
projected = build_transfer_preview_item(task, info)
|
||
self.items[-1].update({key: projected[key] for key in (
|
||
"target", "target_dir", "failure_stage", "recovery_action", "overwrite_skipped",
|
||
)})
|
||
if state == "manual_review":
|
||
self.items[-1]["recovery_action"] = "打开整理队列，先完成人工复核，再决定是否重试"
|
||
|
||
def result(
|
||
self, success: bool, errors: list[str], *, preview: bool,
|
||
preview_items: list[dict[str, Any]],
|
||
) -> tuple[bool, Union[str, dict[str, Any]]]:
|
||
"""
|
||
保持预览与旧调用返回值，同时让管理界面收到完整执行回执。
|
||
"""
|
||
message = "、".join(errors[:2]) + (f"，等{len(errors)}个文件错误！" if len(errors) > 2 else "")
|
||
if self._skipped_history_count:
|
||
message = "；".join(filter(None, [f"已跳过 {self._skipped_history_count} 条成功整理记录", message]))
|
||
if preview:
|
||
preview_items = [*self._preview_rejections, *preview_items]
|
||
return success, {
|
||
"summary": {
|
||
"total": len(preview_items),
|
||
"success": sum(bool(item.get("success")) for item in preview_items),
|
||
"failed": sum(not item.get("success") for item in preview_items),
|
||
},
|
||
"items": preview_items, "message": message,
|
||
}
|
||
for pending in list(self._pending.values()):
|
||
self.record(pending, "skipped" if success else "failed", message or "本次整理未执行")
|
||
if self.enabled and not self.items and self.source:
|
||
self.record(self.source, "skipped" if success else "failed", message or "没有需要执行的整理任务")
|
||
return success, {"items": self.items, "message": message} if self.enabled else message
|