/
/
/
1"""
2Deezer provider - provider facade.
3
4Thin facade that delegates to specialized manager classes:
5- DeezerMediaManager: library, search, item getters, mutations
6- DeezerBrowseManager: browse tree, recommendations, virtual playlists
7- DeezerStreamingManager: streaming, decryption, playback callbacks
8"""
9
10from __future__ import annotations
11
12import logging
13from collections.abc import AsyncGenerator, Sequence
14from datetime import datetime
15from typing import TYPE_CHECKING
16
17from deezer_python_gql import DeezerGQLClient, GraphQLClientError
18from music_assistant_models.enums import MediaType, ProviderFeature
19from music_assistant_models.errors import LoginFailed
20
21from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
22from music_assistant.models.music_provider import MusicProvider
23from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
24
25from .browse import DeezerBrowseManager
26from .gw_client import DeezerGWError, GWClient
27from .media import DeezerMediaManager
28from .streaming import DeezerStreamingManager
29
30if TYPE_CHECKING:
31 from music_assistant_models.config_entries import ConfigEntry
32 from music_assistant_models.media_items import (
33 Album,
34 Artist,
35 Audiobook,
36 BrowseFolder,
37 ItemMapping,
38 MediaItemType,
39 Playlist,
40 Podcast,
41 PodcastEpisode,
42 Radio,
43 RecommendationFolder,
44 SearchResults,
45 Track,
46 UniqueList,
47 )
48 from music_assistant_models.streamdetails import StreamDetails
49
50SUPPORTED_FEATURES = {
51 ProviderFeature.LIBRARY_ARTISTS,
52 ProviderFeature.LIBRARY_ALBUMS,
53 ProviderFeature.LIBRARY_TRACKS,
54 ProviderFeature.LIBRARY_PLAYLISTS,
55 ProviderFeature.LIBRARY_ALBUMS_EDIT,
56 ProviderFeature.LIBRARY_TRACKS_EDIT,
57 ProviderFeature.LIBRARY_ARTISTS_EDIT,
58 ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
59 ProviderFeature.ALBUM_METADATA,
60 ProviderFeature.TRACK_METADATA,
61 ProviderFeature.ARTIST_METADATA,
62 ProviderFeature.ARTIST_ALBUMS,
63 ProviderFeature.ARTIST_TOPALBUMS,
64 ProviderFeature.ARTIST_TOPTRACKS,
65 ProviderFeature.BROWSE,
66 ProviderFeature.SEARCH,
67 ProviderFeature.PLAYLIST_TRACKS_EDIT,
68 ProviderFeature.PLAYLIST_CREATE,
69 ProviderFeature.RECOMMENDATIONS,
70 ProviderFeature.SIMILAR_TRACKS,
71 ProviderFeature.SIMILAR_ARTISTS,
72 ProviderFeature.LYRICS,
73 ProviderFeature.LIBRARY_PODCASTS,
74 ProviderFeature.LIBRARY_PODCASTS_EDIT,
75 ProviderFeature.LIBRARY_AUDIOBOOKS,
76 ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT,
77}
78
79CONF_ARL_TOKEN = "arl_token"
80
81
82class DeezerProvider(RecommendationPayloadMixin, MusicProvider):
83 """
84 Deezer provider support.
85
86 Delegates to specialized manager classes for clean separation of concerns.
87 """
88
89 gql_client: DeezerGQLClient
90 gw_client: GWClient
91 media_manager: DeezerMediaManager
92 browse_manager: DeezerBrowseManager
93 streaming_manager: DeezerStreamingManager
94 user_id: str
95
96 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
97 """Return Config entries to configure this provider."""
98 return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
99
100 async def handle_async_init(self) -> None:
101 """Handle async init of the Deezer provider."""
102 arl_token = str(self.get_setup_value(CONF_ARL_TOKEN))
103
104 try:
105 self.gql_client = DeezerGQLClient(arl=arl_token, session=self.mass.http_session)
106 logging.getLogger("deezer_python_gql").setLevel(self.logger.level + 10)
107 me = await self.gql_client.get_me()
108 if not me:
109 msg = "Authentication returned no user data"
110 raise GraphQLClientError(msg)
111 self.user_id = me.id
112 self.gw_client = GWClient(self.mass.http_session, arl_token)
113 await self.gw_client.setup()
114 except (GraphQLClientError, DeezerGWError) as err:
115 raise LoginFailed("Deezer authentication failed. Please check your ARL token.") from err
116
117 self.media_manager = DeezerMediaManager(self)
118 self.browse_manager = DeezerBrowseManager(self)
119 self.streaming_manager = DeezerStreamingManager(self)
120
121 async def unload(self, is_removed: bool = False) -> None:
122 """Handle unload/close of the provider."""
123 await super().unload(is_removed)
124
125 # -- Library retrieval --
126
127 async def get_library_artists(self) -> AsyncGenerator[Artist]:
128 """Retrieve all library artists from Deezer."""
129 async for item in self.media_manager.get_library_artists():
130 yield item
131
132 async def get_library_albums(self) -> AsyncGenerator[Album]:
133 """Retrieve all library albums from Deezer."""
134 async for item in self.media_manager.get_library_albums():
135 yield item
136
137 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
138 """Retrieve all library playlists from Deezer."""
139 async for item in self.media_manager.get_library_playlists():
140 yield item
141
142 async def get_library_tracks(self) -> AsyncGenerator[Track]:
143 """Retrieve all library tracks from Deezer."""
144 async for item in self.media_manager.get_library_tracks():
145 yield item
146
147 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
148 """Retrieve library/subscribed podcasts from Deezer."""
149 async for item in self.media_manager.get_library_podcasts():
150 yield item
151
152 async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
153 """Retrieve library/subscribed audiobooks from Deezer."""
154 async for item in self.media_manager.get_library_audiobooks():
155 yield item
156
157 # -- Search --
158
159 async def search(
160 self, search_query: str, media_types: list[MediaType], limit: int = 5
161 ) -> SearchResults:
162 """Perform search on music provider."""
163 return await self.media_manager.search(search_query, media_types, limit)
164
165 # -- Item getters --
166
167 async def get_artist(self, prov_artist_id: str) -> Artist:
168 """Get full artist details by id."""
169 return await self.media_manager.get_artist(prov_artist_id)
170
171 async def get_album(self, prov_album_id: str) -> Album:
172 """Get full album details by id."""
173 return await self.media_manager.get_album(prov_album_id)
174
175 async def get_track(self, prov_track_id: str) -> Track:
176 """Get full track details by id."""
177 return await self.media_manager.get_track(prov_track_id)
178
179 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
180 """Get full playlist details by id."""
181 return await self.media_manager.get_playlist(prov_playlist_id)
182
183 async def get_radio(self, prov_radio_id: str) -> Radio:
184 """Get full radio/livestream details by id."""
185 return await self.media_manager.get_radio(prov_radio_id)
186
187 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
188 """Get full podcast details by id."""
189 return await self.media_manager.get_podcast(prov_podcast_id)
190
191 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
192 """Get (full) podcast episode details by id."""
193 return await self.media_manager.get_podcast_episode(prov_episode_id)
194
195 async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
196 """Get full audiobook details by id."""
197 return await self.media_manager.get_audiobook(prov_audiobook_id)
198
199 # -- Content getters --
200
201 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
202 """Get all tracks in an album."""
203 return await self.media_manager.get_album_tracks(prov_album_id)
204
205 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
206 """Get playlist tracks."""
207 return await self.browse_manager.get_playlist_tracks(prov_playlist_id, page)
208
209 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
210 """Get albums by an artist."""
211 return await self.media_manager.get_artist_albums(prov_artist_id)
212
213 async def get_artist_topalbums(self, prov_artist_id: str) -> list[Album]:
214 """Get top albums of an artist."""
215 return await self.media_manager.get_artist_topalbums(prov_artist_id)
216
217 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
218 """Get top tracks of an artist."""
219 return await self.media_manager.get_artist_toptracks(prov_artist_id)
220
221 async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
222 """Get all episodes for a given podcast."""
223 async for ep in self.media_manager.get_podcast_episodes(prov_podcast_id):
224 yield ep
225
226 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
227 """Retrieve a dynamic list of tracks based on the provided item."""
228 return await self.media_manager.get_similar_tracks(prov_track_id, limit)
229
230 async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
231 """Retrieve a list of artists similar to the provided artist."""
232 return await self.media_manager.get_similar_artists(prov_artist_id, limit)
233
234 # -- Library mutations --
235
236 async def library_add(self, item: MediaItemType) -> bool:
237 """Add an item to the provider's library/favorites."""
238 return await self.media_manager.library_add(item)
239
240 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
241 """Remove an item from the provider's library/favorites."""
242 return await self.media_manager.library_remove(prov_item_id, media_type)
243
244 # -- Playlist CRUD --
245
246 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
247 """Add track(s) to playlist."""
248 await self.media_manager.add_playlist_tracks(prov_playlist_id, prov_track_ids)
249
250 async def remove_playlist_tracks(
251 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
252 ) -> None:
253 """Remove track(s) from playlist."""
254 await self.media_manager.remove_playlist_tracks(prov_playlist_id, positions_to_remove)
255
256 async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
257 """Create a new playlist on provider with given name."""
258 return await self.media_manager.create_playlist(name, media_types)
259
260 # -- Browse & Recommendations --
261
262 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
263 """Browse Deezer content."""
264 return await self.browse_manager.browse(path, super().browse)
265
266 async def get_recommendations(self) -> list[RecommendationFolder]:
267 """Get Deezer's available recommendation rows, without items."""
268 return await self.browse_manager.get_recommendations()
269
270 async def get_recommendation_items(
271 self, item_id: str
272 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
273 """
274 Get the items for a single recommendation row.
275
276 :param item_id: The item_id of the row, as returned by get_recommendations.
277 """
278 return await self.browse_manager.get_recommendation_items(item_id)
279
280 # -- Streaming --
281
282 async def get_resume_position(
283 self, item_id: str, media_type: MediaType
284 ) -> tuple[bool, int, datetime | None]:
285 """Get the resume position for a podcast episode."""
286 return await self.streaming_manager.get_resume_position(item_id, media_type)
287
288 async def on_played(
289 self,
290 media_type: MediaType,
291 prov_item_id: str,
292 fully_played: bool,
293 position: int,
294 media_item: MediaItemType,
295 is_playing: bool = False,
296 ) -> None:
297 """Handle callback when a podcast episode has been played or is playing."""
298 await self.streaming_manager.on_played(
299 media_type, prov_item_id, fully_played, position, media_item, is_playing
300 )
301
302 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
303 """Return the content details for the given track when it will be streamed."""
304 return await self.streaming_manager.get_stream_details(item_id, media_type)
305
306 async def get_audio_stream(
307 self, streamdetails: StreamDetails, seek_position: int = 0
308 ) -> AsyncGenerator[bytes]:
309 """Return the audio stream for the provider item."""
310 async for chunk in self.streaming_manager.get_audio_stream(streamdetails, seek_position):
311 yield chunk
312
313 async def on_streamed(self, streamdetails: StreamDetails) -> None:
314 """Handle callback when an item completed streaming."""
315 await self.streaming_manager.on_streamed(streamdetails)
316
317 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
318 """Fetch the recommendation folders fed by the shared gql recommendations payload."""
319 return await self.browse_manager._fetch_recommendation_payload()
320