music-assistant-server

1"""API client wrapper for Yandex Music."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING
7
8from music_assistant_models.errors import (
9    LoginFailed,
10    ProviderUnavailableError,
11    ResourceTemporarilyUnavailable,
12)
13from yandex_music import Album as YandexAlbum
14from yandex_music import Artist as YandexArtist
15from yandex_music import ClientAsync, Search, TrackShort
16from yandex_music import Playlist as YandexPlaylist
17from yandex_music import Track as YandexTrack
18from yandex_music.exceptions import BadRequestError, NetworkError, UnauthorizedError
19
20if TYPE_CHECKING:
21    from yandex_music import DownloadInfo
22
23from .constants import DEFAULT_LIMIT
24
25LOGGER = logging.getLogger(__name__)
26
27
28class YandexMusicClient:
29    """Wrapper around yandex-music-api ClientAsync."""
30
31    def __init__(self, token: str) -> None:
32        """Initialize the Yandex Music client.
33
34        :param token: Yandex Music OAuth token.
35        """
36        self._token = token
37        self._client: ClientAsync | None = None
38        self._user_id: int | None = None
39
40    @property
41    def user_id(self) -> int:
42        """Return the user ID."""
43        if self._user_id is None:
44            raise ProviderUnavailableError("Client not initialized, call connect() first")
45        return self._user_id
46
47    async def connect(self) -> bool:
48        """Initialize the client and verify token validity.
49
50        :return: True if connection was successful.
51        :raises LoginFailed: If the token is invalid.
52        """
53        try:
54            self._client = await ClientAsync(self._token).init()
55            if self._client.me is None or self._client.me.account is None:
56                raise LoginFailed("Failed to get account info")
57            self._user_id = self._client.me.account.uid
58            LOGGER.debug("Connected to Yandex Music as user %s", self._user_id)
59            return True
60        except UnauthorizedError as err:
61            raise LoginFailed("Invalid Yandex Music token") from err
62        except NetworkError as err:
63            msg = "Network error connecting to Yandex Music"
64            raise ResourceTemporarilyUnavailable(msg) from err
65
66    async def disconnect(self) -> None:
67        """Disconnect the client."""
68        self._client = None
69        self._user_id = None
70
71    def _ensure_connected(self) -> ClientAsync:
72        """Ensure the client is connected and return it."""
73        if self._client is None:
74            raise ProviderUnavailableError("Client not connected, call connect() first")
75        return self._client
76
77    # Library methods
78
79    async def get_liked_tracks(self) -> list[TrackShort]:
80        """Get user's liked tracks.
81
82        :return: List of liked track objects.
83        """
84        client = self._ensure_connected()
85        try:
86            result = await client.users_likes_tracks()
87            if result is None:
88                return []
89            return result.tracks or []
90        except (BadRequestError, NetworkError) as err:
91            LOGGER.error("Error fetching liked tracks: %s", err)
92            raise ResourceTemporarilyUnavailable("Failed to fetch liked tracks") from err
93
94    async def get_liked_albums(self) -> list[YandexAlbum]:
95        """Get user's liked albums.
96
97        :return: List of liked album objects.
98        """
99        client = self._ensure_connected()
100        try:
101            result = await client.users_likes_albums()
102            if result is None:
103                return []
104            return [like.album for like in result if like.album is not None]
105        except (BadRequestError, NetworkError) as err:
106            LOGGER.error("Error fetching liked albums: %s", err)
107            raise ResourceTemporarilyUnavailable("Failed to fetch liked albums") from err
108
109    async def get_liked_artists(self) -> list[YandexArtist]:
110        """Get user's liked artists.
111
112        :return: List of liked artist objects.
113        """
114        client = self._ensure_connected()
115        try:
116            result = await client.users_likes_artists()
117            if result is None:
118                return []
119            return [like.artist for like in result if like.artist is not None]
120        except (BadRequestError, NetworkError) as err:
121            LOGGER.error("Error fetching liked artists: %s", err)
122            raise ResourceTemporarilyUnavailable("Failed to fetch liked artists") from err
123
124    async def get_user_playlists(self) -> list[YandexPlaylist]:
125        """Get user's playlists.
126
127        :return: List of playlist objects.
128        """
129        client = self._ensure_connected()
130        try:
131            result = await client.users_playlists_list()
132            if result is None:
133                return []
134            return list(result)
135        except (BadRequestError, NetworkError) as err:
136            LOGGER.error("Error fetching playlists: %s", err)
137            raise ResourceTemporarilyUnavailable("Failed to fetch playlists") from err
138
139    # Search
140
141    async def search(
142        self,
143        query: str,
144        search_type: str = "all",
145        limit: int = DEFAULT_LIMIT,
146    ) -> Search | None:
147        """Search for tracks, albums, artists, or playlists.
148
149        :param query: Search query string.
150        :param search_type: Type of search ('all', 'track', 'album', 'artist', 'playlist').
151        :param limit: Maximum number of results per type.
152        :return: Search results object.
153        """
154        client = self._ensure_connected()
155        try:
156            return await client.search(query, type_=search_type, page=0, nocorrect=False)
157        except (BadRequestError, NetworkError) as err:
158            LOGGER.error("Search error: %s", err)
159            raise ResourceTemporarilyUnavailable("Search failed") from err
160
161    # Get single items
162
163    async def get_track(self, track_id: str) -> YandexTrack | None:
164        """Get a single track by ID.
165
166        :param track_id: Track ID.
167        :return: Track object or None if not found.
168        """
169        client = self._ensure_connected()
170        try:
171            tracks = await client.tracks([track_id])
172            return tracks[0] if tracks else None
173        except (BadRequestError, NetworkError) as err:
174            LOGGER.error("Error fetching track %s: %s", track_id, err)
175            return None
176
177    async def get_tracks(self, track_ids: list[str]) -> list[YandexTrack]:
178        """Get multiple tracks by IDs.
179
180        :param track_ids: List of track IDs.
181        :return: List of track objects.
182        """
183        client = self._ensure_connected()
184        try:
185            result = await client.tracks(track_ids)
186            return result or []
187        except (BadRequestError, NetworkError) as err:
188            LOGGER.error("Error fetching tracks: %s", err)
189            return []
190
191    async def get_album(self, album_id: str) -> YandexAlbum | None:
192        """Get a single album by ID.
193
194        :param album_id: Album ID.
195        :return: Album object or None if not found.
196        """
197        client = self._ensure_connected()
198        try:
199            albums = await client.albums([album_id])
200            return albums[0] if albums else None
201        except (BadRequestError, NetworkError) as err:
202            LOGGER.error("Error fetching album %s: %s", album_id, err)
203            return None
204
205    async def get_album_with_tracks(self, album_id: str) -> YandexAlbum | None:
206        """Get an album with its tracks.
207
208        :param album_id: Album ID.
209        :return: Album object with tracks or None if not found.
210        """
211        client = self._ensure_connected()
212        try:
213            return await client.albums_with_tracks(album_id)
214        except (BadRequestError, NetworkError) as err:
215            LOGGER.error("Error fetching album with tracks %s: %s", album_id, err)
216            return None
217
218    async def get_artist(self, artist_id: str) -> YandexArtist | None:
219        """Get a single artist by ID.
220
221        :param artist_id: Artist ID.
222        :return: Artist object or None if not found.
223        """
224        client = self._ensure_connected()
225        try:
226            artists = await client.artists([artist_id])
227            return artists[0] if artists else None
228        except (BadRequestError, NetworkError) as err:
229            LOGGER.error("Error fetching artist %s: %s", artist_id, err)
230            return None
231
232    async def get_artist_albums(
233        self, artist_id: str, limit: int = DEFAULT_LIMIT
234    ) -> list[YandexAlbum]:
235        """Get artist's albums.
236
237        :param artist_id: Artist ID.
238        :param limit: Maximum number of albums.
239        :return: List of album objects.
240        """
241        client = self._ensure_connected()
242        try:
243            result = await client.artists_direct_albums(artist_id, page=0, page_size=limit)
244            if result is None:
245                return []
246            return result.albums or []
247        except (BadRequestError, NetworkError) as err:
248            LOGGER.error("Error fetching artist albums %s: %s", artist_id, err)
249            return []
250
251    async def get_artist_tracks(
252        self, artist_id: str, limit: int = DEFAULT_LIMIT
253    ) -> list[YandexTrack]:
254        """Get artist's top tracks.
255
256        :param artist_id: Artist ID.
257        :param limit: Maximum number of tracks.
258        :return: List of track objects.
259        """
260        client = self._ensure_connected()
261        try:
262            result = await client.artists_tracks(artist_id, page=0, page_size=limit)
263            if result is None:
264                return []
265            return result.tracks or []
266        except (BadRequestError, NetworkError) as err:
267            LOGGER.error("Error fetching artist tracks %s: %s", artist_id, err)
268            return []
269
270    async def get_playlist(self, user_id: str, playlist_id: str) -> YandexPlaylist | None:
271        """Get a playlist by ID.
272
273        :param user_id: User ID (owner of the playlist).
274        :param playlist_id: Playlist ID (kind).
275        :return: Playlist object or None if not found.
276        """
277        client = self._ensure_connected()
278        try:
279            result = await client.users_playlists(kind=int(playlist_id), user_id=user_id)
280            if isinstance(result, list):
281                return result[0] if result else None
282            return result
283        except (BadRequestError, NetworkError) as err:
284            LOGGER.error("Error fetching playlist %s/%s: %s", user_id, playlist_id, err)
285            return None
286
287    # Streaming
288
289    async def get_track_download_info(
290        self, track_id: str, get_direct_links: bool = True
291    ) -> list[DownloadInfo]:
292        """Get download info for a track.
293
294        :param track_id: Track ID.
295        :param get_direct_links: Whether to get direct download links.
296        :return: List of download info objects.
297        """
298        client = self._ensure_connected()
299        try:
300            result = await client.tracks_download_info(track_id, get_direct_links=get_direct_links)
301            return result or []
302        except (BadRequestError, NetworkError) as err:
303            LOGGER.error("Error fetching download info for track %s: %s", track_id, err)
304            return []
305
306    # Library modifications
307
308    async def like_track(self, track_id: str) -> bool:
309        """Add a track to liked tracks.
310
311        :param track_id: Track ID to like.
312        :return: True if successful.
313        """
314        client = self._ensure_connected()
315        try:
316            result = await client.users_likes_tracks_add(track_id)
317            return result is not None
318        except (BadRequestError, NetworkError) as err:
319            LOGGER.error("Error liking track %s: %s", track_id, err)
320            return False
321
322    async def unlike_track(self, track_id: str) -> bool:
323        """Remove a track from liked tracks.
324
325        :param track_id: Track ID to unlike.
326        :return: True if successful.
327        """
328        client = self._ensure_connected()
329        try:
330            result = await client.users_likes_tracks_remove(track_id)
331            return result is not None
332        except (BadRequestError, NetworkError) as err:
333            LOGGER.error("Error unliking track %s: %s", track_id, err)
334            return False
335
336    async def like_album(self, album_id: str) -> bool:
337        """Add an album to liked albums.
338
339        :param album_id: Album ID to like.
340        :return: True if successful.
341        """
342        client = self._ensure_connected()
343        try:
344            result = await client.users_likes_albums_add(album_id)
345            return result is not None
346        except (BadRequestError, NetworkError) as err:
347            LOGGER.error("Error liking album %s: %s", album_id, err)
348            return False
349
350    async def unlike_album(self, album_id: str) -> bool:
351        """Remove an album from liked albums.
352
353        :param album_id: Album ID to unlike.
354        :return: True if successful.
355        """
356        client = self._ensure_connected()
357        try:
358            result = await client.users_likes_albums_remove(album_id)
359            return result is not None
360        except (BadRequestError, NetworkError) as err:
361            LOGGER.error("Error unliking album %s: %s", album_id, err)
362            return False
363
364    async def like_artist(self, artist_id: str) -> bool:
365        """Add an artist to liked artists.
366
367        :param artist_id: Artist ID to like.
368        :return: True if successful.
369        """
370        client = self._ensure_connected()
371        try:
372            result = await client.users_likes_artists_add(artist_id)
373            return result is not None
374        except (BadRequestError, NetworkError) as err:
375            LOGGER.error("Error liking artist %s: %s", artist_id, err)
376            return False
377
378    async def unlike_artist(self, artist_id: str) -> bool:
379        """Remove an artist from liked artists.
380
381        :param artist_id: Artist ID to unlike.
382        :return: True if successful.
383        """
384        client = self._ensure_connected()
385        try:
386            result = await client.users_likes_artists_remove(artist_id)
387            return result is not None
388        except (BadRequestError, NetworkError) as err:
389            LOGGER.error("Error unliking artist %s: %s", artist_id, err)
390            return False
391