/
/
/
1"""
2Shared helper functions for the Deezer provider.
3
4Utility functions used across multiple modules (parsers, browse, media, streaming).
5"""
6
7from __future__ import annotations
8
9from dataclasses import dataclass
10from typing import TYPE_CHECKING
11
12from music_assistant_models.enums import ImageType, MediaType
13from music_assistant_models.media_items import (
14 MediaItemImage,
15 MediaItemMetadata,
16 Playlist,
17 ProviderMapping,
18 UniqueList,
19)
20
21from .constants import (
22 AUDIOBOOK_CHAPTERS_PAGE_SIZE,
23 FLOW_CONFIG_PREFIX,
24 FLOW_PLAYLIST_ID,
25 PERSONAL_SONGS_PLAYLIST_ID,
26 RECOMMENDED_TRACKS_PLAYLIST_ID,
27 SHAKER_CURATED_PREFIX,
28 SHAKER_PREFIX,
29 SMART_TRACKLIST_PREFIX,
30 TOP_CHARTS_PLAYLIST_ID,
31 USER_TOP_TRACKS_PLAYLIST_ID,
32)
33
34if TYPE_CHECKING:
35 from deezer_python_gql import DeezerGQLClient
36 from deezer_python_gql.generated.get_audiobook import (
37 GetAudiobookAudiobookChaptersEdges,
38 GetAudiobookAudiobookChaptersPageInfo,
39 )
40
41 from .provider import DeezerProvider
42
43
44@dataclass(frozen=True)
45class VirtualPlaylistMeta:
46 """Canonical metadata for a virtual playlist type."""
47
48 name: str
49 is_dynamic: bool = False
50 # translation_key for localizing the (fixed) display name; only set for exact-id
51 # entries whose name is stable. Prefix-based entries carry dynamic names and stay None.
52 translation_key: str | None = None
53
54
55# Registry of virtual playlist types with their canonical name and is_dynamic flag.
56# Keyed by exact playlist ID for fixed IDs, and by prefix for parameterized IDs.
57VIRTUAL_PLAYLIST_TYPES: dict[str, VirtualPlaylistMeta] = {
58 FLOW_PLAYLIST_ID: VirtualPlaylistMeta("Flow", is_dynamic=True, translation_key="flow"),
59 FLOW_CONFIG_PREFIX: VirtualPlaylistMeta("Flow", is_dynamic=True),
60 SMART_TRACKLIST_PREFIX: VirtualPlaylistMeta("Mix"),
61 RECOMMENDED_TRACKS_PLAYLIST_ID: VirtualPlaylistMeta(
62 "Hot Tracks", translation_key="recommended_tracks"
63 ),
64 TOP_CHARTS_PLAYLIST_ID: VirtualPlaylistMeta("Top Charts", translation_key="top_charts"),
65 USER_TOP_TRACKS_PLAYLIST_ID: VirtualPlaylistMeta(
66 "Your Top Tracks", translation_key="user_top_tracks"
67 ),
68 PERSONAL_SONGS_PLAYLIST_ID: VirtualPlaylistMeta("My Uploads", translation_key="personal_songs"),
69 SHAKER_PREFIX: VirtualPlaylistMeta("Mix", is_dynamic=True),
70 SHAKER_CURATED_PREFIX: VirtualPlaylistMeta("Playlist"),
71}
72
73
74def get_virtual_playlist_meta(item_id: str) -> VirtualPlaylistMeta | None:
75 """
76 Look up canonical metadata for a virtual playlist by its item_id.
77
78 Tries exact match first, then longest prefix match.
79 """
80 if item_id in VIRTUAL_PLAYLIST_TYPES:
81 return VIRTUAL_PLAYLIST_TYPES[item_id]
82 # Sort by prefix length descending so longer prefixes match first
83 # (e.g. "shaker_curated_" before "shaker_")
84 for prefix, meta in sorted(
85 VIRTUAL_PLAYLIST_TYPES.items(), key=lambda x: len(x[0]), reverse=True
86 ):
87 if prefix.endswith("_") and item_id.startswith(prefix):
88 return meta
89 return None
90
91
92def create_virtual_playlist(
93 provider: DeezerProvider,
94 item_id: str,
95 name: str,
96 image_url: str | None = None,
97 is_dynamic: bool | None = None,
98) -> Playlist:
99 """
100 Create a virtual playlist for Flow, recommended content, etc.
101
102 :param provider: The Deezer provider instance.
103 :param item_id: The unique identifier (e.g., "flow", "smart_tracklist_123").
104 :param name: Display name for the playlist.
105 :param image_url: Optional cover image URL.
106 :param is_dynamic: Whether the playlist returns fresh tracks on each fetch.
107 If None, the value is looked up from the virtual playlist registry.
108 """
109 if is_dynamic is None:
110 meta = get_virtual_playlist_meta(item_id)
111 is_dynamic = meta.is_dynamic if meta else False
112 # Only exact-id (fixed-name) virtual playlists are localizable; prefix-based ones
113 # (flow configs, shaker groups, smart tracklists) carry dynamic names from the caller.
114 exact_meta = VIRTUAL_PLAYLIST_TYPES.get(item_id)
115 translation_key = exact_meta.translation_key if exact_meta else None
116 images: UniqueList[MediaItemImage] = UniqueList()
117 if image_url:
118 images.append(
119 MediaItemImage(
120 type=ImageType.THUMB,
121 path=image_url,
122 provider=provider.instance_id,
123 remotely_accessible=True,
124 )
125 )
126 return Playlist(
127 item_id=item_id,
128 provider=provider.instance_id,
129 name=name,
130 media_type=MediaType.PLAYLIST,
131 provider_mappings={
132 ProviderMapping(
133 item_id=item_id,
134 provider_domain=provider.domain,
135 provider_instance=provider.instance_id,
136 )
137 },
138 metadata=MediaItemMetadata(images=images) if images else MediaItemMetadata(),
139 is_editable=False,
140 is_dynamic=is_dynamic,
141 translation_key=translation_key,
142 owner="Deezer",
143 )
144
145
146async def fetch_all_audiobook_chapter_edges(
147 gql_client: DeezerGQLClient,
148 audiobook_id: str,
149 page_size: int = AUDIOBOOK_CHAPTERS_PAGE_SIZE,
150 *,
151 initial_edges: list[GetAudiobookAudiobookChaptersEdges] | None = None,
152 initial_page_info: GetAudiobookAudiobookChaptersPageInfo | None = None,
153) -> list[GetAudiobookAudiobookChaptersEdges]:
154 """
155 Paginate through all chapters of an audiobook and return the full edge list.
156
157 :param gql_client: The Deezer GQL client to use.
158 :param audiobook_id: The audiobook ID to fetch chapters for.
159 :param page_size: Number of chapters per page.
160 :param initial_edges: Pre-fetched edges to avoid re-fetching the first page.
161 :param initial_page_info: Page info from the pre-fetched result.
162 """
163 if initial_edges is not None and initial_page_info is not None:
164 all_edges = list(initial_edges)
165 page_info = initial_page_info
166 else:
167 result = await gql_client.get_audiobook(audiobook_id=audiobook_id, chapters_first=page_size)
168 if result is None:
169 return []
170 all_edges = list(result.chapters.edges)
171 page_info = result.chapters.page_info
172 while page_info.has_next_page:
173 next_page = await gql_client.get_audiobook(
174 audiobook_id=audiobook_id,
175 chapters_first=page_size,
176 chapters_after=page_info.end_cursor,
177 )
178 if next_page is None:
179 break
180 all_edges.extend(next_page.chapters.edges)
181 page_info = next_page.chapters.page_info
182 return all_edges
183
184
185async def fetch_all_bookmarks(gql_client: DeezerGQLClient) -> dict[str, tuple[bool, int]]:
186 """
187 Paginate through all podcast episode bookmarks and return a lookup dict.
188
189 :param gql_client: The Deezer GQL client to use.
190 :returns: Dict mapping episode ID to (is_played, position_ms).
191 """
192 bookmarks: dict[str, tuple[bool, int]] = {}
193 cursor: str | None = None
194 while True:
195 result = await gql_client.get_podcast_episode_bookmarks(first=50, after=cursor)
196 if not result:
197 break
198 for edge in result.podcast_episode_bookmarks.edges:
199 if edge.node is not None:
200 bookmarks[edge.node.episode.id] = (
201 edge.node.is_played,
202 edge.node.position * 1000,
203 )
204 if not result.podcast_episode_bookmarks.page_info.has_next_page:
205 break
206 cursor = result.podcast_episode_bookmarks.page_info.end_cursor
207 return bookmarks
208