From c356d47807a368410a608110bb230a6393771e6b Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 7 Aug 2026 23:46:54 +0800 Subject: [PATCH] fix(v3): complete music recognition and scraping --- app/api/endpoints/media.py | 45 +++++- app/api/endpoints/music.py | 19 ++- app/chain/music.py | 228 +++++++++++++++++++++++++++- app/core/config.py | 2 + app/helper/audio.py | 129 +++++++++++++++- tests/test_audio_metadata.py | 40 +++++ tests/test_media_scrape_endpoint.py | 67 +++++++- tests/test_music_chain.py | 63 +++++++- tests/test_music_endpoint.py | 21 ++- tests/test_musicbrainz_module.py | 7 + tests/test_subscribe_oper.py | 24 +++ 11 files changed, 630 insertions(+), 15 deletions(-) diff --git a/app/api/endpoints/media.py b/app/api/endpoints/media.py index df0b75ed2..253323619 100644 --- a/app/api/endpoints/media.py +++ b/app/api/endpoints/media.py @@ -1,13 +1,16 @@ from pathlib import Path from typing import Annotated, Any, List, Optional, Union +from uuid import UUID from fastapi import APIRouter, Depends from app import schemas from app.chain.media import MediaChain +from app.chain.music import MusicChain from app.chain.tmdb import TmdbChain from app.core.config import settings from app.core.context import Context +from app.core.music import MusicInfo from app.core.event import eventmanager from app.core.meta import MetaBase from app.core.metainfo import MetaInfo, MetaInfoPath @@ -23,6 +26,17 @@ router = APIRouter() MediaSource = str +def _is_valid_source_media_id(source: Optional[str], media_id: str) -> bool: + """按媒体数据源校验原生 ID，MusicBrainz 使用 UUID，其它现有来源使用数字 ID。""" + if source == "musicbrainz": + try: + UUID(media_id) + return True + except (TypeError, ValueError): + return False + return media_id.isdigit() + + def _build_recognize_metainfo( title: str, subtitle: Optional[str] = None, @@ -148,6 +162,12 @@ async def recognize_file( """ 根据文件路径识别媒体信息 """ + if MusicChain.is_audio_path(path) or source == "musicbrainz": + meta_info, media_info = await MusicChain().async_recognize_by_path( + path=path, + source=source or "musicbrainz", + ) + return Context(meta_info=meta_info, media_info=media_info).to_dict() # 识别媒体信息 context = await MediaChain().async_recognize_by_path(path, source=source) if context: @@ -255,9 +275,32 @@ def scrape( return schemas.Response( success=False, message="指定媒体ID时必须同时指定媒体数据源" ) - if normalized_media_id and not normalized_media_id.isdigit(): + if normalized_media_id and not _is_valid_source_media_id(media_source, normalized_media_id): return schemas.Response(success=False, message="媒体ID格式无效") + is_music = ( + type_name == MediaType.MUSIC + or media_source == "musicbrainz" + or MusicChain.is_audio_path(fileitem.path) + ) + if is_music: + if type_name not in (None, MediaType.MUSIC): + return schemas.Response(success=False, message="MusicBrainz 只能用于音乐刮削") + music_info: Optional[MusicInfo] = None + if normalized_media_id: + music_info = MusicChain().recognize( + source=media_source or "musicbrainz", + media_id=normalized_media_id, + ) + if not music_info: + return schemas.Response(success=False, message="刮削失败，无法识别音乐信息") + success, message = MusicChain().scrape_metadata( + fileitem=fileitem, + mediainfo=music_info, + overwrite=True, + ) + return schemas.Response(success=success, message=message) + chain = MediaChain() if normalized_media_id: meta_info = MetaInfoPath(Path(fileitem.path)) diff --git a/app/api/endpoints/music.py b/app/api/endpoints/music.py index 0946b4504..448390caf 100644 --- a/app/api/endpoints/music.py +++ b/app/api/endpoints/music.py @@ -10,6 +10,14 @@ from app.core.security import verify_token router = APIRouter() CountParam = Annotated[int, Query(ge=1, le=100)] +MusicRangeParam = Annotated[ + str, + Query(pattern="^(this_week|this_month|this_year|all_time)$"), +] +MusicSortParam = Annotated[ + str, + Query(pattern="^listen_count\\.(desc|asc)$"), +] def _serialize_music(info: MusicInfo) -> schemas.MusicInfo: @@ -59,12 +67,19 @@ async def recognize_music( async def explore_music( page: Annotated[int, Query(ge=1)] = 1, count: CountParam = 30, + range_name: MusicRangeParam = "this_month", + sort_by: MusicSortParam = "listen_count.desc", + min_listen_count: Annotated[int, Query(ge=0)] = 0, + with_cover: bool = False, _: schemas.TokenPayload = Depends(verify_token), ) -> list[schemas.MusicInfo]: - """按月度全站收听榜单分页返回可搜索和订阅的音乐候选。""" + """按周期、热度和封面条件返回可搜索和订阅的音乐候选。""" results = await MusicChain().async_chart( - range_name="this_month", + range_name=range_name, page=page, count=count, + sort_by=sort_by, + min_listen_count=min_listen_count, + with_cover=with_cover, ) return [_serialize_music(info) for info in results] diff --git a/app/chain/music.py b/app/chain/music.py index af5b80897..f318daec5 100644 --- a/app/chain/music.py +++ b/app/chain/music.py @@ -1,8 +1,16 @@ import re +from pathlib import Path +from tempfile import TemporaryDirectory from typing import Any, Iterable, Optional +from app import schemas from app.chain import ChainBase +from app.chain.storage import StorageChain +from app.core.config import settings from app.core.music import MusicInfo, MusicMeta +from app.helper.audio import AudioMetadataHelper +from app.log import logger +from app.utils.http import RequestUtils class MusicChain(ChainBase): @@ -119,15 +127,175 @@ class MusicChain(ChainBase): range_name: str, page: int = 1, count: int = 30, + sort_by: str = "listen_count.desc", + min_listen_count: int = 0, + with_cover: bool = False, ) -> list[MusicInfo]: - """异步读取 ListenBrainz 全站音乐榜单并标准化分页结果。""" + """异步读取 ListenBrainz 榜单，并应用音乐探索筛选和排序。""" candidates = await self.async_run_module( "music_chart", range_name=range_name, offset=max(page - 1, 0) * count, count=count, ) - return self.normalize_candidates(candidates, limit=count) + results = self.normalize_candidates(candidates) + if min_listen_count > 0: + results = [ + info for info in results + if (info.listen_count or 0) >= min_listen_count + ] + if with_cover: + results = [info for info in results if info.cover_url] + results.sort( + key=lambda info: info.listen_count or 0, + reverse=sort_by != "listen_count.asc", + ) + return results[:count] + + @classmethod + def is_audio_path(cls, path: str | Path) -> bool: + """判断路径是否指向系统支持的音频文件。""" + return Path(path).suffix.lower() in settings.RMT_AUDIOEXT + + @classmethod + def read_path_meta(cls, path: str | Path) -> MusicMeta: + """读取本地音频标签，不可访问时按文件名构造最小音乐元数据。""" + file_path = Path(path) + if file_path.exists() and file_path.is_file(): + return AudioMetadataHelper.read(file_path) + return cls.parse_query(file_path.stem) + + async def async_recognize_by_path( + self, + path: str | Path, + source: str = "musicbrainz", + ) -> tuple[MusicMeta, MusicInfo]: + """根据音频标签和文件名识别音乐，远端不可用时仍返回最小音乐信息。""" + meta = self.read_path_meta(path) + candidates = await self.async_run_module( + "search_music", + meta=meta, + limit=10, + ) + results = self.normalize_candidates(candidates, limit=10) + matched = self._select_path_candidate(meta, results, source=source) + if matched: + return meta, matched + return meta, self._info_from_meta(meta) + + def recognize_by_path( + self, + path: str | Path, + source: str = "musicbrainz", + ) -> tuple[MusicMeta, MusicInfo]: + """同步根据音频标签和文件名识别音乐，并保留离线最小结果。""" + meta = self.read_path_meta(path) + candidates = self.run_module("search_music", meta=meta, limit=10) + results = self.normalize_candidates(candidates, limit=10) + matched = self._select_path_candidate(meta, results, source=source) + return meta, matched or self._info_from_meta(meta) + + def scrape_metadata( + self, + fileitem: schemas.FileItem, + mediainfo: Optional[MusicInfo] = None, + overwrite: bool = True, + ) -> tuple[bool, str]: + """为音频文件或目录写入音乐标签和封面，复用现有存储下载上传能力。""" + files = self._audio_fileitems(fileitem) + if not files: + return False, "刮削路径中没有支持的音频文件" + if mediainfo and len(files) > 1: + return False, "指定 MusicBrainz ID 时仅支持刮削单个音频文件" + + failures: list[str] = [] + for audio_item in files: + info = mediainfo + if not info: + _, info = self.recognize_by_path(audio_item.path) + if not info or not info.title: + failures.append(f"{audio_item.name or audio_item.path} 无法识别音乐信息") + continue + if not self._scrape_audio_file(audio_item, info, overwrite=overwrite): + failures.append(f"{audio_item.name or audio_item.path} 标签写入失败") + if failures: + return False, "；".join(failures[:3]) + return True, f"已刮削 {len(files)} 个音频文件" + + @staticmethod + def _download_cover(url: Optional[str]) -> tuple[Optional[bytes], str]: + """通过统一请求封装下载封面，并返回图片内容与 MIME 类型。""" + if not url: + return None, "image/jpeg" + response = RequestUtils( + proxies=settings.PROXY, + ua=settings.NORMAL_USER_AGENT, + timeout=20, + ).get_res(url) + if not response: + return None, "image/jpeg" + try: + if response.status_code != 200: + logger.warning(f"音乐封面下载失败：{response.status_code} {url}") + return None, "image/jpeg" + mime = (response.headers.get("Content-Type") or "image/jpeg").split(";", 1)[0] + return response.content, mime + finally: + response.close() + + def _audio_fileitems(self, fileitem: schemas.FileItem) -> list[schemas.FileItem]: + """展开待刮削目录并过滤系统支持的音频文件。""" + if fileitem.type != "dir": + return [fileitem] if self.is_audio_path(fileitem.path or "") else [] + return [ + item + for item in StorageChain().list_files(fileitem, recursion=True) or [] + if item.type == "file" and self.is_audio_path(item.path or "") + ] + + def _scrape_audio_file( + self, + fileitem: schemas.FileItem, + mediainfo: MusicInfo, + overwrite: bool, + ) -> bool: + """下载单个音频文件、写入标签，并在远端存储场景上传覆盖原文件。""" + cover_data, cover_mime = self._download_cover(mediainfo.cover_url) + storage = StorageChain() + if fileitem.storage == "local": + local_path = storage.download_file(fileitem) + return bool( + local_path + and AudioMetadataHelper.write( + local_path, + mediainfo, + cover_data=cover_data, + cover_mime=cover_mime, + overwrite=overwrite, + ) + ) + + with TemporaryDirectory(prefix="moviepilot-music-scrape-") as temp_dir: + local_path = storage.download_file(fileitem, path=Path(temp_dir)) + if not local_path or not AudioMetadataHelper.write( + local_path, + mediainfo, + cover_data=cover_data, + cover_mime=cover_mime, + overwrite=overwrite, + ): + return False + parent = storage.get_parent_item(fileitem) + if not parent: + logger.warning(f"无法获取远端音频父目录：{fileitem.path}") + return False + return bool( + storage.upload_file( + parent, + local_path, + new_name=fileitem.name or local_path.name, + ) + ) @classmethod def to_meta(cls, info: MusicInfo) -> MusicMeta: @@ -148,6 +316,62 @@ class MusicChain(ChainBase): media_id=info.media_id, ) + @classmethod + def _select_path_candidate( + cls, + meta: MusicMeta, + 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 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: MusicMeta) -> 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 _candidate_identity(cls, info: MusicInfo) -> tuple[str, ...]: """构造跨来源稳定的候选去重键。""" diff --git a/app/core/config.py b/app/core/config.py index d869befc6..3ccc3b990 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -517,6 +517,8 @@ class ConfigModel(BaseModel): "ykimg.com", "qpic.cn", "anilist.co", + "coverartarchive.org", + "archive.org", ] ) # 图片代理允许访问的非公网 IP/CIDR，默认不放行任何非公网解析结果 diff --git a/app/helper/audio.py b/app/helper/audio.py index 0de9cc225..f8eddd3c8 100644 --- a/app/helper/audio.py +++ b/app/helper/audio.py @@ -1,14 +1,17 @@ from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, Union from mutagen import File as MutagenFile +from mutagen.flac import FLAC, Picture +from mutagen.id3 import APIC +from mutagen.mp4 import MP4, MP4Cover -from app.core.music import MusicMeta +from app.core.music import MusicInfo, MusicMeta from app.log import logger class AudioMetadataHelper: - """读取音频标签和技术参数并转换为标准 MusicMeta。""" + """读取和写入音频标签，并转换为标准音乐元数据。""" @classmethod def read(cls, path: Path) -> MusicMeta: @@ -50,6 +53,126 @@ class AudioMetadataHelper: isrc=cls._first(tags, "isrc"), ) + @classmethod + def write( + cls, + path: Path, + music: Union[MusicMeta, MusicInfo], + cover_data: Optional[bytes] = None, + cover_mime: str = "image/jpeg", + overwrite: bool = True, + ) -> bool: + """把标准音乐字段写入音频标签，并为常见格式嵌入专辑封面。""" + try: + audio = MutagenFile(path, easy=True) + if not audio: + logger.warning(f"无法写入音频标签：{path}") + return False + if audio.tags is None: + audio.add_tags() + for key, value in cls._tag_values(music).items(): + if value in (None, "", []): + continue + if not overwrite and audio.tags.get(key): + continue + try: + audio[key] = value if isinstance(value, list) else [str(value)] + except (KeyError, TypeError, ValueError) as err: + logger.debug(f"音频格式不支持标签 {key}：{path} - {err}") + audio.save() + if cover_data: + cls._write_cover( + path=path, + cover_data=cover_data, + cover_mime=cover_mime, + overwrite=overwrite, + ) + return True + except Exception as err: + logger.warning(f"写入音频标签失败：{path} - {err}") + return False + + @classmethod + def _tag_values(cls, music: Union[MusicMeta, MusicInfo]) -> dict[str, Any]: + """把标准音乐对象转换为 Mutagen Easy 标签字典。""" + track_number = cls._number_text( + getattr(music, "track_number", None), + getattr(music, "total_tracks", None), + ) + disc_number = cls._number_text( + getattr(music, "disc_number", None), + getattr(music, "total_discs", None), + ) + return { + "title": getattr(music, "title", None), + "artist": list(getattr(music, "artists", None) or []), + "album": getattr(music, "album", None), + "albumartist": getattr(music, "album_artist", None), + "date": getattr(music, "year", None), + "tracknumber": track_number, + "discnumber": disc_number, + "isrc": getattr(music, "isrc", None), + } + + @staticmethod + def _number_text(current: Optional[int], total: Optional[int]) -> Optional[str]: + """把曲序或碟号转换为常见的 current/total 标签文本。""" + if current is None: + return None + return f"{current}/{total}" if total else str(current) + + @staticmethod + def _write_cover( + path: Path, + cover_data: bytes, + cover_mime: str, + overwrite: bool, + ) -> None: + """为 MP3、FLAC 和 MP4/M4A 写入内嵌封面，其它格式保留标签写入结果。""" + audio = MutagenFile(path) + if isinstance(audio, FLAC): + if audio.pictures and not overwrite: + return + picture = Picture() + picture.type = 3 + picture.mime = cover_mime + picture.desc = "Cover" + picture.data = cover_data + if overwrite: + audio.clear_pictures() + audio.add_picture(picture) + audio.save() + return + if isinstance(audio, MP4): + if audio.tags is None: + audio.add_tags() + if audio.tags.get("covr") and not overwrite: + return + image_format = ( + MP4Cover.FORMAT_PNG + if cover_mime == "image/png" + else MP4Cover.FORMAT_JPEG + ) + audio.tags["covr"] = [MP4Cover(cover_data, imageformat=image_format)] + audio.save() + return + tags = getattr(audio, "tags", None) + if tags is not None and hasattr(tags, "add"): + if tags.getall("APIC") and not overwrite: + return + if overwrite: + tags.delall("APIC") + tags.add( + APIC( + encoding=3, + mime=cover_mime, + type=3, + desc="Cover", + data=cover_data, + ) + ) + audio.save() + @staticmethod def _values(tags: Any, key: str) -> list[str]: """从 Mutagen Easy 标签中提取非空字符串列表。""" diff --git a/tests/test_audio_metadata.py b/tests/test_audio_metadata.py index b399abf8c..a062aa340 100644 --- a/tests/test_audio_metadata.py +++ b/tests/test_audio_metadata.py @@ -1,6 +1,7 @@ from pathlib import Path from types import SimpleNamespace +from app.core.music import MusicInfo from app.helper.audio import AudioMetadataHelper @@ -46,3 +47,42 @@ def test_read_audio_metadata_falls_back_to_filename(monkeypatch): assert meta.title == "Unknown Track" assert meta.audio_format == "MP3" + + +def test_write_audio_metadata_maps_music_info_to_easy_tags(monkeypatch): + """音乐刮削应把标准歌曲、专辑和曲序字段写回音频标签。""" + class FakeAudio: + """记录 Mutagen Easy 标签写入结果。""" + + def __init__(self): + self.tags = {} + self.saved = False + + def __setitem__(self, key, value): + self.tags[key] = value + + def save(self): + self.saved = True + + audio = FakeAudio() + monkeypatch.setattr("app.helper.audio.MutagenFile", lambda *_args, **_kwargs: audio) + + success = AudioMetadataHelper.write( + Path("/music/08 - Get Lucky.flac"), + MusicInfo( + title="Get Lucky", + artists=["Daft Punk", "Pharrell Williams"], + album="Random Access Memories", + album_artist="Daft Punk", + year=2013, + track_number=8, + total_tracks=13, + isrc="USQX91300105", + ), + ) + + assert success is True + assert audio.saved is True + assert audio.tags["title"] == ["Get Lucky"] + assert audio.tags["artist"] == ["Daft Punk", "Pharrell Williams"] + assert audio.tags["tracknumber"] == ["8/13"] diff --git a/tests/test_media_scrape_endpoint.py b/tests/test_media_scrape_endpoint.py index 9c8fa1e59..ece1df244 100644 --- a/tests/test_media_scrape_endpoint.py +++ b/tests/test_media_scrape_endpoint.py @@ -1,8 +1,9 @@ -from unittest.mock import Mock, patch +from unittest.mock import AsyncMock, Mock, patch -from app.api.endpoints.media import scrape +from app.api.endpoints.media import recognize_file, scrape from app.core.context import Context, MediaInfo from app.core.meta import MetaBase +from app.core.music import MusicInfo, MusicMeta from app.schemas import FileItem, MediaType @@ -80,3 +81,65 @@ def test_scrape_rejects_media_id_without_source() -> None: assert result.success is False assert result.message == "指定媒体ID时必须同时指定媒体数据源" + + +def test_recognize_file_routes_audio_to_music_chain() -> None: + """文件管理识别音频文件时应返回音乐专属上下文。""" + music_chain = Mock() + music_chain.async_recognize_by_path = AsyncMock( + return_value=( + MusicMeta(title="晴天", artists=["周杰伦"]), + MusicInfo( + source="musicbrainz", + media_id="977e6978-139d-425c-bb98-6b0c62d1e45e", + title="晴天", + artists=["周杰伦"], + ), + ) + ) + + import asyncio + + with patch("app.api.endpoints.media.MusicChain", return_value=music_chain): + result = asyncio.run(recognize_file(path="/music/晴天.flac", _=Mock())) + + assert result["meta_info"]["type"] == "音乐" + assert result["media_info"]["title"] == "晴天" + music_chain.async_recognize_by_path.assert_awaited_once_with( + path="/music/晴天.flac", + source="musicbrainz", + ) + + +def test_scrape_music_uses_musicbrainz_uuid_and_music_scraper() -> None: + """手动音乐刮削应接受 MusicBrainz UUID 并进入音乐标签写入流程。""" + fileitem = FileItem(storage="local", path="/music/晴天.flac", type="file") + info = MusicInfo( + source="musicbrainz", + media_id="977e6978-139d-425c-bb98-6b0c62d1e45e", + title="晴天", + ) + chain = Mock() + chain.recognize.return_value = info + chain.scrape_metadata.return_value = (True, "已刮削 1 个音频文件") + + with patch("app.api.endpoints.media.MusicChain", return_value=chain): + result = scrape( + fileitem=fileitem, + storage="local", + media_source="musicbrainz", + media_id="977e6978-139d-425c-bb98-6b0c62d1e45e", + type_name=MediaType.MUSIC, + _=Mock(), + ) + + assert result.success is True + chain.recognize.assert_called_once_with( + source="musicbrainz", + media_id="977e6978-139d-425c-bb98-6b0c62d1e45e", + ) + chain.scrape_metadata.assert_called_once_with( + fileitem=fileitem, + mediainfo=info, + overwrite=True, + ) diff --git a/tests/test_music_chain.py b/tests/test_music_chain.py index b690b5d18..3a2bde03f 100644 --- a/tests/test_music_chain.py +++ b/tests/test_music_chain.py @@ -1,5 +1,5 @@ from app.chain.music import MusicChain -from app.core.music import MusicInfo +from app.core.music import MusicInfo, MusicMeta def test_parse_query_supports_artist_title_format(): @@ -110,3 +110,64 @@ def test_chart_converts_page_to_listenbrainz_offset(monkeypatch): "count": 30, } assert len(results) == 1 + + +def test_async_chart_applies_music_explore_filters(monkeypatch): + """音乐探索应按收听次数、封面条件和升序设置筛选榜单。""" + chain = MusicChain() + + async def fake_async_run_module(method, **kwargs): + """返回包含不同热度和封面状态的榜单候选。""" + assert method == "music_chart" + return [ + MusicInfo(media_id="1", source="musicbrainz", title="A", listen_count=300), + MusicInfo( + media_id="2", + source="musicbrainz", + title="B", + listen_count=120, + cover_url="https://coverartarchive.org/release/2/front-500", + ), + MusicInfo( + media_id="3", + source="musicbrainz", + title="C", + listen_count=240, + cover_url="https://coverartarchive.org/release/3/front-500", + ), + ] + + monkeypatch.setattr(chain, "async_run_module", fake_async_run_module) + + import asyncio + + results = asyncio.run( + chain.async_chart( + range_name="this_month", + count=30, + sort_by="listen_count.asc", + min_listen_count=100, + with_cover=True, + ) + ) + + assert [item.title for item in results] == ["B", "C"] + + +def test_select_path_candidate_prefers_matching_audio_tags(): + """文件识别应优先选择标题、艺术家和专辑均匹配的 MusicBrainz 候选。""" + meta = MusicMeta(title="晴天", artists=["周杰伦"], album="叶惠美") + candidates = [ + MusicInfo(source="musicbrainz", media_id="1", title="晴天", artists=["其他歌手"]), + MusicInfo( + source="musicbrainz", + media_id="2", + title="晴天", + artists=["周杰伦"], + album="叶惠美", + ), + ] + + selected = MusicChain._select_path_candidate(meta, candidates, source="musicbrainz") + + assert selected is candidates[1] diff --git a/tests/test_music_endpoint.py b/tests/test_music_endpoint.py index 73b8e6084..b9ac23411 100644 --- a/tests/test_music_endpoint.py +++ b/tests/test_music_endpoint.py @@ -90,8 +90,8 @@ def test_recognize_music_returns_404_for_unknown_item(): assert error.value.status_code == 404 -def test_explore_music_serializes_monthly_chart(): - """音乐探索接口应按月度榜单分页并保留收听统计。""" +def test_explore_music_forwards_filters_and_serializes_chart(): + """音乐探索接口应传递周期、排序、热度和封面筛选条件。""" chain = Mock() chain.async_chart = AsyncMock( return_value=[ @@ -106,11 +106,24 @@ def test_explore_music_serializes_monthly_chart(): ) with patch("app.api.endpoints.music.MusicChain", return_value=chain): - result = asyncio.run(explore_music(page=2, count=20, _=Mock())) + result = asyncio.run( + explore_music( + page=2, + count=20, + range_name="this_week", + sort_by="listen_count.asc", + min_listen_count=100, + with_cover=True, + _=Mock(), + ) + ) assert result[0].listen_count == 123 chain.async_chart.assert_awaited_once_with( - range_name="this_month", + range_name="this_week", page=2, count=20, + sort_by="listen_count.asc", + min_listen_count=100, + with_cover=True, ) diff --git a/tests/test_musicbrainz_module.py b/tests/test_musicbrainz_module.py index ddcac7488..83bf917c6 100644 --- a/tests/test_musicbrainz_module.py +++ b/tests/test_musicbrainz_module.py @@ -1,5 +1,12 @@ from app.core.music import MusicMeta from app.modules.musicbrainz import MusicBrainzModule +from app.core.config import settings + + +def test_musicbrainz_cover_domains_are_allowed_by_image_proxy(): + """MusicBrainz 封面及其归档重定向域名应进入图片代理安全列表。""" + assert "coverartarchive.org" in settings.SECURITY_IMAGE_DOMAINS + assert "archive.org" in settings.SECURITY_IMAGE_DOMAINS def test_build_query_uses_structured_music_fields(): diff --git a/tests/test_subscribe_oper.py b/tests/test_subscribe_oper.py index 0fb6001d9..cb75ef30d 100644 --- a/tests/test_subscribe_oper.py +++ b/tests/test_subscribe_oper.py @@ -8,6 +8,7 @@ import pytest from app.db.models.subscribe import Subscribe from app.db.models.subscribehistory import SubscribeHistory from app.db.subscribe_oper import SubscribeOper +from app.core.music import MusicInfo from app.schemas.types import MediaType @@ -96,6 +97,29 @@ def test_add_scopes_duplicate_lookup_by_episode_group(episode_group): created.create.assert_called_once() +def test_music_subscribe_persists_release_cover_as_poster_and_backdrop(): + """音乐订阅应把 MusicBrainz 发行封面写入订阅海报和背景字段。""" + persisted = SimpleNamespace(id=92) + created = SimpleNamespace(create=MagicMock()) + media = MusicInfo( + source="musicbrainz", + media_id="977e6978-139d-425c-bb98-6b0c62d1e45e", + title="晴天", + cover_url="https://coverartarchive.org/release-group/example/front-500", + ) + + with patch("app.db.subscribe_oper.Subscribe") as subscribe_model: + subscribe_model.exists.side_effect = [None, persisted] + subscribe_model.return_value = created + + sid, _ = SubscribeOper(db=object()).add(mediainfo=media, season=None) + + assert sid == 92 + payload = subscribe_model.call_args.kwargs + assert payload["poster"] == media.cover_url + assert payload["backdrop"] == media.cover_url + + @pytest.mark.parametrize("episode_group", [None, "eg-1"]) def test_async_add_scopes_duplicate_lookup_by_episode_group(episode_group): """异步新增与同步路径使用相同的剧集组身份契约。"""