from typing import Any from pydantic import SecretStr from resonance_engine.adapters.deezer import DeezerClient from resonance_engine.dsp.backends.base import DSPBackend from resonance_engine.dsp.enums import DSPId from resonance_engine.dsp.types import TokenResult from resonance_engine.utils.functional import exclude # Image/link noise and PII stripped from every Deezer object before the sink. _EXCLUDE_KEYS = frozenset( { "birthday", "picture", "picture_small", "picture_medium", "picture_big", "picture_xl", "picture_type", "cover", "cover_small", "cover_medium", "cover_big", "cover_xl", "md5_image", "tracklist", "link", "share", "checksum", } ) class DeezerDSPBackend(DSPBackend): dsp_id = DSPId.deezer def __init__(self, client: DeezerClient) -> None: self._client = client def close(self) -> None: self._client.close() def refresh_token(self, refresh_token: SecretStr) -> TokenResult: response = self._client.refresh_token(refresh_token) return TokenResult(access_token=response["access_token"]) def get_profile(self, access_token: SecretStr) -> dict[str, Any]: return exclude( self._client.get_current_user_profile(access_token), keys=_EXCLUDE_KEYS, recursive=True, ) def get_top_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_top_artists(access_token, max_items=50), keys=_EXCLUDE_KEYS, recursive=True, ) def get_top_tracks(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_top_tracks(access_token, max_items=50), keys=_EXCLUDE_KEYS, recursive=True, ) def get_recently_played( self, access_token: SecretStr, *, after: int | None = None, # noqa: ARG002 — Deezer history has no after cursor ) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_history(access_token, max_items=50), keys=_EXCLUDE_KEYS, recursive=True, ) def get_playlists(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_playlists(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, ) def get_saved_albums(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_albums(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, ) def get_saved_tracks(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_tracks(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, ) def get_followed_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_current_user_artists(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, )