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