music-assistant-server

1"""Media retrieval operations for Tidal."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, Any, cast
7
8from aiohttp.client_exceptions import ClientError
9from music_assistant_models.enums import MediaType
10from music_assistant_models.errors import (
11    MediaNotFoundError,
12    MusicAssistantError,
13)
14from music_assistant_models.media_items import SearchResults
15
16from .constants import FAVORITE_TRACKS_PLAYLIST_ID, PAGES_MIX, PLAYLISTS, SKIPPABLE_ITEM_ERRORS
17from .parsers import (
18    parse_favorite_tracks_playlist,
19    parse_playlist,
20    parse_track,
21)
22from .parsers_v2 import parse_album as parse_album_v2
23from .parsers_v2 import parse_artist as parse_artist_v2
24from .parsers_v2 import parse_playlist as parse_playlist_v2
25from .parsers_v2 import parse_track as parse_track_v2
26
27if TYPE_CHECKING:
28    from collections.abc import Callable
29
30    from music_assistant_models.media_items import Album, Artist, Playlist, Track
31
32    from .jsonapi import JsonApiDocument
33    from .provider import TidalProvider
34
35
36class TidalMediaManager:
37    """Handles retrieval of media items from Tidal."""
38
39    def __init__(self, provider: TidalProvider):
40        """Initialize media retriever."""
41        self.provider = provider
42        self.api = provider.api
43        self.logger = provider.logger
44
45    async def search(
46        self, search_query: str, media_types: list[MediaType], limit: int = 5
47    ) -> SearchResults:
48        """Perform search on Tidal."""
49        results = SearchResults()
50        wanted = set(media_types)
51
52        # Build the includes for the requested types only, keeping under the
53        # official API's 10-included-resource cap. Track album covers are
54        # included so track results carry artwork; standalone album results
55        # trade artist names for staying within the cap.
56        includes: list[str] = []
57        if MediaType.TRACK in wanted:
58            includes += ["tracks.artists", "tracks.albums.coverArt"]
59        if MediaType.ALBUM in wanted:
60            includes.append("albums.coverArt")
61        if MediaType.ARTIST in wanted:
62            includes.append("artists.profileArt")
63        if MediaType.PLAYLIST in wanted:
64            includes.append("playlists.coverArt")
65        if not includes:
66            return results
67
68        # Since spec 1.10.101 search is a collection endpoint taking the query as
69        # a filter and returning exactly one searchResults resource (with an
70        # opaque id); the old /searchResults/{query} path 400s.
71        doc = await self.api.get_jsonapi(
72            "searchResults", params={"filter[query]": search_query}, include=includes
73        )
74        if not doc.data_list:
75            return results
76        data = doc.data_list[0]
77
78        # Slice the resources before parsing so we only parse up to `limit` items.
79        if MediaType.TRACK in wanted:
80            results.tracks = [
81                track
82                for res in doc.related(data, "tracks")[:limit]
83                if (track := self._parse_or_skip(parse_track_v2, doc, res)) is not None
84            ]
85        if MediaType.ALBUM in wanted:
86            results.albums = [
87                album
88                for res in doc.related(data, "albums")[:limit]
89                if (album := self._parse_or_skip(parse_album_v2, doc, res)) is not None
90            ]
91        if MediaType.ARTIST in wanted:
92            results.artists = [
93                artist
94                for res in doc.related(data, "artists")[:limit]
95                if (artist := self._parse_or_skip(parse_artist_v2, doc, res)) is not None
96            ]
97        if MediaType.PLAYLIST in wanted:
98            results.playlists = [
99                playlist
100                for res in doc.related(data, "playlists")[:limit]
101                if (playlist := self._parse_or_skip(parse_playlist_v2, doc, res)) is not None
102            ]
103        return results
104
105    async def get_artist(self, prov_artist_id: str) -> Artist:
106        """Get artist details."""
107        try:
108            doc = await self.api.get_jsonapi(
109                f"artists/{prov_artist_id}", include=["profileArt", "biography"]
110            )
111            return parse_artist_v2(self.provider, doc, doc.data)
112        except (ClientError, KeyError, ValueError) as err:
113            raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
114
115    async def get_album(self, prov_album_id: str) -> Album:
116        """Get album details."""
117        try:
118            doc = await self.api.get_jsonapi(
119                f"albums/{prov_album_id}", include=["artists", "coverArt", "genres"]
120            )
121            return parse_album_v2(self.provider, doc, doc.data)
122        except (ClientError, KeyError, ValueError) as err:
123            raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
124
125    async def get_track(self, prov_track_id: str) -> Track:
126        """Get track details."""
127        try:
128            # The album cover is resolved via the albums.coverArt include so the
129            # track carries an image, matching the unofficial API's behaviour.
130            doc = await self.api.get_jsonapi(
131                f"tracks/{prov_track_id}",
132                include=["artists", "albums", "albums.coverArt", "genres", "credits"],
133            )
134            track = parse_track_v2(self.provider, doc, doc.data)
135        except (ClientError, KeyError, ValueError) as err:
136            raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
137
138        # Lyrics remain on the unofficial API (not exposed at the official
139        # third-party tier). A lyrics failure must not fail the track lookup.
140        if lyrics := await self._get_lyrics(prov_track_id):
141            if plain := lyrics.get("lyrics"):
142                track.metadata.lyrics = plain
143            if synced := lyrics.get("subtitles"):
144                track.metadata.lrc_lyrics = synced
145
146        return track
147
148    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
149        """Get playlist details."""
150        if prov_playlist_id == FAVORITE_TRACKS_PLAYLIST_ID:
151            return parse_favorite_tracks_playlist(self.provider)
152
153        if prov_playlist_id.startswith("mix_"):
154            return await self._get_mix_details(prov_playlist_id[4:])
155
156        try:
157            data = await self.api.get(f"{PLAYLISTS}/{prov_playlist_id}")
158            return parse_playlist(self.provider, data)
159        except MediaNotFoundError:
160            return await self._get_mix_details(prov_playlist_id)
161        except (ClientError, KeyError, ValueError) as err:
162            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") from err
163
164    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
165        """Get album tracks."""
166        tracks: list[Track] = []
167        async for doc in self.api.paginate_jsonapi(
168            f"albums/{prov_album_id}/relationships/items",
169            include=["items.artists", "items.albums.coverArt"],
170            replace_media="items",
171        ):
172            for item in doc.data_list:
173                # The items relationship is mixed-type: an album's music
174                # videos appear here too, and parsing one as a track would
175                # yield an id that 404s on playback and shift trackNumber.
176                if item.get("type") != "tracks":
177                    continue
178                if not (resource := doc.resolve(item)):
179                    continue
180                if (track := self._parse_or_skip(parse_track_v2, doc, resource)) is None:
181                    continue
182                item_meta = item.get("meta") or {}
183                track.track_number = item_meta.get("trackNumber", 0) or 0
184                track.disc_number = item_meta.get("volumeNumber", 0) or 0
185                tracks.append(track)
186        return tracks
187
188    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
189        """Get artist albums."""
190        albums: list[Album] = []
191        async for doc in self.api.paginate_jsonapi(
192            f"artists/{prov_artist_id}/relationships/albums",
193            include=["albums.artists", "albums.coverArt"],
194            replace_media="albums",
195        ):
196            albums.extend(self._parse_items(parse_album_v2, doc))
197        return albums
198
199    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
200        """Get artist top tracks."""
201        # Top tracks are a bounded, ranked list: the first page is enough.
202        doc = await self.api.get_jsonapi(
203            f"artists/{prov_artist_id}/relationships/tracks",
204            params={"collapseBy": "FINGERPRINT"},
205            include=["tracks.artists", "tracks.albums.coverArt"],
206            replace_media="tracks",
207        )
208        return self._parse_items(parse_track_v2, doc)
209
210    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
211        """Get similar tracks."""
212        # Similar tracks are a bounded, ranked list: the first page is enough.
213        doc = await self.api.get_jsonapi(
214            f"tracks/{prov_track_id}/relationships/similarTracks",
215            include=["similarTracks.artists", "similarTracks.albums.coverArt"],
216            replace_media="similarTracks",
217        )
218        return self._parse_items(parse_track_v2, doc)[:limit]
219
220    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
221        """Get similar artists."""
222        # Similar artists are a bounded, ranked list: the first page is enough.
223        doc = await self.api.get_jsonapi(
224            f"artists/{prov_artist_id}/relationships/similarArtists",
225            include=["similarArtists.profileArt"],
226        )
227        return self._parse_items(parse_artist_v2, doc)[:limit]
228
229    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
230        """Get playlist tracks."""
231        page_size = 200
232        offset = page * page_size
233
234        if prov_playlist_id == FAVORITE_TRACKS_PLAYLIST_ID:
235            return await self._get_favorite_tracks(offset)
236
237        if prov_playlist_id.startswith("mix_"):
238            return await self._get_mix_tracks(prov_playlist_id[4:], page_size, offset)
239
240        try:
241            data = await self.api.get(
242                f"{PLAYLISTS}/{prov_playlist_id}/tracks",
243                params={"limit": page_size, "offset": offset},
244            )
245            return self._process_tracks(data.get("items", []), offset)
246        except MediaNotFoundError:
247            return await self._get_mix_tracks(prov_playlist_id, page_size, offset)
248
249    async def _get_mix_details(self, prov_mix_id: str) -> Playlist:
250        """Get details for a Tidal Mix."""
251        try:
252            tidal_mix = await self._fetch_mix_page(prov_mix_id)
253            mix_obj = {
254                "id": prov_mix_id,
255                "title": tidal_mix.get("title", "Unknown Mix"),
256                "updated": tidal_mix.get("lastUpdated", ""),
257                "subTitle": tidal_mix.get("subTitle", ""),
258                "images": {},
259            }
260            if module := self._find_mix_module(tidal_mix.get("rows", []), "mix"):
261                mix_obj["images"] = (module.get("mix") or {}).get("images", {})
262            return parse_playlist(self.provider, mix_obj, is_mix=True)
263        except (ClientError, KeyError, ValueError) as err:
264            raise MediaNotFoundError(f"Mix {prov_mix_id} not found") from err
265
266    async def _get_favorite_tracks(self, offset: int) -> list[Track]:
267        """Get the user's favorite tracks from the official user collection (newest first)."""
268        # The official collection is cursor-paginated, which does not map to MA's
269        # page-based interface. Walk the whole collection on the first page (cached
270        # by get_playlist_tracks) and return nothing for later pages.
271        if offset > 0:
272            return []
273        tracks: list[Track] = []
274        async for doc in self.api.paginate_jsonapi(
275            "userCollectionTracks/me/relationships/items",
276            include=["items.artists", "items.albums.coverArt"],
277            # Request the newest-first order explicitly rather than relying on the
278            # server default, since the positions below encode it.
279            params={"sort": "-addedAt"},
280            replace_media="items",
281        ):
282            for item in doc.data_list:
283                if not (resource := doc.resolve(item)):
284                    continue
285                if (track := self._parse_or_skip(parse_track_v2, doc, resource)) is None:
286                    continue
287                track.position = len(tracks) + 1
288                tracks.append(track)
289                # Feed the stale->live pairs Tidal computed for this read into
290                # the churn cache, as the library walk already does.
291                self.provider.note_replaced_track(item)
292        return tracks
293
294    async def _get_mix_tracks(self, mix_id: str, limit: int, offset: int) -> list[Track]:
295        """Get tracks from a mix."""
296        try:
297            data = await self._fetch_mix_page(mix_id)
298            module = self._find_mix_module(data.get("rows", []), "pagedList")
299            if not module:
300                raise MediaNotFoundError(f"Mix {mix_id} has no tracks")
301            all_items = module["pagedList"].get("items", [])
302            # The mix feed is not itself paginated, so slice MA's page window in memory.
303            paged_items = all_items[offset : offset + limit]
304            return self._process_tracks(paged_items, offset)
305        except (KeyError, ValueError) as err:
306            raise MediaNotFoundError(f"Mix {mix_id} not found") from err
307
308    async def _get_lyrics(self, prov_track_id: str) -> dict[str, str] | None:
309        """Get lyrics for a track, returning None when unavailable."""
310        # Lyrics are optional enrichment: never fail the track lookup on
311        # a missing/failed lyrics response.
312        try:
313            return await self.api.get(f"tracks/{prov_track_id}/lyrics")
314        except (ClientError, MusicAssistantError) as err:
315            self.logger.debug("Failed to fetch lyrics for track %s: %s", prov_track_id, err)
316            return None
317
318    def _parse_items[ItemT](
319        self,
320        parser: Callable[[TidalProvider, JsonApiDocument, dict[str, Any]], ItemT],
321        doc: JsonApiDocument,
322    ) -> list[ItemT]:
323        """
324        Parse the document's primary collection, leaving out the items that cannot be parsed.
325
326        Only for a collection of a single resource type: every resource is handed to the
327        same parser, so a mixed-type relationship has to be filtered by the caller.
328
329        :param parser: The parser to apply to each resolved resource.
330        :param doc: The JSON:API document holding the collection.
331        """
332        items: list[ItemT] = []
333        for identifier in doc.data_list:
334            if not (resource := doc.resolve(identifier)):
335                continue
336            if (item := self._parse_or_skip(parser, doc, resource)) is not None:
337                items.append(item)
338        return items
339
340    def _parse_or_skip[ItemT](
341        self,
342        parser: Callable[[TidalProvider, JsonApiDocument, dict[str, Any]], ItemT],
343        doc: JsonApiDocument,
344        resource: dict[str, Any],
345    ) -> ItemT | None:
346        """
347        Parse a resource into a media item, or return None if it cannot be parsed.
348
349        :param parser: The parser to apply to the resource.
350        :param doc: The JSON:API document holding the resource.
351        :param resource: The resource object to parse.
352        """
353        try:
354            return parser(self.provider, doc, resource)
355        except SKIPPABLE_ITEM_ERRORS as err:
356            # Tidal sending one item in a shape its parser can not read must not discard
357            # the rest of the collection. Log the traceback (on debug) as well, so a
358            # parser bug caught here is still traceable to its origin.
359            self.logger.warning(
360                "Skipping Tidal %s %s: %s",
361                resource.get("type", "item"),
362                resource.get("id", "[no id]"),
363                err,
364                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
365            )
366            return None
367
368    def _process_tracks(self, items: list[dict[str, Any]], offset: int) -> list[Track]:
369        result = []
370        for idx, item in enumerate(items, 1):
371            try:
372                track = parse_track(self.provider, item)
373                track.position = offset + idx
374                result.append(track)
375            except SKIPPABLE_ITEM_ERRORS as err:
376                track_data = item.get("item", item) if isinstance(item, dict) else item
377                self.logger.warning(
378                    "Skipping Tidal track %s: %s",
379                    track_data.get("id", "[no id]") if isinstance(track_data, dict) else "[no id]",
380                    err,
381                    exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
382                )
383                continue
384        return result
385
386    async def _fetch_mix_page(self, mix_id: str) -> dict[str, Any]:
387        """
388        Fetch the raw pages/mix feed for a mix, cached and shared.
389
390        The single feed carries both the mix header and its track list, so caching it
391        here lets get_playlist (details) and get_playlist_tracks share one upstream
392        request per mix instead of fetching the same feed twice.
393        """
394        cache = self.provider.mass.cache
395        cache_key = f"mix_page.{mix_id}"
396        if (cached := await cache.get(cache_key, provider=self.provider.instance_id)) is not None:
397            return cast("dict[str, Any]", cached)
398        data = await self.api.get(PAGES_MIX, params={"mixId": mix_id, "deviceType": "BROWSER"})
399        # Await the store: details and tracks are read back-to-back on a mix open,
400        # and a background write could lose that race and refetch the feed.
401        await cache.set(cache_key, data, expiration=3600 * 3, provider=self.provider.instance_id)
402        return data
403
404    @staticmethod
405    def _find_mix_module(rows: list[dict[str, Any]], key: str) -> dict[str, Any] | None:
406        """
407        Return the first pages/mix module carrying the given key.
408
409        The mix header (``mix``) and track list (``pagedList``) live in separate rows
410        whose order Tidal does not guarantee, so locate them by content rather than by a
411        fixed row/module index.
412        """
413        for row in rows:
414            for module in row.get("modules") or []:
415                if key in module:
416                    return cast("dict[str, Any]", module)
417        return None
418