From 415ccda6a2d051d5b8f7158d9f6c92fcc032aa14 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 8 Aug 2026 12:57:09 +0800 Subject: [PATCH] feat(v3): enhance music server integration with Navidrome support --- app/api/endpoints/dashboard.py | 1 + app/api/endpoints/music.py | 151 ++++++++-- app/chain/mediaserver.py | 6 +- app/chain/music.py | 138 ++++++++- app/core/config.py | 2 + app/core/music.py | 326 ++++++++++++++++++++++ app/locales/en-US.json | 4 + app/locales/zh-TW.json | 4 + app/modules/emby/__init__.py | 14 + app/modules/emby/emby.py | 45 ++- app/modules/jellyfin/__init__.py | 14 + app/modules/jellyfin/jellyfin.py | 45 ++- app/modules/listenbrainz/__init__.py | 226 +++++++++++++-- app/modules/musicbrainz/__init__.py | 401 ++++++++++++++++++++++++++- app/modules/navidrome/__init__.py | 205 ++++++++++++++ app/modules/navidrome/navidrome.py | 232 ++++++++++++++++ app/modules/plex/__init__.py | 14 + app/modules/plex/plex.py | 40 ++- app/modules/trimemedia/__init__.py | 14 + app/modules/trimemedia/trimemedia.py | 3 + app/modules/ugreen/__init__.py | 14 + app/modules/ugreen/ugreen.py | 31 +++ app/modules/zspace/__init__.py | 14 + app/modules/zspace/zspace.py | 34 ++- app/schemas/dashboard.py | 2 + app/schemas/mediaserver.py | 4 +- app/schemas/music.py | 90 ++++++ app/schemas/system.py | 2 +- app/schemas/types.py | 2 + tests/test_listenbrainz_module.py | 103 ++++++- tests/test_music_chain.py | 113 +++++++- tests/test_music_endpoint.py | 187 ++++++++++++- tests/test_music_subscribe.py | 1 + tests/test_musicbrainz_module.py | 252 +++++++++++++++++ tests/test_navidrome_module.py | 32 +++ 35 files changed, 2676 insertions(+), 90 deletions(-) create mode 100644 app/modules/navidrome/__init__.py create mode 100644 app/modules/navidrome/navidrome.py create mode 100644 tests/test_navidrome_module.py diff --git a/app/api/endpoints/dashboard.py b/app/api/endpoints/dashboard.py index 096624778..a6828aeb2 100644 --- a/app/api/endpoints/dashboard.py +++ b/app/api/endpoints/dashboard.py @@ -33,6 +33,7 @@ def _build_statistic(db: Session, name: Optional[str] = None) -> schemas.Statist for media_statistic in media_statistics: ret_statistic.movie_count += media_statistic.movie_count or 0 ret_statistic.tv_count += media_statistic.tv_count or 0 + ret_statistic.music_count += media_statistic.music_count or 0 ret_statistic.user_count += media_statistic.user_count or 0 if media_statistic.episode_count is not None: ret_statistic.episode_count += media_statistic.episode_count or 0 diff --git a/app/api/endpoints/music.py b/app/api/endpoints/music.py index 448390caf..8a15237ec 100644 --- a/app/api/endpoints/music.py +++ b/app/api/endpoints/music.py @@ -1,22 +1,32 @@ -from typing import Annotated +from typing import Annotated, Optional from fastapi import APIRouter, Depends, HTTPException, Query from app import schemas from app.chain.music import MusicChain -from app.core.music import MusicInfo +from app.core.music import MusicAlbumInfo, MusicArtistInfo, MusicInfo from app.core.security import verify_token +from app.modules.listenbrainz import ( + LISTENBRAINZ_CHART_RANGES, + LISTENBRAINZ_FRESH_MAX_DAYS, + LISTENBRAINZ_FRESH_SORTS, +) 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)$"), +PageParam = Annotated[int, Query(ge=1)] +MusicSourceParam = Annotated[str, Query(pattern="^musicbrainz$")] +MusicModeParam = Annotated[str, Query(pattern="^(chart|fresh)$")] +MusicEntityParam = Annotated[str, Query(pattern="^(recording|album)$")] +MusicRangeParam = Annotated[str, Query(pattern=f"^({'|'.join(LISTENBRAINZ_CHART_RANGES)})$")] +MusicSortParam = Annotated[str, Query(pattern="^listen_count\\.(desc|asc)$")] +MusicFreshSortParam = Annotated[str, Query(pattern=f"^({'|'.join(LISTENBRAINZ_FRESH_SORTS)})$")] +MusicDaysParam = Annotated[int, Query(ge=1, le=LISTENBRAINZ_FRESH_MAX_DAYS)] +# MusicBrainz 浏览接口支持的专辑类型筛选 +MusicAlbumTypeParam = Annotated[ + Optional[str], + Query(pattern="^(album|single|ep|broadcast|other|compilation|soundtrack|live|remix)$"), ] @@ -25,6 +35,16 @@ def _serialize_music(info: MusicInfo) -> schemas.MusicInfo: return schemas.MusicInfo(**info.to_dict()) +def _serialize_album(info: MusicAlbumInfo) -> schemas.MusicAlbumInfo: + """将内部专辑信息转换为 REST 响应模型。""" + return schemas.MusicAlbumInfo(**info.to_dict()) + + +def _serialize_artist(info: MusicArtistInfo) -> schemas.MusicArtistInfo: + """将内部艺术家信息转换为 REST 响应模型。""" + return schemas.MusicArtistInfo(**info.to_dict()) + + @router.get( "/search", summary="搜索音乐元数据", @@ -61,25 +81,122 @@ async def recognize_music( @router.get( "/explore", - summary="探索热门音乐", + summary="探索音乐", response_model=list[schemas.MusicInfo], ) async def explore_music( - page: Annotated[int, Query(ge=1)] = 1, + page: PageParam = 1, count: CountParam = 30, + mode: MusicModeParam = "chart", + entity: MusicEntityParam = "recording", range_name: MusicRangeParam = "this_month", sort_by: MusicSortParam = "listen_count.desc", + sort: MusicFreshSortParam = "release_date", + days: MusicDaysParam = 14, + past: bool = True, + future: bool = True, 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=range_name, + """按 ListenBrainz 官方热门榜单或新发行两种模式返回可订阅的音乐候选。""" + chain = MusicChain() + if mode == "fresh": + results = await chain.async_fresh_releases( + days=days, + sort=sort, + past=past, + future=future, + page=page, + count=count, + with_cover=with_cover, + ) + else: + results = await chain.async_chart( + range_name=range_name, + page=page, + count=count, + sort_by=sort_by, + min_listen_count=min_listen_count, + with_cover=with_cover, + entity=entity, + ) + return [_serialize_music(info) for info in results] + + +@router.get( + "/album/{album_id}", + summary="查询音乐专辑详情", + response_model=schemas.MusicAlbumInfo, +) +async def music_album( + album_id: str, + source: MusicSourceParam = "musicbrainz", + _: schemas.TokenPayload = Depends(verify_token), +) -> schemas.MusicAlbumInfo: + """按专辑标准 ID 返回专辑详情、曲目列表和发行版本。""" + info = await MusicChain().async_album(source=source, media_id=album_id) + if not info: + raise HTTPException(status_code=404, detail="未识别到专辑信息") + return _serialize_album(info) + + +@router.get( + "/artist/{artist_id}/albums", + summary="查询艺术家的专辑列表", + response_model=list[schemas.MusicInfo], +) +async def music_artist_albums( + artist_id: str, + page: PageParam = 1, + count: CountParam = 30, + album_type: MusicAlbumTypeParam = None, + source: MusicSourceParam = "musicbrainz", + _: schemas.TokenPayload = Depends(verify_token), +) -> list[schemas.MusicInfo]: + """按艺术家标准 ID 分页返回其专辑、EP 和单曲。""" + results = await MusicChain().async_artist_albums( + source=source, + media_id=artist_id, page=page, count=count, - sort_by=sort_by, - min_listen_count=min_listen_count, - with_cover=with_cover, + album_type=album_type, ) return [_serialize_music(info) for info in results] + + +@router.get( + "/artist/{artist_id}/related", + summary="查询关联艺术家", + response_model=list[schemas.MusicArtistInfo], +) +async def music_artist_related( + artist_id: str, + count: CountParam = 24, + source: MusicSourceParam = "musicbrainz", + _: schemas.TokenPayload = Depends(verify_token), +) -> list[schemas.MusicArtistInfo]: + """按艺术家关系返回可继续浏览的关联艺术家。""" + results = await MusicChain().async_artist_related( + source=source, + media_id=artist_id, + count=count, + ) + return [_serialize_artist(info) for info in results] + + +@router.get( + "/artist/{artist_id}", + summary="查询音乐艺术家详情", + response_model=schemas.MusicArtistInfo, +) +async def music_artist( + artist_id: str, + source: MusicSourceParam = "musicbrainz", + _: schemas.TokenPayload = Depends(verify_token), +) -> schemas.MusicArtistInfo: + """按艺术家标准 ID 返回艺术家详情。""" + info = await MusicChain().async_artist(source=source, media_id=artist_id) + if not info: + raise HTTPException(status_code=404, detail="未识别到艺术家信息") + return _serialize_artist(info) diff --git a/app/chain/mediaserver.py b/app/chain/mediaserver.py index 7143e9a49..433feaecc 100644 --- a/app/chain/mediaserver.py +++ b/app/chain/mediaserver.py @@ -129,7 +129,7 @@ class MediaServerChain(ChainBase): def media_count(self, server: str) -> Optional[int]: """ - 获取指定媒体服务器可同步的电影和电视剧总数 + 获取指定媒体服务器可同步的电影、电视剧和音乐总数 :param server: 媒体服务器名称 :return: 电影和电视剧总数，无法获取时返回None @@ -138,7 +138,9 @@ class MediaServerChain(ChainBase): if not statistics: return None return sum( - (statistic.movie_count or 0) + (statistic.tv_count or 0) + (statistic.movie_count or 0) + + (statistic.tv_count or 0) + + (statistic.music_count or 0) for statistic in statistics ) diff --git a/app/chain/music.py b/app/chain/music.py index f318daec5..f15dac483 100644 --- a/app/chain/music.py +++ b/app/chain/music.py @@ -7,7 +7,13 @@ 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.core.music import ( + MUSIC_ENTITY_RECORDING, + MusicAlbumInfo, + MusicArtistInfo, + MusicInfo, + MusicMeta, +) from app.helper.audio import AudioMetadataHelper from app.log import logger from app.utils.http import RequestUtils @@ -130,15 +136,135 @@ class MusicChain(ChainBase): sort_by: str = "listen_count.desc", min_listen_count: int = 0, with_cover: bool = False, + entity: str = MUSIC_ENTITY_RECORDING, ) -> list[MusicInfo]: - """异步读取 ListenBrainz 榜单，并应用音乐探索筛选和排序。""" + """异步读取 ListenBrainz 热门榜单，并应用音乐探索筛选和排序。""" candidates = await self.async_run_module( "music_chart", range_name=range_name, offset=max(page - 1, 0) * count, count=count, + entity=entity, ) - results = self.normalize_candidates(candidates) + results = self._filter_candidates( + self.normalize_candidates(candidates), + min_listen_count=min_listen_count, + with_cover=with_cover, + ) + results.sort( + key=lambda info: info.listen_count or 0, + reverse=sort_by != "listen_count.asc", + ) + return results[:count] + + async def async_fresh_releases( + self, + days: int = 14, + sort: str = "release_date", + past: bool = True, + future: bool = True, + page: int = 1, + count: int = 30, + with_cover: bool = False, + ) -> list[MusicInfo]: + """异步读取 ListenBrainz 官方新发行专辑，排序由官方接口决定。""" + candidates = await self.async_run_module( + "music_fresh_releases", + days=days, + sort=sort, + past=past, + future=future, + offset=max(page - 1, 0) * count, + count=count, + ) + results = self._filter_candidates( + self.normalize_candidates(candidates), + min_listen_count=0, + with_cover=with_cover, + ) + return results[:count] + + async def async_album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]: + """异步按来源和专辑 ID 获取标准化专辑详情及曲目。""" + result = await self.async_run_module( + "music_album", + source=source, + media_id=media_id, + ) + if isinstance(result, MusicAlbumInfo): + return result + if isinstance(result, dict): + return MusicAlbumInfo.from_dict(result) + return None + + async def async_artist(self, source: str, media_id: str) -> Optional[MusicArtistInfo]: + """异步按来源和艺术家 ID 获取标准化艺术家详情。""" + result = await self.async_run_module( + "music_artist", + source=source, + media_id=media_id, + ) + if isinstance(result, MusicArtistInfo): + return result + if isinstance(result, dict): + return MusicArtistInfo.from_dict(result) + return None + + async def async_artist_albums( + self, + source: str, + media_id: str, + page: int = 1, + count: int = 30, + album_type: Optional[str] = None, + ) -> list[MusicInfo]: + """异步分页读取艺术家名下的专辑、EP 和单曲。""" + candidates = await self.async_run_module( + "music_artist_albums", + source=source, + media_id=media_id, + page=page, + count=count, + album_type=album_type, + ) + return self.normalize_candidates(candidates, limit=count) + + async def async_artist_related( + self, + source: str, + media_id: str, + count: int = 24, + ) -> list[MusicArtistInfo]: + """异步读取关联艺术家，供详情页继续浏览。""" + candidates = await self.async_run_module( + "music_artist_related", + source=source, + media_id=media_id, + count=count, + ) + results: list[MusicArtistInfo] = [] + identities: set[str] = set() + for candidate in candidates or []: + info = ( + candidate + if isinstance(candidate, MusicArtistInfo) + else MusicArtistInfo.from_dict(candidate) + ) + identity = (info.media_id or info.name or "").casefold() + if not identity or identity in identities: + continue + identities.add(identity) + results.append(info) + return results[:count] + + @staticmethod + def _filter_candidates( + candidates: list[MusicInfo], + min_listen_count: int, + with_cover: bool, + ) -> list[MusicInfo]: + """按热度和封面条件过滤音乐探索候选。""" + results = candidates if min_listen_count > 0: results = [ info for info in results @@ -146,11 +272,7 @@ class MusicChain(ChainBase): ] 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] + return list(results) @classmethod def is_audio_path(cls, path: str | Path) -> bool: diff --git a/app/core/config.py b/app/core/config.py index 3ccc3b990..92b8cf12b 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -519,6 +519,8 @@ class ConfigModel(BaseModel): "anilist.co", "coverartarchive.org", "archive.org", + "commons.wikimedia.org", + "upload.wikimedia.org", ] ) # 图片代理允许访问的非公网 IP/CIDR，默认不放行任何非公网解析结果 diff --git a/app/core/music.py b/app/core/music.py index 86eca56ff..f1e77fc6c 100644 --- a/app/core/music.py +++ b/app/core/music.py @@ -3,6 +3,11 @@ from typing import Any, Self from app.schemas.types import MediaType +# 音乐可浏览实体类型：单曲（Recording）、专辑（Release Group）、艺术家（Artist） +MUSIC_ENTITY_RECORDING = "recording" +MUSIC_ENTITY_ALBUM = "album" +MUSIC_ENTITY_ARTIST = "artist" + def _validate_music_type(value: object) -> None: if value in {None, MediaType.MUSIC, MediaType.MUSIC.value, "music"}: @@ -20,6 +25,15 @@ def _string_list(value: object) -> list[str]: return [str(value)] +def _aligned_list(value: object) -> list[str]: + """保留原始位置的字符串列表，用于与艺术家名称按下标对应的 ID 列表。""" + if value is None: + return [] + if isinstance(value, (list, tuple)): + return [str(item or "") for item in value] + return [str(value or "")] + + def _optional_int(value: object) -> int | None: if value in {None, ""}: return None @@ -29,6 +43,21 @@ def _optional_int(value: object) -> int | None: return None +def _optional_float(value: object) -> float: + if value in {None, ""}: + return 0.0 + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +def _year_of(release_date: object) -> int | None: + """从 MusicBrainz 可变精度日期（YYYY / YYYY-MM / YYYY-MM-DD）提取年份。""" + text = str(release_date or "")[:4] + return int(text) if text.isdigit() else None + + def _init_values(model: type, data: dict[str, Any]) -> dict[str, Any]: init_names = {item.name for item in fields(model) if item.init} return {key: value for key, value in data.items() if key in init_names} @@ -64,6 +93,11 @@ class MusicMeta: """返回搜索和展示使用的音乐名称。""" return self.album or self.title or "" + @property + def original_name(self) -> str: + """返回未经过通用识别词处理的原始名称，兼容影视识别链的公共访问。""" + return self.org_string or self.title or self.album or "" + @property def artist(self) -> str: """返回兼容现有展示组件的艺术家文本。""" @@ -189,10 +223,18 @@ class MusicInfo: type: MediaType = field(default=MediaType.MUSIC, init=False) source: str | None = None media_id: str | None = None + # 音乐实体类型，用于区分单曲、专辑和艺术家三类可浏览对象 + music_type: str = MUSIC_ENTITY_RECORDING title: str | None = None artists: list[str] = field(default_factory=list) + # 艺术家标准 ID，顺序与 artists 一致，供详情页关联跳转 + artist_ids: list[str] = field(default_factory=list) album: str | None = None album_artist: str | None = None + # 所属专辑标准 ID（MusicBrainz Release Group） + album_id: str | None = None + # 专辑主类型：Album、EP、Single 等 + album_type: str | None = None year: int | None = None release_date: str | None = None disc_number: int | None = None @@ -204,6 +246,7 @@ class MusicInfo: lyrics: str | None = None version: str | None = None category: str = "" + genres: list[str] = field(default_factory=list) names: list[str] = field(default_factory=list) detail_link: str | None = None listen_count: int | None = None @@ -321,7 +364,10 @@ class MusicInfo: _validate_music_type(data.get("type")) values = _init_values(cls, data) values["artists"] = _string_list(values.get("artists") or data.get("artist")) + values["artist_ids"] = _aligned_list(values.get("artist_ids")) + values["genres"] = _string_list(values.get("genres")) values["names"] = _string_list(values.get("names")) + values["music_type"] = str(values.get("music_type") or MUSIC_ENTITY_RECORDING) values["raw_data"] = dict(values.get("raw_data") or {}) for key in ( "year", @@ -333,3 +379,283 @@ class MusicInfo: ): values[key] = _optional_int(values.get(key)) return cls(**values) + + +@dataclass +class MusicRelease: + """音乐专辑下的单个发行版本（MusicBrainz Release）。""" + + media_id: str | None = None + title: str | None = None + date: str | None = None + country: str | None = None + status: str | None = None + packaging: str | None = None + formats: list[str] = field(default_factory=list) + track_count: int | None = None + cover_url: str | None = None + + @property + def year(self) -> int | None: + """返回发行版本年份。""" + return _year_of(self.date) + + def to_dict(self) -> dict[str, Any]: + """转换为可传输的字典。""" + payload = asdict(self) + payload["year"] = self.year + return payload + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Self: + """从字典恢复发行版本信息。""" + values = _init_values(cls, data) + values["formats"] = _string_list(values.get("formats")) + values["track_count"] = _optional_int(values.get("track_count")) + return cls(**values) + + +@dataclass +class MusicAlbumInfo: + """标准化音乐专辑信息（MusicBrainz Release Group）。""" + + type: MediaType = field(default=MediaType.MUSIC, init=False) + music_type: str = field(default=MUSIC_ENTITY_ALBUM, init=False) + source: str | None = None + media_id: str | None = None + title: str | None = None + artists: list[str] = field(default_factory=list) + artist_ids: list[str] = field(default_factory=list) + # 专辑主类型：Album、EP、Single、Broadcast、Other + album_type: str | None = None + # 专辑副类型：Live、Compilation、Soundtrack、Remix 等 + secondary_types: list[str] = field(default_factory=list) + release_date: str | None = None + cover_url: str | None = None + genres: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + rating: float = 0.0 + rating_votes: int | None = None + detail_link: str | None = None + # 专辑内的音乐，按碟号和音轨号排序 + tracks: list[MusicInfo] = field(default_factory=list) + # 同一专辑下的其它发行版本 + releases: list[MusicRelease] = field(default_factory=list) + raw_data: dict[str, Any] = field(default_factory=dict) + + @property + def artist(self) -> str: + """返回兼容现有展示组件的艺术家文本。""" + return " / ".join(self.artists) + + @property + def year(self) -> int | None: + """返回专辑首次发行年份。""" + return _year_of(self.release_date) + + @property + def category(self) -> str: + """返回专辑主类型与副类型组合成的分类文本。""" + return " / ".join(part for part in [self.album_type, *self.secondary_types] if part) + + @property + def track_count(self) -> int: + """返回专辑内已解析的音乐数量。""" + return len(self.tracks) + + @property + def duration(self) -> int | None: + """返回专辑内所有音乐时长之和。""" + durations = [track.duration for track in self.tracks if track.duration] + return sum(durations) if durations else None + + @property + def title_year(self) -> str: + """返回包含年份的专辑展示标题。""" + if not self.title: + return "" + return f"{self.title} ({self.year})" if self.year else self.title + + @property + def poster_path(self) -> str | None: + """返回兼容现有媒体卡片的封面地址。""" + return self.cover_url + + @property + def backdrop_path(self) -> str | None: + """返回兼容现有详情页背景的封面地址。""" + return self.cover_url + + @property + def overview(self) -> str: + """返回专辑摘要，供卡片和通知复用。""" + parts = [self.artist, self.category, self.release_date, " / ".join(self.genres[:3])] + return " · ".join(part for part in parts if part) + + def to_dict(self) -> dict[str, Any]: + """转换为兼容前端 MediaInfo 结构的字典。""" + payload = { + key: value + for key, value in asdict(self).items() + if key not in {"tracks", "releases", "type"} + } + payload.update( + { + "type": self.type.value, + "artist": self.artist, + "album": self.title, + "year": self.year, + "category": self.category, + "duration": self.duration, + "total_tracks": self.track_count, + "title_year": self.title_year, + "poster_path": self.poster_path, + "backdrop_path": self.backdrop_path, + "mediaid_prefix": self.source, + "overview": self.overview, + "vote_average": self.rating, + "tracks": [track.to_dict() for track in self.tracks], + "releases": [release.to_dict() for release in self.releases], + } + ) + return payload + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Self: + """从字典恢复标准化专辑信息。""" + _validate_music_type(data.get("type")) + values = _init_values(cls, data) + for key in ("artists", "secondary_types", "genres", "tags"): + values[key] = _string_list(values.get(key)) + values["artist_ids"] = _aligned_list(values.get("artist_ids")) + values["rating"] = _optional_float(values.get("rating")) + values["rating_votes"] = _optional_int(values.get("rating_votes")) + values["tracks"] = [ + item if isinstance(item, MusicInfo) else MusicInfo.from_dict(item) + for item in data.get("tracks") or [] + ] + values["releases"] = [ + item if isinstance(item, MusicRelease) else MusicRelease.from_dict(item) + for item in data.get("releases") or [] + ] + values["raw_data"] = dict(values.get("raw_data") or {}) + return cls(**values) + + def to_music_info(self) -> MusicInfo: + """转换为专辑卡片使用的音乐信息，供列表接口统一返回。""" + return MusicInfo( + source=self.source, + media_id=self.media_id, + music_type=MUSIC_ENTITY_ALBUM, + title=self.title, + artists=list(self.artists), + artist_ids=list(self.artist_ids), + album=self.title, + album_artist=self.artist or None, + album_id=self.media_id, + album_type=self.album_type, + year=self.year, + release_date=self.release_date, + total_tracks=self.track_count or None, + duration=self.duration, + cover_url=self.cover_url, + category=self.category, + genres=list(self.genres), + names=[name for name in (self.title,) if name], + detail_link=self.detail_link, + ) + + +@dataclass +class MusicArtistInfo: + """标准化音乐艺术家信息（MusicBrainz Artist）。""" + + type: MediaType = field(default=MediaType.MUSIC, init=False) + music_type: str = field(default=MUSIC_ENTITY_ARTIST, init=False) + source: str | None = None + media_id: str | None = None + name: str | None = None + sort_name: str | None = None + # MusicBrainz 消歧义说明，同名艺术家依靠该字段区分 + disambiguation: str | None = None + # 艺术家类型：Person、Group、Orchestra、Choir、Character、Other + artist_type: str | None = None + gender: str | None = None + country: str | None = None + area: str | None = None + begin_date: str | None = None + end_date: str | None = None + ended: bool = False + genres: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + aliases: list[str] = field(default_factory=list) + # 关联艺术家场景下的关系文本，例如乐队成员、子团体 + relation: str | None = None + image_url: str | None = None + detail_link: str | None = None + # 外部站点链接，键为关系类型，值为地址 + external_links: dict[str, str] = field(default_factory=dict) + album_count: int | None = None + raw_data: dict[str, Any] = field(default_factory=dict) + + @property + def title(self) -> str | None: + """返回兼容通用媒体展示组件的标题。""" + return self.name + + @property + def life_span(self) -> str: + """返回艺术家活跃时间区间文本。""" + if not self.begin_date and not self.end_date: + return "" + end = self.end_date or ("" if self.ended else "…") + return f"{self.begin_date or '?'} - {end}" if end else (self.begin_date or "") + + @property + def overview(self) -> str: + """返回艺术家摘要，供卡片和详情页复用。""" + parts = [ + self.artist_type, + self.disambiguation, + self.area or self.country, + self.life_span, + " / ".join(self.genres[:3]), + ] + return " · ".join(part for part in parts if part) + + @property + def poster_path(self) -> str | None: + """返回兼容现有卡片的艺术家图片地址。""" + return self.image_url + + def to_dict(self) -> dict[str, Any]: + """转换为可传输的字典。""" + payload = {key: value for key, value in asdict(self).items() if key != "type"} + payload.update( + { + "type": self.type.value, + "title": self.title, + "life_span": self.life_span, + "overview": self.overview, + "poster_path": self.poster_path, + "mediaid_prefix": self.source, + } + ) + return payload + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Self: + """从字典恢复标准化艺术家信息。""" + _validate_music_type(data.get("type")) + values = _init_values(cls, data) + for key in ("genres", "tags", "aliases"): + values[key] = _string_list(values.get(key)) + values["ended"] = bool(values.get("ended")) + values["album_count"] = _optional_int(values.get("album_count")) + values["external_links"] = { + str(key): str(value) + for key, value in (values.get("external_links") or {}).items() + if value + } + values["raw_data"] = dict(values.get("raw_data") or {}) + return cls(**values) diff --git a/app/locales/en-US.json b/app/locales/en-US.json index c464eed10..369319501 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -110,6 +110,8 @@ "无法识别媒体信息": "Unable to recognize media information", "未识别到媒体信息": "Unable to recognize media information", "未识别到音乐信息": "Unable to recognize music information", + "未识别到专辑信息": "Unable to recognize album information", + "未识别到艺术家信息": "Unable to recognize artist information", "记录不存在": "Record does not exist", "MoviePilot智能助手未启用": "MoviePilot Assistant is not enabled", "整理记录不存在": "Organization record does not exist", @@ -123,6 +125,8 @@ "刮削路径无效": "Scraping path is invalid", "指定媒体ID时必须同时指定媒体数据源": "The media source must be specified together with the media ID", "媒体ID格式无效": "Invalid media ID format", + "MusicBrainz 只能用于音乐刮削": "MusicBrainz can only be used for music scraping", + "刮削失败，无法识别音乐信息": "Scraping failed: unable to recognize music information", "刮削失败，无法识别媒体信息": "Scraping failed: unable to recognize media information", "刮削路径不存在": "Scraping path does not exist", "保存成功": "Saved successfully", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 966a649ae..3154a54dc 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -110,6 +110,8 @@ "无法识别媒体信息": "無法識別媒體資訊", "未识别到媒体信息": "未識別到媒體資訊", "未识别到音乐信息": "未識別到音樂資訊", + "未识别到专辑信息": "未識別到專輯資訊", + "未识别到艺术家信息": "未識別到藝術家資訊", "记录不存在": "記錄不存在", "MoviePilot智能助手未启用": "MoviePilot 智慧助手未啟用", "整理记录不存在": "整理記錄不存在", @@ -123,6 +125,8 @@ "刮削路径无效": "刮削路徑無效", "指定媒体ID时必须同时指定媒体数据源": "指定媒體ID時必須同時指定媒體資料源", "媒体ID格式无效": "媒體ID格式無效", + "MusicBrainz 只能用于音乐刮削": "MusicBrainz 只能用於音樂刮削", + "刮削失败，无法识别音乐信息": "刮削失敗，無法識別音樂資訊", "刮削失败，无法识别媒体信息": "刮削失敗，無法識別媒體資訊", "刮削路径不存在": "刮削路徑不存在", "保存成功": "儲存成功", diff --git a/app/modules/emby/__init__.py b/app/modules/emby/__init__.py index f25e9faaf..2a12966c7 100644 --- a/app/modules/emby/__init__.py +++ b/app/modules/emby/__init__.py @@ -151,6 +151,20 @@ class EmbyModule(_ModuleBase, _MediaServerBase[Emby]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = s.get_music( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="emby", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/emby/emby.py b/app/modules/emby/emby.py index e68424cd3..0f395abb2 100644 --- a/app/modules/emby/emby.py +++ b/app/modules/emby/emby.py @@ -164,6 +164,8 @@ class Emby: continue if library.get("CollectionType") == "movies": library_type = MediaType.MOVIE.value + elif library.get("CollectionType") in ("music", "musicvideos", "audio"): + library_type = MediaType.MUSIC.value elif library.get("CollectionType") == "tvshows": library_type = MediaType.TV.value else: @@ -180,7 +182,10 @@ class Emby: name=library.get("Name"), path=library.get("Path"), type=library_type, - item_count=self.get_items_count(library.get("Id")), + item_count=self.get_items_count( + library.get("Id"), + "MusicAlbum" if library_type == MediaType.MUSIC.value else "Movie,Series", + ), image=image, link=f'{self._playhost or self._host}web/index.html' f'#!/videos?{server_query}parentId={library.get("Id")}', @@ -326,7 +331,9 @@ class Emby: return schemas.Statistic( movie_count=result.get("MovieCount") or 0, tv_count=result.get("SeriesCount") or 0, - episode_count=result.get("EpisodeCount") or 0 + episode_count=result.get("EpisodeCount") or 0, + music_count=result.get("MusicAlbumCount") or result.get("AlbumCount") + or result.get("SongCount") or 0, ) else: logger.error(f"Items/Counts 未获取到返回数据") @@ -412,7 +419,33 @@ class Emby: except Exception as e: logger.error(f"连接Items出错：" + str(e)) return None - return [] + return None + + def get_music( + self, title: Optional[str] = None, artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询 Emby 音乐条目。""" + if not self._host or not self._apikey or not self.user: + return [] + query = " ".join(filter(None, [title, artist, album])).strip() + if not query: + return [] + url = f"{self._host}emby/Users/{self.user}/Items" + params = { + "IncludeItemTypes": "MusicAlbum,Audio", + "SearchTerm": query, + "Recursive": "true", + "Limit": 20, + "api_key": self._apikey, + } + try: + res = RequestUtils().get_res(url, params) + items = res.json().get("Items", []) if res and res.status_code == 200 else [] + return [item for item in (self.__format_item_info(item) for item in items) if item] + except Exception as e: + logger.debug(f"查询Emby音乐出错：{e}") + return [] def get_tv_episodes(self, item_id: Optional[str] = None, @@ -742,7 +775,7 @@ class Emby: logger.error(f"连接/Users/{self.user}/Items/{itemid}出错：" + str(e)) return None - def get_items_count(self, parent: Union[str, int]) -> Optional[int]: + def get_items_count(self, parent: Union[str, int], include_item_types: str = "Movie,Series") -> Optional[int]: """ 获取指定媒体库可同步的电影和剧集总数 @@ -755,7 +788,7 @@ class Emby: params = { "ParentId": parent, "Recursive": "true", - "IncludeItemTypes": "Movie,Series", + "IncludeItemTypes": include_item_types, "Limit": 0, "api_key": self._apikey, } @@ -804,7 +837,7 @@ class Emby: if "Folder" in item.get("Type"): for items in self.get_items(parent=item.get('Id')): yield items - elif item.get("Type") in ["Movie", "Series"]: + elif item.get("Type") in ["Movie", "Series", "MusicAlbum"]: yield self.__format_item_info(item) except Exception as e: logger.error(f"连接Users/Items出错：" + str(e)) diff --git a/app/modules/jellyfin/__init__.py b/app/modules/jellyfin/__init__.py index ba738d808..33aaa46e0 100644 --- a/app/modules/jellyfin/__init__.py +++ b/app/modules/jellyfin/__init__.py @@ -152,6 +152,20 @@ class JellyfinModule(_ModuleBase, _MediaServerBase[Jellyfin]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = s.get_music( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="jellyfin", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/jellyfin/jellyfin.py b/app/modules/jellyfin/jellyfin.py index 42777c812..63f3b93d2 100644 --- a/app/modules/jellyfin/jellyfin.py +++ b/app/modules/jellyfin/jellyfin.py @@ -180,6 +180,10 @@ class Jellyfin: library_type = MediaType.TV.value link = f"{self._playhost or self._host}web/index.html#!" \ f"/tv.html?topParentId={library.get('Id')}" + elif library.get("CollectionType") in ("music", "musicvideos"): + library_type = MediaType.MUSIC.value + link = f"{self._playhost or self._host}web/index.html#!" \ + f"/music.html?topParentId={library.get('Id')}" else: library_type = MediaType.UNKNOWN.value link = f"{self._playhost or self._host}web/index.html#!" \ @@ -192,7 +196,10 @@ class Jellyfin: name=library.get("Name"), path=library.get("Path"), type=library_type, - item_count=self.get_items_count(library.get("Id")), + item_count=self.get_items_count( + library.get("Id"), + include_item_types="MusicAlbum" if library_type == MediaType.MUSIC.value else "Movie,Series", + ), image=image, link=link, server_type="jellyfin" @@ -359,7 +366,9 @@ class Jellyfin: return schemas.Statistic( movie_count=result.get("MovieCount") or 0, tv_count=result.get("SeriesCount") or 0, - episode_count=result.get("EpisodeCount") or 0 + episode_count=result.get("EpisodeCount") or 0, + music_count=result.get("MusicAlbumCount") or result.get("AlbumCount") + or result.get("SongCount") or 0, ) else: logger.error(f"Items/Counts 未获取到返回数据") @@ -392,6 +401,8 @@ class Jellyfin: elif collection_type == "tvshows": stat.tv_count += self.get_items_count(library_id, include_item_types="Series") or 0 stat.episode_count += self.get_items_count(library_id, include_item_types="Episode") or 0 + elif collection_type in ("music", "musicvideos"): + stat.music_count += self.get_items_count(library_id, include_item_types="MusicAlbum") or 0 return stat def __get_jellyfin_series_id_by_name(self, name: str, year: str) -> Optional[str]: @@ -464,7 +475,33 @@ class Jellyfin: except Exception as e: logger.error(f"连接Items出错：" + str(e)) return None - return [] + return None + + def get_music( + self, title: Optional[str] = None, artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询 Jellyfin 音乐条目。""" + if not self._host or not self._apikey or not self.user: + return [] + query = " ".join(filter(None, [title, artist, album])).strip() + if not query: + return [] + url = f"{self._host}Users/{self.user}/Items" + params = { + "IncludeItemTypes": "MusicAlbum,Audio", + "searchTerm": query, + "Recursive": "true", + "Limit": 20, + "api_key": self._apikey, + } + try: + res = self._request().get_res(url, params) + items = res.json().get("Items", []) if res and res.status_code == 200 else [] + return [item for item in (self.__format_item_info(item) for item in items) if item] + except Exception as e: + logger.debug(f"查询Jellyfin音乐出错：{e}") + return [] def get_tv_episodes(self, item_id: Optional[str] = None, @@ -958,7 +995,7 @@ class Jellyfin: if "Folder" in item.get("Type"): for items in self.get_items(item.get("Id")): yield items - elif item.get("Type") in ["Movie", "Series"]: + elif item.get("Type") in ["Movie", "Series", "MusicAlbum"]: yield self.__format_item_info(item) except Exception as e: logger.error(f"连接Users/Items出错：" + str(e)) diff --git a/app/modules/listenbrainz/__init__.py b/app/modules/listenbrainz/__init__.py index 2275f5e31..3a4f0079f 100644 --- a/app/modules/listenbrainz/__init__.py +++ b/app/modules/listenbrainz/__init__.py @@ -1,26 +1,55 @@ from typing import Any, Optional, Tuple, Union +from app.core.cache import cached from app.core.config import settings -from app.core.music import MusicInfo +from app.core.music import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MusicInfo from app.log import logger from app.modules import _ModuleBase from app.schemas.types import ModuleType, OtherModulesType from app.utils.http import RequestUtils +# ListenBrainz 全站统计支持的周期，取值与官方统计页面完全一致 +LISTENBRAINZ_CHART_RANGES = ( + "this_week", + "this_month", + "this_year", + "week", + "month", + "quarter", + "half_yearly", + "year", + "all_time", +) +# ListenBrainz 官方新发行页面支持的排序方式 +LISTENBRAINZ_FRESH_SORTS = ( + "release_date", + "artist_credit_name", + "release_name", +) +# ListenBrainz 新发行页面允许回溯或预告的最大天数 +LISTENBRAINZ_FRESH_MAX_DAYS = 90 + class ListenBrainzModule(_ModuleBase): - """通过 ListenBrainz 全站统计提供音乐推荐与探索榜单。""" + """通过 ListenBrainz 全站统计与新发行数据提供音乐探索能力。""" _base_url = "https://api.listenbrainz.org/1" _detail_url = "https://musicbrainz.org/recording" - _cover_url = "https://coverartarchive.org/release" + _album_detail_url = "https://musicbrainz.org/release-group" + _release_cover_url = "https://coverartarchive.org/release" + _release_group_cover_url = "https://coverartarchive.org/release-group" _source = "musicbrainz" + # 全站统计按实体分为不同接口，键为音乐实体类型，值为接口路径与数据字段 + _chart_entities = { + MUSIC_ENTITY_RECORDING: ("recordings", "recordings"), + MUSIC_ENTITY_ALBUM: ("release-groups", "release_groups"), + } def init_module(self) -> None: - """初始化无状态的 ListenBrainz 榜单模块。""" + """初始化无状态的 ListenBrainz 探索模块。""" def init_setting(self) -> Optional[Tuple[str, Union[str, bool]]]: - """ListenBrainz 公共榜单无需独立密钥或启用开关。""" + """ListenBrainz 公共数据无需独立密钥或启用开关。""" return None def stop(self) -> None: @@ -28,7 +57,7 @@ class ListenBrainzModule(_ModuleBase): def test(self) -> Tuple[bool, str]: """测试 ListenBrainz 全站榜单接口连通性。""" - result = self._request_chart(range_name="this_week", offset=0, count=1) + result = self._request_chart(entity="recordings", range_name="this_week", offset=0, count=1) return (True, "") if result is not None else (False, "ListenBrainz 网络连接失败") @staticmethod @@ -56,28 +85,92 @@ class ListenBrainzModule(_ModuleBase): range_name: str, offset: int = 0, count: int = 30, + entity: str = MUSIC_ENTITY_RECORDING, ) -> list[MusicInfo]: - """读取指定统计周期的全站录音榜单。""" + """读取指定周期的全站热门单曲或热门专辑榜单。""" + path, field = self._chart_entities.get( + entity, self._chart_entities[MUSIC_ENTITY_RECORDING] + ) payload = self._request_chart( - range_name=range_name, + entity=path, + range_name=range_name if range_name in LISTENBRAINZ_CHART_RANGES else "this_month", offset=max(offset, 0), count=max(1, min(count, 100)), ) - recordings = ((payload or {}).get("payload") or {}).get("recordings") or [] - return [ - info - for item in recordings - if (info := self._recording_to_info(item)) - ] + items = ((payload or {}).get("payload") or {}).get(field) or [] + convert = ( + self._release_group_to_info + if entity == MUSIC_ENTITY_ALBUM + else self._recording_to_info + ) + return [info for item in items if (info := convert(item))] + + def music_fresh_releases( + self, + days: int = 14, + sort: str = "release_date", + past: bool = True, + future: bool = True, + offset: int = 0, + count: int = 30, + ) -> list[MusicInfo]: + """读取 ListenBrainz 官方新发行专辑，并按官方排序方式分页返回。""" + releases = self._fresh_releases( + days=max(1, min(days, LISTENBRAINZ_FRESH_MAX_DAYS)), + sort=sort if sort in LISTENBRAINZ_FRESH_SORTS else "release_date", + past=bool(past), + future=bool(future), + ) + # 官方接口一次性返回整个时间窗口的全部发行，分页只能在结果集上切片 + start = max(offset, 0) + window = releases[start: start + max(1, min(count, 100))] + return [info for item in window if (info := self._fresh_release_to_info(item))] + + @cached(maxsize=32, ttl=1800, skip_empty=True) + def _fresh_releases( + self, + days: int, + sort: str, + past: bool, + future: bool, + ) -> list[dict[str, Any]]: + """缓存官方新发行原始结果，避免翻页时重复拉取整个时间窗口。""" + payload = self._request_json( + "/explore/fresh-releases/", + params={ + "days": days, + "sort": sort, + "past": str(past).lower(), + "future": str(future).lower(), + }, + ) + return ((payload or {}).get("payload") or {}).get("releases") or [] @classmethod def _request_chart( cls, + entity: str, range_name: str, offset: int, count: int, ) -> Optional[dict[str, Any]]: - """请求 ListenBrainz 全站录音统计并统一处理异常响应。""" + """请求 ListenBrainz 全站统计接口。""" + return cls._request_json( + f"/stats/sitewide/{entity}", + params={ + "range": range_name, + "offset": offset, + "count": count, + }, + ) + + @classmethod + def _request_json( + cls, + path: str, + params: Optional[dict[str, Any]] = None, + ) -> Optional[dict[str, Any]]: + """请求 ListenBrainz JSON 接口并统一处理网络和响应错误。""" response = RequestUtils( headers={ "User-Agent": f"{settings.USER_AGENT} (https://github.com/jxxghp/MoviePilot)", @@ -85,17 +178,14 @@ class ListenBrainzModule(_ModuleBase): }, proxies=settings.PROXY, timeout=20, - ).get_res( - f"{cls._base_url}/stats/sitewide/recordings", - params={ - "range": range_name, - "offset": offset, - "count": count, - }, - ) + ).get_res(f"{cls._base_url}{path}", params=params) if not response: return None try: + if response.status_code == 204: + # 统计尚未生成时官方返回 204，视为空结果而不是失败 + logger.debug(f"ListenBrainz 统计尚未生成：{path}") + return {} if response.status_code != 200: logger.warning( f"ListenBrainz 请求失败：{response.status_code} {response.text[:200]}" @@ -110,7 +200,7 @@ class ListenBrainzModule(_ModuleBase): @classmethod def _recording_to_info(cls, recording: dict[str, Any]) -> Optional[MusicInfo]: - """将 ListenBrainz 录音统计转换为标准音乐信息。""" + """将 ListenBrainz 热门单曲统计转换为标准音乐信息。""" media_id = recording.get("recording_mbid") title = recording.get("track_name") if not media_id or not title: @@ -121,16 +211,102 @@ class ListenBrainzModule(_ModuleBase): return MusicInfo( source=cls._source, media_id=str(media_id), + music_type=MUSIC_ENTITY_RECORDING, title=str(title), artists=[artist_name] if artist_name else [], + artist_ids=cls._primary_artist_ids(recording.get("artist_mbids")), album=release_name or None, - cover_url=f"{cls._cover_url}/{release_mbid}/front-500" if release_mbid else None, + cover_url=cls._release_cover(release_mbid), names=[name for name in (title, release_name) if name], detail_link=f"{cls._detail_url}/{media_id}", listen_count=cls._optional_int(recording.get("listen_count")), raw_data=recording, ) + @classmethod + def _release_group_to_info(cls, release_group: dict[str, Any]) -> Optional[MusicInfo]: + """将 ListenBrainz 热门专辑统计转换为标准音乐信息。""" + media_id = release_group.get("release_group_mbid") + title = release_group.get("release_group_name") + if not media_id or not title: + return None + artist_name = str(release_group.get("artist_name") or "").strip() + return MusicInfo( + source=cls._source, + media_id=str(media_id), + music_type=MUSIC_ENTITY_ALBUM, + title=str(title), + artists=[artist_name] if artist_name else [], + artist_ids=cls._primary_artist_ids(release_group.get("artist_mbids")), + album=str(title), + album_artist=artist_name or None, + album_id=str(media_id), + cover_url=cls._release_cover(release_group.get("caa_release_mbid")) + or f"{cls._release_group_cover_url}/{media_id}/front-500", + names=[str(title)], + detail_link=f"{cls._album_detail_url}/{media_id}", + listen_count=cls._optional_int(release_group.get("listen_count")), + raw_data=release_group, + ) + + @classmethod + def _fresh_release_to_info(cls, release: dict[str, Any]) -> Optional[MusicInfo]: + """将 ListenBrainz 新发行条目转换为标准专辑信息。""" + media_id = release.get("release_group_mbid") + title = release.get("release_name") + if not media_id or not title: + return None + artist_name = str(release.get("artist_credit_name") or "").strip() + release_date = release.get("release_date") or None + category_parts = [ + release.get("release_group_primary_type"), + release.get("release_group_secondary_type"), + ] + return MusicInfo( + source=cls._source, + media_id=str(media_id), + music_type=MUSIC_ENTITY_ALBUM, + title=str(title), + artists=[artist_name] if artist_name else [], + artist_ids=cls._primary_artist_ids(release.get("artist_mbids")), + album=str(title), + album_artist=artist_name or None, + album_id=str(media_id), + album_type=release.get("release_group_primary_type") or None, + year=cls._year(release_date), + release_date=release_date, + cover_url=cls._release_cover(release.get("caa_release_mbid")) + or f"{cls._release_group_cover_url}/{media_id}/front-500", + category=" / ".join(str(part) for part in category_parts if part), + genres=[str(tag) for tag in release.get("release_tags") or [] if tag], + names=[str(title)], + detail_link=f"{cls._album_detail_url}/{media_id}", + listen_count=cls._optional_int(release.get("listen_count")), + raw_data=release, + ) + + @classmethod + def _release_cover(cls, release_mbid: Any) -> Optional[str]: + """根据发行版本 ID 构造 Cover Art Archive 封面地址。""" + return f"{cls._release_cover_url}/{release_mbid}/front-500" if release_mbid else None + + @staticmethod + def _primary_artist_ids(artist_mbids: Any) -> list[str]: + """返回可跳转的主艺术家 ID。 + + ListenBrainz 的艺术家名称是合并署名文本，无法与多个 ID 一一对应， + 因此只保留首个 ID 作为卡片上的主艺术家入口。 + """ + if not isinstance(artist_mbids, (list, tuple)): + return [] + return [str(artist_mbids[0])] if artist_mbids and artist_mbids[0] else [] + + @staticmethod + def _year(release_date: Any) -> Optional[int]: + """从 ListenBrainz 发行日期提取年份。""" + text = str(release_date or "")[:4] + return int(text) if text.isdigit() else None + @staticmethod def _optional_int(value: Any) -> Optional[int]: """把 ListenBrainz 统计值转换为可选整数。""" diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index 0fcc4f102..18df44f7f 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/app/modules/musicbrainz/__init__.py @@ -1,10 +1,16 @@ import threading import time -from datetime import datetime from typing import Any, Optional, Tuple, Union from app.core.config import settings -from app.core.music import MusicInfo, MusicMeta +from app.core.music import ( + MUSIC_ENTITY_ALBUM, + MusicAlbumInfo, + MusicArtistInfo, + MusicInfo, + MusicMeta, + MusicRelease, +) from app.log import logger from app.modules import _ModuleBase from app.schemas.types import MediaRecognizeType, ModuleType @@ -17,10 +23,40 @@ class MusicBrainzModule(_ModuleBase): _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 模块。""" @@ -76,17 +112,99 @@ class MusicBrainzModule(_ModuleBase): ] def recognize_music(self, source: str, media_id: str) -> Optional[MusicInfo]: - """按 MusicBrainz Recording ID 获取标准化音乐详情。""" + """按 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", + "inc": "artists+releases+release-groups+isrcs+genres", "fmt": "json", }, ) - return self._recording_to_info(payload) if payload else None + if payload: + return self._recording_to_info(payload) + # 订阅只持久化来源和 ID，无法区分单曲与专辑，因此按专辑再查一次 + 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: MusicMeta) -> str: @@ -119,8 +237,8 @@ class MusicBrainzModule(_ModuleBase): release_group = (release or {}).get("release-group") or {} release_date = cls._release_date(recording, release) album = (release or {}).get("title") - artists = cls._artist_names(recording.get("artist-credit")) - album_artists = cls._artist_names((release or {}).get("artist-credit")) + 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( @@ -128,8 +246,11 @@ class MusicBrainzModule(_ModuleBase): 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")), @@ -137,21 +258,164 @@ class MusicBrainzModule(_ModuleBase): 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_names(artist_credit: Optional[list[dict[str, Any]]]) -> list[str]: - """从 MusicBrainz artist-credit 提取有序艺术家名称。""" - results = [] + 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 name and name not in results: - results.append(str(name)) - return results + 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]: @@ -193,6 +457,113 @@ class MusicBrainzModule(_ModuleBase): 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 封面地址。""" @@ -229,6 +600,10 @@ class MusicBrainzModule(_ModuleBase): 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]}" diff --git a/app/modules/navidrome/__init__.py b/app/modules/navidrome/__init__.py new file mode 100644 index 000000000..6f9026337 --- /dev/null +++ b/app/modules/navidrome/__init__.py @@ -0,0 +1,205 @@ +"""Navidrome 媒体服务器模块。""" + +from typing import Any, Dict, Generator, List, Optional, Tuple, Union + +from app import schemas +from app.core.context import MediaInfo +from app.core.event import eventmanager +from app.log import logger +from app.modules import _MediaServerBase, _ModuleBase +from app.modules.navidrome.navidrome import Navidrome +from app.schemas import AuthCredentials, AuthInterceptCredentials +from app.schemas.types import ChainEventType, MediaServerType, MediaType, ModuleType + + +class NavidromeModule(_ModuleBase, _MediaServerBase[Navidrome]): + """将 Navidrome 接入 MoviePilot 媒体服务器统一接口。""" + + def init_module(self) -> None: + """读取配置并初始化启用的 Navidrome 实例。""" + super().init_service( + service_name=Navidrome.__name__.lower(), + service_type=lambda conf: Navidrome(**conf.config, sync_libraries=conf.sync_libraries), + ) + + @staticmethod + def get_name() -> str: + """返回模块显示名称。""" + return "Navidrome" + + @staticmethod + def get_type() -> ModuleType: + """返回模块类型。""" + return ModuleType.MediaServer + + @staticmethod + def get_subtype() -> MediaServerType: + """返回媒体服务器子类型。""" + return MediaServerType.Navidrome + + @staticmethod + def get_priority() -> int: + """返回模块优先级。""" + return 7 + + def stop(self) -> None: + """释放 Navidrome 模块资源。""" + + def init_setting(self) -> Optional[Tuple[str, Union[str, bool]]]: + """Navidrome 与其它媒体服务器一致，由服务配置控制启用，无需系统开关。""" + return None + + def scheduler_job(self) -> None: + """定时检查 Navidrome 连接状态并尝试重连。""" + for name, server in self.get_instances().items(): + if server.is_inactive(): + logger.info(f"Navidrome {name} 服务器连接断开，尝试重连 ...") + server.reconnect() + + def test(self) -> Optional[Tuple[bool, str]]: + """测试所有已启用 Navidrome 配置的连通性。""" + if not self.get_instances(): + return None + for name, server in self.get_instances().items(): + if server.is_inactive(): + server.reconnect() + if not server.get_user(): + return False, f"无法连接Navidrome服务器：{name}" + return True, "" + + def user_authenticate( + self, credentials: AuthCredentials, service_name: Optional[str] = None + ) -> Optional[AuthCredentials]: + """使用 Navidrome 用户凭据完成辅助认证。""" + if not credentials or credentials.grant_type != "password": + return None + servers = ( + [(service_name, self.get_instance(service_name))] + if service_name and self.get_instance(service_name) + else list(self.get_instances().items()) + ) + for name, server in servers: + if not server: + continue + intercept_event = eventmanager.send_event( + etype=ChainEventType.AuthIntercept, + data=AuthInterceptCredentials( + username=credentials.username, + channel=self.get_name(), + service=name, + status="triggered", + ), + ) + if intercept_event and intercept_event.event_data and intercept_event.event_data.cancel: + continue + token = server.authenticate(credentials.username, credentials.password) + if token: + credentials.channel = self.get_name() + credentials.service = name + credentials.token = token + return credentials + return None + + def media_exists( + self, mediainfo: MediaInfo, itemid: Optional[str] = None, server: Optional[str] = None + ) -> Optional[schemas.ExistMediaInfo]: + """判断音乐是否已存在于 Navidrome 音乐库。""" + if mediainfo.type != MediaType.MUSIC: + return None + servers = ( + [(server, self.get_instance(server))] + if server + else list(self.get_instances().items()) + ) + for name, service in servers: + if not service: + continue + if itemid and service.get_iteminfo(str(itemid)): + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="navidrome", + server=name, + itemid=itemid, + ) + matches = service.search_music( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="navidrome", + server=name, + itemid=matches[0].item_id, + ) + return None + + def media_statistic(self, server: Optional[str] = None) -> Optional[List[schemas.Statistic]]: + """返回 Navidrome 音乐数量统计。""" + servers = [self.get_instance(server)] if server else list(self.get_instances().values()) + result: List[schemas.Statistic] = [] + for service in servers: + if not service: + continue + statistic = service.get_medias_count() + statistic.user_count = service.get_user_count() + result.append(statistic) + return result + + def mediaserver_librarys( + self, server: str, username: Optional[str] = None, hidden: Optional[bool] = False + ) -> Optional[List[schemas.MediaServerLibrary]]: + """返回 Navidrome 的虚拟音乐库。""" + service = self.get_instance(server) + return service.get_librarys(hidden=hidden) if service else None + + def mediaserver_items( + self, server: str, library_id: Union[str, int], start_index: Optional[int] = 0, + limit: Optional[int] = -1, + ) -> Optional[Generator]: + """获取 Navidrome 专辑条目。""" + service = self.get_instance(server) + return service.get_items(start_index, limit) if service else None + + def mediaserver_items_count(self, server: str, library_id: Union[str, int]) -> Optional[int]: + """获取 Navidrome 音乐库条目数量。""" + service = self.get_instance(server) + return service.get_items_count(str(library_id)) if service else None + + def mediaserver_iteminfo(self, server: str, item_id: str) -> Optional[schemas.MediaServerItem]: + """获取 Navidrome 专辑详情。""" + service = self.get_instance(server) + return service.get_iteminfo(item_id) if service else None + + def mediaserver_tv_episodes(self, server: str, item_id: Union[str, int]) -> List[schemas.MediaServerSeasonInfo]: + """音乐服务器没有剧集信息，返回空列表。""" + return [] + + def mediaserver_playing( + self, server: str, count: Optional[int] = 20, username: Optional[str] = None + ) -> Optional[List[schemas.MediaServerPlayItem]]: + """获取 Navidrome 当前播放条目。""" + service = self.get_instance(server) + return service.get_resume(count) if service else None + + def mediaserver_play_url(self, server: str, item_id: Union[str, int]) -> Optional[str]: + """生成 Navidrome 音频流地址。""" + service = self.get_instance(server) + return service.get_play_url(str(item_id)) if service else None + + def mediaserver_latest( + self, server: Optional[str] = None, count: Optional[int] = 20, + username: Optional[str] = None, + ) -> Optional[List[schemas.MediaServerPlayItem]]: + """获取 Navidrome 最近新增专辑。""" + service = self.get_instance(server) + return service.get_latest(count) if service else None + + def mediaserver_latest_images( + self, server: Optional[str] = None, count: Optional[int] = 10, + username: Optional[str] = None, remote: Optional[bool] = False, + ) -> List[str]: + """获取最近新增专辑封面。""" + return [item.image for item in (self.mediaserver_latest(server, count, username) or []) if item.image] + diff --git a/app/modules/navidrome/navidrome.py b/app/modules/navidrome/navidrome.py new file mode 100644 index 000000000..5be202ed3 --- /dev/null +++ b/app/modules/navidrome/navidrome.py @@ -0,0 +1,232 @@ +"""Navidrome Subsonic/OpenSubsonic API 客户端。""" + +import hashlib +import secrets +from typing import Any, Dict, Generator, List, Optional +from urllib.parse import urlencode + +from app import schemas +from app.log import logger +from app.schemas.types import MediaType +from app.utils.http import RequestUtils +from app.utils.url import UrlUtils + + +class Navidrome: + """通过 Subsonic API 访问 Navidrome 音乐库。""" + + def __init__( + self, + host: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, + play_host: Optional[str] = None, + sync_libraries: Optional[list] = None, + **kwargs: Any, + ) -> None: + """初始化 Navidrome 连接配置。""" + self._host = UrlUtils.standardize_base_url(host) if host else None + self._play_host = UrlUtils.standardize_base_url(play_host) if play_host else self._host + self._username = username or "" + self._password = password or "" + self._sync_libraries = sync_libraries or [] + self.user = self.get_user() + + def _api_url(self) -> Optional[str]: + """返回 Subsonic REST 根地址。""" + if not self._host: + return None + return f"{self._host.rstrip('/')}/rest" + + def _params(self, **kwargs: Any) -> Dict[str, Any]: + """构造 Subsonic token 鉴权参数。""" + salt = secrets.token_hex(8) + token = hashlib.md5(f"{self._password}{salt}".encode("utf-8")).hexdigest() + params: Dict[str, Any] = { + "u": self._username, + "t": token, + "s": salt, + "v": "1.16.1", + "c": "MoviePilot", + "f": "json", + } + params.update({key: value for key, value in kwargs.items() if value is not None}) + return params + + def _call(self, method: str, **kwargs: Any) -> Optional[dict]: + """调用 Subsonic 方法并返回成功响应体。""" + url = self._api_url() + if not url or not self._username or not self._password: + return None + response = None + try: + response = RequestUtils().get_res(f"{url}/{method}", self._params(**kwargs)) + if not response: + return None + payload = response.json().get("subsonic-response", {}) + if payload.get("status") != "ok": + logger.warning(f"Navidrome API {method} 返回失败：{payload.get('error')}") + return None + return payload + except Exception as err: + logger.debug(f"Navidrome API {method} 调用失败：{err}") + return None + finally: + if response: + response.close() + + def is_inactive(self) -> bool: + """判断当前连接是否不可用。""" + return bool(self._host and self._username and self._password and not self.user) + + def reconnect(self) -> None: + """重新验证 Navidrome 连接。""" + self.user = self.get_user() + + def get_user(self) -> Optional[str]: + """验证凭据并返回配置用户名。""" + return self._username if self._call("ping") else None + + def authenticate(self, username: str, password: str) -> Optional[str]: + """验证用户凭据，成功时返回 Navidrome 用户名。""" + if not username or not password or not self._host: + return None + original_username, original_password = self._username, self._password + try: + self._username, self._password = username, password + return username if self._call("ping") else None + finally: + self._username, self._password = original_username, original_password + + def get_user_count(self) -> int: + """返回当前凭据可确认的用户数量。""" + return 1 if self.user else 0 + + @staticmethod + def _album_name(album: dict) -> str: + """从 Subsonic 专辑对象提取稳定标题。""" + return album.get("name") or album.get("album") or "" + + def _album_to_item(self, album: dict) -> schemas.MediaServerItem: + """将 Subsonic 专辑转换为统一媒体条目。""" + album_id = str(album.get("id") or "") + return schemas.MediaServerItem( + id=album_id, + item_id=album_id, + title=self._album_name(album), + original_title=self._album_name(album), + year=album.get("year") or album.get("created"), + item_type=MediaType.MUSIC.value, + server_type="navidrome", + path=album.get("path"), + note={"artist": album.get("artist"), "song_count": album.get("songCount")}, + ) + + def _album_cover(self, album: dict) -> Optional[str]: + """返回专辑封面地址。""" + cover_id = album.get("coverArt") + if not cover_id: + return None + url = self._api_url() + if not url: + return None + return f"{url}/getCoverArt?{urlencode(self._params(id=cover_id, size=512))}" + + def _albums(self, list_type: str = "alphabeticalByName", size: int = 500) -> List[dict]: + """分页读取专辑列表。""" + albums: List[dict] = [] + offset = 0 + while True: + payload = self._call("getAlbumList2", type=list_type, size=size, offset=offset) + page = ((payload or {}).get("albumList2") or {}).get("album") or [] + if not page: + break + albums.extend(page) + if len(page) < size: + break + offset += len(page) + return albums + + def get_medias_count(self) -> schemas.Statistic: + """统计 Navidrome 专辑数量并映射为音乐数量。""" + return schemas.Statistic(music_count=len(self._albums())) if self.user else schemas.Statistic() + + def get_librarys(self, hidden: Optional[bool] = False) -> Optional[List[schemas.MediaServerLibrary]]: + """返回单个虚拟音乐库。""" + if not self.user or (hidden and self._sync_libraries and "all" not in self._sync_libraries): + return [] + count = self.get_items_count("music") + return [schemas.MediaServerLibrary( + server="navidrome", + id="music", + item_id="music", + name="音乐", + type=MediaType.MUSIC.value, + item_count=count, + link=self._play_host, + server_type="navidrome", + )] + + def get_items_count(self, library_id: str = "music") -> int: + """返回虚拟音乐库中的专辑数量。""" + return len(self._albums()) if self.user else 0 + + def get_items(self, start_index: int = 0, limit: int = -1) -> Generator[schemas.MediaServerItem, None, None]: + """逐条返回 Navidrome 专辑。""" + albums = self._albums() + end = None if limit is None or limit == -1 else start_index + limit + for album in albums[start_index:end]: + yield self._album_to_item(album) + + def get_iteminfo(self, item_id: str) -> Optional[schemas.MediaServerItem]: + """获取 Navidrome 专辑详情。""" + payload = self._call("getAlbum", id=item_id) + album = ((payload or {}).get("album") or {}) + return self._album_to_item(album) if album else None + + def search_music( + self, title: Optional[str] = None, artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询音乐条目。""" + query = " ".join(filter(None, [title, artist, album])).strip() + if not query: + return [] + payload = self._call("search3", query=query, artistCount=0, albumCount=20, songCount=20) + result = ((payload or {}).get("searchResult3") or {}) + albums = result.get("album") or [] + songs = result.get("song") or [] + candidates = albums or songs + return [self._album_to_item(item) for item in candidates] + + def _to_play_item(self, album: dict) -> schemas.MediaServerPlayItem: + """将专辑转换为仪表盘播放/最新条目。""" + album_id = str(album.get("id") or "") + return schemas.MediaServerPlayItem( + id=album_id, + item_id=album_id, + title=self._album_name(album), + subtitle=album.get("artist") or album.get("albumArtist"), + type=MediaType.MUSIC.value, + image=self._album_cover(album), + link=self._play_host, + server_type="navidrome", + ) + + def get_latest(self, count: int = 20) -> List[schemas.MediaServerPlayItem]: + """返回最近新增专辑。""" + return [self._to_play_item(album) for album in self._albums("newest")[:count]] + + def get_resume(self, count: int = 20) -> List[schemas.MediaServerPlayItem]: + """返回当前用户正在播放的音乐。""" + payload = self._call("getNowPlaying") + items = ((payload or {}).get("nowPlaying") or {}).get("entry") or [] + return [self._to_play_item(item) for item in items[:count]] + + def get_play_url(self, item_id: str) -> Optional[str]: + """返回 Navidrome 流播放地址。""" + url = self._api_url() + if not url: + return None + return f"{url}/stream?{urlencode(self._params(id=item_id))}" + diff --git a/app/modules/plex/__init__.py b/app/modules/plex/__init__.py index 9d6839116..990d9ae32 100644 --- a/app/modules/plex/__init__.py +++ b/app/modules/plex/__init__.py @@ -161,6 +161,20 @@ class PlexModule(_ModuleBase, _MediaServerBase[Plex]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = s.get_music( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="plex", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/plex/plex.py b/app/modules/plex/plex.py index f507ceb21..5fda76735 100644 --- a/app/modules/plex/plex.py +++ b/app/modules/plex/plex.py @@ -144,6 +144,9 @@ class Plex: elif library.type == "show": library_type = MediaType.TV.value image_list = self.__get_library_images(library.key, 2) + elif library.type in ("artist", "music"): + library_type = MediaType.MUSIC.value + image_list = self.__get_library_images(library.key, 8) else: continue libraries.append( @@ -169,7 +172,7 @@ class Plex: if not self._plex: return schemas.Statistic() sections = self._plex.library.sections() - movie_count = tv_count = episode_count = 0 + movie_count = tv_count = episode_count = music_count = 0 # 媒体库白名单 allow_library = [str(lib.id) for lib in self.get_librarys(hidden=True) or []] for sec in sections: @@ -180,10 +183,13 @@ class Plex: if sec.type == "show": tv_count += sec.totalSize episode_count += sec.totalViewSize(libtype="episode") + if sec.type in ("artist", "music"): + music_count += sec.totalSize return schemas.Statistic( movie_count=movie_count, tv_count=tv_count, - episode_count=episode_count + episode_count=episode_count, + music_count=music_count, ) def get_movies(self, @@ -242,6 +248,36 @@ class Plex: ) return ret_movies + def get_music( + self, title: Optional[str] = None, artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询 Plex 音乐条目。""" + if not self._plex: + return [] + query = " ".join(filter(None, [title, artist, album])).strip() + if not query: + return [] + results: List[schemas.MediaServerItem] = [] + try: + for library in self._plex.library.sections(): + if library.type not in ("artist", "music"): + continue + for item in library.search(title=query): + results.append(schemas.MediaServerItem( + server="plex", + library=library.key, + item_id=getattr(item, "ratingKey", None) or getattr(item, "key", None), + item_type=MediaType.MUSIC.value, + title=getattr(item, "title", None), + original_title=getattr(item, "title", None), + year=getattr(item, "year", None), + path=getattr(item, "file", None), + )) + except Exception as e: + logger.debug(f"查询Plex音乐出错：{e}") + return results + def get_tv_episodes(self, item_id: Optional[str] = None, title: Optional[str] = None, diff --git a/app/modules/trimemedia/__init__.py b/app/modules/trimemedia/__init__.py index 058ba95a1..039513aa5 100644 --- a/app/modules/trimemedia/__init__.py +++ b/app/modules/trimemedia/__init__.py @@ -180,6 +180,20 @@ class TrimeMediaModule(_ModuleBase, _MediaServerBase[TrimeMedia]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = getattr(s, "get_music", lambda **_: [])( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="trimemedia", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/trimemedia/trimemedia.py b/app/modules/trimemedia/trimemedia.py index b4d90166f..6c91b3ddb 100644 --- a/app/modules/trimemedia/trimemedia.py +++ b/app/modules/trimemedia/trimemedia.py @@ -193,6 +193,8 @@ class TrimeMedia: library_type = MediaType.MOVIE.value elif library.category == fnapi.Category.TV: library_type = MediaType.TV.value + elif getattr(library.category, "value", "").lower() in ("music", "audio"): + library_type = MediaType.MUSIC.value elif library.category == fnapi.Category.OTHERS: # 忽略这个库 continue @@ -240,6 +242,7 @@ class TrimeMedia: return schemas.Statistic( movie_count=info.movie, tv_count=info.tv, + music_count=getattr(info, "music", 0) or getattr(info, "audio", 0) or 0, ) def authenticate(self, username: str, password: str) -> Optional[str]: diff --git a/app/modules/ugreen/__init__.py b/app/modules/ugreen/__init__.py index a791c0386..895836922 100644 --- a/app/modules/ugreen/__init__.py +++ b/app/modules/ugreen/__init__.py @@ -162,6 +162,20 @@ class UgreenModule(_ModuleBase, _MediaServerBase[Ugreen]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = getattr(s, "get_music", lambda **_: [])( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="ugreen", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/ugreen/ugreen.py b/app/modules/ugreen/ugreen.py index cfbca672a..ca44c09e4 100644 --- a/app/modules/ugreen/ugreen.py +++ b/app/modules/ugreen/ugreen.py @@ -413,6 +413,8 @@ class Ugreen: return MediaType.TV.value if "电影" in path or "电影" in name: return MediaType.MOVIE.value + if any(key in name.lower() for key in ["music", "audio", "音乐"]): + return MediaType.MUSIC.value return MediaType.UNKNOWN.value def __is_library_blocked(self, library_id: str) -> bool: @@ -594,12 +596,14 @@ class Ugreen: movie_data = self._api.video_all(classification=-102, page=1, page_size=1) or {} tv_data = self._api.video_all(classification=-103, page=1, page_size=1) or {} + music_data = self._api.video_all(classification=-104, page=1, page_size=1) or {} return schemas.Statistic( movie_count=int(movie_data.get("total_num") or 0), tv_count=int(tv_data.get("total_num") or 0), # 绿联当前不统计剧集总数，返回 None 由前端展示“未获取”。 episode_count=None, + music_count=int(music_data.get("total_num") or 0), ) def authenticate(self, username: str, password: str) -> Optional[str]: @@ -654,6 +658,33 @@ class Ugreen: movies.append(media_item) return movies + def get_music( + self, + title: Optional[str] = None, + artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询绿联影视音乐条目。""" + if not self.is_authenticated() or not self._api: + return [] + query = " ".join(filter(None, [album, title, artist])).strip() + if not query: + return [] + + data = self._api.search(query) + if not data: + return [] + + # 绿联搜索按媒体类型分桶返回，音乐相关桶名在不同固件版本上并不统一 + music_buckets = ("music_list", "audio_list", "album_list", "songs_list") + results: List[schemas.MediaServerItem] = [] + for bucket in music_buckets: + for info in self.__extract_video_info_list(data.get(bucket)): + media_item = self.__build_media_server_item(info) + if media_item: + results.append(media_item) + return results + def __search_tv_item(self, title: str, year: Optional[str] = None, tmdb_id: Optional[int] = None) -> Optional[dict]: if not self._api: return None diff --git a/app/modules/zspace/__init__.py b/app/modules/zspace/__init__.py index 3e03524ea..634032caf 100644 --- a/app/modules/zspace/__init__.py +++ b/app/modules/zspace/__init__.py @@ -145,6 +145,20 @@ class ZSpaceModule(_ModuleBase, _MediaServerBase[ZSpace]): for name, s in servers: if not s: continue + if mediainfo.type == MediaType.MUSIC: + matches = getattr(s, "get_music", lambda **_: [])( + title=getattr(mediainfo, "title", None), + artist=getattr(mediainfo, "artist", None), + album=getattr(mediainfo, "album", None), + ) + if matches: + return schemas.ExistMediaInfo( + type=MediaType.MUSIC, + server_type="zspace", + server=name, + itemid=matches[0].item_id, + ) + continue if mediainfo.type == MediaType.MOVIE: if itemid: movie = s.get_iteminfo(itemid) diff --git a/app/modules/zspace/zspace.py b/app/modules/zspace/zspace.py index cfbeeaecf..0044f17d6 100644 --- a/app/modules/zspace/zspace.py +++ b/app/modules/zspace/zspace.py @@ -245,6 +245,8 @@ class ZSpace: library_type = MediaType.MOVIE.value elif library.get("CollectionType") == "tvshows": library_type = MediaType.TV.value + elif library.get("CollectionType") in ("music", "musicvideos", "audio"): + library_type = MediaType.MUSIC.value else: library_type = MediaType.UNKNOWN.value image = self.__get_local_image_by_id(library.get("Id")) @@ -388,7 +390,9 @@ class ZSpace: return schemas.Statistic( movie_count=result.get("MovieCount") or 0, tv_count=result.get("SeriesCount") or 0, - episode_count=result.get("EpisodeCount") or 0 + episode_count=result.get("EpisodeCount") or 0, + music_count=result.get("MusicAlbumCount") or result.get("AlbumCount") + or result.get("SongCount") or 0, ) logger.debug("Items/Counts 未获取到返回数据，回退到按媒体库累计 TotalRecordCount") except Exception as e: @@ -550,6 +554,34 @@ class ZSpace: return None return [] + def get_music( + self, + title: Optional[str] = None, + artist: Optional[str] = None, + album: Optional[str] = None, + ) -> List[schemas.MediaServerItem]: + """按歌曲、艺术家或专辑名称查询极影视音乐条目。""" + if not self._host or not self._apikey: + return [] + query = " ".join(filter(None, [title, artist, album])).strip() + if not query: + return [] + url = f"{self._host}emby/Items" + params = { + "IncludeItemTypes": "MusicAlbum,Audio", + "Fields": "ProviderIds,OriginalTitle,ProductionYear,Path,ParentId", + "SearchTerm": query, + "Recursive": "true", + "Limit": 20, + } + try: + res = self.__request_utils().get_res(url, params=params) + items = res.json().get("Items", []) if res else [] + return [item for item in (self.__format_item_info(item) for item in items) if item] + except Exception as e: + logger.debug(f"查询极影视音乐出错：{e}") + return [] + def get_tv_episodes(self, item_id: Optional[str] = None, title: Optional[str] = None, diff --git a/app/schemas/dashboard.py b/app/schemas/dashboard.py index 2f9a01959..7fec0d496 100644 --- a/app/schemas/dashboard.py +++ b/app/schemas/dashboard.py @@ -14,6 +14,8 @@ class Statistic(BaseModel): tv_count: Optional[int] = 0 # 集数量 episode_count: Optional[int] = 0 + # 音乐数量 + music_count: Optional[int] = 0 # 用户数量 user_count: Optional[int] = 0 # 本月新增电影数量 diff --git a/app/schemas/mediaserver.py b/app/schemas/mediaserver.py index baafb5d7a..6598fe77d 100644 --- a/app/schemas/mediaserver.py +++ b/app/schemas/mediaserver.py @@ -10,11 +10,11 @@ class ExistMediaInfo(BaseModel): """ 媒体服务器存在媒体信息 """ - # 类型 电影、电视剧 + # 类型 电影、电视剧、音乐 type: Optional[MediaType] = None # 季 seasons: Optional[Dict[int, list]] = Field(default_factory=dict) - # 媒体服务器类型：plex、jellyfin、emby、zspace、trimemedia、ugreen + # 媒体服务器类型：plex、jellyfin、emby、zspace、trimemedia、ugreen、navidrome server_type: Optional[str] = None # 媒体服务器名称 server: Optional[str] = None diff --git a/app/schemas/music.py b/app/schemas/music.py index e59e5075a..c4977fbb3 100644 --- a/app/schemas/music.py +++ b/app/schemas/music.py @@ -33,13 +33,18 @@ class MusicInfo(BaseModel): """标准化音乐元数据信息。""" type: Literal["音乐"] = "音乐" + # 音乐实体类型：recording 单曲、album 专辑、artist 艺术家 + music_type: Literal["recording", "album", "artist"] = "recording" source: Optional[str] = None media_id: Optional[str] = None title: Optional[str] = None artists: list[str] = Field(default_factory=list) artist: Optional[str] = None + artist_ids: list[str] = Field(default_factory=list) album: Optional[str] = None album_artist: Optional[str] = None + album_id: Optional[str] = None + album_type: Optional[str] = None year: Optional[int] = None release_date: Optional[str] = None disc_number: Optional[int] = None @@ -51,6 +56,7 @@ class MusicInfo(BaseModel): lyrics: Optional[str] = None version: Optional[str] = None category: Optional[str] = "" + genres: list[str] = Field(default_factory=list) names: list[str] = Field(default_factory=list) detail_link: Optional[str] = None listen_count: Optional[int] = None @@ -63,6 +69,90 @@ class MusicInfo(BaseModel): vote_average: float = 0.0 +class MusicRelease(BaseModel): + """音乐专辑下的单个发行版本。""" + + media_id: Optional[str] = None + title: Optional[str] = None + date: Optional[str] = None + year: Optional[int] = None + country: Optional[str] = None + status: Optional[str] = None + packaging: Optional[str] = None + formats: list[str] = Field(default_factory=list) + track_count: Optional[int] = None + cover_url: Optional[str] = None + + +class MusicAlbumInfo(BaseModel): + """标准化音乐专辑信息。""" + + type: Literal["音乐"] = "音乐" + music_type: Literal["album"] = "album" + source: Optional[str] = None + media_id: Optional[str] = None + title: Optional[str] = None + artists: list[str] = Field(default_factory=list) + artist: Optional[str] = None + artist_ids: list[str] = Field(default_factory=list) + album: Optional[str] = None + album_type: Optional[str] = None + secondary_types: list[str] = Field(default_factory=list) + year: Optional[int] = None + release_date: Optional[str] = None + total_tracks: Optional[int] = None + duration: Optional[int] = None + cover_url: Optional[str] = None + genres: list[str] = Field(default_factory=list) + tags: list[str] = Field(default_factory=list) + category: Optional[str] = "" + rating: float = 0.0 + rating_votes: Optional[int] = None + detail_link: Optional[str] = None + tracks: list[MusicInfo] = Field(default_factory=list) + releases: list[MusicRelease] = Field(default_factory=list) + raw_data: dict[str, Any] = Field(default_factory=dict) + title_year: Optional[str] = None + poster_path: Optional[str] = None + backdrop_path: Optional[str] = None + mediaid_prefix: Optional[str] = None + overview: Optional[str] = None + vote_average: float = 0.0 + + +class MusicArtistInfo(BaseModel): + """标准化音乐艺术家信息。""" + + type: Literal["音乐"] = "音乐" + music_type: Literal["artist"] = "artist" + source: Optional[str] = None + media_id: Optional[str] = None + name: Optional[str] = None + title: Optional[str] = None + sort_name: Optional[str] = None + disambiguation: Optional[str] = None + artist_type: Optional[str] = None + gender: Optional[str] = None + country: Optional[str] = None + area: Optional[str] = None + begin_date: Optional[str] = None + end_date: Optional[str] = None + ended: bool = False + life_span: Optional[str] = None + genres: list[str] = Field(default_factory=list) + tags: list[str] = Field(default_factory=list) + aliases: list[str] = Field(default_factory=list) + relation: Optional[str] = None + image_url: Optional[str] = None + detail_link: Optional[str] = None + external_links: dict[str, str] = Field(default_factory=dict) + album_count: Optional[int] = None + raw_data: dict[str, Any] = Field(default_factory=dict) + poster_path: Optional[str] = None + mediaid_prefix: Optional[str] = None + overview: Optional[str] = None + + class MusicRecognizeRequest(BaseModel): """音乐元数据详情识别请求。""" diff --git a/app/schemas/system.py b/app/schemas/system.py index 1f3f1f0f7..e09d03cb6 100644 --- a/app/schemas/system.py +++ b/app/schemas/system.py @@ -29,7 +29,7 @@ class MediaServerConf(BaseModel): # 名称 name: Optional[str] = None - # 类型 emby/zspace/jellyfin/plex/trimemedia/ugreen + # 类型 emby/zspace/jellyfin/plex/trimemedia/ugreen/navidrome type: Optional[str] = None # 配置 config: Optional[dict] = Field(default_factory=dict) diff --git a/app/schemas/types.py b/app/schemas/types.py index 59f0ba3d3..cb39a300c 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -392,6 +392,8 @@ class MediaServerType(Enum): TrimeMedia = "TrimeMedia" # 绿联影视 Ugreen = "Ugreen" + # Navidrome 音乐服务器 + Navidrome = "Navidrome" # 识别器类型 diff --git a/tests/test_listenbrainz_module.py b/tests/test_listenbrainz_module.py index 63ff5dafa..d2ece7cd1 100644 --- a/tests/test_listenbrainz_module.py +++ b/tests/test_listenbrainz_module.py @@ -24,13 +24,13 @@ def test_recording_to_info_maps_listenbrainz_payload(): def test_music_chart_requests_requested_page(monkeypatch): - """音乐榜单模块应传递周期、偏移量和数量并过滤无身份记录。""" + """音乐榜单模块应传递实体、周期、偏移量和数量并过滤无身份记录。""" module = ListenBrainzModule() requested = {} - def fake_request(range_name, offset, count): + def fake_request(entity, range_name, offset, count): """记录榜单请求参数并返回一条有效录音。""" - requested.update(range_name=range_name, offset=offset, count=count) + requested.update(entity=entity, range_name=range_name, offset=offset, count=count) return { "payload": { "recordings": [ @@ -48,5 +48,100 @@ def test_music_chart_requests_requested_page(monkeypatch): results = module.music_chart(range_name="this_month", offset=30, count=30) - assert requested == {"range_name": "this_month", "offset": 30, "count": 30} + assert requested == { + "entity": "recordings", + "range_name": "this_month", + "offset": 30, + "count": 30, + } assert [item.media_id for item in results] == ["recording-1"] + + +def test_music_chart_supports_album_entity(monkeypatch): + """热门专辑榜单应请求官方 release-groups 接口并返回专辑实体。""" + module = ListenBrainzModule() + requested = {} + + def fake_request(entity, range_name, offset, count): + """记录请求实体并返回一条热门专辑。""" + requested.update(entity=entity, range_name=range_name) + return { + "payload": { + "release_groups": [ + { + "artist_name": "BTS", + "artist_mbids": ["artist-1"], + "listen_count": 999, + "release_group_mbid": "release-group-1", + "release_group_name": "ARIRANG", + } + ] + } + } + + monkeypatch.setattr(module, "_request_chart", fake_request) + + results = module.music_chart(range_name="week", offset=0, count=10, entity="album") + + assert requested == {"entity": "release-groups", "range_name": "week"} + assert [item.media_id for item in results] == ["release-group-1"] + assert results[0].music_type == "album" + assert results[0].album_id == "release-group-1" + assert results[0].artist_ids == ["artist-1"] + + +def test_music_chart_falls_back_to_supported_range(monkeypatch): + """非官方周期应回退到默认周期，避免请求被官方接口拒绝。""" + module = ListenBrainzModule() + requested = {} + + def fake_request(entity, range_name, offset, count): + """仅记录周期取值。""" + requested.update(range_name=range_name) + return {"payload": {"recordings": []}} + + monkeypatch.setattr(module, "_request_chart", fake_request) + module.music_chart(range_name="last_decade") + + assert requested == {"range_name": "this_month"} + + +def test_music_fresh_releases_pages_official_window(monkeypatch): + """新发行探索应按官方排序请求并在结果集上分页。""" + module = ListenBrainzModule() + requested = {} + + def fake_releases(days, sort, past, future): + """记录官方新发行请求参数并返回两条发行。""" + requested.update(days=days, sort=sort, past=past, future=future) + return [ + { + "artist_credit_name": "Artist A", + "artist_mbids": ["artist-1"], + "release_date": "2026-08-01", + "release_group_mbid": "release-group-1", + "release_group_primary_type": "Album", + "release_name": "First", + }, + { + "artist_credit_name": "Artist B", + "release_date": "2026-08-02", + "release_group_mbid": "release-group-2", + "release_name": "Second", + }, + ] + + monkeypatch.setattr(module, "_fresh_releases", fake_releases) + + results = module.music_fresh_releases( + days=200, + sort="unsupported", + past=True, + future=False, + offset=1, + count=1, + ) + + assert requested == {"days": 90, "sort": "release_date", "past": True, "future": False} + assert [item.media_id for item in results] == ["release-group-2"] + assert results[0].music_type == "album" diff --git a/tests/test_music_chain.py b/tests/test_music_chain.py index 3a2bde03f..d53f4c4a3 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, MusicMeta +from app.core.music import MusicAlbumInfo, MusicArtistInfo, MusicInfo, MusicMeta def test_parse_query_supports_artist_title_format(): @@ -171,3 +171,114 @@ def test_select_path_candidate_prefers_matching_audio_tags(): selected = MusicChain._select_path_candidate(meta, candidates, source="musicbrainz") assert selected is candidates[1] + + +def test_async_chart_forwards_album_entity(monkeypatch): + """热门专辑探索应把实体类型透传给 ListenBrainz 榜单模块。""" + chain = MusicChain() + requested = {} + + async def fake_async_run_module(method, **kwargs): + """记录榜单请求参数并返回一个专辑候选。""" + requested.update(method=method, **kwargs) + return [ + MusicInfo( + media_id="release-group-1", + source="musicbrainz", + music_type="album", + title="ARIRANG", + listen_count=10, + ) + ] + + monkeypatch.setattr(chain, "async_run_module", fake_async_run_module) + + import asyncio + + results = asyncio.run(chain.async_chart(range_name="week", page=3, count=20, entity="album")) + + assert requested["method"] == "music_chart" + assert requested["entity"] == "album" + assert requested["offset"] == 40 + assert results[0].music_type == "album" + + +def test_async_fresh_releases_keeps_official_order(monkeypatch): + """新发行探索应保留官方排序，只按封面条件过滤。""" + chain = MusicChain() + requested = {} + + async def fake_async_run_module(method, **kwargs): + """记录新发行请求参数并返回带封面与不带封面的候选。""" + requested.update(method=method, **kwargs) + return [ + MusicInfo(media_id="b", source="musicbrainz", music_type="album", title="B"), + MusicInfo( + media_id="a", + source="musicbrainz", + music_type="album", + title="A", + cover_url="https://coverartarchive.org/release/a/front-500", + ), + ] + + monkeypatch.setattr(chain, "async_run_module", fake_async_run_module) + + import asyncio + + results = asyncio.run( + chain.async_fresh_releases(days=30, sort="release_name", page=2, count=10, with_cover=True) + ) + + assert requested["method"] == "music_fresh_releases" + assert requested["offset"] == 10 + assert requested["sort"] == "release_name" + assert [item.title for item in results] == ["A"] + + +def test_async_artist_related_deduplicates_artists(monkeypatch): + """关联艺术家应按标准 ID 去重，避免同一成员重复出现。""" + chain = MusicChain() + + async def fake_async_run_module(method, **kwargs): + """返回重复的关联艺术家候选。""" + assert method == "music_artist_related" + return [ + MusicArtistInfo(source="musicbrainz", media_id="artist-1", name="Brian May"), + MusicArtistInfo(source="musicbrainz", media_id="artist-1", name="Brian May"), + MusicArtistInfo(source="musicbrainz", media_id="artist-2", name="John Deacon"), + ] + + monkeypatch.setattr(chain, "async_run_module", fake_async_run_module) + + import asyncio + + results = asyncio.run(chain.async_artist_related(source="musicbrainz", media_id="artist-0")) + + assert [item.media_id for item in results] == ["artist-1", "artist-2"] + + +def test_async_album_restores_dataclass_from_plugin_dict(monkeypatch): + """插件返回字典时专辑链应恢复为标准专辑对象。""" + chain = MusicChain() + + async def fake_async_run_module(method, **kwargs): + """模拟插件模块以字典形式返回专辑详情。""" + assert method == "music_album" + return MusicAlbumInfo( + source="musicbrainz", + media_id="release-group-1", + title="A Night at the Opera", + artists=["Queen"], + release_date="1975-11-21", + ).to_dict() + + monkeypatch.setattr(chain, "async_run_module", fake_async_run_module) + + import asyncio + + album = asyncio.run(chain.async_album(source="musicbrainz", media_id="release-group-1")) + + assert album is not None + assert album.year == 1975 + assert album.artists == ["Queen"] diff --git a/tests/test_music_endpoint.py b/tests/test_music_endpoint.py index b9ac23411..c54c150a6 100644 --- a/tests/test_music_endpoint.py +++ b/tests/test_music_endpoint.py @@ -5,18 +5,36 @@ import pytest from fastapi import HTTPException from app.api.apiv1 import api_router -from app.api.endpoints.music import explore_music, recognize_music, search_music -from app.core.music import MusicInfo +from app.api.endpoints.music import ( + explore_music, + music_album, + music_artist, + music_artist_albums, + music_artist_related, + recognize_music, + search_music, +) +from app.core.music import MusicAlbumInfo, MusicArtistInfo, MusicInfo, MusicRelease from app.schemas.music import MusicRecognizeRequest def test_music_routes_are_registered(): - """V1 API 应注册音乐搜索和详情识别路由。""" + """V1 API 应注册音乐搜索、详情识别、探索及艺术家专辑浏览路由。""" routes = {(route.path, tuple(route.methods or [])) for route in api_router.routes} assert any(path == "/music/search" and "GET" in methods for path, methods in routes) 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/artist/{artist_id}" and "GET" in methods for path, methods in routes) + assert any( + path == "/music/artist/{artist_id}/albums" and "GET" in methods + for path, methods in routes + ) + assert any( + path == "/music/artist/{artist_id}/related" and "GET" in methods + for path, methods in routes + ) def test_search_music_serializes_chain_results(): @@ -91,7 +109,7 @@ def test_recognize_music_returns_404_for_unknown_item(): def test_explore_music_forwards_filters_and_serializes_chart(): - """音乐探索接口应传递周期、排序、热度和封面筛选条件。""" + """音乐探索接口应传递实体、周期、排序、热度和封面筛选条件。""" chain = Mock() chain.async_chart = AsyncMock( return_value=[ @@ -126,4 +144,165 @@ def test_explore_music_forwards_filters_and_serializes_chart(): sort_by="listen_count.asc", min_listen_count=100, with_cover=True, + entity="recording", + ) + + +def test_explore_music_supports_official_fresh_release_mode(): + """新发行模式应按 ListenBrainz 官方排序和时间窗口请求探索数据。""" + chain = Mock() + chain.async_fresh_releases = AsyncMock( + return_value=[ + MusicInfo( + source="musicbrainz", + media_id="release-group-1", + music_type="album", + title="ARIRANG", + artists=["BTS"], + ) + ] + ) + + with patch("app.api.endpoints.music.MusicChain", return_value=chain): + result = asyncio.run( + explore_music( + page=1, + count=30, + mode="fresh", + sort="artist_credit_name", + days=30, + past=True, + future=False, + with_cover=True, + _=Mock(), + ) + ) + + assert result[0].music_type == "album" + chain.async_fresh_releases.assert_awaited_once_with( + days=30, + sort="artist_credit_name", + past=True, + future=False, + page=1, + count=30, + with_cover=True, + ) + + +def test_music_album_returns_tracks_and_releases(): + """专辑接口应返回专辑详情、曲目和发行版本。""" + chain = Mock() + chain.async_album = AsyncMock( + return_value=MusicAlbumInfo( + source="musicbrainz", + media_id="release-group-1", + title="A Night at the Opera", + artists=["Queen"], + artist_ids=["artist-1"], + album_type="Album", + release_date="1975-11-21", + tracks=[MusicInfo(source="musicbrainz", media_id="recording-1", title="Love of My Life")], + releases=[MusicRelease(media_id="release-1", title="A Night at the Opera", date="1975")], + ) + ) + + with patch("app.api.endpoints.music.MusicChain", return_value=chain): + result = asyncio.run(music_album(album_id="release-group-1", _=Mock())) + + assert result.music_type == "album" + assert result.year == 1975 + assert result.total_tracks == 1 + assert result.tracks[0].media_id == "recording-1" + assert result.releases[0].media_id == "release-1" + chain.async_album.assert_awaited_once_with(source="musicbrainz", media_id="release-group-1") + + +def test_music_album_returns_404_for_unknown_album(): + """专辑不存在时接口应返回 404。""" + chain = Mock() + chain.async_album = AsyncMock(return_value=None) + + with ( + patch("app.api.endpoints.music.MusicChain", return_value=chain), + pytest.raises(HTTPException) as error, + ): + asyncio.run(music_album(album_id="missing", _=Mock())) + + assert error.value.status_code == 404 + + +def test_music_artist_returns_detail(): + """艺术家接口应返回名称、类型和活跃时间。""" + chain = Mock() + chain.async_artist = AsyncMock( + return_value=MusicArtistInfo( + source="musicbrainz", + media_id="artist-1", + name="Queen", + artist_type="Group", + begin_date="1970-06-27", + ) + ) + + with patch("app.api.endpoints.music.MusicChain", return_value=chain): + result = asyncio.run(music_artist(artist_id="artist-1", _=Mock())) + + assert result.name == "Queen" + assert result.title == "Queen" + assert result.music_type == "artist" + chain.async_artist.assert_awaited_once_with(source="musicbrainz", media_id="artist-1") + + +def test_music_artist_albums_forwards_pagination_and_type(): + """艺术家专辑接口应传递分页和专辑类型筛选。""" + chain = Mock() + chain.async_artist_albums = AsyncMock( + return_value=[ + MusicInfo( + source="musicbrainz", + media_id="release-group-1", + music_type="album", + title="News of the World", + ) + ] + ) + + with patch("app.api.endpoints.music.MusicChain", return_value=chain): + result = asyncio.run( + music_artist_albums(artist_id="artist-1", page=2, count=10, album_type="ep", _=Mock()) + ) + + assert result[0].media_id == "release-group-1" + chain.async_artist_albums.assert_awaited_once_with( + source="musicbrainz", + media_id="artist-1", + page=2, + count=10, + album_type="ep", + ) + + +def test_music_artist_related_returns_relationship_text(): + """关联艺术家接口应返回关系说明，供详情页展示。""" + chain = Mock() + chain.async_artist_related = AsyncMock( + return_value=[ + MusicArtistInfo( + source="musicbrainz", + media_id="artist-2", + name="Freddie Mercury", + relation="member of band", + ) + ] + ) + + with patch("app.api.endpoints.music.MusicChain", return_value=chain): + result = asyncio.run(music_artist_related(artist_id="artist-1", count=5, _=Mock())) + + assert result[0].relation == "member of band" + chain.async_artist_related.assert_awaited_once_with( + source="musicbrainz", + media_id="artist-1", + count=5, ) diff --git a/tests/test_music_subscribe.py b/tests/test_music_subscribe.py index 4323a1167..566704a94 100644 --- a/tests/test_music_subscribe.py +++ b/tests/test_music_subscribe.py @@ -62,6 +62,7 @@ def test_build_subscribe_meta_returns_music_meta(): assert isinstance(meta, MusicMeta) assert meta.type == MediaType.MUSIC assert meta.media_id == "recording-1" + assert meta.original_name == "晴天" def test_music_subscribe_reuses_search_download_and_finish_flow(): diff --git a/tests/test_musicbrainz_module.py b/tests/test_musicbrainz_module.py index 83bf917c6..36dd8ade6 100644 --- a/tests/test_musicbrainz_module.py +++ b/tests/test_musicbrainz_module.py @@ -116,3 +116,255 @@ def test_recognize_music_fetches_recording_detail(monkeypatch): assert result is not None assert result.media_id == "recording-1" assert result.title == "晴天" + + +def test_recognize_music_falls_back_to_album(monkeypatch): + """单曲 ID 不存在时应按专辑再识别一次，保证专辑订阅可恢复目标。""" + module = MusicBrainzModule() + requested = [] + + def fake_request(path, params=None): + """单曲请求返回空，专辑请求返回最小 Release Group 数据。""" + requested.append(path) + if path.startswith("/recording/"): + return None + if path.startswith("/release-group/"): + return { + "id": "release-group-1", + "title": "A Night at the Opera", + "primary-type": "Album", + "first-release-date": "1975-11-21", + "artist-credit": [{"artist": {"id": "artist-1", "name": "Queen"}}], + "releases": [], + } + return None + + monkeypatch.setattr(module, "_request_json", fake_request) + monkeypatch.setattr(MusicBrainzModule, "_request_json", staticmethod(fake_request)) + + info = module.recognize_music("musicbrainz", "release-group-1") + + assert info is not None + assert info.music_type == "album" + assert info.album_id == "release-group-1" + assert info.artists == ["Queen"] + assert info.artist_ids == ["artist-1"] + assert requested[0].startswith("/recording/") + + +def test_music_album_builds_tracks_and_release_variants(monkeypatch): + """专辑详情应带上曲目列表、发行版本和 10 分制评分。""" + module = MusicBrainzModule() + + def fake_request(path, params=None): + """按路径分别返回 Release Group 与代表性 Release 数据。""" + if path == "/release-group/release-group-1": + return { + "id": "release-group-1", + "title": "A Night at the Opera", + "primary-type": "Album", + "secondary-types": ["Live"], + "first-release-date": "1975-11-21", + "rating": {"value": 4.25, "votes-count": 44}, + "genres": [{"name": "rock", "count": 13}, {"name": "art rock", "count": 4}], + "tags": [{"name": "british", "count": 2}], + "artist-credit": [{"artist": {"id": "artist-1", "name": "Queen"}}], + "releases": [ + { + "id": "release-early", + "title": "A Night at the Opera", + "status": "Official", + "date": "1975-11-21", + "country": "GB", + "packaging": "Gatefold Cover", + "media": [{"format": "12\" Vinyl", "track-count": 2}], + }, + { + "id": "release-late", + "title": "A Night at the Opera", + "status": "Official", + "date": "1991", + "media": [{"format": "CD", "track-count": 2}], + }, + ], + } + if path == "/release/release-early": + return { + "media": [ + { + "position": 1, + "track-count": 2, + "tracks": [ + { + "position": 1, + "title": "Death on Two Legs", + "length": 223733, + "recording": {"id": "recording-1", "title": "Death on Two Legs"}, + }, + { + "position": 2, + "title": "Lazing on a Sunday Afternoon", + "recording": {"id": "recording-2", "length": 68000}, + }, + ], + } + ] + } + return None + + monkeypatch.setattr(MusicBrainzModule, "_request_json", staticmethod(fake_request)) + + album = module.music_album("musicbrainz", "release-group-1") + + assert album is not None + assert album.category == "Album / Live" + assert album.rating == 8.5 + assert album.genres[:2] == ["rock", "art rock"] + assert [track.media_id for track in album.tracks] == ["recording-1", "recording-2"] + assert album.tracks[0].duration == 224 + assert album.tracks[0].track_number == 1 + assert album.tracks[0].disc_number == 1 + assert album.tracks[0].album_id == "release-group-1" + assert album.tracks[1].title == "Lazing on a Sunday Afternoon" + assert [release.media_id for release in album.releases] == ["release-early", "release-late"] + assert album.releases[0].formats == ['12" Vinyl'] + assert album.track_count == 2 + + +def test_music_artist_maps_profile_links_and_image(monkeypatch): + """艺术家详情应整理活跃时间、别名、外链并把维基共享页转为图片直链。""" + module = MusicBrainzModule() + monkeypatch.setattr( + MusicBrainzModule, + "_request_json", + staticmethod( + lambda path, params=None: { + "id": "artist-1", + "name": "Queen", + "sort-name": "Queen", + "type": "Group", + "disambiguation": "UK rock group", + "country": "GB", + "area": {"name": "United Kingdom"}, + "life-span": {"begin": "1970-06-27", "ended": True}, + "genres": [{"name": "rock", "count": 20}, {"name": "glam rock", "count": 9}], + "tags": [{"name": "british", "count": 15}], + "aliases": [{"name": "皇后乐队", "count": 0}], + "relations": [ + { + "type": "image", + "target-type": "url", + "url": {"resource": "https://commons.wikimedia.org/wiki/File:Queen.jpg"}, + }, + { + "type": "official homepage", + "target-type": "url", + "url": {"resource": "http://www.queenonline.com/"}, + }, + { + "type": "wikidata", + "target-type": "url", + "url": {"resource": "https://www.wikidata.org/wiki/Q15862"}, + }, + { + "type": "creative commons licensed download", + "target-type": "url", + "url": {"resource": "https://example.com/ignored"}, + }, + ], + } + ), + ) + + artist = module.music_artist("musicbrainz", "artist-1") + + assert artist is not None + assert artist.artist_type == "Group" + assert artist.area == "United Kingdom" + assert artist.life_span == "1970-06-27" + assert artist.genres == ["rock", "glam rock"] + assert artist.aliases == ["皇后乐队"] + assert artist.image_url == ( + "https://commons.wikimedia.org/wiki/Special:FilePath/Queen.jpg?width=500" + ) + assert set(artist.external_links) == {"official homepage", "wikidata"} + + +def test_music_artist_albums_sorts_page_by_release_date(monkeypatch): + """艺术家专辑列表应按发行日期倒序，并带上专辑类型筛选参数。""" + module = MusicBrainzModule() + requested = {} + + def fake_request(path, params=None): + """记录浏览请求参数并返回两个乱序专辑。""" + requested.update(path=path, params=params) + return { + "release-groups": [ + { + "id": "release-group-old", + "title": "Queen", + "primary-type": "Album", + "first-release-date": "1973-07-13", + "artist-credit": [{"artist": {"id": "artist-1", "name": "Queen"}}], + }, + { + "id": "release-group-new", + "title": "News of the World", + "primary-type": "Album", + "first-release-date": "1977-10-28", + "artist-credit": [{"artist": {"id": "artist-1", "name": "Queen"}}], + }, + ] + } + + monkeypatch.setattr(MusicBrainzModule, "_request_json", staticmethod(fake_request)) + + albums = module.music_artist_albums("musicbrainz", "artist-1", page=2, count=10, album_type="album") + + assert requested["path"] == "/release-group" + assert requested["params"]["artist"] == "artist-1" + assert requested["params"]["offset"] == 10 + assert requested["params"]["type"] == "album" + assert [album.media_id for album in albums] == ["release-group-new", "release-group-old"] + assert albums[0].music_type == "album" + + +def test_music_artist_related_prefers_meaningful_relations(monkeypatch): + """关联艺术家应优先返回成员与子团体关系，致敬乐队排在最后。""" + module = MusicBrainzModule() + monkeypatch.setattr( + MusicBrainzModule, + "_request_json", + staticmethod( + lambda path, params=None: { + "relations": [ + { + "type": "tribute", + "target-type": "artist", + "artist": {"id": "artist-tribute", "name": "Queen Tribute"}, + }, + { + "type": "member of band", + "target-type": "artist", + "artist": {"id": "artist-member", "name": "Brian May"}, + }, + { + "type": "member of band", + "target-type": "artist", + "artist": {"id": "artist-member", "name": "Brian May"}, + }, + { + "type": "allmusic", + "target-type": "url", + "url": {"resource": "https://example.com"}, + }, + ] + } + ), + ) + + related = module.music_artist_related("musicbrainz", "artist-1", count=5) + + assert [item.media_id for item in related] == ["artist-member", "artist-tribute"] + assert related[0].relation == "member of band" + assert related[0].music_type == "artist" diff --git a/tests/test_navidrome_module.py b/tests/test_navidrome_module.py new file mode 100644 index 000000000..8d4ebc972 --- /dev/null +++ b/tests/test_navidrome_module.py @@ -0,0 +1,32 @@ +"""Navidrome 媒体服务器模块接入测试。""" +from app.core.module import ModuleManager +from app.modules.navidrome import NavidromeModule +from app.schemas.types import MediaServerType, ModuleType + + +def test_navidrome_module_declares_media_server_identity(): + """Navidrome 应以媒体服务器身份注册，供统一媒体服务器链调用。""" + assert NavidromeModule.get_name() == "Navidrome" + assert NavidromeModule.get_type() == ModuleType.MediaServer + assert NavidromeModule.get_subtype() == MediaServerType.Navidrome + + +def test_navidrome_module_has_no_system_switch(): + """Navidrome 由服务配置控制启用，不能返回无效的系统开关名。""" + assert NavidromeModule().init_setting() is None + + +def test_navidrome_module_is_loaded_by_module_manager(): + """模块管理器应能加载 Navidrome，否则媒体服务器列表里不会出现该类型。""" + assert "NavidromeModule" in ModuleManager()._running_modules + + +def test_navidrome_module_ignores_non_music_media(): + """Navidrome 只管理音乐，影视存在性检查应交给其它媒体服务器。""" + from app.core.context import MediaInfo + from app.schemas.types import MediaType + + mediainfo = MediaInfo() + mediainfo.type = MediaType.MOVIE + + assert NavidromeModule().media_exists(mediainfo) is None