GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-14 10:14:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(music): add TheAudioDB and Douban discovery

Browse Source
This commit is contained in:
jxxghp
2026-08-12 10:50:13 +08:00
parent a1dd259143
commit d1028a6902
15 changed files with 769 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
app/api/endpoints/music.py
View File

@@ -133,6 +133,7 @@ async def clear_music_recognition_cache(
async def explore_music(
page: PageParam = 1,
count: CountParam = 30,
source: MusicSourceParam = "musicbrainz",
mode: MusicModeParam = "chart",
entity: MusicEntityParam = "recording",
range_name: MusicRangeParam = "this_month",
@@ -143,11 +144,20 @@ async def explore_music(
future: bool = True,
min_listen_count: Annotated[int, Query(ge=0)] = 0,
with_cover: bool = False,
country: Annotated[str, Query(pattern="^[A-Za-z]{2}$")] = "us",
_: schemas.TokenPayload = Depends(verify_token),
) -> list[schemas.MusicInfo]:
"""按 ListenBrainz 官方热门榜单或新发行两种模式返回可订阅的音乐候选。"""
"""按音乐来源返回可订阅的榜单或新发行候选。"""
chain = MusicChain()
if mode == "fresh":
if source != "musicbrainz":
results = await chain.async_discover(
source=source,
page=page,
count=count,
entity=entity,
country=country,
)
elif mode == "fresh":
results = await chain.async_fresh_releases(
days=days,
sort=sort,
@@ -167,6 +177,8 @@ async def explore_music(
with_cover=with_cover,
entity=entity,
)
if source != "musicbrainz" and with_cover:
results = [info for info in results if info.cover_url or info.poster_path]
return [_serialize_music(info) for info in results]
@@ -187,6 +199,26 @@ async def music_album(
return _serialize_album(info)
@router.get(
"/album/{album_id}/related",
summary="查询关联音乐专辑",
response_model=list[schemas.MusicInfo],
)
async def music_album_related(
album_id: str,
count: CountParam = 24,
source: MusicSourceParam = "musicbrainz",
_: schemas.TokenPayload = Depends(verify_token),
) -> list[schemas.MusicInfo]:
"""按来源和专辑 ID 返回可继续浏览的关联专辑。"""
results = await MusicChain().async_album_related(
source=source,
media_id=album_id,
count=count,
)
return [_serialize_music(info) for info in results]
@router.get(
"/artist/{artist_id}/albums",
summary="查询艺术家的专辑列表",

52
app/api/endpoints/recommend.py
View File

@@ -74,6 +74,58 @@ async def music_weekly(
return await RecommendChain().async_music_weekly(page=page, count=count)
@router.get(
"/music_theaudiodb_albums",
summary="TheAudioDB 热门专辑",
response_model=List[schemas.MusicInfo],
)
async def music_theaudiodb_albums(
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
_: schemas.TokenPayload = Depends(verify_token),
) -> Any:
"""浏览 TheAudioDB 指定国家或地区的热门专辑。"""
return await RecommendChain().async_music_theaudiodb_albums(
page=page,
count=count,
country=country,
)
@router.get(
"/music_theaudiodb_tracks",
summary="TheAudioDB 热门单曲",
response_model=List[schemas.MusicInfo],
)
async def music_theaudiodb_tracks(
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
_: schemas.TokenPayload = Depends(verify_token),
) -> Any:
"""浏览 TheAudioDB 指定国家或地区的热门单曲。"""
return await RecommendChain().async_music_theaudiodb_tracks(
page=page,
count=count,
country=country,
)
@router.get(
"/music_douban",
summary="豆瓣音乐推荐",
response_model=List[schemas.MusicInfo],
)
async def music_douban(
page: Optional[int] = 1,
count: Optional[int] = 30,
_: schemas.TokenPayload = Depends(verify_token),
) -> Any:
"""浏览豆瓣音乐推荐合集。"""
return await RecommendChain().async_music_douban(page=page, count=count)
@router.get(
"/douban_showing", summary="豆瓣正在热映", response_model=List[schemas.MediaInfo]
)

2
app/api/endpoints/subscribe.py
View File

@@ -613,7 +613,7 @@ async def subscribe_history(
current_user: User = Depends(get_current_active_user_async),
) -> Any:
"""
查询电影/电视剧订阅历史
查询电影、电视剧或音乐订阅历史
"""
if current_user.is_superuser:
histories = await SubscribeHistory.async_list_by_type(

53
app/chain/music.py
View File

@@ -314,6 +314,44 @@ class MusicChain(ChainBase):
)
return results[:count]
def discover(
self,
source: str,
page: int = 1,
count: int = 30,
entity: str = MUSIC_ENTITY_ALBUM,
country: str = "us",
) -> list[MusicInfo]:
"""按指定音乐源读取推荐榜单，并统一分页候选结构。"""
candidates = self.run_module(
"music_discover",
source=source,
page=page,
count=count,
entity=entity,
country=country,
)
return self.normalize_candidates(candidates, limit=count)
async def async_discover(
self,
source: str,
page: int = 1,
count: int = 30,
entity: str = MUSIC_ENTITY_ALBUM,
country: str = "us",
) -> list[MusicInfo]:
"""异步按指定音乐源读取推荐榜单，并统一分页候选结构。"""
candidates = await self.async_run_module(
"music_discover",
source=source,
page=page,
count=count,
entity=entity,
country=country,
)
return self.normalize_candidates(candidates, limit=count)
async def async_album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]:
"""异步按来源和专辑 ID 获取标准化专辑详情及曲目。"""
result = await self.async_run_module(
@@ -340,6 +378,21 @@ class MusicChain(ChainBase):
return MusicAlbumInfo.from_dict(result)
return None
async def async_album_related(
self,
source: str,
media_id: str,
count: int = 24,
) -> list[MusicInfo]:
"""异步读取指定来源的关联专辑，供专辑详情继续浏览。"""
candidates = await self.async_run_module(
"music_album_related",
source=source,
media_id=media_id,
count=count,
)
return self.normalize_candidates(candidates, limit=count)
def lyrics(self, music: MetaMusic | MusicInfo) -> Optional[MusicLyrics]:
"""按单曲元数据调用已启用的歌词模块并返回标准歌词。"""
result = self.run_module("music_lyrics", music=music)

108
app/chain/recommend.py
View File

@@ -12,6 +12,7 @@ from app.core.config import settings, global_vars
from app.helper.image import ImageHelper
from app.log import logger
from app.schemas import MediaType
from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING
from app.utils.common import log_execution_time
from app.utils.singleton import Singleton
@@ -57,6 +58,9 @@ class RecommendChain(ChainBase, metaclass=Singleton):
self.douban_movie_hot,
self.douban_tv_hot,
self.music_weekly,
self.music_theaudiodb_albums,
self.music_theaudiodb_tracks,
self.music_douban,
]
# 缓存并刷新所有推荐数据
@@ -187,6 +191,58 @@ class RecommendChain(ChainBase, metaclass=Singleton):
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
def music_theaudiodb_albums(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
) -> List[dict]:
"""返回 TheAudioDB 指定国家或地区的热门专辑。"""
medias = MusicChain().discover(
source="theaudiodb",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_ALBUM,
country=country,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
def music_theaudiodb_tracks(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
) -> List[dict]:
"""返回 TheAudioDB 指定国家或地区的热门单曲。"""
medias = MusicChain().discover(
source="theaudiodb",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_RECORDING,
country=country,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
def music_douban(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
) -> List[dict]:
"""返回豆瓣音乐推荐合集。"""
medias = MusicChain().discover(
source="doubanmusic",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_ALBUM,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
def tmdb_tvs(self, sort_by: Optional[str] = "popularity.desc",
@@ -416,6 +472,58 @@ class RecommendChain(ChainBase, metaclass=Singleton):
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
async def async_music_theaudiodb_albums(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
) -> List[dict]:
"""异步返回 TheAudioDB 指定国家或地区的热门专辑。"""
medias = await MusicChain().async_discover(
source="theaudiodb",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_ALBUM,
country=country,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
async def async_music_theaudiodb_tracks(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
country: str = "us",
) -> List[dict]:
"""异步返回 TheAudioDB 指定国家或地区的热门单曲。"""
medias = await MusicChain().async_discover(
source="theaudiodb",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_RECORDING,
country=country,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
async def async_music_douban(
self,
page: Optional[int] = 1,
count: Optional[int] = 30,
) -> List[dict]:
"""异步返回豆瓣音乐推荐合集。"""
medias = await MusicChain().async_discover(
source="doubanmusic",
page=page or 1,
count=count or 30,
entity=MUSIC_ENTITY_ALBUM,
)
return [media.to_dict() for media in medias]
@log_execution_time(logger=logger)
@cached(ttl=recommend_ttl, region=recommend_cache_region, skip_empty=True)
async def async_douban_movies(self, sort: Optional[str] = "R", tags: Optional[str] = "",

67
app/modules/douban/__init__.py
View File

@@ -136,6 +136,40 @@ class DoubanModule(_ModuleBase):
info = self.doubanapi.music_detail(subject_id=str(media_id))
return self._douban_music_to_album(info) if info else None
def music_discover(
self,
source: str,
page: int = 1,
count: int = 30,
entity: str = MUSIC_ENTITY_ALBUM,
country: str = "us",
) -> Optional[List[MusicInfo]]:
"""分页读取豆瓣音乐推荐合集，并保留豆瓣条目原生身份。"""
if source != self._music_source:
return None
del entity, country
result = self.doubanapi.music_single(
start=max(page - 1, 0) * max(1, count),
count=max(1, count),
)
return self._build_music_search_results(result)
def music_album_related(
self,
source: str,
media_id: str,
count: int = 24,
) -> Optional[List[MusicInfo]]:
"""按豆瓣音乐专辑 ID 返回相关推荐条目。"""
if source != self._music_source or not media_id:
return None
result = self.doubanapi.music_recommendations(
subject_id=str(media_id),
start=0,
count=max(1, count),
)
return self._build_music_search_results(result)
def _recognize_music_media(
self,
meta: Optional[MetaMusic],
@@ -278,17 +312,39 @@ class DoubanModule(_ModuleBase):
return candidates[0]
@classmethod
def _build_music_search_results(cls, result: Optional[dict]) -> List[MusicInfo]:
def _build_music_search_results(
cls,
result: Optional[dict | list],
) -> List[MusicInfo]:
"""把豆瓣音乐搜索响应转换为专辑候选列表。"""
items = (result or {}).get("items") or (result or {}).get("musics") or []
payload = result or {}
if isinstance(payload, list):
items = payload
else:
items = (
payload.get("subject_collection_items")
or payload.get("recommendations")
or payload.get("subjects")
or payload.get("items")
or payload.get("musics")
or []
)
candidates = []
for item in items:
if not isinstance(item, dict):
continue
target_type = str(item.get("target_type") or item.get("type") or "").casefold()
target = item.get("target") if isinstance(item.get("target"), dict) else item
target_type = str(item.get("target_type") or "").casefold()
if isinstance(item.get("target"), dict):
target = item["target"]
elif isinstance(item.get("subject"), dict):
target = item["subject"]
else:
target = item
type_name = str(target.get("type_name") or target.get("subtype") or "")
if target_type and target_type not in {"music", "音乐"}:
target_subject_type = str(target.get("type") or "").casefold()
if target_type and target_type not in {"music", "音乐", "subject"}:
continue
if target_subject_type and target_subject_type not in {"music", "音乐"}:
continue
if type_name and type_name not in {"音乐", "music"}:
continue
@@ -473,6 +529,7 @@ class DoubanModule(_ModuleBase):
cover_img.get("url"),
cover.get("large"),
cover.get("normal"),
cover.get("url"),
info.get("cover_url"),
info.get("image"),
]

38
app/modules/douban/apiv2.py
View File

@@ -623,6 +623,44 @@ class DoubanApi(metaclass=WeakSingleton):
"""异步获取豆瓣音乐详情。"""
return await self.__async_invoke_search(self._urls["music_detail"] + subject_id)
def music_single(self, start: int = 0, count: int = 20) -> dict:
"""分页获取豆瓣音乐推荐合集。"""
return self.__invoke_recommend(
self._urls["music_single"], start=start, count=count
)
async def async_music_single(self, start: int = 0, count: int = 20) -> dict:
"""异步分页获取豆瓣音乐推荐合集。"""
return await self.__async_invoke_recommend(
self._urls["music_single"], start=start, count=count
)
def music_recommendations(
self,
subject_id: str,
start: int = 0,
count: int = 20,
) -> dict:
"""获取豆瓣音乐条目的相关推荐。"""
return self.__invoke_recommend(
self._urls["music_recommendations"] % subject_id,
start=start,
count=count,
)
async def async_music_recommendations(
self,
subject_id: str,
start: int = 0,
count: int = 20,
) -> dict:
"""异步获取豆瓣音乐条目的相关推荐。"""
return await self.__async_invoke_recommend(
self._urls["music_recommendations"] % subject_id,
start=start,
count=count,
)
def movie_top250(self, start: Optional[int] = 0, count: Optional[int] = 20,
ts=datetime.strftime(datetime.now(), '%Y%m%d')):
"""

64
app/modules/theaudiodb/__init__.py
View File

@@ -274,6 +274,70 @@ class TheAudioDbModule(_ModuleBase):
start = max(page - 1, 0) * max(1, count)
return [album.to_music_info() for album in albums[start:start + max(1, count)]]
def music_discover(
self,
source: str,
page: int = 1,
count: int = 30,
entity: str = MUSIC_ENTITY_ALBUM,
country: str = "us",
) -> Optional[list[MusicInfo]]:
"""读取 TheAudioDB iTunes 趋势榜并转换为可继续浏览的音乐实体。"""
if source != self._source:
return None
payload = self._request_json(
"trending.php",
{
"country": country.casefold(),
"type": "itunes",
"format": "singles" if entity == MUSIC_ENTITY_RECORDING else "albums",
},
)
items = self._entities(payload, "trending")
items.sort(
key=lambda item: self._optional_int(item.get("intChartPlace")) or 10_000
)
candidates = []
for item in items:
if entity == MUSIC_ENTITY_RECORDING:
info = self._track_to_info(item)
else:
info = self._album_to_info(item).to_music_info()
if not info.media_id or not info.title:
continue
info.category = self._text(item.get("strType")) or "iTunes"
info.raw_data.update(
{
"chart_position": self._optional_int(item.get("intChartPlace")),
"chart_country": self._text(item.get("strCountry")),
}
)
candidates.append(info)
start = max(page - 1, 0) * max(1, count)
return candidates[start:start + max(1, count)]
def music_album_related(
self,
source: str,
media_id: str,
count: int = 24,
) -> Optional[list[MusicInfo]]:
"""按专辑主艺术家返回 TheAudioDB 同艺人专辑，供详情页关联浏览。"""
if source != self._source or not media_id:
return None
payload = self._request_json("album.php", {"m": media_id})
album_item = self._first_entity(payload, "album", "albums")
artist_id = self._text((album_item or {}).get("idArtist"))
if not artist_id:
return []
albums_payload = self._request_json("album.php", {"i": artist_id})
albums = [
self._album_to_info(item).to_music_info()
for item in self._entities(albums_payload, "album", "albums")
if self._text(item.get("idAlbum") or item.get("id")) != str(media_id)
]
return albums[:max(1, count)]
def clear_cache(self) -> None:
"""清除 TheAudioDB 请求缓存。"""
self._request_json.cache_clear()

8
docs/mcp-api.md
View File

@@ -196,12 +196,16 @@ AniList 榜单、探索、详情、人物和推荐接口优先通过 `anilist-ch
| :--- | :--- | :--- |
| GET | `/api/v1/media/search` | 当 `type=music` 或 `source=musicbrainz` 时按歌曲、专辑或歌手关键词搜索音乐元数据，参数：`title`、`type`、`count` |
| POST | `/api/v1/music/recognize` | 按 `source` + `media_id` 识别音乐详情，请求体：`MusicRecognizeRequest` |
| GET | `/api/v1/music/explore` | 浏览 ListenBrainz 热门单曲/专辑或新发行专辑，参数：`mode=chart|fresh`、`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}` | 按 MusicBrainz 专辑 ID 查询专辑详情、完整曲目和发行版本，参数：`source` |
| GET | `/api/v1/music/explore` | 按来源浏览音乐；`source=musicbrainz` 支持热门榜单与新发行，`source=theaudiodb` 支持国家/地区趋势专辑或单曲，`source=doubanmusic` 支持豆瓣音乐推荐。参数：`source`、`mode=chart|fresh`、`entity=recording|album`、`country`、`range_name`、`sort_by`、`sort`、`days`、`past`、`future`、`min_listen_count`、`with_cover`、`page`、`count` |
| GET | `/api/v1/music/album/{album_id}` | 按来源专辑 ID 查询专辑详情、完整曲目和发行版本，参数：`source` |
| GET | `/api/v1/music/album/{album_id}/related` | 按来源查询关联专辑，参数：`source`、`count` |
| GET | `/api/v1/music/artist/{artist_id}` | 查询艺术家详情；艺术家为只读浏览实体，参数：`source` |
| GET | `/api/v1/music/artist/{artist_id}/albums` | 分页查询艺术家的专辑、EP 和单曲，参数：`source`、`page`、`count`、`album_type` |
| GET | `/api/v1/music/artist/{artist_id}/related` | 查询关联艺术家，参数：`source`、`count` |
| GET | `/api/v1/recommend/music_weekly` | 浏览本周热门音乐，参数：`page`、`count` |
| GET | `/api/v1/recommend/music_theaudiodb_albums` | 浏览 TheAudioDB 热门专辑，参数：`country`、`page`、`count` |
| GET | `/api/v1/recommend/music_theaudiodb_tracks` | 浏览 TheAudioDB 热门单曲，参数：`country`、`page`、`count` |
| GET | `/api/v1/recommend/music_douban` | 浏览豆瓣音乐推荐，参数：`page`、`count` |
专辑下载与订阅按“整包”处理：下载层会读取种子文件清单并以专辑 `total_tracks` 校验独立音频文件数量；未确认完整覆盖时不会把专辑订阅销订，也不会把部分曲目报告为完整专辑已入库。音乐刮削遵循 `music` 的标签、封面和歌词策略，歌词通过带有界 TTL/LRU 缓存的 LRCLIB 模块保存为同名 `.lrc` 或 `.txt` 旁挂文件。

8
skills/moviepilot-api/SKILL.md
View File

@@ -189,8 +189,9 @@ music on configured music-capable media servers; it does not manage playlists.
|--------|------|-------------|
| GET | `/api/v1/media/search` | Search tracks, albums, or artists with `type=music` or `source=musicbrainz`. Params: `title`, `type`, `count` |
| POST | `/api/v1/music/recognize` | Resolve music metadata. Body: `source`, `media_id` |
| GET | `/api/v1/music/explore` | Explore ListenBrainz charts or fresh albums. Params: `mode`, `entity`, `range_name`, `sort_by`, `sort`, `days`, `past`, `future`, `min_listen_count`, `with_cover`, `page`, `count` |
| GET | `/api/v1/music/explore` | Explore music by `source`: MusicBrainz charts/fresh releases, TheAudioDB country trends, or Douban music recommendations. Params: `source`, `mode`, `entity`, `country`, `range_name`, `sort_by`, `sort`, `days`, `past`, `future`, `min_listen_count`, `with_cover`, `page`, `count` |
| GET | `/api/v1/music/album/{album_id}` | Album detail with tracks and releases. Params: `source` |
| GET | `/api/v1/music/album/{album_id}/related` | Related albums for the selected source. Params: `source`, `count` |
| GET | `/api/v1/music/artist/{artist_id}` | Browse artist detail. Params: `source` |
| GET | `/api/v1/music/artist/{artist_id}/albums` | Browse artist albums/EPs/singles. Params: `source`, `page`, `count`, `album_type` |
| GET | `/api/v1/music/artist/{artist_id}/related` | Browse related artists. Params: `source`, `count` |
@@ -466,13 +467,16 @@ Streaming search sends `{"type":"heartbeat"}` every 15 seconds without business
| GET | `/api/v1/discover/tmdb_movies` | Discover TMDB movies. Params: `sort_by`, `with_genres`, `with_original_language`, `page` |
| GET | `/api/v1/discover/tmdb_tvs` | Discover TMDB TV. Params: same as movies |
### Recommend (15 endpoints)
### Recommend (18 endpoints)
| Method | Path | Description |
|--------|------|-------------|
| GET | `/api/v1/recommend/source` | Recommendation data sources |
| GET | `/api/v1/recommend/bangumi_calendar` | Bangumi daily schedule. Params: `page`, `count` |
| GET | `/api/v1/recommend/music_weekly` | ListenBrainz weekly site-wide music chart. Params: `page`, `count` |
| GET | `/api/v1/recommend/music_theaudiodb_albums` | TheAudioDB trending albums. Params: `country`, `page`, `count` |
| GET | `/api/v1/recommend/music_theaudiodb_tracks` | TheAudioDB trending tracks. Params: `country`, `page`, `count` |
| GET | `/api/v1/recommend/music_douban` | Douban music recommendations. Params: `page`, `count` |
| GET | `/api/v1/recommend/douban_showing` | Douban now showing. Params: `page`, `count` |
| GET | `/api/v1/recommend/douban_movies` | Douban movies. Params: `sort`, `tags`, `page`, `count` |
| GET | `/api/v1/recommend/douban_tvs` | Douban TV. Params: `sort`, `tags`, `page`, `count` |

111
tests/test_music_endpoint.py
View File

@@ -5,15 +5,16 @@ import pytest
from fastapi import HTTPException
from app.api.apiv1 import api_router
from app.api.endpoints import media as media_endpoints
from app.api.endpoints.music import (
explore_music,
music_album,
music_album_related,
music_artist,
music_artist_albums,
music_artist_related,
recognize_music,
)
from app.api.endpoints import media as media_endpoints
from app.core.context import MusicAlbumInfo, MusicArtistInfo, MusicInfo, MusicRelease
from app.schemas.music import MusicRecognizeRequest
from app.schemas.types import MediaType
@@ -26,6 +27,10 @@ def test_music_routes_are_registered():
assert any(path == "/music/recognize" and "POST" in methods for path, methods in routes)
assert any(path == "/music/explore" and "GET" in methods for path, methods in routes)
assert any(path == "/music/album/{album_id}" and "GET" in methods for path, methods in routes)
assert any(
path == "/music/album/{album_id}/related" and "GET" in methods
for path, methods in routes
)
assert any(path == "/music/artist/{artist_id}" and "GET" in methods for path, methods in routes)
assert any(
path == "/music/artist/{artist_id}/albums" and "GET" in methods
@@ -38,6 +43,15 @@ def test_music_routes_are_registered():
assert any(
path == "/media/search" and "GET" in methods for path, methods in routes
)
for recommend_path in (
"/recommend/music_theaudiodb_albums",
"/recommend/music_theaudiodb_tracks",
"/recommend/music_douban",
):
assert any(
path == recommend_path and "GET" in methods
for path, methods in routes
)
def test_media_search_routes_music_queries_with_query_kwarg():
@@ -244,6 +258,69 @@ def test_explore_music_supports_official_fresh_release_mode():
)
def test_explore_music_forwards_selected_metadata_source():
"""TheAudioDB 与豆瓣探索应走可扩展发现链而不是 ListenBrainz。"""
chain = Mock()
chain.async_discover = AsyncMock(
return_value=[
MusicInfo(
source="theaudiodb",
media_id="album-1",
music_type="album",
title="Parachutes",
)
]
)
with patch("app.api.endpoints.music.MusicChain", return_value=chain):
result = asyncio.run(
explore_music(
source="theaudiodb",
entity="album",
country="gb",
page=2,
count=20,
_=Mock(),
)
)
assert result[0].source == "theaudiodb"
chain.async_discover.assert_awaited_once_with(
source="theaudiodb",
page=2,
count=20,
entity="album",
country="gb",
)
def test_explore_music_filters_missing_covers_for_external_sources():
"""外部音乐源选择仅有封面时应在统一响应层过滤无图条目。"""
chain = Mock()
chain.async_discover = AsyncMock(
return_value=[
MusicInfo(source="doubanmusic", media_id="album-1", title="No Cover"),
MusicInfo(
source="doubanmusic",
media_id="album-2",
title="With Cover",
cover_url="https://img.example/album-2.jpg",
),
]
)
with patch("app.api.endpoints.music.MusicChain", return_value=chain):
result = asyncio.run(
explore_music(
source="doubanmusic",
with_cover=True,
_=Mock(),
)
)
assert [item.media_id for item in result] == ["album-2"]
def test_music_album_returns_tracks_and_releases():
"""专辑接口应返回专辑详情、曲目和发行版本。"""
chain = Mock()
@@ -286,6 +363,38 @@ def test_music_album_returns_404_for_unknown_album():
assert error.value.status_code == 404
def test_music_album_related_returns_source_results():
"""专辑关联浏览接口应传递来源和数量并序列化结果。"""
chain = Mock()
chain.async_album_related = AsyncMock(
return_value=[
MusicInfo(
source="doubanmusic",
media_id="album-2",
music_type="album",
title="依然范特西",
)
]
)
with patch("app.api.endpoints.music.MusicChain", return_value=chain):
result = asyncio.run(
music_album_related(
album_id="album-1",
count=12,
source="doubanmusic",
_=Mock(),
)
)
assert result[0].media_id == "album-2"
chain.async_album_related.assert_awaited_once_with(
source="doubanmusic",
media_id="album-1",
count=12,
)
def test_music_artist_returns_detail():
"""艺术家接口应返回名称、类型和活跃时间。"""
chain = Mock()

91
tests/test_music_metadata_sources.py
View File

@@ -81,6 +81,64 @@ def test_theaudiodb_detail_respects_requested_entity(monkeypatch):
request.assert_called_once_with("album.php", {"m": "2109619"})
def test_theaudiodb_discover_maps_and_sorts_trending_albums(monkeypatch):
"""TheAudioDB 探索应按榜位排序并保留趋势来源元数据。"""
module = TheAudioDbModule()
request = Mock(return_value={
"trending": [
{
"idAlbum": "album-2",
"strAlbum": "Second",
"strArtist": "Artist",
"intChartPlace": "2",
"strCountry": "GB",
},
{
"idAlbum": "album-1",
"strAlbum": "First",
"strArtist": "Artist",
"intChartPlace": "1",
"strCountry": "GB",
},
]
})
monkeypatch.setattr(module, "_request_json", request)
results = module.music_discover(
source="theaudiodb",
entity=MUSIC_ENTITY_ALBUM,
country="GB",
)
assert results and [item.media_id for item in results] == ["album-1", "album-2"]
assert results[0].source == "theaudiodb"
assert results[0].raw_data["chart_position"] == 1
request.assert_called_once_with(
"trending.php",
{"country": "gb", "type": "itunes", "format": "albums"},
)
def test_theaudiodb_album_related_excludes_current_album(monkeypatch):
"""TheAudioDB 关联专辑应按当前专辑艺术家查询并排除自身。"""
module = TheAudioDbModule()
request = Mock(side_effect=[
{"album": [{"idAlbum": "album-1", "idArtist": "artist-1"}]},
{
"album": [
{"idAlbum": "album-1", "strAlbum": "Current"},
{"idAlbum": "album-2", "strAlbum": "Related"},
]
},
])
monkeypatch.setattr(module, "_request_json", request)
results = module.music_album_related("theaudiodb", "album-1", count=10)
assert results and [item.media_id for item in results] == ["album-2"]
assert request.call_args_list[1].args == ("album.php", {"i": "artist-1"})
def test_douban_detail_rejects_album_id_as_recording(monkeypatch):
"""豆瓣单曲使用专辑加曲序复合 ID，纯专辑 ID 不能作为 Recording。"""
module = DoubanModule()
@@ -153,6 +211,39 @@ def test_douban_music_search_and_album_mapping(monkeypatch):
assert album.tracks[1].cover_url == "https://img.example/track.jpg"
def test_douban_music_discover_and_related_accept_collection_wrappers(monkeypatch):
"""豆瓣音乐合集与相关推荐应兼容 subject 包装并保留专辑身份。"""
module = DoubanModule()
module.doubanapi = Mock()
wrapped_item = {
"type": "subject_collection_item",
"subject": {
"id": "1401853",
"type": "music",
"title": "范特西",
"artists": [{"name": "周杰伦"}],
"cover": {"url": "https://img.example/fantasy.jpg"},
},
}
module.doubanapi.music_single.return_value = {
"subject_collection_items": [wrapped_item]
}
module.doubanapi.music_recommendations.return_value = [wrapped_item["subject"]]
discovered = module.music_discover("doubanmusic", page=2, count=10)
related = module.music_album_related("doubanmusic", "album-1", count=6)
assert discovered and discovered[0].media_id == "1401853"
assert discovered[0].cover_url == "https://img.example/fantasy.jpg"
assert related and related[0].source == "doubanmusic"
module.doubanapi.music_single.assert_called_once_with(start=10, count=10)
module.doubanapi.music_recommendations.assert_called_once_with(
subject_id="album-1",
start=0,
count=6,
)
def test_douban_music_recognize_expands_album_to_matching_track(monkeypatch):
"""自动文件识别有专辑线索时，豆瓣应返回专辑内音轨而不是专辑实体。"""
module = DoubanModule()

55
tests/test_music_subscribe.py
View File

@@ -1,6 +1,8 @@
from types import SimpleNamespace
from unittest.mock import Mock, patch
import pytest
from app.chain.music import MusicChain
from app.chain.subscribe import SubscribeChain, build_subscribe_meta
from app.core.context import (
@@ -8,10 +10,10 @@ from app.core.context import (
MUSIC_ENTITY_ARTIST,
MUSIC_ENTITY_RECORDING,
Context,
MusicInfo,
TorrentInfo,
)
from app.core.meta import MetaMusic
from app.core.context import MusicInfo
from app.schemas.types import MediaType
@@ -736,6 +738,57 @@ def test_subscribe_add_music_uses_explicit_entity_recognize():
media_chain.recognize_by_meta.assert_not_called()
@pytest.mark.parametrize(
("source", "media_id", "title"),
[
("theaudiodb", "2109619", "Parachutes"),
("doubanmusic", "1401853", "范特西"),
],
)
def test_subscribe_add_music_routes_new_album_sources(
source: str,
media_id: str,
title: str,
):
"""新增音乐源的专辑订阅应保留来源、原生 ID 与实体类型。"""
target = MusicInfo(
source=source,
media_id=media_id,
music_type=MUSIC_ENTITY_ALBUM,
title=title,
album=title,
total_tracks=10,
)
media_chain = Mock()
media_chain.recognize_media = Mock(return_value=target)
subscribe_oper = Mock()
subscribe_oper.add.return_value = (1, "")
with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \
patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \
patch("app.chain.subscribe.MoviePilotServerHelper"), \
patch("app.chain.subscribe.eventmanager"):
sid, err_msg = SubscribeChain().add(
title=title,
year="2000",
mtype=MediaType.MUSIC,
media_source=source,
media_id=media_id,
music_type=MUSIC_ENTITY_ALBUM,
message=False,
)
assert sid == 1
assert err_msg == ""
media_chain.recognize_media.assert_called_once()
assert media_chain.recognize_media.call_args.kwargs["source"] == source
assert media_chain.recognize_media.call_args.kwargs["mediaid"] == media_id
assert media_chain.recognize_media.call_args.kwargs["music_type"] == MUSIC_ENTITY_ALBUM
assert subscribe_oper.add.call_args.kwargs["media_source"] == source
assert subscribe_oper.add.call_args.kwargs["media_id"] == media_id
media_chain.recognize_by_meta.assert_not_called()
def test_subscribe_add_rejects_music_entity_mismatch_before_database_write():
"""请求专辑却识别为单曲时必须中止，不能创建完成语义错误的订阅。"""
media_chain = Mock()

76
tests/test_recommend_chain.py
View File

@@ -5,9 +5,9 @@ from unittest.mock import AsyncMock, patch
import pytest
from app.chain.recommend import RecommendChain
from app.core.context import MusicInfo
from app.core.cache import TTLCache
from app.core.context import MusicInfo
from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING
SYNC_EMPTY_CACHE_CASES = [
("tmdb_movies", "app.chain.recommend.TmdbChain", "tmdb_discover"),
@@ -132,3 +132,75 @@ def test_async_music_weekly_uses_music_chart():
page=1,
count=30,
)
@pytest.mark.parametrize(
("method_name", "source", "entity", "country"),
[
("music_theaudiodb_albums", "theaudiodb", MUSIC_ENTITY_ALBUM, "gb"),
("music_theaudiodb_tracks", "theaudiodb", MUSIC_ENTITY_RECORDING, "gb"),
("music_douban", "doubanmusic", MUSIC_ENTITY_ALBUM, "us"),
],
)
def test_music_source_recommendations_use_discover(
method_name: str,
source: str,
entity: str,
country: str,
):
"""新增音乐推荐入口应保留来源与实体，并输出统一媒体字典。"""
chain = RecommendChain()
with patch("app.chain.recommend.MusicChain") as music_chain:
music_chain.return_value.discover.return_value = [
MusicInfo(source=source, media_id="music-1", music_type=entity, title="Music")
]
kwargs = {"page": 2, "count": 10}
if source == "theaudiodb":
kwargs["country"] = country
result = getattr(chain, method_name)(**kwargs)
assert result[0]["source"] == source
expected_kwargs = {
"source": source,
"page": 2,
"count": 10,
"entity": entity,
}
if source == "theaudiodb":
expected_kwargs["country"] = country
music_chain.return_value.discover.assert_called_once_with(**expected_kwargs)
@pytest.mark.parametrize(
("method_name", "source", "entity"),
[
("async_music_theaudiodb_albums", "theaudiodb", MUSIC_ENTITY_ALBUM),
("async_music_theaudiodb_tracks", "theaudiodb", MUSIC_ENTITY_RECORDING),
("async_music_douban", "doubanmusic", MUSIC_ENTITY_ALBUM),
],
)
def test_async_music_source_recommendations_use_discover(
method_name: str,
source: str,
entity: str,
):
"""异步音乐推荐入口应调用统一发现链并保留来源。"""
chain = RecommendChain()
with patch("app.chain.recommend.MusicChain") as music_chain:
music_chain.return_value.async_discover = AsyncMock(
return_value=[MusicInfo(source=source, media_id="music-1", title="Music")]
)
result = asyncio.run(getattr(chain, method_name)(page=1, count=30))
assert result[0]["source"] == source
expected_kwargs = {
"source": source,
"page": 1,
"count": 30,
"entity": entity,
}
if source == "theaudiodb":
expected_kwargs["country"] = "us"
music_chain.return_value.async_discover.assert_awaited_once_with(**expected_kwargs)

20
tests/test_search_media_sources.py
View File

@@ -57,7 +57,19 @@ def test_resource_search_forwards_custom_plugin_source(monkeypatch) -> None:
assert captured["mtype"] == MediaType.TV
def test_resource_search_forwards_music_entity_namespace(monkeypatch) -> None:
@pytest.mark.parametrize(
("source", "media_id"),
[
("musicbrainz", "release-group-1"),
("theaudiodb", "2109619"),
("doubanmusic", "1401853"),
],
)
def test_resource_search_forwards_music_entity_namespace(
monkeypatch,
source: str,
media_id: str,
) -> None:
"""音乐资源搜索 API 应在识别前传递单曲或专辑实体类型。"""
captured = {}
@@ -81,7 +93,7 @@ def test_resource_search_forwards_music_entity_namespace(monkeypatch) -> None:
response = asyncio.run(
search_endpoint.search_by_id(
mediaid="musicbrainz:release-group-1",
mediaid=f"{source}:{media_id}",
mtype="music",
music_type="album",
_=None,
@@ -89,8 +101,8 @@ def test_resource_search_forwards_music_entity_namespace(monkeypatch) -> None:
)
assert response.success
assert captured["source"] == "musicbrainz"
assert captured["mediaid"] == "release-group-1"
assert captured["source"] == source
assert captured["mediaid"] == media_id
assert captured["mtype"] == MediaType.MUSIC
assert captured["music_type"] == "album"