/
/
/
1"""Qobuz musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6import datetime
7import hashlib
8import time
9from contextlib import suppress
10from datetime import UTC
11from typing import TYPE_CHECKING, Any, cast
12
13from aiohttp import client_exceptions
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import (
16 AlbumType,
17 ConfigEntryType,
18 ContentType,
19 ExternalID,
20 ImageType,
21 MediaType,
22 ProviderFeature,
23 StreamType,
24)
25from music_assistant_models.errors import (
26 InvalidDataError,
27 LoginFailed,
28 MediaNotFoundError,
29 RateLimited,
30 ResourceTemporarilyUnavailable,
31)
32from music_assistant_models.media_items import (
33 Album,
34 Artist,
35 AudioFormat,
36 MediaItemImage,
37 MediaItemType,
38 Playlist,
39 ProviderMapping,
40 SearchResults,
41 Track,
42)
43from music_assistant_models.streamdetails import StreamDetails
44
45from music_assistant.constants import (
46 CONF_ENTRY_UNOFFICIAL_PROVIDER,
47 CONF_PASSWORD,
48 CONF_USERNAME,
49 VARIOUS_ARTISTS_MBID,
50 VARIOUS_ARTISTS_NAME,
51)
52from music_assistant.controllers.cache import use_cache
53from music_assistant.helpers.app_vars import app_var
54from music_assistant.helpers.json import json_loads
55from music_assistant.helpers.throttle_retry import (
56 ThrottlerManager,
57 parse_retry_after,
58 throttle_with_retries,
59)
60from music_assistant.helpers.util import (
61 infer_album_type,
62 lock,
63 parse_title_and_version,
64 try_parse_int,
65)
66from music_assistant.models.music_provider import MusicProvider
67
68if TYPE_CHECKING:
69 from collections.abc import AsyncGenerator
70
71 from music_assistant_models.config_entries import ProviderConfig
72 from music_assistant_models.provider import ProviderManifest
73
74 from music_assistant import MusicAssistant
75 from music_assistant.models import ProviderInstanceType
76
77
78SUPPORTED_FEATURES = {
79 ProviderFeature.LIBRARY_ARTISTS,
80 ProviderFeature.LIBRARY_ALBUMS,
81 ProviderFeature.LIBRARY_TRACKS,
82 ProviderFeature.LIBRARY_PLAYLISTS,
83 ProviderFeature.LIBRARY_ARTISTS_EDIT,
84 ProviderFeature.LIBRARY_ALBUMS_EDIT,
85 ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
86 ProviderFeature.LIBRARY_TRACKS_EDIT,
87 ProviderFeature.PLAYLIST_TRACKS_EDIT,
88 ProviderFeature.PLAYLIST_CREATE,
89 ProviderFeature.BROWSE,
90 ProviderFeature.SEARCH,
91 ProviderFeature.ARTIST_ALBUMS,
92 ProviderFeature.ARTIST_TOPTRACKS,
93}
94
95VARIOUS_ARTISTS_ID = "145383"
96
97CONF_QUALITY = "quality"
98
99
100async def setup(
101 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
102) -> ProviderInstanceType:
103 """Initialize provider(instance) with given configuration."""
104 return QobuzProvider(mass, manifest, config, SUPPORTED_FEATURES)
105
106
107class QobuzProvider(MusicProvider):
108 """Provider for the Qobuz music service."""
109
110 _user_auth_info: dict[str, Any] | None = None
111 # Class-level throttler shared across all instances of this provider.
112 # This ensures a single rate limit even if multiple Qobuz accounts are configured.
113 throttler = ThrottlerManager(rate_limit=2, period=1)
114
115 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
116 """Return Config entries to configure this provider."""
117 return (
118 CONF_ENTRY_UNOFFICIAL_PROVIDER,
119 ConfigEntry(
120 key=CONF_QUALITY,
121 type=ConfigEntryType.STRING,
122 default_value="27",
123 options=[
124 ConfigValueOption("27"),
125 ConfigValueOption("7"),
126 ConfigValueOption("6"),
127 ConfigValueOption("5"),
128 ],
129 ),
130 )
131
132 async def handle_async_init(self) -> None:
133 """Handle async initialization of the provider."""
134 if not self.get_setup_value(CONF_USERNAME) or not self.get_setup_value(CONF_PASSWORD):
135 msg = "Invalid login credentials"
136 raise LoginFailed(msg)
137 # try to get a token, raise if that fails
138 token = await self._auth_token()
139 if not token:
140 msg = f"Login failed for user {self.get_setup_value(CONF_USERNAME)}"
141 raise LoginFailed(msg)
142
143 @use_cache(3600 * 24 * 14) # Cache for 14 days
144 async def search(
145 self, search_query: str, media_types: list[MediaType], limit: int = 5
146 ) -> SearchResults:
147 """
148 Perform search on musicprovider.
149
150 :param search_query: Search query.
151 :param media_types: A list of media_types to include. All types if None.
152 :param limit: Number of items to return in the search (per type).
153 """
154 result = SearchResults()
155 media_types = [
156 x
157 for x in media_types
158 if x in (MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST)
159 ]
160 if not media_types:
161 return result
162 params: dict[str, Any] = {"query": search_query, "limit": limit}
163 if len(media_types) == 1:
164 # qobuz does not support multiple searchtypes, falls back to all if no type given
165 if media_types[0] == MediaType.ARTIST:
166 params["type"] = "artists"
167 if media_types[0] == MediaType.ALBUM:
168 params["type"] = "albums"
169 if media_types[0] == MediaType.TRACK:
170 params["type"] = "tracks"
171 if media_types[0] == MediaType.PLAYLIST:
172 params["type"] = "playlists"
173 if searchresult := await self._get_data("catalog/search", **params):
174 if "artists" in searchresult and MediaType.ARTIST in media_types:
175 result.artists = [
176 self._parse_artist(item)
177 for item in searchresult["artists"]["items"]
178 if (item and item["id"])
179 ]
180 if "albums" in searchresult and MediaType.ALBUM in media_types:
181 result.albums = [
182 await self._parse_album(item)
183 for item in searchresult["albums"]["items"]
184 if (item and item["id"])
185 ]
186 if "tracks" in searchresult and MediaType.TRACK in media_types:
187 result.tracks = [
188 await self._parse_track(item)
189 for item in searchresult["tracks"]["items"]
190 if (item and item["id"])
191 ]
192 if "playlists" in searchresult and MediaType.PLAYLIST in media_types:
193 result.playlists = [
194 self._parse_playlist(item)
195 for item in searchresult["playlists"]["items"]
196 if (item and item["id"])
197 ]
198 return result
199
200 async def get_library_artists(self) -> AsyncGenerator[Artist]:
201 """Retrieve all library artists from Qobuz."""
202 endpoint = "favorite/getUserFavorites"
203 for item in await self._get_all_items(endpoint, key="artists", type="artists"):
204 if item and item["id"]:
205 yield self._parse_artist(item)
206
207 async def get_library_albums(self) -> AsyncGenerator[Album]:
208 """Retrieve all library albums from Qobuz."""
209 endpoint = "favorite/getUserFavorites"
210 for item in await self._get_all_items(endpoint, key="albums", type="albums"):
211 if item and item["id"]:
212 yield await self._parse_album(item)
213
214 async def get_library_tracks(self) -> AsyncGenerator[Track]:
215 """Retrieve library tracks from Qobuz."""
216 endpoint = "favorite/getUserFavorites"
217 for item in await self._get_all_items(endpoint, key="tracks", type="tracks"):
218 if item and item["id"]:
219 yield await self._parse_track(item)
220
221 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
222 """Retrieve all library playlists from the provider."""
223 endpoint = "playlist/getUserPlaylists"
224 for item in await self._get_all_items(endpoint, key="playlists"):
225 if item and item["id"]:
226 yield self._parse_playlist(item)
227
228 @use_cache(3600 * 24 * 30) # Cache for 30 days
229 async def get_artist(self, prov_artist_id: str) -> Artist:
230 """Get full artist details by id."""
231 params: dict[str, Any] = {"artist_id": prov_artist_id}
232 artist_obj = await self._get_data("artist/get", **params)
233 if artist_obj and artist_obj.get("id"):
234 return self._parse_artist(artist_obj)
235 msg = f"Item {prov_artist_id} not found"
236 raise MediaNotFoundError(msg)
237
238 @use_cache(3600 * 24 * 30) # Cache for 30 days
239 async def get_album(self, prov_album_id: str) -> Album:
240 """Get full album details by id."""
241 params: dict[str, Any] = {"album_id": prov_album_id}
242 album_obj = await self._get_data("album/get", **params)
243 if album_obj and album_obj.get("id"):
244 return await self._parse_album(album_obj)
245 msg = f"Item {prov_album_id} not found"
246 raise MediaNotFoundError(msg)
247
248 @use_cache(3600 * 24 * 30) # Cache for 30 days
249 async def get_track(self, prov_track_id: str) -> Track:
250 """Get full track details by id."""
251 params: dict[str, Any] = {"track_id": prov_track_id}
252 track_obj = await self._get_data("track/get", **params)
253 if track_obj and track_obj.get("id"):
254 return await self._parse_track(track_obj)
255 msg = f"Item {prov_track_id} not found"
256 raise MediaNotFoundError(msg)
257
258 @use_cache(3600 * 24 * 30) # Cache for 30 days
259 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
260 """Get full playlist details by id."""
261 params: dict[str, Any] = {"playlist_id": prov_playlist_id}
262 playlist_obj = await self._get_data("playlist/get", **params)
263 if playlist_obj and playlist_obj.get("id"):
264 return self._parse_playlist(playlist_obj)
265 msg = f"Item {prov_playlist_id} not found"
266 raise MediaNotFoundError(msg)
267
268 async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
269 """Create a new playlist on Qobuz with the given name."""
270 playlist_obj = await self._get_data(
271 "playlist/create",
272 name=name,
273 description="",
274 is_public=0,
275 is_collaborative=0,
276 )
277 if not playlist_obj or not playlist_obj.get("id"):
278 msg = f"Failed to create playlist: {name}"
279 raise InvalidDataError(
280 msg,
281 translation_key="create_playlist_failed",
282 translation_owner=self.translation_owner,
283 translation_args=[name],
284 )
285 return self._parse_playlist(playlist_obj)
286
287 @use_cache(3600 * 24 * 30, allow_expired_cache=True) # Cache for 30 days
288 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
289 """Get all album tracks for given album id."""
290 params = {"album_id": prov_album_id}
291 result: list[Track] = []
292 for index, item in enumerate(
293 await self._get_all_items("album/get", **params, key="tracks")
294 ):
295 if not (item and item["id"]):
296 continue
297 result.append(await self._parse_track(item))
298 if index % 10 == 0:
299 await asyncio.sleep(0)
300 return result
301
302 @use_cache(3600 * 3, allow_expired_cache=True) # Cache for 3 hours
303 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
304 """Get playlist tracks."""
305 result: list[Track] = []
306 page_size = 100
307 offset = page * page_size
308 qobuz_result = await self._get_data(
309 "playlist/get",
310 key="tracks",
311 playlist_id=prov_playlist_id,
312 extra="tracks",
313 offset=offset,
314 limit=page_size,
315 )
316 if not qobuz_result:
317 return result
318
319 for index, track_obj in enumerate(qobuz_result["tracks"]["items"], 1):
320 if not (track_obj and track_obj["id"]):
321 continue
322 track = await self._parse_track(track_obj)
323 track.position = index + offset
324 result.append(track)
325 if index % 10 == 0:
326 await asyncio.sleep(0)
327 return result
328
329 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
330 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
331 """Get a list of albums for the given artist."""
332 result = await self._get_data(
333 "artist/get",
334 artist_id=prov_artist_id,
335 extra="albums",
336 offset=0,
337 limit=100,
338 )
339 if not result:
340 return []
341 return [
342 await self._parse_album(item)
343 for item in result["albums"]["items"]
344 if (
345 item
346 and item["id"]
347 and item.get("artist")
348 and str(item["artist"]["id"]) == prov_artist_id
349 )
350 ]
351
352 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
353 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
354 """Get a list of most popular tracks for the given artist."""
355 result = await self._get_data(
356 "artist/get",
357 artist_id=prov_artist_id,
358 extra="playlists",
359 offset=0,
360 limit=25,
361 )
362 if result and result.get("playlists"):
363 return [
364 await self._parse_track(item)
365 for item in result["playlists"][0]["tracks"]["items"]
366 if (item and item["id"])
367 ]
368 # fallback to search
369 artist = await self.get_artist(prov_artist_id)
370 searchresult = await self._get_data(
371 "catalog/search", query=artist.name, limit=25, type="tracks"
372 )
373 if not searchresult:
374 return []
375
376 return [
377 await self._parse_track(item)
378 for item in searchresult["tracks"]["items"]
379 if (
380 item
381 and item["id"]
382 and "performer" in item
383 and str(item["performer"]["id"]) == str(prov_artist_id)
384 )
385 ]
386
387 async def library_add(self, item: MediaItemType) -> bool:
388 """Add item to library."""
389 result = None
390 if item.media_type == MediaType.ARTIST:
391 result = await self._get_data("favorite/create", artist_id=item.item_id)
392 elif item.media_type == MediaType.ALBUM:
393 result = await self._get_data("favorite/create", album_ids=item.item_id)
394 elif item.media_type == MediaType.TRACK:
395 result = await self._get_data("favorite/create", track_ids=item.item_id)
396 elif item.media_type == MediaType.PLAYLIST:
397 result = await self._get_data("playlist/subscribe", playlist_id=item.item_id)
398 return result is not None
399
400 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
401 """Remove item from library."""
402 result = None
403 if media_type == MediaType.ARTIST:
404 result = await self._get_data("favorite/delete", artist_ids=prov_item_id)
405 elif media_type == MediaType.ALBUM:
406 result = await self._get_data("favorite/delete", album_ids=prov_item_id)
407 elif media_type == MediaType.TRACK:
408 result = await self._get_data("favorite/delete", track_ids=prov_item_id)
409 elif media_type == MediaType.PLAYLIST:
410 playlist = await self.get_playlist(prov_item_id)
411 if playlist.is_editable:
412 result = await self._get_data("playlist/delete", playlist_id=prov_item_id)
413 else:
414 result = await self._get_data("playlist/unsubscribe", playlist_id=prov_item_id)
415 return result is not None
416
417 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
418 """Add track(s) to playlist."""
419 await self._get_data(
420 "playlist/addTracks",
421 playlist_id=prov_playlist_id,
422 track_ids=",".join(prov_track_ids),
423 playlist_track_ids=",".join(prov_track_ids),
424 )
425
426 async def remove_playlist_tracks(
427 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
428 ) -> None:
429 """Remove track(s) from playlist."""
430 playlist_track_ids = set()
431 for pos in positions_to_remove:
432 idx = pos - 1
433 qobuz_result = await self._get_data(
434 "playlist/get",
435 key="tracks",
436 playlist_id=prov_playlist_id,
437 extra="tracks",
438 offset=idx,
439 limit=1,
440 )
441 if not qobuz_result:
442 continue
443 playlist_track_id = qobuz_result["tracks"]["items"][0]["playlist_track_id"]
444 playlist_track_ids.add(str(playlist_track_id))
445
446 await self._get_data(
447 "playlist/deleteTracks",
448 playlist_id=prov_playlist_id,
449 playlist_track_ids=",".join(playlist_track_ids),
450 )
451
452 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
453 """Return the content details for the given track when it will be streamed."""
454 max_quality = int(cast("str", self.config.get_value(CONF_QUALITY)) or "27")
455 # Quality order from highest to lowest
456 quality_order = [27, 7, 6, 5]
457 # Only try qualities up to the user's maximum setting
458 allowed_qualities = [q for q in quality_order if q <= max_quality]
459
460 streamdata: dict[str, Any] | None = None
461 for format_id in allowed_qualities:
462 # it seems that simply requesting for highest available quality does not work
463 # from time to time the api response is empty for this request ?!
464 result = await self._get_data(
465 "track/getFileUrl",
466 sign_request=True,
467 format_id=format_id,
468 track_id=item_id,
469 intent="stream",
470 )
471 if result and result.get("url"):
472 streamdata = result
473 break
474 if not streamdata:
475 msg = f"Unable to retrieve stream details for {item_id}"
476 raise MediaNotFoundError(msg)
477 if streamdata["mime_type"] == "audio/mpeg":
478 content_type = ContentType.MPEG
479 elif streamdata["mime_type"] == "audio/flac":
480 content_type = ContentType.FLAC
481 else:
482 msg = f"Unsupported mime type for {item_id}"
483 raise MediaNotFoundError(msg)
484 self.mass.create_task(self._report_playback_started(streamdata))
485 return StreamDetails(
486 item_id=str(item_id),
487 provider=self.instance_id,
488 audio_format=AudioFormat(
489 content_type=content_type,
490 sample_rate=int(streamdata["sampling_rate"] * 1000),
491 bit_depth=streamdata["bit_depth"],
492 ),
493 stream_type=StreamType.HTTP,
494 duration=streamdata["duration"],
495 data=streamdata, # we need these details for reporting playback
496 path=streamdata["url"],
497 can_seek=True,
498 allow_seek=True,
499 )
500
501 async def on_streamed(
502 self,
503 streamdetails: StreamDetails,
504 ) -> None:
505 """Handle callback when an item completed streaming."""
506 if self._user_auth_info is None:
507 msg = "User auth info not available"
508 raise LoginFailed(msg)
509 user_id = self._user_auth_info["user"]["id"]
510 async with self.throttler.bypass():
511 await self._get_data(
512 "track/reportStreamingEnd",
513 user_id=user_id,
514 track_id=str(streamdetails.item_id),
515 duration=try_parse_int(streamdetails.seconds_streamed),
516 )
517
518 async def _report_playback_started(self, streamdata: dict[str, Any]) -> None:
519 """Report playback start to qobuz."""
520 # TODO: need to figure out if the streamed track is purchased by user
521 # https://www.qobuz.com/api.json/0.2/purchase/getUserPurchasesIds?limit=5000&user_id=xxxxxxx
522 # {"albums":{"total":0,"items":[]},
523 # "tracks":{"total":0,"items":[]},"user":{"id":xxxx,"login":"xxxxx"}}
524 assert self._user_auth_info is not None # for type checking
525 device_id = self._user_auth_info["user"]["device"]["id"]
526 credential_id = self._user_auth_info["user"]["credential"]["id"]
527 user_id = self._user_auth_info["user"]["id"]
528 format_id = streamdata["format_id"]
529 timestamp = int(time.time())
530 events = [
531 {
532 "online": True,
533 "sample": False,
534 "intent": "stream",
535 "device_id": device_id,
536 "track_id": streamdata["track_id"],
537 "purchase": False,
538 "date": timestamp,
539 "credential_id": credential_id,
540 "user_id": user_id,
541 "local": False,
542 "format_id": format_id,
543 }
544 ]
545 async with self.throttler.bypass():
546 await self._post_data("track/reportStreamingStart", data=events)
547
548 def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
549 """Parse qobuz artist object to generic layout."""
550 artist = Artist(
551 item_id=str(artist_obj["id"]),
552 provider=self.domain,
553 name=artist_obj["name"],
554 provider_mappings={
555 ProviderMapping(
556 item_id=str(artist_obj["id"]),
557 provider_domain=self.domain,
558 provider_instance=self.instance_id,
559 url=f"https://open.qobuz.com/artist/{artist_obj['id']}",
560 )
561 },
562 )
563 if artist.item_id == VARIOUS_ARTISTS_ID:
564 artist.mbid = VARIOUS_ARTISTS_MBID
565 artist.name = VARIOUS_ARTISTS_NAME
566 if img := self.__get_image(artist_obj):
567 artist.metadata.add_image(
568 MediaItemImage(
569 type=ImageType.THUMB,
570 path=img,
571 provider=self.instance_id,
572 remotely_accessible=True,
573 )
574 )
575 if biography := artist_obj.get("biography"):
576 artist.metadata.description = biography.get("content")
577 artist.metadata.description_language = biography.get("language")
578 if favorited_at := artist_obj.get("favorited_at"):
579 artist.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
580 return artist
581
582 async def _parse_album(
583 self, album_obj: dict[str, Any], artist_obj: dict[str, Any] | None = None
584 ) -> Album:
585 """Parse qobuz album object to generic layout."""
586 if not artist_obj and "artist" not in album_obj:
587 # artist missing in album info, return full abum instead
588 return await self.get_album(album_obj["id"])
589 name, version = parse_title_and_version(album_obj["title"], album_obj.get("version"))
590 album = Album(
591 item_id=str(album_obj["id"]),
592 provider=self.domain,
593 name=name,
594 version=version,
595 provider_mappings={
596 ProviderMapping(
597 item_id=str(album_obj["id"]),
598 provider_domain=self.domain,
599 provider_instance=self.instance_id,
600 available=album_obj["streamable"] and album_obj["displayable"],
601 audio_format=AudioFormat(
602 content_type=ContentType.FLAC,
603 sample_rate=album_obj["maximum_sampling_rate"] * 1000,
604 bit_depth=album_obj["maximum_bit_depth"],
605 ),
606 url=f"https://open.qobuz.com/album/{album_obj['id']}",
607 )
608 },
609 )
610 if upc := album_obj.get("upc"):
611 album.external_ids.add((ExternalID.BARCODE, upc))
612 album.artists.append(self._parse_artist(artist_obj or album_obj["artist"]))
613 if (
614 album_obj.get("product_type", "") == "single"
615 or album_obj.get("release_type", "") == "single"
616 ):
617 album.album_type = AlbumType.SINGLE
618 elif (
619 album_obj.get("product_type", "") == "compilation"
620 or album.artists[0].item_id == VARIOUS_ARTISTS_ID
621 ):
622 album.album_type = AlbumType.COMPILATION
623 elif (
624 album_obj.get("product_type", "") == "album"
625 or album_obj.get("release_type", "") == "album"
626 ):
627 album.album_type = AlbumType.ALBUM
628
629 # Try inference - override if it finds something more specific
630 inferred_type = infer_album_type(name, version)
631 if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
632 album.album_type = inferred_type
633
634 if "genre" in album_obj:
635 album.metadata.genres = {album_obj["genre"]["name"]}
636 if img := self.__get_image(album_obj):
637 album.metadata.add_image(
638 MediaItemImage(
639 provider=self.instance_id,
640 type=ImageType.THUMB,
641 path=img,
642 remotely_accessible=True,
643 )
644 )
645 if "label" in album_obj:
646 album.metadata.label = album_obj["label"]["name"]
647 if released_at := album_obj.get("released_at"):
648 with suppress(ValueError):
649 album.year = datetime.datetime.fromtimestamp(released_at, tz=UTC).year
650 if album_obj.get("copyright"):
651 album.metadata.copyright = album_obj["copyright"]
652 if album_obj.get("description"):
653 album.metadata.description = album_obj["description"]
654 if album_obj.get("parental_warning"):
655 album.metadata.explicit = True
656 if favorited_at := album_obj.get("favorited_at"):
657 album.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
658 return album
659
660 async def _parse_track(self, track_obj: dict[str, Any]) -> Track:
661 """Parse qobuz track object to generic layout."""
662 name, version = parse_title_and_version(track_obj["title"], track_obj.get("version"))
663 track = Track(
664 item_id=str(track_obj["id"]),
665 provider=self.domain,
666 name=name,
667 version=version,
668 duration=track_obj["duration"],
669 provider_mappings={
670 ProviderMapping(
671 item_id=str(track_obj["id"]),
672 provider_domain=self.domain,
673 provider_instance=self.instance_id,
674 available=track_obj["streamable"] and track_obj["displayable"],
675 audio_format=AudioFormat(
676 content_type=ContentType.FLAC,
677 sample_rate=track_obj["maximum_sampling_rate"] * 1000,
678 bit_depth=track_obj["maximum_bit_depth"],
679 ),
680 url=f"https://open.qobuz.com/track/{track_obj['id']}",
681 )
682 },
683 disc_number=track_obj.get("media_number", 0),
684 track_number=track_obj.get("track_number", 0),
685 )
686 if isrc := track_obj.get("isrc"):
687 track.external_ids.add((ExternalID.ISRC, isrc))
688 if (
689 track_obj.get("performer")
690 and str(track_obj["performer"].get("id", "")) != VARIOUS_ARTISTS_ID
691 ):
692 artist = self._parse_artist(track_obj["performer"])
693 if artist:
694 track.artists.append(artist)
695 # try to grab artist from album
696 if not track.artists and (
697 track_obj.get("album")
698 and track_obj["album"].get("artist")
699 and str(track_obj["album"]["artist"].get("id", "")) != VARIOUS_ARTISTS_ID
700 ):
701 artist = self._parse_artist(track_obj["album"]["artist"])
702 if artist:
703 track.artists.append(artist)
704 if not track.artists:
705 # last resort: parse from performers string
706 for performer_str in track_obj.get("performers", "").split(" - "):
707 if ", " not in performer_str:
708 continue
709 role = performer_str.split(", ")[1]
710 name = performer_str.split(", ")[0]
711 if "artist" in role.lower():
712 artist = Artist(
713 item_id=name,
714 provider=self.domain,
715 name=name,
716 provider_mappings={
717 ProviderMapping(
718 item_id=name,
719 provider_domain=self.domain,
720 provider_instance=self.instance_id,
721 )
722 },
723 )
724 track.artists.append(artist)
725 # TODO: fix grabbing composer from details
726
727 if "album" in track_obj:
728 album = await self._parse_album(track_obj["album"])
729 if album:
730 track.album = album
731 if track_obj.get("performers"):
732 track.metadata.performers = {x.strip() for x in track_obj["performers"].split("-")}
733 if track_obj.get("copyright"):
734 track.metadata.copyright = track_obj["copyright"]
735 if track_obj.get("parental_warning"):
736 track.metadata.explicit = True
737 if img := self.__get_image(track_obj):
738 track.metadata.add_image(
739 MediaItemImage(
740 type=ImageType.THUMB,
741 path=img,
742 provider=self.instance_id,
743 remotely_accessible=True,
744 )
745 )
746 if favorited_at := track_obj.get("favorited_at"):
747 track.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
748 return track
749
750 def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
751 """Parse qobuz playlist object to generic layout."""
752 if self._user_auth_info is None:
753 msg = "User auth info not available"
754 raise LoginFailed(msg)
755
756 is_editable = (
757 playlist_obj["owner"]["id"] == self._user_auth_info["user"]["id"]
758 or playlist_obj["is_collaborative"]
759 )
760 playlist = Playlist(
761 item_id=str(playlist_obj["id"]),
762 provider=self.instance_id,
763 name=playlist_obj["name"],
764 owner=playlist_obj["owner"]["name"],
765 provider_mappings={
766 ProviderMapping(
767 item_id=str(playlist_obj["id"]),
768 provider_domain=self.domain,
769 provider_instance=self.instance_id,
770 url=f"https://open.qobuz.com/playlist/{playlist_obj['id']}",
771 is_unique=is_editable, # user-owned playlists are unique
772 )
773 },
774 is_editable=is_editable,
775 )
776 if img := self.__get_image(playlist_obj):
777 playlist.metadata.add_image(
778 MediaItemImage(
779 type=ImageType.THUMB,
780 path=img,
781 provider=self.instance_id,
782 remotely_accessible=True,
783 )
784 )
785 # subscribed_at for playlists the user subscribed to, created_at for user-owned ones
786 if timestamp := playlist_obj.get("subscribed_at") or playlist_obj.get("created_at"):
787 playlist.date_added = datetime.datetime.fromtimestamp(timestamp, tz=datetime.UTC)
788 return playlist
789
790 @lock
791 async def _auth_token(self) -> str | None:
792 """Login to qobuz and store the token."""
793 if self._user_auth_info:
794 return str(self._user_auth_info["user_auth_token"])
795 # TODO: move credentials from query string to POST body to remove the
796 # residual exposure via HTTP session tracing / upstream proxy logs.
797 params: dict[str, Any] = {
798 "username": self.get_setup_value(CONF_USERNAME),
799 "password": self.get_setup_value(CONF_PASSWORD),
800 "device_manufacturer_id": "music_assistant",
801 }
802 details = await self._get_data("user/login", **params)
803 if details and "user" in details:
804 self._user_auth_info = details
805 self.logger.info(
806 "Successfully logged in to Qobuz as %s", details["user"]["display_name"]
807 )
808 self.mass.metadata.set_default_preferred_language(details["user"]["country_code"])
809 return str(details["user_auth_token"])
810 return None
811
812 async def _get_all_items(
813 self, endpoint: str, key: str = "tracks", **kwargs: Any
814 ) -> list[dict[str, Any]]:
815 """Get all items from a paged list."""
816 limit = 500
817 offset = 0
818 all_items: list[dict[str, Any]] = []
819 while True:
820 kwargs["limit"] = limit
821 kwargs["offset"] = offset
822 result = await self._get_data(endpoint, **kwargs)
823 offset += limit
824 if not result:
825 break
826 if not result.get(key) or not result[key].get("items"):
827 break
828 all_items.extend(result[key]["items"])
829 total = result[key].get("total", 0)
830 items_received = len(result[key]["items"])
831 if items_received < limit:
832 # If the API returned fewer items than requested but reports more exist,
833 # the server silently capped our limit. Continue paginating.
834 if items_received > 0 and total > len(all_items):
835 continue
836 break
837 return all_items
838
839 @throttle_with_retries
840 async def _get_data(
841 self, endpoint: str, sign_request: bool = False, **kwargs: Any
842 ) -> dict[str, Any] | None:
843 """Get data from api."""
844 self.logger.debug("Handling GET request to %s", endpoint)
845 url = f"https://www.qobuz.com/api.json/0.2/{endpoint}"
846 headers = {"X-App-Id": app_var("qobuz_app_id")}
847 locale = self.mass.metadata.locale.replace("_", "-")
848 language = locale.split("-")[0]
849 headers["Accept-Language"] = f"{locale}, {language};q=0.9, *;q=0.5"
850 if endpoint != "user/login":
851 auth_token = await self._auth_token()
852 if not auth_token:
853 self.logger.debug("Not logged in")
854 return None
855 headers["X-User-Auth-Token"] = auth_token
856 if sign_request:
857 signing_data = "".join(endpoint.split("/"))
858 keys = list(kwargs.keys())
859 keys.sort()
860 for key in keys:
861 signing_data += f"{key}{kwargs[key]}"
862 request_ts = str(time.time())
863 request_sig = signing_data + request_ts + app_var("qobuz_app_secret")
864 # Qobuz signs API requests with MD5; usedforsecurity=False as this is mandated by
865 # their API, not a security measure on our side.
866 request_sig = str(hashlib.md5(request_sig.encode(), usedforsecurity=False).hexdigest())
867 kwargs["request_ts"] = request_ts
868 kwargs["request_sig"] = request_sig
869 kwargs["app_id"] = app_var("qobuz_app_id")
870 kwargs["user_auth_token"] = await self._auth_token()
871 async with (
872 self.mass.http_session.get(url, headers=headers, params=kwargs) as response,
873 ):
874 # handle rate limiter
875 if response.status == 429:
876 retry_after = response.headers.get("Retry-After")
877 backoff_time = parse_retry_after(retry_after)
878 self.logger.warning(
879 "Rate limited by Qobuz API (429) on %s, Retry-After: %s",
880 endpoint,
881 retry_after or "not provided",
882 )
883 raise RateLimited("Rate Limiter", backoff_time=backoff_time)
884 # handle temporary server error
885 if response.status in (502, 503):
886 raise ResourceTemporarilyUnavailable(backoff_time=30)
887 # handle 404 not found, convert to MediaNotFoundError
888 if response.status == 404:
889 raise MediaNotFoundError(f"{endpoint} not found")
890 # deliberately no raise_for_status here: its exception message embeds
891 # the full request URL, which on /user/login carries the username and
892 # password as query params (and the user_auth_token on signed requests),
893 # so those would end up in the logs.
894 if response.status == 401:
895 if endpoint == "user/login":
896 raise LoginFailed("Invalid Qobuz credentials")
897 self._user_auth_info = None
898 raise LoginFailed("Qobuz session expired")
899 if response.status >= 400:
900 msg = f"Error {response.status} ({response.reason}) while handling {endpoint}"
901 raise InvalidDataError(msg)
902 try:
903 return cast("dict[str, Any]", await response.json(loads=json_loads))
904 except client_exceptions.ContentTypeError as err:
905 text = err.message or await response.text() or err.status
906 msg = f"Error while handling {endpoint}: {text}"
907 raise InvalidDataError(msg)
908
909 @throttle_with_retries
910 async def _post_data(
911 self,
912 endpoint: str,
913 params: dict[str, Any] | None = None,
914 data: dict[str, Any] | list[dict[str, Any]] | None = None,
915 ) -> dict[str, Any]:
916 """Post data to api."""
917 self.logger.debug("Handling POST request to %s", endpoint)
918 if not params:
919 params = {}
920 if not data:
921 data = {}
922 url = f"https://www.qobuz.com/api.json/0.2/{endpoint}"
923 params["app_id"] = app_var("qobuz_app_id")
924 auth_token = await self._auth_token()
925 if auth_token is None:
926 msg = "Authentication token is required"
927 raise LoginFailed(msg)
928 params["user_auth_token"] = auth_token
929 async with self.mass.http_session.post(url, params=params, json=data) as response:
930 # handle rate limiter
931 if response.status == 429:
932 retry_after = response.headers.get("Retry-After")
933 backoff_time = parse_retry_after(retry_after)
934 self.logger.warning(
935 "Rate limited by Qobuz API (429) on %s, Retry-After: %s",
936 endpoint,
937 retry_after or "not provided",
938 )
939 raise RateLimited("Rate Limiter", backoff_time=backoff_time)
940 # handle temporary server error
941 if response.status in (502, 503):
942 raise ResourceTemporarilyUnavailable(backoff_time=30)
943 # handle 404 not found, convert to MediaNotFoundError
944 if response.status == 404:
945 raise MediaNotFoundError(f"{endpoint} not found")
946 # deliberately no raise_for_status here: its exception message embeds
947 # the full request URL, which carries the user_auth_token as a query
948 # param, so it would end up in the logs.
949 if response.status == 401:
950 self._user_auth_info = None
951 raise LoginFailed("Qobuz session expired")
952 if response.status >= 400:
953 msg = f"Error {response.status} ({response.reason}) while handling {endpoint}"
954 raise InvalidDataError(msg)
955 return cast("dict[str, Any]", await response.json(loads=json_loads))
956
957 def __get_image(self, obj: dict[str, Any]) -> str | None:
958 """Try to parse image from Qobuz media object."""
959 if obj.get("image"):
960 for key in ["extralarge", "large", "medium", "small"]:
961 if obj["image"].get(key):
962 img_value: str = obj["image"][key]
963 if "2a96cbd8b46e442fc41c2b86b821562f" in img_value:
964 continue
965 return img_value
966 if obj.get("images300"):
967 # playlists seem to use this strange format
968 return str(obj["images300"][0])
969 if obj.get("album"):
970 return self.__get_image(obj["album"])
971 if obj.get("artist"):
972 return self.__get_image(obj["artist"])
973 return None
974