/
/
1"""Parsing utilities to convert Spotify API responses into Music Assistant model objects."""
2
3from __future__ import annotations
4
5import contextlib
6from datetime import datetime
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import AlbumType, ContentType, ExternalID, ImageType
10from music_assistant_models.media_items import (
11 Album,
12 Artist,
13 Audiobook,
14 AudioFormat,
15 MediaItemImage,
16 Playlist,
17 Podcast,
18 PodcastEpisode,
19 ProviderMapping,
20 Track,
21)
22from music_assistant_models.unique_list import UniqueList
23
24from music_assistant.helpers.util import infer_album_type, parse_title_and_version
25
26if TYPE_CHECKING:
27 from .provider import SpotifyProvider
28
29
30def parse_images(
31 images_list: list[dict[str, Any]], instance_id: str, exclude_generic: bool = False
32) -> UniqueList[MediaItemImage]:
33 """Parse images list into MediaItemImage objects."""
34 if not images_list:
35 return UniqueList([])
36
37 # Filter out generic images if requested (for artists)
38 filtered_images = []
39 for img in images_list:
40 img_url = img["url"]
41 if exclude_generic and "2a96cbd8b46e442fc41c2b86b821562f" in img_url:
42 continue
43 filtered_images.append(img)
44
45 if not filtered_images:
46 return UniqueList([])
47
48 # Spotify images come in various sizes (typically 640x640, 300x300, 64x64)
49 # Find the largest image available
50 best_image = max(
51 filtered_images, key=lambda img: img.get("height", 0), default=filtered_images[0]
52 )
53
54 return UniqueList(
55 [
56 MediaItemImage(
57 type=ImageType.THUMB,
58 path=best_image["url"],
59 provider=instance_id,
60 remotely_accessible=True,
61 )
62 ]
63 )
64
65
66def parse_artist(artist_obj: dict[str, Any], provider: SpotifyProvider) -> Artist:
67 """Parse spotify artist object to generic layout."""
68 artist = Artist(
69 item_id=artist_obj["id"],
70 provider=provider.instance_id,
71 name=artist_obj["name"] or artist_obj["id"],
72 provider_mappings={
73 ProviderMapping(
74 item_id=artist_obj["id"],
75 provider_domain=provider.domain,
76 provider_instance=provider.instance_id,
77 url=artist_obj["external_urls"]["spotify"],
78 )
79 },
80 )
81 if "genres" in artist_obj:
82 artist.metadata.genres = set(artist_obj["genres"])
83
84 # Use unified image parsing with generic exclusion
85 artist.metadata.images = parse_images(
86 artist_obj.get("images", []), provider.instance_id, exclude_generic=True
87 )
88 return artist
89
90
91def parse_album(album_obj: dict[str, Any], provider: SpotifyProvider) -> Album:
92 """Parse spotify album object to generic layout."""
93 name, version = parse_title_and_version(album_obj["name"])
94 album = Album(
95 item_id=album_obj["id"],
96 provider=provider.instance_id,
97 name=name,
98 version=version,
99 provider_mappings={
100 ProviderMapping(
101 item_id=album_obj["id"],
102 provider_domain=provider.domain,
103 provider_instance=provider.instance_id,
104 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=320),
105 url=album_obj["external_urls"]["spotify"],
106 )
107 },
108 )
109 if "external_ids" in album_obj and album_obj["external_ids"].get("upc"):
110 album.external_ids.add((ExternalID.BARCODE, album_obj["external_ids"]["upc"]))
111 if "external_ids" in album_obj and album_obj["external_ids"].get("ean"):
112 album.external_ids.add((ExternalID.BARCODE, album_obj["external_ids"]["ean"]))
113
114 for artist_obj in album_obj["artists"]:
115 if not artist_obj.get("name") or not artist_obj.get("id"):
116 continue
117 album.artists.append(parse_artist(artist_obj, provider))
118
119 with contextlib.suppress(ValueError):
120 album.album_type = AlbumType(album_obj["album_type"])
121
122 # Override with inferred type if version indicates it
123 inferred_type = infer_album_type(album.name, album.version)
124 if inferred_type in (AlbumType.LIVE, AlbumType.SOUNDTRACK):
125 album.album_type = inferred_type
126
127 if "genres" in album_obj:
128 album.metadata.genres = set(album_obj["genres"])
129
130 album.metadata.images = parse_images(album_obj.get("images", []), provider.instance_id)
131
132 if "label" in album_obj:
133 album.metadata.label = album_obj["label"]
134 if album_obj.get("release_date"):
135 album.year = int(album_obj["release_date"].split("-")[0])
136 if album_obj.get("copyrights"):
137 album.metadata.copyright = album_obj["copyrights"][0]["text"]
138 if album_obj.get("explicit"):
139 album.metadata.explicit = album_obj["explicit"]
140 return album
141
142
143def parse_track(
144 track_obj: dict[str, Any],
145 provider: SpotifyProvider,
146 artist: Artist | None = None,
147) -> Track:
148 """Parse spotify track object to generic layout."""
149 name, version = parse_title_and_version(track_obj["name"])
150 track = Track(
151 item_id=track_obj["id"],
152 provider=provider.instance_id,
153 name=name,
154 version=version,
155 duration=int(track_obj["duration_ms"] / 1000),
156 provider_mappings={
157 ProviderMapping(
158 item_id=track_obj["id"],
159 provider_domain=provider.domain,
160 provider_instance=provider.instance_id,
161 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=320),
162 url=track_obj["external_urls"]["spotify"],
163 available=not track_obj["is_local"] and track_obj["is_playable"],
164 )
165 },
166 disc_number=track_obj.get("disc_number", 0),
167 track_number=track_obj.get("track_number", 0),
168 )
169 if isrc := track_obj.get("external_ids", {}).get("isrc"):
170 track.external_ids.add((ExternalID.ISRC, isrc))
171
172 if artist:
173 track.artists.append(artist)
174 for track_artist in track_obj.get("artists", []):
175 if not track_artist.get("name") or not track_artist.get("id"):
176 continue
177 artist_parsed = parse_artist(track_artist, provider)
178 if artist_parsed and artist_parsed.item_id not in {x.item_id for x in track.artists}:
179 track.artists.append(artist_parsed)
180
181 track.metadata.explicit = track_obj["explicit"]
182 if "preview_url" in track_obj:
183 track.metadata.preview = track_obj["preview_url"]
184 if "album" in track_obj:
185 track.album = parse_album(track_obj["album"], provider)
186 track.metadata.images = parse_images(
187 track_obj["album"].get("images", []), provider.instance_id
188 )
189 if track_obj.get("copyright"):
190 track.metadata.copyright = track_obj["copyright"]
191 if track_obj.get("explicit"):
192 track.metadata.explicit = True
193 if track_obj.get("popularity"):
194 track.metadata.popularity = track_obj["popularity"]
195 return track
196
197
198def parse_playlist(playlist_obj: dict[str, Any], provider: SpotifyProvider) -> Playlist:
199 """Parse spotify playlist object to generic layout."""
200 owner_id = playlist_obj["owner"].get("id", "")
201 is_editable = (
202 provider._sp_user is not None and owner_id == provider._sp_user["id"]
203 ) or playlist_obj["collaborative"]
204
205 # Spotify-owned playlists (Daily Mix, Discover Weekly, etc.) are personalized per user
206 is_spotify_owned = owner_id.lower() == "spotify"
207
208 # Get owner name with fallback
209 owner_name = playlist_obj["owner"].get("display_name")
210 if owner_name is None and provider._sp_user is not None:
211 owner_name = provider._sp_user["display_name"]
212
213 # Mark as unique if user-owned/editable OR if it's a Spotify personalized playlist
214 is_unique = is_editable or is_spotify_owned
215
216 playlist = Playlist(
217 item_id=playlist_obj["id"],
218 provider=provider.instance_id,
219 name=playlist_obj["name"],
220 owner=owner_name,
221 provider_mappings={
222 ProviderMapping(
223 item_id=playlist_obj["id"],
224 provider_domain=provider.domain,
225 provider_instance=provider.instance_id,
226 url=playlist_obj["external_urls"]["spotify"],
227 is_unique=is_unique,
228 )
229 },
230 is_editable=is_editable,
231 )
232
233 playlist.metadata.images = parse_images(playlist_obj.get("images", []), provider.instance_id)
234 return playlist
235
236
237def parse_podcast(podcast_obj: dict[str, Any], provider: SpotifyProvider) -> Podcast:
238 """Parse spotify podcast (show) object to generic layout."""
239 podcast = Podcast(
240 item_id=podcast_obj["id"],
241 provider=provider.instance_id,
242 name=podcast_obj["name"],
243 provider_mappings={
244 ProviderMapping(
245 item_id=podcast_obj["id"],
246 provider_domain=provider.domain,
247 provider_instance=provider.instance_id,
248 url=podcast_obj["external_urls"]["spotify"],
249 )
250 },
251 publisher=podcast_obj.get("publisher"),
252 total_episodes=podcast_obj.get("total_episodes"),
253 )
254
255 # Set metadata
256 if podcast_obj.get("description"):
257 podcast.metadata.description = podcast_obj["description"]
258
259 podcast.metadata.images = parse_images(podcast_obj.get("images", []), provider.instance_id)
260
261 if "explicit" in podcast_obj:
262 podcast.metadata.explicit = podcast_obj["explicit"]
263
264 if podcast_obj.get("languages"):
265 podcast.metadata.languages = UniqueList(podcast_obj["languages"])
266
267 podcast.metadata.genres = {"Spoken Word"}
268
269 return podcast
270
271
272def parse_podcast_episode(
273 episode_obj: dict[str, Any], provider: SpotifyProvider, podcast: Podcast | None = None
274) -> PodcastEpisode:
275 """Parse spotify podcast episode object to generic layout."""
276 # Get or create a basic podcast reference if not provided
277 if podcast is None and "show" in episode_obj:
278 podcast = Podcast(
279 item_id=episode_obj["show"]["id"],
280 provider=provider.instance_id,
281 name=episode_obj["show"]["name"],
282 provider_mappings={
283 ProviderMapping(
284 item_id=episode_obj["show"]["id"],
285 provider_domain=provider.domain,
286 provider_instance=provider.instance_id,
287 url=episode_obj["show"]["external_urls"]["spotify"],
288 )
289 },
290 )
291 elif podcast is None:
292 # Create a minimal podcast reference if none available
293 podcast = Podcast(
294 item_id="unknown",
295 provider=provider.instance_id,
296 name="Unknown Podcast",
297 translation_key="unknown_podcast",
298 provider_mappings=set(),
299 )
300
301 episode = PodcastEpisode(
302 item_id=episode_obj["id"],
303 provider=provider.instance_id,
304 name=episode_obj["name"],
305 duration=episode_obj["duration_ms"] // 1000 if episode_obj.get("duration_ms") else 0,
306 podcast=podcast,
307 position=0,
308 provider_mappings={
309 ProviderMapping(
310 item_id=episode_obj["id"],
311 provider_domain=provider.domain,
312 provider_instance=provider.instance_id,
313 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=160),
314 url=episode_obj["external_urls"]["spotify"],
315 )
316 },
317 )
318
319 # Set description in metadata
320 if episode_obj.get("description"):
321 episode.metadata.description = episode_obj["description"]
322
323 # Add release date to metadata
324 if episode_obj.get("release_date"):
325 with contextlib.suppress(ValueError, TypeError):
326 date_str = episode_obj["release_date"].strip()
327
328 if len(date_str) == 4:
329 # Year only: "2023" -> "2023-01-01T00:00:00+00:00"
330 date_str = f"{date_str}-01-01T00:00:00+00:00"
331 elif len(date_str) == 10:
332 # Date only: "2023-12-25" -> "2023-12-25T00:00:00+00:00"
333 date_str = f"{date_str}T00:00:00+00:00"
334
335 episode.metadata.release_date = datetime.fromisoformat(date_str)
336
337 episode.metadata.images = parse_images(episode_obj.get("images", []), provider.instance_id)
338
339 # Use podcast artwork if episode has none
340 if not episode.metadata.images and isinstance(podcast, Podcast) and podcast.metadata.images:
341 episode.metadata.images = podcast.metadata.images
342
343 if "explicit" in episode_obj:
344 episode.metadata.explicit = episode_obj["explicit"]
345
346 if "audio_preview_url" in episode_obj:
347 episode.metadata.preview = episode_obj["audio_preview_url"]
348
349 return episode
350
351
352def parse_audiobook(audiobook_obj: dict[str, Any], provider: SpotifyProvider) -> Audiobook:
353 """Parse spotify audiobook object to generic layout."""
354 audiobook = Audiobook(
355 item_id=audiobook_obj["id"],
356 provider=provider.instance_id,
357 name=audiobook_obj["name"],
358 provider_mappings={
359 ProviderMapping(
360 item_id=audiobook_obj["id"],
361 provider_domain=provider.domain,
362 provider_instance=provider.instance_id,
363 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=320),
364 url=audiobook_obj["external_urls"]["spotify"],
365 )
366 },
367 )
368
369 if "duration_ms" in audiobook_obj:
370 provider.logger.debug(
371 f"Found duration_ms in audiobook object: {audiobook_obj['duration_ms']}"
372 )
373 audiobook.duration = audiobook_obj["duration_ms"] // 1000
374 else:
375 provider.logger.debug(
376 "No duration_ms found in main audiobook object - will calculate from chapters"
377 )
378 # Don't set duration here - let get_audiobook calculate it from chapters
379 audiobook.duration = 0
380
381 # Set authors
382 if "authors" in audiobook_obj:
383 for author_obj in audiobook_obj["authors"]:
384 if author_obj.get("name"):
385 audiobook.authors.append(author_obj["name"])
386
387 # Set narrators
388 if "narrators" in audiobook_obj:
389 for narrator_obj in audiobook_obj["narrators"]:
390 if narrator_obj.get("name"):
391 audiobook.narrators.append(narrator_obj["name"])
392
393 # Set metadata
394 if audiobook_obj.get("description"):
395 audiobook.metadata.description = audiobook_obj["description"]
396
397 if audiobook_obj.get("publisher"):
398 audiobook.publisher = audiobook_obj["publisher"]
399
400 audiobook.metadata.images = parse_images(audiobook_obj.get("images", []), provider.instance_id)
401
402 if audiobook_obj.get("explicit"):
403 audiobook.metadata.explicit = audiobook_obj["explicit"]
404
405 if audiobook_obj.get("languages"):
406 audiobook.metadata.languages = UniqueList(audiobook_obj["languages"])
407
408 audiobook.metadata.genres = {"Spoken Word"}
409
410 # Set publication date if available
411 if audiobook_obj.get("publication_date"):
412 with contextlib.suppress(ValueError, TypeError):
413 date_str = audiobook_obj["publication_date"].strip()
414 if len(date_str) == 4:
415 # Year only: "2023" -> "2023-01-01T00:00:00+00:00"
416 date_str = f"{date_str}-01-01T00:00:00+00:00"
417 elif len(date_str) == 10:
418 # Date only: "2023-12-25" -> "2023-12-25T00:00:00+00:00"
419 date_str = f"{date_str}T00:00:00+00:00"
420 audiobook.metadata.release_date = datetime.fromisoformat(date_str)
421
422 return audiobook
423