/
/
/
1"""Parsers for official Tidal API (openapi.tidal.com/v2) JSON:API responses."""
2
3from __future__ import annotations
4
5import logging
6import re
7from collections.abc import Mapping
8from contextlib import suppress
9from datetime import datetime
10from typing import TYPE_CHECKING, Any
11
12from music_assistant_models.enums import (
13 AlbumType,
14 ContentType,
15 ExternalID,
16 ImageType,
17 LinkType,
18 MediaType,
19)
20from music_assistant_models.media_items import (
21 Album,
22 Artist,
23 AudioFormat,
24 AudioMetadata,
25 MediaItemImage,
26 MediaItemLink,
27 Playlist,
28 ProviderMapping,
29 Track,
30 UniqueList,
31)
32
33from music_assistant.helpers.util import infer_album_type, parse_title_and_version
34
35from .constants import SKIPPABLE_ITEM_ERRORS
36
37if TYPE_CHECKING:
38 from collections.abc import Callable
39
40 from ._openapi_models import (
41 AlbumsAttributes,
42 ArtistsAttributes,
43 PlaylistsAttributes,
44 TracksAttributes,
45 )
46 from .jsonapi import JsonApiDocument
47 from .provider import TidalProvider
48
49# Preferred artwork width; Tidal serves square art in a range of sizes.
50_PREFERRED_IMAGE_WIDTH = 750
51
52_ISO_DURATION_RE = re.compile(
53 r"^P(?:\d+Y)?(?:\d+M)?(?:\d+W)?(?:\d+D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?$"
54)
55
56# Tidal biographies embed internal cross-reference markup, e.g.
57# [wimpLink artistId="123"]Some Artist[/wimpLink]. Strip the tags, keep the text.
58_WIMP_LINK_RE = re.compile(r"\[/?wimpLink[^\]]*\]")
59
60# Tidal externalLinks meta.type -> generic LinkType. Types not listed here
61# (Tidal share links, autoplay and payment/claim redirects) are not exposed.
62_LINK_TYPE = {
63 "OFFICIAL_HOMEPAGE": LinkType.WEBSITE,
64 "FACEBOOK": LinkType.FACEBOOK,
65 "TWITTER": LinkType.TWITTER,
66 "INSTAGRAM": LinkType.INSTAGRAM,
67 "TIKTOK": LinkType.TIKTOK,
68 "SNAPCHAT": LinkType.SNAPCHAT,
69}
70
71# Tidal key enum -> pitch-class notation used in Metadata.musical_key.
72_KEY_PITCH = {
73 "C": "C",
74 "CSharp": "C#",
75 "D": "D",
76 "Eb": "Eb",
77 "E": "E",
78 "F": "F",
79 "FSharp": "F#",
80 "G": "G",
81 "Ab": "Ab",
82 "A": "A",
83 "Bb": "Bb",
84 "B": "B",
85}
86
87
88def parse_artist(provider: TidalProvider, doc: JsonApiDocument, resource: dict[str, Any]) -> Artist:
89 """Parse an official Tidal artist resource to a generic Artist."""
90 artist_id = str(resource["id"])
91 attributes: ArtistsAttributes = resource.get("attributes", {})
92 artist = Artist(
93 item_id=artist_id,
94 provider=provider.instance_id,
95 name=attributes.get("name", ""),
96 provider_mappings={
97 ProviderMapping(
98 item_id=artist_id,
99 provider_domain=provider.domain,
100 provider_instance=provider.instance_id,
101 url=f"https://tidal.com/artist/{artist_id}",
102 )
103 },
104 )
105 if attributes.get("popularity") is not None:
106 artist.metadata.popularity = _scale_popularity(attributes["popularity"])
107 if links := _parse_links(attributes):
108 artist.metadata.links = links
109 if image := _resolve_image(provider, doc, resource, "profileArt"):
110 artist.metadata.images = UniqueList([image])
111 if biography := doc.related_one(resource, "biography"):
112 if text := biography.get("attributes", {}).get("text"):
113 artist.metadata.description = _clean_biography(text)
114 return artist
115
116
117def parse_album(provider: TidalProvider, doc: JsonApiDocument, resource: dict[str, Any]) -> Album:
118 """Parse an official Tidal album resource to a generic Album."""
119 album_id = str(resource["id"])
120 attributes: AlbumsAttributes = resource.get("attributes", {})
121 name, version = _split_title_version(
122 attributes.get("title", "Unknown Album"), attributes.get("version") or None
123 )
124 availability = attributes.get("availability")
125 available = "STREAM" in availability if availability is not None else True
126 album = Album(
127 item_id=album_id,
128 provider=provider.instance_id,
129 name=name,
130 version=version,
131 provider_mappings={
132 ProviderMapping(
133 item_id=album_id,
134 provider_domain=provider.domain,
135 provider_instance=provider.instance_id,
136 audio_format=AudioFormat(content_type=ContentType.FLAC),
137 url=f"https://tidal.com/album/{album_id}",
138 available=available,
139 )
140 },
141 )
142
143 various_artists = False
144 for artist_resource in doc.related(resource, "artists"):
145 artist = parse_artist(provider, doc, artist_resource)
146 if artist.name == "Various Artists":
147 various_artists = True
148 album.artists.append(artist)
149
150 album.album_type = _map_album_type(attributes.get("albumType"), name, version, various_artists)
151
152 if release_date := attributes.get("releaseDate"):
153 with suppress(ValueError, IndexError):
154 album.year = int(release_date.split("-")[0])
155 with suppress(ValueError):
156 album.metadata.release_date = datetime.fromisoformat(release_date)
157
158 if barcode := attributes.get("barcodeId"):
159 album.external_ids.add((ExternalID.BARCODE, barcode))
160 if copyright_data := attributes.get("copyright"):
161 album.metadata.copyright = copyright_data.get("text", "")
162 album.metadata.explicit = attributes.get("explicit", False)
163 if attributes.get("popularity") is not None:
164 album.metadata.popularity = _scale_popularity(attributes["popularity"])
165 if genres := _parse_genres(doc, resource):
166 album.metadata.genres = genres
167 if links := _parse_links(attributes):
168 album.metadata.links = links
169 if image := _resolve_image(provider, doc, resource, "coverArt"):
170 album.metadata.images = UniqueList([image])
171
172 return album
173
174
175def parse_track(provider: TidalProvider, doc: JsonApiDocument, resource: dict[str, Any]) -> Track:
176 """Parse an official Tidal track resource to a generic Track."""
177 track_id = str(resource["id"])
178 attributes: TracksAttributes = resource.get("attributes", {})
179 name, version = _split_title_version(
180 attributes.get("title", "Unknown"), attributes.get("version") or None
181 )
182 hi_res_lossless = "HIRES_LOSSLESS" in (attributes.get("mediaTags") or [])
183 availability = attributes.get("availability")
184 available = "STREAM" in availability if availability is not None else True
185 track = Track(
186 item_id=track_id,
187 provider=provider.instance_id,
188 name=name,
189 version=version,
190 duration=_parse_iso_duration(attributes.get("duration", "")),
191 provider_mappings={
192 ProviderMapping(
193 item_id=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=available,
202 )
203 },
204 )
205
206 if isrc := attributes.get("isrc"):
207 track.external_ids.add((ExternalID.ISRC, isrc))
208
209 track.artists = UniqueList(
210 [
211 parse_artist(provider, doc, artist_resource)
212 for artist_resource in doc.related(resource, "artists")
213 ]
214 )
215
216 track.metadata.explicit = attributes.get("explicit", False)
217 if attributes.get("popularity") is not None:
218 track.metadata.popularity = _scale_popularity(attributes["popularity"])
219 if copyright_data := attributes.get("copyright"):
220 track.metadata.copyright = copyright_data.get("text", "")
221
222 if genres := _parse_genres(doc, resource):
223 track.metadata.genres = genres
224 if links := _parse_links(attributes):
225 track.metadata.links = links
226 if performers := _parse_credits(doc, resource):
227 track.metadata.performers = performers
228
229 bpm = attributes.get("bpm")
230 musical_key = _parse_musical_key(attributes.get("key"), attributes.get("keyScale"))
231 if bpm is not None or musical_key:
232 track.audio_metadata = AudioMetadata(bpm=bpm, musical_key=musical_key)
233
234 # The album relationship carries a minimal album resource; use an ItemMapping
235 # (as with the unofficial API) and take the track image from the album cover.
236 if album_resource := doc.related_one(resource, "albums"):
237 album_attributes: dict[str, Any] = album_resource.get("attributes", {})
238 track.album = provider.get_item_mapping(
239 media_type=MediaType.ALBUM,
240 key=str(album_resource["id"]),
241 name=album_attributes.get("title") or "",
242 )
243 if image := _resolve_image(provider, doc, album_resource, "coverArt"):
244 track.metadata.images = UniqueList([image])
245
246 return track
247
248
249def parse_playlist(
250 provider: TidalProvider, doc: JsonApiDocument, resource: dict[str, Any]
251) -> Playlist:
252 """Parse an official Tidal playlist resource to a generic Playlist."""
253 raw_id = str(resource["id"])
254 attributes: PlaylistsAttributes = resource.get("attributes", {})
255 # Mixes are exposed as playlists but keep the "mix_" item id so they open via
256 # the existing (unofficial) mix flow.
257 is_mix = attributes.get("playlistType") == "MIX"
258 if is_mix:
259 playlist_id = f"mix_{raw_id}"
260 owner_name = "Created by Tidal"
261 is_editable = False
262 url = f"https://tidal.com/mix/{raw_id}"
263 else:
264 playlist_id = raw_id
265 # A playlist is editable when the authenticated user is one of its owners.
266 # This needs the "owners" relationship to be present; search omits it (it
267 # would exceed the include cap), so search results are non-editable there.
268 owner_ids = doc.linkage_ids(resource, "owners")
269 user_id = str(provider.auth.user_id) if provider.auth.user_id else None
270 is_editable = bool(user_id and user_id in owner_ids)
271 owner_name = "Tidal"
272 if is_editable:
273 owner_name = (
274 provider.auth.user.profile_name or provider.auth.user.user_name or str(user_id)
275 )
276 url = f"https://tidal.com/playlist/{raw_id}"
277
278 playlist = Playlist(
279 item_id=playlist_id,
280 provider=provider.instance_id,
281 name=attributes.get("name", "Unknown"),
282 owner=owner_name,
283 provider_mappings={
284 ProviderMapping(
285 item_id=playlist_id,
286 provider_domain=provider.domain,
287 provider_instance=provider.instance_id,
288 url=url,
289 is_unique=is_editable,
290 )
291 },
292 is_editable=is_editable,
293 )
294 if description := attributes.get("description"):
295 playlist.metadata.description = description
296 if image := _resolve_image(provider, doc, resource, "coverArt"):
297 playlist.metadata.images = UniqueList([image])
298 return playlist
299
300
301def _parse_items[ItemT](
302 parser: Callable[[TidalProvider, JsonApiDocument, dict[str, Any]], ItemT],
303 provider: TidalProvider,
304 doc: JsonApiDocument,
305) -> list[ItemT]:
306 """
307 Parse the document's primary collection, leaving out items that cannot be parsed.
308
309 Only use this for a collection of a single resource type.
310
311 :param parser: The parser to apply to each resolved resource.
312 :param provider: The Tidal provider instance.
313 :param doc: The JSON:API document holding the collection.
314 """
315 items: list[ItemT] = []
316 for identifier in doc.data_list:
317 if not (resource := doc.resolve(identifier)):
318 continue
319 if (item := _parse_or_skip(parser, provider, doc, resource)) is not None:
320 items.append(item)
321 return items
322
323
324def _parse_or_skip[ItemT](
325 parser: Callable[[TidalProvider, JsonApiDocument, dict[str, Any]], ItemT],
326 provider: TidalProvider,
327 doc: JsonApiDocument,
328 resource: dict[str, Any],
329 sync_media_type: MediaType | None = None,
330 sync_item_id: str | None = None,
331) -> ItemT | None:
332 """
333 Parse a resource into a media item, or return None if it cannot be parsed.
334
335 :param parser: The parser to apply to the resource.
336 :param provider: The Tidal provider instance.
337 :param doc: The JSON:API document holding the resource.
338 :param resource: The resource object to parse.
339 :param sync_media_type: The library media type to protect from sync cleanup.
340 :param sync_item_id: The provider item ID to protect from sync cleanup.
341 """
342 try:
343 return parser(provider, doc, resource)
344 except SKIPPABLE_ITEM_ERRORS as err:
345 if sync_media_type is not None:
346 provider.report_skipped_sync_item(
347 sync_media_type,
348 sync_item_id,
349 err,
350 )
351 else:
352 provider.logger.warning(
353 "Skipping Tidal %s %s: %s",
354 resource.get("type", "item"),
355 resource.get("id", "[no id]"),
356 err,
357 exc_info=err if provider.logger.isEnabledFor(logging.DEBUG) else None,
358 )
359 return None
360
361
362def _split_title_version(title: str, version: str | None) -> tuple[str, str]:
363 """
364 Split a title into (name, version).
365
366 The official API provides an explicit version qualifier (e.g. "International
367 Version"). When present, trust it: strip that exact qualifier from the end of
368 the title to form the name. This is more reliable than the shared heuristic,
369 which guesses the version from bracketed parts of the title and mishandles
370 titles carrying more than one parenthetical. Fall back to the heuristic only
371 when no explicit version is given.
372 """
373 if not version:
374 return parse_title_and_version(title)
375 # Strip a trailing "(<version>)", "[<version>]" or " - <version>" qualifier.
376 pattern = rf"\s*(?:-\s*)?[(\[]?\s*{re.escape(version)}\s*[)\]]?\s*$"
377 name = re.sub(pattern, "", title, flags=re.IGNORECASE).strip()
378 return name or title, version
379
380
381def _clean_biography(text: str) -> str:
382 """Strip Tidal's internal [wimpLink] cross-reference markup from bio text."""
383 return _WIMP_LINK_RE.sub("", text).strip()
384
385
386def _parse_credits(doc: JsonApiDocument, resource: dict[str, Any]) -> set[str] | None:
387 """Resolve the credits relationship to a set of contributor names."""
388 names = {
389 name
390 for credit in doc.related(resource, "credits")
391 if (name := credit.get("attributes", {}).get("name"))
392 }
393 return names or None
394
395
396def _parse_genres(doc: JsonApiDocument, resource: dict[str, Any]) -> set[str] | None:
397 """Resolve the genres relationship to a set of genre names."""
398 genres = {
399 name
400 for genre in doc.related(resource, "genres")
401 if (name := genre.get("attributes", {}).get("genreName"))
402 }
403 return genres or None
404
405
406def _parse_links(attributes: Mapping[str, Any]) -> set[MediaItemLink] | None:
407 """Map the externalLinks attribute to a set of generic MediaItemLinks."""
408 links = {
409 MediaItemLink(type=link_type, url=href)
410 for link in attributes.get("externalLinks") or []
411 if (link_type := _LINK_TYPE.get(link.get("meta", {}).get("type", "")))
412 and (href := link.get("href"))
413 }
414 return links or None
415
416
417def _scale_popularity(value: float) -> int:
418 """Convert the official API's 0..1 popularity to the 0..100 scale MA uses."""
419 return round(value * 100)
420
421
422def _parse_iso_duration(value: str) -> int:
423 """Parse an ISO-8601 duration (e.g. "PT3M57S") into whole seconds."""
424 if not value or not (match := _ISO_DURATION_RE.match(value)):
425 return 0
426 hours, minutes, seconds = match.groups()
427 total = int(hours or 0) * 3600 + int(minutes or 0) * 60 + float(seconds or 0)
428 return int(total)
429
430
431def _parse_musical_key(key: str | None, scale: str | None) -> str | None:
432 """Build a Metadata.musical_key value (e.g. "F# minor") from Tidal's enums."""
433 pitch = _KEY_PITCH.get(key or "")
434 if not pitch:
435 return None
436 if scale and scale != "UNKNOWN":
437 return f"{pitch} {scale.lower().replace('_', ' ')}"
438 return pitch
439
440
441def _map_album_type(
442 album_type: str | None, name: str, version: str | None, various_artists: bool
443) -> AlbumType:
444 """Map the official albumType (plus inference) to a generic AlbumType."""
445 if various_artists:
446 return AlbumType.COMPILATION
447 inferred = infer_album_type(name, version or "")
448 if inferred in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
449 return inferred
450 return {
451 "ALBUM": AlbumType.ALBUM,
452 "EP": AlbumType.EP,
453 "SINGLE": AlbumType.SINGLE,
454 }.get(album_type or "ALBUM", AlbumType.ALBUM)
455
456
457def _resolve_image(
458 provider: TidalProvider,
459 doc: JsonApiDocument,
460 resource: dict[str, Any],
461 relationship: str,
462) -> MediaItemImage | None:
463 """Resolve an artwork relationship to a MediaItemImage, if available."""
464 artwork = doc.related_one(resource, relationship)
465 if not artwork:
466 return None
467 files = artwork.get("attributes", {}).get("files") or []
468 if not (url := _select_image_url(files)):
469 return None
470 return MediaItemImage(
471 type=ImageType.THUMB,
472 path=url,
473 provider=provider.instance_id,
474 remotely_accessible=True,
475 )
476
477
478def _select_image_url(files: list[dict[str, Any]]) -> str | None:
479 """Pick the artwork file nearest the preferred width (preferring larger)."""
480 usable = [f for f in files if f.get("href")]
481 if not usable:
482 return None
483
484 def sort_key(file: dict[str, Any]) -> tuple[bool, int]:
485 width = file.get("meta", {}).get("width", 0)
486 # Prefer the smallest file at or above the preferred width; if none reach
487 # it, fall back to the largest available.
488 below = width < _PREFERRED_IMAGE_WIDTH
489 distance = abs(width - _PREFERRED_IMAGE_WIDTH)
490 return (below, distance)
491
492 return str(min(usable, key=sort_key)["href"])
493