From: Julian Daberkow Date: Tue, 17 Feb 2026 09:06:19 +0000 (+0100) Subject: feat(deezer): Genre and mood flows (#3171) X-Git-Url: https://git.kitaultman.com/?a=commitdiff_plain;h=9eda5abf3b35fd55e70d2f59bed3162b86164469;p=music-assistant-server.git feat(deezer): Genre and mood flows (#3171) * Add basic mood and genre flow support for Deezer * Fetch Deezer flows and their images dynamically * Address PR review comments - Narrow exception handling from `Exception` to `DeezerErrorResponse` for recommended albums/artists endpoints - Parse GW track data directly via `_parse_gw_track()` instead of making individual REST API calls per track - Add early return check in `get_user_radio` when data key is missing - Remove unnecessary `:param: None` docstring in `get_home_flows` - Add defensive `.get()` check for "sections" key in `get_home_flows` to prevent KeyError * Shrink function documentation of _parse_gw_track * Revert to previous approach of fetching gw tracks seperately --------- Co-authored-by: jdaberkow <13017916+jdaberkow@users.noreply.github.com> --- diff --git a/music_assistant/providers/deezer/__init__.py b/music_assistant/providers/deezer/__init__.py index a7a50917..396c7b1d 100644 --- a/music_assistant/providers/deezer/__init__.py +++ b/music_assistant/providers/deezer/__init__.py @@ -100,6 +100,7 @@ FLOW_PLAYLIST_ID = "flow" RECOMMENDED_TRACKS_PLAYLIST_ID = "recommended_tracks" TOP_CHARTS_PLAYLIST_ID = "top_charts" RADIO_PLAYLIST_PREFIX = "radio_" +MOOD_FLOW_PREFIX = "mood_flow_" # Curated Deezer radio station IDs CURATED_RADIO_IDS = [ @@ -237,6 +238,34 @@ class DeezerProvider(MusicProvider): chart = await self.client.get_chart() return list(chart.tracks[:100]) if chart.tracks else [] + @use_cache(3600) # Cache for 1 hour + async def _get_mood_flow_tracks(self, config_id: str) -> list[dict[str, Any]]: + """Get cached mood/genre Flow tracks from the GW API. + + :param config_id: The Flow config identifier (e.g. "happy", "chill", "genre-rock"). + """ + return await self.gw_client.get_user_radio(config_id) + + @use_cache(3600 * 24) # Cache for 24 hours + async def _get_available_flows(self) -> list[tuple[str, str, str | None]]: + """Discover available mood/genre Flow variants from the Deezer home page. + + Genre flows have config_ids starting with 'genre-'. + Returns a list of (config_id, display_name, cover_url) tuples. + """ + items = await self.gw_client.get_home_flows() + flows: list[tuple[str, str, str | None]] = [] + for item in items: + config_id = item["data"]["id"] + if config_id == "default": + continue + title = f"Flow: {item['title']}" + cover_url = None + if pictures := item.get("pictures"): + cover_url = f"https://e-cdns-images.dzcdn.net/images/misc/{pictures[0]['md5']}/264x264-000000-80-0-0.jpg" + flows.append((config_id, title, cover_url)) + return flows + @use_cache(3600 * 24 * 7) # Cache for 7 days async def search( self, search_query: str, media_types: list[MediaType], limit: int = 5 @@ -364,6 +393,12 @@ class DeezerProvider(MusicProvider): except Exception as err: self.logger.warning("Failed getting radio %s: %s", radio_id, err) raise MediaNotFoundError(f"Radio {prov_playlist_id} not found on Deezer") from err + if prov_playlist_id.startswith(MOOD_FLOW_PREFIX): + config_id = prov_playlist_id.removeprefix(MOOD_FLOW_PREFIX) + all_flows = await self._get_available_flows() + flow_info = {cid: (name, cover) for cid, name, cover in all_flows} + name, cover_url = flow_info.get(config_id, (f"Flow: {config_id}", None)) + return self._create_virtual_playlist(prov_playlist_id, name, image_url=cover_url) try: return self.parse_playlist( playlist=await self.client.get_playlist(playlist_id=int(prov_playlist_id)), @@ -423,6 +458,11 @@ class DeezerProvider(MusicProvider): self.logger.debug("Failed to get radio tracks %s: %s", radio_id, err) return [] + if prov_playlist_id.startswith(MOOD_FLOW_PREFIX): + config_id = prov_playlist_id.removeprefix(MOOD_FLOW_PREFIX) + gw_tracks = await self._get_mood_flow_tracks(config_id) + return [await self.get_track(track["SNG_ID"]) for track in gw_tracks] + # Regular Deezer playlists (cached separately) return await self._get_regular_playlist_tracks(prov_playlist_id) @@ -569,32 +609,64 @@ class DeezerProvider(MusicProvider): ) # Recommended albums - recommended_albums = list(await self.client.get_user_recommended_albums()) - if recommended_albums: - result.append( - RecommendationFolder( - item_id="recommended_albums", - provider=self.instance_id, - name="Recommended albums", - items=UniqueList( - [self.parse_album(album=album) for album in recommended_albums] - ), + try: + recommended_albums = list(await self.client.get_user_recommended_albums()) + if recommended_albums: + result.append( + RecommendationFolder( + item_id="recommended_albums", + provider=self.instance_id, + name="Recommended albums", + items=UniqueList( + [self.parse_album(album=album) for album in recommended_albums] + ), + ) ) - ) + except deezer_exceptions.DeezerErrorResponse as err: + self.logger.debug("Failed to get recommended albums: %s", err) # Recommended artists - recommended_artists = list(await self.client.get_user_recommended_artists()) - if recommended_artists: - result.append( - RecommendationFolder( - item_id="recommended_artists", - provider=self.instance_id, - name="Recommended artists", - items=UniqueList( - [self.parse_artist(artist=artist) for artist in recommended_artists] - ), + try: + recommended_artists = list(await self.client.get_user_recommended_artists()) + if recommended_artists: + result.append( + RecommendationFolder( + item_id="recommended_artists", + provider=self.instance_id, + name="Recommended artists", + items=UniqueList( + [self.parse_artist(artist=artist) for artist in recommended_artists] + ), + ) + ) + except deezer_exceptions.DeezerErrorResponse as err: + self.logger.debug("Failed to get recommended artists: %s", err) + + # Deezer Mood and Genre Flows - personalized playlists (dynamically discovered) + all_flows = await self._get_available_flows() + mood_flows = [(c, n, img) for c, n, img in all_flows if not c.startswith("genre-")] + genre_flows = [(c, n, img) for c, n, img in all_flows if c.startswith("genre-")] + for folder_id, folder_name, flows in [ + ("mood_flows", "Deezer Mood Flows", mood_flows), + ("genre_flows", "Deezer Genre Flows", genre_flows), + ]: + flow_playlists = [ + self._create_virtual_playlist( + item_id=f"{MOOD_FLOW_PREFIX}{config_id}", + name=display_name, + image_url=cover_url, + ) + for config_id, display_name, cover_url in flows + ] + if flow_playlists: + result.append( + RecommendationFolder( + item_id=folder_id, + provider=self.instance_id, + name=folder_name, + items=UniqueList(flow_playlists), + ) ) - ) # Deezer Radios - curated selection (as virtual playlists in one folder) radio_playlists: list[Playlist] = [] diff --git a/music_assistant/providers/deezer/gw_client.py b/music_assistant/providers/deezer/gw_client.py index f4f109d0..82306554 100644 --- a/music_assistant/providers/deezer/gw_client.py +++ b/music_assistant/providers/deezer/gw_client.py @@ -5,6 +5,7 @@ cookie based on the api_token. """ import datetime +import json from collections.abc import Mapping from http.cookies import BaseCookie, Morsel from typing import Any, cast @@ -36,6 +37,7 @@ class GWClient: _gw_csrf_token: str | None _license: str | None _license_expiration_timestamp: int + _user_id: int session: ClientSession formats: list[dict[str, str]] = [ {"cipher": "BF_CBC_STRIPE", "format": "MP3_128"}, @@ -67,6 +69,7 @@ class GWClient: raise DeezerGWError(msg) self._gw_csrf_token = user_data["results"]["checkForm"] + self._user_id = int(user_data["results"]["USER"]["USER_ID"]) self._license = user_data["results"]["USER"]["OPTIONS"]["license_token"] self._license_expiration_timestamp = user_data["results"]["USER"]["OPTIONS"][ "expiration_timestamp" @@ -127,6 +130,38 @@ class GWClient: raise DeezerGWError(msg, result_json["error"]) return cast("dict[str, Any]", result_json) + async def get_user_radio(self, config_id: str) -> list[dict[str, Any]]: + """Get personalized Flow tracks for a specific mood or genre. + + :param config_id: The Flow config identifier (e.g. "happy", "chill", "genre-rock"). + """ + result = await self._gw_api_call( + "radio.getUserRadio", + args={"config_id": config_id, "user_id": self._user_id}, + ) + if "data" not in result["results"]: + return [] + return cast("list[dict[str, Any]]", result["results"]["data"]) + + async def get_home_flows(self) -> list[dict[str, Any]]: + """Discover available Flow variants from the Deezer home page.""" + gateway_input = json.dumps( + { + "PAGE": "home", + "VERSION": "2.5", + "SUPPORT": {"filterable-grid": ["flow"]}, + } + ) + result = await self._gw_api_call( + "page.get", + params={"gateway_input": gateway_input}, + ) + sections = result["results"].get("sections", []) + for section in sections: + if section.get("layout") == "filterable-grid": + return cast("list[dict[str, Any]]", section["items"]) + return [] + async def get_song_data(self, track_id: str) -> dict[str, Any]: """Get data such as the track token for a given track.""" return await self._gw_api_call("song.getData", args={"SNG_ID": track_id})