/
/
/
1"""API client wrapper for Yandex Music."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, Any, cast
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
19from yandex_music.utils.sign_request import get_sign_request
20
21if TYPE_CHECKING:
22 from yandex_music import DownloadInfo
23
24from .constants import DEFAULT_LIMIT
25
26# get-file-info with quality=lossless returns FLAC; default /tracks/.../download-info often does not
27# Prefer flac-mp4/aac-mp4 (Yandex API moved to these formats around 2025)
28GET_FILE_INFO_CODECS = "flac-mp4,flac,aac-mp4,aac,he-aac,mp3,he-aac-mp4"
29# get-file-info: same host as library (all requests go through one API)
30GET_FILE_INFO_BASE_URL = "https://api.music.yandex.net"
31
32LOGGER = logging.getLogger(__name__)
33
34
35class YandexMusicClient:
36 """Wrapper around yandex-music-api ClientAsync."""
37
38 def __init__(self, token: str) -> None:
39 """Initialize the Yandex Music client.
40
41 :param token: Yandex Music OAuth token.
42 """
43 self._token = token
44 self._client: ClientAsync | None = None
45 self._user_id: int | None = None
46
47 @property
48 def user_id(self) -> int:
49 """Return the user ID."""
50 if self._user_id is None:
51 raise ProviderUnavailableError("Client not initialized, call connect() first")
52 return self._user_id
53
54 async def connect(self) -> bool:
55 """Initialize the client and verify token validity.
56
57 :return: True if connection was successful.
58 :raises LoginFailed: If the token is invalid.
59 """
60 try:
61 self._client = await ClientAsync(self._token).init()
62 if self._client.me is None or self._client.me.account is None:
63 raise LoginFailed("Failed to get account info")
64 self._user_id = self._client.me.account.uid
65 LOGGER.debug("Connected to Yandex Music as user %s", self._user_id)
66 return True
67 except UnauthorizedError as err:
68 raise LoginFailed("Invalid Yandex Music token") from err
69 except NetworkError as err:
70 msg = "Network error connecting to Yandex Music"
71 raise ResourceTemporarilyUnavailable(msg) from err
72
73 async def disconnect(self) -> None:
74 """Disconnect the client."""
75 self._client = None
76 self._user_id = None
77
78 def _ensure_connected(self) -> ClientAsync:
79 """Ensure the client is connected and return it."""
80 if self._client is None:
81 raise ProviderUnavailableError("Client not connected, call connect() first")
82 return self._client
83
84 # Library methods
85
86 async def get_liked_tracks(self) -> list[TrackShort]:
87 """Get user's liked tracks.
88
89 :return: List of liked track objects.
90 """
91 client = self._ensure_connected()
92 try:
93 result = await client.users_likes_tracks()
94 if result is None:
95 return []
96 return result.tracks or []
97 except (BadRequestError, NetworkError) as err:
98 LOGGER.error("Error fetching liked tracks: %s", err)
99 raise ResourceTemporarilyUnavailable("Failed to fetch liked tracks") from err
100
101 async def get_liked_albums(self) -> list[YandexAlbum]:
102 """Get user's liked albums.
103
104 :return: List of liked album objects.
105 """
106 client = self._ensure_connected()
107 try:
108 result = await client.users_likes_albums()
109 if result is None:
110 return []
111 return [like.album for like in result if like.album is not None]
112 except (BadRequestError, NetworkError) as err:
113 LOGGER.error("Error fetching liked albums: %s", err)
114 raise ResourceTemporarilyUnavailable("Failed to fetch liked albums") from err
115
116 async def get_liked_artists(self) -> list[YandexArtist]:
117 """Get user's liked artists.
118
119 :return: List of liked artist objects.
120 """
121 client = self._ensure_connected()
122 try:
123 result = await client.users_likes_artists()
124 if result is None:
125 return []
126 return [like.artist for like in result if like.artist is not None]
127 except (BadRequestError, NetworkError) as err:
128 LOGGER.error("Error fetching liked artists: %s", err)
129 raise ResourceTemporarilyUnavailable("Failed to fetch liked artists") from err
130
131 async def get_user_playlists(self) -> list[YandexPlaylist]:
132 """Get user's playlists.
133
134 :return: List of playlist objects.
135 """
136 client = self._ensure_connected()
137 try:
138 result = await client.users_playlists_list()
139 if result is None:
140 return []
141 return list(result)
142 except (BadRequestError, NetworkError) as err:
143 LOGGER.error("Error fetching playlists: %s", err)
144 raise ResourceTemporarilyUnavailable("Failed to fetch playlists") from err
145
146 # Search
147
148 async def search(
149 self,
150 query: str,
151 search_type: str = "all",
152 limit: int = DEFAULT_LIMIT,
153 ) -> Search | None:
154 """Search for tracks, albums, artists, or playlists.
155
156 :param query: Search query string.
157 :param search_type: Type of search ('all', 'track', 'album', 'artist', 'playlist').
158 :param limit: Maximum number of results per type.
159 :return: Search results object.
160 """
161 client = self._ensure_connected()
162 try:
163 return await client.search(query, type_=search_type, page=0, nocorrect=False)
164 except (BadRequestError, NetworkError) as err:
165 LOGGER.error("Search error: %s", err)
166 raise ResourceTemporarilyUnavailable("Search failed") from err
167
168 # Get single items
169
170 async def get_track(self, track_id: str) -> YandexTrack | None:
171 """Get a single track by ID.
172
173 :param track_id: Track ID.
174 :return: Track object or None if not found.
175 """
176 client = self._ensure_connected()
177 try:
178 tracks = await client.tracks([track_id])
179 return tracks[0] if tracks else None
180 except (BadRequestError, NetworkError) as err:
181 LOGGER.error("Error fetching track %s: %s", track_id, err)
182 return None
183
184 async def get_tracks(self, track_ids: list[str]) -> list[YandexTrack]:
185 """Get multiple tracks by IDs.
186
187 :param track_ids: List of track IDs.
188 :return: List of track objects.
189 """
190 client = self._ensure_connected()
191 try:
192 result = await client.tracks(track_ids)
193 return result or []
194 except (BadRequestError, NetworkError) as err:
195 LOGGER.error("Error fetching tracks: %s", err)
196 return []
197
198 async def get_album(self, album_id: str) -> YandexAlbum | None:
199 """Get a single album by ID.
200
201 :param album_id: Album ID.
202 :return: Album object or None if not found.
203 """
204 client = self._ensure_connected()
205 try:
206 albums = await client.albums([album_id])
207 return albums[0] if albums else None
208 except (BadRequestError, NetworkError) as err:
209 LOGGER.error("Error fetching album %s: %s", album_id, err)
210 return None
211
212 async def get_album_with_tracks(self, album_id: str) -> YandexAlbum | None:
213 """Get an album with its tracks.
214
215 Uses the same semantics as the web client: albums/{id}/with-tracks
216 with resumeStream, richTracks, withListeningFinished when the library
217 passes them through.
218
219 :param album_id: Album ID.
220 :return: Album object with tracks or None if not found.
221 """
222 client = self._ensure_connected()
223 try:
224 return await client.albums_with_tracks(
225 album_id,
226 resumeStream=True,
227 richTracks=True,
228 withListeningFinished=True,
229 )
230 except TypeError:
231 # Older yandex-music may not accept these kwargs
232 return await client.albums_with_tracks(album_id)
233 except (BadRequestError, NetworkError) as err:
234 LOGGER.error("Error fetching album with tracks %s: %s", album_id, err)
235 return None
236
237 async def get_artist(self, artist_id: str) -> YandexArtist | None:
238 """Get a single artist by ID.
239
240 :param artist_id: Artist ID.
241 :return: Artist object or None if not found.
242 """
243 client = self._ensure_connected()
244 try:
245 artists = await client.artists([artist_id])
246 return artists[0] if artists else None
247 except (BadRequestError, NetworkError) as err:
248 LOGGER.error("Error fetching artist %s: %s", artist_id, err)
249 return None
250
251 async def get_artist_albums(
252 self, artist_id: str, limit: int = DEFAULT_LIMIT
253 ) -> list[YandexAlbum]:
254 """Get artist's albums.
255
256 :param artist_id: Artist ID.
257 :param limit: Maximum number of albums.
258 :return: List of album objects.
259 """
260 client = self._ensure_connected()
261 try:
262 result = await client.artists_direct_albums(artist_id, page=0, page_size=limit)
263 if result is None:
264 return []
265 return result.albums or []
266 except (BadRequestError, NetworkError) as err:
267 LOGGER.error("Error fetching artist albums %s: %s", artist_id, err)
268 return []
269
270 async def get_artist_tracks(
271 self, artist_id: str, limit: int = DEFAULT_LIMIT
272 ) -> list[YandexTrack]:
273 """Get artist's top tracks.
274
275 :param artist_id: Artist ID.
276 :param limit: Maximum number of tracks.
277 :return: List of track objects.
278 """
279 client = self._ensure_connected()
280 try:
281 result = await client.artists_tracks(artist_id, page=0, page_size=limit)
282 if result is None:
283 return []
284 return result.tracks or []
285 except (BadRequestError, NetworkError) as err:
286 LOGGER.error("Error fetching artist tracks %s: %s", artist_id, err)
287 return []
288
289 async def get_playlist(self, user_id: str, playlist_id: str) -> YandexPlaylist | None:
290 """Get a playlist by ID.
291
292 :param user_id: User ID (owner of the playlist).
293 :param playlist_id: Playlist ID (kind).
294 :return: Playlist object or None if not found.
295 """
296 client = self._ensure_connected()
297 try:
298 result = await client.users_playlists(kind=int(playlist_id), user_id=user_id)
299 if isinstance(result, list):
300 return result[0] if result else None
301 return result
302 except (BadRequestError, NetworkError) as err:
303 LOGGER.error("Error fetching playlist %s/%s: %s", user_id, playlist_id, err)
304 return None
305
306 # Streaming
307
308 async def get_track_download_info(
309 self, track_id: str, get_direct_links: bool = True
310 ) -> list[DownloadInfo]:
311 """Get download info for a track.
312
313 :param track_id: Track ID.
314 :param get_direct_links: Whether to get direct download links.
315 :return: List of download info objects.
316 """
317 client = self._ensure_connected()
318 try:
319 result = await client.tracks_download_info(track_id, get_direct_links=get_direct_links)
320 return result or []
321 except (BadRequestError, NetworkError) as err:
322 LOGGER.error("Error fetching download info for track %s: %s", track_id, err)
323 return []
324
325 async def get_track_file_info_lossless(self, track_id: str) -> dict[str, Any] | None:
326 """Request lossless stream via get-file-info (quality=lossless).
327
328 The /tracks/{id}/download-info endpoint often returns only MP3; get-file-info
329 with quality=lossless and codecs=flac,... returns FLAC when available.
330
331 :param track_id: Track ID.
332 :return: Parsed downloadInfo dict (url, codec, urls, ...) or None on error.
333 """
334 client = self._ensure_connected()
335 sign = get_sign_request(track_id)
336 base_params = {
337 "ts": sign.timestamp,
338 "trackId": track_id,
339 "quality": "lossless",
340 "codecs": GET_FILE_INFO_CODECS,
341 "sign": sign.value,
342 }
343
344 def _parse_file_info_result(raw: dict[str, Any] | None) -> dict[str, Any] | None:
345 if not raw or not isinstance(raw, dict):
346 return None
347 download_info = raw.get("download_info")
348 if not download_info or not download_info.get("url"):
349 return None
350 return cast("dict[str, Any]", download_info)
351
352 url = f"{GET_FILE_INFO_BASE_URL}/get-file-info"
353 params_encraw = {**base_params, "transports": "encraw"}
354 try:
355 result = await client._request.get(url, params=params_encraw)
356 return _parse_file_info_result(result)
357 except (BadRequestError, NetworkError) as err:
358 LOGGER.debug(
359 "get-file-info lossless for track %s: %s %s",
360 track_id,
361 type(err).__name__,
362 getattr(err, "message", str(err)) or repr(err),
363 )
364 return None
365 except UnauthorizedError as err:
366 LOGGER.debug(
367 "get-file-info lossless for track %s (transports=encraw): %s %s",
368 track_id,
369 type(err).__name__,
370 getattr(err, "message", str(err)) or repr(err),
371 )
372 LOGGER.debug(
373 "If you have Yandex Music Plus and this track has lossless, "
374 "try a token from the web client (music.yandex.ru)."
375 )
376 params_raw = {**base_params, "transports": "raw"}
377 try:
378 result = await client._request.get(url, params=params_raw)
379 return _parse_file_info_result(result)
380 except (BadRequestError, NetworkError, UnauthorizedError) as retry_err:
381 LOGGER.debug(
382 "get-file-info lossless for track %s (transports=raw): %s %s",
383 track_id,
384 type(retry_err).__name__,
385 getattr(retry_err, "message", str(retry_err)) or repr(retry_err),
386 )
387 return None
388
389 # Library modifications
390
391 async def like_track(self, track_id: str) -> bool:
392 """Add a track to liked tracks.
393
394 :param track_id: Track ID to like.
395 :return: True if successful.
396 """
397 client = self._ensure_connected()
398 try:
399 result = await client.users_likes_tracks_add(track_id)
400 return result is not None
401 except (BadRequestError, NetworkError) as err:
402 LOGGER.error("Error liking track %s: %s", track_id, err)
403 return False
404
405 async def unlike_track(self, track_id: str) -> bool:
406 """Remove a track from liked tracks.
407
408 :param track_id: Track ID to unlike.
409 :return: True if successful.
410 """
411 client = self._ensure_connected()
412 try:
413 result = await client.users_likes_tracks_remove(track_id)
414 return result is not None
415 except (BadRequestError, NetworkError) as err:
416 LOGGER.error("Error unliking track %s: %s", track_id, err)
417 return False
418
419 async def like_album(self, album_id: str) -> bool:
420 """Add an album to liked albums.
421
422 :param album_id: Album ID to like.
423 :return: True if successful.
424 """
425 client = self._ensure_connected()
426 try:
427 result = await client.users_likes_albums_add(album_id)
428 return result is not None
429 except (BadRequestError, NetworkError) as err:
430 LOGGER.error("Error liking album %s: %s", album_id, err)
431 return False
432
433 async def unlike_album(self, album_id: str) -> bool:
434 """Remove an album from liked albums.
435
436 :param album_id: Album ID to unlike.
437 :return: True if successful.
438 """
439 client = self._ensure_connected()
440 try:
441 result = await client.users_likes_albums_remove(album_id)
442 return result is not None
443 except (BadRequestError, NetworkError) as err:
444 LOGGER.error("Error unliking album %s: %s", album_id, err)
445 return False
446
447 async def like_artist(self, artist_id: str) -> bool:
448 """Add an artist to liked artists.
449
450 :param artist_id: Artist ID to like.
451 :return: True if successful.
452 """
453 client = self._ensure_connected()
454 try:
455 result = await client.users_likes_artists_add(artist_id)
456 return result is not None
457 except (BadRequestError, NetworkError) as err:
458 LOGGER.error("Error liking artist %s: %s", artist_id, err)
459 return False
460
461 async def unlike_artist(self, artist_id: str) -> bool:
462 """Remove an artist from liked artists.
463
464 :param artist_id: Artist ID to unlike.
465 :return: True if successful.
466 """
467 client = self._ensure_connected()
468 try:
469 result = await client.users_likes_artists_remove(artist_id)
470 return result is not None
471 except (BadRequestError, NetworkError) as err:
472 LOGGER.error("Error unliking artist %s: %s", artist_id, err)
473 return False
474