__init__.py

"""Qobuz musicprovider support for MusicAssistant."""

from __future__ import annotations

import asyncio
import datetime
import hashlib
import time
from contextlib import suppress
from datetime import UTC
from typing import TYPE_CHECKING, Any, cast

from aiohttp import client_exceptions
from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
from music_assistant_models.enums import (
    AlbumType,
    ConfigEntryType,
    ContentType,
    ExternalID,
    ImageType,
    MediaType,
    ProviderFeature,
    StreamType,
)
from music_assistant_models.errors import (
    InvalidDataError,
    LoginFailed,
    MediaNotFoundError,
    RateLimited,
    ResourceTemporarilyUnavailable,
)
from music_assistant_models.media_items import (
    Album,
    Artist,
    AudioFormat,
    MediaItemImage,
    MediaItemType,
    Playlist,
    ProviderMapping,
    SearchResults,
    Track,
)
from music_assistant_models.streamdetails import StreamDetails

from music_assistant.constants import (
    CONF_ENTRY_UNOFFICIAL_PROVIDER,
    CONF_PASSWORD,
    CONF_USERNAME,
    VARIOUS_ARTISTS_MBID,
    VARIOUS_ARTISTS_NAME,
)
from music_assistant.controllers.cache import use_cache
from music_assistant.helpers.app_vars import app_var
from music_assistant.helpers.json import json_loads
from music_assistant.helpers.throttle_retry import (
    ThrottlerManager,
    parse_retry_after,
    throttle_with_retries,
)
from music_assistant.helpers.util import (
    infer_album_type,
    lock,
    parse_title_and_version,
    try_parse_int,
)
from music_assistant.models.music_provider import MusicProvider

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator

    from music_assistant_models.config_entries import ProviderConfig
    from music_assistant_models.provider import ProviderManifest

    from music_assistant import MusicAssistant
    from music_assistant.models import ProviderInstanceType


SUPPORTED_FEATURES = {
    ProviderFeature.LIBRARY_ARTISTS,
    ProviderFeature.LIBRARY_ALBUMS,
    ProviderFeature.LIBRARY_TRACKS,
    ProviderFeature.LIBRARY_PLAYLISTS,
    ProviderFeature.LIBRARY_ARTISTS_EDIT,
    ProviderFeature.LIBRARY_ALBUMS_EDIT,
    ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
    ProviderFeature.LIBRARY_TRACKS_EDIT,
    ProviderFeature.PLAYLIST_TRACKS_EDIT,
    ProviderFeature.PLAYLIST_CREATE,
    ProviderFeature.BROWSE,
    ProviderFeature.SEARCH,
    ProviderFeature.ARTIST_ALBUMS,
    ProviderFeature.ARTIST_TOPTRACKS,
}

VARIOUS_ARTISTS_ID = "145383"

CONF_QUALITY = "quality"


async def setup(
    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
) -> ProviderInstanceType:
    """Initialize provider(instance) with given configuration."""
    return QobuzProvider(mass, manifest, config, SUPPORTED_FEATURES)


class QobuzProvider(MusicProvider):
    """Provider for the Qobuz music service."""

    _user_auth_info: dict[str, Any] | None = None
    # Class-level throttler shared across all instances of this provider.
    # This ensures a single rate limit even if multiple Qobuz accounts are configured.
    throttler = ThrottlerManager(rate_limit=2, period=1)

    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
        """Return Config entries to configure this provider."""
        return (
            CONF_ENTRY_UNOFFICIAL_PROVIDER,
            ConfigEntry(
                key=CONF_QUALITY,
                type=ConfigEntryType.STRING,
                default_value="27",
                options=[
                    ConfigValueOption("27"),
                    ConfigValueOption("7"),
                    ConfigValueOption("6"),
                    ConfigValueOption("5"),
                ],
            ),
        )

    async def handle_async_init(self) -> None:
        """Handle async initialization of the provider."""
        if not self.get_setup_value(CONF_USERNAME) or not self.get_setup_value(CONF_PASSWORD):
            msg = "Invalid login credentials"
            raise LoginFailed(msg)
        # try to get a token, raise if that fails
        token = await self._auth_token()
        if not token:
            msg = f"Login failed for user {self.get_setup_value(CONF_USERNAME)}"
            raise LoginFailed(msg)

    @use_cache(3600 * 24 * 14)  # Cache for 14 days
    async def search(
        self, search_query: str, media_types: list[MediaType], limit: int = 5
    ) -> SearchResults:
        """
        Perform search on musicprovider.

        :param search_query: Search query.
        :param media_types: A list of media_types to include. All types if None.
        :param limit: Number of items to return in the search (per type).
        """
        result = SearchResults()
        media_types = [
            x
            for x in media_types
            if x in (MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST)
        ]
        if not media_types:
            return result
        params: dict[str, Any] = {"query": search_query, "limit": limit}
        if len(media_types) == 1:
            # qobuz does not support multiple searchtypes, falls back to all if no type given
            if media_types[0] == MediaType.ARTIST:
                params["type"] = "artists"
            if media_types[0] == MediaType.ALBUM:
                params["type"] = "albums"
            if media_types[0] == MediaType.TRACK:
                params["type"] = "tracks"
            if media_types[0] == MediaType.PLAYLIST:
                params["type"] = "playlists"
        if searchresult := await self._get_data("catalog/search", **params):
            if "artists" in searchresult and MediaType.ARTIST in media_types:
                result.artists = [
                    self._parse_artist(item)
                    for item in searchresult["artists"]["items"]
                    if (item and item["id"])
                ]
            if "albums" in searchresult and MediaType.ALBUM in media_types:
                result.albums = [
                    await self._parse_album(item)
                    for item in searchresult["albums"]["items"]
                    if (item and item["id"])
                ]
            if "tracks" in searchresult and MediaType.TRACK in media_types:
                result.tracks = [
                    await self._parse_track(item)
                    for item in searchresult["tracks"]["items"]
                    if (item and item["id"])
                ]
            if "playlists" in searchresult and MediaType.PLAYLIST in media_types:
                result.playlists = [
                    self._parse_playlist(item)
                    for item in searchresult["playlists"]["items"]
                    if (item and item["id"])
                ]
        return result

    async def get_library_artists(self) -> AsyncGenerator[Artist]:
        """Retrieve all library artists from Qobuz."""
        endpoint = "favorite/getUserFavorites"
        for item in await self._get_all_items(endpoint, key="artists", type="artists"):
            if item and item["id"]:
                yield self._parse_artist(item)

    async def get_library_albums(self) -> AsyncGenerator[Album]:
        """Retrieve all library albums from Qobuz."""
        endpoint = "favorite/getUserFavorites"
        for item in await self._get_all_items(endpoint, key="albums", type="albums"):
            if item and item["id"]:
                yield await self._parse_album(item)

    async def get_library_tracks(self) -> AsyncGenerator[Track]:
        """Retrieve library tracks from Qobuz."""
        endpoint = "favorite/getUserFavorites"
        for item in await self._get_all_items(endpoint, key="tracks", type="tracks"):
            if item and item["id"]:
                yield await self._parse_track(item)

    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
        """Retrieve all library playlists from the provider."""
        endpoint = "playlist/getUserPlaylists"
        for item in await self._get_all_items(endpoint, key="playlists"):
            if item and item["id"]:
                yield self._parse_playlist(item)

    @use_cache(3600 * 24 * 30)  # Cache for 30 days
    async def get_artist(self, prov_artist_id: str) -> Artist:
        """Get full artist details by id."""
        params: dict[str, Any] = {"artist_id": prov_artist_id}
        artist_obj = await self._get_data("artist/get", **params)
        if artist_obj and artist_obj.get("id"):
            return self._parse_artist(artist_obj)
        msg = f"Item {prov_artist_id} not found"
        raise MediaNotFoundError(msg)

    @use_cache(3600 * 24 * 30)  # Cache for 30 days
    async def get_album(self, prov_album_id: str) -> Album:
        """Get full album details by id."""
        params: dict[str, Any] = {"album_id": prov_album_id}
        album_obj = await self._get_data("album/get", **params)
        if album_obj and album_obj.get("id"):
            return await self._parse_album(album_obj)
        msg = f"Item {prov_album_id} not found"
        raise MediaNotFoundError(msg)

    @use_cache(3600 * 24 * 30)  # Cache for 30 days
    async def get_track(self, prov_track_id: str) -> Track:
        """Get full track details by id."""
        params: dict[str, Any] = {"track_id": prov_track_id}
        track_obj = await self._get_data("track/get", **params)
        if track_obj and track_obj.get("id"):
            return await self._parse_track(track_obj)
        msg = f"Item {prov_track_id} not found"
        raise MediaNotFoundError(msg)

    @use_cache(3600 * 24 * 30)  # Cache for 30 days
    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
        """Get full playlist details by id."""
        params: dict[str, Any] = {"playlist_id": prov_playlist_id}
        playlist_obj = await self._get_data("playlist/get", **params)
        if playlist_obj and playlist_obj.get("id"):
            return self._parse_playlist(playlist_obj)
        msg = f"Item {prov_playlist_id} not found"
        raise MediaNotFoundError(msg)

    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
        """Create a new playlist on Qobuz with the given name."""
        playlist_obj = await self._get_data(
            "playlist/create",
            name=name,
            description="",
            is_public=0,
            is_collaborative=0,
        )
        if not playlist_obj or not playlist_obj.get("id"):
            msg = f"Failed to create playlist: {name}"
            raise InvalidDataError(
                msg,
                translation_key="create_playlist_failed",
                translation_owner=self.translation_owner,
                translation_args=[name],
            )
        return self._parse_playlist(playlist_obj)

    @use_cache(3600 * 24 * 30, allow_expired_cache=True)  # Cache for 30 days
    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
        """Get all album tracks for given album id."""
        params = {"album_id": prov_album_id}
        result: list[Track] = []
        for index, item in enumerate(
            await self._get_all_items("album/get", **params, key="tracks")
        ):
            if not (item and item["id"]):
                continue
            result.append(await self._parse_track(item))
            if index % 10 == 0:
                await asyncio.sleep(0)
        return result

    @use_cache(3600 * 3, allow_expired_cache=True)  # Cache for 3 hours
    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
        """Get playlist tracks."""
        result: list[Track] = []
        page_size = 100
        offset = page * page_size
        qobuz_result = await self._get_data(
            "playlist/get",
            key="tracks",
            playlist_id=prov_playlist_id,
            extra="tracks",
            offset=offset,
            limit=page_size,
        )
        if not qobuz_result:
            return result

        for index, track_obj in enumerate(qobuz_result["tracks"]["items"], 1):
            if not (track_obj and track_obj["id"]):
                continue
            track = await self._parse_track(track_obj)
            track.position = index + offset
            result.append(track)
            if index % 10 == 0:
                await asyncio.sleep(0)
        return result

    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
        """Get a list of albums for the given artist."""
        result = await self._get_data(
            "artist/get",
            artist_id=prov_artist_id,
            extra="albums",
            offset=0,
            limit=100,
        )
        if not result:
            return []
        return [
            await self._parse_album(item)
            for item in result["albums"]["items"]
            if (
                item
                and item["id"]
                and item.get("artist")
                and str(item["artist"]["id"]) == prov_artist_id
            )
        ]

    @use_cache(3600 * 24 * 14, allow_expired_cache=True)  # Cache for 14 days
    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
        """Get a list of most popular tracks for the given artist."""
        result = await self._get_data(
            "artist/get",
            artist_id=prov_artist_id,
            extra="playlists",
            offset=0,
            limit=25,
        )
        if result and result.get("playlists"):
            return [
                await self._parse_track(item)
                for item in result["playlists"][0]["tracks"]["items"]
                if (item and item["id"])
            ]
        # fallback to search
        artist = await self.get_artist(prov_artist_id)
        searchresult = await self._get_data(
            "catalog/search", query=artist.name, limit=25, type="tracks"
        )
        if not searchresult:
            return []

        return [
            await self._parse_track(item)
            for item in searchresult["tracks"]["items"]
            if (
                item
                and item["id"]
                and "performer" in item
                and str(item["performer"]["id"]) == str(prov_artist_id)
            )
        ]

    async def library_add(self, item: MediaItemType) -> bool:
        """Add item to library."""
        result = None
        if item.media_type == MediaType.ARTIST:
            result = await self._get_data("favorite/create", artist_id=item.item_id)
        elif item.media_type == MediaType.ALBUM:
            result = await self._get_data("favorite/create", album_ids=item.item_id)
        elif item.media_type == MediaType.TRACK:
            result = await self._get_data("favorite/create", track_ids=item.item_id)
        elif item.media_type == MediaType.PLAYLIST:
            result = await self._get_data("playlist/subscribe", playlist_id=item.item_id)
        return result is not None

    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
        """Remove item from library."""
        result = None
        if media_type == MediaType.ARTIST:
            result = await self._get_data("favorite/delete", artist_ids=prov_item_id)
        elif media_type == MediaType.ALBUM:
            result = await self._get_data("favorite/delete", album_ids=prov_item_id)
        elif media_type == MediaType.TRACK:
            result = await self._get_data("favorite/delete", track_ids=prov_item_id)
        elif media_type == MediaType.PLAYLIST:
            playlist = await self.get_playlist(prov_item_id)
            if playlist.is_editable:
                result = await self._get_data("playlist/delete", playlist_id=prov_item_id)
            else:
                result = await self._get_data("playlist/unsubscribe", playlist_id=prov_item_id)
        return result is not None

    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
        """Add track(s) to playlist."""
        await self._get_data(
            "playlist/addTracks",
            playlist_id=prov_playlist_id,
            track_ids=",".join(prov_track_ids),
            playlist_track_ids=",".join(prov_track_ids),
        )

    async def remove_playlist_tracks(
        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
    ) -> None:
        """Remove track(s) from playlist."""
        playlist_track_ids = set()
        for pos in positions_to_remove:
            idx = pos - 1
            qobuz_result = await self._get_data(
                "playlist/get",
                key="tracks",
                playlist_id=prov_playlist_id,
                extra="tracks",
                offset=idx,
                limit=1,
            )
            if not qobuz_result:
                continue
            playlist_track_id = qobuz_result["tracks"]["items"][0]["playlist_track_id"]
            playlist_track_ids.add(str(playlist_track_id))

        await self._get_data(
            "playlist/deleteTracks",
            playlist_id=prov_playlist_id,
            playlist_track_ids=",".join(playlist_track_ids),
        )

    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
        """Return the content details for the given track when it will be streamed."""
        max_quality = int(cast("str", self.config.get_value(CONF_QUALITY)) or "27")
        # Quality order from highest to lowest
        quality_order = [27, 7, 6, 5]
        # Only try qualities up to the user's maximum setting
        allowed_qualities = [q for q in quality_order if q <= max_quality]

        streamdata: dict[str, Any] | None = None
        for format_id in allowed_qualities:
            # it seems that simply requesting for highest available quality does not work
            # from time to time the api response is empty for this request ?!
            result = await self._get_data(
                "track/getFileUrl",
                sign_request=True,
                format_id=format_id,
                track_id=item_id,
                intent="stream",
            )
            if result and result.get("url"):
                streamdata = result
                break
        if not streamdata:
            msg = f"Unable to retrieve stream details for {item_id}"
            raise MediaNotFoundError(msg)
        if streamdata["mime_type"] == "audio/mpeg":
            content_type = ContentType.MPEG
        elif streamdata["mime_type"] == "audio/flac":
            content_type = ContentType.FLAC
        else:
            msg = f"Unsupported mime type for {item_id}"
            raise MediaNotFoundError(msg)
        self.mass.create_task(self._report_playback_started(streamdata))
        return StreamDetails(
            item_id=str(item_id),
            provider=self.instance_id,
            audio_format=AudioFormat(
                content_type=content_type,
                sample_rate=int(streamdata["sampling_rate"] * 1000),
                bit_depth=streamdata["bit_depth"],
            ),
            stream_type=StreamType.HTTP,
            duration=streamdata["duration"],
            data=streamdata,  # we need these details for reporting playback
            path=streamdata["url"],
            can_seek=True,
            allow_seek=True,
        )

    async def on_streamed(
        self,
        streamdetails: StreamDetails,
    ) -> None:
        """Handle callback when an item completed streaming."""
        if self._user_auth_info is None:
            msg = "User auth info not available"
            raise LoginFailed(msg)
        user_id = self._user_auth_info["user"]["id"]
        async with self.throttler.bypass():
            await self._get_data(
                "track/reportStreamingEnd",
                user_id=user_id,
                track_id=str(streamdetails.item_id),
                duration=try_parse_int(streamdetails.seconds_streamed),
            )

    async def _report_playback_started(self, streamdata: dict[str, Any]) -> None:
        """Report playback start to qobuz."""
        # TODO: need to figure out if the streamed track is purchased by user
        # https://www.qobuz.com/api.json/0.2/purchase/getUserPurchasesIds?limit=5000&user_id=xxxxxxx
        # {"albums":{"total":0,"items":[]},
        # "tracks":{"total":0,"items":[]},"user":{"id":xxxx,"login":"xxxxx"}}
        assert self._user_auth_info is not None  # for type checking
        device_id = self._user_auth_info["user"]["device"]["id"]
        credential_id = self._user_auth_info["user"]["credential"]["id"]
        user_id = self._user_auth_info["user"]["id"]
        format_id = streamdata["format_id"]
        timestamp = int(time.time())
        events = [
            {
                "online": True,
                "sample": False,
                "intent": "stream",
                "device_id": device_id,
                "track_id": streamdata["track_id"],
                "purchase": False,
                "date": timestamp,
                "credential_id": credential_id,
                "user_id": user_id,
                "local": False,
                "format_id": format_id,
            }
        ]
        async with self.throttler.bypass():
            await self._post_data("track/reportStreamingStart", data=events)

    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
        """Parse qobuz artist object to generic layout."""
        artist = Artist(
            item_id=str(artist_obj["id"]),
            provider=self.domain,
            name=artist_obj["name"],
            provider_mappings={
                ProviderMapping(
                    item_id=str(artist_obj["id"]),
                    provider_domain=self.domain,
                    provider_instance=self.instance_id,
                    url=f"https://open.qobuz.com/artist/{artist_obj['id']}",
                )
            },
        )
        if artist.item_id == VARIOUS_ARTISTS_ID:
            artist.mbid = VARIOUS_ARTISTS_MBID
            artist.name = VARIOUS_ARTISTS_NAME
        if img := self.__get_image(artist_obj):
            artist.metadata.add_image(
                MediaItemImage(
                    type=ImageType.THUMB,
                    path=img,
                    provider=self.instance_id,
                    remotely_accessible=True,
                )
            )
        if biography := artist_obj.get("biography"):
            artist.metadata.description = biography.get("content")
            artist.metadata.description_language = biography.get("language")
        if favorited_at := artist_obj.get("favorited_at"):
            artist.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
        return artist

    async def _parse_album(
        self, album_obj: dict[str, Any], artist_obj: dict[str, Any] | None = None
    ) -> Album:
        """Parse qobuz album object to generic layout."""
        if not artist_obj and "artist" not in album_obj:
            # artist missing in album info, return full abum instead
            return await self.get_album(album_obj["id"])
        name, version = parse_title_and_version(album_obj["title"], album_obj.get("version"))
        album = Album(
            item_id=str(album_obj["id"]),
            provider=self.domain,
            name=name,
            version=version,
            provider_mappings={
                ProviderMapping(
                    item_id=str(album_obj["id"]),
                    provider_domain=self.domain,
                    provider_instance=self.instance_id,
                    available=album_obj["streamable"] and album_obj["displayable"],
                    audio_format=AudioFormat(
                        content_type=ContentType.FLAC,
                        sample_rate=album_obj["maximum_sampling_rate"] * 1000,
                        bit_depth=album_obj["maximum_bit_depth"],
                    ),
                    url=f"https://open.qobuz.com/album/{album_obj['id']}",
                )
            },
        )
        if upc := album_obj.get("upc"):
            album.external_ids.add((ExternalID.BARCODE, upc))
        album.artists.append(self._parse_artist(artist_obj or album_obj["artist"]))
        if (
            album_obj.get("product_type", "") == "single"
            or album_obj.get("release_type", "") == "single"
        ):
            album.album_type = AlbumType.SINGLE
        elif (
            album_obj.get("product_type", "") == "compilation"
            or album.artists[0].item_id == VARIOUS_ARTISTS_ID
        ):
            album.album_type = AlbumType.COMPILATION
        elif (
            album_obj.get("product_type", "") == "album"
            or album_obj.get("release_type", "") == "album"
        ):
            album.album_type = AlbumType.ALBUM

        # Try inference - override if it finds something more specific
        inferred_type = infer_album_type(name, version)
        if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
            album.album_type = inferred_type

        if "genre" in album_obj:
            album.metadata.genres = {album_obj["genre"]["name"]}
        if img := self.__get_image(album_obj):
            album.metadata.add_image(
                MediaItemImage(
                    provider=self.instance_id,
                    type=ImageType.THUMB,
                    path=img,
                    remotely_accessible=True,
                )
            )
        if "label" in album_obj:
            album.metadata.label = album_obj["label"]["name"]
        if released_at := album_obj.get("released_at"):
            with suppress(ValueError):
                album.year = datetime.datetime.fromtimestamp(released_at, tz=UTC).year
        if album_obj.get("copyright"):
            album.metadata.copyright = album_obj["copyright"]
        if album_obj.get("description"):
            album.metadata.description = album_obj["description"]
        if album_obj.get("parental_warning"):
            album.metadata.explicit = True
        if favorited_at := album_obj.get("favorited_at"):
            album.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
        return album

    async def _parse_track(self, track_obj: dict[str, Any]) -> Track:
        """Parse qobuz track object to generic layout."""
        name, version = parse_title_and_version(track_obj["title"], track_obj.get("version"))
        track = Track(
            item_id=str(track_obj["id"]),
            provider=self.domain,
            name=name,
            version=version,
            duration=track_obj["duration"],
            provider_mappings={
                ProviderMapping(
                    item_id=str(track_obj["id"]),
                    provider_domain=self.domain,
                    provider_instance=self.instance_id,
                    available=track_obj["streamable"] and track_obj["displayable"],
                    audio_format=AudioFormat(
                        content_type=ContentType.FLAC,
                        sample_rate=track_obj["maximum_sampling_rate"] * 1000,
                        bit_depth=track_obj["maximum_bit_depth"],
                    ),
                    url=f"https://open.qobuz.com/track/{track_obj['id']}",
                )
            },
            disc_number=track_obj.get("media_number", 0),
            track_number=track_obj.get("track_number", 0),
        )
        if isrc := track_obj.get("isrc"):
            track.external_ids.add((ExternalID.ISRC, isrc))
        if (
            track_obj.get("performer")
            and str(track_obj["performer"].get("id", "")) != VARIOUS_ARTISTS_ID
        ):
            artist = self._parse_artist(track_obj["performer"])
            if artist:
                track.artists.append(artist)
        # try to grab artist from album
        if not track.artists and (
            track_obj.get("album")
            and track_obj["album"].get("artist")
            and str(track_obj["album"]["artist"].get("id", "")) != VARIOUS_ARTISTS_ID
        ):
            artist = self._parse_artist(track_obj["album"]["artist"])
            if artist:
                track.artists.append(artist)
        if not track.artists:
            # last resort: parse from performers string
            for performer_str in track_obj.get("performers", "").split(" - "):
                if ", " not in performer_str:
                    continue
                role = performer_str.split(", ")[1]
                name = performer_str.split(", ")[0]
                if "artist" in role.lower():
                    artist = Artist(
                        item_id=name,
                        provider=self.domain,
                        name=name,
                        provider_mappings={
                            ProviderMapping(
                                item_id=name,
                                provider_domain=self.domain,
                                provider_instance=self.instance_id,
                            )
                        },
                    )
                    track.artists.append(artist)
        # TODO: fix grabbing composer from details

        if "album" in track_obj:
            album = await self._parse_album(track_obj["album"])
            if album:
                track.album = album
        if track_obj.get("performers"):
            track.metadata.performers = {x.strip() for x in track_obj["performers"].split("-")}
        if track_obj.get("copyright"):
            track.metadata.copyright = track_obj["copyright"]
        if track_obj.get("parental_warning"):
            track.metadata.explicit = True
        if img := self.__get_image(track_obj):
            track.metadata.add_image(
                MediaItemImage(
                    type=ImageType.THUMB,
                    path=img,
                    provider=self.instance_id,
                    remotely_accessible=True,
                )
            )
        if favorited_at := track_obj.get("favorited_at"):
            track.date_added = datetime.datetime.fromtimestamp(favorited_at, tz=datetime.UTC)
        return track

    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
        """Parse qobuz playlist object to generic layout."""
        if self._user_auth_info is None:
            msg = "User auth info not available"
            raise LoginFailed(msg)

        is_editable = (
            playlist_obj["owner"]["id"] == self._user_auth_info["user"]["id"]
            or playlist_obj["is_collaborative"]
        )
        playlist = Playlist(
            item_id=str(playlist_obj["id"]),
            provider=self.instance_id,
            name=playlist_obj["name"],
            owner=playlist_obj["owner"]["name"],
            provider_mappings={
                ProviderMapping(
                    item_id=str(playlist_obj["id"]),
                    provider_domain=self.domain,
                    provider_instance=self.instance_id,
                    url=f"https://open.qobuz.com/playlist/{playlist_obj['id']}",
                    is_unique=is_editable,  # user-owned playlists are unique
                )
            },
            is_editable=is_editable,
        )
        if img := self.__get_image(playlist_obj):
            playlist.metadata.add_image(
                MediaItemImage(
                    type=ImageType.THUMB,
                    path=img,
                    provider=self.instance_id,
                    remotely_accessible=True,
                )
            )
        # subscribed_at for playlists the user subscribed to, created_at for user-owned ones
        if timestamp := playlist_obj.get("subscribed_at") or playlist_obj.get("created_at"):
            playlist.date_added = datetime.datetime.fromtimestamp(timestamp, tz=datetime.UTC)
        return playlist

    @lock
    async def _auth_token(self) -> str | None:
        """Login to qobuz and store the token."""
        if self._user_auth_info:
            return str(self._user_auth_info["user_auth_token"])
        # TODO: move credentials from query string to POST body to remove the
        # residual exposure via HTTP session tracing / upstream proxy logs.
        params: dict[str, Any] = {
            "username": self.get_setup_value(CONF_USERNAME),
            "password": self.get_setup_value(CONF_PASSWORD),
            "device_manufacturer_id": "music_assistant",
        }
        details = await self._get_data("user/login", **params)
        if details and "user" in details:
            self._user_auth_info = details
            self.logger.info(
                "Successfully logged in to Qobuz as %s", details["user"]["display_name"]
            )
            self.mass.metadata.set_default_preferred_language(details["user"]["country_code"])
            return str(details["user_auth_token"])
        return None

    async def _get_all_items(
        self, endpoint: str, key: str = "tracks", **kwargs: Any
    ) -> list[dict[str, Any]]:
        """Get all items from a paged list."""
        limit = 500
        offset = 0
        all_items: list[dict[str, Any]] = []
        while True:
            kwargs["limit"] = limit
            kwargs["offset"] = offset
            result = await self._get_data(endpoint, **kwargs)
            offset += limit
            if not result:
                break
            if not result.get(key) or not result[key].get("items"):
                break
            all_items.extend(result[key]["items"])
            total = result[key].get("total", 0)
            items_received = len(result[key]["items"])
            if items_received < limit:
                # If the API returned fewer items than requested but reports more exist,
                # the server silently capped our limit. Continue paginating.
                if items_received > 0 and total > len(all_items):
                    continue
                break
        return all_items

    @throttle_with_retries
    async def _get_data(
        self, endpoint: str, sign_request: bool = False, **kwargs: Any
    ) -> dict[str, Any] | None:
        """Get data from api."""
        self.logger.debug("Handling GET request to %s", endpoint)
        url = f"https://www.qobuz.com/api.json/0.2/{endpoint}"
        headers = {"X-App-Id": app_var("qobuz_app_id")}
        locale = self.mass.metadata.locale.replace("_", "-")
        language = locale.split("-")[0]
        headers["Accept-Language"] = f"{locale}, {language};q=0.9, *;q=0.5"
        if endpoint != "user/login":
            auth_token = await self._auth_token()
            if not auth_token:
                self.logger.debug("Not logged in")
                return None
            headers["X-User-Auth-Token"] = auth_token
        if sign_request:
            signing_data = "".join(endpoint.split("/"))
            keys = list(kwargs.keys())
            keys.sort()
            for key in keys:
                signing_data += f"{key}{kwargs[key]}"
            request_ts = str(time.time())
            request_sig = signing_data + request_ts + app_var("qobuz_app_secret")
            # Qobuz signs API requests with MD5; usedforsecurity=False as this is mandated by
            # their API, not a security measure on our side.
            request_sig = str(hashlib.md5(request_sig.encode(), usedforsecurity=False).hexdigest())
            kwargs["request_ts"] = request_ts
            kwargs["request_sig"] = request_sig
            kwargs["app_id"] = app_var("qobuz_app_id")
            kwargs["user_auth_token"] = await self._auth_token()
        async with (
            self.mass.http_session.get(url, headers=headers, params=kwargs) as response,
        ):
            # handle rate limiter
            if response.status == 429:
                retry_after = response.headers.get("Retry-After")
                backoff_time = parse_retry_after(retry_after)
                self.logger.warning(
                    "Rate limited by Qobuz API (429) on %s, Retry-After: %s",
                    endpoint,
                    retry_after or "not provided",
                )
                raise RateLimited("Rate Limiter", backoff_time=backoff_time)
            # handle temporary server error
            if response.status in (502, 503):
                raise ResourceTemporarilyUnavailable(backoff_time=30)
            # handle 404 not found, convert to MediaNotFoundError
            if response.status == 404:
                raise MediaNotFoundError(f"{endpoint} not found")
            # deliberately no raise_for_status here: its exception message embeds
            # the full request URL, which on /user/login carries the username and
            # password as query params (and the user_auth_token on signed requests),
            # so those would end up in the logs.
            if response.status == 401:
                if endpoint == "user/login":
                    raise LoginFailed("Invalid Qobuz credentials")
                self._user_auth_info = None
                raise LoginFailed("Qobuz session expired")
            if response.status >= 400:
                msg = f"Error {response.status} ({response.reason}) while handling {endpoint}"
                raise InvalidDataError(msg)
            try:
                return cast("dict[str, Any]", await response.json(loads=json_loads))
            except client_exceptions.ContentTypeError as err:
                text = err.message or await response.text() or err.status
                msg = f"Error while handling {endpoint}: {text}"
                raise InvalidDataError(msg)

    @throttle_with_retries
    async def _post_data(
        self,
        endpoint: str,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        """Post data to api."""
        self.logger.debug("Handling POST request to %s", endpoint)
        if not params:
            params = {}
        if not data:
            data = {}
        url = f"https://www.qobuz.com/api.json/0.2/{endpoint}"
        params["app_id"] = app_var("qobuz_app_id")
        auth_token = await self._auth_token()
        if auth_token is None:
            msg = "Authentication token is required"
            raise LoginFailed(msg)
        params["user_auth_token"] = auth_token
        async with self.mass.http_session.post(url, params=params, json=data) as response:
            # handle rate limiter
            if response.status == 429:
                retry_after = response.headers.get("Retry-After")
                backoff_time = parse_retry_after(retry_after)
                self.logger.warning(
                    "Rate limited by Qobuz API (429) on %s, Retry-After: %s",
                    endpoint,
                    retry_after or "not provided",
                )
                raise RateLimited("Rate Limiter", backoff_time=backoff_time)
            # handle temporary server error
            if response.status in (502, 503):
                raise ResourceTemporarilyUnavailable(backoff_time=30)
            # handle 404 not found, convert to MediaNotFoundError
            if response.status == 404:
                raise MediaNotFoundError(f"{endpoint} not found")
            # deliberately no raise_for_status here: its exception message embeds
            # the full request URL, which carries the user_auth_token as a query
            # param, so it would end up in the logs.
            if response.status == 401:
                self._user_auth_info = None
                raise LoginFailed("Qobuz session expired")
            if response.status >= 400:
                msg = f"Error {response.status} ({response.reason}) while handling {endpoint}"
                raise InvalidDataError(msg)
            return cast("dict[str, Any]", await response.json(loads=json_loads))

    def __get_image(self, obj: dict[str, Any]) -> str | None:
        """Try to parse image from Qobuz media object."""
        if obj.get("image"):
            for key in ["extralarge", "large", "medium", "small"]:
                if obj["image"].get(key):
                    img_value: str = obj["image"][key]
                    if "2a96cbd8b46e442fc41c2b86b821562f" in img_value:
                        continue
                    return img_value
        if obj.get("images300"):
            # playlists seem to use this strange format
            return str(obj["images300"][0])
        if obj.get("album"):
            return self.__get_image(obj["album"])
        if obj.get("artist"):
            return self.__get_image(obj["artist"])
        return None