music-assistant-server

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