--- /dev/null
+"""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,
+ ),
+ )
--- /dev/null
+"""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
--- /dev/null
+"""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] = ":"
--- /dev/null
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 144 144">
+ <path fill="#FFBC0D" d="m130.863 57.739-.468-2.327-19.788-3.457 11.498-15.557-1.337-1.462-16.913 8.11 2.139-21.54-1.738-.997-10.295 17.418L82.395 12H80.39l2.74 25.064-29.08-23.269-2.474.732 22.396 28.122-44.323-14.76-2.006 2.261L67.22 52.686l-54.618 4.521-.602 3.39 56.757 6.184-47.33 39.157 2.005 2.726 56.356-30.648-11.164 53.983h3.41l21.592-50.792 13.17 39.756 2.34-1.795-5.415-40.42 20.524 23.268 1.337-2.128-15.711-28.853 21.928 8.111.201-2.46-19.655-14.493 18.518-4.454Z"/>
+</svg>
--- /dev/null
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 144 144">
+ <path fill="#FFFFFF" d="m130.863 57.739-.468-2.327-19.788-3.457 11.498-15.557-1.337-1.462-16.913 8.11 2.139-21.54-1.738-.997-10.295 17.418L82.395 12H80.39l2.74 25.064-29.08-23.269-2.474.732 22.396 28.122-44.323-14.76-2.006 2.261L67.22 52.686l-54.618 4.521-.602 3.39 56.757 6.184-47.33 39.157 2.005 2.726 56.356-30.648-11.164 53.983h3.41l21.592-50.792 13.17 39.756 2.34-1.795-5.415-40.42 20.524 23.268 1.337-2.128-15.711-28.853 21.928 8.111.201-2.46-19.655-14.493 18.518-4.454Z"/>
+</svg>
--- /dev/null
+{
+ "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
+}
--- /dev/null
+"""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
--- /dev/null
+"""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)
--- /dev/null
+"""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
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
projects / music-assistant-server.git / commitdiff
