music-assistant-server

1"""Parsers for Apple Music API response objects."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast
6from urllib.parse import urlparse
7
8from music_assistant_models.enums import AlbumType, ContentType, ExternalID, ImageType, MediaType
9from music_assistant_models.media_items import (
10    Album,
11    Artist,
12    AudioFormat,
13    ItemMapping,
14    MediaItemImage,
15    Playlist,
16    ProviderMapping,
17    Track,
18    UniqueList,
19)
20
21from music_assistant.helpers.util import (
22    infer_album_type,
23    normalize_unicode,
24    parse_title_and_version,
25)
26
27from .constants import BLOBSTORE_DOMAIN, MAX_ARTWORK_DIMENSION, UNKNOWN_PLAYLIST_NAME
28from .helpers.utils import is_library_id
29
30if TYPE_CHECKING:
31    from .provider import AppleMusicProvider
32
33
34def is_remotely_accessible_artwork_url(url: str) -> bool:
35    """
36    Check if artwork URL is remotely accessible without caching.
37
38    Blobstore URLs have AWS signatures that expire after 24h, so they must be cached immediately.
39    mzstatic.com URLs are permanent CDN URLs that can be used directly.
40
41    :param url: The artwork URL to check.
42    :return: True if the URL is remotely accessible (permanent), False if it needs caching.
43    """
44    hostname = urlparse(url).hostname or ""
45    return BLOBSTORE_DOMAIN not in hostname
46
47
48def format_artwork_url(attributes: dict[str, Any]) -> str | None:
49    """
50    Return the artwork URL from a raw api item object's (resolved) attributes, if any.
51
52    :param attributes: The attributes dict of the raw api item object.
53    """
54    if not (artwork := attributes.get("artwork")):
55        return None
56    if not (url := artwork.get("url")):
57        return None
58    if artwork.get("width") and artwork.get("height"):
59        url = url.format(
60            w=min(artwork["width"], MAX_ARTWORK_DIMENSION),
61            h=min(artwork["height"], MAX_ARTWORK_DIMENSION),
62        )
63    return cast("str", url)
64
65
66def parse_artwork_image(
67    provider: AppleMusicProvider,
68    media_type: MediaType,
69    item_id: str,
70    attributes: dict[str, Any],
71) -> MediaItemImage | None:
72    """
73    Parse the artwork of a raw api item object into a MediaItemImage, if any.
74
75    :param provider: The Apple Music provider instance.
76    :param media_type: The media type of the item the artwork belongs to.
77    :param item_id: The provider item id of the item the artwork belongs to.
78    :param attributes: The (resolved) attributes dict of the raw api item object.
79    """
80    if (url := format_artwork_url(attributes)) is None:
81        return None
82    if is_remotely_accessible_artwork_url(url):
83        return MediaItemImage(
84            provider=provider.instance_id,
85            type=ImageType.THUMB,
86            path=url,
87            remotely_accessible=True,
88        )
89    # blobstore artwork URLs are presigned with a ~24h expiry and must never be
90    # persisted: store a stable token instead, which is resolved to a freshly
91    # signed URL on demand (see AppleMusicProvider.resolve_image)
92    return MediaItemImage(
93        provider=provider.instance_id,
94        type=ImageType.THUMB,
95        path=f"{media_type.value}/{item_id}",
96        remotely_accessible=False,
97    )
98
99
100def parse_artist(provider: AppleMusicProvider, artist_obj: dict[str, Any]) -> Artist | ItemMapping:
101    """Parse artist object to generic layout."""
102    relationships = artist_obj.get("relationships", {})
103    if (
104        artist_obj.get("type") == "library-artists"
105        and relationships.get("catalog", {}).get("data", []) != []
106    ):
107        artist_id = relationships["catalog"]["data"][0]["id"]
108        attributes = relationships["catalog"]["data"][0]["attributes"]
109    elif "attributes" in artist_obj:
110        artist_id = artist_obj["id"]
111        attributes = artist_obj["attributes"]
112    else:
113        artist_id = artist_obj["id"]
114        provider.logger.debug("No attributes found for artist %s", artist_obj)
115        return ItemMapping(
116            media_type=MediaType.ARTIST,
117            provider=provider.instance_id,
118            item_id=artist_id,
119            name=artist_id,
120        )
121    artist = Artist(
122        item_id=artist_id,
123        name=cast("str", normalize_unicode(attributes.get("name"))),
124        provider=provider.domain,
125        provider_mappings={
126            ProviderMapping(
127                item_id=artist_id,
128                provider_domain=provider.domain,
129                provider_instance=provider.instance_id,
130                url=attributes.get("url"),
131            )
132        },
133    )
134    if image := parse_artwork_image(provider, MediaType.ARTIST, artist_id, attributes):
135        artist.metadata.add_image(image)
136    if genres := attributes.get("genreNames"):
137        artist.metadata.genres = set(genres)
138    if notes := attributes.get("editorialNotes"):
139        artist.metadata.description = notes.get("standard") or notes.get("short")
140    return artist
141
142
143def parse_album(
144    provider: AppleMusicProvider,
145    album_obj: dict[str, Any],
146    is_favourite: bool | None = None,
147) -> Album | ItemMapping | None:
148    """Parse album object to generic layout."""
149    relationships = album_obj.get("relationships", {})
150    catalog_data = relationships.get("catalog", {}).get("data", [])
151    response_type = album_obj.get("type")
152    if response_type == "library-albums" and catalog_data != [] and "attributes" in catalog_data[0]:
153        album_id = catalog_data[0]["id"]
154        attributes = catalog_data[0]["attributes"]
155    elif "attributes" in album_obj:
156        album_id = album_obj["id"]
157        attributes = album_obj["attributes"]
158    else:
159        album_id = album_obj["id"]
160        return ItemMapping(
161            media_type=MediaType.ALBUM,
162            provider=provider.instance_id,
163            item_id=album_id,
164            name=album_id,
165        )
166    name, version = parse_title_and_version(attributes["name"])
167    album = Album(
168        item_id=album_id,
169        provider=provider.domain,
170        name=cast("str", normalize_unicode(name)),
171        version=version,
172        provider_mappings={
173            ProviderMapping(
174                item_id=album_id,
175                provider_domain=provider.domain,
176                provider_instance=provider.instance_id,
177                url=attributes.get("url"),
178                available=_is_available(attributes),
179            )
180        },
181    )
182    album_artists = _parse_album_artists(provider, attributes, relationships)
183    if album_artists:
184        album.artists = album_artists
185    if release_date := attributes.get("releaseDate"):
186        album.year = int(release_date.split("-")[0])
187    if genres := attributes.get("genreNames"):
188        album.metadata.genres = set(genres)
189    if image := parse_artwork_image(provider, MediaType.ALBUM, album_id, attributes):
190        album.metadata.add_image(image)
191    if album_copyright := attributes.get("copyright"):
192        album.metadata.copyright = album_copyright
193    if record_label := attributes.get("recordLabel"):
194        album.metadata.label = record_label
195    if upc := attributes.get("upc"):
196        album.external_ids.add((ExternalID.BARCODE, upc))
197    if notes := attributes.get("editorialNotes"):
198        album.metadata.description = notes.get("standard") or notes.get("short")
199    if content_rating := attributes.get("contentRating"):
200        album.metadata.explicit = content_rating == "explicit"
201    album_type = AlbumType.ALBUM
202    if attributes.get("isSingle"):
203        album_type = AlbumType.SINGLE
204    elif attributes.get("isCompilation"):
205        album_type = AlbumType.COMPILATION
206    album.album_type = album_type
207    # Try inference — override if it finds something more specific
208    inferred_type = infer_album_type(album.name, "")
209    if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
210        album.album_type = inferred_type
211    album.favorite = is_favourite or False
212    return album
213
214
215def parse_track(
216    provider: AppleMusicProvider,
217    track_obj: dict[str, Any],
218    is_favourite: bool | None = None,
219) -> Track:
220    """Parse track object to generic layout."""
221    relationships = track_obj.get("relationships", {})
222    raw_attributes = track_obj.get("attributes", {})
223    if (
224        track_obj.get("type") == "library-songs"
225        and relationships.get("catalog", {}).get("data", []) != []
226    ):
227        track_id = relationships.get("catalog", {})["data"][0]["id"]
228        attributes = relationships.get("catalog", {})["data"][0]["attributes"]
229    elif "attributes" in track_obj:
230        track_id = track_obj["id"]
231        attributes = track_obj["attributes"]
232    else:
233        track_id = track_obj["id"]
234        attributes = {}
235    name, version = parse_title_and_version(attributes.get("name", ""))
236    track = Track(
237        item_id=track_id,
238        provider=provider.domain,
239        name=cast("str", normalize_unicode(name)),
240        version=version,
241        duration=int(attributes.get("durationInMillis", 0) / 1000),
242        provider_mappings={
243            ProviderMapping(
244                item_id=track_id,
245                provider_domain=provider.domain,
246                provider_instance=provider.instance_id,
247                audio_format=AudioFormat(content_type=ContentType.AAC),
248                url=attributes.get("url"),
249                available=_is_available(attributes),
250            )
251        },
252    )
253    if disc_number := attributes.get("discNumber"):
254        track.disc_number = disc_number
255    if track_number := attributes.get("trackNumber"):
256        track.track_number = track_number
257    # Prefer catalog information over library information for artists.
258    # The artists relationship is empty when the artists are not in the user's library.
259    if artists_data := relationships.get("artists", {}).get("data"):
260        track.artists = UniqueList([parse_artist(provider, artist) for artist in artists_data])
261    elif artist_name := normalize_unicode(
262        attributes.get("artistName") or raw_attributes.get("artistName")
263    ):
264        track.artists = UniqueList(
265            [
266                ItemMapping(
267                    media_type=MediaType.ARTIST,
268                    item_id=artist_name,
269                    provider=provider.instance_id,
270                    name=artist_name,
271                )
272            ]
273        )
274    if albums := relationships.get("albums"):
275        if "data" in albums and len(albums["data"]) > 0:
276            parsed_album = parse_album(provider, albums["data"][0])
277            if parsed_album:
278                track.album = parsed_album
279    elif album_name := normalize_unicode(
280        attributes.get("albumName") or raw_attributes.get("albumName")
281    ):
282        track.album = ItemMapping(
283            media_type=MediaType.ALBUM,
284            item_id=album_name,
285            provider=provider.instance_id,
286            name=album_name,
287        )
288    if image := parse_artwork_image(provider, MediaType.TRACK, track_id, attributes):
289        track.metadata.add_image(image)
290    if genres := attributes.get("genreNames"):
291        track.metadata.genres = set(genres)
292    if composers := attributes.get("composerName"):
293        track.metadata.performers = set(composers.split(", "))
294    if content_rating := attributes.get("contentRating"):
295        track.metadata.explicit = content_rating == "explicit"
296    if isrc := attributes.get("isrc"):
297        track.external_ids.add((ExternalID.ISRC, isrc))
298    track.favorite = is_favourite or False
299    return track
300
301
302def parse_playlist(
303    provider: AppleMusicProvider,
304    playlist_obj: dict[str, Any],
305    is_favourite: bool | None = None,
306    can_edit_hint: bool | None = None,
307    library_id_override: str | None = None,
308) -> Playlist:
309    """Parse Apple Music playlist object to generic layout."""
310    attributes = playlist_obj["attributes"]
311    raw_playlist_id = playlist_obj["id"]
312    play_params = attributes.get("playParams", {})
313    # Prefer write-safe library IDs when available.
314    playlist_id = (
315        library_id_override
316        or (raw_playlist_id if is_library_id(raw_playlist_id) else play_params.get("globalId"))
317        or raw_playlist_id
318    )
319    is_editable = can_edit_hint if can_edit_hint is not None else attributes.get("canEdit", False)
320    playlist = Playlist(
321        item_id=playlist_id,
322        provider=provider.instance_id,
323        name=attributes.get("name", UNKNOWN_PLAYLIST_NAME),
324        owner=attributes.get("curatorName", "me"),
325        provider_mappings={
326            ProviderMapping(
327                item_id=playlist_id,
328                provider_domain=provider.domain,
329                provider_instance=provider.instance_id,
330                url=attributes.get("url"),
331                is_unique=is_editable,
332            )
333        },
334        is_editable=is_editable,
335    )
336    if image := parse_artwork_image(provider, MediaType.PLAYLIST, playlist_id, attributes):
337        playlist.metadata.add_image(image)
338    if description := attributes.get("description"):
339        playlist.metadata.description = description.get("standard")
340    playlist.favorite = is_favourite or False
341    return playlist
342
343
344def parse_station_as_playlist(
345    provider: AppleMusicProvider,
346    station_obj: dict[str, Any],
347) -> Playlist:
348    """Parse a station object into a dynamic Playlist."""
349    station_id = station_obj["id"]
350    attributes = station_obj.get("attributes", {})
351    name = attributes.get("name", station_id)
352    playlist = Playlist(
353        item_id=station_id,
354        provider=provider.instance_id,
355        name=name,
356        is_dynamic=True,
357        provider_mappings={
358            ProviderMapping(
359                item_id=station_id,
360                provider_domain=provider.domain,
361                provider_instance=provider.instance_id,
362            )
363        },
364    )
365    if image := parse_artwork_image(provider, MediaType.PLAYLIST, station_id, attributes):
366        playlist.metadata.add_image(image)
367    return playlist
368
369
370def _parse_album_artists(
371    provider: AppleMusicProvider,
372    attributes: dict[str, Any],
373    relationships: dict[str, Any],
374) -> UniqueList[Artist | ItemMapping] | None:
375    """Parse the album artists from an album's attributes and relationships."""
376    album_artist_name = normalize_unicode(attributes.get("artistName"))
377    # Skip relationships that cannot produce a named artist.
378    artist_objs = [
379        artist
380        for artist in relationships.get("artists", {}).get("data", [])
381        if _has_artist_details(artist)
382    ]
383    artists = UniqueList([parse_artist(provider, artist) for artist in artist_objs])
384    if album_artist_name and attributes.get("isCompilation"):
385        # A lone related artist can be a contributor rather than the album artist.
386        if len(artists) == 1 and artists[0].name != album_artist_name:
387            artists = UniqueList()
388    if artists:
389        return artists
390    if album_artist_name:
391        return UniqueList(
392            [
393                ItemMapping(
394                    media_type=MediaType.ARTIST,
395                    provider=provider.instance_id,
396                    item_id=album_artist_name,
397                    name=album_artist_name,
398                )
399            ]
400        )
401    return None
402
403
404def _has_artist_details(artist_obj: dict[str, Any]) -> bool:
405    """Check if an artist object holds enough details to parse it."""
406    relationships = artist_obj.get("relationships", {})
407    catalog_data = relationships.get("catalog", {}).get("data", [])
408    if artist_obj.get("type") == "library-artists" and catalog_data:
409        attributes = catalog_data[0].get("attributes", {})
410    else:
411        attributes = artist_obj.get("attributes", {})
412    return bool(normalize_unicode(attributes.get("name")))
413
414
415def _is_available(attributes: dict[str, Any]) -> bool:
416    """
417    Return whether Apple will actually serve a stream for this item.
418
419    ``playParams`` is absent entirely for items Apple has withdrawn. It is present
420    but carries a ``purchasedId`` with no ``catalogId`` for purchase-only items
421    (iTunes purchases, and the 2014 U2 giveaway), which the stream endpoint also
422    refuses. Uploads carry neither marker, so they stay available - see
423    music-assistant/support#6032 and #4108.
424    """
425    play_params = attributes.get("playParams") or {}
426    if play_params.get("id") is None:
427        return False
428    return not (play_params.get("purchasedId") is not None and play_params.get("catalogId") is None)
429