/
/
/
1"""
2Parsers for Deezer API response objects (GQL + GW).
3
4Standalone functions that convert Deezer GQL Pydantic models and GW API
5dicts into Music Assistant media item models.
6"""
7
8from __future__ import annotations
9
10import re
11from collections.abc import Sequence
12from datetime import datetime
13from typing import TYPE_CHECKING, Any, Protocol
14
15from deezer_python_gql.generated.enums import AlbumType as DeezerAlbumType
16from deezer_python_gql.generated.enums import AudiobookContributorRoles
17from deezer_python_gql.generated.get_recently_played import (
18 GetRecentlyPlayedMeRecentlyPlayedEdges,
19 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeAlbum,
20 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeArtist,
21 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlow,
22 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlowConfig,
23 GetRecentlyPlayedMeRecentlyPlayedEdgesNodePlaylist,
24 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeSmartTracklist,
25)
26from deezer_python_gql.generated.get_track import GetTrackTrack
27from music_assistant_models.enums import (
28 AlbumType,
29 ExternalID,
30 ImageType,
31 MediaType,
32)
33from music_assistant_models.media_items import (
34 Album,
35 Artist,
36 Audiobook,
37 ItemMapping,
38 MediaItemChapter,
39 MediaItemImage,
40 MediaItemMetadata,
41 MediaItemType,
42 Playlist,
43 Podcast,
44 PodcastEpisode,
45 ProviderMapping,
46 Radio,
47 Track,
48 UniqueList,
49)
50
51from music_assistant.helpers.util import infer_album_type, parse_title_and_version
52
53from .constants import (
54 FLOW_CONFIG_PREFIX,
55 FLOW_PLAYLIST_ID,
56 PERSONAL_ALBUM_PREFIX,
57 PERSONAL_ARTIST_PREFIX,
58 SMART_TRACKLIST_PREFIX,
59)
60from .helpers import (
61 create_virtual_playlist,
62)
63
64if TYPE_CHECKING:
65 from deezer_python_gql.generated.fragments import (
66 AlbumFields,
67 ArtistFields,
68 AudiobookFields,
69 LivestreamFields,
70 PlaylistFields,
71 PodcastEpisodeFields,
72 PodcastFields,
73 TrackFields,
74 )
75 from deezer_python_gql.generated.get_audiobook import (
76 GetAudiobookAudiobookChaptersEdges,
77 )
78 from deezer_python_gql.generated.get_flow_config_tracks import (
79 GetFlowConfigTracksFlowConfig,
80 )
81 from deezer_python_gql.generated.get_flow_configs import (
82 GetFlowConfigsMeFlowConfigsGenresEdgesNode,
83 GetFlowConfigsMeFlowConfigsMoodsEdgesNode,
84 )
85 from deezer_python_gql.generated.search import (
86 SearchSearchResultsAlbumsEdgesNode,
87 SearchSearchResultsArtistsEdgesNode,
88 SearchSearchResultsLivestreamsEdgesNode,
89 SearchSearchResultsPlaylistsEdgesNode,
90 SearchSearchResultsPodcastsEdgesNode,
91 SearchSearchResultsTracksEdgesNode,
92 )
93 from deezer_python_gql.generated.search_flows import (
94 SearchFlowsSearchResultsFlowConfigsEdgesNode,
95 )
96
97 from .provider import DeezerProvider
98
99# Deezer CDN image URL pattern
100DEEZER_CDN_IMAGE = "https://e-cdns-images.dzcdn.net/images"
101
102
103# -- Protocols for typed GQL attributes --
104
105
106class _CoverLike(Protocol):
107 """Protocol for GQL cover/picture objects with a `.urls` list."""
108
109 @property
110 def urls(self) -> list[str]: ...
111
112
113class _HasUrl(Protocol):
114 """Protocol for GQL result objects with a `.url` field."""
115
116 @property
117 def url(self) -> object: ...
118
119
120def _cover_image(provider: DeezerProvider, cover: _CoverLike | None) -> MediaItemImage | None:
121 """
122 Create a THUMB image from a GQL cover/picture object with a `.urls` list.
123
124 Returns None when the cover is absent or has no URLs.
125 """
126 if cover and cover.urls:
127 return MediaItemImage(
128 type=ImageType.THUMB,
129 path=cover.urls[0],
130 provider=provider.instance_id,
131 remotely_accessible=True,
132 )
133 return None
134
135
136def _provider_mapping(
137 provider: DeezerProvider,
138 item_id: str,
139 *,
140 available: bool = True,
141 is_unique: bool | None = None,
142) -> ProviderMapping:
143 """Create a ProviderMapping for the given item."""
144 return ProviderMapping(
145 item_id=item_id,
146 provider_domain=provider.domain,
147 provider_instance=provider.instance_id,
148 available=available,
149 is_unique=is_unique,
150 )
151
152
153# -- GQL model parsers --
154
155
156def _track_available(
157 track: TrackFields | SearchSearchResultsTracksEdgesNode | GetTrackTrack,
158) -> bool:
159 """Determine if a track is available for streaming."""
160 media = track.media
161 if media is None:
162 return False
163 if media.rights.sub is None:
164 return False
165 return bool(media.rights.sub.available)
166
167
168def parse_track(
169 provider: DeezerProvider,
170 track: TrackFields | SearchSearchResultsTracksEdgesNode | GetTrackTrack,
171 position: int = 0,
172) -> Track:
173 """
174 Parse a GQL track model to Music Assistant Track.
175
176 :param provider: The Deezer provider instance.
177 :param track: A GQL track model (fragment-based or slim search result).
178 :param position: Position in a track list (for playlist ordering).
179 """
180 artists: UniqueList[Artist | ItemMapping] = UniqueList()
181 for edge in track.contributors.edges:
182 if edge.node is not None:
183 artists.append(
184 ItemMapping(
185 media_type=MediaType.ARTIST,
186 item_id=edge.node.id,
187 provider=provider.instance_id,
188 name=edge.node.name,
189 )
190 )
191
192 album: Album | None = None
193 if track.album is not None:
194 album_images: UniqueList[MediaItemImage] = UniqueList()
195 if img := _cover_image(provider, track.album.cover):
196 album_images.append(img)
197 # Use track contributors as album artists since the track-level album
198 # sub-query doesn't include its own contributors. This ensures the
199 # album gets stored with artist references when added to the library.
200 album_artists: UniqueList[Artist | ItemMapping] = UniqueList()
201 for edge in track.contributors.edges:
202 if edge.node is not None and edge.roles and "MAIN" in edge.roles:
203 album_artists.append(
204 ItemMapping(
205 media_type=MediaType.ARTIST,
206 item_id=edge.node.id,
207 provider=provider.instance_id,
208 name=edge.node.name,
209 )
210 )
211 year: int | None = None
212 if release_date := getattr(track, "release_date", None):
213 if parsed := parse_date(str(release_date)):
214 year = parsed.year
215 album = Album(
216 item_id=track.album.id,
217 provider=provider.instance_id,
218 name=track.album.display_title,
219 year=year,
220 artists=album_artists,
221 provider_mappings={_provider_mapping(provider, track.album.id)},
222 metadata=MediaItemMetadata(images=album_images)
223 if album_images
224 else MediaItemMetadata(),
225 )
226
227 name, version = parse_title_and_version(track.title)
228 disc_number = 0
229 track_number = position
230 if (disk_info := getattr(track, "disk_info", None)) is not None:
231 disc_number = disk_info.disk_number or 0
232 track_number = disk_info.track_number or position
233
234 item = Track(
235 item_id=track.id,
236 provider=provider.instance_id,
237 name=name,
238 version=version,
239 duration=track.duration,
240 artists=artists,
241 album=album,
242 provider_mappings={
243 _provider_mapping(provider, track.id, available=_track_available(track))
244 },
245 metadata=_parse_track_metadata(provider, track),
246 track_number=track_number,
247 position=position,
248 disc_number=disc_number,
249 )
250 if isrc := getattr(track, "isrc", None):
251 item.external_ids.add((ExternalID.ISRC, isrc))
252 if getattr(track, "is_favorite", False):
253 item.favorite = True
254 if (popularity := getattr(track, "popularity", None)) is not None:
255 item.metadata.popularity = int(popularity)
256 return item
257
258
259def _parse_track_metadata(
260 provider: DeezerProvider,
261 track: TrackFields | SearchSearchResultsTracksEdgesNode | GetTrackTrack,
262) -> MediaItemMetadata:
263 """Parse track metadata (images, explicit flag, lyrics) from a GQL track model."""
264 metadata = MediaItemMetadata(explicit=track.is_explicit)
265 if track.album is not None:
266 if img := _cover_image(provider, track.album.cover):
267 metadata.add_image(img)
268 # Lyrics (only present on GetTrackTrack, not on fragment-based models)
269 if isinstance(track, GetTrackTrack) and track.lyrics is not None:
270 if track.lyrics.text:
271 metadata.lyrics = track.lyrics.text
272 if track.lyrics.synchronized_lines:
273 lrc_lines = [
274 f"{line.lrc_timestamp} {line.line}"
275 for line in track.lyrics.synchronized_lines
276 if line.lrc_timestamp
277 ]
278 if lrc_lines:
279 metadata.lrc_lyrics = "\n".join(lrc_lines)
280 return metadata
281
282
283def parse_artist(
284 provider: DeezerProvider, artist: ArtistFields | SearchSearchResultsArtistsEdgesNode
285) -> Artist:
286 """Parse a GQL artist model to Music Assistant Artist."""
287 images: UniqueList[MediaItemImage] = UniqueList()
288 if img := _cover_image(provider, artist.picture):
289 images.append(img)
290 metadata = MediaItemMetadata(images=images) if images else MediaItemMetadata()
291 if bio := getattr(artist, "bio", None):
292 metadata.description = re.sub(r"<[^>]+>", "", bio.full).strip()
293 if (fans_count := getattr(artist, "fans_count", None)) is not None:
294 metadata.popularity = int(fans_count)
295 item = Artist(
296 item_id=artist.id,
297 provider=provider.instance_id,
298 name=artist.name,
299 media_type=MediaType.ARTIST,
300 provider_mappings={_provider_mapping(provider, artist.id)},
301 metadata=metadata,
302 )
303 if getattr(artist, "is_favorite", False):
304 item.favorite = True
305 return item
306
307
308def parse_album(
309 provider: DeezerProvider, album: AlbumFields | SearchSearchResultsAlbumsEdgesNode
310) -> Album:
311 """Parse a GQL album model to Music Assistant Album."""
312 name, version = parse_title_and_version(album.display_title)
313 artists: UniqueList[Artist | ItemMapping] = UniqueList()
314 for edge in album.contributors.edges:
315 if edge.node is not None:
316 artists.append(
317 ItemMapping(
318 media_type=MediaType.ARTIST,
319 item_id=edge.node.id,
320 provider=provider.instance_id,
321 name=edge.node.name,
322 )
323 )
324
325 images: UniqueList[MediaItemImage] = UniqueList()
326 if img := _cover_image(provider, album.cover):
327 images.append(img)
328
329 deezer_type = album.type_
330 album_type = map_album_type(deezer_type, name)
331
332 year: int | None = None
333 if album.release_date:
334 if parsed := parse_date(str(album.release_date)):
335 year = parsed.year
336
337 item = Album(
338 album_type=album_type,
339 item_id=album.id,
340 provider=provider.instance_id,
341 name=name,
342 version=version,
343 year=year,
344 artists=artists,
345 media_type=MediaType.ALBUM,
346 provider_mappings={_provider_mapping(provider, album.id)},
347 metadata=MediaItemMetadata(
348 explicit=getattr(album, "is_explicit", None),
349 images=images,
350 label=getattr(album, "label", None),
351 copyright=getattr(album, "copyright", None),
352 ),
353 )
354 if (fans_count := getattr(album, "fans_count", None)) is not None:
355 item.metadata.popularity = int(fans_count)
356 if getattr(album, "is_favorite", False):
357 item.favorite = True
358 return item
359
360
361def parse_playlist(
362 provider: DeezerProvider,
363 playlist: PlaylistFields | SearchSearchResultsPlaylistsEdgesNode,
364 is_editable: bool = False,
365) -> Playlist:
366 """
367 Parse a GQL playlist model to Music Assistant Playlist.
368
369 :param provider: The Deezer provider instance.
370 :param playlist: A GQL playlist model (fragment-based or slim search result).
371 :param is_editable: Whether the current user owns this playlist.
372 """
373 images: UniqueList[MediaItemImage] = UniqueList()
374 if img := _cover_image(provider, playlist.picture):
375 images.append(img)
376 owner_name = "Unknown"
377 if playlist.owner:
378 owner_name = playlist.owner.name
379 if not is_editable and playlist.owner.id == provider.user_id:
380 is_editable = True
381
382 metadata = MediaItemMetadata(images=images) if images else MediaItemMetadata()
383 if (fans_count := getattr(playlist, "fans_count", None)) is not None:
384 metadata.popularity = int(fans_count)
385 if description := getattr(playlist, "description", None):
386 metadata.description = description
387 item = Playlist(
388 item_id=playlist.id,
389 provider=provider.instance_id,
390 name=playlist.title,
391 media_type=MediaType.PLAYLIST,
392 provider_mappings={_provider_mapping(provider, playlist.id, is_unique=is_editable)},
393 metadata=metadata,
394 is_editable=is_editable,
395 owner=owner_name,
396 )
397 if getattr(playlist, "is_favorite", False) or is_editable:
398 item.favorite = True
399 return item
400
401
402def parse_radio(
403 provider: DeezerProvider, livestream: LivestreamFields | SearchSearchResultsLivestreamsEdgesNode
404) -> Radio:
405 """
406 Parse a GQL Livestream model to Music Assistant Radio.
407
408 :param provider: The Deezer provider instance.
409 :param livestream: A GQL livestream model (fragment-based or slim search result).
410 """
411 images: UniqueList[MediaItemImage] = UniqueList()
412 if img := _cover_image(provider, livestream.cover):
413 images.append(img)
414 metadata = MediaItemMetadata(images=images) if images else MediaItemMetadata()
415 if description := getattr(livestream, "description", None):
416 metadata.description = description
417 return Radio(
418 item_id=livestream.id,
419 provider=provider.instance_id,
420 name=livestream.name,
421 media_type=MediaType.RADIO,
422 provider_mappings={_provider_mapping(provider, livestream.id)},
423 metadata=metadata,
424 )
425
426
427def parse_podcast(
428 provider: DeezerProvider, podcast: PodcastFields | SearchSearchResultsPodcastsEdgesNode
429) -> Podcast:
430 """
431 Parse a GQL podcast model to Music Assistant Podcast.
432
433 :param provider: The Deezer provider instance.
434 :param podcast: A GQL podcast model (fragment-based or slim search result).
435 """
436 images: UniqueList[MediaItemImage] = UniqueList()
437 if img := _cover_image(provider, podcast.cover):
438 images.append(img)
439 metadata = MediaItemMetadata(
440 images=images,
441 explicit=getattr(podcast, "is_explicit", None),
442 )
443 if description := getattr(podcast, "description", None):
444 metadata.description = description
445 item = Podcast(
446 item_id=podcast.id,
447 provider=provider.instance_id,
448 name=podcast.display_title,
449 media_type=MediaType.PODCAST,
450 provider_mappings={_provider_mapping(provider, podcast.id)},
451 metadata=metadata,
452 )
453 if getattr(podcast, "is_favorite", False):
454 item.favorite = True
455 return item
456
457
458def parse_podcast_episode(
459 provider: DeezerProvider,
460 episode: PodcastEpisodeFields,
461 podcast: Podcast | ItemMapping,
462 position: int = 0,
463 podcast_image_url: str | None = None,
464) -> PodcastEpisode:
465 """
466 Parse a GQL podcast episode model to Music Assistant PodcastEpisode.
467
468 :param provider: The Deezer provider instance.
469 :param episode: A GQL podcast episode model inheriting from PodcastEpisodeFields.
470 :param podcast: Parent podcast reference.
471 :param position: Sort position / episode number.
472 :param podcast_image_url: Fallback image from parent podcast if episode has none.
473 """
474 images: UniqueList[MediaItemImage] = UniqueList()
475 if img := _cover_image(provider, episode.cover):
476 images.append(img)
477 elif podcast_image_url:
478 images.append(
479 MediaItemImage(
480 type=ImageType.THUMB,
481 path=podcast_image_url,
482 provider=provider.instance_id,
483 remotely_accessible=True,
484 )
485 )
486 metadata = MediaItemMetadata(images=images)
487 if episode.description:
488 metadata.description = episode.description
489 episode_name = episode.title
490 if episode.publication_date:
491 if pub_date := parse_date(str(episode.publication_date)):
492 metadata.release_date = pub_date
493 episode_name = f"{pub_date.strftime('%Y-%m-%d')} - {episode.title}"
494 return PodcastEpisode(
495 item_id=episode.id,
496 provider=provider.instance_id,
497 name=episode_name,
498 duration=episode.duration,
499 podcast=podcast,
500 position=position,
501 media_type=MediaType.PODCAST_EPISODE,
502 provider_mappings={_provider_mapping(provider, episode.id)},
503 metadata=metadata,
504 )
505
506
507def parse_audiobook(provider: DeezerProvider, audiobook: AudiobookFields) -> Audiobook:
508 """
509 Parse a GQL audiobook model to Music Assistant Audiobook.
510
511 :param provider: The Deezer provider instance.
512 :param audiobook: A GQL audiobook model inheriting from AudiobookFields.
513 """
514 images: UniqueList[MediaItemImage] = UniqueList()
515 if img := _cover_image(provider, audiobook.cover):
516 images.append(img)
517 metadata = MediaItemMetadata(
518 images=images,
519 explicit=audiobook.is_explicit,
520 )
521 if audiobook.description:
522 metadata.description = audiobook.description
523 if audiobook.fans_count:
524 metadata.popularity = int(audiobook.fans_count)
525
526 authors: UniqueList[str | Artist | ItemMapping] = UniqueList()
527 narrators: UniqueList[str | Artist | ItemMapping] = UniqueList()
528 for edge in audiobook.contributors.edges:
529 if AudiobookContributorRoles.NARRATOR in edge.roles:
530 narrators.append(edge.node.name)
531 else:
532 authors.append(edge.node.name)
533
534 item = Audiobook(
535 item_id=audiobook.id,
536 provider=provider.instance_id,
537 name=audiobook.display_title or audiobook.id,
538 duration=audiobook.duration,
539 publisher=audiobook.publisher,
540 authors=authors,
541 narrators=narrators,
542 media_type=MediaType.AUDIOBOOK,
543 provider_mappings={_provider_mapping(provider, audiobook.id)},
544 metadata=metadata,
545 )
546 if audiobook.is_favorite:
547 item.favorite = True
548 return item
549
550
551def parse_audiobook_from_album(
552 provider: DeezerProvider, album: AlbumFields | SearchSearchResultsAlbumsEdgesNode
553) -> Audiobook:
554 """
555 Create an Audiobook from an AlbumFields result (search context).
556
557 :param provider: The Deezer provider instance.
558 :param album: A GQL album model (fragment-based or slim search result).
559 """
560 images: UniqueList[MediaItemImage] = UniqueList()
561 if img := _cover_image(provider, album.cover):
562 images.append(img)
563
564 authors: UniqueList[str | Artist | ItemMapping] = UniqueList()
565 for edge in album.contributors.edges:
566 if edge.node is not None:
567 authors.append(edge.node.name)
568
569 return Audiobook(
570 item_id=album.id,
571 provider=provider.instance_id,
572 name=album.display_title,
573 authors=authors,
574 media_type=MediaType.AUDIOBOOK,
575 provider_mappings={_provider_mapping(provider, album.id)},
576 metadata=MediaItemMetadata(
577 images=images,
578 explicit=getattr(album, "is_explicit", None),
579 ),
580 )
581
582
583def parse_audiobook_chapters(
584 chapter_edges: Sequence[GetAudiobookAudiobookChaptersEdges],
585) -> list[MediaItemChapter]:
586 """
587 Build MediaItemChapter list from audiobook chapter edges.
588
589 :param chapter_edges: List of chapter edge objects with `.node` attribute.
590 """
591 chapters: list[MediaItemChapter] = []
592 cumulative_seconds = 0.0
593 for idx, edge in enumerate(chapter_edges):
594 if edge.node is None:
595 continue
596 duration = float(edge.node.duration)
597 chapters.append(
598 MediaItemChapter(
599 position=idx + 1,
600 name=edge.node.display_title,
601 start=cumulative_seconds,
602 end=cumulative_seconds + duration,
603 )
604 )
605 cumulative_seconds += duration
606 return chapters
607
608
609def parse_recently_played_edges(
610 provider: DeezerProvider,
611 edges: list[GetRecentlyPlayedMeRecentlyPlayedEdges],
612) -> list[MediaItemType]:
613 """Parse recently played edges into MediaItemType list."""
614 items: list[MediaItemType] = []
615 for edge in edges:
616 node = edge.node
617 if node is None:
618 continue
619 if isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodeAlbum):
620 items.append(parse_album(provider, node))
621 elif isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodePlaylist):
622 items.append(parse_playlist(provider, node))
623 elif isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodeArtist):
624 items.append(parse_artist(provider, node))
625 elif isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlow):
626 cover = node.cover.urls[0] if node.cover and node.cover.urls else None
627 items.append(
628 create_virtual_playlist(provider, FLOW_PLAYLIST_ID, node.title, image_url=cover)
629 )
630 elif isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlowConfig):
631 cover = get_flow_config_image(node)
632 playlist_id = f"{FLOW_CONFIG_PREFIX}{node.id}"
633 items.append(
634 create_virtual_playlist(
635 provider, playlist_id, f"Flow: {node.title}", image_url=cover
636 )
637 )
638 elif isinstance(node, GetRecentlyPlayedMeRecentlyPlayedEdgesNodeSmartTracklist):
639 cover = node.cover.urls[0] if node.cover and node.cover.urls else None
640 items.append(
641 create_virtual_playlist(
642 provider,
643 f"{SMART_TRACKLIST_PREFIX}{node.id}",
644 node.title,
645 image_url=cover,
646 )
647 )
648 return items
649
650
651# -- GW API parsers --
652
653
654def parse_gw_item(provider: DeezerProvider, item: dict[str, Any]) -> MediaItemType | None:
655 """Parse a GW page item to a Music Assistant media item."""
656 item_type = item.get("type")
657 data = item.get("data", {})
658 try:
659 if item_type == "album" and data.get("ALB_ID"):
660 return parse_gw_audiobook(provider, data)
661 if item_type == "playlist" and data.get("PLAYLIST_ID"):
662 return parse_gw_playlist(provider, data)
663 if item_type == "artist" and data.get("ART_ID"):
664 return parse_gw_artist(provider, data)
665 except KeyError:
666 provider.logger.debug("Incomplete GW item data for type=%s, skipping", item_type)
667 return None
668
669
670def parse_gw_audiobook(provider: DeezerProvider, data: dict[str, Any]) -> Audiobook:
671 """Parse a GW page album item to Music Assistant Audiobook."""
672 album_id = str(data["ALB_ID"])
673 title = data.get("ALB_TITLE", "")
674
675 images: UniqueList[MediaItemImage] = UniqueList()
676 if md5 := data.get("ALB_PICTURE"):
677 images.append(
678 MediaItemImage(
679 type=ImageType.THUMB,
680 path=deezer_cover_url(md5, "cover"),
681 provider=provider.instance_id,
682 remotely_accessible=True,
683 )
684 )
685
686 authors: UniqueList[str | Artist | ItemMapping] = UniqueList()
687 if art_name := data.get("ART_NAME"):
688 authors.append(art_name)
689
690 return Audiobook(
691 item_id=album_id,
692 provider=provider.instance_id,
693 name=title,
694 authors=authors,
695 media_type=MediaType.AUDIOBOOK,
696 provider_mappings={_provider_mapping(provider, album_id)},
697 metadata=MediaItemMetadata(images=images),
698 )
699
700
701def parse_gw_playlist(provider: DeezerProvider, data: dict[str, Any]) -> Playlist:
702 """Parse a GW page playlist item to Music Assistant Playlist."""
703 playlist_id = str(data["PLAYLIST_ID"])
704
705 images: UniqueList[MediaItemImage] = UniqueList()
706 if md5 := data.get("PLAYLIST_PICTURE"):
707 images.append(
708 MediaItemImage(
709 type=ImageType.THUMB,
710 path=deezer_cover_url(md5, "playlist"),
711 provider=provider.instance_id,
712 remotely_accessible=True,
713 )
714 )
715
716 metadata = MediaItemMetadata(images=images) if images else MediaItemMetadata()
717 if desc := data.get("DESCRIPTION"):
718 metadata.description = desc
719
720 return Playlist(
721 item_id=playlist_id,
722 provider=provider.instance_id,
723 name=data.get("TITLE", ""),
724 media_type=MediaType.PLAYLIST,
725 provider_mappings={_provider_mapping(provider, playlist_id)},
726 metadata=metadata,
727 owner=data.get("PARENT_USERNAME", "Deezer"),
728 )
729
730
731def parse_gw_artist(provider: DeezerProvider, data: dict[str, Any]) -> Artist:
732 """Parse a GW page artist item to Music Assistant Artist."""
733 artist_id = str(data["ART_ID"])
734
735 images: UniqueList[MediaItemImage] = UniqueList()
736 if md5 := data.get("ART_PICTURE"):
737 images.append(
738 MediaItemImage(
739 type=ImageType.THUMB,
740 path=deezer_cover_url(md5, "artist"),
741 provider=provider.instance_id,
742 remotely_accessible=True,
743 )
744 )
745
746 return Artist(
747 item_id=artist_id,
748 provider=provider.instance_id,
749 name=data.get("ART_NAME", ""),
750 media_type=MediaType.ARTIST,
751 provider_mappings={_provider_mapping(provider, artist_id)},
752 metadata=MediaItemMetadata(images=images),
753 )
754
755
756def parse_gw_track(provider: DeezerProvider, song: dict[str, Any], position: int = 0) -> Track:
757 """
758 Parse a GW API song dict into a Music Assistant Track.
759
760 :param provider: The Deezer provider instance.
761 :param song: Raw song dict from the GW API (personal_song.getList, etc.).
762 :param position: Position in a track list.
763 """
764 song_id = str(song["SNG_ID"])
765 is_personal = int(song_id) < 0
766 artists: UniqueList[Artist | ItemMapping] = UniqueList()
767 art_name = song.get("ART_NAME", "")
768 art_id = str(song.get("ART_ID", "0"))
769 if art_name:
770 if is_personal or art_id == "0":
771 # Personal tracks have ART_ID=0 which doesn't exist on Deezer.
772 # Create a full Artist object so MA doesn't try to resolve it.
773 # Use a prefixed ID to avoid collisions with the track's provider mapping.
774 personal_art_id = f"{PERSONAL_ARTIST_PREFIX}{song_id}"
775 artists.append(
776 Artist(
777 item_id=personal_art_id,
778 provider=provider.instance_id,
779 name=art_name,
780 favorite=True,
781 provider_mappings={_provider_mapping(provider, personal_art_id)},
782 )
783 )
784 else:
785 artists.append(
786 ItemMapping(
787 media_type=MediaType.ARTIST,
788 item_id=art_id,
789 provider=provider.instance_id,
790 name=art_name,
791 )
792 )
793
794 album: Album | ItemMapping | None = None
795 alb_title = song.get("ALB_TITLE", "")
796 alb_id = str(song.get("ALB_ID", 0))
797 if alb_title:
798 if is_personal or alb_id == "0":
799 # Personal tracks have ALB_ID=0 which doesn't exist on Deezer.
800 # Create a full Album object so MA doesn't try to resolve it.
801 # Use a prefixed ID to avoid collisions with the track's provider mapping.
802 personal_alb_id = f"{PERSONAL_ALBUM_PREFIX}{song_id}"
803 album = Album(
804 item_id=personal_alb_id,
805 provider=provider.instance_id,
806 name=alb_title,
807 favorite=True,
808 artists=artists,
809 provider_mappings={_provider_mapping(provider, personal_alb_id)},
810 )
811 else:
812 album = ItemMapping(
813 media_type=MediaType.ALBUM,
814 item_id=alb_id,
815 provider=provider.instance_id,
816 name=alb_title,
817 )
818
819 name, version = parse_title_and_version(song.get("SNG_TITLE", ""))
820 return Track(
821 item_id=song_id,
822 provider=provider.instance_id,
823 name=name,
824 version=version,
825 duration=int(song.get("DURATION", 0)),
826 favorite=is_personal,
827 artists=artists,
828 album=album,
829 provider_mappings={_provider_mapping(provider, song_id, available=True)},
830 position=position,
831 )
832
833
834# -- Helper functions --
835
836
837def deezer_cover_url(md5: str, image_type: str = "cover", size: int = 500) -> str:
838 """Construct a Deezer CDN image URL from an MD5 hash."""
839 return f"{DEEZER_CDN_IMAGE}/{image_type}/{md5}/{size}x{size}-000000-80-0-0.jpg"
840
841
842def get_flow_config_image(
843 node: GetFlowConfigTracksFlowConfig
844 | GetFlowConfigsMeFlowConfigsMoodsEdgesNode
845 | GetFlowConfigsMeFlowConfigsGenresEdgesNode
846 | SearchFlowsSearchResultsFlowConfigsEdgesNode
847 | GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlowConfig,
848) -> str | None:
849 """Extract the square icon URL from a FlowConfig node's visuals."""
850 icon = node.visuals.hardware_square_icon
851 if icon and icon.urls:
852 url: str = icon.urls[0]
853 return url
854 return None
855
856
857def get_gw_item_image(provider: DeezerProvider, item: dict[str, Any]) -> MediaItemImage | None:
858 """Extract a cover image from a GW page item."""
859 data = item.get("data", {})
860 item_type = item.get("type")
861 md5: str | None = None
862 img_type = "cover"
863
864 if item_type == "album":
865 md5 = data.get("ALB_PICTURE")
866 elif item_type == "playlist":
867 md5 = data.get("PLAYLIST_PICTURE")
868 img_type = "playlist"
869 elif item_type == "artist":
870 md5 = data.get("ART_PICTURE")
871 img_type = "artist"
872 elif item_type == "channel":
873 pictures = item.get("pictures", [])
874 if pictures:
875 md5 = pictures[0].get("md5")
876 img_type = pictures[0].get("type", "misc")
877
878 if md5:
879 return MediaItemImage(
880 type=ImageType.THUMB,
881 path=deezer_cover_url(md5, img_type),
882 provider=provider.instance_id,
883 remotely_accessible=True,
884 )
885 return None
886
887
888def apply_web_url(item: Artist | Album, gql_result: _HasUrl) -> None:
889 """Set web URL on provider mappings if available in the GQL result."""
890 if web_url := getattr(gql_result.url, "web_url", None):
891 for pm in item.provider_mappings:
892 pm.url = web_url
893
894
895def map_album_type(deezer_type: DeezerAlbumType | None, title: str) -> AlbumType:
896 """Map Deezer album type to Music Assistant AlbumType."""
897 inferred = infer_album_type(title, "")
898 if inferred in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
899 return inferred
900 if deezer_type is None:
901 return AlbumType.UNKNOWN
902 match deezer_type:
903 case DeezerAlbumType.ALBUM:
904 return AlbumType.ALBUM
905 case DeezerAlbumType.SINGLES:
906 return AlbumType.SINGLE
907 case DeezerAlbumType.EP:
908 return AlbumType.EP
909 case DeezerAlbumType.COMPILATIONS:
910 return AlbumType.COMPILATION
911 case _:
912 return AlbumType.UNKNOWN
913
914
915def parse_date(date_value: str | None) -> datetime | None:
916 """Parse a date value from the GQL API to a timezone-aware datetime."""
917 try:
918 return datetime.fromisoformat(str(date_value))
919 except ValueError, TypeError:
920 return None
921