/
/
/
1"""
2Browse and recommendations manager for the Deezer provider.
3
4Handles browse tree routing, recommendation folders, virtual playlist
5infrastructure, and all track-fetching methods for virtual playlists.
6"""
7
8from __future__ import annotations
9
10from collections.abc import Callable, Coroutine, Sequence
11from typing import TYPE_CHECKING
12
13from deezer_python_gql.generated.enums import (
14 MusicTogetherRefreshSuggestedTracklistMoodInput,
15 MusicTogetherSuggestedTracklistMoodInput,
16)
17from deezer_python_gql.generated.get_made_for_me import (
18 GetMadeForMeMeMadeForMeEdgesNodeSmartTracklist,
19)
20from music_assistant_models.media_items import (
21 BrowseFolder,
22 ItemMapping,
23 MediaItemType,
24 Playlist,
25 RecommendationFolder,
26 Track,
27 UniqueList,
28)
29
30from music_assistant.controllers.cache import use_cache
31from music_assistant.helpers.track_filter import filter_tracks
32
33from .constants import (
34 BROWSE_ALL_FLOWS,
35 BROWSE_AUDIOBOOKS,
36 BROWSE_EXPLORE,
37 BROWSE_GENRES,
38 BROWSE_MADE_FOR_YOU,
39 BROWSE_MOODS,
40 BROWSE_PERSONALIZED_PLAYLISTS,
41 BROWSE_RECENTLY_PLAYED,
42 BROWSE_RECOMMENDED_ARTIST_PLAYLISTS,
43 BROWSE_RECOMMENDED_PLAYLISTS,
44 BROWSE_SHAKER,
45 BROWSE_TOP_ALBUMS,
46 BROWSE_TOP_ARTISTS,
47 BROWSE_TOP_PLAYLISTS,
48 BROWSE_YOUR_TOP_ALBUMS,
49 BROWSE_YOUR_TOP_ARTISTS,
50 DEFAULT_FLOW_CONFIG_ID,
51 FLOW_BATCH_COUNT,
52 FLOW_CONFIG_PREFIX,
53 FLOW_PLAYLIST_ID,
54 PERSONAL_SONGS_PLAYLIST_ID,
55 RECOMMENDED_TRACKS_PLAYLIST_ID,
56 SHAKER_CURATED_PREFIX,
57 SHAKER_MIX_COVER,
58 SHAKER_PREFIX,
59 SMART_TRACKLIST_PREFIX,
60 TOP_CHARTS_PLAYLIST_ID,
61 USER_TOP_TRACKS_PLAYLIST_ID,
62)
63from .helpers import (
64 create_virtual_playlist,
65)
66from .parsers import (
67 get_flow_config_image,
68 get_gw_item_image,
69 parse_album,
70 parse_artist,
71 parse_gw_item,
72 parse_gw_track,
73 parse_playlist,
74 parse_recently_played_edges,
75 parse_track,
76)
77
78if TYPE_CHECKING:
79 from deezer_python_gql.generated.get_flow_configs import (
80 GetFlowConfigsMeFlowConfigsGenresEdges,
81 GetFlowConfigsMeFlowConfigsMoodsEdges,
82 )
83 from deezer_python_gql.generated.get_recommendations import GetRecommendationsMe
84 from deezer_python_gql.generated.search_flows import (
85 SearchFlowsSearchResultsFlowConfigsEdges,
86 )
87
88 from .provider import DeezerProvider
89
90AUDIOBOOKS_CHANNEL = "channels/audiobooks"
91
92# row item_ids fed by the single shared gql get_recommendations payload fetch
93SHARED_PAYLOAD_ROW_IDS = (
94 "recommended_playlists",
95 "recommended_artist_playlists",
96 "recommended_tracks",
97 "new_releases",
98)
99
100
101class DeezerBrowseManager:
102 """Handles browse tree, recommendations, and virtual playlist content."""
103
104 def __init__(self, provider: DeezerProvider) -> None:
105 """Initialize browse manager."""
106 self.provider = provider
107 self.mass = provider.mass
108 self.instance_id = provider.instance_id
109 self.domain = provider.domain
110 self.logger = provider.logger
111 self._browse_slug_cache: dict[str, str] = {}
112
113 # -- Browse routing --
114
115 async def browse(
116 self,
117 path: str,
118 base_browse: Callable[
119 [str], Coroutine[None, None, Sequence[MediaItemType | ItemMapping | BrowseFolder]]
120 ],
121 ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
122 """
123 Browse Deezer content.
124
125 :param path: The browse path.
126 :param base_browse: Coroutine for the base class browse method.
127 """
128 path_parts = path.split("://")[1].split("/") if "://" in path else []
129 subpath = path_parts[0] if path_parts else None
130 sub_subpath = path_parts[1] if len(path_parts) > 1 else None
131
132 if subpath == BROWSE_MADE_FOR_YOU:
133 return await self._browse_made_for_you(path, sub_subpath)
134
135 if subpath == BROWSE_EXPLORE:
136 if sub_subpath:
137 return await self._browse_explore_category(sub_subpath)
138 return await self._browse_explore_root(path)
139
140 if subpath == BROWSE_RECENTLY_PLAYED:
141 return await self._get_recently_played_items()
142
143 if subpath == BROWSE_SHAKER:
144 if sub_subpath:
145 group_id = self._browse_slug_cache.get(f"shaker/{sub_subpath}", sub_subpath)
146 return await self._browse_shaker_group(group_id)
147 return await self._browse_shaker_root(path)
148
149 if subpath == BROWSE_AUDIOBOOKS:
150 if sub_subpath:
151 page_path = self._browse_slug_cache.get(
152 f"audiobooks/{sub_subpath}", f"channels/{sub_subpath}"
153 )
154 return await self._browse_audiobooks_page(page_path)
155 return await self._browse_audiobooks_root(path)
156
157 if not subpath:
158 # Root: add custom folders alongside standard ones
159 # Filter out the Recommendations folder â our custom folders cover that content
160 base_items = [
161 item
162 for item in await base_browse(path)
163 if not (isinstance(item, BrowseFolder) and item.item_id == "recommendations")
164 ]
165 base = path if path.endswith("//") else path.rstrip("/") + "/"
166 base_items.extend(
167 [
168 BrowseFolder(
169 item_id="made_for_me",
170 provider=self.instance_id,
171 path=f"{base}{BROWSE_MADE_FOR_YOU}",
172 name=BROWSE_MADE_FOR_YOU,
173 translation_key="made_for_me",
174 ),
175 BrowseFolder(
176 item_id="explore",
177 provider=self.instance_id,
178 path=f"{base}{BROWSE_EXPLORE}",
179 name=BROWSE_EXPLORE,
180 translation_key="explore",
181 ),
182 BrowseFolder(
183 item_id="recently_played",
184 provider=self.instance_id,
185 path=f"{base}{BROWSE_RECENTLY_PLAYED}",
186 name=BROWSE_RECENTLY_PLAYED,
187 translation_key="recently_played",
188 ),
189 BrowseFolder(
190 item_id="shaker",
191 provider=self.instance_id,
192 path=f"{base}{BROWSE_SHAKER}",
193 name=BROWSE_SHAKER,
194 translation_key="shaker",
195 ),
196 BrowseFolder(
197 item_id="discover_audiobooks",
198 provider=self.instance_id,
199 path=f"{base}{BROWSE_AUDIOBOOKS}",
200 name=BROWSE_AUDIOBOOKS,
201 translation_key="discover_audiobooks",
202 ),
203 create_virtual_playlist(
204 self.provider, PERSONAL_SONGS_PLAYLIST_ID, "My Uploads"
205 ),
206 ]
207 )
208 return base_items
209
210 # Standard paths handled by base class
211 return list(await base_browse(path))
212
213 # -- Recommendations --
214
215 async def get_recommendations(self) -> list[RecommendationFolder]:
216 """Get the static descriptors of Deezer's recommendation rows, without items."""
217 return [
218 RecommendationFolder(
219 item_id="made_for_you",
220 provider=self.instance_id,
221 name=BROWSE_MADE_FOR_YOU,
222 translation_key="made_for_you",
223 ),
224 RecommendationFolder(
225 item_id="recommended_playlists",
226 provider=self.instance_id,
227 name=BROWSE_RECOMMENDED_PLAYLISTS,
228 translation_key="recommended_playlists",
229 ),
230 RecommendationFolder(
231 item_id="recommended_artist_playlists",
232 provider=self.instance_id,
233 name=BROWSE_RECOMMENDED_ARTIST_PLAYLISTS,
234 translation_key="recommended_artist_playlists",
235 ),
236 RecommendationFolder(
237 item_id="recommended_tracks",
238 provider=self.instance_id,
239 name="Hot Tracks",
240 translation_key="recommended_tracks",
241 ),
242 RecommendationFolder(
243 item_id="new_releases",
244 provider=self.instance_id,
245 name="New Releases",
246 translation_key="new_releases",
247 ),
248 RecommendationFolder(
249 item_id="mood_flows",
250 provider=self.instance_id,
251 name="Deezer Mood Flows",
252 translation_key="mood_flows",
253 ),
254 RecommendationFolder(
255 item_id="genre_flows",
256 provider=self.instance_id,
257 name="Deezer Genre Flows",
258 translation_key="genre_flows",
259 ),
260 RecommendationFolder(
261 item_id="recently_played",
262 provider=self.instance_id,
263 name=BROWSE_RECENTLY_PLAYED,
264 translation_key="recently_played",
265 ),
266 ]
267
268 async def get_recommendation_items(
269 self, item_id: str
270 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
271 """
272 Get the items for a single recommendation row.
273
274 :param item_id: The item_id of the row, as returned by get_recommendations
275 (empty result if unknown).
276 """
277 if item_id in SHARED_PAYLOAD_ROW_IDS:
278 return await self.provider._recommendation_items_from_payload(item_id)
279 if item_id == "made_for_you":
280 folders: list[RecommendationFolder] = []
281 await self._add_made_for_you(folders)
282 return folders[0].items if folders else UniqueList()
283 if item_id in ("mood_flows", "genre_flows"):
284 return next(
285 (
286 folder.items
287 for folder in await self._get_flow_folders()
288 if folder.item_id == item_id
289 ),
290 UniqueList(),
291 )
292 if item_id == "recently_played":
293 return UniqueList(await self._get_recently_played_items())
294 return UniqueList()
295
296 # -- Made For You --
297
298 async def _browse_made_for_you(
299 self, path: str, sub_subpath: str | None
300 ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
301 """Route Made For You sub-paths or return the root listing."""
302 if sub_subpath in (BROWSE_MOODS, BROWSE_GENRES):
303 return await self._browse_flow_configs(sub_subpath.lower())
304 if sub_subpath == BROWSE_YOUR_TOP_ARTISTS:
305 return await self._browse_user_charts_category("your_top_artists")
306 if sub_subpath == BROWSE_YOUR_TOP_ALBUMS:
307 return await self._browse_user_charts_category("your_top_albums")
308 if sub_subpath == BROWSE_RECOMMENDED_PLAYLISTS:
309 return await self._browse_editorial_playlists()
310 if sub_subpath == BROWSE_RECOMMENDED_ARTIST_PLAYLISTS:
311 return await self._browse_artist_playlists()
312 if sub_subpath == BROWSE_PERSONALIZED_PLAYLISTS:
313 return await self._get_smart_tracklist_playlists()
314 return await self._browse_made_for_me(path)
315
316 async def _browse_made_for_me(self, path: str) -> list[MediaItemType | BrowseFolder]:
317 """Return Made For You sub-items: Moods, Genres, Top stats, Mixes, Playlists."""
318 base = path if path.endswith("/") else path + "/"
319 items: list[MediaItemType | BrowseFolder] = [
320 BrowseFolder(
321 item_id="moods",
322 provider=self.instance_id,
323 path=f"{base}{BROWSE_MOODS}",
324 name=BROWSE_MOODS,
325 translation_key="moods",
326 ),
327 BrowseFolder(
328 item_id="genres",
329 provider=self.instance_id,
330 path=f"{base}{BROWSE_GENRES}",
331 name=BROWSE_GENRES,
332 translation_key="genres",
333 ),
334 create_virtual_playlist(self.provider, USER_TOP_TRACKS_PLAYLIST_ID, "Your Top Tracks"),
335 create_virtual_playlist(self.provider, RECOMMENDED_TRACKS_PLAYLIST_ID, "Hot Tracks"),
336 BrowseFolder(
337 item_id="your_top_artists",
338 provider=self.instance_id,
339 path=f"{base}{BROWSE_YOUR_TOP_ARTISTS}",
340 name=BROWSE_YOUR_TOP_ARTISTS,
341 translation_key="your_top_artists",
342 ),
343 BrowseFolder(
344 item_id="your_top_albums",
345 provider=self.instance_id,
346 path=f"{base}{BROWSE_YOUR_TOP_ALBUMS}",
347 name=BROWSE_YOUR_TOP_ALBUMS,
348 translation_key="your_top_albums",
349 ),
350 BrowseFolder(
351 item_id="mixes",
352 provider=self.instance_id,
353 path=f"{base}{BROWSE_PERSONALIZED_PLAYLISTS}",
354 name=BROWSE_PERSONALIZED_PLAYLISTS,
355 translation_key="mixes",
356 ),
357 BrowseFolder(
358 item_id="recommended_playlists",
359 provider=self.instance_id,
360 path=f"{base}{BROWSE_RECOMMENDED_PLAYLISTS}",
361 name=BROWSE_RECOMMENDED_PLAYLISTS,
362 translation_key="recommended_playlists",
363 ),
364 BrowseFolder(
365 item_id="recommended_artist_playlists",
366 provider=self.instance_id,
367 path=f"{base}{BROWSE_RECOMMENDED_ARTIST_PLAYLISTS}",
368 name=BROWSE_RECOMMENDED_ARTIST_PLAYLISTS,
369 translation_key="recommended_artist_playlists",
370 ),
371 ]
372 return items
373
374 async def _browse_editorial_playlists(self) -> list[Playlist]:
375 """Fetch personalized editorial playlists from Deezer recommendations."""
376 recs = await self.provider.gql_client.get_recommendations(
377 playlists_first=50,
378 artist_playlists_first=0,
379 new_releases_first=0,
380 artists_first=0,
381 hot_tracks_limit=0,
382 )
383 if not recs or not recs.recommendations:
384 return []
385 return [
386 parse_playlist(self.provider, edge.node)
387 for edge in recs.recommendations.playlists.edges
388 if edge.node is not None
389 ]
390
391 async def _browse_artist_playlists(self) -> list[Playlist]:
392 """Fetch personalized artist playlists from Deezer recommendations."""
393 recs = await self.provider.gql_client.get_recommendations(
394 playlists_first=0,
395 artist_playlists_first=50,
396 new_releases_first=0,
397 artists_first=0,
398 hot_tracks_limit=0,
399 )
400 if not recs or not recs.recommendations:
401 return []
402 return [
403 parse_playlist(self.provider, edge.node)
404 for edge in recs.recommendations.artist_playlists.edges
405 if edge.node is not None
406 ]
407
408 # -- Flow configs --
409
410 def _flow_configs_to_playlists(
411 self,
412 edges: Sequence[
413 GetFlowConfigsMeFlowConfigsMoodsEdges
414 | GetFlowConfigsMeFlowConfigsGenresEdges
415 | SearchFlowsSearchResultsFlowConfigsEdges
416 ],
417 ) -> list[Playlist]:
418 """Convert FlowConfig edges to virtual playlists."""
419 return [
420 create_virtual_playlist(
421 self.provider,
422 f"{FLOW_CONFIG_PREFIX}{edge.node.id}",
423 f"Flow: {edge.node.title}",
424 image_url=get_flow_config_image(edge.node),
425 )
426 for edge in edges
427 if edge.node is not None
428 ]
429
430 async def _browse_flow_configs(self, category: str) -> list[Playlist]:
431 """
432 Fetch mood or genre flow configs and return as virtual playlists.
433
434 :param category: Either "moods" or "genres".
435 """
436 is_moods = category == "moods"
437 all_edges: list[
438 GetFlowConfigsMeFlowConfigsMoodsEdges | GetFlowConfigsMeFlowConfigsGenresEdges
439 ] = []
440 cursor: str | None = None
441 while True:
442 flow_configs = await self.provider.gql_client.get_flow_configs(
443 moods_first=50 if is_moods else 0,
444 moods_after=cursor if is_moods else None,
445 genres_first=0 if is_moods else 50,
446 genres_after=None if is_moods else cursor,
447 )
448 if not flow_configs or not flow_configs.flow_configs:
449 break
450 connection = (
451 flow_configs.flow_configs.moods if is_moods else flow_configs.flow_configs.genres
452 )
453 all_edges.extend(connection.edges)
454 if not connection.page_info.has_next_page:
455 break
456 cursor = connection.page_info.end_cursor
457 return self._flow_configs_to_playlists(all_edges)
458
459 async def _browse_all_flows(self) -> list[Playlist]:
460 """Fetch all available Deezer flows via search and return as virtual playlists."""
461 all_edges: list[SearchFlowsSearchResultsFlowConfigsEdges] = []
462 cursor: str | None = None
463 while True:
464 result = await self.provider.gql_client.search_flows(
465 query="flow", first=100, after=cursor
466 )
467 if not result:
468 break
469 edges = result.results.flow_configs.edges
470 all_edges.extend(edges)
471 if not result.results.flow_configs.page_info.has_next_page:
472 break
473 cursor = result.results.flow_configs.page_info.end_cursor
474 return self._flow_configs_to_playlists(all_edges)
475
476 # -- Explore --
477
478 async def _browse_explore_root(self, path: str) -> list[MediaItemType | BrowseFolder]:
479 """Return Explore section: charts, top content, all flows."""
480 base = path if path.endswith("/") else path + "/"
481 charts_cover = None
482 charts = await self.provider.gql_client.get_charts(tracks_first=1)
483 if charts and charts.country and charts.country.tracks:
484 for edge in charts.country.tracks.edges:
485 if edge.node and edge.node.album and edge.node.album.cover:
486 if edge.node.album.cover.urls:
487 charts_cover = edge.node.album.cover.urls[0]
488 break
489 return [
490 create_virtual_playlist(
491 self.provider, TOP_CHARTS_PLAYLIST_ID, "Top Charts", image_url=charts_cover
492 ),
493 BrowseFolder(
494 item_id="top_albums",
495 provider=self.instance_id,
496 path=f"{base}{BROWSE_TOP_ALBUMS}",
497 name=BROWSE_TOP_ALBUMS,
498 translation_key="top_albums",
499 ),
500 BrowseFolder(
501 item_id="top_artists",
502 provider=self.instance_id,
503 path=f"{base}{BROWSE_TOP_ARTISTS}",
504 name=BROWSE_TOP_ARTISTS,
505 translation_key="top_artists",
506 ),
507 BrowseFolder(
508 item_id="top_playlists",
509 provider=self.instance_id,
510 path=f"{base}{BROWSE_TOP_PLAYLISTS}",
511 name=BROWSE_TOP_PLAYLISTS,
512 translation_key="top_playlists",
513 ),
514 BrowseFolder(
515 item_id="all_flows",
516 provider=self.instance_id,
517 path=f"{base}{BROWSE_ALL_FLOWS}",
518 name=BROWSE_ALL_FLOWS,
519 translation_key="all_flows",
520 ),
521 ]
522
523 async def _browse_explore_category(self, category: str) -> list[MediaItemType]:
524 """Fetch items for an Explore sub-category."""
525 if category == BROWSE_ALL_FLOWS:
526 return list(await self._browse_all_flows())
527 items: list[MediaItemType] = []
528 if category in (BROWSE_TOP_ALBUMS, BROWSE_TOP_ARTISTS, BROWSE_TOP_PLAYLISTS):
529 charts = await self.provider.gql_client.get_charts(tracks_first=0)
530 if not charts or not charts.country:
531 return []
532 country = charts.country
533 if category == BROWSE_TOP_ALBUMS and country.albums:
534 for album_edge in country.albums.edges:
535 if album_edge.node is not None:
536 items.append(parse_album(self.provider, album_edge.node))
537 elif category == BROWSE_TOP_ARTISTS and country.artists:
538 for artist_edge in country.artists.edges:
539 if artist_edge.node is not None:
540 items.append(parse_artist(self.provider, artist_edge.node))
541 elif category == BROWSE_TOP_PLAYLISTS and country.playlists:
542 for playlist_edge in country.playlists.edges:
543 if playlist_edge.node is not None:
544 items.append(parse_playlist(self.provider, playlist_edge.node))
545 return items
546
547 async def _browse_user_charts_category(self, category: str) -> list[MediaItemType]:
548 """Fetch user chart items (top artists/albums)."""
549 result = await self.provider.gql_client.get_user_charts()
550 if not result:
551 return []
552 charts = result.charts
553 items: list[MediaItemType] = []
554 if category == "your_top_artists" and charts.artists:
555 for artist_edge in charts.artists.edges:
556 if artist_edge.node is not None:
557 items.append(parse_artist(self.provider, artist_edge.node))
558 elif category == "your_top_albums" and charts.albums:
559 for album_edge in charts.albums.edges:
560 if album_edge.node is not None:
561 items.append(parse_album(self.provider, album_edge.node))
562 return items
563
564 # -- Shaker (Music Together) --
565
566 async def _browse_shaker_root(self, path: str) -> list[BrowseFolder]:
567 """Return Shaker (Music Together) groups as browse folders."""
568 base = path if path.endswith("/") else path + "/"
569 folders: list[BrowseFolder] = []
570 cursor: str | None = None
571 while True:
572 result = await self.provider.gql_client.get_music_together_groups(
573 first=50, after=cursor
574 )
575 if not result:
576 break
577 for edge in result.music_together_groups.edges:
578 if edge.node is None:
579 continue
580 group = edge.node
581 members = group.estimated_members_count
582 name = f"{group.name} ({members} member{'s' if members != 1 else ''})"
583 path_name = group.name.replace("/", "-")
584 self._browse_slug_cache[f"shaker/{path_name}"] = group.id
585 folders.append(
586 BrowseFolder(
587 item_id=f"shaker_{group.id}",
588 provider=self.instance_id,
589 path=f"{base}{path_name}",
590 name=name,
591 )
592 )
593 if not result.music_together_groups.page_info.has_next_page:
594 break
595 cursor = result.music_together_groups.page_info.end_cursor
596 return folders
597
598 async def _browse_shaker_group(self, group_id: str) -> list[MediaItemType]:
599 """Return playlists for a Shaker group: mix + curated tracklist."""
600 group = await self.provider.gql_client.get_music_together_group(
601 group_id=group_id,
602 mood=MusicTogetherSuggestedTracklistMoodInput.NONE,
603 tracks_first=1,
604 )
605 if group is None:
606 return []
607 items: list[MediaItemType] = []
608 if group.suggested_tracklist and group.suggested_tracklist.tracklist:
609 items.append(
610 create_virtual_playlist(
611 self.provider,
612 f"{SHAKER_PREFIX}{group_id}",
613 f"{group.name} - Mix",
614 image_url=SHAKER_MIX_COVER,
615 )
616 )
617 if group.curated_tracklist:
618 cover_url = (
619 group.curated_tracklist.picture.urls[0]
620 if group.curated_tracklist.picture and group.curated_tracklist.picture.urls
621 else None
622 )
623 items.append(
624 create_virtual_playlist(
625 self.provider,
626 f"{SHAKER_CURATED_PREFIX}{group_id}",
627 f"{group.name} - Playlist",
628 image_url=cover_url,
629 )
630 )
631 return items
632
633 # -- Audiobooks --
634
635 async def _browse_audiobooks_root(self, path: str) -> list[MediaItemType | BrowseFolder]:
636 """Return audiobook sections from the Deezer audiobooks channel page."""
637 page_data = await self.provider.gw_client.get_page(AUDIOBOOKS_CHANNEL)
638 sections = page_data.get("sections", [])
639 base = path if path.endswith("/") else path + "/"
640 items: list[MediaItemType | BrowseFolder] = []
641
642 for section in sections:
643 title = section.get("title", "")
644 if not title:
645 continue
646 section_items = section.get("items", [])
647 if not section_items:
648 continue
649
650 first_item = section_items[0]
651 if first_item.get("type") == "channel":
652 for item in section_items:
653 data = item.get("data", {})
654 target = data.get("target", "")
655 if not target:
656 continue
657 slug = target.removeprefix("/channels/")
658 channel_name = data.get("name", slug)
659 path_name = channel_name.replace("/", "-")
660 self._browse_slug_cache[f"audiobooks/{path_name}"] = f"channels/{slug}"
661 folder = BrowseFolder(
662 item_id=f"audiobooks_{slug}",
663 provider=self.instance_id,
664 path=f"{base}{path_name}",
665 name=channel_name,
666 )
667 folder.image = get_gw_item_image(self.provider, item)
668 items.append(folder)
669 else:
670 module_id = section.get("module_id", "")
671 if not module_id:
672 continue
673 path_name = title.replace("/", "-")
674 self._browse_slug_cache[f"audiobooks/{path_name}"] = f"channels/module/{module_id}"
675 folder = BrowseFolder(
676 item_id=f"audiobooks_section_{module_id}",
677 provider=self.instance_id,
678 path=f"{base}{path_name}",
679 name=title,
680 )
681 folder.image = get_gw_item_image(self.provider, first_item)
682 items.append(folder)
683
684 return items
685
686 async def _browse_audiobooks_page(self, page_path: str) -> list[MediaItemType | BrowseFolder]:
687 """Fetch a Deezer channel page and return its items."""
688 page_data = await self.provider.gw_client.get_page(page_path)
689 sections = page_data.get("sections", [])
690 items: list[MediaItemType | BrowseFolder] = []
691 for section in sections:
692 for item in section.get("items", []):
693 if parsed := parse_gw_item(self.provider, item):
694 items.append(parsed)
695 return items
696
697 def _add_recommended_tracks(
698 self,
699 result: list[RecommendationFolder],
700 recs: GetRecommendationsMe | None,
701 ) -> None:
702 """Add Hot Tracks section with tracks rendered directly."""
703 if not recs or not recs.recommendations.hot_tracks:
704 return
705 track_items = [parse_track(self.provider, ht) for ht in recs.recommendations.hot_tracks]
706 if track_items:
707 result.append(
708 RecommendationFolder(
709 item_id="recommended_tracks",
710 provider=self.instance_id,
711 name="Hot Tracks",
712 translation_key="recommended_tracks",
713 items=UniqueList(track_items),
714 )
715 )
716
717 def _add_recommended_playlists(
718 self,
719 result: list[RecommendationFolder],
720 recs: GetRecommendationsMe | None,
721 ) -> None:
722 """Add Recommended Playlists section (editorial playlists)."""
723 if not recs or not recs.recommendations:
724 return
725 items = [
726 parse_playlist(self.provider, edge.node)
727 for edge in recs.recommendations.playlists.edges
728 if edge.node is not None
729 ]
730 if items:
731 result.append(
732 RecommendationFolder(
733 item_id="recommended_playlists",
734 provider=self.instance_id,
735 name=BROWSE_RECOMMENDED_PLAYLISTS,
736 translation_key="recommended_playlists",
737 items=UniqueList(items),
738 )
739 )
740
741 def _add_recommended_artist_playlists(
742 self,
743 result: list[RecommendationFolder],
744 recs: GetRecommendationsMe | None,
745 ) -> None:
746 """Add Recommended Artist Playlists section."""
747 if not recs or not recs.recommendations:
748 return
749 items = [
750 parse_playlist(self.provider, edge.node)
751 for edge in recs.recommendations.artist_playlists.edges
752 if edge.node is not None
753 ]
754 if items:
755 result.append(
756 RecommendationFolder(
757 item_id="recommended_artist_playlists",
758 provider=self.instance_id,
759 name=BROWSE_RECOMMENDED_ARTIST_PLAYLISTS,
760 translation_key="recommended_artist_playlists",
761 items=UniqueList(items),
762 )
763 )
764
765 def _add_new_releases(
766 self,
767 result: list[RecommendationFolder],
768 recs: GetRecommendationsMe | None,
769 ) -> None:
770 """Add New Releases section to recommendations."""
771 if recs is None:
772 return
773 new_release_items = [
774 parse_album(self.provider, edge.node)
775 for edge in recs.recommendations.new_releases.edges
776 if edge.node is not None
777 ]
778 if new_release_items:
779 result.append(
780 RecommendationFolder(
781 item_id="new_releases",
782 provider=self.instance_id,
783 name="New Releases",
784 translation_key="new_releases",
785 items=UniqueList(new_release_items),
786 )
787 )
788
789 async def _add_flow_configs(self, result: list[RecommendationFolder]) -> None:
790 """Add Mood and Genre Flow sections to recommendations."""
791 flow_configs = await self.provider.gql_client.get_flow_configs(
792 moods_first=20, genres_first=20
793 )
794 if not flow_configs or not flow_configs.flow_configs:
795 return
796 configs = flow_configs.flow_configs
797 for folder_id, folder_name, edges in (
798 ("mood_flows", "Deezer Mood Flows", configs.moods.edges),
799 ("genre_flows", "Deezer Genre Flows", configs.genres.edges),
800 ):
801 playlists = self._flow_configs_to_playlists(list(edges))
802 if playlists:
803 result.append(
804 RecommendationFolder(
805 item_id=folder_id,
806 provider=self.instance_id,
807 name=folder_name,
808 translation_key=folder_id,
809 items=UniqueList(playlists),
810 )
811 )
812
813 # -- Recently played (shared by browse and recommendations) --
814
815 @use_cache(3600)
816 async def _get_recently_played_items(self) -> list[MediaItemType]:
817 """Get recently played items (cached)."""
818 result = await self.provider.gql_client.get_recently_played(first=50)
819 if not result:
820 return []
821 return parse_recently_played_edges(self.provider, result.recently_played.edges)
822
823 # -- Virtual playlist metadata --
824
825 async def get_virtual_playlist(self, prov_playlist_id: str) -> Playlist | None:
826 """Return a virtual playlist, or None if the ID is not virtual."""
827 if prov_playlist_id == FLOW_PLAYLIST_ID:
828 cover = await self._get_flow_cover()
829 return create_virtual_playlist(self.provider, FLOW_PLAYLIST_ID, "Flow", image_url=cover)
830 if prov_playlist_id == RECOMMENDED_TRACKS_PLAYLIST_ID:
831 return create_virtual_playlist(
832 self.provider, RECOMMENDED_TRACKS_PLAYLIST_ID, "Hot Tracks"
833 )
834 if prov_playlist_id == TOP_CHARTS_PLAYLIST_ID:
835 return create_virtual_playlist(self.provider, TOP_CHARTS_PLAYLIST_ID, "Top Charts")
836 if prov_playlist_id == USER_TOP_TRACKS_PLAYLIST_ID:
837 return create_virtual_playlist(
838 self.provider, USER_TOP_TRACKS_PLAYLIST_ID, "Your Top Tracks"
839 )
840 if prov_playlist_id == PERSONAL_SONGS_PLAYLIST_ID:
841 return create_virtual_playlist(self.provider, PERSONAL_SONGS_PLAYLIST_ID, "My Uploads")
842 if prov_playlist_id.startswith(FLOW_CONFIG_PREFIX):
843 config_id = prov_playlist_id.removeprefix(FLOW_CONFIG_PREFIX)
844 flow_config = await self.provider.gql_client.get_flow_config_tracks(
845 flow_config_id=config_id
846 )
847 name = f"Flow: {flow_config.title}" if flow_config else f"Flow: {config_id}"
848 cover = get_flow_config_image(flow_config) if flow_config else None
849 return create_virtual_playlist(self.provider, prov_playlist_id, name, image_url=cover)
850 if prov_playlist_id.startswith(SMART_TRACKLIST_PREFIX):
851 tracklist_id = prov_playlist_id.removeprefix(SMART_TRACKLIST_PREFIX)
852 tracklist = await self.provider.gql_client.get_smart_tracklist(
853 smart_tracklist_id=tracklist_id, first=1
854 )
855 name = tracklist.title if tracklist else f"Mix {tracklist_id}"
856 cover = (
857 tracklist.cover.urls[0]
858 if tracklist and tracklist.cover and tracklist.cover.urls
859 else None
860 )
861 return create_virtual_playlist(
862 self.provider,
863 prov_playlist_id,
864 name,
865 image_url=cover,
866 )
867 if prov_playlist_id.startswith(SHAKER_CURATED_PREFIX):
868 group_id = prov_playlist_id.removeprefix(SHAKER_CURATED_PREFIX)
869 group = await self.provider.gql_client.get_music_together_group(
870 group_id=group_id,
871 mood=MusicTogetherSuggestedTracklistMoodInput.NONE,
872 tracks_first=1,
873 )
874 name = f"{group.name} - Playlist" if group else f"Shaker {group_id}"
875 cover_url: str | None = None
876 if (
877 group
878 and group.curated_tracklist
879 and group.curated_tracklist.picture
880 and group.curated_tracklist.picture.urls
881 ):
882 cover_url = group.curated_tracklist.picture.urls[0]
883 return create_virtual_playlist(
884 self.provider,
885 prov_playlist_id,
886 name,
887 image_url=cover_url,
888 )
889 if prov_playlist_id.startswith(SHAKER_PREFIX):
890 group_id = prov_playlist_id.removeprefix(SHAKER_PREFIX)
891 group = await self.provider.gql_client.get_music_together_group(
892 group_id=group_id,
893 mood=MusicTogetherSuggestedTracklistMoodInput.NONE,
894 tracks_first=1,
895 )
896 name = f"{group.name} - Mix" if group else f"Shaker {group_id}"
897 return create_virtual_playlist(
898 self.provider, prov_playlist_id, name, image_url=SHAKER_MIX_COVER
899 )
900 return None
901
902 # -- Virtual playlist track fetchers --
903
904 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
905 """Get playlist tracks, routing virtual playlist IDs to their fetchers."""
906 if page > 0:
907 return []
908 if prov_playlist_id == FLOW_PLAYLIST_ID:
909 return await self._get_flow_config_tracks(DEFAULT_FLOW_CONFIG_ID)
910 if prov_playlist_id == RECOMMENDED_TRACKS_PLAYLIST_ID:
911 return await self._get_recommended_tracks()
912 if prov_playlist_id == TOP_CHARTS_PLAYLIST_ID:
913 return await self._get_chart_tracks()
914 if prov_playlist_id == USER_TOP_TRACKS_PLAYLIST_ID:
915 return await self._get_user_chart_tracks()
916 if prov_playlist_id == PERSONAL_SONGS_PLAYLIST_ID:
917 return await self._get_personal_songs()
918 if prov_playlist_id.startswith(FLOW_CONFIG_PREFIX):
919 return await self._get_flow_config_tracks(
920 prov_playlist_id.removeprefix(FLOW_CONFIG_PREFIX)
921 )
922 if prov_playlist_id.startswith(SMART_TRACKLIST_PREFIX):
923 tracklist_id = prov_playlist_id.removeprefix(SMART_TRACKLIST_PREFIX)
924 return await self._get_smart_tracklist_tracks(tracklist_id)
925 if prov_playlist_id.startswith(SHAKER_CURATED_PREFIX):
926 group_id = prov_playlist_id.removeprefix(SHAKER_CURATED_PREFIX)
927 return await self._get_shaker_curated_tracks(group_id)
928 if prov_playlist_id.startswith(SHAKER_PREFIX):
929 shaker_id = prov_playlist_id.removeprefix(SHAKER_PREFIX)
930 return await self._get_shaker_tracks(shaker_id)
931 return await self._get_regular_playlist_tracks(prov_playlist_id)
932
933 @use_cache(3600)
934 async def _get_smart_tracklist_playlists(self) -> list[Playlist]:
935 """Get SmartTracklist items from Made For Me as virtual playlists."""
936 made_for_me = await self.provider.gql_client.get_made_for_me(first=20)
937 if not made_for_me or not made_for_me.made_for_me:
938 return []
939 playlists: list[Playlist] = []
940 for edge in made_for_me.made_for_me.edges:
941 if edge.node is None:
942 continue
943 if isinstance(edge.node, GetMadeForMeMeMadeForMeEdgesNodeSmartTracklist):
944 cover = (
945 edge.node.cover.urls[0] if edge.node.cover and edge.node.cover.urls else None
946 )
947 playlists.append(
948 create_virtual_playlist(
949 self.provider,
950 f"{SMART_TRACKLIST_PREFIX}{edge.node.id}",
951 edge.node.title,
952 image_url=cover,
953 )
954 )
955 return playlists
956
957 @use_cache(3600)
958 async def _get_recommended_tracks(self) -> list[Track]:
959 """Get cached recommended tracks (hot tracks)."""
960 recs = await self.provider.gql_client.get_recommendations(
961 playlists_first=0,
962 artist_playlists_first=0,
963 new_releases_first=0,
964 artists_first=0,
965 hot_tracks_limit=50,
966 )
967 if recs is None or recs.recommendations.hot_tracks is None:
968 return []
969 return [parse_track(self.provider, ht) for ht in recs.recommendations.hot_tracks]
970
971 @use_cache(3600)
972 async def _get_chart_tracks(self) -> list[Track]:
973 """Get cached chart tracks."""
974 charts = await self.provider.gql_client.get_charts(
975 country_code=self.provider.gw_client.user_country,
976 tracks_first=100,
977 )
978 if charts is None or charts.country is None or charts.country.tracks is None:
979 return []
980 return [
981 parse_track(self.provider, edge.node)
982 for edge in charts.country.tracks.edges
983 if edge.node is not None
984 ]
985
986 async def _get_flow_config_tracks(self, config_id: str) -> list[Track]:
987 """
988 Get fresh batches of tracks for a Flow config, merged deduplicated.
989
990 :param config_id: The Flow config identifier
991 (e.g. "default", "happy", "chill", "genre-rock").
992 """
993 seen: set[str] = set()
994 tracks: list[Track] = []
995 for _ in range(FLOW_BATCH_COUNT):
996 result = await self.provider.gql_client.get_flow_config_tracks(flow_config_id=config_id)
997 if result is None or not result.tracks:
998 break
999 for ft in result.tracks:
1000 if ft.track is not None and ft.track.id not in seen:
1001 seen.add(ft.track.id)
1002 tracks.append(parse_track(self.provider, ft.track))
1003 return filter_tracks(tracks)
1004
1005 @use_cache(3600)
1006 async def _get_smart_tracklist_tracks(self, tracklist_id: str) -> list[Track]:
1007 """
1008 Get tracks for a SmartTracklist.
1009
1010 :param tracklist_id: The SmartTracklist identifier.
1011 """
1012 all_tracks: list[Track] = []
1013 cursor: str | None = None
1014 while True:
1015 result = await self.provider.gql_client.get_smart_tracklist(
1016 smart_tracklist_id=tracklist_id, first=50, after=cursor
1017 )
1018 if result is None:
1019 break
1020 all_tracks.extend(
1021 parse_track(self.provider, edge.node)
1022 for edge in result.tracks.edges
1023 if edge.node is not None
1024 )
1025 if not result.tracks.page_info.has_next_page:
1026 break
1027 cursor = result.tracks.page_info.end_cursor
1028 return all_tracks
1029
1030 async def _get_shaker_tracks(self, group_id: str) -> list[Track]:
1031 """
1032 Get suggested tracks for a Shaker (Music Together) group.
1033
1034 :param group_id: The Music Together group identifier.
1035 """
1036 # Refresh the suggested tracklist to get a fresh set of tracks
1037 await self.provider.gql_client.music_together_refresh_suggested_tracklist(
1038 group_id=group_id,
1039 mood=MusicTogetherRefreshSuggestedTracklistMoodInput.NONE,
1040 )
1041 group = await self.provider.gql_client.get_music_together_group(
1042 group_id=group_id,
1043 mood=MusicTogetherSuggestedTracklistMoodInput.NONE,
1044 tracks_first=50,
1045 )
1046 if group is None or group.suggested_tracklist is None:
1047 return []
1048 tracklist = group.suggested_tracklist.tracklist
1049 if tracklist is None:
1050 return []
1051 return filter_tracks(
1052 [parse_track(self.provider, edge.node) for edge in tracklist.tracks.edges if edge.node]
1053 )
1054
1055 async def _get_shaker_curated_tracks(self, group_id: str) -> list[Track]:
1056 """
1057 Get curated playlist tracks for a Shaker (Music Together) group.
1058
1059 :param group_id: The Music Together group identifier.
1060 """
1061 all_tracks: list[Track] = []
1062 cursor: str | None = None
1063 while True:
1064 group = await self.provider.gql_client.get_music_together_group(
1065 group_id=group_id,
1066 mood=MusicTogetherSuggestedTracklistMoodInput.NONE,
1067 tracks_first=50,
1068 tracks_after=cursor,
1069 )
1070 if group is None or group.curated_tracklist is None:
1071 break
1072 tracks_conn = group.curated_tracklist.tracks
1073 all_tracks.extend(
1074 parse_track(self.provider, edge.node) for edge in tracks_conn.edges if edge.node
1075 )
1076 if not tracks_conn.page_info.has_next_page:
1077 break
1078 cursor = tracks_conn.page_info.end_cursor
1079 return all_tracks
1080
1081 @use_cache(3600)
1082 async def _get_user_chart_tracks(self) -> list[Track]:
1083 """Get the user's most listened tracks."""
1084 result = await self.provider.gql_client.get_user_charts(tracks_first=50)
1085 if not result or not result.charts.tracks:
1086 return []
1087 return [
1088 parse_track(self.provider, edge.node)
1089 for edge in result.charts.tracks.edges
1090 if edge.node is not None
1091 ]
1092
1093 async def _get_personal_songs(self) -> list[Track]:
1094 """Get user-uploaded personal songs via the GW API."""
1095 songs = await self.provider.media_manager._get_personal_songs()
1096 return [
1097 parse_gw_track(self.provider, song, position=idx) for idx, song in enumerate(songs, 1)
1098 ]
1099
1100 async def invalidate_playlist_cache(self, prov_playlist_id: str) -> None:
1101 """Invalidate the cached playlist tracks after a mutation."""
1102 cache_key = f"_get_regular_playlist_tracks.{prov_playlist_id}"
1103 await self.mass.cache.delete(key=cache_key, provider=self.instance_id)
1104
1105 @use_cache(3600 * 3)
1106 async def _get_regular_playlist_tracks(self, prov_playlist_id: str) -> list[Track]:
1107 """Get tracks for regular Deezer playlists (cached)."""
1108 result = await self.provider.gql_client.get_playlist(playlist_id=prov_playlist_id)
1109 if result is None:
1110 return []
1111 all_edges = list(result.tracks.edges)
1112 while result.tracks.page_info.has_next_page:
1113 result = await self.provider.gql_client.get_playlist(
1114 playlist_id=prov_playlist_id,
1115 tracks_after=result.tracks.page_info.end_cursor,
1116 )
1117 if result is None:
1118 break
1119 all_edges.extend(result.tracks.edges)
1120 return [
1121 parse_track(self.provider, edge.node, position=idx)
1122 for idx, edge in enumerate(all_edges, 1)
1123 if edge.node is not None
1124 ]
1125
1126 @use_cache(3600)
1127 async def _get_flow_cover(self) -> str | None:
1128 """Get the cover URL for the user's default Flow."""
1129 result = await self.provider.gql_client.get_flow_config_tracks(
1130 flow_config_id=DEFAULT_FLOW_CONFIG_ID
1131 )
1132 return get_flow_config_image(result) if result else None
1133
1134 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
1135 """Fetch the recommendation folders fed by the shared gql recommendations payload."""
1136 recs = await self.provider.gql_client.get_recommendations(
1137 playlists_first=50,
1138 artist_playlists_first=50,
1139 new_releases_first=10,
1140 artists_first=0,
1141 hot_tracks_limit=50,
1142 )
1143 result: list[RecommendationFolder] = []
1144 self._add_recommended_playlists(result, recs)
1145 self._add_recommended_artist_playlists(result, recs)
1146 self._add_recommended_tracks(result, recs)
1147 self._add_new_releases(result, recs)
1148 return result
1149
1150 @use_cache(3600, base_class=RecommendationFolder)
1151 async def _get_flow_folders(self) -> list[RecommendationFolder]:
1152 """Get the mood and genre flow rows, built from one shared backend fetch (cached)."""
1153 result: list[RecommendationFolder] = []
1154 await self._add_flow_configs(result)
1155 return result
1156
1157 async def _add_made_for_you(self, result: list[RecommendationFolder]) -> None:
1158 """Add Made For You section to recommendations."""
1159 made_for_me_items: list[Playlist] = [
1160 create_virtual_playlist(
1161 self.provider, FLOW_PLAYLIST_ID, "Flow", image_url=await self._get_flow_cover()
1162 )
1163 ]
1164 made_for_me_items.extend(await self._get_smart_tracklist_playlists())
1165 if made_for_me_items:
1166 result.append(
1167 RecommendationFolder(
1168 item_id="made_for_you",
1169 provider=self.instance_id,
1170 name=BROWSE_MADE_FOR_YOU,
1171 translation_key="made_for_you",
1172 items=UniqueList(made_for_me_items),
1173 )
1174 )
1175