From 1f728996ebfbc27a6a5b031667b0f4d640bbbdda Mon Sep 17 00:00:00 2001 From: yanjunpu <541726449@qq.com> Date: Tue, 8 Sep 2026 05:20:02 +0800 Subject: [PATCH 1/3] feat(music): add release region and script preferences --- app/api/endpoints/transfer.py | 4 + app/application/configuration.py | 2 + app/chain/media/album.py | 77 +++++++- app/chain/media/catalog.py | 18 +- app/chain/media/contract.py | 4 + app/chain/media/facade.py | 1 + app/chain/musicbrainz.py | 36 +++- app/chain/transfer/filter.py | 11 +- app/chain/transfer/history.py | 33 +++- app/chain/transfer/workflow.py | 16 +- app/modules/musicbrainz/__init__.py | 231 ++++++++++++++++++++--- app/runtime/config.py | 4 + app/schemas/transfer.py | 23 +++ app/startup/composition/configuration.py | 6 + docs/cli.md | 3 + docs/mcp-api.md | 2 +- tests/test_manual_transfer_history.py | 26 +++ tests/test_music_album_match.py | 68 ++++++- tests/test_music_recognize_routing.py | 6 + tests/test_transfer_planning_compat.py | 2 + 20 files changed, 526 insertions(+), 47 deletions(-) diff --git a/app/api/endpoints/transfer.py b/app/api/endpoints/transfer.py index 50b613d43..4f9f37e43 100644 --- a/app/api/endpoints/transfer.py +++ b/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, diff --git a/app/application/configuration.py b/app/application/configuration.py index dcb7b3ffd..a3f8dc24f 100644 --- a/app/application/configuration.py +++ b/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 diff --git a/app/chain/media/album.py b/app/chain/media/album.py index 6007bfdfa..95c2d9fd7 100644 --- a/app/chain/media/album.py +++ b/app/chain/media/album.py @@ -22,9 +22,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): @@ -159,11 +163,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 = 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 +190,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 +216,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 +226,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 +254,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 +264,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 +288,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 diff --git a/app/chain/media/catalog.py b/app/chain/media/catalog.py index d02605a52..10b524a45 100644 --- a/app/chain/media/catalog.py +++ b/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), diff --git a/app/chain/media/contract.py b/app/chain/media/contract.py index e83c9ae71..d4aadcf41 100644 --- a/app/chain/media/contract.py +++ b/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]: """异步识别音乐专辑目录。""" ... diff --git a/app/chain/media/facade.py b/app/chain/media/facade.py index 20fd64a2a..e53c4ba0e 100644 --- a/app/chain/media/facade.py +++ b/app/chain/media/facade.py @@ -118,6 +118,7 @@ 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] diff --git a/app/chain/musicbrainz.py b/app/chain/musicbrainz.py index 52e7803a5..de612b4e5 100644 --- a/app/chain/musicbrainz.py +++ b/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) diff --git a/app/chain/transfer/filter.py b/app/chain/transfer/filter.py index 771a91a17..021185f80 100644 --- a/app/chain/transfer/filter.py +++ b/app/chain/transfer/filter.py @@ -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 diff --git a/app/chain/transfer/history.py b/app/chain/transfer/history.py index efccdb28b..bb721bba4 100644 --- a/app/chain/transfer/history.py +++ b/app/chain/transfer/history.py @@ -13,6 +13,7 @@ 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, @@ -140,6 +141,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 +169,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,13 +178,23 @@ class TransferHistoryOwner(_TransferOwnerBase): return False, "手动整理需要同时提供 media_source 和 media_id" if media_source and media_id: # 有输入媒体ID时预先识别，音乐与影视统一走 recognize_media 按类型分发 - mediainfo = MediaChain().recognize_media( - media_source=media_source, - media_id=media_id, - music_type=music_type, - mtype=mtype, - episode_group=episode_group, - ) + mediainfo: Optional[Union[MediaInfo, MusicInfo]] + 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, + ) + mediainfo = album.to_music_info() if album else None + else: + mediainfo = MediaChain().recognize_media( + media_source=media_source, + media_id=media_id, + music_type=music_type, + mtype=mtype, + episode_group=episode_group, + ) if not mediainfo: return ( False, @@ -215,6 +230,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 +262,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 diff --git a/app/chain/transfer/workflow.py b/app/chain/transfer/workflow.py index 4e03574e9..d80da601a 100644 --- a/app/chain/transfer/workflow.py +++ b/app/chain/transfer/workflow.py @@ -383,6 +383,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 +419,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]]: @@ -452,6 +456,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,6 +488,8 @@ 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( @@ -603,6 +611,8 @@ 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, ) except OperationInterrupted: return False, f"{fileitem.name} 已取消" @@ -664,6 +674,8 @@ 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]], ) -> Tuple[List[TransferTask], bool, List[str], int, set[Tuple[str, str]]]: """从冻结候选构建任务，并集中执行历史去重与 durable 绑定。""" _build_file_meta = build_file_meta @@ -824,7 +836,9 @@ class TransferWorkflowOwner(_TransferOwnerBase): ) 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): diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index bb2aa7ad0..c665f649e 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/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 diff --git a/app/runtime/config.py b/app/runtime/config.py index e9d52396c..a0e2a95a3 100644 --- a/app/runtime/config.py +++ b/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 的自建实例 diff --git a/app/schemas/transfer.py b/app/schemas/transfer.py index 581048633..942a1c487 100644 --- a/app/schemas/transfer.py +++ b/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): """ diff --git a/app/startup/composition/configuration.py b/app/startup/composition/configuration.py index af4998d7d..90794dd37 100644 --- a/app/startup/composition/configuration.py +++ b/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, diff --git a/docs/cli.md b/docs/cli.md index c1b51f0f8..cee8fd475 100644 --- a/docs/cli.md +++ b/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` 显示单个配置项的类型、默认值和当前值 ## 数据库备份命令 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 179904bb0..227cc4d05 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -278,7 +278,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 结算写入；响应仅返回任务、操作、决策、后续状态和复核修订号 | diff --git a/tests/test_manual_transfer_history.py b/tests/test_manual_transfer_history.py index 754c13415..4e8a378d7 100644 --- a/tests/test_manual_transfer_history.py +++ b/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}) diff --git a/tests/test_music_album_match.py b/tests/test_music_album_match.py index 803a212e8..ccdb427cc 100644 --- a/tests/test_music_album_match.py +++ b/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( diff --git a/tests/test_music_recognize_routing.py b/tests/test_music_recognize_routing.py index 92ae43bdf..5fcf39245 100644 --- a/tests/test_music_recognize_routing.py +++ b/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( diff --git a/tests/test_transfer_planning_compat.py b/tests/test_transfer_planning_compat.py index c3476d17b..5ce99586c 100644 --- a/tests/test_transfer_planning_compat.py +++ b/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), ) From e84418ceb311ce731d56821794d2d9e4d6af045e Mon Sep 17 00:00:00 2001 From: yanjunpu <541726449@qq.com> Date: Tue, 8 Sep 2026 13:41:30 +0800 Subject: [PATCH 2/3] fix(music): apply selected album tracks during transfer --- app/chain/media/album.py | 18 ++ app/chain/media/facade.py | 3 + app/chain/transfer/contract.py | 3 + app/chain/transfer/filter.py | 61 ++++- app/chain/transfer/history.py | 58 +++-- app/chain/transfer/request.py | 21 ++ app/chain/transfer/workflow.py | 134 +++++------ tests/test_transfer_music_album_selection.py | 228 +++++++++++++++++++ 8 files changed, 434 insertions(+), 92 deletions(-) create mode 100644 tests/test_transfer_music_album_selection.py diff --git a/app/chain/media/album.py b/app/chain/media/album.py index 95c2d9fd7..ca922f7cf 100644 --- a/app/chain/media/album.py +++ b/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 @@ -159,6 +160,23 @@ 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, diff --git a/app/chain/media/facade.py b/app/chain/media/facade.py index e53c4ba0e..91ca726ca 100644 --- a/app/chain/media/facade.py +++ b/app/chain/media/facade.py @@ -123,6 +123,9 @@ class MediaChain(ChainBase, metaclass=Singleton): _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( diff --git a/app/chain/transfer/contract.py b/app/chain/transfer/contract.py index 92571667e..3c559aa78 100644 --- a/app/chain/transfer/contract.py +++ b/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] diff --git a/app/chain/transfer/filter.py b/app/chain/transfer/filter.py index 021185f80..8ac6d8749 100644 --- a/app/chain/transfer/filter.py +++ b/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 @@ -193,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: @@ -235,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], diff --git a/app/chain/transfer/history.py b/app/chain/transfer/history.py index bb721bba4..46c8e6503 100644 --- a/app/chain/transfer/history.py +++ b/app/chain/transfer/history.py @@ -6,7 +6,7 @@ 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 @@ -23,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]], +) -> tuple[Optional[Union[MediaInfo, MusicInfo]], Optional[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.to_music_info() if album else None), album + return MediaChain().recognize_media( + media_source=media_source, + media_id=media_id, + music_type=music_type, + mtype=mtype, + episode_group=episode_group, + ), None + + class TransferHistoryOwner(_TransferOwnerBase): """唯一持有手动历史、重整命令和通知公开入口。""" @@ -174,27 +202,20 @@ class TransferHistoryOwner(_TransferOwnerBase): """ logger.info(f"手动整理：{fileitem.path} ...") explicit_identity = media_source is not None or media_id is not None + selected_music_album: Optional[MusicAlbumInfo] = None if explicit_identity and (not media_source or not media_id): return False, "手动整理需要同时提供 media_source 和 media_id" if media_source and media_id: # 有输入媒体ID时预先识别，音乐与影视统一走 recognize_media 按类型分发 - mediainfo: Optional[Union[MediaInfo, MusicInfo]] - 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, - ) - mediainfo = album.to_music_info() if album else None - else: - mediainfo = MediaChain().recognize_media( - media_source=media_source, - media_id=media_id, - music_type=music_type, - mtype=mtype, - episode_group=episode_group, - ) + mediainfo, selected_music_album = _recognize_manual_media( + media_source=media_source, + media_id=media_id, + 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, @@ -232,6 +253,7 @@ class TransferHistoryOwner(_TransferOwnerBase): cleanup_dest_fileitem=cleanup_dest_fileitem, music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, + selected_music_album=selected_music_album, ) if not state: return False, errmsg diff --git a/app/chain/transfer/request.py b/app/chain/transfer/request.py index 28efd8c78..f92e21143 100644 --- a/app/chain/transfer/request.py +++ b/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: """持有一次整理请求的只读候选规划上下文。""" diff --git a/app/chain/transfer/workflow.py b/app/chain/transfer/workflow.py index d80da601a..9a215efa4 100644 --- a/app/chain/transfer/workflow.py +++ b/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, @@ -385,6 +364,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): reorganize: Optional[bool] = False, music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, + selected_music_album: Optional[MusicAlbumInfo] = None, ) -> Tuple[bool, Union[str, dict]]: """ 兼容公开整理入口，委托给内部批次执行阶段。 @@ -421,6 +401,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): reorganize=reorganize, music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, + selected_music_album=selected_music_album, ) def _execute_transfer(self, *args: Any, **kwargs: Any) -> Tuple[bool, Union[str, dict]]: @@ -458,6 +439,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): recovery_admission: Optional[TransferAdmission] = None, music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, + selected_music_album: Optional[MusicAlbumInfo] = None, ) -> Tuple[bool, Union[str, dict]]: """ 执行一个复杂目录的整理操作 @@ -490,9 +472,10 @@ class TransferWorkflowOwner(_TransferOwnerBase): :param recovery_admission: 内部恢复调用绑定的既有 durable 记录 :param music_release_regions: 本次音乐整理的发行地区优先级 :param music_release_scripts: 本次音乐整理的文字字形优先级 + :param selected_music_album: 手动指定的音乐专辑，用于按已选发行版逐曲对齐 返回：成功标识，错误信息 """ - mediainfo, media_source, media_id, identity_error = self._normalize_transfer_identity( + mediainfo, media_source, media_id, identity_error = _normalize_transfer_identity( mediainfo=mediainfo, mtype=mtype, media_source=media_source, @@ -570,6 +553,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: @@ -613,6 +602,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): 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} 已取消" @@ -676,6 +666,7 @@ class TransferWorkflowOwner(_TransferOwnerBase): 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 @@ -829,10 +820,9 @@ 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): # 无标签音频或误带单曲身份的整包按目录级专辑匹配；命中结果带缓存不会逐文件重复请求 diff --git a/tests/test_transfer_music_album_selection.py b/tests/test_transfer_music_album_selection.py new file mode 100644 index 000000000..3c6263ca2 --- /dev/null +++ b/tests/test_transfer_music_album_selection.py @@ -0,0 +1,228 @@ +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.to_music_info(), + mtype=MediaType.MUSIC, + media_source=MediaSource.MusicBrainz, + media_id=album.media_id, + selected_music_album=album, + 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["selected_music_album"] is album + assert captured["mediainfo"].title == "七里香" + assert [track.title for track in captured["selected_music_album"].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.to_music_info(), + mtype=MediaType.MUSIC, + media_source=MediaSource.MusicBrainz, + media_id=album.media_id, + selected_music_album=album, + background=False, + preview=True, + ) + + assert state is False + assert "只能对齐 2 / 4" in message + assert "重复版本" in message From 981be2935e070bbe81992b4fa7431670471d9793 Mon Sep 17 00:00:00 2001 From: yanjunpu <541726449@qq.com> Date: Tue, 8 Sep 2026 13:49:31 +0800 Subject: [PATCH 3/3] fix(music): preserve transfer entry compatibility --- app/chain/transfer/history.py | 14 ++++++-------- app/chain/transfer/workflow.py | 14 ++++++++------ tests/test_transfer_music_album_selection.py | 10 ++++------ 3 files changed, 18 insertions(+), 20 deletions(-) diff --git a/app/chain/transfer/history.py b/app/chain/transfer/history.py index 46c8e6503..96691ac0d 100644 --- a/app/chain/transfer/history.py +++ b/app/chain/transfer/history.py @@ -32,7 +32,7 @@ def _recognize_manual_media( episode_group: Optional[str], music_release_regions: Optional[list[str]], music_release_scripts: Optional[list[str]], -) -> tuple[Optional[Union[MediaInfo, MusicInfo]], Optional[MusicAlbumInfo]]: +) -> Optional[Union[MediaInfo, MusicInfo, MusicAlbumInfo]]: """识别手动指定的媒体，并为音乐专辑保留完整曲目表。""" if mtype == MediaType.MUSIC and music_type == MUSIC_ENTITY_ALBUM: album = MediaChain().get_music_album( @@ -41,14 +41,14 @@ def _recognize_manual_media( music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, ) - return (album.to_music_info() if album else None), album + return album return MediaChain().recognize_media( media_source=media_source, media_id=media_id, music_type=music_type, mtype=mtype, episode_group=episode_group, - ), None + ) class TransferHistoryOwner(_TransferOwnerBase): @@ -202,12 +202,11 @@ class TransferHistoryOwner(_TransferOwnerBase): """ logger.info(f"手动整理：{fileitem.path} ...") explicit_identity = media_source is not None or media_id is not None - selected_music_album: Optional[MusicAlbumInfo] = None if explicit_identity and (not media_source or not media_id): return False, "手动整理需要同时提供 media_source 和 media_id" if media_source and media_id: # 有输入媒体ID时预先识别，音乐与影视统一走 recognize_media 按类型分发 - mediainfo, selected_music_album = _recognize_manual_media( + mediainfo = _recognize_manual_media( media_source=media_source, media_id=media_id, mtype=mtype, @@ -221,9 +220,9 @@ class TransferHistoryOwner(_TransferOwnerBase): 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) # 开始整理 @@ -253,7 +252,6 @@ class TransferHistoryOwner(_TransferOwnerBase): cleanup_dest_fileitem=cleanup_dest_fileitem, music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, - selected_music_album=selected_music_album, ) if not state: return False, errmsg diff --git a/app/chain/transfer/workflow.py b/app/chain/transfer/workflow.py index 9a215efa4..950ed4e55 100644 --- a/app/chain/transfer/workflow.py +++ b/app/chain/transfer/workflow.py @@ -338,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, @@ -364,7 +364,6 @@ class TransferWorkflowOwner(_TransferOwnerBase): reorganize: Optional[bool] = False, music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, - selected_music_album: Optional[MusicAlbumInfo] = None, ) -> Tuple[bool, Union[str, dict]]: """ 兼容公开整理入口，委托给内部批次执行阶段。 @@ -401,7 +400,6 @@ class TransferWorkflowOwner(_TransferOwnerBase): reorganize=reorganize, music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, - selected_music_album=selected_music_album, ) def _execute_transfer(self, *args: Any, **kwargs: Any) -> Tuple[bool, Union[str, dict]]: @@ -412,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, @@ -439,7 +437,6 @@ class TransferWorkflowOwner(_TransferOwnerBase): recovery_admission: Optional[TransferAdmission] = None, music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, - selected_music_album: Optional[MusicAlbumInfo] = None, ) -> Tuple[bool, Union[str, dict]]: """ 执行一个复杂目录的整理操作 @@ -472,9 +469,14 @@ class TransferWorkflowOwner(_TransferOwnerBase): :param recovery_admission: 内部恢复调用绑定的既有 durable 记录 :param music_release_regions: 本次音乐整理的发行地区优先级 :param music_release_scripts: 本次音乐整理的文字字形优先级 - :param selected_music_album: 手动指定的音乐专辑，用于按已选发行版逐曲对齐 返回：成功标识，错误信息 """ + 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, diff --git a/tests/test_transfer_music_album_selection.py b/tests/test_transfer_music_album_selection.py index 3c6263ca2..b573e6b21 100644 --- a/tests/test_transfer_music_album_selection.py +++ b/tests/test_transfer_music_album_selection.py @@ -134,11 +134,10 @@ def test_selected_album_tracks_override_source_tag_names(tmp_path, monkeypatch): state, message = TransferChain.do_transfer( chain, fileitem=_directory_item(album_dir), - mediainfo=album.to_music_info(), + mediainfo=album, mtype=MediaType.MUSIC, media_source=MediaSource.MusicBrainz, media_id=album.media_id, - selected_music_album=album, background=False, ) @@ -179,9 +178,9 @@ def test_manual_album_identity_forwards_full_selected_album(monkeypatch): assert state is True assert message == "" - assert captured["selected_music_album"] is album + assert captured["mediainfo"] is album assert captured["mediainfo"].title == "七里香" - assert [track.title for track in captured["selected_music_album"].tracks] == [ + assert [track.title for track in captured["mediainfo"].tracks] == [ "我的地盘", "借口", ] @@ -214,11 +213,10 @@ def test_selected_album_rejects_duplicate_local_editions(tmp_path, monkeypatch): state, message = TransferChain.do_transfer( chain, fileitem=_directory_item(album_dir), - mediainfo=album.to_music_info(), + mediainfo=album, mtype=MediaType.MUSIC, media_source=MediaSource.MusicBrainz, media_id=album.media_id, - selected_music_album=album, background=False, preview=True, )