music-assistant-server

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