mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-19 22:14:28 +08:00
796 lines
32 KiB
Python
796 lines
32 KiB
Python
import re
|
||
import threading
|
||
import time
|
||
from typing import Any, Iterable, Optional, Tuple, Union
|
||
|
||
from fastapi.concurrency import run_in_threadpool
|
||
|
||
from app.core.cache import cached
|
||
from app.core.config import settings
|
||
from app.core.context import (
|
||
MUSIC_ENTITY_ALBUM,
|
||
MusicAlbumInfo,
|
||
MusicArtistInfo,
|
||
MusicInfo,
|
||
MusicRelease,
|
||
)
|
||
from app.core.meta import MetaBase, MetaMusic
|
||
from app.log import logger
|
||
from app.modules import _ModuleBase
|
||
from app.schemas.types import MediaRecognizeType, MediaType, ModuleType
|
||
from app.utils.http import RequestUtils
|
||
|
||
|
||
class MusicBrainzModule(_ModuleBase):
|
||
"""通过 MusicBrainz 提供音乐元数据搜索和详情识别。"""
|
||
|
||
_source = "musicbrainz"
|
||
_base_url = "https://musicbrainz.org/ws/2"
|
||
_detail_url = "https://musicbrainz.org/recording"
|
||
_album_detail_url = "https://musicbrainz.org/release-group"
|
||
_artist_detail_url = "https://musicbrainz.org/artist"
|
||
_cover_url = "https://coverartarchive.org/release-group"
|
||
_request_interval = 1.0
|
||
_request_lock = threading.Lock()
|
||
_last_request_at = 0.0
|
||
# 关联艺术家按关系可读性排序，纪念性质的致敬关系数量庞大且价值低，放到最后
|
||
_artist_relation_priority = (
|
||
"member of band",
|
||
"subgroup",
|
||
"collaboration",
|
||
"founder",
|
||
"artist rename",
|
||
"supporting musician",
|
||
"conductor position",
|
||
"involved with",
|
||
"teacher",
|
||
"sibling",
|
||
"parent",
|
||
"married",
|
||
)
|
||
# 艺术家外链只保留对用户有意义的官方与流媒体入口
|
||
_artist_link_types = (
|
||
"official homepage",
|
||
"wikidata",
|
||
"wikipedia",
|
||
"discogs",
|
||
"allmusic",
|
||
"social network",
|
||
"free streaming",
|
||
"streaming",
|
||
"youtube",
|
||
"purchase for download",
|
||
)
|
||
|
||
def init_module(self) -> None:
|
||
"""初始化无状态的 MusicBrainz 模块。"""
|
||
|
||
def init_setting(self) -> Optional[Tuple[str, Union[str, bool]]]:
|
||
"""MusicBrainz 无需独立密钥或启用开关。"""
|
||
return None
|
||
|
||
def stop(self) -> None:
|
||
"""停止模块；当前实现没有需要释放的持久资源。"""
|
||
|
||
def test(self) -> Tuple[bool, str]:
|
||
"""测试 MusicBrainz 搜索接口连通性。"""
|
||
result = self._request_json(
|
||
"/recording",
|
||
params={"query": "recording:test", "limit": 1, "fmt": "json"},
|
||
)
|
||
return (True, "") if result is not None else (False, "MusicBrainz 网络连接失败")
|
||
|
||
@staticmethod
|
||
def get_name() -> str:
|
||
"""返回模块展示名称。"""
|
||
return "MusicBrainz"
|
||
|
||
@staticmethod
|
||
def get_type() -> ModuleType:
|
||
"""返回模块所属的媒体识别类型。"""
|
||
return ModuleType.MediaRecognize
|
||
|
||
@staticmethod
|
||
def get_subtype() -> MediaRecognizeType:
|
||
"""返回 MusicBrainz 模块子类型。"""
|
||
return MediaRecognizeType.MusicBrainz
|
||
|
||
@staticmethod
|
||
def get_priority() -> int:
|
||
"""音乐识别在所有 MediaRecognize 模块中最先响应，避免音乐请求被影视模块误识别。"""
|
||
return 0
|
||
|
||
def search_music(self, meta: MetaMusic, limit: int = 20) -> list[MusicInfo]:
|
||
"""搜索单曲、专辑和艺术家，并交错返回可浏览的 MusicBrainz 候选。"""
|
||
normalized_limit = max(1, min(limit, 100))
|
||
recordings = self._search_recordings(meta, limit=normalized_limit)
|
||
albums = self._search_albums(meta, limit=normalized_limit)
|
||
artists = self._search_artists(meta, limit=normalized_limit)
|
||
return self._interleave_results(
|
||
recordings,
|
||
albums,
|
||
artists,
|
||
limit=normalized_limit,
|
||
)
|
||
|
||
def _search_recordings(self, meta: MetaMusic, limit: int) -> list[MusicInfo]:
|
||
"""按音频标签条件搜索 Recording，供全局搜索和文件识别复用。"""
|
||
query = self._build_query(meta)
|
||
if not query:
|
||
return []
|
||
payload = self._request_json(
|
||
"/recording",
|
||
params={"query": query, "limit": max(1, min(limit, 100)), "fmt": "json"},
|
||
)
|
||
return [
|
||
info
|
||
for item in (payload or {}).get("recordings") or []
|
||
if (info := self._recording_to_info(item))
|
||
]
|
||
|
||
def _search_albums(self, meta: MetaMusic, limit: int) -> list[MusicInfo]:
|
||
"""按标题和可选艺术家搜索 Release Group 专辑候选。"""
|
||
title = meta.album or meta.title
|
||
if not title:
|
||
return []
|
||
clauses = [f'releasegroup:"{self._escape_query(title)}"']
|
||
if meta.artists:
|
||
clauses.append(f'artist:"{self._escape_query(meta.artists[0])}"')
|
||
payload = self._request_json(
|
||
"/release-group",
|
||
params={
|
||
"query": " AND ".join(clauses),
|
||
"limit": max(1, min(limit, 100)),
|
||
"fmt": "json",
|
||
},
|
||
)
|
||
return [
|
||
album.to_music_info()
|
||
for item in (payload or {}).get("release-groups") or []
|
||
if (album := self._release_group_to_album(item))
|
||
]
|
||
|
||
def _search_artists(self, meta: MetaMusic, limit: int) -> list[MusicInfo]:
|
||
"""按用户输入中的艺术家部分搜索 Artist 浏览候选。"""
|
||
artist_name = meta.artists[0] if meta.artists else meta.title
|
||
if not artist_name:
|
||
return []
|
||
payload = self._request_json(
|
||
"/artist",
|
||
params={
|
||
"query": f'artist:"{self._escape_query(artist_name)}"',
|
||
"limit": max(1, min(limit, 100)),
|
||
"fmt": "json",
|
||
},
|
||
)
|
||
return [
|
||
artist.to_music_info()
|
||
for item in (payload or {}).get("artists") or []
|
||
if (artist := self._artist_to_info(item, include_raw=True))
|
||
]
|
||
|
||
@staticmethod
|
||
def _interleave_results(*groups: list[MusicInfo], limit: int) -> list[MusicInfo]:
|
||
"""按实体轮询合并结果，避免单曲数量占满全局搜索页。"""
|
||
results: list[MusicInfo] = []
|
||
index = 0
|
||
while len(results) < limit and any(index < len(group) for group in groups):
|
||
for group in groups:
|
||
if index < len(group):
|
||
results.append(group[index])
|
||
if len(results) >= limit:
|
||
break
|
||
index += 1
|
||
return results
|
||
|
||
def recognize_media(
|
||
self,
|
||
meta: MetaBase = None,
|
||
mtype: MediaType = None,
|
||
source: Optional[str] = None,
|
||
mediaid: Optional[str] = None,
|
||
**kwargs,
|
||
) -> Optional[MusicInfo]:
|
||
"""跟随统一媒体识别分发，仅在音乐类型请求下返回 MusicBrainz 识别结果。"""
|
||
# 非音乐请求交给影视识别模块，不占用识别管线
|
||
if not isinstance(meta, MetaMusic) and mtype != MediaType.MUSIC and source != self._source:
|
||
return None
|
||
# 无 MetaMusic 元数据时仅响应本数据源的详情识别请求
|
||
if not isinstance(meta, MetaMusic):
|
||
if source == self._source and mediaid:
|
||
return self.recognize_music(source, str(mediaid))
|
||
return None
|
||
# 携带数据源与原生 ID 的请求优先按详情识别
|
||
resolved_source = source or meta.media_source
|
||
if resolved_source and (mediaid or meta.media_id):
|
||
info = self.recognize_music(resolved_source, str(mediaid or meta.media_id))
|
||
if info:
|
||
return info
|
||
# 无身份时按标题搜索并挑选可信候选，检索不到时返回元数据兜底
|
||
# 文件识别只能从 Recording 中挑选，专辑或艺术家同名结果不能成为音轨身份。
|
||
candidates = self._search_recordings(meta, limit=10)
|
||
matched = self._select_candidate(meta, candidates, source=resolved_source or self._source)
|
||
return matched or self._info_from_meta(meta)
|
||
|
||
async def async_recognize_media(
|
||
self,
|
||
meta: MetaBase = None,
|
||
mtype: MediaType = None,
|
||
source: Optional[str] = None,
|
||
mediaid: Optional[str] = None,
|
||
**kwargs,
|
||
) -> Optional[MusicInfo]:
|
||
"""同步分发到音乐识别的异步版本，避免阻塞共享事件循环。"""
|
||
return await run_in_threadpool(
|
||
self.recognize_media,
|
||
meta,
|
||
mtype=mtype,
|
||
source=source,
|
||
mediaid=mediaid,
|
||
**kwargs,
|
||
)
|
||
|
||
@classmethod
|
||
def _select_candidate(cls, meta: MetaMusic, candidates: Iterable[MusicInfo], source: str) -> Optional[MusicInfo]:
|
||
"""按标题、艺术家和专辑匹配度选择最可信的搜索候选。"""
|
||
normalized_source = cls._normalize_text(source).casefold()
|
||
ranked: list[tuple[int, MusicInfo]] = []
|
||
for candidate in candidates:
|
||
if normalized_source and (candidate.source or "").casefold() != normalized_source:
|
||
continue
|
||
score = 0
|
||
if meta.title and cls._same_text(meta.title, candidate.title):
|
||
score += 4
|
||
if meta.artists and any(
|
||
cls._same_text(meta.artists[0], artist)
|
||
for artist in candidate.artists
|
||
):
|
||
score += 3
|
||
if meta.album and cls._same_text(meta.album, candidate.album):
|
||
score += 2
|
||
if meta.isrc and cls._same_text(meta.isrc, candidate.isrc):
|
||
score += 5
|
||
ranked.append((score, candidate))
|
||
if not ranked:
|
||
return None
|
||
ranked.sort(key=lambda item: item[0], reverse=True)
|
||
return ranked[0][1] if ranked[0][0] > 0 else None
|
||
|
||
@classmethod
|
||
def _info_from_meta(cls, meta: MetaMusic) -> MusicInfo:
|
||
"""音乐识别无候选时，把元数据转换为可展示的最小信息。"""
|
||
return MusicInfo(
|
||
source=meta.media_source,
|
||
media_id=meta.media_id,
|
||
title=meta.title,
|
||
artists=list(meta.artists),
|
||
album=meta.album,
|
||
album_artist=meta.album_artist,
|
||
year=meta.year,
|
||
disc_number=meta.disc_number,
|
||
track_number=meta.track_number,
|
||
total_tracks=meta.total_tracks,
|
||
duration=meta.duration,
|
||
isrc=meta.isrc,
|
||
version=meta.version,
|
||
names=[name for name in (meta.title, meta.album) if name],
|
||
)
|
||
|
||
@classmethod
|
||
def _same_text(cls, left: Optional[str], right: Optional[str]) -> bool:
|
||
"""忽略大小写比较两个音乐文本字段。"""
|
||
return cls._normalize_text(left).casefold() == cls._normalize_text(right).casefold()
|
||
|
||
@classmethod
|
||
def _normalize_text(cls, value: Optional[str]) -> str:
|
||
"""清理音乐检索文本中的多余空白。"""
|
||
return re.sub(r"\s+", " ", str(value or "")).strip()
|
||
|
||
def recognize_music(self, source: str, media_id: str) -> Optional[MusicInfo]:
|
||
"""按 MusicBrainz 标准 ID 获取音乐详情，单曲不存在时回退到专辑。"""
|
||
if source != self._source or not media_id:
|
||
return None
|
||
payload = self._request_json(
|
||
f"/recording/{media_id}",
|
||
params={
|
||
"inc": "artists+releases+release-groups+isrcs+genres",
|
||
"fmt": "json",
|
||
},
|
||
)
|
||
if payload:
|
||
return self._recording_to_info(payload)
|
||
# MusicBrainz 各实体共用 UUID 形式，统一详情入口在 Recording 未命中后继续探测专辑。
|
||
album = self.music_album(source, media_id)
|
||
return album.to_music_info() if album else None
|
||
|
||
def music_album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]:
|
||
"""按 MusicBrainz Release Group ID 获取标准化专辑详情及曲目。"""
|
||
if source != self._source or not media_id:
|
||
return None
|
||
payload = self._request_json(
|
||
f"/release-group/{media_id}",
|
||
params={
|
||
"inc": "artists+releases+media+genres+tags+ratings",
|
||
"fmt": "json",
|
||
},
|
||
)
|
||
if not payload:
|
||
return None
|
||
album = self._release_group_to_album(payload)
|
||
if not album:
|
||
return None
|
||
album.releases = self._release_variants(payload.get("releases") or [])
|
||
album.tracks = self._album_tracks(album, payload.get("releases") or [])
|
||
return album
|
||
|
||
def music_artist(self, source: str, media_id: str) -> Optional[MusicArtistInfo]:
|
||
"""按 MusicBrainz Artist ID 获取标准化艺术家详情。"""
|
||
if source != self._source or not media_id:
|
||
return None
|
||
payload = self._request_json(
|
||
f"/artist/{media_id}",
|
||
params={"inc": "url-rels+genres+tags+aliases", "fmt": "json"},
|
||
)
|
||
return self._artist_to_info(payload) if payload else None
|
||
|
||
def music_artist_albums(
|
||
self,
|
||
source: str,
|
||
media_id: str,
|
||
page: int = 1,
|
||
count: int = 30,
|
||
album_type: Optional[str] = None,
|
||
) -> list[MusicInfo]:
|
||
"""按 MusicBrainz Artist ID 分页浏览该艺术家的专辑、EP 和单曲。"""
|
||
if source != self._source or not media_id:
|
||
return []
|
||
limit = max(1, min(count, 100))
|
||
params: dict[str, Any] = {
|
||
"artist": media_id,
|
||
"inc": "artist-credits",
|
||
"limit": limit,
|
||
"offset": max(page - 1, 0) * limit,
|
||
"fmt": "json",
|
||
}
|
||
if album_type:
|
||
params["type"] = album_type
|
||
payload = self._request_json("/release-group", params=params)
|
||
albums = [
|
||
album
|
||
for item in (payload or {}).get("release-groups") or []
|
||
if (album := self._release_group_to_album(item))
|
||
]
|
||
# MusicBrainz 浏览接口不支持排序，只能在当前页内按发行日期倒序，保证首页是最新作品
|
||
albums.sort(key=lambda item: item.release_date or "", reverse=True)
|
||
return [album.to_music_info() for album in albums]
|
||
|
||
def music_artist_related(
|
||
self,
|
||
source: str,
|
||
media_id: str,
|
||
count: int = 24,
|
||
) -> list[MusicArtistInfo]:
|
||
"""按 MusicBrainz 艺术家关系返回可继续浏览的关联艺术家。"""
|
||
if source != self._source or not media_id:
|
||
return []
|
||
payload = self._request_json(
|
||
f"/artist/{media_id}",
|
||
params={"inc": "artist-rels", "fmt": "json"},
|
||
)
|
||
if not payload:
|
||
return []
|
||
return self._related_artists(payload.get("relations") or [], count=count)
|
||
|
||
@classmethod
|
||
def _build_query(cls, meta: MetaMusic) -> str:
|
||
"""构造 MusicBrainz Recording 搜索表达式。"""
|
||
clauses = []
|
||
if meta.title:
|
||
clauses.append(f'recording:"{cls._escape_query(meta.title)}"')
|
||
if meta.artists:
|
||
clauses.append(f'artist:"{cls._escape_query(meta.artists[0])}"')
|
||
if meta.album:
|
||
clauses.append(f'release:"{cls._escape_query(meta.album)}"')
|
||
if meta.isrc:
|
||
clauses.append(f'isrc:"{cls._escape_query(meta.isrc)}"')
|
||
return " AND ".join(clauses)
|
||
|
||
@staticmethod
|
||
def _escape_query(value: str) -> str:
|
||
"""转义 MusicBrainz 查询中的引号和反斜线。"""
|
||
return value.replace("\\", "\\\\").replace('"', '\\"').strip()
|
||
|
||
@classmethod
|
||
def _recording_to_info(cls, recording: dict[str, Any]) -> Optional[MusicInfo]:
|
||
"""将 MusicBrainz Recording 响应转换为标准音乐信息。"""
|
||
media_id = recording.get("id")
|
||
title = recording.get("title")
|
||
if not media_id or not title:
|
||
return None
|
||
releases = recording.get("releases") or []
|
||
release = cls._select_release(releases)
|
||
release_group = (release or {}).get("release-group") or {}
|
||
release_date = cls._release_date(recording, release)
|
||
album = (release or {}).get("title")
|
||
artists, artist_ids = cls._artist_credits(recording.get("artist-credit"))
|
||
album_artists, _ = cls._artist_credits((release or {}).get("artist-credit"))
|
||
category_parts = [release_group.get("primary-type")]
|
||
category_parts.extend(release_group.get("secondary-types") or [])
|
||
return MusicInfo(
|
||
source=cls._source,
|
||
media_id=str(media_id),
|
||
title=str(title),
|
||
artists=artists,
|
||
artist_ids=artist_ids,
|
||
album=album,
|
||
album_artist=" / ".join(album_artists) if album_artists else None,
|
||
album_id=str(release_group["id"]) if release_group.get("id") else None,
|
||
album_type=release_group.get("primary-type"),
|
||
year=cls._year(release_date),
|
||
release_date=release_date,
|
||
duration=cls._duration_seconds(recording.get("length")),
|
||
isrc=next(iter(recording.get("isrcs") or []), None),
|
||
cover_url=cls._build_cover_url(release_group.get("id")),
|
||
version=recording.get("disambiguation") or None,
|
||
category=" / ".join(str(part) for part in category_parts if part),
|
||
genres=cls._names_of(recording.get("genres")),
|
||
names=[name for name in (title, album) if name],
|
||
detail_link=f"{cls._detail_url}/{media_id}",
|
||
raw_data=recording,
|
||
)
|
||
|
||
@classmethod
|
||
def _release_group_to_album(cls, release_group: dict[str, Any]) -> Optional[MusicAlbumInfo]:
|
||
"""将 MusicBrainz Release Group 响应转换为标准专辑信息。"""
|
||
media_id = release_group.get("id")
|
||
title = release_group.get("title")
|
||
if not media_id or not title:
|
||
return None
|
||
artists, artist_ids = cls._artist_credits(release_group.get("artist-credit"))
|
||
rating = release_group.get("rating") or {}
|
||
return MusicAlbumInfo(
|
||
source=cls._source,
|
||
media_id=str(media_id),
|
||
title=str(title),
|
||
artists=artists,
|
||
artist_ids=artist_ids,
|
||
album_type=release_group.get("primary-type"),
|
||
secondary_types=[str(item) for item in release_group.get("secondary-types") or []],
|
||
release_date=release_group.get("first-release-date") or None,
|
||
cover_url=cls._build_cover_url(media_id),
|
||
genres=cls._names_of(release_group.get("genres")),
|
||
tags=cls._names_of(release_group.get("tags")),
|
||
# MusicBrainz 评分是 5 分制，统一放大到与影视一致的 10 分制展示
|
||
rating=round(float(rating["value"]) * 2, 1) if rating.get("value") else 0.0,
|
||
rating_votes=rating.get("votes-count"),
|
||
detail_link=f"{cls._album_detail_url}/{media_id}",
|
||
raw_data=release_group,
|
||
)
|
||
|
||
@classmethod
|
||
def _release_variants(cls, releases: list[dict[str, Any]]) -> list[MusicRelease]:
|
||
"""整理同一专辑下的发行版本，供详情页对比介质和地区。"""
|
||
variants = []
|
||
for release in releases:
|
||
if not release.get("id"):
|
||
continue
|
||
media = release.get("media") or []
|
||
variants.append(
|
||
MusicRelease(
|
||
media_id=str(release["id"]),
|
||
title=release.get("title"),
|
||
date=release.get("date") or None,
|
||
country=release.get("country") or None,
|
||
status=release.get("status") or None,
|
||
packaging=release.get("packaging") or None,
|
||
formats=[str(item["format"]) for item in media if item.get("format")],
|
||
track_count=sum(int(item.get("track-count") or 0) for item in media) or None,
|
||
)
|
||
)
|
||
variants.sort(key=lambda item: (cls._date_sort_key(item.date), item.title or ""))
|
||
return variants
|
||
|
||
@staticmethod
|
||
def _artist_credits(
|
||
artist_credit: Optional[list[dict[str, Any]]],
|
||
) -> Tuple[list[str], list[str]]:
|
||
"""从 MusicBrainz artist-credit 提取有序艺术家名称和按位置对齐的标准 ID。"""
|
||
names: list[str] = []
|
||
ids: list[str] = []
|
||
for credit in artist_credit or []:
|
||
artist = credit.get("artist") or {}
|
||
name = artist.get("name") or credit.get("name")
|
||
if not name or str(name) in names:
|
||
continue
|
||
names.append(str(name))
|
||
# 名称与 ID 按下标一一对应，缺少 ID 时补空串，前端据此决定是否可跳转
|
||
ids.append(str(artist.get("id") or ""))
|
||
return names, ids
|
||
|
||
@staticmethod
|
||
def _names_of(items: Optional[list[dict[str, Any]]]) -> list[str]:
|
||
"""提取 MusicBrainz 风格、标签或别名列表的名称，热度高的排在前面。"""
|
||
entries = [item for item in items or [] if item.get("name")]
|
||
entries.sort(key=lambda item: (-int(item.get("count") or 0), str(item["name"])))
|
||
return [str(item["name"]) for item in entries]
|
||
|
||
@classmethod
|
||
def _select_track_release(cls, releases: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""选择曲目最完整且发行最早的正式版本，作为专辑曲目来源。"""
|
||
candidates = [
|
||
release
|
||
for release in releases
|
||
if release.get("id")
|
||
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")), {})
|
||
official = [release for release in candidates if release.get("status") == "Official"]
|
||
return min(
|
||
official or candidates,
|
||
key=lambda release: cls._date_sort_key(release.get("date")),
|
||
)
|
||
|
||
@classmethod
|
||
def _album_tracks(
|
||
cls,
|
||
album: MusicAlbumInfo,
|
||
releases: list[dict[str, Any]],
|
||
) -> list[MusicInfo]:
|
||
"""读取专辑代表性发行版本的曲目，作为专辑内音乐列表。"""
|
||
release = cls._select_track_release(releases)
|
||
if not release.get("id"):
|
||
return []
|
||
payload = cls._request_json(
|
||
f"/release/{release['id']}",
|
||
params={"inc": "recordings+artist-credits", "fmt": "json"},
|
||
)
|
||
tracks: list[MusicInfo] = []
|
||
for medium in (payload or {}).get("media") or []:
|
||
for track in medium.get("tracks") or []:
|
||
info = cls._track_to_info(album, medium, track)
|
||
if info:
|
||
tracks.append(info)
|
||
return tracks
|
||
|
||
@classmethod
|
||
def _track_to_info(
|
||
cls,
|
||
album: MusicAlbumInfo,
|
||
medium: dict[str, Any],
|
||
track: dict[str, Any],
|
||
) -> Optional[MusicInfo]:
|
||
"""将发行版本中的单条曲目转换为可继续浏览的音乐信息。"""
|
||
recording = track.get("recording") or {}
|
||
media_id = recording.get("id")
|
||
title = track.get("title") or recording.get("title")
|
||
if not media_id or not title:
|
||
return None
|
||
artists, artist_ids = cls._artist_credits(
|
||
track.get("artist-credit") or recording.get("artist-credit")
|
||
)
|
||
return MusicInfo(
|
||
source=cls._source,
|
||
media_id=str(media_id),
|
||
title=str(title),
|
||
artists=artists or list(album.artists),
|
||
artist_ids=artist_ids or list(album.artist_ids),
|
||
album=album.title,
|
||
album_artist=album.artist or None,
|
||
album_id=album.media_id,
|
||
album_type=album.album_type,
|
||
year=album.year,
|
||
release_date=recording.get("first-release-date") or album.release_date,
|
||
disc_number=cls._optional_int(medium.get("position")),
|
||
track_number=cls._optional_int(track.get("position")),
|
||
total_tracks=cls._optional_int(medium.get("track-count")),
|
||
duration=cls._duration_seconds(track.get("length") or recording.get("length")),
|
||
cover_url=album.cover_url,
|
||
version=recording.get("disambiguation") or None,
|
||
category=album.category,
|
||
names=[str(title)],
|
||
detail_link=f"{cls._detail_url}/{media_id}",
|
||
)
|
||
|
||
@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
|
||
return min(
|
||
candidates,
|
||
key=lambda release: cls._date_sort_key(release.get("date")),
|
||
)
|
||
|
||
@staticmethod
|
||
def _date_sort_key(value: Optional[str]) -> tuple[int, str]:
|
||
"""将完整或不完整发行日期转换为稳定排序键。"""
|
||
return (0, value) if value else (1, "")
|
||
|
||
@staticmethod
|
||
def _release_date(recording: dict[str, Any], release: dict[str, Any]) -> Optional[str]:
|
||
"""从录音和发行信息中选择最可靠的发行日期。"""
|
||
return recording.get("first-release-date") or release.get("date")
|
||
|
||
@staticmethod
|
||
def _year(release_date: Optional[str]) -> Optional[int]:
|
||
"""从 MusicBrainz 的可变精度日期提取年份。"""
|
||
if not release_date:
|
||
return None
|
||
try:
|
||
return int(release_date[:4])
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _duration_seconds(value: Any) -> Optional[int]:
|
||
"""将 MusicBrainz 毫秒时长转换为整数秒。"""
|
||
try:
|
||
return round(int(value) / 1000) if value is not None else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _optional_int(value: Any) -> Optional[int]:
|
||
"""将 MusicBrainz 的碟号、音轨号等计数字段转换为可选整数。"""
|
||
try:
|
||
return int(value) if value not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@classmethod
|
||
def _artist_to_info(
|
||
cls,
|
||
artist: dict[str, Any],
|
||
relation: Optional[str] = None,
|
||
include_raw: bool = False,
|
||
) -> Optional[MusicArtistInfo]:
|
||
"""将 MusicBrainz Artist 响应转换为标准艺术家信息。"""
|
||
media_id = artist.get("id")
|
||
name = artist.get("name")
|
||
if not media_id or not name:
|
||
return None
|
||
life_span = artist.get("life-span") or {}
|
||
area = artist.get("area") or {}
|
||
begin_area = artist.get("begin-area") or {}
|
||
relations = artist.get("relations") or []
|
||
return MusicArtistInfo(
|
||
source=cls._source,
|
||
media_id=str(media_id),
|
||
name=str(name),
|
||
sort_name=artist.get("sort-name") or None,
|
||
disambiguation=artist.get("disambiguation") or None,
|
||
artist_type=artist.get("type") or None,
|
||
gender=artist.get("gender") or None,
|
||
country=artist.get("country") or None,
|
||
area=area.get("name") or begin_area.get("name") or None,
|
||
begin_date=life_span.get("begin") or None,
|
||
end_date=life_span.get("end") or None,
|
||
ended=bool(life_span.get("ended")),
|
||
genres=cls._names_of(artist.get("genres")),
|
||
tags=cls._names_of(artist.get("tags")),
|
||
aliases=cls._names_of(artist.get("aliases")),
|
||
relation=relation,
|
||
image_url=cls._artist_image(relations),
|
||
detail_link=f"{cls._artist_detail_url}/{media_id}",
|
||
external_links=cls._artist_links(relations),
|
||
raw_data=artist if include_raw else {},
|
||
)
|
||
|
||
@classmethod
|
||
def _artist_image(cls, relations: list[dict[str, Any]]) -> Optional[str]:
|
||
"""从艺术家 image 关系解析可直接展示的图片地址。"""
|
||
for relation in relations:
|
||
if relation.get("type") != "image":
|
||
continue
|
||
resource = (relation.get("url") or {}).get("resource") or ""
|
||
# MusicBrainz 记录的是维基共享资源页地址，需要转成文件直链才能展示
|
||
if "commons.wikimedia.org/wiki/File:" in resource:
|
||
file_name = resource.rsplit("File:", 1)[-1]
|
||
return (
|
||
"https://commons.wikimedia.org/wiki/Special:FilePath/"
|
||
f"{file_name}?width=500"
|
||
)
|
||
if resource:
|
||
return resource
|
||
return None
|
||
|
||
@classmethod
|
||
def _artist_links(cls, relations: list[dict[str, Any]]) -> dict[str, str]:
|
||
"""整理艺术家可对外跳转的官方与流媒体链接。"""
|
||
links: dict[str, str] = {}
|
||
for relation in relations:
|
||
relation_type = str(relation.get("type") or "")
|
||
resource = (relation.get("url") or {}).get("resource")
|
||
if relation_type in cls._artist_link_types and resource:
|
||
links.setdefault(relation_type, str(resource))
|
||
return links
|
||
|
||
@classmethod
|
||
def _related_artists(
|
||
cls,
|
||
relations: list[dict[str, Any]],
|
||
count: int,
|
||
) -> list[MusicArtistInfo]:
|
||
"""按关系类型优先级整理关联艺术家，并按来源去重。"""
|
||
ranked: list[tuple[int, MusicArtistInfo]] = []
|
||
seen: set[str] = set()
|
||
fallback_priority = len(cls._artist_relation_priority)
|
||
for relation in relations:
|
||
if relation.get("target-type") != "artist":
|
||
continue
|
||
artist = relation.get("artist") or {}
|
||
artist_id = str(artist.get("id") or "")
|
||
if not artist_id or artist_id in seen:
|
||
continue
|
||
relation_type = str(relation.get("type") or "")
|
||
info = cls._artist_to_info(artist, relation=relation_type or None)
|
||
if not info:
|
||
continue
|
||
seen.add(artist_id)
|
||
priority = (
|
||
cls._artist_relation_priority.index(relation_type)
|
||
if relation_type in cls._artist_relation_priority
|
||
else fallback_priority
|
||
)
|
||
ranked.append((priority, info))
|
||
ranked.sort(key=lambda item: (item[0], item[1].name or ""))
|
||
return [info for _, info in ranked[: max(1, count)]]
|
||
|
||
@classmethod
|
||
def _build_cover_url(cls, release_group_id: Optional[str]) -> Optional[str]:
|
||
"""根据 Release Group ID 构造 Cover Art Archive 封面地址。"""
|
||
if not release_group_id:
|
||
return None
|
||
# 支持配置音乐封面代理地址，解决 coverartarchive.org 无法访问的问题
|
||
base = (settings.MUSIC_COVER_PROXY or "https://coverartarchive.org").rstrip("/")
|
||
return f"{base}/release-group/{release_group_id}/front-500"
|
||
|
||
@classmethod
|
||
def _wait_for_rate_limit(cls) -> None:
|
||
"""串行控制 MusicBrainz 公共接口的最小请求间隔。"""
|
||
with cls._request_lock:
|
||
now = time.monotonic()
|
||
remaining = cls._request_interval - (now - cls._last_request_at)
|
||
if remaining > 0:
|
||
time.sleep(remaining)
|
||
cls._last_request_at = time.monotonic()
|
||
|
||
@classmethod
|
||
@cached(maxsize=settings.CONF.musicbrainz, ttl=settings.CONF.meta, skip_none=True)
|
||
def _request_json(
|
||
cls,
|
||
path: str,
|
||
params: Optional[dict[str, Any]] = None,
|
||
) -> Optional[dict[str, Any]]:
|
||
"""请求 MusicBrainz JSON 接口并统一处理网络和响应错误。"""
|
||
cls._wait_for_rate_limit()
|
||
response = RequestUtils(
|
||
headers={
|
||
"User-Agent": f"{settings.USER_AGENT} (https://github.com/jxxghp/MoviePilot)",
|
||
"Accept": "application/json",
|
||
},
|
||
proxies=settings.PROXY,
|
||
timeout=20,
|
||
).get_res(f"{cls._base_url}{path}", params=params)
|
||
if not response:
|
||
return None
|
||
try:
|
||
if response.status_code == 404:
|
||
# 单曲与专辑共用同一套 ID 入口，404 属于正常的探测结果
|
||
logger.debug(f"MusicBrainz 资源不存在：{path}")
|
||
return None
|
||
if response.status_code != 200:
|
||
logger.warning(
|
||
f"MusicBrainz 请求失败：{response.status_code} {response.text[:200]}"
|
||
)
|
||
return None
|
||
return response.json()
|
||
except (TypeError, ValueError) as err:
|
||
logger.warning(f"MusicBrainz 响应解析失败：{err}")
|
||
return None
|
||
finally:
|
||
response.close()
|