From 6d7f831b48de5fa840176b72f8f0bc2c000733dc Mon Sep 17 00:00:00 2001 From: Mikhail Nevskiy <139659391+trudenboy@users.noreply.github.com> Date: Wed, 28 Jan 2026 15:30:23 +0300 Subject: [PATCH] Add Yandex Music provider (#3002) MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Add Yandex Music provider Implement a new music provider for Yandex Music streaming service using the unofficial yandex-music-api library. Features include: - Library sync (artists, albums, tracks, playlists) - Library editing (add/remove tracks, albums, artists) - Search across all media types - Artist albums and top tracks - HTTP streaming with quality selection (320kbps MP3 / FLAC) - Token-based authentication Co-Authored-By: Claude Opus 4.5 * Fix Yandex Music provider config entries Remove duplicate token field that was causing the Save button to remain disabled during provider setup. Co-Authored-By: Claude Opus 4.5 * Fix Yandex Music provider based on PR review feedback - Move SUPPORTED_FEATURES to __init__.py and pass to setup() - Remove code duplication in get_config_entries - Remove label_instructions (use documentation instead) - Replace RuntimeError with ProviderUnavailableError - Replace generic Exception catches with InvalidDataError - Remove non-existent AlbumType.PODCAST - Fix audio_format to respect quality config setting (MP3/FLAC) Co-Authored-By: Claude Opus 4.5 * Add yandex-music to requirements_all.txt Co-Authored-By: Claude Opus 4.5 * Update manifest.json * Improve Yandex Music provider based on PR review - Use ContentType.UNKNOWN for unknown codecs instead of assuming MP3 - Add PLAYLIST_ID_SPLITTER constant and remove unused cache TTL constants - Optimize search to use specific type when only one media type requested - Remove unused imports Co-Authored-By: Claude Opus 4.5 * Fix mypy type errors in Yandex Music provider - Add cast for users_playlists_list return type in api_client.py - Add type annotations for _select_best_quality method in streaming.py - Convert quality config value to str | None before passing to method - Remove non-existent MediaItemMetadata attributes (track_count, has_lyrics) - Use spread operator instead of append for Sequence types in search results Co-Authored-By: Claude Opus 4.5 * Enable CI workflow for feature branches Allow test workflow to run on feature/* branches to validate changes before pushing to upstream PR. Co-Authored-By: Claude Opus 4.5 * Fix ruff TC006 error - use list() instead of cast() Co-Authored-By: Claude Opus 4.5 * Remove CI workflow changes unrelated to Yandex Music provider Reverts changes to .github/workflows/test.yml as requested in PR review. Co-Authored-By: Claude Opus 4.5 * Add Yandex Music provider icons Co-Authored-By: Claude Opus 4.5 * Update music_assistant/providers/yandex_music/manifest.json Co-authored-by: OzGav --------- Co-authored-by: Михаил Невский Co-authored-by: Claude Opus 4.5 Co-authored-by: Marvin Schenkel Co-authored-by: OzGav --- .../providers/yandex_music/__init__.py | 96 ++++ .../providers/yandex_music/api_client.py | 390 ++++++++++++++++ .../providers/yandex_music/constants.py | 32 ++ .../providers/yandex_music/icon.svg | 3 + .../yandex_music/icon_monochrome.svg | 3 + .../providers/yandex_music/manifest.json | 11 + .../providers/yandex_music/parsers.py | 370 +++++++++++++++ .../providers/yandex_music/provider.py | 420 ++++++++++++++++++ .../providers/yandex_music/streaming.py | 121 +++++ requirements_all.txt | 1 + 10 files changed, 1447 insertions(+) create mode 100644 music_assistant/providers/yandex_music/__init__.py create mode 100644 music_assistant/providers/yandex_music/api_client.py create mode 100644 music_assistant/providers/yandex_music/constants.py create mode 100644 music_assistant/providers/yandex_music/icon.svg create mode 100644 music_assistant/providers/yandex_music/icon_monochrome.svg create mode 100644 music_assistant/providers/yandex_music/manifest.json create mode 100644 music_assistant/providers/yandex_music/parsers.py create mode 100644 music_assistant/providers/yandex_music/provider.py create mode 100644 music_assistant/providers/yandex_music/streaming.py diff --git a/music_assistant/providers/yandex_music/__init__.py b/music_assistant/providers/yandex_music/__init__.py new file mode 100644 index 00000000..b179e992 --- /dev/null +++ b/music_assistant/providers/yandex_music/__init__.py @@ -0,0 +1,96 @@ +"""Yandex 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 YandexMusicProvider + +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 YandexMusicProvider(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="Yandex Music Token", + description="Enter your Yandex 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/yandex_music/api_client.py b/music_assistant/providers/yandex_music/api_client.py new file mode 100644 index 00000000..65e1c321 --- /dev/null +++ b/music_assistant/providers/yandex_music/api_client.py @@ -0,0 +1,390 @@ +"""API client wrapper for Yandex Music.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +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 + +if TYPE_CHECKING: + from yandex_music import DownloadInfo + +from .constants import DEFAULT_LIMIT + +LOGGER = logging.getLogger(__name__) + + +class YandexMusicClient: + """Wrapper around yandex-music-api ClientAsync.""" + + def __init__(self, token: str) -> None: + """Initialize the Yandex Music client. + + :param token: Yandex 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).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 Yandex Music as user %s", self._user_id) + return True + except UnauthorizedError as err: + raise LoginFailed("Invalid Yandex Music token") from err + except NetworkError as err: + msg = "Network error connecting to Yandex 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. + + :return: List of liked album objects. + """ + client = self._ensure_connected() + try: + result = await client.users_likes_albums() + if result is None: + return [] + return [like.album for like in result if like.album is not None] + 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. + """ + client = self._ensure_connected() + try: + result = await client.tracks(track_ids) + return result or [] + except (BadRequestError, NetworkError) 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. + + :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) + 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 [] + + # 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/yandex_music/constants.py b/music_assistant/providers/yandex_music/constants.py new file mode 100644 index 00000000..b82796c2 --- /dev/null +++ b/music_assistant/providers/yandex_music/constants.py @@ -0,0 +1,32 @@ +"""Constants for the Yandex 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/yandex_music/icon.svg b/music_assistant/providers/yandex_music/icon.svg new file mode 100644 index 00000000..2eb3d2aa --- /dev/null +++ b/music_assistant/providers/yandex_music/icon.svg @@ -0,0 +1,3 @@ + + + diff --git a/music_assistant/providers/yandex_music/icon_monochrome.svg b/music_assistant/providers/yandex_music/icon_monochrome.svg new file mode 100644 index 00000000..d68fc6bc --- /dev/null +++ b/music_assistant/providers/yandex_music/icon_monochrome.svg @@ -0,0 +1,3 @@ + + + diff --git a/music_assistant/providers/yandex_music/manifest.json b/music_assistant/providers/yandex_music/manifest.json new file mode 100644 index 00000000..4fdbad63 --- /dev/null +++ b/music_assistant/providers/yandex_music/manifest.json @@ -0,0 +1,11 @@ +{ + "type": "music", + "domain": "yandex_music", + "stage": "beta", + "name": "Yandex Music", + "description": "Stream music from Yandex Music service.", + "codeowners": ["@TrudenBoy"], + "documentation": "https://music-assistant.io/music-providers/yandex/", + "requirements": ["yandex-music==2.2.0"], + "multi_instance": true +} diff --git a/music_assistant/providers/yandex_music/parsers.py b/music_assistant/providers/yandex_music/parsers.py new file mode 100644 index 00000000..7c6d5202 --- /dev/null +++ b/music_assistant/providers/yandex_music/parsers.py @@ -0,0 +1,370 @@ +"""Parsers for Yandex 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 YandexMusicProvider + + +def _get_content_type(provider: YandexMusicProvider) -> ContentType: + """Get content type based on provider quality setting. + + :param provider: The Yandex 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 Yandex cover URI to full URL. + + :param cover_uri: Yandex 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: YandexMusicProvider, artist_obj: YandexArtist) -> Artist: + """Parse Yandex artist object to MA Artist model. + + :param provider: The Yandex Music provider instance. + :param artist_obj: Yandex 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.yandex.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: YandexMusicProvider, album_obj: YandexAlbum) -> Album: + """Parse Yandex album object to MA Album model. + + :param provider: The Yandex Music provider instance. + :param album_obj: Yandex 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.yandex.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: YandexMusicProvider, track_obj: YandexTrack) -> Track: + """Parse Yandex track object to MA Track model. + + :param provider: The Yandex Music provider instance. + :param track_obj: Yandex 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 in Yandex API + 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.yandex.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 (minimal data) + if track_obj.albums and len(track_obj.albums) > 0: + album = track_obj.albums[0] + track.album = provider.get_item_mapping( + media_type="album", + key=str(album.id), + name=album.title or "Unknown Album", + ) + # Get image from album if available + if album.cover_uri: + image_url = _get_image_url(album.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: YandexMusicProvider, playlist_obj: YandexPlaylist, owner_name: str | None = None +) -> Playlist: + """Parse Yandex playlist object to MA Playlist model. + + :param provider: The Yandex Music provider instance. + :param playlist_obj: Yandex playlist object. + :param owner_name: Optional owner name override. + :return: Music Assistant Playlist model. + """ + # Playlist ID in Yandex 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 = "Yandex 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.yandex.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/yandex_music/provider.py b/music_assistant/providers/yandex_music/provider.py new file mode 100644 index 00000000..36185433 --- /dev/null +++ b/music_assistant/providers/yandex_music/provider.py @@ -0,0 +1,420 @@ +"""Yandex 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, +) +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 YandexMusicClient +from .constants import CONF_TOKEN, PLAYLIST_ID_SPLITTER +from .parsers import parse_album, parse_artist, parse_playlist, parse_track +from .streaming import YandexMusicStreamingManager + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + from music_assistant_models.streamdetails import StreamDetails + + +class YandexMusicProvider(MusicProvider): + """Implementation of a Yandex Music MusicProvider.""" + + _client: YandexMusicClient | None = None + _streaming: YandexMusicStreamingManager | None = None + + @property + def client(self) -> YandexMusicClient: + """Return the Yandex Music client.""" + if self._client is None: + raise ProviderUnavailableError("Provider not initialized") + return self._client + + @property + def streaming(self) -> YandexMusicStreamingManager: + """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 Yandex Music token provided") + + self._client = YandexMusicClient(str(token)) + await self._client.connect() + self._streaming = YandexMusicStreamingManager(self) + self.logger.info("Successfully connected to Yandex 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 Yandex 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 Yandex 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. + """ + track = await self.client.get_track(prov_track_id) + if not track: + raise MediaNotFoundError(f"Track {prov_track_id} not found") + return parse_track(self, 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. + """ + # 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 or not playlist.tracks: + return [] + + # Yandex 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 playlist.tracks + if track + ] + + if not track_ids: + return [] + + # Fetch full track details + full_tracks = await self.client.get_tracks(track_ids) + 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) + 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 Yandex 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 Yandex 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 Yandex 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 Yandex 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/yandex_music/streaming.py b/music_assistant/providers/yandex_music/streaming.py new file mode 100644 index 00000000..852bf8d3 --- /dev/null +++ b/music_assistant/providers/yandex_music/streaming.py @@ -0,0 +1,121 @@ +"""Streaming operations for Yandex 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 QUALITY_LOSSLESS + +if TYPE_CHECKING: + from yandex_music import DownloadInfo + + from .provider import YandexMusicProvider + + +class YandexMusicStreamingManager: + """Manages Yandex Music streaming operations.""" + + def __init__(self, provider: YandexMusicProvider) -> None: + """Initialize streaming manager. + + :param provider: The Yandex 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") + + # Get download info + 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}") + + # Select best quality based on config + quality = self.provider.config.get_value("quality") + quality_str = str(quality) if quality is not None else None + 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}") + + # Determine content type + 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 setting. + :return: Best matching DownloadInfo or None. + """ + if not download_infos: + return None + + # Sort by bitrate descending + sorted_infos = sorted( + download_infos, + key=lambda x: x.bitrate_in_kbps or 0, + reverse=True, + ) + + # If user wants lossless, try to find FLAC first + if preferred_quality == QUALITY_LOSSLESS: + for info in sorted_infos: + if info.codec and info.codec.lower() == "flac": + return info + + # 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 Yandex API. + :return: ContentType enum value. + """ + if not codec: + return ContentType.UNKNOWN + + codec_lower = codec.lower() + if codec_lower == "flac": + 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/requirements_all.txt b/requirements_all.txt index 4f14d08e..a9523c44 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -75,5 +75,6 @@ unidecode==1.4.0 uv>=0.8.0 websocket-client==1.9.0 xmltodict==1.0.2 +yandex-music==2.2.0 ytmusicapi==1.11.3 zeroconf==0.148.0 -- 2.34.1