mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-03 13:55:00 +08:00
fix(music): scope and rank manual search results
This commit is contained in:
@@ -34,7 +34,7 @@ from app.schemas.event import MediaSourceInfo as _SchemaMediaSourceInfo
|
||||
from app.schemas.media import normalize_media_source, resolve_media_identity
|
||||
from app.schemas.response import Response as _SchemaResponse
|
||||
from app.schemas.token import TokenPayload as _SchemaTokenPayload
|
||||
from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource, MediaType
|
||||
from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource, MediaType, MusicEntityType
|
||||
from app.schemas.workflow import Context as _SchemaContext
|
||||
from app.schemas.workflow import FileItem as _SchemaFileItem
|
||||
from app.schemas.workflow import MediaInfo as _SchemaMediaInfo
|
||||
@@ -285,6 +285,7 @@ async def search(
|
||||
page: int = 1,
|
||||
count: int = 8,
|
||||
media_source: MediaSourceQuery = (),
|
||||
music_type: Optional[MusicEntityType] = None,
|
||||
_: _SchemaTokenPayload = Depends(verify_token),
|
||||
) -> Any:
|
||||
"""
|
||||
@@ -295,6 +296,7 @@ async def search(
|
||||
:param page: 页码
|
||||
:param count: 每页数量
|
||||
:param media_source: 请求级搜索数据源枚举;可重复传入,逗号格式仅用于兼容旧客户端
|
||||
:param music_type: 可选音乐实体类型,用于限制 MusicBrainz 只搜索单曲、发行组或艺术家
|
||||
:param _: Token校验
|
||||
:return: 搜索结果列表
|
||||
"""
|
||||
@@ -318,12 +320,19 @@ async def search(
|
||||
|
||||
media_chain = MediaChain()
|
||||
is_music = type == "music" or any(is_music_media_source(source) for source in selected_sources)
|
||||
if type == "media" or is_music:
|
||||
_, medias = await media_chain.async_search(
|
||||
title=title, media_source=source_selection,
|
||||
**({"mtype": MediaType.MUSIC, "limit": count} if is_music else {}),
|
||||
if is_music:
|
||||
music_results = await media_chain.async_search_music(
|
||||
query=title,
|
||||
limit=count,
|
||||
media_source=source_selection,
|
||||
music_types=(music_type,) if music_type else None,
|
||||
)
|
||||
result = [media.to_dict() for media in medias] if medias else []
|
||||
result = [media.to_dict() for media in music_results] if music_results else []
|
||||
elif type == "media":
|
||||
_media_meta, media_results = await media_chain.async_search(
|
||||
title=title, media_source=source_selection,
|
||||
)
|
||||
result = [media.to_dict() for media in media_results] if media_results else []
|
||||
elif type == "collection":
|
||||
collections = await media_chain.async_search_collections(name=title, media_source=source_selection)
|
||||
result = [collection.to_dict() for collection in collections] if collections else []
|
||||
|
||||
@@ -7,7 +7,7 @@ from typing import Any, Callable, Iterable, Optional
|
||||
from app.domain.context import MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.schemas.media import normalize_media_source
|
||||
from app.schemas.types import MediaSource, MediaSourceSelection
|
||||
from app.schemas.types import MediaSource, MediaSourceSelection, MusicEntityType
|
||||
|
||||
|
||||
class MusicCatalogService:
|
||||
@@ -81,16 +81,21 @@ class MusicCatalogService:
|
||||
query: str | MetaMusic,
|
||||
limit: int = 20,
|
||||
media_source: Optional[MediaSourceSelection] = None,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""顺序搜索一个或多个音乐来源,隔离单一来源失败。"""
|
||||
meta = query if isinstance(query, MetaMusic) else MetaMusic.parse_query(query)
|
||||
normalized_music_types = tuple(music_types) if music_types is not None else None
|
||||
candidates = []
|
||||
for source in self.search_sources(media_source):
|
||||
chain = self._source_resolver(source)
|
||||
if not chain:
|
||||
continue
|
||||
try:
|
||||
candidates.append(chain.search_music(meta, limit=limit))
|
||||
search_options: dict[str, Any] = {"limit": limit}
|
||||
if normalized_music_types is not None:
|
||||
search_options["music_types"] = normalized_music_types
|
||||
candidates.append(chain.search_music(meta, **search_options))
|
||||
except Exception as error:
|
||||
self._warning(f"音乐来源 {source} 搜索失败:{str(error)}")
|
||||
return self.merge_sources(candidates, limit=limit)
|
||||
@@ -100,14 +105,16 @@ class MusicCatalogService:
|
||||
query: str | MetaMusic,
|
||||
limit: int = 20,
|
||||
media_source: Optional[MediaSourceSelection] = None,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""并行搜索一个或多个音乐来源,隔离单一来源失败。"""
|
||||
meta = query if isinstance(query, MetaMusic) else MetaMusic.parse_query(query)
|
||||
normalized_music_types = tuple(music_types) if music_types is not None else None
|
||||
searches = []
|
||||
for source in self.search_sources(media_source):
|
||||
chain = self._source_resolver(source)
|
||||
if chain:
|
||||
searches.append(self._async_search_source(chain, source, meta, limit))
|
||||
searches.append(self._async_search_source(chain, source, meta, limit, normalized_music_types))
|
||||
source_results = await asyncio.gather(*searches) if searches else []
|
||||
return self.merge_sources(source_results, limit=limit)
|
||||
|
||||
@@ -124,10 +131,14 @@ class MusicCatalogService:
|
||||
source: MediaSource,
|
||||
meta: MetaMusic,
|
||||
limit: int,
|
||||
music_types: Optional[Iterable[MusicEntityType]],
|
||||
) -> list[MusicInfo]:
|
||||
"""异步搜索单个来源,并把异常降级为空候选。"""
|
||||
try:
|
||||
return await chain.async_search_music(meta, limit=limit)
|
||||
search_options: dict[str, Any] = {"limit": limit}
|
||||
if music_types is not None:
|
||||
search_options["music_types"] = music_types
|
||||
return await chain.async_search_music(meta, **search_options)
|
||||
except Exception as error:
|
||||
self._warning(f"音乐来源 {source} 搜索失败:{str(error)}")
|
||||
return []
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Iterable, List, Optional
|
||||
|
||||
from app.chain.base import ChainBase
|
||||
from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.schemas.context import MediaPerson as _SchemaMediaPerson
|
||||
from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType
|
||||
from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType, MusicEntityType
|
||||
|
||||
|
||||
class DoubanChain(ChainBase):
|
||||
@@ -14,7 +14,12 @@ class DoubanChain(ChainBase):
|
||||
|
||||
music_source = MediaSource.DoubanMusic
|
||||
|
||||
def search_music(self, meta: MetaMusic, limit: int = 20) -> list[MusicInfo]:
|
||||
def search_music(
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int = 20,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""按音乐元数据搜索豆瓣音乐候选。"""
|
||||
result = self.run_module(
|
||||
"search_music",
|
||||
@@ -22,12 +27,15 @@ class DoubanChain(ChainBase):
|
||||
limit=limit,
|
||||
media_source=self.music_source,
|
||||
)
|
||||
return self._music_infos(result, limit=limit)
|
||||
infos = self._music_infos(result, limit=limit)
|
||||
selected_types = set(music_types or [])
|
||||
return [info for info in infos if info.music_type in selected_types] if selected_types else infos
|
||||
|
||||
async def async_search_music(
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int = 20,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""异步按音乐元数据搜索豆瓣音乐候选。"""
|
||||
result = await self.async_run_module(
|
||||
@@ -36,7 +44,9 @@ class DoubanChain(ChainBase):
|
||||
limit=limit,
|
||||
media_source=self.music_source,
|
||||
)
|
||||
return self._music_infos(result, limit=limit)
|
||||
infos = self._music_infos(result, limit=limit)
|
||||
selected_types = set(music_types or [])
|
||||
return [info for info in infos if info.music_type in selected_types] if selected_types else infos
|
||||
|
||||
def recognize_music(
|
||||
self,
|
||||
|
||||
@@ -22,7 +22,7 @@ from app.schemas.types import (
|
||||
MUSIC_ENTITY_ALBUM,
|
||||
MediaSource,
|
||||
MediaSourceSelection,
|
||||
MediaType,
|
||||
MusicEntityType,
|
||||
)
|
||||
|
||||
|
||||
@@ -100,20 +100,32 @@ class MediaCatalogOwner(_MediaOwnerBase):
|
||||
query: str,
|
||||
limit: int = 20,
|
||||
media_source: Optional[MediaSourceSelection] = None,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""按一个或多个音乐来源搜索候选,未指定时使用 MusicBrainz。"""
|
||||
_, candidates = self.search(title=query, media_source=media_source, mtype=MediaType.MUSIC, limit=limit)
|
||||
return cast(list[MusicInfo], candidates)
|
||||
meta = MetaMusic.parse_query(query)
|
||||
return self._music_catalog().search(
|
||||
meta,
|
||||
limit=limit,
|
||||
media_source=media_source,
|
||||
music_types=music_types,
|
||||
)
|
||||
|
||||
async def async_search_music(
|
||||
self,
|
||||
query: str,
|
||||
limit: int = 20,
|
||||
media_source: Optional[MediaSourceSelection] = None,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""并行搜索一个或多个音乐来源,单一来源失败不影响其它结果。"""
|
||||
_, candidates = await self.async_search(title=query, media_source=media_source, mtype=MediaType.MUSIC, limit=limit)
|
||||
return cast(list[MusicInfo], candidates)
|
||||
meta = MetaMusic.parse_query(query)
|
||||
return await self._music_catalog().async_search(
|
||||
meta,
|
||||
limit=limit,
|
||||
media_source=media_source,
|
||||
music_types=music_types,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _validate_music_result(
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
from typing import Any, Optional
|
||||
from typing import Any, Iterable, Optional
|
||||
|
||||
from app.chain.base import ChainBase
|
||||
from app.domain.context import MusicAlbumInfo, MusicArtistInfo, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.schemas.types import MediaSource, MediaType
|
||||
from app.schemas.types import MediaSource, MediaType, MusicEntityType
|
||||
|
||||
|
||||
class MusicMetadataSourceChain(ChainBase):
|
||||
@@ -11,25 +11,35 @@ class MusicMetadataSourceChain(ChainBase):
|
||||
|
||||
source: MediaSource
|
||||
|
||||
def search_music(self, meta: MetaMusic, limit: int = 20) -> list[MusicInfo]:
|
||||
def search_music(
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int = 20,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""按音乐元数据搜索当前来源候选。"""
|
||||
result = self.run_module(
|
||||
"search_music",
|
||||
meta=meta,
|
||||
limit=limit,
|
||||
media_source=self.source,
|
||||
)
|
||||
return self._music_infos(result, limit=limit)
|
||||
kwargs: dict[str, Any] = {"meta": meta, "limit": limit, "media_source": self.source}
|
||||
if self.source == MediaSource.MusicBrainz and music_types is not None:
|
||||
kwargs["music_types"] = music_types
|
||||
result = self.run_module("search_music", **kwargs)
|
||||
infos = self._music_infos(result, limit=limit)
|
||||
selected_types = set(music_types or [])
|
||||
return [info for info in infos if info.music_type in selected_types] if selected_types else infos
|
||||
|
||||
async def async_search_music(self, meta: MetaMusic, limit: int = 20) -> list[MusicInfo]:
|
||||
async def async_search_music(
|
||||
self,
|
||||
meta: MetaMusic,
|
||||
limit: int = 20,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> list[MusicInfo]:
|
||||
"""异步按音乐元数据搜索当前来源候选。"""
|
||||
result = await self.async_run_module(
|
||||
"search_music",
|
||||
meta=meta,
|
||||
limit=limit,
|
||||
media_source=self.source,
|
||||
)
|
||||
return self._music_infos(result, limit=limit)
|
||||
kwargs: dict[str, Any] = {"meta": meta, "limit": limit, "media_source": self.source}
|
||||
if self.source == MediaSource.MusicBrainz and music_types is not None:
|
||||
kwargs["music_types"] = music_types
|
||||
result = await self.async_run_module("search_music", **kwargs)
|
||||
infos = self._music_infos(result, limit=limit)
|
||||
selected_types = set(music_types or [])
|
||||
return [info for info in infos if info.music_type in selected_types] if selected_types else infos
|
||||
|
||||
def recognize_music(
|
||||
self,
|
||||
|
||||
@@ -42,12 +42,15 @@ from app.schemas.category import (
|
||||
)
|
||||
from app.schemas.types import (
|
||||
MUSIC_ENTITY_ALBUM,
|
||||
MUSIC_ENTITY_ARTIST,
|
||||
MUSIC_ENTITY_RECORDING,
|
||||
MUSIC_ENTITY_TYPES,
|
||||
MediaRecognizeType,
|
||||
MediaSource,
|
||||
MediaSourceSelection,
|
||||
MediaType,
|
||||
ModuleType,
|
||||
MusicEntityType,
|
||||
)
|
||||
|
||||
|
||||
@@ -366,20 +369,46 @@ class MusicBrainzModule(_ModuleBase):
|
||||
meta: MetaMusic,
|
||||
limit: int = 20,
|
||||
media_source: Optional[MediaSourceSelection] = None,
|
||||
music_types: Optional[Iterable[MusicEntityType]] = None,
|
||||
) -> Optional[list[MusicInfo]]:
|
||||
"""搜索单曲、专辑和艺术家,并交错返回可浏览的 MusicBrainz 候选。"""
|
||||
"""按请求的音乐实体搜索 MusicBrainz,多实体时交错返回候选。"""
|
||||
if not is_media_source_selected(media_source, self._source):
|
||||
return None
|
||||
normalized_limit = max(1, min(limit, 100))
|
||||
requested_types = {
|
||||
music_type
|
||||
for music_type in (music_types or MUSIC_ENTITY_TYPES)
|
||||
if music_type in MUSIC_ENTITY_TYPES
|
||||
}
|
||||
if not requested_types:
|
||||
return []
|
||||
# 中文逐字召回可能包含很多局部命中,扩大单次窗口后按完整名称重新排序。
|
||||
fetch_limit = min(100, normalized_limit * 3) if self._QUERY_CJK_RE.search(meta.title or "") else normalized_limit
|
||||
recordings = self._rank_search_candidates(meta, self._search_recordings(meta, limit=fetch_limit))
|
||||
albums = self._rank_search_candidates(meta, self._search_albums(meta, limit=fetch_limit))
|
||||
artists = self._rank_search_candidates(meta, self._search_artists(meta, limit=fetch_limit))
|
||||
# 同名发行组远多于一页;专辑实体单独搜索时扩大召回,再用年份、艺人与发行类型重排。
|
||||
album_fetch_limit = min(100, normalized_limit * 5) if requested_types == {MUSIC_ENTITY_ALBUM} else fetch_limit
|
||||
recordings = (
|
||||
self._rank_search_candidates(meta, self._search_recordings(meta, limit=fetch_limit))
|
||||
if MUSIC_ENTITY_RECORDING in requested_types else []
|
||||
)
|
||||
albums = (
|
||||
self._rank_album_search_candidates(meta, self._search_albums(meta, limit=album_fetch_limit))
|
||||
if MUSIC_ENTITY_ALBUM in requested_types else []
|
||||
)
|
||||
artists = (
|
||||
self._rank_search_candidates(meta, self._search_artists(meta, limit=fetch_limit))
|
||||
if MUSIC_ENTITY_ARTIST in requested_types else []
|
||||
)
|
||||
groups = {
|
||||
MUSIC_ENTITY_RECORDING: recordings,
|
||||
MUSIC_ENTITY_ALBUM: albums,
|
||||
MUSIC_ENTITY_ARTIST: artists,
|
||||
}
|
||||
return self._interleave_results(
|
||||
recordings,
|
||||
albums,
|
||||
artists,
|
||||
*(groups[music_type] for music_type in (
|
||||
MUSIC_ENTITY_RECORDING,
|
||||
MUSIC_ENTITY_ALBUM,
|
||||
MUSIC_ENTITY_ARTIST,
|
||||
) if music_type in requested_types),
|
||||
limit=normalized_limit,
|
||||
)
|
||||
|
||||
@@ -403,6 +432,31 @@ class MusicBrainzModule(_ModuleBase):
|
||||
|
||||
return sorted(candidates, key=score, reverse=True)
|
||||
|
||||
@staticmethod
|
||||
def _rank_album_search_candidates(meta: MetaMusic, candidates: list[MusicInfo]) -> list[MusicInfo]:
|
||||
"""按完整专辑名、艺人、年份和发行类型排序 Release Group 候选。"""
|
||||
expected = music_text_key(meta.album or meta.title)
|
||||
expected_artists = {music_text_key(artist) for artist in meta.artists}
|
||||
|
||||
def score(info: MusicInfo) -> tuple[bool, bool, bool, bool, bool, float]:
|
||||
names = [info.title, *(info.title_aliases or [])]
|
||||
keys = [music_text_key(name) for name in names if name]
|
||||
candidate_artists = {
|
||||
music_text_key(artist)
|
||||
for artist in [*info.artists, *(info.artist_aliases or [])]
|
||||
}
|
||||
similarity = max((SequenceMatcher(None, expected, key).ratio() for key in keys), default=0.0)
|
||||
return (
|
||||
expected in keys,
|
||||
bool(expected_artists and expected_artists & candidate_artists),
|
||||
bool(meta.year and info.year and int(meta.year) == int(info.year)),
|
||||
str(info.album_type or "").casefold() == "album",
|
||||
not bool(info.secondary_types),
|
||||
similarity,
|
||||
)
|
||||
|
||||
return sorted(candidates, key=score, reverse=True)
|
||||
|
||||
def _search_recordings(self, meta: MetaMusic, limit: int, require_match: bool = False) -> list[MusicInfo]:
|
||||
"""查询 Recording;自动识别须确认身份才停止,手动浏览仍保留原始候选。"""
|
||||
for query in self._recording_queries(meta):
|
||||
|
||||
@@ -102,7 +102,7 @@ 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,441 / 513 文件 | Agent API 重构后的现状基线;canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 |
|
||||
| 全量 mypy 历史债务 | 9,440 / 513 文件 | Agent API 重构后的现状基线;canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 |
|
||||
| Ruff 历史诊断 | 537 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 覆盖率固定基线 | Application 80.00%,Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
|
||||
|
||||
|
||||
@@ -275,7 +275,7 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`,不再返
|
||||
|
||||
| 方法 | 路径 | 说明 |
|
||||
| :--- | :--- | :--- |
|
||||
| GET | `/api/v1/media/search` | 按标题搜索媒体、合集、人物或音乐,参数:`title`、`type`、`page`、`count`,可重复传入可选 `media_source`;内置模块只处理自身支持的来源,插件模块可以处理其注册的扩展来源,旧客户端的逗号格式仅在输入边界兼容 |
|
||||
| GET | `/api/v1/media/search` | 按标题搜索媒体、合集、人物或音乐,参数:`title`、`type`、`page`、`count`,可重复传入可选 `media_source`;音乐搜索可用 `music_type` 限定单曲、专辑或艺术家实体;内置模块只处理自身支持的来源,插件模块可以处理其注册的扩展来源,旧客户端的逗号格式仅在输入边界兼容 |
|
||||
| GET | `/api/v1/media/recognize` | 识别标题,参数:`title`、`subtitle`、`custom_words`,可选 `media_source`;当 `title` 为含目录的媒体文件路径时,会合并父目录中的名称、年份等信息 |
|
||||
| GET | `/api/v1/media/recognize_file` | 识别文件路径,参数:`path`,可选 `media_source` |
|
||||
| GET | `/api/v1/media/{media_id}` | 按原生 ID 查询影视或音乐详情;必填参数:`media_source`、`type_name`,其中 `media_source` 与路径中的 `media_id` 组成统一媒体身份,`type_name` 支持电影、电视剧和音乐 |
|
||||
@@ -351,7 +351,7 @@ SSE 的 `candidate_items` 是站点原始返回数量,`match_counts` 记录身
|
||||
|
||||
| 方法 | 路径 | 说明 |
|
||||
| :--- | :--- | :--- |
|
||||
| GET | `/api/v1/media/search` | 当 `type=music` 或指定音乐 `media_source` 时按歌曲、专辑或歌手关键词搜索音乐元数据,参数:`title`、`type`、`count`、可重复的 `media_source` 枚举 |
|
||||
| GET | `/api/v1/media/search` | 当 `type=music` 或指定音乐 `media_source` 时按歌曲、专辑或歌手关键词搜索音乐元数据,参数:`title`、`type`、`count`、可重复的 `media_source` 枚举,以及可选的 `music_type` 实体过滤 |
|
||||
| POST | `/api/v1/music/recognize` | 按 `media_source` + `media_id` 识别音乐详情,请求体:`MusicRecognizeRequest` |
|
||||
| GET | `/api/v1/music/explore` | 按来源浏览音乐;`media_source=musicbrainz` 支持 `mode=chart|fresh` 榜单与新发行,`media_source=doubanmusic` 固定按官方标签分类浏览,使用 `tags` 和 `douban_sort=U|S|R|O` 筛选。其它参数:`entity=recording|album`、`range_name`、`sort_by`、`sort`、`days`、`past`、`future`、`min_listen_count`、`with_cover`、`page`、`count` |
|
||||
| GET | `/api/v1/music/album/{album_id}` | 按来源专辑 ID 查询专辑详情、完整曲目和发行版本,参数:`media_source` |
|
||||
|
||||
@@ -457,7 +457,7 @@
|
||||
},
|
||||
"app/api/endpoints/media.py": {
|
||||
"arg-type": 2,
|
||||
"assignment": 8,
|
||||
"assignment": 7,
|
||||
"attr-defined": 3,
|
||||
"misc": 13,
|
||||
"no-untyped-call": 5,
|
||||
|
||||
@@ -71,7 +71,7 @@ def test_media_search_endpoint_forwards_multi_source() -> None:
|
||||
async def test_media_search_route_accepts_comma_separated_music_sources() -> None:
|
||||
"""真实路由在旧逗号格式兼容边界后应把每项转换为 MediaSource。"""
|
||||
chain = Mock()
|
||||
chain.async_search = AsyncMock(return_value=(None, []))
|
||||
chain.async_search_music = AsyncMock(return_value=[])
|
||||
app = FastAPI()
|
||||
app.include_router(media_endpoints.router, prefix="/api/v1/media")
|
||||
app.dependency_overrides[verify_token] = lambda: Mock()
|
||||
@@ -87,16 +87,17 @@ async def test_media_search_route_accepts_comma_separated_music_sources() -> Non
|
||||
"title": "周杰伦",
|
||||
"type": "music",
|
||||
"count": 30,
|
||||
"music_type": "album",
|
||||
"media_source": "musicbrainz,theaudiodb,doubanmusic",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"success": True, "message": "", "data": []}
|
||||
chain.async_search.assert_awaited_once_with(
|
||||
title="周杰伦",
|
||||
mtype=MediaType.MUSIC,
|
||||
chain.async_search_music.assert_awaited_once_with(
|
||||
query="周杰伦",
|
||||
limit=30,
|
||||
music_types=("album",),
|
||||
media_source=(
|
||||
MediaSource.MusicBrainz,
|
||||
MediaSource.TheAudioDB,
|
||||
|
||||
@@ -52,6 +52,28 @@ def test_music_catalog_service_isolates_failed_source():
|
||||
assert errors and "broken" in errors[0]
|
||||
|
||||
|
||||
def test_music_catalog_service_forwards_stable_entity_filter_to_every_source():
|
||||
"""实体过滤即使由生成器传入,也应完整转发给每个音乐来源。"""
|
||||
received = []
|
||||
|
||||
class _FilteredSource:
|
||||
def search_music(self, _meta, limit=20, music_types=None):
|
||||
received.append((limit, music_types))
|
||||
return []
|
||||
|
||||
service = MusicCatalogService(
|
||||
source_resolver=lambda _source: _FilteredSource(),
|
||||
warning=lambda _message: None,
|
||||
)
|
||||
|
||||
assert service.search(
|
||||
"Hotel California",
|
||||
media_source=(MediaSource.MusicBrainz, MediaSource.DoubanMusic),
|
||||
music_types=(item for item in ["album"]),
|
||||
) == []
|
||||
assert received == [(20, ("album",)), (20, ("album",))]
|
||||
|
||||
|
||||
def test_music_catalog_merge_keeps_later_source_with_full_first_page():
|
||||
"""第一来源达到条数上限也不能挤掉后来来源的准确目标。"""
|
||||
first = [MusicInfo(media_source=MediaSource.MusicBrainz, media_id=str(index), title=f"Other {index}") for index in range(30)]
|
||||
|
||||
@@ -345,6 +345,74 @@ def test_search_music_interleaves_recordings_albums_and_artists(monkeypatch):
|
||||
assert requested[2][1]["query"] == 'artist:("周杰伦" OR "周杰倫")'
|
||||
|
||||
|
||||
def test_search_music_limits_album_selector_to_release_groups_and_ranks_identity(monkeypatch):
|
||||
"""专辑选择器只应请求 Release Group,并把艺人、年份与主类型匹配的录音室专辑排在前面。"""
|
||||
module = MusicBrainzModule()
|
||||
requested = []
|
||||
|
||||
def fake_request(path, params=None):
|
||||
requested.append((path, params))
|
||||
return {
|
||||
"release-groups": [
|
||||
{
|
||||
"id": "live-single",
|
||||
"title": "Hotel California",
|
||||
"primary-type": "Single",
|
||||
"secondary-types": ["Live"],
|
||||
"artist-credit": [{"artist": {"id": "eagles", "name": "Eagles"}}],
|
||||
},
|
||||
{
|
||||
"id": "cover-album",
|
||||
"title": "Hotel California",
|
||||
"first-release-date": "2004",
|
||||
"primary-type": "Album",
|
||||
"artist-credit": [{"artist": {"id": "cover", "name": "Banda Dos"}}],
|
||||
},
|
||||
{
|
||||
"id": "studio-album",
|
||||
"title": "Hotel California",
|
||||
"first-release-date": "1976-12-08",
|
||||
"primary-type": "Album",
|
||||
"artist-credit": [{"artist": {"id": "eagles", "name": "Eagles"}}],
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
monkeypatch.setattr(module, "_request_json", fake_request)
|
||||
|
||||
results = module.search_music(
|
||||
MetaMusic(title="Hotel California", artists=["Eagles"], year=1976),
|
||||
limit=20,
|
||||
music_types=("album",),
|
||||
)
|
||||
|
||||
assert [item.music_type for item in results] == ["album", "album", "album"]
|
||||
assert results[0].media_id == "studio-album"
|
||||
assert [path for path, _params in requested] == ["/release-group"]
|
||||
assert requested[0][1]["limit"] == 100
|
||||
|
||||
|
||||
def test_search_music_limits_recording_selector_to_recordings(monkeypatch):
|
||||
"""单曲选择器不应额外请求发行组或艺术家。"""
|
||||
module = MusicBrainzModule()
|
||||
requested_paths = []
|
||||
|
||||
def fake_request(path, params=None):
|
||||
requested_paths.append(path)
|
||||
return {"recordings": [{"id": "recording-1", "title": "Hotel California"}]}
|
||||
|
||||
monkeypatch.setattr(module, "_request_json", fake_request)
|
||||
|
||||
results = module.search_music(
|
||||
MetaMusic(title="Hotel California"),
|
||||
limit=20,
|
||||
music_types=("recording",),
|
||||
)
|
||||
|
||||
assert [item.media_id for item in results] == ["recording-1"]
|
||||
assert requested_paths == ["/recording"]
|
||||
|
||||
|
||||
def test_file_recognition_searches_recordings_only(monkeypatch):
|
||||
"""本地音轨识别不得把同名专辑或艺术家候选当成 Recording。"""
|
||||
module = MusicBrainzModule()
|
||||
|
||||
Reference in New Issue
Block a user