fix(music): preserve album context for manual batches

2026-11-02 13:28:32 +08:00 · 2026-09-08 15:06:36 +08:00
parent f9f2601144
commit 4d7bd96c6b
10 changed files with 365 additions and 13 deletions
--- a/app/api/endpoints/transfer.py
+++ b/app/api/endpoints/transfer.py
@@ -1,3 +1,4 @@
+import re
 from pathlib import Path
 from typing import Annotated, Any, List, Literal, Optional, cast

@@ -115,6 +116,58 @@ def _merge_transfer_messages(messages: List[str]) -> str:
    )


+def _is_music_file_batch(fileitems: List[FileItem]) -> bool:
+    """判断显式选中项是否为可按专辑识别的多音轨批次。"""
+    if len(fileitems) < 2 or any(fileitem.type != "file" for fileitem in fileitems):
+        return False
+    audio_extensions = set(get_api_runtime_config_snapshot().audio_extensions)
+    return all(
+        Path(fileitem.path or fileitem.name or "").suffix.lower() in audio_extensions
+        for fileitem in fileitems
+    )
+
+
+def _selected_music_fileitems(
+    fileitems: List[FileItem],
+    *,
+    explicitly_selected: bool,
+    media_type: Optional[MediaType],
+) -> Optional[List[FileItem]]:
+    """返回需要共享专辑上下文的显式多选音轨。"""
+    if not explicitly_selected or media_type not in (None, MediaType.MUSIC):
+        return None
+    if not _is_music_file_batch(fileitems):
+        return None
+    paths = [Path(fileitem.path) for fileitem in fileitems if fileitem.path]
+    parent_counts: dict[Path, int] = {}
+    for path in paths:
+        parent_counts[path.parent] = parent_counts.get(path.parent, 0) + 1
+    album_roots = [
+        parent
+        for parent in parent_counts
+        if all(path.parent == parent or parent in path.parents for path in paths)
+    ]
+    if not album_roots:
+        return fileitems
+    album_root = max(album_roots, key=lambda path: len(path.parts))
+    filtered = []
+    for fileitem, path in zip(fileitems, paths):
+        if path.parent == album_root:
+            filtered.append(fileitem)
+            continue
+        relative_parent = path.parent.relative_to(album_root)
+        first_directory = relative_parent.parts[0] if relative_parent.parts else ""
+        if re.fullmatch(r"(?:cd|disc|disk)\s*\d{1,2}", first_directory.strip(), re.IGNORECASE):
+            filtered.append(fileitem)
+    if filtered and len(filtered) < len(fileitems):
+        logger.info(
+            "音乐专辑批次忽略 %s 个非碟片子目录中的重复或附加音频",
+            len(fileitems) - len(filtered),
+        )
+        return filtered
+    return fileitems
+
+
 def _manual_review_actor(current_user: object) -> str:
    """按名称、用户名和用户 ID 的稳定顺序提取人工复核操作者。"""
    for attribute in ("name", "username", "id"):
@@ -674,6 +727,11 @@ def _execute_manual_transfer(
            offset=transer_item.episode_offset,
        )
    explicit_selected_files = bool(transer_item.fileitems)
+    selected_music_fileitems = _selected_music_fileitems(
+        src_fileitems, explicitly_selected=explicit_selected_files, media_type=mtype
+    )
+    mtype = MediaType.MUSIC if selected_music_fileitems is not None else mtype
+    explicit_selected_files = explicit_selected_files and selected_music_fileitems is None

    # 前端显式传入文件列表时，按选中的文件逐个处理，避免将目录整体展开。
    if explicit_selected_files:
@@ -784,7 +842,11 @@ def _execute_manual_transfer(
        target_path=target_path,
        media_source=transer_item.media_source,
        media_id=transer_item.media_id,
-        music_type=_resolve_music_type(src_fileitem),
+        music_type=(
+            MUSIC_ENTITY_ALBUM
+            if selected_music_fileitems is not None
+            else _resolve_music_type(src_fileitem)
+        ),
        music_release_regions=transer_item.music_release_regions,
        music_release_scripts=transer_item.music_release_scripts,
        mtype=mtype,
@@ -802,8 +864,9 @@ def _execute_manual_transfer(
        download_hash=download_hash,
        preview=transer_item.preview,
        reorganize=transer_item.reorganize,
-        sync_extra_files=True,
+        sync_extra_files=selected_music_fileitems is None,
        cleanup_dest_fileitem=cleanup_dest_fileitem,
+        selected_fileitems=selected_music_fileitems,
    )
    # 失败
    if not state:
--- a/app/chain/media/album.py
+++ b/app/chain/media/album.py
@@ -1,6 +1,7 @@
 """专辑目录扫描、曲目对齐与缓存编排 owner。"""

 import os
+from copy import deepcopy
 from pathlib import Path
 from typing import Optional, Union, cast

@@ -62,7 +63,8 @@ class MediaAlbumOwner(_MediaOwnerBase):
        except OSError:
            subdirectories = []
        for subdirectory in subdirectories:
-            collect(subdirectory)
+            if MetaMusic.parse_disc_dir(subdirectory.name) is not None:
+                collect(subdirectory)
        return files

    @staticmethod
@@ -177,6 +179,24 @@ class MediaAlbumOwner(_MediaOwnerBase):
            ).items()
        }

+    @classmethod
+    def _album_track_map(
+        cls,
+        files: list[Path],
+        metas: list[MetaMusic],
+        album: MusicAlbumInfo,
+    ) -> dict[str, MusicInfo]:
+        """将专辑分类与识别事实传递给每条已对位曲目。"""
+        aligned = cls._align_music_album_tracks(files, metas, album.tracks)
+        for info in aligned.values():
+            info.set_library_category(album.library_category)
+            info.classification = deepcopy(album.classification)
+            info.classification_facts = dict(album.classification_facts)
+        return {
+            str(file.resolve()): info
+            for file, info in aligned.items()
+        }
+
    def _match_music_album_directory(
        self,
        directory: Path,
@@ -199,10 +219,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
        )
        if not album or not album.tracks:
            return {}
-        return {
-            str(file.resolve()): info
-            for file, info in self._align_music_album_tracks(files, metas, album.tracks).items()
-        }
+        return self._album_track_map(files, metas, album)

    async def _async_match_music_album_directory(
        self,
@@ -226,10 +243,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
        )
        if not album or not album.tracks:
            return {}
-        return {
-            str(file.resolve()): info
-            for file, info in self._align_music_album_tracks(files, metas, album.tracks).items()
-        }
+        return self._album_track_map(files, metas, album)

    def recognize_music_album_directory(
        self,
--- a/app/chain/media/contract.py
+++ b/app/chain/media/contract.py
@@ -173,6 +173,16 @@ if TYPE_CHECKING:
            """同步识别音乐专辑目录。"""
            ...

+        @classmethod
+        def _album_track_map(
+            cls,
+            files: list[Path],
+            metas: list[MetaMusic],
+            album: MusicAlbumInfo,
+        ) -> dict[str, MusicInfo]:
+            """将专辑级分类上下文传递给对位曲目。"""
+            ...
+
        async def async_recognize_music_album_directory(
            self,
            path: Union[str, Path],
--- a/app/chain/media/facade.py
+++ b/app/chain/media/facade.py
@@ -126,6 +126,9 @@ class MediaChain(ChainBase, metaclass=Singleton):
    _align_selected_music_album = classmethod(  # type: ignore[var-annotated]
        MediaAlbumOwner._align_selected_music_album.__func__  # type: ignore[attr-defined]
    )
+    _album_track_map = classmethod(  # type: ignore[var-annotated]
+        MediaAlbumOwner._album_track_map.__func__  # type: ignore[attr-defined]
+    )
    _match_music_album_directory = MediaAlbumOwner._match_music_album_directory
    _async_match_music_album_directory = MediaAlbumOwner._async_match_music_album_directory
    recognize_music_album_directory = cast(
--- a/app/chain/transfer/filter.py
+++ b/app/chain/transfer/filter.py
@@ -235,6 +235,7 @@ class FileFilterMixin(_TransferOwnerBase):
        merged_info.set_library_category(info.library_category)
        merged_info.metadata_category = info.metadata_category
        merged_info.classification = deepcopy(info.classification)
+        merged_info.classification_facts = dict(info.classification_facts)
        merged_info.genres = list(info.genres)
        merged_info.tags = list(info.tags)
        merged_info.artist_country = info.artist_country
--- a/app/chain/transfer/history.py
+++ b/app/chain/transfer/history.py
@@ -171,6 +171,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
            music_type: Optional[str] = None,
            music_release_regions: Optional[list[str]] = None,
            music_release_scripts: Optional[list[str]] = None,
+            selected_fileitems: Optional[list[FileItem]] = None,
    ) -> Tuple[bool, Union[str, dict[str, Any]]]:
        """
        手动整理，支持复杂条件，带进度显示
@@ -199,6 +200,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
        :param music_type: 音乐实体类型；为保持位置参数兼容，必须追加在签名末尾
        :param music_release_regions: 本次音乐整理的发行地区优先级，空值继承系统设置
        :param music_release_scripts: 本次音乐整理的文字字形优先级，空值继承系统设置
+        :param selected_fileitems: 前端显式选中的批量文件
        """
        logger.info(f"手动整理：{fileitem.path} ...")
        explicit_identity = media_source is not None or media_id is not None
@@ -252,6 +254,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
                cleanup_dest_fileitem=cleanup_dest_fileitem,
                music_release_regions=music_release_regions,
                music_release_scripts=music_release_scripts,
+                selected_fileitems=selected_fileitems,
            )
            if not state:
                return False, errmsg
@@ -284,6 +287,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
                cleanup_dest_fileitem=cleanup_dest_fileitem,
                music_release_regions=music_release_regions,
                music_release_scripts=music_release_scripts,
+                selected_fileitems=selected_fileitems,
            )
            return state, errmsg

--- a/app/chain/transfer/workflow.py
+++ b/app/chain/transfer/workflow.py
@@ -276,6 +276,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        epformat: Optional[EpisodeFormat],
        season: Optional[int],
        continue_callback: Optional[Callable],
+        selected_fileitems: Optional[list[FileItem]] = None,
    ) -> Tuple[List[Tuple[FileItem, bool]], bool]:
        """
        收集并过滤本次整理的候选文件。
@@ -329,7 +330,15 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                return False
            return not self._is_blocked_by_exclude_words(item.path, exclude_words)

-        candidates = self._TransferChain__get_trans_fileitems(fileitem, predicate=keep_candidate)
+        if selected_fileitems is None:
+            candidates = self._TransferChain__get_trans_fileitems(fileitem, predicate=keep_candidate)
+        else:
+            storage_chain = StorageChain()
+            candidates = [
+                (latest_fileitem, False)
+                for selected_fileitem in selected_fileitems
+                if (latest_fileitem := storage_chain.get_item(selected_fileitem))
+            ]
        return [
            (item, is_bluray_dir) for item, is_bluray_dir in candidates if is_allowed(item, is_bluray_dir)
        ], matched_template
@@ -364,6 +373,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        reorganize: Optional[bool] = False,
        music_release_regions: Optional[list[str]] = None,
        music_release_scripts: Optional[list[str]] = None,
+        selected_fileitems: Optional[list[FileItem]] = None,
    ) -> Tuple[bool, Union[str, dict]]:
        """
        兼容公开整理入口，委托给内部批次执行阶段。
@@ -400,6 +410,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
            reorganize=reorganize,
            music_release_regions=music_release_regions,
            music_release_scripts=music_release_scripts,
+            selected_fileitems=selected_fileitems,
        )

    def _execute_transfer(self, *args: Any, **kwargs: Any) -> Tuple[bool, Union[str, dict]]:
@@ -437,6 +448,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        recovery_admission: Optional[TransferAdmission] = None,
        music_release_regions: Optional[list[str]] = None,
        music_release_scripts: Optional[list[str]] = None,
+        selected_fileitems: Optional[list[FileItem]] = None,
    ) -> Tuple[bool, Union[str, dict]]:
        """
        执行一个复杂目录的整理操作
@@ -469,6 +481,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        :param recovery_admission: 内部恢复调用绑定的既有 durable 记录
        :param music_release_regions: 本次音乐整理的发行地区优先级
        :param music_release_scripts: 本次音乐整理的文字字形优先级
+        :param selected_fileitems: 前端显式选中的文件，按同一批次规划
        返回：成功标识，错误信息
        """
        selected_music_album: Optional[MusicAlbumInfo]
@@ -487,7 +500,6 @@ class TransferWorkflowOwner(_TransferOwnerBase):
        if identity_error:
            return False, identity_error

-        # 是否全部成功
        all_success = True
        transfer_batch_id = str(uuid.uuid4())
        batch_mtype = getattr(mediainfo, "type", None)
@@ -536,6 +548,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                epformat=epformat,
                season=season,
                continue_callback=continue_callback,
+                selected_fileitems=selected_fileitems,
            )
        except OperationInterrupted:
            return False, f"{fileitem.name} 已取消"
--- a/tests/test_music_album_match.py
+++ b/tests/test_music_album_match.py
@@ -203,6 +203,7 @@ def test_recognize_album_directory_maps_files(tmp_path, media_chain, monkeypatch
        media_id="rg-1",
        title="七里香",
        artists=["周杰伦"],
+        library_category="Album",
        tracks=[
            MusicInfo(
                media_source="musicbrainz",
@@ -230,11 +231,30 @@ def test_recognize_album_directory_maps_files(tmp_path, media_chain, monkeypatch
        info = matched[str(file.resolve())]
        assert info.media_id == f"rec-{index + 1}"
        assert info.title == ALBUM_TRACKS[index][0]
+        assert info.library_category == "Album"
    # 同一目录再次识别直接命中缓存，不重复请求模块
    assert media_chain.recognize_music_album_directory(album_dir) == matched
    source_chain.match_music_album.assert_called_once()


+def test_directory_audio_files_only_includes_disc_subdirectories(tmp_path):
+    """专辑根目录只应合并 CD/Disc 子目录，不得把附加版本当成额外碟。"""
+    album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
+    disc_dir = album_dir / "CD2"
+    alternate_dir = album_dir / "附加原版"
+    disc_dir.mkdir(parents=True)
+    alternate_dir.mkdir()
+    root_track = album_dir / "01.flac"
+    disc_track = disc_dir / "01.flac"
+    alternate_track = alternate_dir / "01.flac"
+    for path in (root_track, disc_track, alternate_track):
+        path.write_bytes(b"audio")
+
+    files = MediaChain._directory_audio_files(album_dir)
+
+    assert files == [root_track, disc_track]
+
+
 def test_align_album_tracks_prefers_exact_titles_over_conflicting_positions():
    """本地曲序与发行版本冲突时，精确曲名必须优先，避免整张专辑错位。"""
    files = [
--- a/tests/test_transfer_history_retransfer.py
+++ b/tests/test_transfer_history_retransfer.py
@@ -429,6 +429,120 @@ def test_manual_transfer_preview_multi_select_collects_failures(monkeypatch):
    assert resp.data["items"][1]["success"] is False


+def test_manual_transfer_music_files_share_one_album_batch(monkeypatch):
+    """多选音轨应一次进入整理链，避免按单曲丢失专辑分类上下文。"""
+    selected_fileitems = [
+        {
+            "storage": "local",
+            "path": f"/downloads/七里香/{index:02d}.flac",
+            "name": f"{index:02d}.flac",
+            "extension": "flac",
+            "type": "file",
+        }
+        for index in (1, 2)
+    ]
+    captured = []
+
+    class FakeTransferChain:
+        def manual_transfer(self, **kwargs):
+            captured.append(kwargs)
+            return True, {
+                "summary": {"total": 2, "success": 2, "failed": 0},
+                "items": [
+                    {
+                        "source": item.path,
+                        "target": f"/library/Album/周杰伦/七里香/{item.name}",
+                        "target_dir": "/library/Album/周杰伦/七里香",
+                        "success": True,
+                        "message": "",
+                        "type": "音乐",
+                        "title": "七里香 (2004)",
+                    }
+                    for item in kwargs["selected_fileitems"]
+                ],
+                "message": "",
+            }
+
+    monkeypatch.setattr("app.api.endpoints.transfer.TransferChain", FakeTransferChain)
+    monkeypatch.setattr(
+        "app.api.endpoints.transfer.get_api_runtime_config_snapshot",
+        lambda: SimpleNamespace(audio_extensions=(".flac",)),
+    )
+
+    resp = manual_transfer(
+        transer_item=ManualTransferItem(
+            fileitems=selected_fileitems,
+            preview=True,
+            type_name="自动",
+        ),
+        background=False,
+        history_query=SimpleNamespace(get=lambda _history_id: None),
+        _="token",
+    )
+
+    assert resp.success is True
+    assert resp.data["summary"] == {"total": 2, "success": 2, "failed": 0}
+    assert len(captured) == 1
+    assert captured[0]["mtype"].value == "音乐"
+    assert captured[0]["music_type"] == "album"
+    assert [item.path for item in captured[0]["selected_fileitems"]] == [
+        item["path"] for item in selected_fileitems
+    ]
+
+
+def test_manual_transfer_music_batch_ignores_non_disc_alternate_directory(monkeypatch):
+    """根目录已有完整专辑时，不得把“附加原版”等任意子目录当成第二张碟。"""
+    selected_fileitems = [
+        {
+            "storage": "local",
+            "path": path,
+            "name": path.rsplit("/", 1)[-1],
+            "extension": "flac",
+            "type": "file",
+        }
+        for path in (
+            "/downloads/七里香/01.flac",
+            "/downloads/七里香/02.flac",
+            "/downloads/七里香/附加原版/01.flac",
+            "/downloads/七里香/附加原版/02.flac",
+        )
+    ]
+    captured = []
+
+    class FakeTransferChain:
+        def manual_transfer(self, **kwargs):
+            captured.append(kwargs)
+            return True, {
+                "summary": {"total": 2, "success": 2, "failed": 0},
+                "items": [],
+                "message": "",
+            }
+
+    monkeypatch.setattr("app.api.endpoints.transfer.TransferChain", FakeTransferChain)
+    monkeypatch.setattr(
+        "app.api.endpoints.transfer.get_api_runtime_config_snapshot",
+        lambda: SimpleNamespace(audio_extensions=(".flac",)),
+    )
+
+    response = manual_transfer(
+        transer_item=ManualTransferItem(
+            fileitems=selected_fileitems,
+            preview=True,
+            type_name="自动",
+        ),
+        background=False,
+        history_query=SimpleNamespace(get=lambda _history_id: None),
+        _="token",
+    )
+
+    assert response.success is True
+    assert len(captured) == 1
+    assert [item.path for item in captured[0]["selected_fileitems"]] == [
+        "/downloads/七里香/01.flac",
+        "/downloads/七里香/02.flac",
+    ]
+
+
 def test_match_manual_transfer_target_path_returns_directory_match(monkeypatch):
    captured = {}

--- a/tests/test_transfer_music_album_selection.py
+++ b/tests/test_transfer_music_album_selection.py
@@ -149,6 +149,116 @@ def test_selected_album_tracks_override_source_tag_names(tmp_path, monkeypatch):
    ]


+def test_selected_music_fileitems_keep_album_batch_context(tmp_path, monkeypatch):
+    """显式多选音轨应在单个批次中应用专辑曲目和分类。"""
+    album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
+    album_dir.mkdir()
+    paths = [album_dir / "01.flac", album_dir / "02.flac"]
+    for path in paths:
+        path.write_bytes(b"audio")
+    fileitems = [make_fileitem(path.as_posix()) for path in paths]
+    local_metas = {
+        paths[0]: MetaMusic(title="我的地盤", artists=["周杰倫"], track_number=1),
+        paths[1]: MetaMusic(title="藉口", artists=["周杰倫"], track_number=2),
+    }
+    chain = _prepare_chain(monkeypatch, fileitems)
+    monkeypatch.setattr(
+        "app.chain.transfer.workflow.StorageChain.get_item",
+        lambda _self, item: item,
+    )
+    monkeypatch.setattr(
+        "app.chain.media.album.AudioMetadataHelper.read_many",
+        lambda requested: [deepcopy(local_metas[path]) for path in requested],
+    )
+    monkeypatch.setattr(
+        MediaChain,
+        "read_path_meta",
+        staticmethod(lambda path: deepcopy(local_metas[path])),
+    )
+    planned = []
+
+    def handle_transfer(task, callback=None):
+        del callback
+        planned.append((task.meta.title, task.mediainfo.library_category))
+        return True, ""
+
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", handle_transfer)
+
+    state, message = TransferChain.do_transfer(
+        chain,
+        fileitem=fileitems[0],
+        selected_fileitems=fileitems,
+        mediainfo=_album(),
+        mtype=MediaType.MUSIC,
+        media_source=MediaSource.MusicBrainz,
+        media_id="release-group-1",
+        background=False,
+    )
+
+    assert state is True
+    assert message == ""
+    assert planned == [("我的地盘", "Album"), ("借口", "Album")]
+
+
+def test_automatic_music_fileitems_receive_album_identity_and_category(tmp_path, monkeypatch):
+    """自动多选音轨应用目录级专辑识别结果覆盖本地繁体标签。"""
+    album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
+    album_dir.mkdir()
+    paths = [album_dir / "01.flac", album_dir / "02.flac"]
+    for path in paths:
+        path.write_bytes(b"audio")
+    fileitems = [make_fileitem(path.as_posix()) for path in paths]
+    local_metas = {
+        paths[0]: MetaMusic(title="我的地盤", artists=["周杰倫"], track_number=1),
+        paths[1]: MetaMusic(title="藉口", artists=["周杰倫"], track_number=2),
+    }
+    album = _album()
+    matched = {}
+    for path, track in zip(paths, album.tracks):
+        matched_track = deepcopy(track)
+        matched_track.set_library_category(album.library_category)
+        matched_track.classification = deepcopy(album.classification)
+        matched[str(path.resolve())] = matched_track
+    chain = _prepare_chain(monkeypatch, fileitems)
+    monkeypatch.setattr(
+        "app.chain.transfer.workflow.StorageChain.get_item",
+        lambda _self, item: item,
+    )
+    monkeypatch.setattr(
+        MediaChain,
+        "read_path_meta",
+        staticmethod(lambda path: deepcopy(local_metas[path])),
+    )
+    monkeypatch.setattr(
+        MediaChain,
+        "recognize_music_album_directory",
+        lambda _self, _path, **_kwargs: matched,
+    )
+    planned = []
+
+    def handle_transfer(task, callback=None):
+        del callback
+        planned.append((task.meta.title, task.mediainfo.album, task.mediainfo.library_category))
+        return True, ""
+
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", handle_transfer)
+
+    state, message = TransferChain.do_transfer(
+        chain,
+        fileitem=fileitems[0],
+        selected_fileitems=fileitems,
+        mtype=MediaType.MUSIC,
+        background=False,
+    )
+
+    assert state is True
+    assert message == ""
+    assert planned == [
+        ("我的地盘", "七里香", "Album"),
+        ("借口", "七里香", "Album"),
+    ]
+
+
 def test_manual_album_identity_forwards_full_selected_album(monkeypatch):
    """手动指定专辑 ID 时不应在进入整理链前丢失曲目表。"""
    chain = make_transfer_chain()