feat(music): add release region and script preferences

This commit is contained in:
yanjunpu
2026-09-08 05:20:02 +08:00
parent 7ccb1d9dee
commit 1f728996eb
20 changed files with 526 additions and 47 deletions

View File

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

View File

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

View File

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

View File

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

View File

@@ -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]:
"""异步识别音乐专辑目录。"""
...

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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 的自建实例

View File

@@ -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):
"""

View File

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

View File

@@ -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` 显示单个配置项的类型、默认值和当前值
## 数据库备份命令

View File

@@ -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 结算写入;响应仅返回任务、操作、决策、后续状态和复核修订号 |

View File

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

View File

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

View File

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

View File

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