music-assistant-server

1"""Media retrieval operations for Tidal."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
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, FAVORITES_TRACKS, PAGES_MIX, PLAYLISTS
16from .parsers import (
17    parse_album,
18    parse_artist,
19    parse_favorite_tracks_playlist,
20    parse_playlist,
21    parse_track,
22)
23
24if TYPE_CHECKING:
25    from music_assistant_models.media_items import Album, Artist, Playlist, Track
26
27    from .provider import TidalProvider
28
29
30class TidalMediaManager:
31    """Handles retrieval of media items from Tidal."""
32
33    def __init__(self, provider: TidalProvider):
34        """Initialize media retriever."""
35        self.provider = provider
36        self.api = provider.api
37        self.logger = provider.logger
38
39    async def search(
40        self, search_query: str, media_types: list[MediaType], limit: int = 5
41    ) -> SearchResults:
42        """Perform search on Tidal."""
43        parsed_results = SearchResults()
44        media_type_strings = []
45
46        if MediaType.ARTIST in media_types:
47            media_type_strings.append("artists")
48        if MediaType.ALBUM in media_types:
49            media_type_strings.append("albums")
50        if MediaType.TRACK in media_types:
51            media_type_strings.append("tracks")
52        if MediaType.PLAYLIST in media_types:
53            media_type_strings.append("playlists")
54
55        if not media_type_strings:
56            return parsed_results
57
58        results = await self.api.get(
59            "search",
60            params={
61                "query": search_query.replace("'", ""),
62                "limit": limit,
63                "types": ",".join(media_type_strings),
64            },
65        )
66
67        if "artists" in results and results["artists"].get("items"):
68            parsed_results.artists = [
69                parse_artist(self.provider, x) for x in results["artists"]["items"]
70            ]
71        if "albums" in results and results["albums"].get("items"):
72            parsed_results.albums = [
73                parse_album(self.provider, x) for x in results["albums"]["items"]
74            ]
75        if "playlists" in results and results["playlists"].get("items"):
76            parsed_results.playlists = [
77                parse_playlist(self.provider, x) for x in results["playlists"]["items"]
78            ]
79        if "tracks" in results and results["tracks"].get("items"):
80            parsed_results.tracks = [
81                parse_track(self.provider, x) for x in results["tracks"]["items"]
82            ]
83        return parsed_results
84
85    async def get_artist(self, prov_artist_id: str) -> Artist:
86        """Get artist details."""
87        try:
88            data = await self.api.get(f"artists/{prov_artist_id}")
89            return parse_artist(self.provider, data)
90        except (ClientError, KeyError, ValueError) as err:
91            raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
92
93    async def get_album(self, prov_album_id: str) -> Album:
94        """Get album details."""
95        try:
96            data = await self.api.get(f"albums/{prov_album_id}")
97            return parse_album(self.provider, data)
98        except (ClientError, KeyError, ValueError) as err:
99            raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
100
101    async def get_track(self, prov_track_id: str) -> Track:
102        """Get track details."""
103        try:
104            # Fetch the track first so a nonexistent track doesn't cost an
105            # extra (throttled) lyrics request.
106            track_obj = await self.api.get(f"tracks/{prov_track_id}")
107            lyrics = await self._get_lyrics(prov_track_id)
108            return parse_track(self.provider, track_obj, lyrics=lyrics)
109        except (ClientError, KeyError, ValueError) as err:
110            raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
111
112    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
113        """Get playlist details."""
114        if prov_playlist_id == FAVORITE_TRACKS_PLAYLIST_ID:
115            return parse_favorite_tracks_playlist(self.provider)
116
117        if prov_playlist_id.startswith("mix_"):
118            return await self._get_mix_details(prov_playlist_id[4:])
119
120        try:
121            data = await self.api.get(f"{PLAYLISTS}/{prov_playlist_id}")
122            return parse_playlist(self.provider, data)
123        except MediaNotFoundError:
124            return await self._get_mix_details(prov_playlist_id)
125        except (ClientError, KeyError, ValueError) as err:
126            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") from err
127
128    async def _get_mix_details(self, prov_mix_id: str) -> Playlist:
129        """Get details for a Tidal Mix."""
130        try:
131            params = {"mixId": prov_mix_id, "deviceType": "BROWSER"}
132            tidal_mix = await self.api.get(PAGES_MIX, params=params)
133
134            mix_obj = {
135                "id": prov_mix_id,
136                "title": tidal_mix.get("title", "Unknown Mix"),
137                "updated": tidal_mix.get("lastUpdated", ""),
138                "images": {},
139            }
140
141            # Try to extract images from rows/modules structure
142            rows = tidal_mix.get("rows", [])
143            if rows and (modules := rows[0].get("modules")):
144                if mix_data := modules[0].get("mix"):
145                    mix_obj["images"] = mix_data.get("images", {})
146
147            if "subTitle" not in mix_obj:
148                mix_obj["subTitle"] = tidal_mix.get("subTitle", "")
149
150            return parse_playlist(self.provider, mix_obj, is_mix=True)
151        except (ClientError, KeyError, ValueError) as err:
152            raise MediaNotFoundError(f"Mix {prov_mix_id} not found") from err
153
154    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
155        """Get album tracks."""
156        try:
157            data = await self.api.get(f"albums/{prov_album_id}/tracks", params={"limit": 250})
158            return [parse_track(self.provider, x) for x in data.get("items", [])]
159        except (ClientError, KeyError, ValueError) as err:
160            raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
161
162    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
163        """Get artist albums."""
164        try:
165            data = await self.api.get(f"artists/{prov_artist_id}/albums", params={"limit": 250})
166            return [parse_album(self.provider, x) for x in data.get("items", [])]
167        except (ClientError, KeyError, ValueError) as err:
168            raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
169
170    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
171        """Get artist top tracks."""
172        try:
173            data = await self.api.get(
174                f"artists/{prov_artist_id}/toptracks", params={"limit": 10, "offset": 0}
175            )
176            return [parse_track(self.provider, x) for x in data.get("items", [])]
177        except (ClientError, KeyError, ValueError) as err:
178            raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
179
180    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
181        """Get similar tracks."""
182        try:
183            data = await self.api.get(f"tracks/{prov_track_id}/radio", params={"limit": limit})
184            return [parse_track(self.provider, x) for x in data.get("items", [])]
185        except (ClientError, KeyError, ValueError) as err:
186            raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
187
188    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
189        """Get playlist tracks."""
190        page_size = 200
191        offset = page * page_size
192
193        if prov_playlist_id == FAVORITE_TRACKS_PLAYLIST_ID:
194            return await self._get_favorite_tracks(page_size, offset)
195
196        if prov_playlist_id.startswith("mix_"):
197            return await self._get_mix_tracks(prov_playlist_id[4:], page_size, offset)
198
199        try:
200            data = await self.api.get(
201                f"{PLAYLISTS}/{prov_playlist_id}/tracks",
202                params={"limit": page_size, "offset": offset},
203            )
204            return self._process_tracks(data.get("items", []), offset)
205        except MediaNotFoundError:
206            return await self._get_mix_tracks(prov_playlist_id, page_size, offset)
207
208    async def _get_favorite_tracks(self, limit: int, offset: int) -> list[Track]:
209        """Get the user's favorite tracks in descending order (newest first)."""
210        try:
211            data = await self.api.get(
212                f"users/{self.provider.auth.user_id}/{FAVORITES_TRACKS}",
213                params={
214                    "limit": limit,
215                    "offset": offset,
216                    "order": "DATE",
217                    "orderDirection": "DESC",
218                },
219            )
220            return self._process_tracks(data.get("items", []), offset)
221        except (ClientError, KeyError, ValueError) as err:
222            raise MediaNotFoundError("Starred tracks not found") from err
223
224    async def _get_mix_tracks(self, mix_id: str, limit: int, offset: int) -> list[Track]:
225        """Get tracks from a mix."""
226        try:
227            params = {"mixId": mix_id, "deviceType": "BROWSER"}
228            data = await self.api.get(PAGES_MIX, params=params)
229
230            # Mix tracks are usually in the second row
231            rows = data.get("rows", [])
232            if len(rows) < 2:
233                raise MediaNotFoundError(f"Mix {mix_id} has no tracks")
234
235            modules = rows[1].get("modules", [])
236            if not modules or "pagedList" not in modules[0]:
237                raise MediaNotFoundError(f"Mix {mix_id} has no tracks")
238
239            all_items = modules[0]["pagedList"].get("items", [])
240            # Manual pagination for mixes
241            paged_items = all_items[offset : offset + limit]
242            return self._process_tracks(paged_items, offset)
243        except (ClientError, KeyError, ValueError) as err:
244            raise MediaNotFoundError(f"Mix {mix_id} not found") from err
245
246    async def _get_lyrics(self, prov_track_id: str) -> dict[str, str] | None:
247        """Get lyrics for a track, returning None when unavailable."""
248        # Lyrics are optional enrichment: never fail the track lookup on
249        # a missing/failed lyrics response.
250        try:
251            return await self.api.get(f"tracks/{prov_track_id}/lyrics")
252        except (ClientError, MusicAssistantError) as err:
253            self.logger.debug("Failed to fetch lyrics for track %s: %s", prov_track_id, err)
254            return None
255
256    def _process_tracks(self, items: list[dict[str, Any]], offset: int) -> list[Track]:
257        result = []
258        for idx, item in enumerate(items, 1):
259            try:
260                track = parse_track(self.provider, item)
261                track.position = offset + idx
262                result.append(track)
263            except KeyError, TypeError:
264                continue
265        return result
266