music-assistant-server
12 KB•PY
parsers.py
12 KB • 340 lines • python
1"""Parsers for Tidal API responses."""
2
3from __future__ import annotations
4
5from contextlib import suppress
6from datetime import datetime
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import (
10 AlbumType,
11 ContentType,
12 ExternalID,
13 ImageType,
14 MediaType,
15)
16from music_assistant_models.media_items import (
17 Album,
18 Artist,
19 AudioFormat,
20 MediaItemImage,
21 Playlist,
22 ProviderMapping,
23 Track,
24 UniqueList,
25)
26
27from music_assistant.helpers.util import infer_album_type, parse_title_and_version
28
29from .constants import BROWSE_URL, RESOURCES_URL
30
31if TYPE_CHECKING:
32 from .provider import TidalProvider
33
34
35def parse_artist(provider: TidalProvider, artist_obj: dict[str, Any]) -> Artist:
36 """Parse tidal artist object to generic layout."""
37 # Handle both full artist objects and nested ones coming from albums/tracks
38 artist_obj_data = artist_obj.get("item", artist_obj)
39 artist_id = str(artist_obj_data["id"])
40 artist = Artist(
41 item_id=artist_id,
42 provider=provider.instance_id,
43 name=artist_obj_data["name"],
44 provider_mappings={
45 ProviderMapping(
46 item_id=artist_id,
47 provider_domain=provider.domain,
48 provider_instance=provider.instance_id,
49 # NOTE: don't use the /browse endpoint as it's
50 # not working for musicbrainz lookups
51 url=f"https://tidal.com/artist/{artist_id}",
52 )
53 },
54 )
55 # metadata
56 if "created" in artist_obj:
57 with suppress(ValueError):
58 artist.date_added = datetime.fromisoformat(artist_obj["created"])
59 if artist_obj_data["picture"]:
60 picture_id = artist_obj_data["picture"].replace("-", "/")
61 image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
62 artist.metadata.images = UniqueList(
63 [
64 MediaItemImage(
65 type=ImageType.THUMB,
66 path=image_url,
67 provider=provider.instance_id,
68 remotely_accessible=True,
69 )
70 ]
71 )
72
73 return artist
74
75
76def parse_album(provider: TidalProvider, album_obj: dict[str, Any]) -> Album:
77 """Parse tidal album object to generic layout."""
78 album_obj_data = album_obj.get("item", album_obj)
79 name, version = parse_title_and_version(
80 album_obj_data.get("title", "Unknown Album"),
81 album_obj_data.get("version") or None,
82 )
83 album_id = str(album_obj_data.get("id", ""))
84
85 album = Album(
86 item_id=album_id,
87 provider=provider.instance_id,
88 name=name,
89 version=version,
90 provider_mappings={
91 ProviderMapping(
92 item_id=album_id,
93 provider_domain=provider.domain,
94 provider_instance=provider.instance_id,
95 audio_format=AudioFormat(
96 content_type=ContentType.FLAC,
97 ),
98 url=f"https://tidal.com/album/{album_id}",
99 available=album_obj.get("streamReady", True), # Default to available
100 )
101 },
102 )
103
104 # Safely handle artists array
105 various_artist_album: bool = False
106 for artist_obj in album_obj_data.get("artists", []):
107 try:
108 if artist_obj.get("name") == "Various Artists":
109 various_artist_album = True
110 album.artists.append(parse_artist(provider, artist_obj))
111 except (KeyError, TypeError) as err:
112 provider.logger.warning("Error parsing artist in album %s: %s", name, err)
113
114 # Safely determine album type
115 album_type = album_obj_data.get("type", "ALBUM")
116 if album_type == "COMPILATION" or various_artist_album:
117 album.album_type = AlbumType.COMPILATION
118 elif album_type == "ALBUM":
119 album.album_type = AlbumType.ALBUM
120 elif album_type == "EP":
121 album.album_type = AlbumType.EP
122 elif album_type == "SINGLE":
123 album.album_type = AlbumType.SINGLE
124
125 # Try inference - override if it finds something more specific
126 inferred_type = infer_album_type(name, version)
127 if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
128 album.album_type = inferred_type
129
130 # Safely parse year
131 if release_date := album_obj_data.get("releaseDate", ""):
132 try:
133 album.year = int(release_date.split("-")[0])
134 except (ValueError, IndexError):
135 provider.logger.debug("Invalid release date format: %s", release_date)
136 with suppress(ValueError):
137 album.metadata.release_date = datetime.fromisoformat(release_date)
138
139 # Safely set metadata
140 if "created" in album_obj:
141 with suppress(ValueError):
142 album.date_added = datetime.fromisoformat(album_obj["created"])
143 upc = album_obj_data.get("upc")
144 if upc:
145 album.external_ids.add((ExternalID.BARCODE, upc))
146
147 album.metadata.copyright = album_obj_data.get("copyright", "")
148 album.metadata.explicit = album_obj_data.get("explicit", False)
149 album.metadata.popularity = album_obj_data.get("popularity", 0)
150
151 # Safely handle cover image
152 cover = album_obj_data.get("cover")
153 if cover:
154 picture_id = cover.replace("-", "/")
155 image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
156 album.metadata.images = UniqueList(
157 [
158 MediaItemImage(
159 type=ImageType.THUMB,
160 path=image_url,
161 provider=provider.instance_id,
162 remotely_accessible=True,
163 )
164 ]
165 )
166
167 return album
168
169
170def parse_track(
171 provider: TidalProvider,
172 track_obj: dict[str, Any],
173 lyrics: dict[str, str] | None = None,
174) -> Track:
175 """Parse tidal track object to generic layout."""
176 track_obj_data = track_obj.get("item", track_obj)
177 name, version = parse_title_and_version(
178 track_obj_data.get("title", "Unknown"),
179 track_obj_data.get("version") or None,
180 )
181 track_id = str(track_obj_data.get("id", 0))
182 media_metadata = track_obj_data.get("mediaMetadata") or {}
183 tags = media_metadata.get("tags", [])
184 hi_res_lossless = any(tag in tags for tag in ["HIRES_LOSSLESS", "HI_RES_LOSSLESS"])
185 track = Track(
186 item_id=track_id,
187 provider=provider.instance_id,
188 name=name,
189 version=version,
190 duration=track_obj_data.get("duration", 0),
191 provider_mappings={
192 ProviderMapping(
193 item_id=str(track_id),
194 provider_domain=provider.domain,
195 provider_instance=provider.instance_id,
196 audio_format=AudioFormat(
197 content_type=ContentType.FLAC,
198 bit_depth=24 if hi_res_lossless else 16,
199 ),
200 url=f"https://tidal.com/track/{track_id}",
201 available=track_obj_data["streamReady"],
202 )
203 },
204 disc_number=track_obj_data.get("volumeNumber", 0) or 0,
205 track_number=track_obj_data.get("trackNumber", 0) or 0,
206 )
207 if "isrc" in track_obj_data:
208 track.external_ids.add((ExternalID.ISRC, track_obj_data["isrc"]))
209 track.artists = UniqueList()
210 for track_artist in track_obj_data["artists"]:
211 artist = parse_artist(provider, track_artist)
212 track.artists.append(artist)
213 # metadata
214 if "created" in track_obj:
215 with suppress(ValueError):
216 track.date_added = datetime.fromisoformat(track_obj["created"])
217 track.metadata.explicit = track_obj_data["explicit"]
218 track.metadata.popularity = track_obj_data["popularity"]
219 if "copyright" in track_obj_data:
220 track.metadata.copyright = track_obj_data["copyright"]
221 if lyrics and "lyrics" in lyrics:
222 track.metadata.lyrics = lyrics["lyrics"]
223 if lyrics and "subtitles" in lyrics:
224 track.metadata.lrc_lyrics = lyrics["subtitles"]
225 if track_obj_data["album"]:
226 # Here we use an ItemMapping as Tidal returns
227 # minimal data when getting an Album from a Track
228 track.album = provider.get_item_mapping(
229 media_type=MediaType.ALBUM,
230 key=str(track_obj_data["album"]["id"]),
231 name=track_obj_data["album"]["title"],
232 )
233 if track_obj_data["album"]["cover"]:
234 picture_id = track_obj_data["album"]["cover"].replace("-", "/")
235 image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
236 track.metadata.images = UniqueList(
237 [
238 MediaItemImage(
239 type=ImageType.THUMB,
240 path=image_url,
241 provider=provider.instance_id,
242 remotely_accessible=True,
243 )
244 ]
245 )
246 return track
247
248
249def parse_playlist(
250 provider: TidalProvider, playlist_obj: dict[str, Any], is_mix: bool = False
251) -> Playlist:
252 """Parse tidal playlist object to generic layout."""
253 playlist_obj_data = playlist_obj.get("playlist", playlist_obj)
254 # Get ID based on playlist type
255 raw_id = str(playlist_obj_data.get("id" if is_mix else "uuid", ""))
256
257 # Add prefix for mixes to distinguish them
258 playlist_id = f"mix_{raw_id}" if is_mix else raw_id
259
260 # Owner logic differs between types
261 if is_mix:
262 owner_name = "Created by Tidal"
263 is_editable = False
264 else:
265 creator_id = None
266 creator = playlist_obj_data.get("creator", {})
267 if creator:
268 creator_id = creator.get("id")
269 is_editable = bool(creator_id and str(creator_id) == str(provider.auth.user_id))
270
271 owner_name = "Tidal"
272 if is_editable:
273 if provider.auth.user.profile_name:
274 owner_name = provider.auth.user.profile_name
275 elif provider.auth.user.user_name:
276 owner_name = provider.auth.user.user_name
277 elif provider.auth.user_id:
278 owner_name = str(provider.auth.user_id)
279
280 # URL path differs by type - use raw_id for URLs
281 url_path = "mix" if is_mix else "playlist"
282
283 playlist = Playlist(
284 item_id=playlist_id,
285 provider=provider.instance_id,
286 name=playlist_obj_data.get("title", "Unknown"),
287 owner=owner_name,
288 provider_mappings={
289 ProviderMapping(
290 item_id=playlist_id, # Use raw ID for provider mapping
291 provider_domain=provider.domain,
292 provider_instance=provider.instance_id,
293 url=f"{BROWSE_URL}/{url_path}/{raw_id}",
294 is_unique=is_editable, # user-owned playlists are unique
295 )
296 },
297 is_editable=is_editable,
298 )
299
300 # Metadata - different fields based on type
301 if "created" in playlist_obj:
302 with suppress(ValueError):
303 playlist.date_added = datetime.fromisoformat(playlist_obj["created"])
304 # Add the description from the subtitle for mixes
305 if is_mix:
306 subtitle = playlist_obj_data.get("subTitle")
307 if subtitle:
308 playlist.metadata.description = subtitle
309
310 # Handle images differently based on type
311 if is_mix:
312 if pictures := playlist_obj_data.get("images", {}).get("MEDIUM"):
313 image_url = pictures.get("url", "")
314 if image_url:
315 playlist.metadata.images = UniqueList(
316 [
317 MediaItemImage(
318 type=ImageType.THUMB,
319 path=image_url,
320 provider=provider.instance_id,
321 remotely_accessible=True,
322 )
323 ]
324 )
325 elif picture := (playlist_obj_data.get("squareImage") or playlist_obj_data.get("image")):
326 picture_id = picture.replace("-", "/")
327 image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
328 playlist.metadata.images = UniqueList(
329 [
330 MediaItemImage(
331 type=ImageType.THUMB,
332 path=image_url,
333 provider=provider.instance_id,
334 remotely_accessible=True,
335 )
336 ]
337 )
338
339 return playlist
340