music-assistant-server

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