Merge pull request #6601 from dogodefi/codex/music-release-preferences

feat(music): add MusicBrainz release preferences

app/api/endpoints/transfer.py

@@ -688,6 +688,8 @@ def _execute_manual_transfer(
                 media_source=transer_item.media_source,
                 media_id=transer_item.media_id,
                 music_type=_resolve_music_type(src_fileitem),
+                music_release_regions=transer_item.music_release_regions,
+                music_release_scripts=transer_item.music_release_scripts,
                 mtype=mtype,
                 season=transer_item.season,
                 episode_group=transer_item.episode_group,
@@ -783,6 +785,8 @@ def _execute_manual_transfer(
         media_source=transer_item.media_source,
         media_id=transer_item.media_id,
         music_type=_resolve_music_type(src_fileitem),
+        music_release_regions=transer_item.music_release_regions,
+        music_release_scripts=transer_item.music_release_scripts,
         mtype=mtype,
         season=transer_item.season,
         episode_group=transer_item.episode_group,

app/application/configuration.py

@@ -221,6 +221,8 @@ class ChainRuntimeConfig:
     download_subtitle: bool = True
     lyrics_batch_timeout: int = 120
     music_metadata_to_simplified: bool = True
+    music_release_region_priority: tuple[str, ...] = ()
+    music_release_script_priority: tuple[str, ...] = ()
     recognize_plugin_first: bool = False
     ai_agent_enable: bool = False
     ai_agent_global: bool = False

app/chain/media/album.py

@@ -10,6 +10,7 @@ from app.chain.media.cache import AlbumSignature
 from app.chain.media.contract import _MediaOwnerBase
 from app.chain.musicbrainz import MusicBrainzChain
 from app.domain.context import (
+    MusicAlbumInfo,
     MusicInfo,
 )
 from app.domain.meta.metamusic import MetaMusic
@@ -22,9 +23,13 @@ def _is_directory(path: Path) -> bool:
     return path.is_dir()
 
 
-def _album_directory_cache_key(directory: Path) -> str:
-    """返回保留符号链接别名的绝对目录键，避免不同目录语义错误共享结果。"""
-    return os.path.abspath(directory)
+def _album_directory_cache_key(
+    directory: Path,
+    regions: tuple[str, ...],
+    scripts: tuple[str, ...],
+) -> str:
+    """返回包含发行偏好的目录键，避免不同手动选择错误共享结果。"""
+    return "|".join((os.path.abspath(directory), ",".join(regions), ",".join(scripts)))
 
 
 class MediaAlbumOwner(_MediaOwnerBase):
@@ -155,15 +160,43 @@ class MediaAlbumOwner(_MediaOwnerBase):
             matched[file] = track
         return matched
 
+    @classmethod
+    def _align_selected_music_album(
+        cls,
+        files: list[Path],
+        album: MusicAlbumInfo,
+    ) -> dict[str, MusicInfo]:
+        """读取本地标签，并把手动选择发行版的曲目对齐到文件。"""
+        metas = AudioMetadataHelper.read_many(files)
+        return {
+            str(file.resolve()): info
+            for file, info in cls._align_music_album_tracks(
+                files,
+                metas,
+                album.tracks,
+            ).items()
+        }
+
     def _match_music_album_directory(
         self,
         directory: Path,
         files: list[Path],
+        music_release_regions: Optional[tuple[str, ...]] = None,
+        music_release_scripts: Optional[tuple[str, ...]] = None,
     ) -> dict[str, MusicInfo]:
         """同步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
         metas = AudioMetadataHelper.read_many(files)
         album_meta = MetaMusic.from_album_context(directory.name, metas)
-        album = MusicBrainzChain().match_music_album(album_meta, metas)
+        regions, scripts = self._music_release_preferences(
+            list(music_release_regions) if music_release_regions is not None else None,
+            list(music_release_scripts) if music_release_scripts is not None else None,
+        )
+        album = MusicBrainzChain().match_music_album(
+            album_meta,
+            metas,
+            music_release_regions=list(regions),
+            music_release_scripts=list(scripts),
+        )
         if not album or not album.tracks:
             return {}
         return {
@@ -175,11 +208,22 @@ class MediaAlbumOwner(_MediaOwnerBase):
         self,
         directory: Path,
         files: list[Path],
+        music_release_regions: Optional[tuple[str, ...]] = None,
+        music_release_scripts: Optional[tuple[str, ...]] = None,
     ) -> dict[str, MusicInfo]:
         """异步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
         metas = await run_in_threadpool(AudioMetadataHelper.read_many, files)
         album_meta = MetaMusic.from_album_context(directory.name, metas)
-        album = await MusicBrainzChain().async_match_music_album(album_meta, metas)
+        regions, scripts = self._music_release_preferences(
+            list(music_release_regions) if music_release_regions is not None else None,
+            list(music_release_scripts) if music_release_scripts is not None else None,
+        )
+        album = await MusicBrainzChain().async_match_music_album(
+            album_meta,
+            metas,
+            music_release_regions=list(regions),
+            music_release_scripts=list(scripts),
+        )
         if not album or not album.tracks:
             return {}
         return {
@@ -190,6 +234,8 @@ class MediaAlbumOwner(_MediaOwnerBase):
     def recognize_music_album_directory(
         self,
         path: Union[str, Path],
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> dict[str, MusicInfo]:
         """按目录级线索批量识别整张专辑并返回文件到曲目的映射。"""
         directory = Path(path)
@@ -198,12 +244,21 @@ class MediaAlbumOwner(_MediaOwnerBase):
         files = self._directory_audio_files(directory)
         if len(files) < self._album_match_min_files:
             return {}
-        key = _album_directory_cache_key(directory)
+        regions, scripts = self._music_release_preferences(
+            music_release_regions,
+            music_release_scripts,
+        )
+        custom_preference = music_release_regions is not None or music_release_scripts is not None
+        key = _album_directory_cache_key(directory, regions, scripts)
         signature = self._album_directory_signature(directory, files)
         matched = self._album_dir_cache.resolve(
             key,
             signature,
-            lambda: self._match_music_album_directory(directory, files),
+            lambda: (
+                self._match_music_album_directory(directory, files, regions, scripts)
+                if custom_preference
+                else self._match_music_album_directory(directory, files)
+            ),
         )
         simplified = self._simplify_recognized_music_mapping(matched)
         finalized: dict[str, MusicInfo] = {}
@@ -217,6 +272,8 @@ class MediaAlbumOwner(_MediaOwnerBase):
     async def async_recognize_music_album_directory(
         self,
         path: Union[str, Path],
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> dict[str, MusicInfo]:
         """异步按目录级线索批量识别整张专辑。"""
         directory = Path(path)
@@ -225,12 +282,21 @@ class MediaAlbumOwner(_MediaOwnerBase):
         files = await run_in_threadpool(self._directory_audio_files, directory)
         if len(files) < self._album_match_min_files:
             return {}
-        key = _album_directory_cache_key(directory)
+        regions, scripts = self._music_release_preferences(
+            music_release_regions,
+            music_release_scripts,
+        )
+        custom_preference = music_release_regions is not None or music_release_scripts is not None
+        key = _album_directory_cache_key(directory, regions, scripts)
         signature = self._album_directory_signature(directory, files)
         matched = await self._album_dir_cache.async_resolve(
             key,
             signature,
-            lambda: self._async_match_music_album_directory(directory, files),
+            lambda: (
+                self._async_match_music_album_directory(directory, files, regions, scripts)
+                if custom_preference
+                else self._async_match_music_album_directory(directory, files)
+            ),
         )
         simplified = self._simplify_recognized_music_mapping(matched)
         return {
@@ -240,3 +306,14 @@ class MediaAlbumOwner(_MediaOwnerBase):
             )
             for item_path, info in simplified.items()
         }
+
+    @staticmethod
+    def _music_release_preferences(
+        regions: Optional[list[str]],
+        scripts: Optional[list[str]],
+    ) -> tuple[tuple[str, ...], tuple[str, ...]]:
+        """解析请求级发行偏好；未指定时继承一次稳定的系统配置快照。"""
+        runtime = get_chain_runtime_config_snapshot()
+        normalized_regions = tuple(regions) if regions is not None else runtime.music_release_region_priority
+        normalized_scripts = tuple(scripts) if scripts is not None else runtime.music_release_script_priority
+        return normalized_regions, normalized_scripts

app/chain/media/catalog.py

@@ -261,13 +261,20 @@ class MediaCatalogOwner(_MediaOwnerBase):
         self,
         media_source: MediaSource,
         media_id: str,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """按音乐来源和原生 ID 同步获取专辑详情。"""
         source, normalized_id = resolve_media_identity(media_source=media_source, media_id=media_id)
         if not source or not normalized_id:
             return None
         chain = self._music_source_chain(source)
-        result = chain.get_music_album(normalized_id) if chain else None
+        preference_kwargs = {}
+        if music_release_regions is not None:
+            preference_kwargs["music_release_regions"] = music_release_regions
+        if music_release_scripts is not None:
+            preference_kwargs["music_release_scripts"] = music_release_scripts
+        result = chain.get_music_album(normalized_id, **preference_kwargs) if chain else None
         return cast(
             Optional[MusicAlbumInfo],
             self._finalize_recognition_result(result),
@@ -277,13 +284,20 @@ class MediaCatalogOwner(_MediaOwnerBase):
         self,
         media_source: MediaSource,
         media_id: str,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """按音乐来源和原生 ID 异步获取专辑详情。"""
         source, normalized_id = resolve_media_identity(media_source=media_source, media_id=media_id)
         if not source or not normalized_id:
             return None
         chain = self._music_source_chain(source)
-        result = await chain.async_get_music_album(normalized_id) if chain else None
+        preference_kwargs = {}
+        if music_release_regions is not None:
+            preference_kwargs["music_release_regions"] = music_release_regions
+        if music_release_scripts is not None:
+            preference_kwargs["music_release_scripts"] = music_release_scripts
+        result = await chain.async_get_music_album(normalized_id, **preference_kwargs) if chain else None
         return cast(
             Optional[MusicAlbumInfo],
             await self._async_finalize_recognition_result(result),

app/chain/media/contract.py

@@ -167,6 +167,8 @@ if TYPE_CHECKING:
         def recognize_music_album_directory(
             self,
             path: Union[str, Path],
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
         ) -> dict[str, MusicInfo]:
             """同步识别音乐专辑目录。"""
             ...
@@ -174,6 +176,8 @@ if TYPE_CHECKING:
         async def async_recognize_music_album_directory(
             self,
             path: Union[str, Path],
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
         ) -> dict[str, MusicInfo]:
             """异步识别音乐专辑目录。"""
             ...

app/chain/media/facade.py

@@ -118,10 +118,14 @@ class MediaChain(ChainBase, metaclass=Singleton):
         MediaAlbumOwner._directory_audio_files.__func__  # type: ignore[attr-defined]
     )
     _album_directory_signature = staticmethod(MediaAlbumOwner._album_directory_signature)
+    _music_release_preferences = staticmethod(MediaAlbumOwner._music_release_preferences)
     _music_track_title_key = staticmethod(MediaAlbumOwner._music_track_title_key)
     _align_music_album_tracks = classmethod(  # type: ignore[var-annotated]
         MediaAlbumOwner._align_music_album_tracks.__func__  # type: ignore[attr-defined]
     )
+    _align_selected_music_album = classmethod(  # type: ignore[var-annotated]
+        MediaAlbumOwner._align_selected_music_album.__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(

app/chain/musicbrainz.py

@@ -71,27 +71,51 @@ class MusicMetadataSourceChain(ChainBase):
         )
         return self._music_info(result, media_id=normalized_id)
 
-    def get_music_album(self, media_id: str) -> Optional[MusicAlbumInfo]:
+    def get_music_album(
+        self,
+        media_id: str,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
+    ) -> Optional[MusicAlbumInfo]:
         """按当前来源原生 ID 获取专辑详情。"""
         normalized_id = self._normalize_media_id(media_id)
         if not normalized_id:
             return None
+        preference_kwargs = {}
+        if self.source == MediaSource.MusicBrainz:
+            if music_release_regions is not None:
+                preference_kwargs["music_release_regions"] = music_release_regions
+            if music_release_scripts is not None:
+                preference_kwargs["music_release_scripts"] = music_release_scripts
         result = self.run_module(
             "music_album",
             media_source=self.source,
             media_id=normalized_id,
+            **preference_kwargs,
         )
         return self._music_album(result, media_id=normalized_id)
 
-    async def async_get_music_album(self, media_id: str) -> Optional[MusicAlbumInfo]:
+    async def async_get_music_album(
+        self,
+        media_id: str,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
+    ) -> Optional[MusicAlbumInfo]:
         """异步按当前来源原生 ID 获取专辑详情。"""
         normalized_id = self._normalize_media_id(media_id)
         if not normalized_id:
             return None
+        preference_kwargs = {}
+        if self.source == MediaSource.MusicBrainz:
+            if music_release_regions is not None:
+                preference_kwargs["music_release_regions"] = music_release_regions
+            if music_release_scripts is not None:
+                preference_kwargs["music_release_scripts"] = music_release_scripts
         result = await self.async_run_module(
             "music_album",
             media_source=self.source,
             media_id=normalized_id,
+            **preference_kwargs,
         )
         return self._music_album(result, media_id=normalized_id)
 
@@ -263,6 +287,8 @@ class MusicBrainzChain(MusicMetadataSourceChain):
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int = 5,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """按目录元数据与曲目证据匹配 MusicBrainz 发行版本。"""
         result = self.run_module(
@@ -270,6 +296,8 @@ class MusicBrainzChain(MusicMetadataSourceChain):
             meta=meta,
             tracks=tracks,
             limit=limit,
+            music_release_regions=music_release_regions,
+            music_release_scripts=music_release_scripts,
         )
         return self._music_album(result)
 
@@ -278,6 +306,8 @@ class MusicBrainzChain(MusicMetadataSourceChain):
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int = 5,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """异步按目录元数据与曲目证据匹配 MusicBrainz 发行版本。"""
         result = await self.async_run_module(
@@ -285,6 +315,8 @@ class MusicBrainzChain(MusicMetadataSourceChain):
             meta=meta,
             tracks=tracks,
             limit=limit,
+            music_release_regions=music_release_regions,
+            music_release_scripts=music_release_scripts,
         )
         return self._music_album(result)
 

app/chain/transfer/contract.py

@@ -84,6 +84,7 @@ if TYPE_CHECKING:
         _is_primary_media_file: Callable[..., Any]
         _is_subtitle_file: Callable[..., Any]
         _match_music_album_context: Callable[..., Any]
+        _merge_music_track_context: Callable[..., Any]
         _music_info_from_meta: Callable[..., Any]
         _plan_checkpoint_and_execute: Callable[..., Any]
         _re_transfer: Callable[..., Any]
@@ -95,6 +96,8 @@ if TYPE_CHECKING:
         _resolve_download_history: Callable[..., Any]
         _restore_music_download_context: Callable[..., Any]
         _send_metadata_scrape_event: Callable[..., Any]
+        _selected_music_task_context: Callable[..., Any]
+        _selected_music_track_map: Callable[..., Any]
         _should_delete_empty_source_directories: Callable[..., Any]
         _transfer_storage_chain: Callable[..., Any]
         _transfer_media_chain: Callable[..., Any]

app/chain/transfer/filter.py

@@ -2,7 +2,7 @@
 import threading
 from copy import deepcopy
 from pathlib import Path
-from typing import Dict, Optional, Protocol, Union
+from typing import Any, Dict, Optional, Protocol, Union
 
 from app.application.configuration import (
     get_chain_runtime_config_snapshot,
@@ -19,7 +19,7 @@ from app.chain._contracts import TransferMixinHost
 from app.chain.media import MediaChain
 from app.chain.storage import StorageChain
 from app.chain.transfer.contract import _TransferOwnerBase
-from app.domain.context import MediaInfo, MusicInfo
+from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
 from app.domain.media import normalize_music_type
 from app.domain.meta.metamusic import MetaMusic
 from app.runtime.log import logger
@@ -166,6 +166,8 @@ class FileFilterMixin(_TransferOwnerBase):
             file_item: FileItem,
             file_path: Path,
             file_meta: MetaMusic,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> tuple[MetaMusic, Optional[MusicInfo]]:
         """为缺少远端身份的本地音频尝试目录级专辑匹配，命中后回填文件元数据。
 
@@ -176,7 +178,14 @@ class FileFilterMixin(_TransferOwnerBase):
         if file_meta.media_id or getattr(file_item, "storage", "local") != "local":
             return file_meta, None
         try:
-            matched = MediaChain().recognize_music_album_directory(file_path.parent)
+            if music_release_regions is None and music_release_scripts is None:
+                matched = MediaChain().recognize_music_album_directory(file_path.parent)
+            else:
+                matched = MediaChain().recognize_music_album_directory(
+                    file_path.parent,
+                    music_release_regions=music_release_regions,
+                    music_release_scripts=music_release_scripts,
+                )
         except Exception as err:
             logger.debug(f"音乐专辑目录匹配失败：{file_path} - {err}")
             return file_meta, None
@@ -184,6 +193,15 @@ class FileFilterMixin(_TransferOwnerBase):
         if not info or not info.media_id:
             return file_meta, None
         logger.info(f"{file_path.name} 通过专辑目录匹配识别为：{info.artist} - {info.title}")
+        return cls._merge_music_track_context(file_meta, info)
+
+    @classmethod
+    def _merge_music_track_context(
+            cls,
+            file_meta: MetaMusic,
+            info: MusicInfo,
+    ) -> tuple[MetaMusic, MusicInfo]:
+        """以已选发行版的曲目身份更新文件元数据，同时保留本地音频参数。"""
         merged_meta = deepcopy(file_meta)
         # 保留本地音频的实际技术参数，仅回填身份和名称字段
         if info.title:
@@ -226,6 +244,54 @@ class FileFilterMixin(_TransferOwnerBase):
         merged_info.raw_data = deepcopy(info.raw_data)
         return merged_meta, merged_info
 
+    def _selected_music_track_map(
+            self,
+            file_items: list[tuple[FileItem, bool]],
+            album: Optional[MusicAlbumInfo],
+    ) -> tuple[dict[str, MusicInfo], Optional[str]]:
+        """对齐手选发行版的全部曲目，并拒绝混有重复版本的目录。"""
+        if not album:
+            return {}, None
+        audio_paths = [
+            Path(item.path)
+            for item, _ in file_items
+            if item.storage == "local" and item.path and self._is_audio_file(item)
+        ]
+        if not audio_paths:
+            return {}, None
+        aligned_tracks = MediaChain._align_selected_music_album(audio_paths, album)
+        if len(aligned_tracks) != len(audio_paths):
+            return {}, (
+                f"所选专辑只能对齐 {len(aligned_tracks)} / {len(audio_paths)} "
+                "个音频文件，目录中可能包含重复版本或额外曲目；"
+                "请分别选择单个版本后再整理"
+            )
+        selected_tracks: dict[str, MusicInfo] = {}
+        for resolved_path, track in aligned_tracks.items():
+            selected_track = deepcopy(track)
+            selected_track.set_library_category(album.library_category)
+            selected_track.classification = deepcopy(album.classification)
+            selected_tracks[resolved_path] = selected_track
+        return selected_tracks, None
+
+    def _selected_music_task_context(
+            self,
+            file_item: FileItem,
+            file_path: Path,
+            file_meta: Any,
+            selected_tracks: dict[str, MusicInfo],
+            fallback: Optional[Union[MediaInfo, MusicInfo]],
+    ) -> tuple[Any, Optional[Union[MediaInfo, MusicInfo]]]:
+        """为当前任务应用手选发行版曲目，未命中时返回原上下文。"""
+        selected = (
+            selected_tracks.get(str(file_path.resolve()))
+            if file_item.storage == "local"
+            else None
+        )
+        if selected and isinstance(file_meta, MetaMusic):
+            return self._merge_music_track_context(file_meta, selected)
+        return file_meta, fallback
+
     @staticmethod
     def _download_history_music_type(
             download_history: Optional[DownloadHistorySnapshot],

app/chain/transfer/history.py

@@ -6,13 +6,14 @@ from typing import Any, List, Optional, Tuple, Union
 
 from app.chain.media import MediaChain
 from app.chain.transfer.contract import _TransferOwnerBase
-from app.domain.context import MediaInfo, MusicInfo
+from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
 from app.domain.meta.metabase import MetaBase
 from app.runtime.log import logger
 from app.schemas.message import Message
 from app.schemas.tmdb import TmdbEpisode
 from app.schemas.transfer import EpisodeFormat, TransferInfo
 from app.schemas.types import (
+    MUSIC_ENTITY_ALBUM,
     ContentType,
     MediaSource,
     MediaType,
@@ -22,6 +23,34 @@ from app.schemas.types import (
 from app.schemas.workflow import FileItem
 
 
+def _recognize_manual_media(
+    *,
+    media_source: MediaSource,
+    media_id: str,
+    mtype: Optional[MediaType],
+    music_type: Optional[str],
+    episode_group: Optional[str],
+    music_release_regions: Optional[list[str]],
+    music_release_scripts: Optional[list[str]],
+) -> Optional[Union[MediaInfo, MusicInfo, MusicAlbumInfo]]:
+    """识别手动指定的媒体，并为音乐专辑保留完整曲目表。"""
+    if mtype == MediaType.MUSIC and music_type == MUSIC_ENTITY_ALBUM:
+        album = MediaChain().get_music_album(
+            media_source=media_source,
+            media_id=media_id,
+            music_release_regions=music_release_regions,
+            music_release_scripts=music_release_scripts,
+        )
+        return album
+    return MediaChain().recognize_media(
+        media_source=media_source,
+        media_id=media_id,
+        music_type=music_type,
+        mtype=mtype,
+        episode_group=episode_group,
+    )
+
+
 class TransferHistoryOwner(_TransferOwnerBase):
     """唯一持有手动历史、重整命令和通知公开入口。"""
 
@@ -140,6 +169,8 @@ class TransferHistoryOwner(_TransferOwnerBase):
             cleanup_dest_fileitem: Optional[FileItem] = None,
             reorganize: Optional[bool] = False,
             music_type: Optional[str] = None,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Tuple[bool, Union[str, dict[str, Any]]]:
         """
         手动整理，支持复杂条件，带进度显示
@@ -166,6 +197,8 @@ class TransferHistoryOwner(_TransferOwnerBase):
         :param sync_extra_files: 是否同步整理同媒体附加文件
         :param cleanup_dest_fileitem: 确认存在待整理任务后需要清理的旧目标文件
         :param music_type: 音乐实体类型；为保持位置参数兼容，必须追加在签名末尾
+        :param music_release_regions: 本次音乐整理的发行地区优先级，空值继承系统设置
+        :param music_release_scripts: 本次音乐整理的文字字形优先级，空值继承系统设置
         """
         logger.info(f"手动整理：{fileitem.path} ...")
         explicit_identity = media_source is not None or media_id is not None
@@ -173,21 +206,23 @@ class TransferHistoryOwner(_TransferOwnerBase):
             return False, "手动整理需要同时提供 media_source 和 media_id"
         if media_source and media_id:
             # 有输入媒体ID时预先识别，音乐与影视统一走 recognize_media 按类型分发
-            mediainfo = MediaChain().recognize_media(
+            mediainfo = _recognize_manual_media(
                 media_source=media_source,
                 media_id=media_id,
-                music_type=music_type,
                 mtype=mtype,
+                music_type=music_type,
                 episode_group=episode_group,
+                music_release_regions=music_release_regions,
+                music_release_scripts=music_release_scripts,
             )
             if not mediainfo:
                 return (
                     False,
                     "未识别到媒体信息，请检查媒体来源和媒体 ID 后重试",
                 )
-            if media_source and not isinstance(mediainfo, MusicInfo):
+            if media_source and not isinstance(mediainfo, (MusicInfo, MusicAlbumInfo)):
                 mediainfo.scrape_source = media_source
-            if not isinstance(mediainfo, MusicInfo):
+            if not isinstance(mediainfo, (MusicInfo, MusicAlbumInfo)):
                 self.obtain_images(mediainfo=mediainfo)
 
             # 开始整理
@@ -215,6 +250,8 @@ class TransferHistoryOwner(_TransferOwnerBase):
                 reorganize=reorganize,
                 sync_extra_files=sync_extra_files,
                 cleanup_dest_fileitem=cleanup_dest_fileitem,
+                music_release_regions=music_release_regions,
+                music_release_scripts=music_release_scripts,
             )
             if not state:
                 return False, errmsg
@@ -245,6 +282,8 @@ class TransferHistoryOwner(_TransferOwnerBase):
                 reorganize=reorganize,
                 sync_extra_files=sync_extra_files,
                 cleanup_dest_fileitem=cleanup_dest_fileitem,
+                music_release_regions=music_release_regions,
+                music_release_scripts=music_release_scripts,
             )
             return state, errmsg
 

app/chain/transfer/request.py

@@ -15,11 +15,32 @@ from app.domain.meta.metabase import MetaBase
 from app.domain.metainfo import MetaInfoPath
 from app.schemas.exception import OperationInterrupted
 from app.schemas.types import (
+    MUSIC_ENTITY_RECORDING,
+    MediaSource,
     MediaType,
 )
 from app.schemas.workflow import FileItem
 
 
+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)
+    )
+
+
 class _TransferCandidatePlanner:
     """持有一次整理请求的只读候选规划上下文。"""
 

app/chain/transfer/workflow.py

@@ -25,7 +25,7 @@ from app.application.transfer.workflow import (
 from app.chain.media import MediaChain
 from app.chain.storage import StorageChain
 from app.chain.transfer.contract import _TransferOwnerBase
-from app.domain.context import MediaInfo, MusicInfo, TorrentInfo
+from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo, TorrentInfo
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
 from app.runtime.log import logger
@@ -36,7 +36,6 @@ from app.schemas.media import resolve_media_identity
 from app.schemas.system import TransferDirectoryConf
 from app.schemas.transfer import EpisodeFormat, TransferInfo
 from app.schemas.types import (
-    MUSIC_ENTITY_RECORDING,
     MediaSource,
     MediaType,
     ProgressKey,
@@ -44,26 +43,54 @@ from app.schemas.types import (
 )
 from app.schemas.workflow import FileItem
 
-from .request import _TransferCandidatePlanner
+from .request import (
+    _should_discard_batch_recording_identity,
+    _TransferCandidatePlanner,
+)
 
 
-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 _normalize_transfer_identity(
+    mediainfo: Optional[Union[MediaInfo, MusicInfo]],
+    mtype: Optional[MediaType],
+    media_source: Optional[MediaSource],
+    media_id: Optional[str],
+    meta: Optional[MetaBase],
+) -> tuple[
+    Optional[Union[MediaInfo, MusicInfo]],
+    Optional[MediaSource],
+    Optional[str],
+    Optional[str],
+]:
+    """规范整理请求的媒体身份，并在显式身份缺失时短路。"""
+    explicit_identity = media_source is not None or media_id is not None
+    normalized_source, normalized_media_id = resolve_media_identity(
+        media_source=media_source,
+        media_id=media_id,
     )
+    if explicit_identity and (not normalized_source or not normalized_media_id):
+        return (
+            mediainfo,
+            normalized_source,
+            normalized_media_id,
+            "整理任务需要同时提供有效的 media_source 和 media_id",
+        )
+    if not explicit_identity and mediainfo:
+        normalized_source, normalized_media_id = resolve_media_identity(media=mediainfo)
+    if explicit_identity and not mediainfo:
+        mediainfo = MediaChain().recognize_media(
+            mtype=mtype,
+            media_source=normalized_source,
+            media_id=normalized_media_id,
+            music_type=getattr(meta, "music_type", None),
+        )
+        if not mediainfo:
+            return (
+                mediainfo,
+                normalized_source,
+                normalized_media_id,
+                "未识别到媒体信息，请检查媒体来源和媒体 ID 后重试",
+            )
+    return mediainfo, normalized_source, normalized_media_id, None
 
 
 class TransferWorkflowOwner(_TransferOwnerBase):
@@ -241,54 +268,6 @@ class TransferWorkflowOwner(_TransferOwnerBase):
 
         return shared_roots
 
-    @staticmethod
-    def _normalize_transfer_identity(
-        mediainfo: Optional[Union[MediaInfo, MusicInfo]],
-        mtype: Optional[MediaType],
-        media_source: Optional[MediaSource],
-        media_id: Optional[str],
-        meta: Optional[MetaBase],
-    ) -> Tuple[
-        Optional[Union[MediaInfo, MusicInfo]],
-        Optional[MediaSource],
-        Optional[str],
-        Optional[str],
-    ]:
-        """
-        规范整理请求的媒体身份，并在显式身份缺失时短路。
-
-        :return: ``(媒体信息、媒体来源、媒体 ID、错误信息)``；错误信息为空表示可继续执行
-        """
-        explicit_identity = media_source is not None or media_id is not None
-        normalized_source, normalized_media_id = resolve_media_identity(
-            media_source=media_source,
-            media_id=media_id,
-        )
-        if explicit_identity and (not normalized_source or not normalized_media_id):
-            return (
-                mediainfo,
-                normalized_source,
-                normalized_media_id,
-                "整理任务需要同时提供有效的 media_source 和 media_id",
-            )
-        if not explicit_identity and mediainfo:
-            normalized_source, normalized_media_id = resolve_media_identity(media=mediainfo)
-        if explicit_identity and not mediainfo:
-            mediainfo = MediaChain().recognize_media(
-                mtype=mtype,
-                media_source=normalized_source,
-                media_id=normalized_media_id,
-                music_type=getattr(meta, "music_type", None),
-            )
-            if not mediainfo:
-                return (
-                    mediainfo,
-                    normalized_source,
-                    normalized_media_id,
-                    "未识别到媒体信息，请检查媒体来源和媒体 ID 后重试",
-                )
-        return mediainfo, normalized_source, normalized_media_id, None
-
     def _collect_transfer_candidates(
         self,
         fileitem: FileItem,
@@ -359,7 +338,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         self,
         fileitem: FileItem,
         meta: MetaBase = None,
-        mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None,
+        mediainfo: Optional[Union[MediaInfo, MusicInfo, MusicAlbumInfo]] = None,
         mtype: Optional[MediaType] = None,
         media_source: Optional[MediaSource] = None,
         media_id: Optional[str] = None,
@@ -383,6 +362,8 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         cleanup_dest_fileitem: Optional[FileItem] = None,
         continue_callback: Callable = None,
         reorganize: Optional[bool] = False,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> Tuple[bool, Union[str, dict]]:
         """
         兼容公开整理入口，委托给内部批次执行阶段。
@@ -417,6 +398,8 @@ class TransferWorkflowOwner(_TransferOwnerBase):
             cleanup_dest_fileitem=cleanup_dest_fileitem,
             continue_callback=continue_callback,
             reorganize=reorganize,
+            music_release_regions=music_release_regions,
+            music_release_scripts=music_release_scripts,
         )
 
     def _execute_transfer(self, *args: Any, **kwargs: Any) -> Tuple[bool, Union[str, dict]]:
@@ -427,7 +410,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         self,
         fileitem: FileItem,
         meta: MetaBase = None,
-        mediainfo: Optional[Union[MediaInfo, MusicInfo]] = None,
+        mediainfo: Optional[Union[MediaInfo, MusicInfo, MusicAlbumInfo]] = None,
         mtype: Optional[MediaType] = None,
         media_source: Optional[MediaSource] = None,
         media_id: Optional[str] = None,
@@ -452,6 +435,8 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         continue_callback: Callable = None,
         reorganize: Optional[bool] = False,
         recovery_admission: Optional[TransferAdmission] = None,
+        music_release_regions: Optional[list[str]] = None,
+        music_release_scripts: Optional[list[str]] = None,
     ) -> Tuple[bool, Union[str, dict]]:
         """
         执行一个复杂目录的整理操作
@@ -482,9 +467,17 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         :param cleanup_dest_fileitem: 确认存在待整理任务后需要清理的旧目标文件
         :param continue_callback: 继续处理回调
         :param recovery_admission: 内部恢复调用绑定的既有 durable 记录
+        :param music_release_regions: 本次音乐整理的发行地区优先级
+        :param music_release_scripts: 本次音乐整理的文字字形优先级
         返回：成功标识，错误信息
         """
-        mediainfo, media_source, media_id, identity_error = self._normalize_transfer_identity(
+        selected_music_album: Optional[MusicAlbumInfo]
+        if isinstance(mediainfo, MusicAlbumInfo):
+            selected_music_album = mediainfo
+            mediainfo = mediainfo.to_music_info()
+        else:
+            selected_music_album = None
+        mediainfo, media_source, media_id, identity_error = _normalize_transfer_identity(
             mediainfo=mediainfo,
             mtype=mtype,
             media_source=media_source,
@@ -562,6 +555,12 @@ class TransferWorkflowOwner(_TransferOwnerBase):
 
         file_items, inherited_meta_map = candidate_planner._plan_file_items(file_items)
 
+        selected_music_track_map, selected_music_error = self._selected_music_track_map(
+            file_items, selected_music_album
+        )
+        if selected_music_error:
+            return False, selected_music_error
+
         planned_file_count = len(file_items)
 
         if preview:
@@ -603,6 +602,9 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 continue_callback=continue_callback,
                 cleanup_dest_fileitem=cleanup_dest_fileitem,
                 recovery_admission=recovery_admission,
+                music_release_regions=music_release_regions,
+                music_release_scripts=music_release_scripts,
+                selected_music_track_map=selected_music_track_map,
             )
         except OperationInterrupted:
             return False, f"{fileitem.name} 已取消"
@@ -664,6 +666,9 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         continue_callback: Optional[Callable[[], bool]],
         cleanup_dest_fileitem: Optional[FileItem],
         recovery_admission: Optional[TransferAdmission],
+        music_release_regions: Optional[list[str]],
+        music_release_scripts: Optional[list[str]],
+        selected_music_track_map: dict[str, MusicInfo],
     ) -> Tuple[List[TransferTask], bool, List[str], int, set[Tuple[str, str]]]:
         """从冻结候选构建任务，并集中执行历史去重与 durable 绑定。"""
         _build_file_meta = build_file_meta
@@ -817,14 +822,15 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     _download_hash = download_hash
 
                 # 自动整理预载的媒体信息来自整条下载历史；电影合集内文件年份冲突时逐文件识别。
-                task_mediainfo = (
-                    None
-                    if discard_recording_identity
-                    else mediainfo or history_music_info
+                file_meta, task_mediainfo = self._selected_music_task_context(
+                    file_item, file_path, file_meta, selected_music_track_map,
+                    None if discard_recording_identity else mediainfo or history_music_info,
                 )
                 if not task_mediainfo and isinstance(file_meta, MetaMusic):
                     # 无标签音频或误带单曲身份的整包按目录级专辑匹配；命中结果带缓存不会逐文件重复请求
-                    file_meta, task_mediainfo = self._match_music_album_context(file_item, file_path, file_meta)
+                    file_meta, task_mediainfo = self._match_music_album_context(
+                        file_item, file_path, file_meta, music_release_regions, music_release_scripts,
+                    )
                     if not task_mediainfo and discard_recording_identity:
                         task_mediainfo = self._music_info_from_meta(file_meta)
                 if not manual and task_mediainfo and self._is_movie_year_conflict(file_meta, task_mediainfo):

app/modules/musicbrainz/__init__.py

@@ -92,6 +92,14 @@ class _MusicBrainzRecognitionPlan:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class _MusicReleasePreference:
+    """一次 MusicBrainz 发行版本选择使用的地区与文字字形顺序。"""
+
+    regions: tuple[str, ...]
+    scripts: tuple[str, ...]
+
+
 @dataclass(frozen=True, slots=True)
 class _MusicBrainzResponseDecision:
     """描述一次 MusicBrainz 响应的投影结果与退避决策。"""
@@ -157,6 +165,66 @@ class MusicBrainzModule(_ModuleBase):
         "purchase for download",
     )
 
+    @staticmethod
+    def _preference_values(
+            values: Optional[Iterable[str]],
+            setting_name: str,
+            *,
+            upper: bool,
+    ) -> tuple[str, ...]:
+        """规范请求级代码；未传入时读取系统默认并稳定去重。"""
+        configured: Iterable[str]
+        if values is None:
+            configured = str(get_runtime_setting(setting_name, "") or "").split(",")
+        else:
+            configured = values
+        normalized: list[str] = []
+        for value in configured:
+            code = str(value).strip()
+            code = code.upper() if upper else code.title()
+            if code and code not in normalized:
+                normalized.append(code)
+        return tuple(normalized[:3])
+
+    @classmethod
+    def _release_preference(
+            cls,
+            regions: Optional[Iterable[str]] = None,
+            scripts: Optional[Iterable[str]] = None,
+    ) -> _MusicReleasePreference:
+        """返回本次发行选择的显式偏好或部署默认值。"""
+        return _MusicReleasePreference(
+            regions=cls._preference_values(
+                regions,
+                "MUSIC_RELEASE_REGION_PRIORITY",
+                upper=True,
+            ),
+            scripts=cls._preference_values(
+                scripts,
+                "MUSIC_RELEASE_SCRIPT_PRIORITY",
+                upper=False,
+            ),
+        )
+
+    @staticmethod
+    def _release_script(release: dict[str, Any]) -> str:
+        """提取 MusicBrainz Release 的 ISO 15924 文字字形。"""
+        representation = release.get("text-representation") or {}
+        return str(representation.get("script") or "").strip().title()
+
+    @classmethod
+    def _release_preference_sort_key(
+            cls,
+            release: dict[str, Any],
+            preference: _MusicReleasePreference,
+    ) -> tuple[int, int]:
+        """生成地区与字形的稳定升序键；未匹配项排在显式偏好之后。"""
+        country = str(release.get("country") or "").strip().upper()
+        script = cls._release_script(release)
+        region_rank = preference.regions.index(country) if country in preference.regions else len(preference.regions)
+        script_rank = preference.scripts.index(script) if script in preference.scripts else len(preference.scripts)
+        return region_rank, script_rank
+
     def init_module(self) -> None:
         """初始化 MusicBrainz 模块并挂载本地识别缓存。"""
         self.cache = MusicBrainzCache()
@@ -616,6 +684,8 @@ class MusicBrainzModule(_ModuleBase):
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int = 5,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """按目录线索和曲目特征把本地音频集合对位到 MusicBrainz 发行版本。
 
@@ -624,29 +694,45 @@ class MusicBrainzModule(_ModuleBase):
         """
         if not tracks:
             return None
+        preference = self._release_preference(
+            music_release_regions,
+            music_release_scripts,
+        )
         details: list[dict[str, Any]] = []
-        releases = self._search_release_candidates(meta, tracks, limit=limit)
+        releases = self._search_release_candidates(
+            meta,
+            tracks,
+            limit=limit,
+            preference=preference,
+        )
         for request in self._release_detail_requests(releases):
             detail = self._request_json(request.path, params=request.params)
             if not detail:
                 continue
             details.append(detail)
-        return self._select_release_match(meta, tracks, details)
+        return self._select_release_match(meta, tracks, details, preference)
 
     async def async_match_music_album(
             self,
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int = 5,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """异步按目录线索和曲目特征匹配 MusicBrainz 发行版本。"""
         if not tracks:
             return None
+        preference = self._release_preference(
+            music_release_regions,
+            music_release_scripts,
+        )
         details: list[dict[str, Any]] = []
         releases = await self._async_search_release_candidates(
             meta,
             tracks,
             limit=limit,
+            preference=preference,
         )
         for request in self._release_detail_requests(releases):
             detail = await self._async_request_json(
@@ -655,7 +741,7 @@ class MusicBrainzModule(_ModuleBase):
             if not detail:
                 continue
             details.append(detail)
-        return self._select_release_match(meta, tracks, details)
+        return self._select_release_match(meta, tracks, details, preference)
 
     @classmethod
     def _select_release_match(
@@ -663,15 +749,24 @@ class MusicBrainzModule(_ModuleBase):
             meta: MetaMusic,
             tracks: list[MetaMusic],
             details: Iterable[dict[str, Any]],
+            preference: Optional[_MusicReleasePreference] = None,
     ) -> Optional[MusicAlbumInfo]:
         """对已获取的发行详情统一打分并投影最佳专辑。"""
         best_album: Optional[MusicAlbumInfo] = None
         best_score = 0.0
+        best_key: Optional[tuple[float, int, int]] = None
+        selected_preference = preference or cls._release_preference()
         for detail in details:
             summary = cls._release_track_summary(detail)
             score = cls._score_release(meta, tracks, detail, summary)
-            if score > best_score:
+            region_rank, script_rank = cls._release_preference_sort_key(
+                detail,
+                selected_preference,
+            )
+            candidate_key = (score, -region_rank, -script_rank)
+            if best_key is None or candidate_key > best_key:
                 best_score = score
+                best_key = candidate_key
                 best_album = cls._release_to_album(detail)
         # 得分低于阈值时宁可不匹配，避免把曲目写到错误的专辑上
         return best_album if best_score >= cls._album_match_threshold else None
@@ -683,36 +778,68 @@ class MusicBrainzModule(_ModuleBase):
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int,
+            preference: Optional[_MusicReleasePreference] = None,
     ) -> list[dict[str, Any]]:
         """按专辑名和曲名线索搜索候选发行版本，多个查询按命中顺序去重。"""
         releases: list[dict[str, Any]] = []
         seen: set[str] = set()
-        for request in self._release_search_requests(meta, tracks, limit):
+        search_limit = max(25, min(100, limit * 5))
+        for request in self._release_search_requests(meta, tracks, search_limit):
             payload = self._request_json(
                 request.path, params=request.params
             )
             self._merge_release_candidates(releases, seen, payload)
-            if len(releases) >= limit:
+            if len(releases) >= search_limit:
                 break
-        return releases[:limit]
+        return self._rank_release_candidates(
+            releases,
+            preference or self._release_preference(),
+        )[:limit]
 
     async def _async_search_release_candidates(
             self,
             meta: MetaMusic,
             tracks: list[MetaMusic],
             limit: int,
+            preference: Optional[_MusicReleasePreference] = None,
     ) -> list[dict[str, Any]]:
         """异步按专辑名和曲名线索搜索并去重候选发行版本。"""
         releases: list[dict[str, Any]] = []
         seen: set[str] = set()
-        for request in self._release_search_requests(meta, tracks, limit):
+        search_limit = max(25, min(100, limit * 5))
+        for request in self._release_search_requests(meta, tracks, search_limit):
             payload = await self._async_request_json(
                 request.path, params=request.params
             )
             self._merge_release_candidates(releases, seen, payload)
-            if len(releases) >= limit:
+            if len(releases) >= search_limit:
                 break
-        return releases[:limit]
+        return self._rank_release_candidates(
+            releases,
+            preference or self._release_preference(),
+        )[:limit]
+
+    @classmethod
+    def _rank_release_candidates(
+            cls,
+            releases: list[dict[str, Any]],
+            preference: _MusicReleasePreference,
+    ) -> list[dict[str, Any]]:
+        """在相近搜索相关度内优先目标地区与字形，保留强相关性边界。"""
+
+        def sort_key(release: dict[str, Any]) -> tuple[int, int, int, int]:
+            """相关度每五分成组，组内应用发行偏好并保留原始得分。"""
+            try:
+                score = int(release.get("score") or 0)
+            except TypeError, ValueError:
+                score = 0
+            region_rank, script_rank = cls._release_preference_sort_key(
+                release,
+                preference,
+            )
+            return -(score // 5), region_rank, script_rank, -score
+
+        return sorted(releases, key=sort_key)
 
     @staticmethod
     def _merge_release_candidates(
@@ -943,7 +1070,11 @@ class MusicBrainzModule(_ModuleBase):
             cover_url=cls._build_cover_url(group_id),
             genres=cls._names_of(detail.get("genres")),
             detail_link=f"https://musicbrainz.org/release/{release_id}",
-            raw_data={"release_id": str(release_id)},
+            raw_data={
+                "release_id": str(release_id),
+                "release_country": cls._stripped(detail.get("country")),
+                "release_script": cls._release_script(detail) or None,
+            },
         )
         album.tracks = [
             info
@@ -1521,6 +1652,8 @@ class MusicBrainzModule(_ModuleBase):
             self,
             media_source: MediaSource,
             media_id: str,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """异步按 MusicBrainz Release Group ID 获取专辑详情及曲目。"""
         if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
@@ -1536,8 +1669,13 @@ class MusicBrainzModule(_ModuleBase):
         if not album:
             return None
         tracks_payload = await self._async_album_tracks_payload(
-            payload.get("releases") or []
+            payload.get("releases") or [],
+            preference=self._release_preference(
+                music_release_regions,
+                music_release_scripts,
+            ),
         )
+        self._apply_selected_release(album, tracks_payload)
         album.tracks = self._project_album_tracks(album, tracks_payload)
         album.artist_aliases = await self._async_lookup_artist_aliases(album.artist_ids, album.artist_aliases)
         return album
@@ -1546,6 +1684,8 @@ class MusicBrainzModule(_ModuleBase):
             self,
             media_source: MediaSource,
             media_id: str,
+            music_release_regions: Optional[list[str]] = None,
+            music_release_scripts: Optional[list[str]] = None,
     ) -> Optional[MusicAlbumInfo]:
         """按 MusicBrainz Release Group ID 获取标准化专辑详情及曲目。"""
         if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
@@ -1560,7 +1700,14 @@ class MusicBrainzModule(_ModuleBase):
         album = self._project_album_detail(payload)
         if not album:
             return None
-        tracks_payload = self._album_tracks_payload(payload.get("releases") or [])
+        tracks_payload = self._album_tracks_payload(
+            payload.get("releases") or [],
+            preference=self._release_preference(
+                music_release_regions,
+                music_release_scripts,
+            ),
+        )
+        self._apply_selected_release(album, tracks_payload)
         album.tracks = self._project_album_tracks(album, tracks_payload)
         album.artist_aliases = self._lookup_artist_aliases(album.artist_ids, album.artist_aliases)
         return album
@@ -1868,8 +2015,12 @@ class MusicBrainzModule(_ModuleBase):
         return [str(item["name"]) for item in entries]
 
     @classmethod
-    def _select_track_release(cls, releases: list[dict[str, Any]]) -> dict[str, Any]:
-        """选择曲目最完整且发行最早的正式版本，作为专辑曲目来源。"""
+    def _select_track_release(
+            cls,
+            releases: list[dict[str, Any]],
+            preference: Optional[_MusicReleasePreference] = None,
+    ) -> dict[str, Any]:
+        """选择曲目完整的正式版本，并按地区、字形和日期确定代表版本。"""
         candidates = [
             release
             for release in releases
@@ -1877,19 +2028,27 @@ class MusicBrainzModule(_ModuleBase):
             and sum(int(item.get("track-count") or 0) for item in release.get("media") or [])
         ]
         if not candidates:
-            return next((release for release in releases if release.get("id")), {})
+            candidates = [release for release in releases if release.get("id")]
+        if not candidates:
+            return {}
         official = [release for release in candidates if release.get("status") == "Official"]
+        selected_preference = preference or cls._release_preference()
         return min(
             official or candidates,
-            key=lambda release: cls._date_sort_key(release.get("date")),
+            key=lambda release: (
+                *cls._release_preference_sort_key(release, selected_preference),
+                cls._date_sort_key(release.get("date")),
+            ),
         )
 
     @classmethod
     def _album_tracks_payload(
-            cls, releases: list[dict[str, Any]]
+            cls,
+            releases: list[dict[str, Any]],
+            preference: Optional[_MusicReleasePreference] = None,
     ) -> Optional[dict[str, Any]]:
         """同步读取专辑代表性发行版本的原始曲目响应。"""
-        release = cls._select_track_release(releases)
+        release = cls._select_track_release(releases, preference)
         if not release.get("id"):
             return None
         payload = cls._request_json(
@@ -1900,10 +2059,12 @@ class MusicBrainzModule(_ModuleBase):
 
     @classmethod
     async def _async_album_tracks_payload(
-            cls, releases: list[dict[str, Any]]
+            cls,
+            releases: list[dict[str, Any]],
+            preference: Optional[_MusicReleasePreference] = None,
     ) -> Optional[dict[str, Any]]:
         """异步读取专辑代表性发行版本的原始曲目响应。"""
-        release = cls._select_track_release(releases)
+        release = cls._select_track_release(releases, preference)
         if not release.get("id"):
             return None
         payload = await cls._async_request_json(
@@ -1912,6 +2073,26 @@ class MusicBrainzModule(_ModuleBase):
         )
         return payload if isinstance(payload, dict) else None
 
+    @classmethod
+    def _apply_selected_release(
+            cls,
+            album: MusicAlbumInfo,
+            payload: Optional[dict[str, Any]],
+    ) -> None:
+        """把代表 Release 的标题、日期和来源身份回填到 Release Group 专辑。"""
+        if not payload:
+            return
+        if title := cls._stripped(payload.get("title")):
+            album.title = title
+        if release_date := cls._stripped(payload.get("date")):
+            album.release_date = release_date
+        album.raw_data = {
+            **(album.raw_data or {}),
+            "release_id": cls._stripped(payload.get("id")),
+            "release_country": cls._stripped(payload.get("country")),
+            "release_script": cls._release_script(payload) or None,
+        }
+
     @classmethod
     def _project_album_tracks(
             cls,
@@ -1973,14 +2154,18 @@ class MusicBrainzModule(_ModuleBase):
 
     @classmethod
     def _select_release(cls, releases: list[dict[str, Any]]) -> dict[str, Any]:
-        """优先选择正式且日期最早的发行记录。"""
+        """优先选择正式且符合系统地区、字形优先级的发行记录。"""
         if not releases:
             return {}
         official = [release for release in releases if release.get("status") == "Official"]
         candidates = official or releases
+        preference = cls._release_preference()
         return min(
             candidates,
-            key=lambda release: cls._date_sort_key(release.get("date")),
+            key=lambda release: (
+                *cls._release_preference_sort_key(release, preference),
+                cls._date_sort_key(release.get("date")),
+            ),
         )
 
     @staticmethod

app/runtime/config.py

@@ -396,6 +396,10 @@ class ConfigModel(BaseModel):
     ACOUSTID_API_KEY: str = "b1auxfOzAg"
     # 是否将识别到的音乐标题、艺术家、专辑等标准元数据转换为简体中文
     MUSIC_METADATA_TO_SIMPLIFIED: bool = True
+    # MusicBrainz 发行地区优先级，按 ISO 3166-1 两位代码从高到低排列
+    MUSIC_RELEASE_REGION_PRIORITY: str = "CN,TW,HK"
+    # MusicBrainz 发行文字字形优先级，按 ISO 15924 四位代码从高到低排列
+    MUSIC_RELEASE_SCRIPT_PRIORITY: str = "Hans,Hant,Latn"
     # TheAudioDB API Key，默认使用官方公开的免费 V1 Key，可通过环境变量覆盖
     THEAUDIODB_API_KEY: str = "123"
     # LRCLIB 服务地址，可指向兼容官方 API 的自建实例

app/schemas/transfer.py

@@ -276,6 +276,10 @@ class ManualTransferItem(OptionalMediaIdentityMixin, BaseModel):
     media_id: Optional[str] = None
     # 音乐实体类型
     music_type: Optional[MusicTargetEntityType] = None
+    # 本次手动整理的 MusicBrainz 发行地区优先级；空值继承系统设置
+    music_release_regions: Optional[List[str]] = Field(default=None, max_length=3)
+    # 本次手动整理的 MusicBrainz 文字字形优先级；空值继承系统设置
+    music_release_scripts: Optional[List[str]] = Field(default=None, max_length=3)
     # 类型
     type_name: Optional[str] = None
     # 季号
@@ -307,6 +311,25 @@ class ManualTransferItem(OptionalMediaIdentityMixin, BaseModel):
     # 重新整理，清理命中的成功历史及其旧目标
     reorganize: Optional[bool] = False
 
+    @model_validator(mode="after")  # type: ignore[misc]
+    def normalize_music_release_preferences(self) -> "ManualTransferItem":
+        """规范手动发行偏好并拒绝重复或非法 ISO 代码。"""
+        if self.music_release_regions is not None:
+            regions = [str(value).strip().upper() for value in self.music_release_regions]
+            if any(len(value) != 2 or not value.isascii() or not value.isalpha() for value in regions):
+                raise ValueError("音乐发行地区必须使用两位代码")
+            if len(set(regions)) != len(regions):
+                raise ValueError("音乐发行地区优先级不能重复")
+            self.music_release_regions = regions
+        if self.music_release_scripts is not None:
+            scripts = [str(value).strip().title() for value in self.music_release_scripts]
+            if any(len(value) != 4 or not value.isascii() or not value.isalpha() for value in scripts):
+                raise ValueError("音乐文字字形必须使用四位 ISO 15924 代码")
+            if len(set(scripts)) != len(scripts):
+                raise ValueError("音乐文字字形优先级不能重复")
+            self.music_release_scripts = scripts
+        return self
+
 
 class ManualTransferHistoryInfo(BaseModel):
     """

app/startup/composition/configuration.py

@@ -232,6 +232,12 @@ def build_chain_runtime_config(settings: Settings) -> ChainRuntimeConfig:
         data_cleanup_outbox_dead_days=settings.DATA_CLEANUP_OUTBOX_DEAD_DAYS,
         download_subtitle=settings.DOWNLOAD_SUBTITLE,
         music_metadata_to_simplified=settings.MUSIC_METADATA_TO_SIMPLIFIED,
+        music_release_region_priority=tuple(
+            item.strip().upper() for item in settings.MUSIC_RELEASE_REGION_PRIORITY.split(",") if item.strip()
+        ),
+        music_release_script_priority=tuple(
+            item.strip().title() for item in settings.MUSIC_RELEASE_SCRIPT_PRIORITY.split(",") if item.strip()
+        ),
         recognize_plugin_first=settings.RECOGNIZE_PLUGIN_FIRST,
         ai_agent_enable=settings.AI_AGENT_ENABLE,
         ai_agent_global=settings.AI_AGENT_GLOBAL,

docs/cli.md

@@ -468,6 +468,8 @@ moviepilot config set LYRICS_BATCH_TIMEOUT 120
 moviepilot config set LYRICS_PROVIDER_RETRY_MAX_WAIT 5
 moviepilot config set MUSIXMATCH_API_KEY your-authorized-api-key
 moviepilot config set MUSIC_METADATA_TO_SIMPLIFIED true
+moviepilot config set MUSIC_RELEASE_REGION_PRIORITY CN,TW,HK
+moviepilot config set MUSIC_RELEASE_SCRIPT_PRIORITY Hans,Hant,Latn
 ```
 
 查看所有可配置项：
@@ -489,6 +491,7 @@ moviepilot config describe API_TOKEN --show-secrets
 - `LRCLIB_BASE_URL` 默认使用官方实例，也可指向兼容 LRCLIB API 的自建实例；`LYRICS_BATCH_TIMEOUT` 限制单次专辑刮削的在线歌词总预算，`LYRICS_PROVIDER_RETRY_MAX_WAIT` 决定长 `Retry-After` 进入来源冷却而非阻塞批次
 - `THEAUDIODB_API_KEY` 用于音乐元数据及纯文本歌词兜底，默认 `123` 为官方公开 V1 Key；`MUSIXMATCH_API_KEY` 留空时不加载 Musixmatch，配置后只调用官方或 `MUSIXMATCH_BASE_URL` 指定的授权代理，使用者必须遵守对应账户的歌词存储和展示授权
 - `MUSIC_METADATA_TO_SIMPLIFIED` 默认开启；开启后会将识别结果中的曲名、艺术家、专辑和分类等标准音乐元数据转换为简体中文，不转换歌词与来源原始响应
+- `MUSIC_RELEASE_REGION_PRIORITY` 与 `MUSIC_RELEASE_SCRIPT_PRIORITY` 分别使用逗号分隔的 ISO 3166-1 地区代码和 ISO 15924 文字代码；仅在 MusicBrainz 候选结构同样可信时作为发行版本排序依据，最多使用前三项
 - `config describe` 显示单个配置项的类型、默认值和当前值
 
 ## 数据库备份命令

docs/mcp-api.md

@@ -280,7 +280,7 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返
 | POST | `/api/v1/media/scrape/{storage}` | 刮削媒体元数据；请求体为 `FileItem`，可选查询参数 `media_source`、`media_id`、`type_name`（电影/电视剧/音乐）。音乐会按策略处理音频标签、封面和歌词 |
 | POST | `/api/v1/transfer/manual/target-path` | 按源文件与目录配置匹配手动整理目标路径；请求体为 `ManualTransferItem`，该接口不执行媒体识别 |
 | POST | `/api/v1/transfer/manual/history` | 查询文件、批量文件或目录命中的成功整理历史摘要，用于进入手动整理界面时显示重新整理状态 |
-| POST | `/api/v1/transfer/manual` | 手动整理；请求体可用 `media_source` + `media_id` 指定本次识别与刮削数据源；音乐请求未传 `music_type` 时，目录按 `album`、文件按 `recording` 解释；命中持久失败历史，且未指定媒体身份、未开启 `reorganize` 时，由调度器重试原计划（包括 `logid` 历史入口）；显式重整先校验并放弃确定失败任务，再清理旧目标和记录；旧版失败历史仍清理后重试；`reorganize=true` 时清理命中的成功历史和非移动模式旧目标后重新整理 |
+| POST | `/api/v1/transfer/manual` | 手动整理；请求体可用 `media_source` + `media_id` 指定本次识别与刮削数据源；音乐请求未传 `music_type` 时，目录按 `album`、文件按 `recording` 解释；可用最多三项的 `music_release_regions`（ISO 3166-1）和 `music_release_scripts`（ISO 15924）仅覆盖本次 MusicBrainz 发行版本排序，省略时继承系统设置；命中持久失败历史，且未指定媒体身份、未开启 `reorganize` 时，由调度器重试原计划（包括 `logid` 历史入口）；显式重整先校验并放弃确定失败任务，再清理旧目标和记录；旧版失败历史仍清理后重试；`reorganize=true` 时清理命中的成功历史和非移动模式旧目标后重新整理 |
 | GET | `/api/v1/transfer/tasks/manual-reviews` | 管理员分页查询 durable 人工复核任务；`state` 仅允许 `manual_review`（默认）或已经人工判定、等待调度恢复的 `retry_wait`，支持 `page` 与 `page_size`。响应只公开任务、源文件、状态、步骤意图/证据/错误和复核修订号，不返回 lease 或 attempt 身份 |
 | GET | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员查询单个 durable 人工复核任务详情；仅可读取 `manual_review` 或已经人工判定的 `retry_wait` 任务，其余状态按不存在处理 |
 | POST | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员判定处于 `manual_review` 的 durable 整理步骤；请求包含 `operation_id`、`decision=not_applied|applied`、`reason`，`applied` 还必须提供 `result_payload`。`failed` 不属于公开决策，失败终态只能由持租约的 durable 结算写入；响应仅返回任务、操作、决策、后续状态和复核修订号 |

tests/test_manual_transfer_history.py

@@ -900,3 +900,29 @@ def test_forced_manual_reorganize_still_removes_history(monkeypatch):
         ("history", history.id),
     ]
     assert planned == [fileitem.path]
+
+
+def test_manual_transfer_music_release_preferences_are_normalized():
+    """手动整理应规范地区和文字代码，并保留用户给出的优先顺序。"""
+    item = ManualTransferItem(
+        music_release_regions=["cn", "TW"],
+        music_release_scripts=["hans", "Hant"],
+    )
+
+    assert item.music_release_regions == ["CN", "TW"]
+    assert item.music_release_scripts == ["Hans", "Hant"]
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("music_release_regions", ["CHN"]),
+        ("music_release_regions", ["CN", "cn"]),
+        ("music_release_scripts", ["Chinese"]),
+        ("music_release_scripts", ["Hans", "hans"]),
+    ],
+)
+def test_manual_transfer_music_release_preferences_reject_invalid_codes(field, value):
+    """非法或重复的发行偏好不能进入整理链。"""
+    with pytest.raises(ValueError):
+        ManualTransferItem(**{field: value})

tests/test_music_album_match.py

@@ -10,7 +10,7 @@ from app.chain.media import MediaChain
 from app.chain.media.cache import AlbumDirectoryCache
 from app.domain.context import MusicAlbumInfo, MusicInfo
 from app.domain.meta.metamusic import MetaMusic
-from app.modules.musicbrainz import MusicBrainzModule
+from app.modules.musicbrainz import MusicBrainzModule, _MusicReleasePreference
 
 
 def _release_detail(release_id: str, title: str, artist: str, tracks: list[tuple[str, int]]):
@@ -101,6 +101,72 @@ def test_match_music_album_rejects_mismatched_trackset(monkeypatch):
     assert album is None
 
 
+def test_release_match_prefers_mainland_simplified_when_structure_is_equal():
+    """曲目结构等价时应按请求优先选择大陆简体发行版。"""
+    traditional = _release_detail("release-tw", "七里香", "周杰伦", ALBUM_TRACKS)
+    traditional.update({"country": "TW", "text-representation": {"script": "Hant"}})
+    simplified = _release_detail("release-cn", "七里香", "周杰伦", ALBUM_TRACKS)
+    simplified.update({"country": "CN", "text-representation": {"script": "Hans"}})
+
+    album = MusicBrainzModule._select_release_match(
+        MetaMusic(album="七里香", artists=["周杰伦"]),
+        _local_tracks(),
+        [traditional, simplified],
+        _MusicReleasePreference(regions=("CN", "TW"), scripts=("Hans", "Hant")),
+    )
+
+    assert album is not None
+    assert album.raw_data["release_id"] == "release-cn"
+    assert album.raw_data["release_country"] == "CN"
+    assert album.raw_data["release_script"] == "Hans"
+
+
+def test_release_match_keeps_structural_accuracy_above_region_preference():
+    """地区偏好最多只做近似候选排序，不能压过更准确的曲目结构。"""
+    accurate = _release_detail("release-tw", "七里香", "周杰伦", ALBUM_TRACKS)
+    accurate.update({"country": "TW", "text-representation": {"script": "Hant"}})
+    inaccurate_tracks = [(name, length + 45) for name, length in ALBUM_TRACKS]
+    preferred = _release_detail("release-cn", "七里香", "周杰伦", inaccurate_tracks)
+    preferred.update({"country": "CN", "text-representation": {"script": "Hans"}})
+
+    album = MusicBrainzModule._select_release_match(
+        MetaMusic(album="七里香", artists=["周杰伦"]),
+        _local_tracks(),
+        [preferred, accurate],
+        _MusicReleasePreference(regions=("CN", "TW"), scripts=("Hans", "Hant")),
+    )
+
+    assert album is not None
+    assert album.raw_data["release_id"] == "release-tw"
+
+
+def test_select_track_release_uses_manual_preference_before_date():
+    """Release Group 的代表曲目版本应复用同一套请求级偏好。"""
+    selected = MusicBrainzModule._select_track_release(
+        [
+            {
+                "id": "release-tw",
+                "status": "Official",
+                "date": "2004-01-01",
+                "country": "TW",
+                "text-representation": {"script": "Hant"},
+                "media": [{"track-count": 10}],
+            },
+            {
+                "id": "release-cn",
+                "status": "Official",
+                "date": "2005-01-01",
+                "country": "CN",
+                "text-representation": {"script": "Hans"},
+                "media": [{"track-count": 10}],
+            },
+        ],
+        _MusicReleasePreference(regions=("CN", "TW"), scripts=("Hans", "Hant")),
+    )
+
+    assert selected["id"] == "release-cn"
+
+
 def test_release_queries_fallback_to_track_titles():
     """目录名没有专辑线索时应使用代表性曲名反查发行版本。"""
     queries = MusicBrainzModule._release_queries(

tests/test_music_recognize_routing.py

@@ -363,6 +363,12 @@ def test_music_metadata_simplified_conversion_defaults_to_enabled():
     assert ConfigModel.model_fields["MUSIC_METADATA_TO_SIMPLIFIED"].default is True
 
 
+def test_music_release_preference_defaults_favor_mainland_simplified():
+    """默认发行排序应先大陆、再台港，并优先简体字形。"""
+    assert ConfigModel.model_fields["MUSIC_RELEASE_REGION_PRIORITY"].default == "CN,TW,HK"
+    assert ConfigModel.model_fields["MUSIC_RELEASE_SCRIPT_PRIORITY"].default == "Hans,Hant,Latn"
+
+
 def test_media_chain_converts_recognized_music_metadata_without_mutating_source(monkeypatch):
     """开启开关时应转换标准音乐字段，并保留模块缓存对象和歌词原文。"""
     source_info = MusicInfo(

tests/test_transfer_music_album_selection.py

@@ -0,0 +1,226 @@
+from copy import deepcopy
+from types import SimpleNamespace
+
+from app.chain.media import MediaChain
+from app.chain.transfer.facade import TransferChain
+from app.domain.context import MusicAlbumInfo, MusicInfo
+from app.domain.meta.metamusic import MetaMusic
+from app.schemas.category import ClassificationResult, ClassificationSelection
+from app.schemas.file import FileItem
+from app.schemas.types import MediaSource, MediaType
+from tests.test_transfer_sync_extra_files import (
+    bind_empty_history_repositories,
+    make_fileitem,
+    make_transfer_chain,
+)
+
+
+def _album() -> MusicAlbumInfo:
+    """构造已选择的简体中文发行版。"""
+    classification = ClassificationResult(
+        recommended=ClassificationSelection(
+            category_id="music.album",
+            category_path=["Album"],
+            rule_id="music.album.default",
+            source="automatic",
+        ),
+        effective=ClassificationSelection(
+            category_id="music.album",
+            category_path=["Album"],
+            rule_id="music.album.default",
+            source="automatic",
+        ),
+        policy_revision=2,
+        state="complete",
+    )
+    return MusicAlbumInfo(
+        media_source=MediaSource.MusicBrainz,
+        media_id="release-group-1",
+        title="七里香",
+        artists=["周杰伦"],
+        album_type="Album",
+        release_date="2004-08-03",
+        library_category="Album",
+        classification=classification,
+        tracks=[
+            MusicInfo(
+                media_source=MediaSource.MusicBrainz,
+                media_id="recording-1",
+                title="我的地盘",
+                artists=["周杰伦"],
+                album="七里香",
+                album_id="release-group-1",
+                album_type="Album",
+                track_number=1,
+                total_tracks=2,
+            ),
+            MusicInfo(
+                media_source=MediaSource.MusicBrainz,
+                media_id="recording-2",
+                title="借口",
+                artists=["周杰伦"],
+                album="七里香",
+                album_id="release-group-1",
+                album_type="Album",
+                track_number=2,
+                total_tracks=2,
+            ),
+        ],
+    )
+
+
+def _directory_item(path) -> FileItem:
+    """构造手动整理使用的专辑目录项。"""
+    return FileItem(
+        storage="local",
+        path=path.as_posix() + "/",
+        type="dir",
+        name=path.name,
+    )
+
+
+def _prepare_chain(monkeypatch, fileitems):
+    """隔离文件规划外部依赖，返回可同步观察的整理链。"""
+    chain = make_transfer_chain()
+    bind_empty_history_repositories(chain)
+    monkeypatch.setattr(
+        chain,
+        "_TransferChain__get_trans_fileitems",
+        lambda _fileitem, predicate: [(item, False) for item in fileitems],
+    )
+    monkeypatch.setattr(chain, "_TransferChain__put_to_jobview", lambda _task: True)
+    monkeypatch.setattr(chain, "_register_scrape_batch_task", lambda _task: None)
+    monkeypatch.setattr(chain, "_close_scrape_batch", lambda _batch_id: None)
+    monkeypatch.setattr(
+        "app.chain.transfer.workflow.get_configured_system_config",
+        lambda: SimpleNamespace(get=lambda _key: None),
+    )
+    return chain
+
+
+def test_selected_album_tracks_override_source_tag_names(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)
+    planned = []
+    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])),
+    )
+
+    def handle_transfer(task, callback=None):
+        """记录规划后的曲目与分类上下文。"""
+        del callback
+        planned.append((task.meta.title, task.mediainfo.title, task.mediainfo.library_category))
+        return True, ""
+
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", handle_transfer)
+    album = _album()
+
+    state, message = TransferChain.do_transfer(
+        chain,
+        fileitem=_directory_item(album_dir),
+        mediainfo=album,
+        mtype=MediaType.MUSIC,
+        media_source=MediaSource.MusicBrainz,
+        media_id=album.media_id,
+        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()
+    album = _album()
+    captured = {}
+    monkeypatch.setattr(
+        "app.chain.transfer.history.MediaChain",
+        lambda: SimpleNamespace(get_music_album=lambda **_kwargs: album),
+    )
+
+    def do_transfer(**kwargs):
+        """记录手动入口传入的完整专辑上下文。"""
+        captured.update(kwargs)
+        return True, ""
+
+    monkeypatch.setattr(chain, "do_transfer", do_transfer)
+
+    state, message = TransferChain.manual_transfer(
+        chain,
+        fileitem=make_fileitem("/music/七里香/01.flac"),
+        media_source=MediaSource.MusicBrainz,
+        media_id="release-group-1",
+        mtype=MediaType.MUSIC,
+        music_type="album",
+        preview=True,
+    )
+
+    assert state is True
+    assert message == ""
+    assert captured["mediainfo"] is album
+    assert captured["mediainfo"].title == "七里香"
+    assert [track.title for track in captured["mediainfo"].tracks] == [
+        "我的地盘",
+        "借口",
+    ]
+
+
+def test_selected_album_rejects_duplicate_local_editions(tmp_path, monkeypatch):
+    """同一手动批次包含两套同曲序文件时应阻止目标覆盖。"""
+    album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
+    extra_dir = album_dir / "附加原版"
+    extra_dir.mkdir(parents=True)
+    paths = [
+        album_dir / "01.flac",
+        album_dir / "02.flac",
+        extra_dir / "01.flac",
+        extra_dir / "02.flac",
+    ]
+    for path in paths:
+        path.write_bytes(b"audio")
+    fileitems = [make_fileitem(path.as_posix()) for path in paths]
+    chain = _prepare_chain(monkeypatch, fileitems)
+    monkeypatch.setattr(
+        "app.chain.media.album.AudioMetadataHelper.read_many",
+        lambda requested: [
+            MetaMusic(title=path.stem, track_number=int(path.stem))
+            for path in requested
+        ],
+    )
+    album = _album()
+
+    state, message = TransferChain.do_transfer(
+        chain,
+        fileitem=_directory_item(album_dir),
+        mediainfo=album,
+        mtype=MediaType.MUSIC,
+        media_source=MediaSource.MusicBrainz,
+        media_id=album.media_id,
+        background=False,
+        preview=True,
+    )
+
+    assert state is False
+    assert "只能对齐 2 / 4" in message
+    assert "重复版本" in message

tests/test_transfer_planning_compat.py

@@ -125,6 +125,8 @@ def test_transfer_chain_do_transfer_keeps_legacy_signature():
         ("cleanup_dest_fileitem", None),
         ("continue_callback", None),
         ("reorganize", False),
+        ("music_release_regions", None),
+        ("music_release_scripts", None),
     )