music-assistant-server
14.9 KB•PY
parsers.py
14.9 KB • 412 lines • python
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 # The show object embedded in an episode carries the show's description,
277 # artwork and publisher, so parse it in full rather than as a name-only stub.
278 if podcast is None and "show" in episode_obj:
279 podcast = parse_podcast(episode_obj["show"], provider)
280 elif podcast is None:
281 # Create a minimal podcast reference if none available
282 podcast = Podcast(
283 item_id="unknown",
284 provider=provider.instance_id,
285 name="Unknown Podcast",
286 translation_key="unknown_podcast",
287 provider_mappings=set(),
288 )
289
290 episode = PodcastEpisode(
291 item_id=episode_obj["id"],
292 provider=provider.instance_id,
293 name=episode_obj["name"],
294 duration=episode_obj["duration_ms"] // 1000 if episode_obj.get("duration_ms") else 0,
295 podcast=podcast,
296 position=0,
297 provider_mappings={
298 ProviderMapping(
299 item_id=episode_obj["id"],
300 provider_domain=provider.domain,
301 provider_instance=provider.instance_id,
302 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=160),
303 url=episode_obj["external_urls"]["spotify"],
304 )
305 },
306 )
307
308 # Set description in metadata
309 if episode_obj.get("description"):
310 episode.metadata.description = episode_obj["description"]
311
312 # Add release date to metadata
313 if episode_obj.get("release_date"):
314 with contextlib.suppress(ValueError, TypeError):
315 date_str = episode_obj["release_date"].strip()
316
317 if len(date_str) == 4:
318 # Year only: "2023" -> "2023-01-01T00:00:00+00:00"
319 date_str = f"{date_str}-01-01T00:00:00+00:00"
320 elif len(date_str) == 10:
321 # Date only: "2023-12-25" -> "2023-12-25T00:00:00+00:00"
322 date_str = f"{date_str}T00:00:00+00:00"
323
324 episode.metadata.release_date = datetime.fromisoformat(date_str)
325
326 episode.metadata.images = parse_images(episode_obj.get("images", []), provider.instance_id)
327
328 # Use podcast artwork if episode has none
329 if not episode.metadata.images and isinstance(podcast, Podcast) and podcast.metadata.images:
330 episode.metadata.images = podcast.metadata.images
331
332 if "explicit" in episode_obj:
333 episode.metadata.explicit = episode_obj["explicit"]
334
335 if "audio_preview_url" in episode_obj:
336 episode.metadata.preview = episode_obj["audio_preview_url"]
337
338 return episode
339
340
341def parse_audiobook(audiobook_obj: dict[str, Any], provider: SpotifyProvider) -> Audiobook:
342 """Parse spotify audiobook object to generic layout."""
343 audiobook = Audiobook(
344 item_id=audiobook_obj["id"],
345 provider=provider.instance_id,
346 name=audiobook_obj["name"],
347 provider_mappings={
348 ProviderMapping(
349 item_id=audiobook_obj["id"],
350 provider_domain=provider.domain,
351 provider_instance=provider.instance_id,
352 audio_format=AudioFormat(content_type=ContentType.OGG, bit_rate=320),
353 url=audiobook_obj["external_urls"]["spotify"],
354 )
355 },
356 )
357
358 if "duration_ms" in audiobook_obj:
359 provider.logger.debug(
360 f"Found duration_ms in audiobook object: {audiobook_obj['duration_ms']}"
361 )
362 audiobook.duration = audiobook_obj["duration_ms"] // 1000
363 else:
364 provider.logger.debug(
365 "No duration_ms found in main audiobook object - will calculate from chapters"
366 )
367 # Don't set duration here - let get_audiobook calculate it from chapters
368 audiobook.duration = 0
369
370 # Set authors
371 if "authors" in audiobook_obj:
372 for author_obj in audiobook_obj["authors"]:
373 if author_obj.get("name"):
374 audiobook.authors.append(author_obj["name"])
375
376 # Set narrators
377 if "narrators" in audiobook_obj:
378 for narrator_obj in audiobook_obj["narrators"]:
379 if narrator_obj.get("name"):
380 audiobook.narrators.append(narrator_obj["name"])
381
382 # Set metadata
383 if audiobook_obj.get("description"):
384 audiobook.metadata.description = audiobook_obj["description"]
385
386 if audiobook_obj.get("publisher"):
387 audiobook.publisher = audiobook_obj["publisher"]
388
389 audiobook.metadata.images = parse_images(audiobook_obj.get("images", []), provider.instance_id)
390
391 if audiobook_obj.get("explicit"):
392 audiobook.metadata.explicit = audiobook_obj["explicit"]
393
394 if audiobook_obj.get("languages"):
395 audiobook.metadata.languages = UniqueList(audiobook_obj["languages"])
396
397 audiobook.metadata.genres = {"Spoken Word"}
398
399 # Set publication date if available
400 if audiobook_obj.get("publication_date"):
401 with contextlib.suppress(ValueError, TypeError):
402 date_str = audiobook_obj["publication_date"].strip()
403 if len(date_str) == 4:
404 # Year only: "2023" -> "2023-01-01T00:00:00+00:00"
405 date_str = f"{date_str}-01-01T00:00:00+00:00"
406 elif len(date_str) == 10:
407 # Date only: "2023-12-25" -> "2023-12-25T00:00:00+00:00"
408 date_str = f"{date_str}T00:00:00+00:00"
409 audiobook.metadata.release_date = datetime.fromisoformat(date_str)
410
411 return audiobook
412