/
/
/
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 # Check availability: library albums owned by user OR catalog items with playParams
168 is_library_album = is_library_id(album_id) and album_obj.get("type") == "library-albums"
169 has_play_params = attributes.get("playParams", {}).get("id") is not None
170 album = Album(
171 item_id=album_id,
172 provider=provider.domain,
173 name=cast("str", normalize_unicode(name)),
174 version=version,
175 provider_mappings={
176 ProviderMapping(
177 item_id=album_id,
178 provider_domain=provider.domain,
179 provider_instance=provider.instance_id,
180 url=attributes.get("url"),
181 available=is_library_album or has_play_params,
182 )
183 },
184 )
185 album_artists = _parse_album_artists(provider, attributes, relationships)
186 if album_artists:
187 album.artists = album_artists
188 if release_date := attributes.get("releaseDate"):
189 album.year = int(release_date.split("-")[0])
190 if genres := attributes.get("genreNames"):
191 album.metadata.genres = set(genres)
192 if image := parse_artwork_image(provider, MediaType.ALBUM, album_id, attributes):
193 album.metadata.add_image(image)
194 if album_copyright := attributes.get("copyright"):
195 album.metadata.copyright = album_copyright
196 if record_label := attributes.get("recordLabel"):
197 album.metadata.label = record_label
198 if upc := attributes.get("upc"):
199 album.external_ids.add((ExternalID.BARCODE, upc))
200 if notes := attributes.get("editorialNotes"):
201 album.metadata.description = notes.get("standard") or notes.get("short")
202 if content_rating := attributes.get("contentRating"):
203 album.metadata.explicit = content_rating == "explicit"
204 album_type = AlbumType.ALBUM
205 if attributes.get("isSingle"):
206 album_type = AlbumType.SINGLE
207 elif attributes.get("isCompilation"):
208 album_type = AlbumType.COMPILATION
209 album.album_type = album_type
210 # Try inference â override if it finds something more specific
211 inferred_type = infer_album_type(album.name, "")
212 if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
213 album.album_type = inferred_type
214 album.favorite = is_favourite or False
215 return album
216
217
218def parse_track(
219 provider: AppleMusicProvider,
220 track_obj: dict[str, Any],
221 is_favourite: bool | None = None,
222) -> Track:
223 """Parse track object to generic layout."""
224 relationships = track_obj.get("relationships", {})
225 raw_attributes = track_obj.get("attributes", {})
226 if (
227 track_obj.get("type") == "library-songs"
228 and relationships.get("catalog", {}).get("data", []) != []
229 ):
230 track_id = relationships.get("catalog", {})["data"][0]["id"]
231 attributes = relationships.get("catalog", {})["data"][0]["attributes"]
232 elif "attributes" in track_obj:
233 track_id = track_obj["id"]
234 attributes = track_obj["attributes"]
235 else:
236 track_id = track_obj["id"]
237 attributes = {}
238 name, version = parse_title_and_version(attributes.get("name", ""))
239 # Check availability: library tracks owned by user OR catalog items with playParams
240 is_library_track = is_library_id(track_id) and track_obj.get("type") == "library-songs"
241 has_play_params = attributes.get("playParams", {}).get("id") is not None
242 track = Track(
243 item_id=track_id,
244 provider=provider.domain,
245 name=cast("str", normalize_unicode(name)),
246 version=version,
247 duration=int(attributes.get("durationInMillis", 0) / 1000),
248 provider_mappings={
249 ProviderMapping(
250 item_id=track_id,
251 provider_domain=provider.domain,
252 provider_instance=provider.instance_id,
253 audio_format=AudioFormat(content_type=ContentType.AAC),
254 url=attributes.get("url"),
255 available=is_library_track or has_play_params,
256 )
257 },
258 )
259 if disc_number := attributes.get("discNumber"):
260 track.disc_number = disc_number
261 if track_number := attributes.get("trackNumber"):
262 track.track_number = track_number
263 # Prefer catalog information over library information for artists.
264 # The artists relationship is empty when the artists are not in the user's library.
265 if artists_data := relationships.get("artists", {}).get("data"):
266 track.artists = UniqueList([parse_artist(provider, artist) for artist in artists_data])
267 elif artist_name := normalize_unicode(
268 attributes.get("artistName") or raw_attributes.get("artistName")
269 ):
270 track.artists = UniqueList(
271 [
272 ItemMapping(
273 media_type=MediaType.ARTIST,
274 item_id=artist_name,
275 provider=provider.instance_id,
276 name=artist_name,
277 )
278 ]
279 )
280 if albums := relationships.get("albums"):
281 if "data" in albums and len(albums["data"]) > 0:
282 parsed_album = parse_album(provider, albums["data"][0])
283 if parsed_album:
284 track.album = parsed_album
285 elif album_name := normalize_unicode(
286 attributes.get("albumName") or raw_attributes.get("albumName")
287 ):
288 track.album = ItemMapping(
289 media_type=MediaType.ALBUM,
290 item_id=album_name,
291 provider=provider.instance_id,
292 name=album_name,
293 )
294 if image := parse_artwork_image(provider, MediaType.TRACK, track_id, attributes):
295 track.metadata.add_image(image)
296 if genres := attributes.get("genreNames"):
297 track.metadata.genres = set(genres)
298 if composers := attributes.get("composerName"):
299 track.metadata.performers = set(composers.split(", "))
300 if content_rating := attributes.get("contentRating"):
301 track.metadata.explicit = content_rating == "explicit"
302 if isrc := attributes.get("isrc"):
303 track.external_ids.add((ExternalID.ISRC, isrc))
304 track.favorite = is_favourite or False
305 return track
306
307
308def parse_playlist(
309 provider: AppleMusicProvider,
310 playlist_obj: dict[str, Any],
311 is_favourite: bool | None = None,
312 can_edit_hint: bool | None = None,
313 library_id_override: str | None = None,
314) -> Playlist:
315 """Parse Apple Music playlist object to generic layout."""
316 attributes = playlist_obj["attributes"]
317 raw_playlist_id = playlist_obj["id"]
318 play_params = attributes.get("playParams", {})
319 # Prefer write-safe library IDs when available.
320 playlist_id = (
321 library_id_override
322 or (raw_playlist_id if is_library_id(raw_playlist_id) else play_params.get("globalId"))
323 or raw_playlist_id
324 )
325 is_editable = can_edit_hint if can_edit_hint is not None else attributes.get("canEdit", False)
326 playlist = Playlist(
327 item_id=playlist_id,
328 provider=provider.instance_id,
329 name=attributes.get("name", UNKNOWN_PLAYLIST_NAME),
330 owner=attributes.get("curatorName", "me"),
331 provider_mappings={
332 ProviderMapping(
333 item_id=playlist_id,
334 provider_domain=provider.domain,
335 provider_instance=provider.instance_id,
336 url=attributes.get("url"),
337 is_unique=is_editable,
338 )
339 },
340 is_editable=is_editable,
341 )
342 if image := parse_artwork_image(provider, MediaType.PLAYLIST, playlist_id, attributes):
343 playlist.metadata.add_image(image)
344 if description := attributes.get("description"):
345 playlist.metadata.description = description.get("standard")
346 playlist.favorite = is_favourite or False
347 return playlist
348
349
350def parse_station_as_playlist(
351 provider: AppleMusicProvider,
352 station_obj: dict[str, Any],
353) -> Playlist:
354 """Parse a station object into a dynamic Playlist."""
355 station_id = station_obj["id"]
356 attributes = station_obj.get("attributes", {})
357 name = attributes.get("name", station_id)
358 playlist = Playlist(
359 item_id=station_id,
360 provider=provider.instance_id,
361 name=name,
362 is_dynamic=True,
363 provider_mappings={
364 ProviderMapping(
365 item_id=station_id,
366 provider_domain=provider.domain,
367 provider_instance=provider.instance_id,
368 )
369 },
370 )
371 if image := parse_artwork_image(provider, MediaType.PLAYLIST, station_id, attributes):
372 playlist.metadata.add_image(image)
373 return playlist
374
375
376def _parse_album_artists(
377 provider: AppleMusicProvider,
378 attributes: dict[str, Any],
379 relationships: dict[str, Any],
380) -> UniqueList[Artist | ItemMapping] | None:
381 """Parse the album artists from an album's attributes and relationships."""
382 album_artist_name = normalize_unicode(attributes.get("artistName"))
383 # Skip relationships that cannot produce a named artist.
384 artist_objs = [
385 artist
386 for artist in relationships.get("artists", {}).get("data", [])
387 if _has_artist_details(artist)
388 ]
389 artists = UniqueList([parse_artist(provider, artist) for artist in artist_objs])
390 if album_artist_name and attributes.get("isCompilation"):
391 # A lone related artist can be a contributor rather than the album artist.
392 if len(artists) == 1 and artists[0].name != album_artist_name:
393 artists = UniqueList()
394 if artists:
395 return artists
396 if album_artist_name:
397 return UniqueList(
398 [
399 ItemMapping(
400 media_type=MediaType.ARTIST,
401 provider=provider.instance_id,
402 item_id=album_artist_name,
403 name=album_artist_name,
404 )
405 ]
406 )
407 return None
408
409
410def _has_artist_details(artist_obj: dict[str, Any]) -> bool:
411 """Check if an artist object holds enough details to parse it."""
412 relationships = artist_obj.get("relationships", {})
413 catalog_data = relationships.get("catalog", {}).get("data", [])
414 if artist_obj.get("type") == "library-artists" and catalog_data:
415 attributes = catalog_data[0].get("attributes", {})
416 else:
417 attributes = artist_obj.get("attributes", {})
418 return bool(normalize_unicode(attributes.get("name")))
419