/
/
/
1"""API client wrapper for Zvuk Music."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import Awaitable, Callable
7from typing import Any, ParamSpec, TypeVar, cast
8
9from music_assistant_models.errors import (
10 LoginFailed,
11 ProviderUnavailableError,
12 ResourceTemporarilyUnavailable,
13)
14from zvuk_music import Artist as ZvukArtist
15from zvuk_music import ClientAsync, Collection
16from zvuk_music import CollectionItem as ZvukCollectionItem
17from zvuk_music import Playlist as ZvukPlaylist
18from zvuk_music import Release as ZvukRelease
19from zvuk_music import Search as ZvukSearch
20from zvuk_music import SimpleTrack as ZvukSimpleTrack
21from zvuk_music import Stream as ZvukStream
22from zvuk_music import Track as ZvukTrack
23from zvuk_music.exceptions import (
24 BadRequestError,
25 BotDetectedError,
26 GraphQLError,
27 NetworkError,
28 NotFoundError,
29 TimedOutError,
30 UnauthorizedError,
31)
32
33from .constants import DEFAULT_LIMIT
34
35LOGGER = logging.getLogger(__name__)
36
37_P = ParamSpec("_P")
38_R = TypeVar("_R")
39_NOT_FOUND_SENTINEL: Any = object()
40
41
42def handle_zvuk_errors(
43 not_found_return: Any = _NOT_FOUND_SENTINEL,
44) -> Callable[[Callable[_P, Awaitable[_R]]], Callable[_P, Awaitable[_R]]]:
45 """Decorate async methods to map Zvuk API exceptions to MA errors.
46
47 :param not_found_return: Value to return on NotFoundError (e.g. None or []).
48 If not provided, NotFoundError is not caught.
49 """
50
51 def decorator(func: Callable[_P, Awaitable[_R]]) -> Callable[_P, Awaitable[_R]]:
52 async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
53 try:
54 return await func(*args, **kwargs)
55 except UnauthorizedError as err:
56 raise LoginFailed("Invalid Zvuk Music token") from err
57 except (NetworkError, TimedOutError) as err:
58 LOGGER.error("Zvuk API error: %s", err)
59 raise ResourceTemporarilyUnavailable("Zvuk Music request failed") from err
60 except (BadRequestError, GraphQLError) as err:
61 LOGGER.error("Zvuk API error: %s", err)
62 raise ResourceTemporarilyUnavailable("Zvuk Music request failed") from err
63 except BotDetectedError as err:
64 raise ProviderUnavailableError("Bot detected by Zvuk") from err
65 except NotFoundError:
66 if not_found_return is _NOT_FOUND_SENTINEL:
67 raise
68 return cast("_R", not_found_return)
69
70 return wrapper
71
72 return decorator
73
74
75class ZvukMusicClient:
76 """Wrapper around zvuk-music ClientAsync."""
77
78 def __init__(self, token: str) -> None:
79 """Initialize the Zvuk Music client.
80
81 :param token: Zvuk Music X-Auth-Token.
82 """
83 self._token = token
84 self._client: ClientAsync | None = None
85 self._user_id: str | None = None
86
87 @property
88 def user_id(self) -> str:
89 """Return the user ID."""
90 if self._user_id is None:
91 raise ProviderUnavailableError("Client not initialized, call connect() first")
92 return self._user_id
93
94 async def connect(self) -> None:
95 """Initialize the client and verify token validity.
96
97 :raises LoginFailed: If the token is invalid.
98 :raises ResourceTemporarilyUnavailable: If there is a network error.
99 """
100 try:
101 self._client = await ClientAsync(token=self._token).init()
102 if not await self._client.is_authorized():
103 raise LoginFailed("Invalid Zvuk Music token")
104 profile = await self._client.get_profile()
105 if profile and profile.result:
106 self._user_id = str(profile.result.id)
107 LOGGER.debug("Connected to Zvuk Music as user %s", self._user_id)
108 except UnauthorizedError as err:
109 raise LoginFailed("Invalid Zvuk Music token") from err
110 except (NetworkError, TimedOutError) as err:
111 msg = "Network error connecting to Zvuk Music"
112 raise ResourceTemporarilyUnavailable(msg) from err
113
114 async def disconnect(self) -> None:
115 """Disconnect the client."""
116 self._client = None
117 self._user_id = None
118
119 def _ensure_connected(self) -> ClientAsync:
120 """Ensure the client is connected and return it."""
121 if self._client is None:
122 raise ProviderUnavailableError("Client not connected, call connect() first")
123 return self._client
124
125 # Search
126
127 @handle_zvuk_errors(not_found_return=None)
128 async def search(
129 self,
130 query: str,
131 limit: int = DEFAULT_LIMIT,
132 *,
133 search_tracks: bool = True,
134 search_artists: bool = True,
135 search_releases: bool = True,
136 search_playlists: bool = True,
137 ) -> ZvukSearch | None:
138 """Search for tracks, albums, artists, or playlists.
139
140 :param query: Search query string.
141 :param limit: Maximum number of results per type.
142 :param search_tracks: Whether to search for tracks.
143 :param search_artists: Whether to search for artists.
144 :param search_releases: Whether to search for releases.
145 :param search_playlists: Whether to search for playlists.
146 :return: Search results object or None.
147 """
148 client = self._ensure_connected()
149 return await client.search(
150 query,
151 limit=limit,
152 tracks=search_tracks,
153 artists=search_artists,
154 releases=search_releases,
155 playlists=search_playlists,
156 podcasts=False,
157 episodes=False,
158 profiles=False,
159 books=False,
160 )
161
162 # Get single items
163
164 @handle_zvuk_errors(not_found_return=None)
165 async def get_track(self, track_id: str) -> ZvukTrack | None:
166 """Get a single track by ID.
167
168 :param track_id: Track ID.
169 :return: Track object or None if not found.
170 """
171 client = self._ensure_connected()
172 return await client.get_track(track_id)
173
174 @handle_zvuk_errors(not_found_return=[])
175 async def get_tracks(self, track_ids: list[str]) -> list[ZvukTrack]:
176 """Get multiple tracks by IDs.
177
178 :param track_ids: List of track IDs.
179 :return: List of track objects.
180 """
181 client = self._ensure_connected()
182 ids: list[str | int] = list(track_ids)
183 return await client.get_tracks(ids)
184
185 @handle_zvuk_errors(not_found_return=None)
186 async def get_release(self, release_id: str) -> ZvukRelease | None:
187 """Get a single release (album) by ID.
188
189 :param release_id: Release ID.
190 :return: Release object or None if not found.
191 """
192 client = self._ensure_connected()
193 return await client.get_release(release_id)
194
195 @handle_zvuk_errors(not_found_return=[])
196 async def get_releases(self, release_ids: list[str]) -> list[ZvukRelease]:
197 """Get multiple releases by IDs.
198
199 :param release_ids: List of release IDs.
200 :return: List of release objects.
201 """
202 client = self._ensure_connected()
203 ids: list[str | int] = list(release_ids)
204 return await client.get_releases(ids)
205
206 @handle_zvuk_errors(not_found_return=None)
207 async def get_artist(self, artist_id: str) -> ZvukArtist | None:
208 """Get a single artist by ID.
209
210 :param artist_id: Artist ID.
211 :return: Artist object or None if not found.
212 """
213 client = self._ensure_connected()
214 return await client.get_artist(artist_id, with_description=True)
215
216 @handle_zvuk_errors(not_found_return=[])
217 async def get_artists(self, artist_ids: list[str]) -> list[ZvukArtist]:
218 """Get multiple artists by IDs.
219
220 :param artist_ids: List of artist IDs.
221 :return: List of artist objects.
222 """
223 client = self._ensure_connected()
224 ids: list[str | int] = list(artist_ids)
225 return await client.get_artists(ids)
226
227 @handle_zvuk_errors(not_found_return=[])
228 async def get_artist_releases(
229 self, artist_id: str, limit: int = DEFAULT_LIMIT
230 ) -> list[ZvukArtist]:
231 """Get artist's releases.
232
233 :param artist_id: Artist ID.
234 :param limit: Maximum number of releases.
235 :return: List of artist objects with populated releases.
236 """
237 client = self._ensure_connected()
238 return await client.get_artists([artist_id], with_releases=True, releases_limit=limit)
239
240 @handle_zvuk_errors(not_found_return=[])
241 async def get_artist_top_tracks(
242 self, artist_id: str, limit: int = DEFAULT_LIMIT
243 ) -> list[ZvukArtist]:
244 """Get artist's top tracks.
245
246 :param artist_id: Artist ID.
247 :param limit: Maximum number of tracks.
248 :return: List of artist objects with populated popular_tracks.
249 """
250 client = self._ensure_connected()
251 return await client.get_artists([artist_id], with_popular_tracks=True, tracks_limit=limit)
252
253 # Playlists
254
255 @handle_zvuk_errors(not_found_return=None)
256 async def get_playlist(self, playlist_id: str) -> ZvukPlaylist | None:
257 """Get a playlist by ID.
258
259 :param playlist_id: Playlist ID.
260 :return: Playlist object or None if not found.
261 """
262 client = self._ensure_connected()
263 return await client.get_playlist(playlist_id)
264
265 @handle_zvuk_errors(not_found_return=[])
266 async def get_playlists(self, playlist_ids: list[str]) -> list[ZvukPlaylist]:
267 """Get multiple playlists by IDs.
268
269 :param playlist_ids: List of playlist IDs.
270 :return: List of playlist objects.
271 """
272 client = self._ensure_connected()
273 ids: list[str | int] = list(playlist_ids)
274 return await client.get_playlists(ids)
275
276 @handle_zvuk_errors(not_found_return=[])
277 async def get_playlist_tracks(
278 self, playlist_id: str, limit: int = 50, offset: int = 0
279 ) -> list[ZvukSimpleTrack]:
280 """Get playlist tracks.
281
282 :param playlist_id: Playlist ID.
283 :param limit: Maximum number of tracks.
284 :param offset: Offset for pagination.
285 :return: List of SimpleTrack objects.
286 """
287 client = self._ensure_connected()
288 return await client.get_playlist_tracks(playlist_id, limit=limit, offset=offset)
289
290 # Streaming
291
292 @handle_zvuk_errors(not_found_return=[])
293 async def get_stream_urls(self, track_id: str) -> list[ZvukStream]:
294 """Get stream URLs for a track.
295
296 :param track_id: Track ID.
297 :return: List of Stream objects.
298 """
299 client = self._ensure_connected()
300 return await client.get_stream_urls(track_id)
301
302 # Collection (Library)
303
304 @handle_zvuk_errors()
305 async def get_collection(self) -> Collection | None:
306 """Get user's collection (liked items).
307
308 :return: Collection object or None.
309 """
310 client = self._ensure_connected()
311 return await client.get_collection()
312
313 @handle_zvuk_errors(not_found_return=[])
314 async def get_liked_tracks(self) -> list[ZvukTrack]:
315 """Get user's liked tracks.
316
317 :return: List of full Track objects.
318 """
319 client = self._ensure_connected()
320 return await client.get_liked_tracks()
321
322 @handle_zvuk_errors(not_found_return=[])
323 async def get_user_playlists(self) -> list[ZvukCollectionItem]:
324 """Get user's playlists.
325
326 :return: List of CollectionItem objects with playlist IDs.
327 """
328 client = self._ensure_connected()
329 return await client.get_user_playlists()
330
331 # Library modifications
332
333 async def like_track(self, track_id: str) -> bool:
334 """Add a track to liked tracks.
335
336 :param track_id: Track ID.
337 :return: True if successful.
338 """
339 client = self._ensure_connected()
340 try:
341 return await client.like_track(track_id)
342 except (BadRequestError, NetworkError, GraphQLError) as err:
343 LOGGER.error("Error liking track %s: %s", track_id, err)
344 return False
345
346 async def unlike_track(self, track_id: str) -> bool:
347 """Remove a track from liked tracks.
348
349 :param track_id: Track ID.
350 :return: True if successful.
351 """
352 client = self._ensure_connected()
353 try:
354 return await client.unlike_track(track_id)
355 except (BadRequestError, NetworkError, GraphQLError) as err:
356 LOGGER.error("Error unliking track %s: %s", track_id, err)
357 return False
358
359 async def like_release(self, release_id: str) -> bool:
360 """Add a release to liked releases.
361
362 :param release_id: Release ID.
363 :return: True if successful.
364 """
365 client = self._ensure_connected()
366 try:
367 return await client.like_release(release_id)
368 except (BadRequestError, NetworkError, GraphQLError) as err:
369 LOGGER.error("Error liking release %s: %s", release_id, err)
370 return False
371
372 async def unlike_release(self, release_id: str) -> bool:
373 """Remove a release from liked releases.
374
375 :param release_id: Release ID.
376 :return: True if successful.
377 """
378 client = self._ensure_connected()
379 try:
380 return await client.unlike_release(release_id)
381 except (BadRequestError, NetworkError, GraphQLError) as err:
382 LOGGER.error("Error unliking release %s: %s", release_id, err)
383 return False
384
385 async def like_artist(self, artist_id: str) -> bool:
386 """Add an artist to liked artists.
387
388 :param artist_id: Artist ID.
389 :return: True if successful.
390 """
391 client = self._ensure_connected()
392 try:
393 return await client.like_artist(artist_id)
394 except (BadRequestError, NetworkError, GraphQLError) as err:
395 LOGGER.error("Error liking artist %s: %s", artist_id, err)
396 return False
397
398 async def unlike_artist(self, artist_id: str) -> bool:
399 """Remove an artist from liked artists.
400
401 :param artist_id: Artist ID.
402 :return: True if successful.
403 """
404 client = self._ensure_connected()
405 try:
406 return await client.unlike_artist(artist_id)
407 except (BadRequestError, NetworkError, GraphQLError) as err:
408 LOGGER.error("Error unliking artist %s: %s", artist_id, err)
409 return False
410
411 async def like_playlist(self, playlist_id: str) -> bool:
412 """Add a playlist to liked playlists.
413
414 :param playlist_id: Playlist ID.
415 :return: True if successful.
416 """
417 client = self._ensure_connected()
418 try:
419 return await client.like_playlist(playlist_id)
420 except (BadRequestError, NetworkError, GraphQLError) as err:
421 LOGGER.error("Error liking playlist %s: %s", playlist_id, err)
422 return False
423
424 async def unlike_playlist(self, playlist_id: str) -> bool:
425 """Remove a playlist from liked playlists.
426
427 :param playlist_id: Playlist ID.
428 :return: True if successful.
429 """
430 client = self._ensure_connected()
431 try:
432 return await client.unlike_playlist(playlist_id)
433 except (BadRequestError, NetworkError, GraphQLError) as err:
434 LOGGER.error("Error unliking playlist %s: %s", playlist_id, err)
435 return False
436
437 # Playlist management
438
439 @handle_zvuk_errors()
440 async def create_playlist(self, name: str, track_ids: list[str] | None = None) -> str:
441 """Create a new playlist.
442
443 :param name: Playlist name.
444 :param track_ids: Optional list of track IDs to add.
445 :return: New playlist ID.
446 """
447 client = self._ensure_connected()
448 return await client.create_playlist(name, track_ids=track_ids)
449
450 async def delete_playlist(self, playlist_id: str) -> bool:
451 """Delete a playlist.
452
453 :param playlist_id: Playlist ID.
454 :return: True if successful.
455 """
456 client = self._ensure_connected()
457 try:
458 return await client.delete_playlist(playlist_id)
459 except (BadRequestError, NetworkError, GraphQLError) as err:
460 LOGGER.error("Error deleting playlist %s: %s", playlist_id, err)
461 return False
462
463 async def add_tracks_to_playlist(self, playlist_id: str, track_ids: list[str]) -> bool:
464 """Add tracks to a playlist.
465
466 :param playlist_id: Playlist ID.
467 :param track_ids: List of track IDs to add.
468 :return: True if successful.
469 """
470 client = self._ensure_connected()
471 try:
472 return await client.add_tracks_to_playlist(playlist_id, track_ids)
473 except (BadRequestError, NetworkError, GraphQLError) as err:
474 LOGGER.error("Error adding tracks to playlist %s: %s", playlist_id, err)
475 return False
476
477 async def update_playlist(self, playlist_id: str, track_ids: list[str]) -> bool:
478 """Update playlist tracks (used for removing tracks by providing remaining ones).
479
480 :param playlist_id: Playlist ID.
481 :param track_ids: Complete list of track IDs the playlist should contain.
482 :return: True if successful.
483 """
484 client = self._ensure_connected()
485 try:
486 return await client.update_playlist(playlist_id, track_ids)
487 except (BadRequestError, NetworkError, GraphQLError) as err:
488 LOGGER.error("Error updating playlist %s: %s", playlist_id, err)
489 return False
490