From 8982adf88d7fadd40e6f81d5dd42224b3eb63bae Mon Sep 17 00:00:00 2001 From: Mikhail Nevskiy <139659391+trudenboy@users.noreply.github.com> Date: Tue, 10 Feb 2026 16:42:45 +0300 Subject: [PATCH] Add KION Music (MTS Music) provider (#3100) MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * feat(kion_music): add KION Music (MTS Music) provider Add a new music provider for KION Music (music.mts.ru), the MTS telecom music streaming service. The provider adapts the Yandex Music API client (yandex-music library) to work with KION's API endpoint at music.mts.ru/ya_api. Features: - Search (tracks, albums, artists, playlists) - Library access (liked artists, albums, tracks; user playlists) - Add/remove library items (like/unlike) - Streaming with quality selection (High 320kbps / Lossless FLAC) - Lossless streaming via get-file-info API with flac-mp4 codec support - Playlist track loading with pagination fix and batch fetching - Retry on NetworkError to handle transient failures The provider reuses the same yandex-music==2.2.0 dependency already present for the Yandex Music provider — no new dependencies required. Co-authored-by: Cursor * chore(kion_music): update documentation URL in manifest Co-authored-by: Cursor * fix(kion_music): fix missing album cover art in library - In parse_track(), use parse_album() instead of ItemMapping so albums get full metadata (including cover_uri) when tracks are synced - In get_liked_albums(), fetch full album details via client.albums() in batches of 50 (likes endpoint returns minimal data without cover_uri) - Remove unused noqa in provider.py (ruff) Co-authored-by: Cursor * Address PR #3100 review: fix base_url, performance, redundant condition, docstrings; add tests - Pass KION_BASE_URL to ClientAsync in connect() so API calls go to music.mts.ru - Use set(batch) for O(1) lookups in get_liked_albums() fallback path - Remove redundant `track_count > 0` check already guaranteed by outer condition - Replace "Yandex Music" with "KION Music" in docstrings/comments - Add full test suite: 5 api_client unit tests, 18 parser tests, 11 integration tests Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: Михаил Невский Co-authored-by: Cursor Co-authored-by: Claude Opus 4.6 --- .../providers/kion_music/__init__.py | 96 +++ .../providers/kion_music/api_client.py | 511 +++++++++++++++ .../providers/kion_music/constants.py | 32 + music_assistant/providers/kion_music/icon.svg | 12 + .../providers/kion_music/icon_monochrome.svg | 6 + .../providers/kion_music/manifest.json | 11 + .../providers/kion_music/parsers.py | 366 +++++++++++ .../providers/kion_music/provider.py | 475 ++++++++++++++ .../providers/kion_music/streaming.py | 179 ++++++ tests/providers/kion_music/__init__.py | 1 + .../__snapshots__/test_parsers.ambr | 592 ++++++++++++++++++ tests/providers/kion_music/conftest.py | 96 +++ .../kion_music/fixtures/albums/minimal.json | 8 + .../kion_music/fixtures/artists/minimal.json | 4 + .../fixtures/artists/with_cover.json | 8 + .../fixtures/playlists/minimal.json | 9 + .../fixtures/playlists/other_user.json | 10 + .../kion_music/fixtures/tracks/minimal.json | 8 + .../tracks/with_artist_and_album.json | 19 + tests/providers/kion_music/test_api_client.py | 108 ++++ .../providers/kion_music/test_integration.py | 354 +++++++++++ tests/providers/kion_music/test_parsers.py | 247 ++++++++ 22 files changed, 3152 insertions(+) create mode 100644 music_assistant/providers/kion_music/__init__.py create mode 100644 music_assistant/providers/kion_music/api_client.py create mode 100644 music_assistant/providers/kion_music/constants.py create mode 100644 music_assistant/providers/kion_music/icon.svg create mode 100644 music_assistant/providers/kion_music/icon_monochrome.svg create mode 100644 music_assistant/providers/kion_music/manifest.json create mode 100644 music_assistant/providers/kion_music/parsers.py create mode 100644 music_assistant/providers/kion_music/provider.py create mode 100644 music_assistant/providers/kion_music/streaming.py create mode 100644 tests/providers/kion_music/__init__.py create mode 100644 tests/providers/kion_music/__snapshots__/test_parsers.ambr create mode 100644 tests/providers/kion_music/conftest.py create mode 100644 tests/providers/kion_music/fixtures/albums/minimal.json create mode 100644 tests/providers/kion_music/fixtures/artists/minimal.json create mode 100644 tests/providers/kion_music/fixtures/artists/with_cover.json create mode 100644 tests/providers/kion_music/fixtures/playlists/minimal.json create mode 100644 tests/providers/kion_music/fixtures/playlists/other_user.json create mode 100644 tests/providers/kion_music/fixtures/tracks/minimal.json create mode 100644 tests/providers/kion_music/fixtures/tracks/with_artist_and_album.json create mode 100644 tests/providers/kion_music/test_api_client.py create mode 100644 tests/providers/kion_music/test_integration.py create mode 100644 tests/providers/kion_music/test_parsers.py diff --git a/music_assistant/providers/kion_music/__init__.py b/music_assistant/providers/kion_music/__init__.py new file mode 100644 index 00000000..b8e0abeb --- /dev/null +++ b/music_assistant/providers/kion_music/__init__.py @@ -0,0 +1,96 @@ +"""KION Music provider support for Music Assistant.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType +from music_assistant_models.enums import ConfigEntryType, ProviderFeature + +from .constants import ( + CONF_ACTION_CLEAR_AUTH, + CONF_QUALITY, + CONF_TOKEN, + QUALITY_HIGH, + QUALITY_LOSSLESS, +) +from .provider import KionMusicProvider + +if TYPE_CHECKING: + from music_assistant_models.config_entries import ProviderConfig + from music_assistant_models.provider import ProviderManifest + + from music_assistant.mass import MusicAssistant + from music_assistant.models import ProviderInstanceType + + +SUPPORTED_FEATURES = { + ProviderFeature.LIBRARY_ARTISTS, + ProviderFeature.LIBRARY_ALBUMS, + ProviderFeature.LIBRARY_TRACKS, + ProviderFeature.LIBRARY_PLAYLISTS, + ProviderFeature.ARTIST_ALBUMS, + ProviderFeature.ARTIST_TOPTRACKS, + ProviderFeature.SEARCH, + ProviderFeature.LIBRARY_ARTISTS_EDIT, + ProviderFeature.LIBRARY_ALBUMS_EDIT, + ProviderFeature.LIBRARY_TRACKS_EDIT, + ProviderFeature.BROWSE, +} + + +async def setup( + mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig +) -> ProviderInstanceType: + """Initialize provider(instance) with given configuration.""" + return KionMusicProvider(mass, manifest, config, SUPPORTED_FEATURES) + + +async def get_config_entries( + mass: MusicAssistant, # noqa: ARG001 + instance_id: str | None = None, # noqa: ARG001 + action: str | None = None, + values: dict[str, ConfigValueType] | None = None, +) -> tuple[ConfigEntry, ...]: + """Return Config entries to setup this provider.""" + if values is None: + values = {} + + # Handle clear auth action + if action == CONF_ACTION_CLEAR_AUTH: + values[CONF_TOKEN] = None + + # Check if user is authenticated + is_authenticated = bool(values.get(CONF_TOKEN)) + + return ( + ConfigEntry( + key=CONF_TOKEN, + type=ConfigEntryType.SECURE_STRING, + label="KION Music Token", + description="Enter your KION Music OAuth token. " + "See the documentation for how to obtain it.", + required=True, + hidden=is_authenticated, + value=cast("str", values.get(CONF_TOKEN)) if values else None, + ), + ConfigEntry( + key=CONF_ACTION_CLEAR_AUTH, + type=ConfigEntryType.ACTION, + label="Reset authentication", + description="Clear the current authentication details.", + action=CONF_ACTION_CLEAR_AUTH, + hidden=not is_authenticated, + ), + ConfigEntry( + key=CONF_QUALITY, + type=ConfigEntryType.STRING, + label="Audio quality", + description="Select preferred audio quality.", + options=[ + ConfigValueOption("High (320 kbps)", QUALITY_HIGH), + ConfigValueOption("Lossless (FLAC)", QUALITY_LOSSLESS), + ], + default_value=QUALITY_HIGH, + ), + ) diff --git a/music_assistant/providers/kion_music/api_client.py b/music_assistant/providers/kion_music/api_client.py new file mode 100644 index 00000000..a6ba0c98 --- /dev/null +++ b/music_assistant/providers/kion_music/api_client.py @@ -0,0 +1,511 @@ +"""API client wrapper for KION Music (MTS Music).""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, cast + +from music_assistant_models.errors import ( + LoginFailed, + ProviderUnavailableError, + ResourceTemporarilyUnavailable, +) +from yandex_music import Album as YandexAlbum +from yandex_music import Artist as YandexArtist +from yandex_music import ClientAsync, Search, TrackShort +from yandex_music import Playlist as YandexPlaylist +from yandex_music import Track as YandexTrack +from yandex_music.exceptions import BadRequestError, NetworkError, UnauthorizedError +from yandex_music.utils.sign_request import get_sign_request + +if TYPE_CHECKING: + from yandex_music import DownloadInfo + +from .constants import DEFAULT_LIMIT + +# get-file-info with quality=lossless returns FLAC; default /tracks/.../download-info often does not +# Prefer flac-mp4/aac-mp4 +GET_FILE_INFO_CODECS = "flac-mp4,flac,aac-mp4,aac,he-aac,mp3,he-aac-mp4" +# get-file-info: same host as library (all requests go through one API) +KION_BASE_URL = "https://music.mts.ru/ya_api" + +LOGGER = logging.getLogger(__name__) + + +class KionMusicClient: + """Wrapper around yandex-music-api ClientAsync for KION Music.""" + + def __init__(self, token: str) -> None: + """Initialize the KION Music client. + + :param token: KION Music OAuth token. + """ + self._token = token + self._client: ClientAsync | None = None + self._user_id: int | None = None + + @property + def user_id(self) -> int: + """Return the user ID.""" + if self._user_id is None: + raise ProviderUnavailableError("Client not initialized, call connect() first") + return self._user_id + + async def connect(self) -> bool: + """Initialize the client and verify token validity. + + :return: True if connection was successful. + :raises LoginFailed: If the token is invalid. + """ + try: + self._client = await ClientAsync(self._token, base_url=KION_BASE_URL).init() + if self._client.me is None or self._client.me.account is None: + raise LoginFailed("Failed to get account info") + self._user_id = self._client.me.account.uid + LOGGER.debug("Connected to KION Music as user %s", self._user_id) + return True + except UnauthorizedError as err: + raise LoginFailed("Invalid KION Music token") from err + except NetworkError as err: + msg = "Network error connecting to KION Music" + raise ResourceTemporarilyUnavailable(msg) from err + + async def disconnect(self) -> None: + """Disconnect the client.""" + self._client = None + self._user_id = None + + def _ensure_connected(self) -> ClientAsync: + """Ensure the client is connected and return it.""" + if self._client is None: + raise ProviderUnavailableError("Client not connected, call connect() first") + return self._client + + # Library methods + + async def get_liked_tracks(self) -> list[TrackShort]: + """Get user's liked tracks. + + :return: List of liked track objects. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_tracks() + if result is None: + return [] + return result.tracks or [] + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching liked tracks: %s", err) + raise ResourceTemporarilyUnavailable("Failed to fetch liked tracks") from err + + async def get_liked_albums(self) -> list[YandexAlbum]: + """Get user's liked albums with full details (including cover art). + + The users_likes_albums endpoint returns minimal album data without + cover_uri, so we fetch full album details in batches afterwards. + + :return: List of liked album objects with full details. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_albums() + if result is None: + return [] + album_ids = [ + str(like.album.id) for like in result if like.album is not None and like.album.id + ] + if not album_ids: + return [] + # Fetch full album details in batches to get cover_uri and other metadata + batch_size = 50 + full_albums: list[YandexAlbum] = [] + for i in range(0, len(album_ids), batch_size): + batch = album_ids[i : i + batch_size] + try: + batch_result = await client.albums(batch) + if batch_result: + full_albums.extend(batch_result) + except (BadRequestError, NetworkError) as batch_err: + LOGGER.warning("Error fetching album details batch: %s", batch_err) + # Fall back to minimal data for this batch + batch_set = set(batch) + for like in result: + if ( + like.album is not None + and like.album.id + and str(like.album.id) in batch_set + ): + full_albums.append(like.album) + return full_albums + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching liked albums: %s", err) + raise ResourceTemporarilyUnavailable("Failed to fetch liked albums") from err + + async def get_liked_artists(self) -> list[YandexArtist]: + """Get user's liked artists. + + :return: List of liked artist objects. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_artists() + if result is None: + return [] + return [like.artist for like in result if like.artist is not None] + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching liked artists: %s", err) + raise ResourceTemporarilyUnavailable("Failed to fetch liked artists") from err + + async def get_user_playlists(self) -> list[YandexPlaylist]: + """Get user's playlists. + + :return: List of playlist objects. + """ + client = self._ensure_connected() + try: + result = await client.users_playlists_list() + if result is None: + return [] + return list(result) + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching playlists: %s", err) + raise ResourceTemporarilyUnavailable("Failed to fetch playlists") from err + + # Search + + async def search( + self, + query: str, + search_type: str = "all", + limit: int = DEFAULT_LIMIT, + ) -> Search | None: + """Search for tracks, albums, artists, or playlists. + + :param query: Search query string. + :param search_type: Type of search ('all', 'track', 'album', 'artist', 'playlist'). + :param limit: Maximum number of results per type. + :return: Search results object. + """ + client = self._ensure_connected() + try: + return await client.search(query, type_=search_type, page=0, nocorrect=False) + except (BadRequestError, NetworkError) as err: + LOGGER.error("Search error: %s", err) + raise ResourceTemporarilyUnavailable("Search failed") from err + + # Get single items + + async def get_track(self, track_id: str) -> YandexTrack | None: + """Get a single track by ID. + + :param track_id: Track ID. + :return: Track object or None if not found. + """ + client = self._ensure_connected() + try: + tracks = await client.tracks([track_id]) + return tracks[0] if tracks else None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching track %s: %s", track_id, err) + return None + + async def get_tracks(self, track_ids: list[str]) -> list[YandexTrack]: + """Get multiple tracks by IDs. + + :param track_ids: List of track IDs. + :return: List of track objects. + :raises ResourceTemporarilyUnavailable: On network errors after retry. + """ + client = self._ensure_connected() + try: + result = await client.tracks(track_ids) + return result or [] + except NetworkError as err: + # Retry once on network errors (timeout, disconnect, etc.) + LOGGER.warning("Network error fetching tracks, retrying once: %s", err) + try: + result = await client.tracks(track_ids) + return result or [] + except NetworkError as retry_err: + LOGGER.error("Error fetching tracks (retry failed): %s", retry_err) + raise ResourceTemporarilyUnavailable("Failed to fetch tracks") from retry_err + except BadRequestError as err: + LOGGER.error("Error fetching tracks: %s", err) + return [] + + async def get_album(self, album_id: str) -> YandexAlbum | None: + """Get a single album by ID. + + :param album_id: Album ID. + :return: Album object or None if not found. + """ + client = self._ensure_connected() + try: + albums = await client.albums([album_id]) + return albums[0] if albums else None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching album %s: %s", album_id, err) + return None + + async def get_album_with_tracks(self, album_id: str) -> YandexAlbum | None: + """Get an album with its tracks. + + Uses the same semantics as the web client: albums/{id}/with-tracks + with resumeStream, richTracks, withListeningFinished when the library + passes them through. + + :param album_id: Album ID. + :return: Album object with tracks or None if not found. + """ + client = self._ensure_connected() + try: + return await client.albums_with_tracks( + album_id, + resumeStream=True, + richTracks=True, + withListeningFinished=True, + ) + except TypeError: + # Older yandex-music may not accept these kwargs + return await client.albums_with_tracks(album_id) + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching album with tracks %s: %s", album_id, err) + return None + + async def get_artist(self, artist_id: str) -> YandexArtist | None: + """Get a single artist by ID. + + :param artist_id: Artist ID. + :return: Artist object or None if not found. + """ + client = self._ensure_connected() + try: + artists = await client.artists([artist_id]) + return artists[0] if artists else None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching artist %s: %s", artist_id, err) + return None + + async def get_artist_albums( + self, artist_id: str, limit: int = DEFAULT_LIMIT + ) -> list[YandexAlbum]: + """Get artist's albums. + + :param artist_id: Artist ID. + :param limit: Maximum number of albums. + :return: List of album objects. + """ + client = self._ensure_connected() + try: + result = await client.artists_direct_albums(artist_id, page=0, page_size=limit) + if result is None: + return [] + return result.albums or [] + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching artist albums %s: %s", artist_id, err) + return [] + + async def get_artist_tracks( + self, artist_id: str, limit: int = DEFAULT_LIMIT + ) -> list[YandexTrack]: + """Get artist's top tracks. + + :param artist_id: Artist ID. + :param limit: Maximum number of tracks. + :return: List of track objects. + """ + client = self._ensure_connected() + try: + result = await client.artists_tracks(artist_id, page=0, page_size=limit) + if result is None: + return [] + return result.tracks or [] + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching artist tracks %s: %s", artist_id, err) + return [] + + async def get_playlist(self, user_id: str, playlist_id: str) -> YandexPlaylist | None: + """Get a playlist by ID. + + :param user_id: User ID (owner of the playlist). + :param playlist_id: Playlist ID (kind). + :return: Playlist object or None if not found. + """ + client = self._ensure_connected() + try: + result = await client.users_playlists(kind=int(playlist_id), user_id=user_id) + if isinstance(result, list): + return result[0] if result else None + return result + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching playlist %s/%s: %s", user_id, playlist_id, err) + return None + + # Streaming + + async def get_track_download_info( + self, track_id: str, get_direct_links: bool = True + ) -> list[DownloadInfo]: + """Get download info for a track. + + :param track_id: Track ID. + :param get_direct_links: Whether to get direct download links. + :return: List of download info objects. + """ + client = self._ensure_connected() + try: + result = await client.tracks_download_info(track_id, get_direct_links=get_direct_links) + return result or [] + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error fetching download info for track %s: %s", track_id, err) + return [] + + async def get_track_file_info_lossless(self, track_id: str) -> dict[str, Any] | None: + """Request lossless stream via get-file-info (quality=lossless). + + The /tracks/{id}/download-info endpoint often returns only MP3; get-file-info + with quality=lossless and codecs=flac,... returns FLAC when available. + + :param track_id: Track ID. + :return: Parsed downloadInfo dict (url, codec, urls, ...) or None on error. + """ + client = self._ensure_connected() + sign = get_sign_request(track_id) + base_params = { + "ts": sign.timestamp, + "trackId": track_id, + "quality": "lossless", + "codecs": GET_FILE_INFO_CODECS, + "sign": sign.value, + } + + def _parse_file_info_result(raw: dict[str, Any] | None) -> dict[str, Any] | None: + if not raw or not isinstance(raw, dict): + return None + download_info = raw.get("download_info") + if not download_info or not download_info.get("url"): + return None + return cast("dict[str, Any]", download_info) + + url = f"{KION_BASE_URL}/get-file-info" + params_encraw = {**base_params, "transports": "encraw"} + try: + result = await client._request.get(url, params=params_encraw) + return _parse_file_info_result(result) + except (BadRequestError, NetworkError) as err: + LOGGER.debug( + "get-file-info lossless for track %s: %s %s", + track_id, + type(err).__name__, + getattr(err, "message", str(err)) or repr(err), + ) + return None + except UnauthorizedError as err: + LOGGER.debug( + "get-file-info lossless for track %s (transports=encraw): %s %s", + track_id, + type(err).__name__, + getattr(err, "message", str(err)) or repr(err), + ) + LOGGER.debug( + "If you have KION Music Plus and this track has lossless, " + "try a token from the web client (music.mts.ru)." + ) + params_raw = {**base_params, "transports": "raw"} + try: + result = await client._request.get(url, params=params_raw) + return _parse_file_info_result(result) + except (BadRequestError, NetworkError, UnauthorizedError) as retry_err: + LOGGER.debug( + "get-file-info lossless for track %s (transports=raw): %s %s", + track_id, + type(retry_err).__name__, + getattr(retry_err, "message", str(retry_err)) or repr(retry_err), + ) + return None + + # Library modifications + + async def like_track(self, track_id: str) -> bool: + """Add a track to liked tracks. + + :param track_id: Track ID to like. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_tracks_add(track_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error liking track %s: %s", track_id, err) + return False + + async def unlike_track(self, track_id: str) -> bool: + """Remove a track from liked tracks. + + :param track_id: Track ID to unlike. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_tracks_remove(track_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error unliking track %s: %s", track_id, err) + return False + + async def like_album(self, album_id: str) -> bool: + """Add an album to liked albums. + + :param album_id: Album ID to like. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_albums_add(album_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error liking album %s: %s", album_id, err) + return False + + async def unlike_album(self, album_id: str) -> bool: + """Remove an album from liked albums. + + :param album_id: Album ID to unlike. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_albums_remove(album_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error unliking album %s: %s", album_id, err) + return False + + async def like_artist(self, artist_id: str) -> bool: + """Add an artist to liked artists. + + :param artist_id: Artist ID to like. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_artists_add(artist_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error liking artist %s: %s", artist_id, err) + return False + + async def unlike_artist(self, artist_id: str) -> bool: + """Remove an artist from liked artists. + + :param artist_id: Artist ID to unlike. + :return: True if successful. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_artists_remove(artist_id) + return result is not None + except (BadRequestError, NetworkError) as err: + LOGGER.error("Error unliking artist %s: %s", artist_id, err) + return False diff --git a/music_assistant/providers/kion_music/constants.py b/music_assistant/providers/kion_music/constants.py new file mode 100644 index 00000000..39e51f0b --- /dev/null +++ b/music_assistant/providers/kion_music/constants.py @@ -0,0 +1,32 @@ +"""Constants for the KION Music provider.""" + +from __future__ import annotations + +from typing import Final + +# Configuration Keys +CONF_TOKEN = "token" +CONF_QUALITY = "quality" + +# Actions +CONF_ACTION_AUTH = "auth" +CONF_ACTION_CLEAR_AUTH = "clear_auth" + +# Labels +LABEL_TOKEN = "token_label" +LABEL_AUTH_INSTRUCTIONS = "auth_instructions_label" + +# API defaults +DEFAULT_LIMIT: Final[int] = 50 + +# Quality options +QUALITY_HIGH = "high" +QUALITY_LOSSLESS = "lossless" + +# Image sizes +IMAGE_SIZE_SMALL = "200x200" +IMAGE_SIZE_MEDIUM = "400x400" +IMAGE_SIZE_LARGE = "1000x1000" + +# ID separators +PLAYLIST_ID_SPLITTER: Final[str] = ":" diff --git a/music_assistant/providers/kion_music/icon.svg b/music_assistant/providers/kion_music/icon.svg new file mode 100644 index 00000000..f2031622 --- /dev/null +++ b/music_assistant/providers/kion_music/icon.svg @@ -0,0 +1,12 @@ + + + + + + + + + + + + diff --git a/music_assistant/providers/kion_music/icon_monochrome.svg b/music_assistant/providers/kion_music/icon_monochrome.svg new file mode 100644 index 00000000..3f37201e --- /dev/null +++ b/music_assistant/providers/kion_music/icon_monochrome.svg @@ -0,0 +1,6 @@ + + + + + + diff --git a/music_assistant/providers/kion_music/manifest.json b/music_assistant/providers/kion_music/manifest.json new file mode 100644 index 00000000..9b5cdc2d --- /dev/null +++ b/music_assistant/providers/kion_music/manifest.json @@ -0,0 +1,11 @@ +{ + "type": "music", + "domain": "kion_music", + "stage": "beta", + "name": "KION Music", + "description": "Stream music from KION Music (MTS) service.", + "codeowners": ["@TrudenBoy"], + "documentation": "https://music-assistant.io/music-providers/kion-music/", + "requirements": ["yandex-music==2.2.0"], + "multi_instance": true +} diff --git a/music_assistant/providers/kion_music/parsers.py b/music_assistant/providers/kion_music/parsers.py new file mode 100644 index 00000000..8ac52505 --- /dev/null +++ b/music_assistant/providers/kion_music/parsers.py @@ -0,0 +1,366 @@ +"""Parsers for KION Music API responses.""" + +from __future__ import annotations + +from contextlib import suppress +from datetime import datetime +from typing import TYPE_CHECKING + +from music_assistant_models.enums import ( + AlbumType, + ContentType, + ImageType, +) +from music_assistant_models.media_items import ( + Album, + Artist, + AudioFormat, + MediaItemImage, + Playlist, + ProviderMapping, + Track, + UniqueList, +) + +from music_assistant.helpers.util import parse_title_and_version + +from .constants import IMAGE_SIZE_LARGE + +if TYPE_CHECKING: + from yandex_music import Album as YandexAlbum + from yandex_music import Artist as YandexArtist + from yandex_music import Playlist as YandexPlaylist + from yandex_music import Track as YandexTrack + + from .provider import KionMusicProvider + + +def _get_content_type(provider: KionMusicProvider) -> ContentType: + """Get content type based on provider quality setting. + + :param provider: The KION Music provider instance. + :return: ContentType.UNKNOWN as actual codec is determined at stream time. + """ + # Actual codec is determined when getting stream details + # Suppress unused argument warning + _ = provider + return ContentType.UNKNOWN + + +def _get_image_url(cover_uri: str | None, size: str = IMAGE_SIZE_LARGE) -> str | None: + """Convert cover URI to full URL. + + :param cover_uri: Cover URI template. + :param size: Image size (e.g., '1000x1000'). + :return: Full image URL or None. + """ + if not cover_uri: + return None + # Cover URIs come in format "avatars.yandex.net/get-music-content/xxx/yyy/%%" + # Replace %% with the desired size + return f"https://{cover_uri.replace('%%', size)}" + + +def parse_artist(provider: KionMusicProvider, artist_obj: YandexArtist) -> Artist: + """Parse KION artist object to MA Artist model. + + :param provider: The KION Music provider instance. + :param artist_obj: KION artist object. + :return: Music Assistant Artist model. + """ + artist_id = str(artist_obj.id) + artist = Artist( + item_id=artist_id, + provider=provider.instance_id, + name=artist_obj.name or "Unknown Artist", + provider_mappings={ + ProviderMapping( + item_id=artist_id, + provider_domain=provider.domain, + provider_instance=provider.instance_id, + url=f"https://music.mts.ru/artist/{artist_id}", + ) + }, + ) + + # Add image if available + if artist_obj.cover: + image_url = _get_image_url(artist_obj.cover.uri) + if image_url: + artist.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + elif artist_obj.og_image: + image_url = _get_image_url(artist_obj.og_image) + if image_url: + artist.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + + return artist + + +def parse_album(provider: KionMusicProvider, album_obj: YandexAlbum) -> Album: + """Parse KION album object to MA Album model. + + :param provider: The KION Music provider instance. + :param album_obj: KION album object. + :return: Music Assistant Album model. + """ + name, version = parse_title_and_version( + album_obj.title or "Unknown Album", + album_obj.version or None, + ) + album_id = str(album_obj.id) + + # Determine availability + available = album_obj.available or False + + album = Album( + item_id=album_id, + provider=provider.instance_id, + name=name, + version=version, + provider_mappings={ + ProviderMapping( + item_id=album_id, + provider_domain=provider.domain, + provider_instance=provider.instance_id, + audio_format=AudioFormat( + content_type=_get_content_type(provider), + ), + url=f"https://music.mts.ru/album/{album_id}", + available=available, + ) + }, + ) + + # Parse artists + various_artist_album = False + if album_obj.artists: + for artist in album_obj.artists: + if artist.name and artist.name.lower() in ("various artists", "сборник"): + various_artist_album = True + album.artists.append(parse_artist(provider, artist)) + + # Determine album type + album_type_str = album_obj.type or "album" + if album_type_str == "compilation" or various_artist_album: + album.album_type = AlbumType.COMPILATION + elif album_type_str == "single": + album.album_type = AlbumType.SINGLE + else: + album.album_type = AlbumType.ALBUM + + # Parse year + if album_obj.year: + album.year = album_obj.year + if album_obj.release_date: + with suppress(ValueError): + album.metadata.release_date = datetime.fromisoformat(album_obj.release_date) + + # Parse metadata + if album_obj.genre: + album.metadata.genres = {album_obj.genre} + + # Add cover image + if album_obj.cover_uri: + image_url = _get_image_url(album_obj.cover_uri) + if image_url: + album.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + elif album_obj.og_image: + image_url = _get_image_url(album_obj.og_image) + if image_url: + album.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + + return album + + +def parse_track(provider: KionMusicProvider, track_obj: YandexTrack) -> Track: + """Parse KION track object to MA Track model. + + :param provider: The KION Music provider instance. + :param track_obj: KION track object. + :return: Music Assistant Track model. + """ + name, version = parse_title_and_version( + track_obj.title or "Unknown Track", + track_obj.version or None, + ) + track_id = str(track_obj.id) + + # Determine availability + available = track_obj.available or False + + # Duration is in milliseconds + duration = (track_obj.duration_ms or 0) // 1000 + + track = Track( + item_id=track_id, + provider=provider.instance_id, + name=name, + version=version, + duration=duration, + provider_mappings={ + ProviderMapping( + item_id=track_id, + provider_domain=provider.domain, + provider_instance=provider.instance_id, + audio_format=AudioFormat( + content_type=_get_content_type(provider), + ), + url=f"https://music.mts.ru/track/{track_id}", + available=available, + ) + }, + ) + + # Parse artists + if track_obj.artists: + track.artists = UniqueList() + for artist in track_obj.artists: + track.artists.append(parse_artist(provider, artist)) + + # Parse album (full data so album gets cover art in the library) + if track_obj.albums and len(track_obj.albums) > 0: + album_obj = track_obj.albums[0] + track.album = parse_album(provider, album_obj) + # Also set track image from album cover if available + if album_obj.cover_uri: + image_url = _get_image_url(album_obj.cover_uri) + if image_url: + track.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + + # Parse external IDs + if track_obj.real_id: + # real_id can be used as an identifier + pass + + # Metadata + if track_obj.content_warning: + track.metadata.explicit = track_obj.content_warning == "explicit" + + return track + + +def parse_playlist( + provider: KionMusicProvider, playlist_obj: YandexPlaylist, owner_name: str | None = None +) -> Playlist: + """Parse KION playlist object to MA Playlist model. + + :param provider: The KION Music provider instance. + :param playlist_obj: KION playlist object. + :param owner_name: Optional owner name override. + :return: Music Assistant Playlist model. + """ + # Playlist ID is a combination of owner uid and playlist kind + owner_id = str(playlist_obj.owner.uid) if playlist_obj.owner else str(provider.client.user_id) + playlist_kind = str(playlist_obj.kind) + playlist_id = f"{owner_id}:{playlist_kind}" + + # Determine if editable (user owns the playlist) + is_editable = owner_id == str(provider.client.user_id) + + # Get owner name + if owner_name is None: + if playlist_obj.owner and playlist_obj.owner.name: + owner_name = playlist_obj.owner.name + elif is_editable: + owner_name = "Me" + else: + owner_name = "KION Music" + + playlist = Playlist( + item_id=playlist_id, + provider=provider.instance_id, + name=playlist_obj.title or "Unknown Playlist", + owner=owner_name, + provider_mappings={ + ProviderMapping( + item_id=playlist_id, + provider_domain=provider.domain, + provider_instance=provider.instance_id, + url=f"https://music.mts.ru/users/{owner_id}/playlists/{playlist_kind}", + is_unique=is_editable, + ) + }, + is_editable=is_editable, + ) + + # Metadata + if playlist_obj.description: + playlist.metadata.description = playlist_obj.description + + # Add cover image + if playlist_obj.cover: + # Cover can be CoverImage or a string + cover = playlist_obj.cover + if hasattr(cover, "uri") and cover.uri: + image_url = _get_image_url(cover.uri) + if image_url: + playlist.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + elif playlist_obj.og_image: + image_url = _get_image_url(playlist_obj.og_image) + if image_url: + playlist.metadata.images = UniqueList( + [ + MediaItemImage( + type=ImageType.THUMB, + path=image_url, + provider=provider.instance_id, + remotely_accessible=True, + ) + ] + ) + + return playlist diff --git a/music_assistant/providers/kion_music/provider.py b/music_assistant/providers/kion_music/provider.py new file mode 100644 index 00000000..214eae1e --- /dev/null +++ b/music_assistant/providers/kion_music/provider.py @@ -0,0 +1,475 @@ +"""KION Music provider implementation.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from music_assistant_models.enums import MediaType +from music_assistant_models.errors import ( + InvalidDataError, + LoginFailed, + MediaNotFoundError, + ProviderUnavailableError, + ResourceTemporarilyUnavailable, +) +from music_assistant_models.media_items import ( + Album, + Artist, + ItemMapping, + MediaItemType, + Playlist, + SearchResults, + Track, +) + +from music_assistant.controllers.cache import use_cache +from music_assistant.models.music_provider import MusicProvider + +from .api_client import KionMusicClient +from .constants import CONF_TOKEN, PLAYLIST_ID_SPLITTER +from .parsers import parse_album, parse_artist, parse_playlist, parse_track +from .streaming import KionMusicStreamingManager + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + from music_assistant_models.streamdetails import StreamDetails + + +class KionMusicProvider(MusicProvider): + """Implementation of a KION Music MusicProvider.""" + + _client: KionMusicClient | None = None + _streaming: KionMusicStreamingManager | None = None + + @property + def client(self) -> KionMusicClient: + """Return the KION Music client.""" + if self._client is None: + raise ProviderUnavailableError("Provider not initialized") + return self._client + + @property + def streaming(self) -> KionMusicStreamingManager: + """Return the streaming manager.""" + if self._streaming is None: + raise ProviderUnavailableError("Provider not initialized") + return self._streaming + + async def handle_async_init(self) -> None: + """Handle async initialization of the provider.""" + token = self.config.get_value(CONF_TOKEN) + if not token: + raise LoginFailed("No KION Music token provided") + + self._client = KionMusicClient(str(token)) + await self._client.connect() + self._streaming = KionMusicStreamingManager(self) + self.logger.info("Successfully connected to KION Music") + + async def unload(self, is_removed: bool = False) -> None: + """Handle unload/close of the provider. + + :param is_removed: Whether the provider is being removed. + """ + if self._client: + await self._client.disconnect() + self._client = None + self._streaming = None + await super().unload(is_removed) + + def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping: + """Create a generic item mapping. + + :param media_type: The media type. + :param key: The item ID. + :param name: The item name. + :return: An ItemMapping instance. + """ + if isinstance(media_type, str): + media_type = MediaType(media_type) + return ItemMapping( + media_type=media_type, + item_id=key, + provider=self.instance_id, + name=name, + ) + + # Search + + @use_cache(3600 * 24 * 14) + async def search( + self, search_query: str, media_types: list[MediaType], limit: int = 5 + ) -> SearchResults: + """Perform search on KION Music. + + :param search_query: The search query. + :param media_types: List of media types to search for. + :param limit: Maximum number of results per type. + :return: SearchResults with found items. + """ + result = SearchResults() + + # Determine search type based on requested media types + # Map MediaType to KION API search type + type_mapping = { + MediaType.TRACK: "track", + MediaType.ALBUM: "album", + MediaType.ARTIST: "artist", + MediaType.PLAYLIST: "playlist", + } + requested_types = [type_mapping[mt] for mt in media_types if mt in type_mapping] + + # Use specific type if only one requested, otherwise search all + search_type = requested_types[0] if len(requested_types) == 1 else "all" + + search_result = await self.client.search(search_query, search_type=search_type, limit=limit) + if not search_result: + return result + + # Parse tracks + if MediaType.TRACK in media_types and search_result.tracks: + for track in search_result.tracks.results[:limit]: + try: + result.tracks = [*result.tracks, parse_track(self, track)] + except InvalidDataError as err: + self.logger.debug("Error parsing track: %s", err) + + # Parse albums + if MediaType.ALBUM in media_types and search_result.albums: + for album in search_result.albums.results[:limit]: + try: + result.albums = [*result.albums, parse_album(self, album)] + except InvalidDataError as err: + self.logger.debug("Error parsing album: %s", err) + + # Parse artists + if MediaType.ARTIST in media_types and search_result.artists: + for artist in search_result.artists.results[:limit]: + try: + result.artists = [*result.artists, parse_artist(self, artist)] + except InvalidDataError as err: + self.logger.debug("Error parsing artist: %s", err) + + # Parse playlists + if MediaType.PLAYLIST in media_types and search_result.playlists: + for playlist in search_result.playlists.results[:limit]: + try: + result.playlists = [*result.playlists, parse_playlist(self, playlist)] + except InvalidDataError as err: + self.logger.debug("Error parsing playlist: %s", err) + + return result + + # Get single items + + @use_cache(3600 * 24 * 30) + async def get_artist(self, prov_artist_id: str) -> Artist: + """Get artist details by ID. + + :param prov_artist_id: The provider artist ID. + :return: Artist object. + :raises MediaNotFoundError: If artist not found. + """ + artist = await self.client.get_artist(prov_artist_id) + if not artist: + raise MediaNotFoundError(f"Artist {prov_artist_id} not found") + return parse_artist(self, artist) + + @use_cache(3600 * 24 * 30) + async def get_album(self, prov_album_id: str) -> Album: + """Get album details by ID. + + :param prov_album_id: The provider album ID. + :return: Album object. + :raises MediaNotFoundError: If album not found. + """ + album = await self.client.get_album(prov_album_id) + if not album: + raise MediaNotFoundError(f"Album {prov_album_id} not found") + return parse_album(self, album) + + @use_cache(3600 * 24 * 30) + async def get_track(self, prov_track_id: str) -> Track: + """Get track details by ID. + + :param prov_track_id: The provider track ID. + :return: Track object. + :raises MediaNotFoundError: If track not found. + """ + yandex_track = await self.client.get_track(prov_track_id) + if not yandex_track: + raise MediaNotFoundError(f"Track {prov_track_id} not found") + return parse_track(self, yandex_track) + + @use_cache(3600 * 24 * 30) + async def get_playlist(self, prov_playlist_id: str) -> Playlist: + """Get playlist details by ID. + + :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind"). + :return: Playlist object. + :raises MediaNotFoundError: If playlist not found. + """ + # Parse the playlist ID (format: owner_id:kind) + if PLAYLIST_ID_SPLITTER in prov_playlist_id: + owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1) + else: + owner_id = str(self.client.user_id) + kind = prov_playlist_id + + playlist = await self.client.get_playlist(owner_id, kind) + if not playlist: + raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") + return parse_playlist(self, playlist) + + # Get related items + + @use_cache(3600 * 24 * 30) + async def get_album_tracks(self, prov_album_id: str) -> list[Track]: + """Get album tracks. + + :param prov_album_id: The provider album ID. + :return: List of Track objects. + """ + album = await self.client.get_album_with_tracks(prov_album_id) + if not album or not album.volumes: + return [] + + tracks = [] + for volume_index, volume in enumerate(album.volumes): + for track_index, track in enumerate(volume): + try: + parsed_track = parse_track(self, track) + parsed_track.disc_number = volume_index + 1 + parsed_track.track_number = track_index + 1 + tracks.append(parsed_track) + except InvalidDataError as err: + self.logger.debug("Error parsing album track: %s", err) + return tracks + + @use_cache(3600 * 3) + async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]: + """Get playlist tracks. + + :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind"). + :param page: Page number for pagination. + :return: List of Track objects. + """ + # KION Music API returns all playlist tracks in one call (no server-side pagination) + if page > 0: + return [] + + self.logger.debug("get_playlist_tracks called: %s", prov_playlist_id) + # Parse the playlist ID (format: owner_id:kind) + if PLAYLIST_ID_SPLITTER in prov_playlist_id: + owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1) + else: + owner_id = str(self.client.user_id) + kind = prov_playlist_id + + self.logger.debug("Fetching playlist %s/%s from API...", owner_id, kind) + playlist = await self.client.get_playlist(owner_id, kind) + if not playlist: + self.logger.debug("Playlist %s/%s not found", owner_id, kind) + return [] + + # API sometimes returns playlist without tracks; fetch them explicitly if needed + tracks_list = playlist.tracks or [] + track_count = getattr(playlist, "track_count", None) or 0 + self.logger.debug( + "Playlist %s/%s: track_count=%s, tracks_in_response=%s", + owner_id, + kind, + track_count, + len(tracks_list), + ) + if not tracks_list and track_count > 0: + self.logger.debug("No tracks in response, calling fetch_tracks_async...") + try: + tracks_list = await playlist.fetch_tracks_async() + self.logger.debug("fetch_tracks_async returned %s tracks", len(tracks_list or [])) + except Exception as err: + self.logger.warning("fetch_tracks_async failed: %s", err) + if not tracks_list: + self.logger.warning( + "Playlist %s/%s: expected %s tracks but got none", + owner_id, + kind, + track_count, + ) + raise ResourceTemporarilyUnavailable( + "Playlist tracks not available; try again later" + ) from None + + if not tracks_list: + self.logger.debug("Playlist %s/%s has no tracks", owner_id, kind) + return [] + + # API returns TrackShort objects, we need to fetch full track info + track_ids = [ + str(track.track_id) if hasattr(track, "track_id") else str(track.id) + for track in tracks_list + if track + ] + if not track_ids: + return [] + + self.logger.debug("Fetching full details for %s tracks...", len(track_ids)) + # Fetch full track details in batches to avoid timeouts + batch_size = 50 + full_tracks = [] + for i in range(0, len(track_ids), batch_size): + batch = track_ids[i : i + batch_size] + self.logger.debug("Fetching batch %s-%s...", i, i + len(batch)) + batch_result = await self.client.get_tracks(batch) + self.logger.debug("Batch returned %s tracks", len(batch_result or [])) + full_tracks.extend(batch_result or []) + + if track_ids and not full_tracks: + self.logger.warning("Got 0 full tracks for %s IDs", len(track_ids)) + raise ResourceTemporarilyUnavailable( + "Failed to load track details; try again later" + ) from None + + tracks = [] + for track in full_tracks: + try: + tracks.append(parse_track(self, track)) + except InvalidDataError as err: + self.logger.debug("Error parsing playlist track: %s", err) + self.logger.debug("Returning %s parsed tracks", len(tracks)) + return tracks + + @use_cache(3600 * 24 * 7) + async def get_artist_albums(self, prov_artist_id: str) -> list[Album]: + """Get artist's albums. + + :param prov_artist_id: The provider artist ID. + :return: List of Album objects. + """ + albums = await self.client.get_artist_albums(prov_artist_id) + result = [] + for album in albums: + try: + result.append(parse_album(self, album)) + except InvalidDataError as err: + self.logger.debug("Error parsing artist album: %s", err) + return result + + @use_cache(3600 * 24 * 7) + async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]: + """Get artist's top tracks. + + :param prov_artist_id: The provider artist ID. + :return: List of Track objects. + """ + tracks = await self.client.get_artist_tracks(prov_artist_id) + result = [] + for track in tracks: + try: + result.append(parse_track(self, track)) + except InvalidDataError as err: + self.logger.debug("Error parsing artist track: %s", err) + return result + + # Library methods + + async def get_library_artists(self) -> AsyncGenerator[Artist, None]: + """Retrieve library artists from KION Music.""" + artists = await self.client.get_liked_artists() + for artist in artists: + try: + yield parse_artist(self, artist) + except InvalidDataError as err: + self.logger.debug("Error parsing library artist: %s", err) + + async def get_library_albums(self) -> AsyncGenerator[Album, None]: + """Retrieve library albums from KION Music.""" + albums = await self.client.get_liked_albums() + for album in albums: + try: + yield parse_album(self, album) + except InvalidDataError as err: + self.logger.debug("Error parsing library album: %s", err) + + async def get_library_tracks(self) -> AsyncGenerator[Track, None]: + """Retrieve library tracks from KION Music.""" + track_shorts = await self.client.get_liked_tracks() + if not track_shorts: + return + + # Fetch full track details in batches + track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id] + batch_size = 50 + for i in range(0, len(track_ids), batch_size): + batch_ids = track_ids[i : i + batch_size] + full_tracks = await self.client.get_tracks(batch_ids) + for track in full_tracks: + try: + yield parse_track(self, track) + except InvalidDataError as err: + self.logger.debug("Error parsing library track: %s", err) + + async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]: + """Retrieve library playlists from KION Music.""" + playlists = await self.client.get_user_playlists() + for playlist in playlists: + try: + yield parse_playlist(self, playlist) + except InvalidDataError as err: + self.logger.debug("Error parsing library playlist: %s", err) + + # Library edit methods + + async def library_add(self, item: MediaItemType) -> bool: + """Add item to library. + + :param item: The media item to add. + :return: True if successful. + """ + prov_item_id = self._get_provider_item_id(item) + if not prov_item_id: + return False + + if item.media_type == MediaType.TRACK: + return await self.client.like_track(prov_item_id) + if item.media_type == MediaType.ALBUM: + return await self.client.like_album(prov_item_id) + if item.media_type == MediaType.ARTIST: + return await self.client.like_artist(prov_item_id) + return False + + async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool: + """Remove item from library. + + :param prov_item_id: The provider item ID. + :param media_type: The media type. + :return: True if successful. + """ + if media_type == MediaType.TRACK: + return await self.client.unlike_track(prov_item_id) + if media_type == MediaType.ALBUM: + return await self.client.unlike_album(prov_item_id) + if media_type == MediaType.ARTIST: + return await self.client.unlike_artist(prov_item_id) + return False + + def _get_provider_item_id(self, item: MediaItemType) -> str | None: + """Get provider item ID from media item.""" + for mapping in item.provider_mappings: + if mapping.provider_instance == self.instance_id: + return mapping.item_id + return item.item_id if item.provider == self.instance_id else None + + # Streaming + + async def get_stream_details( + self, item_id: str, media_type: MediaType = MediaType.TRACK + ) -> StreamDetails: + """Get stream details for a track. + + :param item_id: The track ID. + :param media_type: The media type (should be TRACK). + :return: StreamDetails for the track. + """ + return await self.streaming.get_stream_details(item_id) diff --git a/music_assistant/providers/kion_music/streaming.py b/music_assistant/providers/kion_music/streaming.py new file mode 100644 index 00000000..54f659e2 --- /dev/null +++ b/music_assistant/providers/kion_music/streaming.py @@ -0,0 +1,179 @@ +"""Streaming operations for KION Music.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from music_assistant_models.enums import ContentType, StreamType +from music_assistant_models.errors import MediaNotFoundError +from music_assistant_models.media_items import AudioFormat +from music_assistant_models.streamdetails import StreamDetails + +from .constants import CONF_QUALITY, QUALITY_LOSSLESS + +if TYPE_CHECKING: + from yandex_music import DownloadInfo + + from .provider import KionMusicProvider + + +class KionMusicStreamingManager: + """Manages KION Music streaming operations.""" + + def __init__(self, provider: KionMusicProvider) -> None: + """Initialize streaming manager. + + :param provider: The KION Music provider instance. + """ + self.provider = provider + self.client = provider.client + self.mass = provider.mass + self.logger = provider.logger + + async def get_stream_details(self, item_id: str) -> StreamDetails: + """Get stream details for a track. + + :param item_id: Track ID. + :return: StreamDetails for the track. + :raises MediaNotFoundError: If stream URL cannot be obtained. + """ + # Get track info first + track = await self.provider.get_track(item_id) + if not track: + raise MediaNotFoundError(f"Track {item_id} not found") + + quality = self.provider.config.get_value(CONF_QUALITY) + quality_str = str(quality) if quality is not None else None + preferred_normalized = (quality_str or "").strip().lower() + want_lossless = ( + QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS + ) + + # When user wants lossless, try get-file-info first (FLAC; download-info often MP3 only) + if want_lossless: + self.logger.debug("Requesting lossless via get-file-info for track %s", item_id) + file_info = await self.client.get_track_file_info_lossless(item_id) + if file_info: + url = file_info.get("url") + codec = file_info.get("codec") or "" + if url and codec.lower() in ("flac", "flac-mp4"): + content_type = self._get_content_type(codec) + self.logger.debug( + "Stream selected for track %s via get-file-info: codec=%s", + item_id, + codec, + ) + return StreamDetails( + item_id=item_id, + provider=self.provider.instance_id, + audio_format=AudioFormat( + content_type=content_type, + bit_rate=0, + ), + stream_type=StreamType.HTTP, + duration=track.duration, + path=url, + can_seek=True, + allow_seek=True, + ) + + # Default: use /tracks/.../download-info and select best quality + download_infos = await self.client.get_track_download_info(item_id, get_direct_links=True) + if not download_infos: + raise MediaNotFoundError(f"No stream info available for track {item_id}") + + codecs_available = [ + (getattr(i, "codec", None), getattr(i, "bitrate_in_kbps", None)) for i in download_infos + ] + self.logger.debug( + "Stream quality for track %s: config quality=%s, available codecs=%s", + item_id, + quality_str, + codecs_available, + ) + selected_info = self._select_best_quality(download_infos, quality_str) + + if not selected_info or not selected_info.direct_link: + raise MediaNotFoundError(f"No stream URL available for track {item_id}") + + self.logger.debug( + "Stream selected for track %s: codec=%s, bitrate=%s", + item_id, + getattr(selected_info, "codec", None), + getattr(selected_info, "bitrate_in_kbps", None), + ) + + content_type = self._get_content_type(selected_info.codec) + bitrate = selected_info.bitrate_in_kbps or 0 + + return StreamDetails( + item_id=item_id, + provider=self.provider.instance_id, + audio_format=AudioFormat( + content_type=content_type, + bit_rate=bitrate, + ), + stream_type=StreamType.HTTP, + duration=track.duration, + path=selected_info.direct_link, + can_seek=True, + allow_seek=True, + ) + + def _select_best_quality( + self, download_infos: list[Any], preferred_quality: str | None + ) -> DownloadInfo | None: + """Select the best quality download info. + + :param download_infos: List of DownloadInfo objects. + :param preferred_quality: User's preferred quality (e.g. "lossless" or "Lossless (FLAC)"). + :return: Best matching DownloadInfo or None. + """ + if not download_infos: + return None + + # Normalize so we accept "lossless", "Lossless (FLAC)", etc. + preferred_normalized = (preferred_quality or "").strip().lower() + want_lossless = ( + QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS + ) + + # Sort by bitrate descending + sorted_infos = sorted( + download_infos, + key=lambda x: x.bitrate_in_kbps or 0, + reverse=True, + ) + + # If user wants lossless, prefer flac-mp4 then flac (API formats ~2025) + if want_lossless: + for codec in ("flac-mp4", "flac"): + for info in sorted_infos: + if info.codec and info.codec.lower() == codec: + return info + self.logger.warning( + "Lossless (FLAC) requested but no FLAC in API response for this " + "track; using best available" + ) + + # Return highest bitrate + return sorted_infos[0] if sorted_infos else None + + def _get_content_type(self, codec: str | None) -> ContentType: + """Determine content type from codec string. + + :param codec: Codec string from API. + :return: ContentType enum value. + """ + if not codec: + return ContentType.UNKNOWN + + codec_lower = codec.lower() + if codec_lower in ("flac", "flac-mp4"): + return ContentType.FLAC + if codec_lower in ("mp3", "mpeg"): + return ContentType.MP3 + if codec_lower == "aac": + return ContentType.AAC + + return ContentType.UNKNOWN diff --git a/tests/providers/kion_music/__init__.py b/tests/providers/kion_music/__init__.py new file mode 100644 index 00000000..40d4adae --- /dev/null +++ b/tests/providers/kion_music/__init__.py @@ -0,0 +1 @@ +"""Tests for KION Music provider.""" diff --git a/tests/providers/kion_music/__snapshots__/test_parsers.ambr b/tests/providers/kion_music/__snapshots__/test_parsers.ambr new file mode 100644 index 00000000..0a7ebc1e --- /dev/null +++ b/tests/providers/kion_music/__snapshots__/test_parsers.ambr @@ -0,0 +1,592 @@ +# serializer version: 1 +# name: test_parse_album_snapshot[minimal] + dict({ + 'album_type': 'album', + 'artists': list([ + ]), + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '300', + 'media_type': 'album', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Album', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '300', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/album/300', + }), + ]), + 'sort_name': 'test album', + 'translation_key': None, + 'uri': 'kion_music_instance://album/300', + 'version': '', + 'year': 2020, + }) +# --- +# name: test_parse_artist_snapshot[minimal] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '100', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Artist', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '100', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/artist/100', + }), + ]), + 'sort_name': 'test artist', + 'translation_key': None, + 'uri': 'kion_music_instance://artist/100', + 'version': '', + }) +# --- +# name: test_parse_artist_snapshot[with_cover] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '200', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': list([ + dict({ + 'path': 'https://avatars.yandex.net/get-music-content/xxx/yyy/1000x1000', + 'provider': 'kion_music_instance', + 'remotely_accessible': True, + 'type': 'thumb', + }), + ]), + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Artist With Cover', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '200', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/artist/200', + }), + ]), + 'sort_name': 'artist with cover', + 'translation_key': None, + 'uri': 'kion_music_instance://artist/200', + 'version': '', + }) +# --- +# name: test_parse_playlist_snapshot[minimal] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_editable': True, + 'is_playable': True, + 'item_id': '12345:3', + 'media_type': 'playlist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'My Playlist', + 'owner': 'Me', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': True, + 'item_id': '12345:3', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/users/12345/playlists/3', + }), + ]), + 'sort_name': 'my playlist', + 'translation_key': None, + 'uri': 'kion_music_instance://playlist/12345:3', + 'version': '', + }) +# --- +# name: test_parse_playlist_snapshot[other_user] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_editable': False, + 'is_playable': True, + 'item_id': '99999:1', + 'media_type': 'playlist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': 'A shared playlist', + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Shared Playlist', + 'owner': 'Other User', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': False, + 'item_id': '99999:1', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/users/99999/playlists/1', + }), + ]), + 'sort_name': 'shared playlist', + 'translation_key': None, + 'uri': 'kion_music_instance://playlist/99999:1', + 'version': '', + }) +# --- +# name: test_parse_track_snapshot[minimal] + dict({ + 'album': None, + 'artists': list([ + ]), + 'date_added': None, + 'disc_number': 0, + 'duration': 180, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '400', + 'last_played': 0, + 'media_type': 'track', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Track', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '400', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/track/400', + }), + ]), + 'sort_name': 'test track', + 'track_number': 0, + 'translation_key': None, + 'uri': 'kion_music_instance://track/400', + 'version': '', + }) +# --- +# name: test_parse_track_snapshot[with_artist_and_album] + dict({ + 'album': dict({ + 'album_type': 'album', + 'artists': list([ + ]), + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '20', + 'media_type': 'album', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': list([ + dict({ + 'path': 'https://avatars.yandex.net/get-music-content/aaa/bbb/1000x1000', + 'provider': 'kion_music_instance', + 'remotely_accessible': True, + 'type': 'thumb', + }), + ]), + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Track Album', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': False, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '20', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/album/20', + }), + ]), + 'sort_name': 'track album', + 'translation_key': None, + 'uri': 'kion_music_instance://album/20', + 'version': '', + 'year': None, + }), + 'artists': list([ + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '10', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Track Artist', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '10', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/artist/10', + }), + ]), + 'sort_name': 'track artist', + 'translation_key': None, + 'uri': 'kion_music_instance://artist/10', + 'version': '', + }), + ]), + 'date_added': None, + 'disc_number': 0, + 'duration': 240, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '500', + 'last_played': 0, + 'media_type': 'track', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': list([ + dict({ + 'path': 'https://avatars.yandex.net/get-music-content/aaa/bbb/1000x1000', + 'provider': 'kion_music_instance', + 'remotely_accessible': True, + 'type': 'thumb', + }), + ]), + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Track With Album', + 'position': None, + 'provider': 'kion_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '500', + 'provider_domain': 'kion_music', + 'provider_instance': 'kion_music_instance', + 'url': 'https://music.mts.ru/track/500', + }), + ]), + 'sort_name': 'track with album', + 'track_number': 0, + 'translation_key': None, + 'uri': 'kion_music_instance://track/500', + 'version': '', + }) +# --- diff --git a/tests/providers/kion_music/conftest.py b/tests/providers/kion_music/conftest.py new file mode 100644 index 00000000..7213c5bd --- /dev/null +++ b/tests/providers/kion_music/conftest.py @@ -0,0 +1,96 @@ +"""Shared fixtures and stubs for KION Music provider tests.""" + +from __future__ import annotations + +import logging + +import pytest +from music_assistant_models.enums import MediaType +from music_assistant_models.media_items import ItemMapping + + +class ProviderStub: + """Minimal provider-like object for parser tests (no Mock). + + Provides the minimal interface needed by parse_* functions. + """ + + domain = "kion_music" + instance_id = "kion_music_instance" + + def __init__(self) -> None: + """Initialize stub with minimal client.""" + self.client = type("ClientStub", (), {"user_id": 12345})() + + def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping: + """Return ItemMapping for the given media type, key and name.""" + return ItemMapping( + media_type=MediaType(media_type) if isinstance(media_type, str) else media_type, + item_id=key, + provider=self.instance_id, + name=name, + ) + + +class StreamingProviderStub: + """Minimal provider stub for streaming tests (no Mock). + + Provides the minimal interface needed by KionMusicStreamingManager. + """ + + domain = "kion_music" + instance_id = "kion_music_instance" + logger = logging.getLogger("kion_music_test_streaming") + + def __init__(self) -> None: + """Initialize stub with minimal client.""" + self.client = type("ClientStub", (), {"user_id": 12345})() + self.mass = type("MassStub", (), {})() + self._warning_count = 0 + + def _count_warning(self, *args: object, **kwargs: object) -> None: + """Track warning calls for test assertions.""" + self._warning_count += 1 + + +class TrackingLogger: + """Logger that tracks calls for test assertions without using Mock.""" + + def __init__(self) -> None: + """Initialize with empty call counters.""" + self._debug_count = 0 + self._info_count = 0 + self._warning_count = 0 + self._error_count = 0 + + def debug(self, *args: object, **kwargs: object) -> None: + """Track debug calls.""" + self._debug_count += 1 + + def info(self, *args: object, **kwargs: object) -> None: + """Track info calls.""" + self._info_count += 1 + + def warning(self, *args: object, **kwargs: object) -> None: + """Track warning calls.""" + self._warning_count += 1 + + def error(self, *args: object, **kwargs: object) -> None: + """Track error calls.""" + self._error_count += 1 + + +# Minimal client-like object for yandex_music de_json (library requires client, not None) +DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})() + + +@pytest.fixture +def provider_stub() -> ProviderStub: + """Return a real provider stub (no Mock).""" + return ProviderStub() + + +@pytest.fixture +def streaming_provider_stub() -> StreamingProviderStub: + """Return a streaming provider stub (no Mock).""" + return StreamingProviderStub() diff --git a/tests/providers/kion_music/fixtures/albums/minimal.json b/tests/providers/kion_music/fixtures/albums/minimal.json new file mode 100644 index 00000000..ec8a82c5 --- /dev/null +++ b/tests/providers/kion_music/fixtures/albums/minimal.json @@ -0,0 +1,8 @@ +{ + "id": 300, + "title": "Test Album", + "available": true, + "artists": [], + "type": "album", + "year": 2020 +} diff --git a/tests/providers/kion_music/fixtures/artists/minimal.json b/tests/providers/kion_music/fixtures/artists/minimal.json new file mode 100644 index 00000000..06296f0a --- /dev/null +++ b/tests/providers/kion_music/fixtures/artists/minimal.json @@ -0,0 +1,4 @@ +{ + "id": 100, + "name": "Test Artist" +} diff --git a/tests/providers/kion_music/fixtures/artists/with_cover.json b/tests/providers/kion_music/fixtures/artists/with_cover.json new file mode 100644 index 00000000..ef6c49ae --- /dev/null +++ b/tests/providers/kion_music/fixtures/artists/with_cover.json @@ -0,0 +1,8 @@ +{ + "id": 200, + "name": "Artist With Cover", + "cover": { + "type": "from-og-image", + "uri": "avatars.yandex.net/get-music-content/xxx/yyy/%%" + } +} diff --git a/tests/providers/kion_music/fixtures/playlists/minimal.json b/tests/providers/kion_music/fixtures/playlists/minimal.json new file mode 100644 index 00000000..6e77c679 --- /dev/null +++ b/tests/providers/kion_music/fixtures/playlists/minimal.json @@ -0,0 +1,9 @@ +{ + "owner": { + "uid": 12345, + "name": "Me", + "login": "me" + }, + "kind": 3, + "title": "My Playlist" +} diff --git a/tests/providers/kion_music/fixtures/playlists/other_user.json b/tests/providers/kion_music/fixtures/playlists/other_user.json new file mode 100644 index 00000000..60fba828 --- /dev/null +++ b/tests/providers/kion_music/fixtures/playlists/other_user.json @@ -0,0 +1,10 @@ +{ + "owner": { + "uid": 99999, + "name": "Other User", + "login": "other_user" + }, + "kind": 1, + "title": "Shared Playlist", + "description": "A shared playlist" +} diff --git a/tests/providers/kion_music/fixtures/tracks/minimal.json b/tests/providers/kion_music/fixtures/tracks/minimal.json new file mode 100644 index 00000000..4aed92b4 --- /dev/null +++ b/tests/providers/kion_music/fixtures/tracks/minimal.json @@ -0,0 +1,8 @@ +{ + "id": 400, + "title": "Test Track", + "available": true, + "duration_ms": 180000, + "artists": [], + "albums": [] +} diff --git a/tests/providers/kion_music/fixtures/tracks/with_artist_and_album.json b/tests/providers/kion_music/fixtures/tracks/with_artist_and_album.json new file mode 100644 index 00000000..2211d3e2 --- /dev/null +++ b/tests/providers/kion_music/fixtures/tracks/with_artist_and_album.json @@ -0,0 +1,19 @@ +{ + "id": 500, + "title": "Track With Album", + "available": true, + "duration_ms": 240000, + "artists": [ + { + "id": 10, + "name": "Track Artist" + } + ], + "albums": [ + { + "id": 20, + "title": "Track Album", + "cover_uri": "avatars.yandex.net/get-music-content/aaa/bbb/%%" + } + ] +} diff --git a/tests/providers/kion_music/test_api_client.py b/tests/providers/kion_music/test_api_client.py new file mode 100644 index 00000000..a75cd930 --- /dev/null +++ b/tests/providers/kion_music/test_api_client.py @@ -0,0 +1,108 @@ +"""Unit tests for the KION Music API client.""" + +from __future__ import annotations + +from unittest import mock + +import pytest +from music_assistant_models.errors import ResourceTemporarilyUnavailable +from yandex_music.exceptions import NetworkError + +from music_assistant.providers.kion_music.api_client import KION_BASE_URL, KionMusicClient + + +@pytest.fixture +def client() -> KionMusicClient: + """Return a KionMusicClient with a fake token.""" + return KionMusicClient("fake_token") + + +async def test_connect_sets_base_url(client: KionMusicClient) -> None: + """Verify connect() passes KION_BASE_URL to ClientAsync.""" + with mock.patch("music_assistant.providers.kion_music.api_client.ClientAsync") as mock_cls: + mock_instance = mock.AsyncMock() + mock_instance.me = type("Me", (), {"account": type("Account", (), {"uid": 42})()})() + mock_instance.init = mock.AsyncMock(return_value=mock_instance) + mock_cls.return_value = mock_instance + + result = await client.connect() + + assert result is True + mock_cls.assert_called_once_with("fake_token", base_url=KION_BASE_URL) + + +async def test_get_liked_albums_batching(client: KionMusicClient) -> None: + """Test that liked albums are fetched in batches of 50.""" + mock_client = mock.AsyncMock() + client._client = mock_client + client._user_id = 1 + + # Create 60 likes so we get 2 batches + likes = [] + for i in range(60): + like = type("Like", (), {"album": type("Album", (), {"id": i + 1})()})() + likes.append(like) + + mock_client.users_likes_albums = mock.AsyncMock(return_value=likes) + + batch1 = [type("Album", (), {"id": i + 1})() for i in range(50)] + batch2 = [type("Album", (), {"id": i + 51})() for i in range(10)] + mock_client.albums = mock.AsyncMock(side_effect=[batch1, batch2]) + + result = await client.get_liked_albums() + + assert len(result) == 60 + assert mock_client.albums.call_count == 2 + + +async def test_get_liked_albums_batch_fallback_on_network_error( + client: KionMusicClient, +) -> None: + """Test fallback to minimal data when batch fetch fails.""" + mock_client = mock.AsyncMock() + client._client = mock_client + client._user_id = 1 + + album_obj = type("Album", (), {"id": 1})() + likes = [type("Like", (), {"album": album_obj})()] + + mock_client.users_likes_albums = mock.AsyncMock(return_value=likes) + mock_client.albums = mock.AsyncMock(side_effect=NetworkError("timeout")) + + result = await client.get_liked_albums() + + assert len(result) == 1 + assert result[0].id == 1 + + +async def test_get_tracks_retry_on_network_error_then_success( + client: KionMusicClient, +) -> None: + """Test that get_tracks retries once on NetworkError and succeeds.""" + mock_client = mock.AsyncMock() + client._client = mock_client + client._user_id = 1 + + track = type("Track", (), {"id": 1})() + mock_client.tracks = mock.AsyncMock(side_effect=[NetworkError("timeout"), [track]]) + + result = await client.get_tracks(["1"]) + + assert len(result) == 1 + assert mock_client.tracks.call_count == 2 + + +async def test_get_tracks_retry_on_network_error_both_fail( + client: KionMusicClient, +) -> None: + """Test that get_tracks raises ResourceTemporarilyUnavailable when retry fails.""" + mock_client = mock.AsyncMock() + client._client = mock_client + client._user_id = 1 + + mock_client.tracks = mock.AsyncMock(side_effect=NetworkError("timeout")) + + with pytest.raises(ResourceTemporarilyUnavailable): + await client.get_tracks(["1"]) + + assert mock_client.tracks.call_count == 2 diff --git a/tests/providers/kion_music/test_integration.py b/tests/providers/kion_music/test_integration.py new file mode 100644 index 00000000..e3e969b7 --- /dev/null +++ b/tests/providers/kion_music/test_integration.py @@ -0,0 +1,354 @@ +"""Integration tests for the KION Music provider with in-process Music Assistant.""" + +from __future__ import annotations + +import json +import pathlib +from collections.abc import AsyncGenerator +from typing import TYPE_CHECKING, Any, cast +from unittest import mock + +import pytest +from music_assistant_models.enums import ContentType, MediaType, StreamType +from yandex_music import Album as YandexAlbum +from yandex_music import Artist as YandexArtist +from yandex_music import Playlist as YandexPlaylist +from yandex_music import Track as YandexTrack + +from music_assistant.mass import MusicAssistant +from music_assistant.models.music_provider import MusicProvider +from tests.common import wait_for_sync_completion + +if TYPE_CHECKING: + from music_assistant_models.config_entries import ProviderConfig + +FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" +_DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})() + + +def _load_json(path: pathlib.Path) -> dict[str, Any]: + """Load JSON fixture.""" + with open(path) as f: + return cast("dict[str, Any]", json.load(f)) + + +def _load_kion_objects() -> tuple[Any, Any, Any, Any]: + """Load Artist, Album, Track, Playlist from fixtures for mock client.""" + artist = YandexArtist.de_json( + _load_json(FIXTURES_DIR / "artists" / "minimal.json"), _DE_JSON_CLIENT + ) + album = YandexAlbum.de_json( + _load_json(FIXTURES_DIR / "albums" / "minimal.json"), _DE_JSON_CLIENT + ) + track = YandexTrack.de_json( + _load_json(FIXTURES_DIR / "tracks" / "minimal.json"), _DE_JSON_CLIENT + ) + playlist = YandexPlaylist.de_json( + _load_json(FIXTURES_DIR / "playlists" / "minimal.json"), _DE_JSON_CLIENT + ) + return artist, album, track, playlist + + +def _make_search_result(track: Any, album: Any, artist: Any, playlist: Any) -> Any: + """Build a Search-like object with .tracks.results, .albums.results, etc.""" + return type( + "Search", + (), + { + "tracks": type("TracksResult", (), {"results": [track]})(), + "albums": type("AlbumsResult", (), {"results": [album]})(), + "artists": type("ArtistsResult", (), {"results": [artist]})(), + "playlists": type("PlaylistsResult", (), {"results": [playlist]})(), + }, + )() + + +def _make_download_info( + codec: str = "mp3", + direct_link: str = "https://example.com/kion_track.mp3", + bitrate_in_kbps: int = 320, +) -> Any: + """Build DownloadInfo-like object for streaming.""" + return type( + "DownloadInfo", + (), + { + "direct_link": direct_link, + "codec": codec, + "bitrate_in_kbps": bitrate_in_kbps, + }, + )() + + +@pytest.fixture +async def kion_music_provider( + mass: MusicAssistant, +) -> AsyncGenerator[ProviderConfig, None]: + """Configure KION Music provider with mocked API client and add to mass.""" + artist, album, track, playlist = _load_kion_objects() + search_result = _make_search_result(track, album, artist, playlist) + download_info = _make_download_info() + + # Album with volumes for get_album_tracks + album_with_volumes = type( + "AlbumWithVolumes", + (), + { + "id": album.id, + "title": album.title, + "volumes": [[track]], + "artists": album.artists if hasattr(album, "artists") else [], + "year": getattr(album, "year", None), + "release_date": getattr(album, "release_date", None), + "genre": getattr(album, "genre", None), + "cover_uri": getattr(album, "cover_uri", None), + "og_image": getattr(album, "og_image", None), + "type": getattr(album, "type", "album"), + "available": getattr(album, "available", True), + }, + )() + + with mock.patch( + "music_assistant.providers.kion_music.provider.KionMusicClient" + ) as mock_client_class: + mock_client = mock.AsyncMock() + mock_client_class.return_value = mock_client + + mock_client.connect = mock.AsyncMock(return_value=True) + mock_client.user_id = 12345 + + mock_client.get_liked_tracks = mock.AsyncMock(return_value=[]) + mock_client.get_liked_albums = mock.AsyncMock(return_value=[]) + mock_client.get_liked_artists = mock.AsyncMock(return_value=[]) + mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist]) + + mock_client.search = mock.AsyncMock(return_value=search_result) + mock_client.get_track = mock.AsyncMock(return_value=track) + mock_client.get_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_album = mock.AsyncMock(return_value=album) + mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes) + mock_client.get_artist = mock.AsyncMock(return_value=artist) + mock_client.get_artist_albums = mock.AsyncMock(return_value=[album]) + mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_playlist = mock.AsyncMock(return_value=playlist) + mock_client.get_track_download_info = mock.AsyncMock(return_value=[download_info]) + + async with wait_for_sync_completion(mass): + config = await mass.config.save_provider_config( + "kion_music", + {"token": "mock_kion_token", "quality": "high"}, + ) + await mass.music.start_sync() + + yield config + + +@pytest.fixture +async def kion_music_provider_lossless( + mass: MusicAssistant, +) -> AsyncGenerator[ProviderConfig, None]: + """Configure KION Music with quality=lossless and mock returning MP3 + FLAC.""" + artist, album, track, playlist = _load_kion_objects() + search_result = _make_search_result(track, album, artist, playlist) + mp3_info = _make_download_info( + codec="mp3", + direct_link="https://example.com/kion_track.mp3", + bitrate_in_kbps=320, + ) + flac_info = _make_download_info( + codec="flac", + direct_link="https://example.com/kion_track.flac", + bitrate_in_kbps=0, + ) + download_infos = [mp3_info, flac_info] + + album_with_volumes = type( + "AlbumWithVolumes", + (), + { + "id": album.id, + "title": album.title, + "volumes": [[track]], + "artists": album.artists if hasattr(album, "artists") else [], + "year": getattr(album, "year", None), + "release_date": getattr(album, "release_date", None), + "genre": getattr(album, "genre", None), + "cover_uri": getattr(album, "cover_uri", None), + "og_image": getattr(album, "og_image", None), + "type": getattr(album, "type", "album"), + "available": getattr(album, "available", True), + }, + )() + + with mock.patch( + "music_assistant.providers.kion_music.provider.KionMusicClient" + ) as mock_client_class: + mock_client = mock.AsyncMock() + mock_client_class.return_value = mock_client + + mock_client.connect = mock.AsyncMock(return_value=True) + mock_client.user_id = 12345 + + mock_client.get_liked_tracks = mock.AsyncMock(return_value=[]) + mock_client.get_liked_albums = mock.AsyncMock(return_value=[]) + mock_client.get_liked_artists = mock.AsyncMock(return_value=[]) + mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist]) + + mock_client.search = mock.AsyncMock(return_value=search_result) + mock_client.get_track = mock.AsyncMock(return_value=track) + mock_client.get_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_album = mock.AsyncMock(return_value=album) + mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes) + mock_client.get_artist = mock.AsyncMock(return_value=artist) + mock_client.get_artist_albums = mock.AsyncMock(return_value=[album]) + mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_playlist = mock.AsyncMock(return_value=playlist) + mock_client.get_track_file_info_lossless = mock.AsyncMock(return_value=None) + mock_client.get_track_download_info = mock.AsyncMock(return_value=download_infos) + + async with wait_for_sync_completion(mass): + config = await mass.config.save_provider_config( + "kion_music", + {"token": "mock_kion_token", "quality": "lossless"}, + ) + await mass.music.start_sync() + + yield config + + +def _get_kion_provider(mass: MusicAssistant) -> MusicProvider | None: + """Get KION Music provider instance from mass.""" + for provider in mass.music.providers: + if provider.domain == "kion_music": + return provider + return None + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_registration_and_sync(mass: MusicAssistant) -> None: + """Test that provider is registered and sync completes.""" + prov = _get_kion_provider(mass) + assert prov is not None + assert prov.domain == "kion_music" + assert prov.instance_id + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_search(mass: MusicAssistant) -> None: + """Test search returns results from kion_music.""" + results = await mass.music.search("test query", [MediaType.TRACK], limit=5) + kion_tracks = [t for t in results.tracks if t.provider and "kion_music" in t.provider] + assert len(kion_tracks) >= 0 + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_artist(mass: MusicAssistant) -> None: + """Test getting artist by id.""" + prov = _get_kion_provider(mass) + assert prov is not None + artist = await prov.get_artist("100") + assert artist is not None + assert artist.name + assert artist.provider == prov.instance_id + assert artist.item_id == "100" + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_album(mass: MusicAssistant) -> None: + """Test getting album by id.""" + prov = _get_kion_provider(mass) + assert prov is not None + album = await prov.get_album("300") + assert album is not None + assert album.name + assert album.provider == prov.instance_id + assert album.item_id == "300" + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_track(mass: MusicAssistant) -> None: + """Test getting track by id.""" + prov = _get_kion_provider(mass) + assert prov is not None + track = await prov.get_track("400") + assert track is not None + assert track.name + assert track.provider == prov.instance_id + assert track.item_id == "400" + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_album_tracks(mass: MusicAssistant) -> None: + """Test getting album tracks.""" + prov = _get_kion_provider(mass) + assert prov is not None + tracks = await prov.get_album_tracks("300") + assert isinstance(tracks, list) + assert len(tracks) >= 0 + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_playlist_tracks(mass: MusicAssistant) -> None: + """Test getting playlist tracks.""" + prov = _get_kion_provider(mass) + assert prov is not None + tracks = await prov.get_playlist_tracks("12345:3", page=0) + assert isinstance(tracks, list) + assert len(tracks) >= 0 + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_playlist_tracks_page_gt_zero_returns_empty(mass: MusicAssistant) -> None: + """Test that page > 0 returns empty list (no server-side pagination).""" + prov = _get_kion_provider(mass) + assert prov is not None + tracks = await prov.get_playlist_tracks("12345:3", page=1) + assert tracks == [] + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_get_stream_details(mass: MusicAssistant) -> None: + """Test stream details retrieval.""" + prov = _get_kion_provider(mass) + assert prov is not None + stream_details = await prov.get_stream_details("400", MediaType.TRACK) + assert stream_details is not None + assert stream_details.stream_type == StreamType.HTTP + assert stream_details.path == "https://example.com/kion_track.mp3" + + +@pytest.mark.usefixtures("kion_music_provider_lossless") +async def test_get_stream_details_returns_flac_when_lossless_selected( + mass: MusicAssistant, +) -> None: + """When quality=lossless and API returns MP3+FLAC, stream details use FLAC.""" + prov = _get_kion_provider(mass) + assert prov is not None + stream_details = await prov.get_stream_details("400", MediaType.TRACK) + assert stream_details is not None + assert stream_details.audio_format.content_type == ContentType.FLAC + assert stream_details.path == "https://example.com/kion_track.flac" + + +@pytest.mark.usefixtures("kion_music_provider") +async def test_library_items(mass: MusicAssistant) -> None: + """Test library artists, albums, tracks, playlists.""" + prov = _get_kion_provider(mass) + assert prov is not None + instance_id = prov.instance_id + + artists = await mass.music.artists.library_items() + kion_artists = [a for a in artists if a.provider == instance_id] + assert len(kion_artists) >= 0 + + albums = await mass.music.albums.library_items() + kion_albums = [a for a in albums if a.provider == instance_id] + assert len(kion_albums) >= 0 + + tracks = await mass.music.tracks.library_items() + kion_tracks = [t for t in tracks if t.provider == instance_id] + assert len(kion_tracks) >= 0 + + playlists = await mass.music.playlists.library_items() + kion_playlists = [p for p in playlists if p.provider == instance_id] + assert len(kion_playlists) >= 0 diff --git a/tests/providers/kion_music/test_parsers.py b/tests/providers/kion_music/test_parsers.py new file mode 100644 index 00000000..2d746a8f --- /dev/null +++ b/tests/providers/kion_music/test_parsers.py @@ -0,0 +1,247 @@ +"""Test we can parse KION Music API objects into Music Assistant models.""" + +from __future__ import annotations + +import json +import pathlib +from typing import TYPE_CHECKING, Any, cast + +import pytest +from yandex_music import Album as YandexAlbum +from yandex_music import Artist as YandexArtist +from yandex_music import Playlist as YandexPlaylist +from yandex_music import Track as YandexTrack + +from music_assistant.providers.kion_music.parsers import ( + parse_album, + parse_artist, + parse_playlist, + parse_track, +) +from music_assistant.providers.kion_music.provider import KionMusicProvider + +from .conftest import DE_JSON_CLIENT + +if TYPE_CHECKING: + from syrupy.assertion import SnapshotAssertion + + from .conftest import ProviderStub + +FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" +ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.json")) +ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.json")) +TRACK_FIXTURES = list(FIXTURES_DIR.glob("tracks/*.json")) +PLAYLIST_FIXTURES = list(FIXTURES_DIR.glob("playlists/*.json")) + + +def _load_json(path: pathlib.Path) -> dict[str, Any]: + """Load JSON fixture.""" + with open(path) as f: + return cast("dict[str, Any]", json.load(f)) + + +def _artist_from_fixture(path: pathlib.Path) -> YandexArtist | None: + """Deserialize Artist from fixture JSON.""" + data = _load_json(path) + return YandexArtist.de_json(data, DE_JSON_CLIENT) + + +def _album_from_fixture(path: pathlib.Path) -> YandexAlbum | None: + """Deserialize Album from fixture JSON.""" + data = _load_json(path) + return YandexAlbum.de_json(data, DE_JSON_CLIENT) + + +def _track_from_fixture(path: pathlib.Path) -> YandexTrack | None: + """Deserialize Track from fixture JSON.""" + data = _load_json(path) + return YandexTrack.de_json(data, DE_JSON_CLIENT) + + +def _playlist_from_fixture(path: pathlib.Path) -> YandexPlaylist | None: + """Deserialize Playlist from fixture JSON.""" + data = _load_json(path) + return YandexPlaylist.de_json(data, DE_JSON_CLIENT) + + +# provider_stub fixture is provided by conftest.py + + +@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_artist(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse artists from fixture JSON.""" + artist_obj = _artist_from_fixture(example) + assert artist_obj is not None + result = parse_artist(cast("KionMusicProvider", provider_stub), artist_obj) + assert result.item_id == str(artist_obj.id) + assert result.name == (artist_obj.name or "Unknown Artist") + assert result.provider == provider_stub.instance_id + assert len(result.provider_mappings) == 1 + mapping = next(iter(result.provider_mappings)) + assert f"music.mts.ru/artist/{artist_obj.id}" in (mapping.url or "") + + +def test_parse_artist_with_cover(provider_stub: ProviderStub) -> None: + """Test parsing artist with cover image.""" + path = FIXTURES_DIR / "artists" / "with_cover.json" + artist_obj = _artist_from_fixture(path) + assert artist_obj is not None + result = parse_artist(cast("KionMusicProvider", provider_stub), artist_obj) + assert result.item_id == "200" + assert result.name == "Artist With Cover" + if artist_obj.cover and artist_obj.cover.uri: + assert result.metadata.images is not None + assert len(result.metadata.images) == 1 + assert "avatars.yandex.net" in (result.metadata.images[0].path or "") + + +@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: val.stem) +def test_parse_album(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse albums from fixture JSON.""" + album_obj = _album_from_fixture(example) + assert album_obj is not None + result = parse_album(cast("KionMusicProvider", provider_stub), album_obj) + assert result.item_id == str(album_obj.id) + assert result.name + assert result.provider == provider_stub.instance_id + mapping = next(iter(result.provider_mappings)) + assert f"music.mts.ru/album/{album_obj.id}" in (mapping.url or "") + if album_obj.year: + assert result.year == album_obj.year + + +@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: val.stem) +def test_parse_track(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse tracks from fixture JSON.""" + track_obj = _track_from_fixture(example) + assert track_obj is not None + result = parse_track(cast("KionMusicProvider", provider_stub), track_obj) + assert result.item_id == str(track_obj.id) + assert result.name + assert result.duration == (track_obj.duration_ms or 0) // 1000 + mapping = next(iter(result.provider_mappings)) + assert f"music.mts.ru/track/{track_obj.id}" in (mapping.url or "") + + +def test_parse_track_with_artist_and_album(provider_stub: ProviderStub) -> None: + """Test parsing track with artist and album.""" + path = FIXTURES_DIR / "tracks" / "with_artist_and_album.json" + track_obj = _track_from_fixture(path) + assert track_obj is not None + result = parse_track(cast("KionMusicProvider", provider_stub), track_obj) + assert result.item_id == "500" + if track_obj.artists: + assert len(result.artists) >= 1 + assert result.artists[0].name == "Track Artist" + if track_obj.albums: + assert result.album is not None + assert result.album.item_id == "20" + assert result.album.name == "Track Album" + + +@pytest.mark.parametrize("example", PLAYLIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_playlist(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse playlists from fixture JSON.""" + playlist_obj = _playlist_from_fixture(example) + assert playlist_obj is not None + result = parse_playlist(cast("KionMusicProvider", provider_stub), playlist_obj) + owner_id = ( + str(playlist_obj.owner.uid) if playlist_obj.owner else str(provider_stub.client.user_id) + ) + kind = str(playlist_obj.kind) + assert result.item_id == f"{owner_id}:{kind}" + assert result.name == (playlist_obj.title or "Unknown Playlist") + mapping = next(iter(result.provider_mappings)) + assert f"music.mts.ru/users/{owner_id}/playlists/{kind}" in (mapping.url or "") + + +def test_parse_playlist_editable(provider_stub: ProviderStub) -> None: + """Test parsing own playlist (editable).""" + path = FIXTURES_DIR / "playlists" / "minimal.json" + playlist_obj = _playlist_from_fixture(path) + assert playlist_obj is not None + result = parse_playlist(cast("KionMusicProvider", provider_stub), playlist_obj) + assert result.owner == "Me" + assert result.is_editable is True + + +def test_parse_playlist_other_user(provider_stub: ProviderStub) -> None: + """Test parsing playlist owned by another user.""" + path = FIXTURES_DIR / "playlists" / "other_user.json" + playlist_obj = _playlist_from_fixture(path) + assert playlist_obj is not None + result = parse_playlist(cast("KionMusicProvider", provider_stub), playlist_obj) + assert result.item_id == "99999:1" + assert result.name == "Shared Playlist" + assert result.owner == "Other User" + assert result.is_editable is False + assert result.metadata.description == "A shared playlist" + + +# --- Snapshot tests --- + + +def _sort_for_snapshot(parsed: dict[str, Any]) -> dict[str, Any]: + """Sort lists in parsed dict for deterministic snapshot comparison.""" + if parsed.get("external_ids"): + parsed["external_ids"] = sorted(parsed["external_ids"]) + if "metadata" in parsed and isinstance(parsed["metadata"], dict): + if parsed["metadata"].get("genres"): + parsed["metadata"]["genres"] = sorted(parsed["metadata"]["genres"]) + return parsed + + +@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_artist_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for artist parsing.""" + artist_obj = _artist_from_fixture(example) + assert artist_obj is not None + result = parse_artist(cast("KionMusicProvider", provider_stub), artist_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: val.stem) +def test_parse_album_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for album parsing.""" + album_obj = _album_from_fixture(example) + assert album_obj is not None + result = parse_album(cast("KionMusicProvider", provider_stub), album_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: val.stem) +def test_parse_track_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for track parsing.""" + track_obj = _track_from_fixture(example) + assert track_obj is not None + result = parse_track(cast("KionMusicProvider", provider_stub), track_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", PLAYLIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_playlist_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for playlist parsing.""" + playlist_obj = _playlist_from_fixture(example) + assert playlist_obj is not None + result = parse_playlist(cast("KionMusicProvider", provider_stub), playlist_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed -- 2.34.1