music-assistant-server
13.9 KB•PY
api_client.py
13.9 KB • 391 lines • python
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