mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-08 17:08:35 +08:00
fix: 对齐音乐候选识别和缓存匹配规则
This commit is contained in:
jxxghp
@@ -878,12 +878,14 @@ class MetaMusic(MetaBase):
|
||||
self.audio_lossless = infer_audio_lossless(self.audio_format, self.audio_lossless)
|
||||
|
||||
def apply_title(self, value: Any) -> None:
|
||||
"""解析种子/文件名标题字符串，提取艺术家、曲名、年份并补充音质参数。
|
||||
"""解析种子/文件名标题字符串，提取艺术家、曲名、年份、版本并补充音质参数。
|
||||
|
||||
公共层先完成字符归一、音质与干扰信息剔除；随后由注册中心依次匹配
|
||||
命名模式和对应解析器，最后统一回填结构化字段并提取曲序前缀。
|
||||
"""
|
||||
raw = str(value or "")
|
||||
if not self.version:
|
||||
self.version = self._resource_version(raw, None)
|
||||
accelerator = get_metainfo_accelerator()
|
||||
if accelerator and MusicNameRegistry._uses_default_components():
|
||||
rust_result = accelerator.parse_metamusic(
|
||||
|
||||
+21
-8
@@ -85,6 +85,16 @@ def music_artists(music: MusicInfo) -> list[str]:
|
||||
return artists
|
||||
|
||||
|
||||
def music_artist_matches(music: MusicInfo, parsed_artists: Iterable[str]) -> bool:
|
||||
"""使用同实体署名和别名核验解析艺人，兼容完整艺名被分隔符拆成多个片段。"""
|
||||
artists = music_artists(music)
|
||||
parsed = unique_music_texts(parsed_artists)
|
||||
keys = {music_text_key(artist) for artist in parsed}
|
||||
if len(parsed) > 1 and any(any(separator in artist for separator in ("/", "&", ",")) for artist in artists):
|
||||
keys.add(music_text_key(" / ".join(parsed)))
|
||||
return bool(keys & {music_text_key(artist) for artist in artists})
|
||||
|
||||
|
||||
def music_base_title(value: Optional[str]) -> str:
|
||||
"""仅剥离已知发行版本后缀，保留未知括号和属于作品本身的文字。"""
|
||||
text = _VERSION_SUFFIX.sub("", _EDITION.sub("", str(value or "")))
|
||||
@@ -134,6 +144,15 @@ def _version_markers(text: str) -> set[str]:
|
||||
return {name for name, pattern in _VERSIONS.items() if re.search(pattern, text, re.I)}
|
||||
|
||||
|
||||
def music_version_matches(music: MusicInfo, meta: MetaMusic) -> bool:
|
||||
"""资源匹配与候选确认共用录音版本约束，不从艺术家字段推断版本。"""
|
||||
target_title = music.album or music.title if music.music_type == MUSIC_ENTITY_ALBUM else music.title
|
||||
# 专辑类型描述整专版本，但单曲的所属专辑类型不能代替该录音自身的版本。
|
||||
album_versions = " ".join(music.secondary_types or []) if music.music_type == MUSIC_ENTITY_ALBUM else ""
|
||||
expected = _version_markers(f"{target_title or ''} {music.version or ''} {album_versions}")
|
||||
return expected == _version_markers(f"{meta.title or ''} {meta.version or ''}")
|
||||
|
||||
|
||||
def match_music_resource(
|
||||
music: MusicInfo,
|
||||
title: str,
|
||||
@@ -154,11 +173,7 @@ def match_music_resource(
|
||||
titles = music_titles(music)
|
||||
title_matched = any(music_text_key(music_base_title(item)) in names for item in titles)
|
||||
content = f"{title} {description}"
|
||||
resource_artist_keys = {music_text_key(artist) for artist in resource.artists}
|
||||
if len(resource.artists) > 1 and any(any(separator in artist for separator in ("/", "&", ",")) for artist in artists):
|
||||
# 带分隔符的完整艺名可能被解析成多个片段，保留整段署名参与比较，不拼接无分隔符艺名。
|
||||
resource_artist_keys.add(music_text_key(resource.artist))
|
||||
artist_matched = bool(resource_artist_keys & {music_text_key(artist) for artist in artists}) if resource.artists \
|
||||
artist_matched = music_artist_matches(music, resource.artists) if resource.artists \
|
||||
else any(_contains_artist(content, artist) for artist in artists)
|
||||
if not title_matched:
|
||||
if music.music_type != MUSIC_ENTITY_ALBUM and artist_matched and any(
|
||||
@@ -182,9 +197,7 @@ def match_music_resource(
|
||||
return MusicMatch("candidate", "partial_album")
|
||||
if music.year and resource.year and str(music.year) != str(resource.year):
|
||||
return MusicMatch("candidate", "year_mismatch")
|
||||
target_title = music.album or music.title if music.music_type == MUSIC_ENTITY_ALBUM else music.title
|
||||
expected_version = _version_markers(f"{target_title or ''} {music.version or ''}")
|
||||
if expected_version != _version_markers(f"{resource.title or ''} {resource.version or ''}"):
|
||||
if not music_version_matches(music, resource):
|
||||
return MusicMatch("candidate", "version_mismatch")
|
||||
if _EDITION.search(music.title or "") and not any(music_text_key(item) in music_text_key(content) for item in titles):
|
||||
return MusicMatch("candidate", "edition_unverified")
|
||||
|
||||
@@ -18,7 +18,14 @@ from app.domain.context import (
|
||||
from app.domain.media import is_media_source_selected
|
||||
from app.domain.meta.metabase import MetaBase
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.domain.music import music_text_key, unique_music_texts
|
||||
from app.domain.music import (
|
||||
music_artist_matches,
|
||||
music_base_title,
|
||||
music_text_key,
|
||||
music_titles,
|
||||
music_version_matches,
|
||||
unique_music_texts,
|
||||
)
|
||||
from app.foundation.text import convert as zhconv_convert
|
||||
from app.modules import _ModuleBase
|
||||
from app.modules.musicbrainz.cache import MusicBrainzCache
|
||||
@@ -325,15 +332,15 @@ class MusicBrainzModule(_ModuleBase):
|
||||
|
||||
return sorted(candidates, key=score, reverse=True)
|
||||
|
||||
def _search_recordings(self, meta: MetaMusic, limit: int) -> list[MusicInfo]:
|
||||
"""按音频标签条件搜索 Recording，供全局搜索和文件识别复用。"""
|
||||
def _search_recordings(self, meta: MetaMusic, limit: int, require_match: bool = False) -> list[MusicInfo]:
|
||||
"""查询 Recording；自动识别须确认身份才停止，手动浏览仍保留原始候选。"""
|
||||
for query in self._recording_queries(meta):
|
||||
payload = self._request_json(
|
||||
"/recording",
|
||||
params={"query": query, "limit": max(1, min(limit, 100)), "fmt": "json"},
|
||||
)
|
||||
results = self._project_recording_search(payload)
|
||||
if results:
|
||||
if results and (not require_match or self._select_candidate(meta, results, self._source)):
|
||||
return results
|
||||
return []
|
||||
|
||||
@@ -341,8 +348,9 @@ class MusicBrainzModule(_ModuleBase):
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int,
|
||||
require_match: bool = False,
|
||||
) -> list[MusicInfo]:
|
||||
"""异步按音频标签条件搜索 Recording 候选。"""
|
||||
"""异步查询 Recording，与同步入口共用候选准入和停止条件。"""
|
||||
for query in self._recording_queries(meta):
|
||||
payload = await self._async_request_json(
|
||||
"/recording",
|
||||
@@ -353,7 +361,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
},
|
||||
)
|
||||
results = self._project_recording_search(payload)
|
||||
if results:
|
||||
if results and (not require_match or self._select_candidate(meta, results, self._source)):
|
||||
return results
|
||||
return []
|
||||
|
||||
@@ -485,8 +493,8 @@ class MusicBrainzModule(_ModuleBase):
|
||||
return remainder
|
||||
return text
|
||||
|
||||
def _search_albums(self, meta: MetaMusic, limit: int) -> list[MusicInfo]:
|
||||
"""按标题和可选艺术家搜索 Release Group 专辑候选，检索式同样逐级放宽。"""
|
||||
def _search_albums(self, meta: MetaMusic, limit: int, require_match: bool = False) -> list[MusicInfo]:
|
||||
"""查询 Release Group；自动识别只在存在可确认专辑时停止检索式回退。"""
|
||||
for query in self._album_queries(meta):
|
||||
payload = self._request_json(
|
||||
"/release-group",
|
||||
@@ -497,7 +505,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
},
|
||||
)
|
||||
results = self._project_album_search(payload)
|
||||
if results:
|
||||
if results and (not require_match or self._select_album_candidate(meta, results)):
|
||||
return results
|
||||
return []
|
||||
|
||||
@@ -505,8 +513,9 @@ class MusicBrainzModule(_ModuleBase):
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int,
|
||||
require_match: bool = False,
|
||||
) -> list[MusicInfo]:
|
||||
"""异步按标题和可选艺术家搜索 Release Group 专辑候选。"""
|
||||
"""异步查询 Release Group，与同步入口共用候选准入和停止条件。"""
|
||||
for query in self._album_queries(meta):
|
||||
payload = await self._async_request_json(
|
||||
"/release-group",
|
||||
@@ -517,7 +526,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
},
|
||||
)
|
||||
results = self._project_album_search(payload)
|
||||
if results:
|
||||
if results and (not require_match or self._select_album_candidate(meta, results)):
|
||||
return results
|
||||
return []
|
||||
|
||||
@@ -895,14 +904,8 @@ class MusicBrainzModule(_ModuleBase):
|
||||
|
||||
@staticmethod
|
||||
def _match_text(value: Optional[str]) -> str:
|
||||
"""移除大小写、空白、标点和繁简差异，生成相似度比较使用的紧凑文本。"""
|
||||
text = str(value or "").casefold()
|
||||
try:
|
||||
# 候选比对统一简体，避免条目繁体写法造成失配
|
||||
text = zhconv_convert(text, "zh-hans")
|
||||
except Exception: # pylint: disable=broad-except
|
||||
pass
|
||||
return re.sub(r"[\W_]+", "", text, flags=re.UNICODE)
|
||||
"""与资源匹配共用繁简、全半角、变音符和标点归一化规则。"""
|
||||
return music_text_key(value)
|
||||
|
||||
@classmethod
|
||||
def _unique_texts(cls, values: Iterable[Optional[str]]) -> list[str]:
|
||||
@@ -979,17 +982,18 @@ class MusicBrainzModule(_ModuleBase):
|
||||
return self._finalize_detail_recognition(plan, info)
|
||||
return self._recognize_from_candidates_sync(plan)
|
||||
|
||||
def _update_recognize_cache(self, meta: MetaMusic, info: Optional[MusicInfo]) -> None:
|
||||
def _update_recognize_cache(self, meta: MetaMusic, info: Optional[MusicInfo],
|
||||
music_type: Optional[str] = None) -> None:
|
||||
"""识别完成后把结果写入本地识别缓存，未挂载缓存时静默跳过。"""
|
||||
if self.cache:
|
||||
self.cache.update(meta, info)
|
||||
self.cache.update(meta, info, music_type=music_type)
|
||||
|
||||
def update_recognize_cache(
|
||||
self,
|
||||
meta: MetaBase,
|
||||
mediainfo: MusicInfo,
|
||||
) -> Optional[bool]:
|
||||
"""回填音乐本地识别缓存，共享识别成功后避免重复回查。"""
|
||||
"""共享识别成功后覆盖未限定请求及已确认实体的负缓存，保持旧回填 ABI。"""
|
||||
if not meta or not mediainfo:
|
||||
return None
|
||||
if not isinstance(meta, MetaMusic) or not isinstance(mediainfo, MusicInfo):
|
||||
@@ -997,6 +1001,8 @@ class MusicBrainzModule(_ModuleBase):
|
||||
if mediainfo.media_source != self._source:
|
||||
return None
|
||||
self._update_recognize_cache(meta, mediainfo)
|
||||
if mediainfo.media_id and mediainfo.music_type in (MUSIC_ENTITY_RECORDING, MUSIC_ENTITY_ALBUM):
|
||||
self._update_recognize_cache(meta, mediainfo, music_type=mediainfo.music_type)
|
||||
return True
|
||||
|
||||
async def async_update_recognize_cache(
|
||||
@@ -1081,9 +1087,11 @@ class MusicBrainzModule(_ModuleBase):
|
||||
meta = plan.require_meta()
|
||||
if not plan.cache_enabled or not self.cache:
|
||||
return None
|
||||
cached_info = self.cache.get(meta)
|
||||
cached_info = self.cache.get(meta, music_type=plan.music_type)
|
||||
if not cached_info:
|
||||
return None
|
||||
if plan.music_type and cached_info.music_type != plan.music_type:
|
||||
return None
|
||||
if cached_info.media_id:
|
||||
logger.info(f"{meta.title} 使用音乐识别缓存：{cached_info.title}")
|
||||
else:
|
||||
@@ -1098,7 +1106,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
) -> Optional[MusicInfo]:
|
||||
"""统一完成显式详情识别后的缓存回填。"""
|
||||
if info and plan.meta:
|
||||
self._update_recognize_cache(plan.meta, info)
|
||||
self._update_recognize_cache(plan.meta, info, music_type=plan.music_type)
|
||||
return info
|
||||
|
||||
@classmethod
|
||||
@@ -1147,7 +1155,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
"""统一生成候选识别兜底并写入本地缓存。"""
|
||||
meta = plan.require_meta()
|
||||
result = matched or self._info_from_meta(meta)
|
||||
self._update_recognize_cache(meta, result)
|
||||
self._update_recognize_cache(meta, result, music_type=plan.music_type)
|
||||
return result
|
||||
|
||||
def _recognize_from_candidates_sync(
|
||||
@@ -1160,12 +1168,12 @@ class MusicBrainzModule(_ModuleBase):
|
||||
if cached_info:
|
||||
return cached_info
|
||||
recordings = (
|
||||
self._search_recordings(meta, limit=10)
|
||||
self._search_recordings(meta, limit=10, require_match=True)
|
||||
if plan.search_recording else []
|
||||
)
|
||||
preliminary = self._select_recognition_candidate(plan, recordings)
|
||||
albums = (
|
||||
self._search_albums(meta, limit=10)
|
||||
self._search_albums(meta, limit=10, require_match=True)
|
||||
if self._should_search_albums(plan, preliminary)
|
||||
else []
|
||||
)
|
||||
@@ -1184,12 +1192,12 @@ class MusicBrainzModule(_ModuleBase):
|
||||
if cached_info:
|
||||
return cached_info
|
||||
recordings = (
|
||||
await self._async_search_recordings(meta, limit=10)
|
||||
await self._async_search_recordings(meta, limit=10, require_match=True)
|
||||
if plan.search_recording else []
|
||||
)
|
||||
preliminary = self._select_recognition_candidate(plan, recordings)
|
||||
albums = (
|
||||
await self._async_search_albums(meta, limit=10)
|
||||
await self._async_search_albums(meta, limit=10, require_match=True)
|
||||
if self._should_search_albums(plan, preliminary)
|
||||
else []
|
||||
)
|
||||
@@ -1205,55 +1213,31 @@ class MusicBrainzModule(_ModuleBase):
|
||||
candidates: Iterable[MusicInfo],
|
||||
media_source: MediaSource,
|
||||
) -> Optional[MusicInfo]:
|
||||
"""按标题、艺术家和专辑匹配度选择最可信的搜索候选。"""
|
||||
"""优先采用同一 ISRC，其他候选须满足完整名称、已有署名和录音版本约束。"""
|
||||
normalized_source = cls._normalize_text(media_source).casefold()
|
||||
# 资源标题携带的音质标记先剥离，再与候选曲名比对；
|
||||
# 曲名开头的艺术家署名前缀是命名习惯，用主体名比对
|
||||
clean_title = cls._strip_artist_prefix(cls._search_title(meta.title), meta.artists)
|
||||
# 条目的影视 tie-in 注释多为全角括号，与资源半角注释无法精确相等，
|
||||
# 去括号后的主体曲名一致视为弱匹配，且需艺术家同时命中才采信；
|
||||
# 卷号后缀（Vol. 3）是发行分卷标记，条目本体不含卷号
|
||||
bare_title = cls._strip_volume_suffix(cls._strip_parenthetical(clean_title))
|
||||
bare_title = music_base_title(clean_title)
|
||||
ranked: list[tuple[int, MusicInfo]] = []
|
||||
for candidate in candidates:
|
||||
if normalized_source and str(candidate.media_source or "").casefold() != normalized_source:
|
||||
continue
|
||||
if meta.isrc and cls._same_text(meta.isrc, candidate.isrc):
|
||||
# 相同 ISRC 是明确录音身份，不能被另一条纯标题命中的得分压过。
|
||||
return candidate
|
||||
score = 0
|
||||
title_match = False
|
||||
# 多艺术家资源任一命中即可，联名候选不会因主艺术家顺序失配
|
||||
artist_match = bool(meta.artists) and any(
|
||||
cls._same_text(artist_name, candidate_artist)
|
||||
for artist_name in meta.artists
|
||||
for candidate_artist in candidate.artists
|
||||
)
|
||||
if clean_title and cls._same_text(clean_title, candidate.title):
|
||||
artist_match = music_artist_matches(candidate, meta.artists)
|
||||
titles = music_titles(candidate)
|
||||
if clean_title and any(cls._same_text(clean_title, title) for title in titles):
|
||||
score += 4
|
||||
title_match = True
|
||||
elif (
|
||||
bare_title
|
||||
and artist_match
|
||||
and (
|
||||
cls._same_text(bare_title, cls._strip_parenthetical(candidate.title))
|
||||
# 条目「天國的情人：鄧麗君逝世十周年…」这类冒号副标题，主标题一致视为弱匹配
|
||||
or cls._same_text(bare_title, cls._main_title(candidate.title))
|
||||
# 条目「为你盛开-许巍《无尽光芒》…」这类连字符前置命名，首段曲名一致视为弱匹配
|
||||
or cls._same_text(bare_title, cls._head_title(candidate.title))
|
||||
# 条目「愛情電影主題曲 雲且留住」这类「主体名 补充说明」结构，首段一致视为弱匹配
|
||||
or (
|
||||
len(cls._match_text(bare_title)) >= 3
|
||||
and cls._same_text(bare_title, cls._lead_token(candidate.title))
|
||||
)
|
||||
# 条目带额外前缀/后缀完整包含资源主体名（好莱坞原声带类），长文本包含视为弱匹配
|
||||
or (
|
||||
len(cls._match_text(bare_title)) >= 6
|
||||
and cls._match_text(bare_title) in cls._match_text(candidate.title)
|
||||
)
|
||||
# 资源标题带演出后缀（S.H.E十七音乐会），条目本体一致视为弱匹配
|
||||
or (
|
||||
cls._performance_title(bare_title)
|
||||
and cls._same_text(cls._performance_title(bare_title), candidate.title)
|
||||
)
|
||||
)
|
||||
and any(cls._same_text(bare_title, music_base_title(title)) for title in titles)
|
||||
):
|
||||
score += 2
|
||||
title_match = True
|
||||
@@ -1261,17 +1245,12 @@ class MusicBrainzModule(_ModuleBase):
|
||||
score += 3
|
||||
if meta.album and cls._same_text(meta.album, candidate.album):
|
||||
score += 2
|
||||
isrc_match = bool(meta.isrc) and cls._same_text(meta.isrc, candidate.isrc)
|
||||
if isrc_match:
|
||||
score += 5
|
||||
# 同名多版本（如不同年份的重发单曲）靠发行年份消歧
|
||||
if meta.year and candidate.year and int(meta.year) == int(candidate.year):
|
||||
score += 1
|
||||
# 已知艺术家时，艺术家未命中的候选不能采信（ISRC 精确身份除外），
|
||||
# 兜住宽检索阶梯下同名异曲的误配；CJK 逐字 OR 检索召回宽，
|
||||
# 标题未命中的候选同样不能仅凭艺术家署名得分（ISRC 除外）
|
||||
if (meta.artists and not artist_match and not isrc_match) or (
|
||||
not title_match and not isrc_match
|
||||
# 非显式身份必须同时满足作品名、已有署名与版本，不能只靠累计得分确认。
|
||||
if (
|
||||
(meta.artists and not artist_match) or not title_match or not music_version_matches(candidate, meta)
|
||||
):
|
||||
score = 0
|
||||
ranked.append((score, candidate))
|
||||
@@ -1299,16 +1278,12 @@ class MusicBrainzModule(_ModuleBase):
|
||||
for album in albums:
|
||||
score = 0
|
||||
album_title = album.title or album.album
|
||||
artist_match = bool(meta.artists) and any(
|
||||
cls._same_text(artist_name, candidate_artist)
|
||||
for artist_name in meta.artists
|
||||
for candidate_artist in album.artists
|
||||
)
|
||||
artist_match = music_artist_matches(album, meta.artists)
|
||||
title_match = False
|
||||
# 资源带卷号时候选卷号不一致（含其他分卷）直接排除，避免 Vol.1 误配 Vol.3
|
||||
if meta_volume and cls._volume_number(album_title) not in (None, meta_volume):
|
||||
pass
|
||||
elif cls._same_text(clean_title, album_title):
|
||||
elif any(cls._same_text(clean_title, title) for title in music_titles(album, album=True)):
|
||||
score += 4
|
||||
title_match = True
|
||||
elif (
|
||||
@@ -1354,7 +1329,7 @@ class MusicBrainzModule(_ModuleBase):
|
||||
if meta.year and album.year and int(meta.year) == int(album.year):
|
||||
score += 1
|
||||
# 标题与艺术家缺一不可，仅有标题相似不能采信
|
||||
ranked.append((score if title_match and artist_match else 0, album))
|
||||
ranked.append((score if title_match and artist_match and music_version_matches(album, meta) else 0, album))
|
||||
if not ranked:
|
||||
return None
|
||||
ranked.sort(key=lambda item: item[0], reverse=True)
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import hashlib
|
||||
import json
|
||||
import pickle
|
||||
import traceback
|
||||
from math import ceil
|
||||
@@ -5,17 +7,16 @@ from threading import RLock
|
||||
from time import time
|
||||
from typing import Optional
|
||||
|
||||
from app.runtime.cache import FileCache, TTLCache
|
||||
from app.runtime.settings import get_runtime_setting
|
||||
|
||||
from app.domain.context import MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.runtime.log import logger
|
||||
from app.schemas.types import MUSIC_ENTITY_RECORDING
|
||||
from app.foundation.singleton import WeakSingleton
|
||||
from app.runtime.cache import FileCache, TTLCache
|
||||
from app.runtime.log import logger
|
||||
from app.runtime.settings import get_runtime_setting
|
||||
from app.schemas.types import MUSIC_ENTITY_RECORDING
|
||||
|
||||
lock = RLock()
|
||||
PERSISTENCE_VERSION = 1
|
||||
PERSISTENCE_VERSION = 2
|
||||
PERSISTENCE_REGION = "recognize"
|
||||
PERSISTENCE_KEY = "musicbrainz"
|
||||
|
||||
@@ -119,20 +120,26 @@ class MusicBrainzCache(metaclass=WeakSingleton):
|
||||
return sorted(cache_items, key=lambda item: item["key"])
|
||||
|
||||
@staticmethod
|
||||
def __get_key(meta: MetaMusic) -> str:
|
||||
"""
|
||||
获取缓存KEY，携带数据源原生 ID 时以 ID 为准身份
|
||||
"""
|
||||
artists = "/".join(meta.artists or [])
|
||||
return f"[音乐]{meta.media_id or meta.title}-{artists}-{meta.album}-{meta.year}"
|
||||
def __get_key(meta: MetaMusic, music_type: Optional[str] = None) -> str:
|
||||
"""按来源身份或完整识别证据及请求实体编码键，避免版本串用和分隔符碰撞。"""
|
||||
source = str(meta.media_source) if meta.media_source else None
|
||||
identity: list[object]
|
||||
if source and meta.media_id:
|
||||
identity = ["id", source, str(meta.media_id)]
|
||||
else:
|
||||
identity = ["meta", source, meta.title, list(meta.artists or []), meta.album,
|
||||
meta.year, meta.version, meta.isrc]
|
||||
payload = json.dumps([music_type, identity], ensure_ascii=False, separators=(",", ":"))
|
||||
return f"[音乐:v{PERSISTENCE_VERSION}]{hashlib.sha256(payload.encode('utf-8')).hexdigest()}"
|
||||
|
||||
def get(self, meta: MetaMusic) -> Optional[MusicInfo]:
|
||||
def get(self, meta: MetaMusic, music_type: Optional[str] = None) -> Optional[MusicInfo]:
|
||||
"""
|
||||
根据元数据获取缓存的音乐识别结果
|
||||
@param meta: 音乐元数据
|
||||
@param music_type: 本次请求的实体范围，未指定时与显式单曲、专辑请求隔离
|
||||
@return: 缓存命中的音乐信息，未命中返回 None
|
||||
"""
|
||||
key = self.__get_key(meta)
|
||||
key = self.__get_key(meta, music_type)
|
||||
with lock:
|
||||
cache_data = self._cache.get(key)
|
||||
if not cache_data and self._expires_at.pop(key, None) is not None:
|
||||
@@ -161,14 +168,14 @@ class MusicBrainzCache(metaclass=WeakSingleton):
|
||||
return cache_data
|
||||
return {}
|
||||
|
||||
def update(self, meta: MetaMusic, info: Optional[MusicInfo]) -> None:
|
||||
def update(self, meta: MetaMusic, info: Optional[MusicInfo], music_type: Optional[str] = None) -> None:
|
||||
"""
|
||||
新增或更新缓存条目，无远端身份的兜底结果也写入内存负缓存，
|
||||
避免批量识别时反复请求 MusicBrainz 触发限流
|
||||
"""
|
||||
if not meta or not info:
|
||||
return
|
||||
key = self.__get_key(meta)
|
||||
key = self.__get_key(meta, music_type)
|
||||
cache_data = info.to_dict()
|
||||
# 上游原始响应体积大且不参与身份恢复，不入缓存
|
||||
cache_data.pop("raw_data", None)
|
||||
|
||||
@@ -75,6 +75,14 @@
|
||||
“晴天”、艺术家“周杰倫”，证明中文完整短语不是零命中。
|
||||
查询字段与短语语义参考 [MusicBrainz 官方检索语法](https://musicbrainz.org/doc/Indexed_Search_Syntax)。
|
||||
- 不根据展示文本臆造 MusicBrainz ID，不更改默认 MusicBrainz 或显式单来源行为。
|
||||
- MusicBrainz 的候选确认与资源匹配复用繁简、变音符、可信别名、完整署名和录音版本规则。
|
||||
单曲不凭首词、包含关系或任意括号剥离认定同一作品；已返回的同一 ISRC 优先于名称打分。
|
||||
专辑的 `secondary_types` 可提供整专版本证据，但不能反向覆盖其中单曲的录音版本。
|
||||
- 自动识别在某个检索式返回的候选全部不匹配时继续尝试后续检索式；手动元数据目录
|
||||
浏览仍保留原始候选。Python/Rust 标题解析进入共同包装层时均保留 `[Live]` 等版本证据。
|
||||
- 音乐识别缓存按请求实体与来源身份或完整标题、署名、专辑、年份、版本、ISRC 编码，
|
||||
不使用可碰撞的分隔符拼接；旧版派生缓存自动重建，负缓存同样不跨实体复用。
|
||||
共享识别成功时，沿用现有模块回填 ABI，同时覆盖已确认实体的负缓存。
|
||||
|
||||
## 订阅与下载
|
||||
|
||||
|
||||
@@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
|
||||
| Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
|
||||
| Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
|
||||
| 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
|
||||
| 全量 mypy 历史债务 | 9,501 / 513 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
|
||||
| Ruff 历史诊断 | 542 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 全量 mypy 历史债务 | 9,500 / 513 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
|
||||
| Ruff 历史诊断 | 541 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
|
||||
|
||||
### 3.3 热点文件
|
||||
|
||||
@@ -414,6 +414,9 @@ TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`
|
||||
`shared_recognized` 和开关字段 `shared_recognize_enabled`。共享命中次数仅在共享结果驱动的二次媒体识别成功后累计。
|
||||
|
||||
音乐识别缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized` 和 `data`；条目字段包括缓存键、`media_id`、`title`、`artists`、`album`、`year`、`music_type` 和 `cover_url`。未携带远端身份的兜底负缓存仅保留在内存，不参与持久化。
|
||||
缓存键为不透明值，管理调用必须使用查询返回的原始键，不自行拼接。识别缓存按请求的
|
||||
单曲、专辑或未限定实体范围隔离，版本及 ISRC 不同的文本识别请求也不会共用结果；
|
||||
旧版未包含这些证据的派生缓存在升级后重新建立，不影响下载历史或订阅数据。
|
||||
|
||||
### 插件补充接口
|
||||
|
||||
|
||||
@@ -163,6 +163,9 @@ Call the gateway with this shape:
|
||||
browse-only. Music recognition-cache operations are administrator-only; call
|
||||
`music.cache.get` before deleting one exact key, and clear all entries only
|
||||
after explicit confirmation.
|
||||
- Cache keys are opaque. Use the exact key returned by `music.cache.get`; do not
|
||||
construct one from a title or artist. Recognition caches separate recording,
|
||||
album, and unspecified requests, as well as version and ISRC evidence.
|
||||
- Music resource metadata records applied recognition rules in `apply_words`.
|
||||
Explicit subtitle versions participate in matching. A track's `album` field
|
||||
does not prove whole-album coverage, even without a track number; keep
|
||||
|
||||
+1
-1
@@ -1942,7 +1942,7 @@
|
||||
"override": 1
|
||||
},
|
||||
"app/modules/musicbrainz/__init__.py": {
|
||||
"arg-type": 2,
|
||||
"arg-type": 1,
|
||||
"assignment": 5,
|
||||
"misc": 4,
|
||||
"no-untyped-call": 4,
|
||||
|
||||
@@ -317,9 +317,6 @@
|
||||
"app/modules/lrclib/__init__.py": {
|
||||
"I001": 1
|
||||
},
|
||||
"app/modules/musicbrainz/cache.py": {
|
||||
"I001": 1
|
||||
},
|
||||
"app/modules/navidrome/__init__.py": {
|
||||
"F401": 2,
|
||||
"I001": 1
|
||||
|
||||
@@ -435,8 +435,8 @@ def test_musicbrainz_candidate_sync_async_decision_parity() -> None:
|
||||
|
||||
assert _music_signature(sync_result) == _music_signature(async_result)
|
||||
assert sync_result and sync_result.media_id == "recording-1"
|
||||
module._search_recordings.assert_called_once_with(meta, limit=10)
|
||||
module._async_search_recordings.assert_awaited_once_with(meta, limit=10)
|
||||
module._search_recordings.assert_called_once_with(meta, limit=10, require_match=True)
|
||||
module._async_search_recordings.assert_awaited_once_with(meta, limit=10, require_match=True)
|
||||
module._search_albums.assert_not_called()
|
||||
module._async_search_albums.assert_not_awaited()
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import copy
|
||||
import pickle
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -11,7 +12,7 @@ from app.chain.search import SearchChain
|
||||
from app.domain.context import Context, MusicAlbumInfo, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.domain.meta.runtime import get_metainfo_accelerator
|
||||
from app.domain.metainfo import MetaInfo
|
||||
from app.domain.metainfo import MetaInfo, MetaInfoPath
|
||||
from app.domain.music import match_music_resource
|
||||
from app.schemas.music import MusicMeta
|
||||
from app.schemas.types import MediaType
|
||||
@@ -226,6 +227,10 @@ def test_resource_evidence_matches_python_and_native_parser_paths(monkeypatch, p
|
||||
assert live.apply_words == ["错误曲名 => 晴天"]
|
||||
album = MusicInfo(music_type="album", title="叶惠美", artists=["周杰伦"])
|
||||
assert match_music_resource(album, "周杰伦 - 晴天 FLAC", "专辑：叶惠美").reason == "partial_album"
|
||||
query = MetaMusic.parse_query("周杰伦 - 晴天 [Live] FLAC")
|
||||
assert query.version == "Live"
|
||||
filename = MetaInfoPath(Path("/music/周杰伦/叶惠美/晴天 [Live].flac"))
|
||||
assert filename.version == "Live"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("factory", [MusicInfo, MusicAlbumInfo])
|
||||
|
||||
@@ -6,7 +6,9 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import pickle
|
||||
from unittest.mock import Mock
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.api.endpoints import music as music_endpoint
|
||||
from app.domain.context import MusicInfo
|
||||
@@ -153,8 +155,19 @@ def test_music_cache_key_prefers_media_id():
|
||||
|
||||
cache.update(meta, _music_info())
|
||||
|
||||
# 缓存键取自请求元数据，携带原生 ID 时以 ID 为主身份
|
||||
assert list(cache._cache.data.keys()) == ["[音乐]rec-1-周杰伦-None-None"]
|
||||
assert next(iter(cache._cache.data)).startswith("[音乐:v2]")
|
||||
renamed = MetaMusic(title="不同展示名", artists=["不同署名"], media_source="musicbrainz", media_id="rec-1")
|
||||
assert cache.get(renamed).media_id == "rec-1"
|
||||
|
||||
|
||||
def test_music_cache_rebuilds_legacy_identity_keys(monkeypatch):
|
||||
"""旧缓存未区分版本和实体范围，升级后不恢复其中可能串用的身份。"""
|
||||
file_cache = _FileCacheStub(pickle.dumps({
|
||||
"version": 1, "items": {"[音乐]legacy": {"expires_at": 2000, "value": _music_info().to_dict()}},
|
||||
}))
|
||||
runtime_cache = _TTLCacheStub()
|
||||
_build_initialized_music_cache(monkeypatch, file_cache, runtime_cache)
|
||||
assert runtime_cache.data == {}
|
||||
|
||||
|
||||
def test_music_cache_update_and_get_roundtrip():
|
||||
@@ -174,6 +187,37 @@ def test_music_cache_update_and_get_roundtrip():
|
||||
assert "raw_data" not in stored
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field,value", [("version", "Live"), ("isrc", "USABC2600001")])
|
||||
def test_music_cache_separates_recording_identity_evidence(field, value):
|
||||
"""相同标题署名但版本或 ISRC 不同的请求不能共用识别结果。"""
|
||||
cache = _build_music_cache({})
|
||||
original = MetaMusic(title="晴天", artists=["周杰伦"])
|
||||
variant = MetaMusic.from_dict(original.to_dict())
|
||||
setattr(variant, field, value)
|
||||
cache.update(original, _music_info())
|
||||
assert cache.get(variant) is None
|
||||
|
||||
|
||||
def test_music_cache_separates_requested_entity_scope():
|
||||
"""单曲、专辑与未限定实体的请求分别缓存，负缓存也不能跨实体阻止回退。"""
|
||||
cache = _build_music_cache({})
|
||||
meta = MetaMusic(title="晴天", artists=["周杰伦"])
|
||||
cache.update(meta, _music_info(music_type="album"), music_type="album")
|
||||
cache.update(meta, MusicInfo.from_meta(meta), music_type="recording")
|
||||
assert cache.get(meta, music_type="album").music_type == "album"
|
||||
assert cache.get(meta, music_type="recording").media_id is None
|
||||
assert cache.get(meta) is None
|
||||
|
||||
|
||||
def test_music_cache_key_does_not_confuse_field_separators():
|
||||
"""名称中的连字符与多艺人分隔符不能把不同字段组合编码成同一个缓存键。"""
|
||||
cache = _build_music_cache({})
|
||||
original = MetaMusic(title="A-B", artists=["C"])
|
||||
other = MetaMusic(title="A", artists=["B-C"])
|
||||
cache.update(original, _music_info())
|
||||
assert cache.get(other) is None
|
||||
|
||||
|
||||
def test_music_cache_get_miss_returns_none():
|
||||
"""未命中的缓存查询应返回 None 而不是抛错。"""
|
||||
cache = _build_music_cache({})
|
||||
@@ -382,6 +426,32 @@ def test_module_recognize_media_hits_cache_without_search(monkeypatch):
|
||||
search_mock.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
def test_module_recording_request_cannot_reuse_album_cache(monkeypatch, async_mode):
|
||||
"""显式单曲识别不能读取此前未限定请求缓存下来的同名专辑。"""
|
||||
cache = _build_music_cache({})
|
||||
module = _build_module_with_cache(cache)
|
||||
meta = MetaMusic(title="晴天", artists=["周杰伦"])
|
||||
cache.update(meta, _music_info(music_type="album", media_id="album-1"))
|
||||
expected = _music_info()
|
||||
monkeypatch.setattr(module, "_search_recordings", Mock(return_value=[expected]))
|
||||
monkeypatch.setattr(module, "_async_search_recordings", AsyncMock(return_value=[expected]))
|
||||
if async_mode:
|
||||
result = asyncio.run(module.async_recognize_media(meta=meta, music_type="recording"))
|
||||
else:
|
||||
result = module.recognize_media(meta=meta, music_type="recording")
|
||||
assert result is expected
|
||||
assert cache.get(meta, music_type="recording").music_type == "recording"
|
||||
if async_mode:
|
||||
cached = asyncio.run(module.async_recognize_media(meta=meta, music_type="recording"))
|
||||
module._async_search_recordings.assert_awaited_once()
|
||||
else:
|
||||
cached = module.recognize_media(meta=meta, music_type="recording")
|
||||
module._search_recordings.assert_called_once()
|
||||
assert cached.media_id == expected.media_id
|
||||
assert cached.recognize_cache_hit is True
|
||||
|
||||
|
||||
def test_module_recognize_media_bypasses_cache_when_disabled(monkeypatch):
|
||||
"""cache=False 时不读取缓存，重新走搜索识别流程。"""
|
||||
cache = _build_music_cache({})
|
||||
@@ -445,3 +515,19 @@ def test_module_update_recognize_cache_only_for_musicbrainz_music():
|
||||
other_source = _music_info(media_source="theaudiodb", media_id="x-1")
|
||||
assert module.update_recognize_cache(meta=meta, mediainfo=other_source) is None
|
||||
assert module.update_recognize_cache(meta=None, mediainfo=_music_info()) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
def test_shared_recognition_replaces_entity_scoped_negative_cache(async_mode):
|
||||
"""共享回填的公开契约没有请求类型，也必须覆盖已确认单曲对应的旧负缓存。"""
|
||||
cache = _build_music_cache({})
|
||||
module = _build_module_with_cache(cache)
|
||||
meta = MetaMusic(title="晴天", artists=["周杰伦"])
|
||||
cache.update(meta, MusicInfo.from_meta(meta), music_type="recording")
|
||||
if async_mode:
|
||||
result = asyncio.run(module.async_update_recognize_cache(meta, _music_info()))
|
||||
else:
|
||||
result = module.update_recognize_cache(meta, _music_info())
|
||||
assert result is True
|
||||
assert cache.get(meta, music_type="recording").media_id == "rec-1"
|
||||
assert cache.get(meta).media_id == "rec-1"
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.domain.context import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
@@ -90,7 +92,7 @@ def test_build_query_strips_audio_quality_tokens():
|
||||
)
|
||||
)
|
||||
|
||||
# CJK 短语在 Lucene 索引中是单一词元，检索式拆为逐字 OR，OR 组带括号避免 AND 优先级歧义
|
||||
# 完整名称的繁简短语组优先，不把长标题拆成单字 OR。
|
||||
assert query == 'recording:("永远是朋友" OR "永遠是朋友") AND artist:"毛阿敏"'
|
||||
|
||||
|
||||
@@ -111,6 +113,124 @@ def test_select_candidate_matches_traditional_chinese_title():
|
||||
assert selected is candidates[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("music_type", ["recording", "album"])
|
||||
def test_recognition_accepts_trusted_title_and_artist_aliases(music_type):
|
||||
"""目录已经返回的同实体别名也必须用于身份确认，不能仅用于搜索展示排序。"""
|
||||
meta = MetaMusic(title="Fine Day", artists=["Jay Chou"])
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="candidate", music_type=music_type,
|
||||
title="晴天", title_aliases=["Fine Day"], artists=["周杰倫"], artist_aliases=["Jay Chou"])
|
||||
if music_type == "album":
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [candidate]) is candidate
|
||||
else:
|
||||
assert MusicBrainzModule._select_candidate(meta, [candidate], "musicbrainz") is candidate
|
||||
|
||||
|
||||
@pytest.mark.parametrize("artist", ["AC/DC", "Earth, Wind & Fire", "Beyoncé"])
|
||||
@pytest.mark.parametrize("music_type", ["recording", "album"])
|
||||
def test_recognition_preserves_compound_and_accented_artist_names(artist, music_type):
|
||||
"""复合艺名的解析拆段和拉丁变音符差异不应导致身份确认漏配。"""
|
||||
meta = MetaMusic.parse_query(f"{artist.replace('é', 'e')} - Example Work FLAC")
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="candidate", music_type=music_type,
|
||||
title="Example Work", artists=[artist])
|
||||
if music_type == "album":
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [candidate]) is candidate
|
||||
else:
|
||||
assert MusicBrainzModule._select_candidate(meta, [candidate], "musicbrainz") is candidate
|
||||
|
||||
|
||||
@pytest.mark.parametrize("candidate_title", ["One Tree Hill", "One - Tree Hill", "One (Other Song)"])
|
||||
def test_recording_recognition_rejects_partial_title_identity(candidate_title):
|
||||
"""单曲确认与资源匹配一样要求作品名称边界，不能借首词或任意括号剥离误配。"""
|
||||
meta = MetaMusic(title="One", artists=["U2"])
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="other", title=candidate_title, artists=["U2"])
|
||||
assert MusicBrainzModule._select_candidate(meta, [candidate], "musicbrainz") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("music_type", ["recording", "album"])
|
||||
@pytest.mark.parametrize("input_version,candidate_version", [(None, "Live"), ("Live", None), ("Live", "Remix")])
|
||||
def test_recognition_rejects_conflicting_recording_versions(music_type, input_version, candidate_version):
|
||||
"""同名同艺人的不同录音版本仍是不同目标，不能以普通标题分数自动确认。"""
|
||||
meta = MetaMusic(title="Example Work", artists=["Artist"], version=input_version)
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="candidate", music_type=music_type,
|
||||
title="Example Work", artists=["Artist"], version=candidate_version)
|
||||
if music_type == "album":
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [candidate]) is None
|
||||
else:
|
||||
assert MusicBrainzModule._select_candidate(meta, [candidate], "musicbrainz") is None
|
||||
|
||||
|
||||
def test_recording_recognition_keeps_isrc_identity_priority():
|
||||
"""来源返回相同 ISRC 时保留显式录音身份优先级，不被不完整标题和署名阻断。"""
|
||||
meta = MetaMusic(title="Unverified", artists=["Unknown"], isrc="USABC2600001")
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="recording", title="Real Title",
|
||||
artists=["Artist"], version="Live", isrc="USABC2600001")
|
||||
misleading = MusicInfo(media_source="musicbrainz", media_id="other", title="Unverified", artists=["Unknown"])
|
||||
assert MusicBrainzModule._select_candidate(meta, [misleading, candidate], "musicbrainz") is candidate
|
||||
|
||||
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
@pytest.mark.parametrize("music_type", ["recording", "album"])
|
||||
@pytest.mark.parametrize("has_match", [False, True])
|
||||
def test_recognition_continues_queries_after_rejected_candidates(monkeypatch, async_mode, music_type, has_match):
|
||||
"""与资源搜索一致，原始候选不能确认身份时继续后续检索式，而非提前宣告无匹配。"""
|
||||
module = MusicBrainzModule()
|
||||
module.cache = None
|
||||
meta = MetaMusic(title="Example Work", artists=["Artist"])
|
||||
queries = []
|
||||
|
||||
def request(_path, params):
|
||||
"""前一检索式仅有无关作品，下一检索式返回同名同署名实体。"""
|
||||
queries.append(params["query"])
|
||||
title = "Example Work" if has_match and len(queries) > 1 else "Other Work"
|
||||
items_key = "recordings" if music_type == "recording" else "release-groups"
|
||||
return {items_key: [{"id": "matched", "title": title, "artist-credit": [{"artist": {"name": "Artist"}}]}]}
|
||||
|
||||
monkeypatch.setattr(module, "_recording_queries", lambda _meta: ["first", "second", "third"])
|
||||
monkeypatch.setattr(module, "_album_queries", lambda _meta: ["first", "second", "third"])
|
||||
monkeypatch.setattr(module, "_request_json", request)
|
||||
monkeypatch.setattr(module, "_async_request_json", AsyncMock(side_effect=request))
|
||||
if async_mode:
|
||||
result = asyncio.run(module.async_recognize_media(meta=meta, music_type=music_type, cache=False))
|
||||
else:
|
||||
result = module.recognize_media(meta=meta, music_type=music_type, cache=False)
|
||||
if has_match:
|
||||
assert result and result.media_id == "matched"
|
||||
else:
|
||||
assert not result or result.media_id is None
|
||||
assert queries == (["first", "second"] if has_match else ["first", "second", "third"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
@pytest.mark.parametrize("music_type", ["recording", "album"])
|
||||
def test_catalog_browsing_keeps_unconfirmed_candidates(monkeypatch, async_mode, music_type):
|
||||
"""手动目录浏览仍展示来源原始候选，自动确认的严格规则不能抹掉浏览结果。"""
|
||||
module = MusicBrainzModule()
|
||||
meta = MetaMusic(title="Example Work", artists=["Artist"])
|
||||
key = "recordings" if music_type == "recording" else "release-groups"
|
||||
payload = {key: [{"id": "related", "title": "Other Work"}]}
|
||||
sync_request = Mock(return_value=payload)
|
||||
async_request = AsyncMock(return_value=payload)
|
||||
monkeypatch.setattr(module, "_request_json", sync_request)
|
||||
monkeypatch.setattr(module, "_async_request_json", async_request)
|
||||
if music_type == "recording":
|
||||
result = asyncio.run(module._async_search_recordings(meta, 10)) if async_mode else module._search_recordings(meta, 10)
|
||||
else:
|
||||
result = asyncio.run(module._async_search_albums(meta, 10)) if async_mode else module._search_albums(meta, 10)
|
||||
assert [item.media_id for item in result] == ["related"]
|
||||
assert sync_request.call_count == (0 if async_mode else 1)
|
||||
assert async_request.await_count == (1 if async_mode else 0)
|
||||
|
||||
|
||||
def test_album_secondary_type_is_version_evidence():
|
||||
"""专辑来源通过 secondary_types 声明现场版时，与标题和独立版本字段同样参与确认。"""
|
||||
candidate = MusicInfo(media_source="musicbrainz", media_id="live-album", music_type="album",
|
||||
title="Example Work", artists=["Artist"], secondary_types=["Live"])
|
||||
meta = MetaMusic(title="Example Work", artists=["Artist"])
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [candidate]) is None
|
||||
meta.version = "Live"
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [candidate]) is candidate
|
||||
|
||||
|
||||
def test_recording_to_info_maps_musicbrainz_payload():
|
||||
"""MusicBrainz Recording 应映射为统一 MusicInfo。"""
|
||||
info = MusicBrainzModule._recording_to_info(
|
||||
@@ -966,7 +1086,7 @@ def test_select_album_candidate_matches_colon_subtitle():
|
||||
|
||||
|
||||
def test_select_album_candidate_matches_head_title():
|
||||
"""条目「曲名-歌手《巡演名》」连字符前置命名应与资源曲名弱匹配命中。"""
|
||||
"""说明性专辑标题可弱匹配，但仍须具有相同现场版本证据。"""
|
||||
meta = MetaMusic(title="为你盛开", artists=["许巍"])
|
||||
album = MusicInfo(
|
||||
media_source="musicbrainz",
|
||||
@@ -976,6 +1096,8 @@ def test_select_album_candidate_matches_head_title():
|
||||
artists=["许巍"],
|
||||
)
|
||||
|
||||
assert MusicBrainzModule._select_album_candidate(meta, [album]) is None
|
||||
meta.version = "现场"
|
||||
matched = MusicBrainzModule._select_album_candidate(meta, [album])
|
||||
|
||||
assert matched is not None
|
||||
|
||||
Reference in New Issue
Block a user