from typing import Any from pydantic import SecretStr from resonance_engine.adapters.apple_music import AppleMusicClient from resonance_engine.dsp.backends.base import DSPBackend from resonance_engine.dsp.enums import DSPId from resonance_engine.dsp.exceptions import DSPResourceUnsupportedError from resonance_engine.dsp.types import TokenResult from resonance_engine.utils.functional import exclude # Artwork URLs and self-links stripped from every object before the sink. _EXCLUDE_KEYS = frozenset( { "artwork", "href", "playParams", "previews", "discNumber", "trackNumber", "hasLyrics", "isAppleDigitalMaster", } ) class AppleMusicDSPBackend(DSPBackend): dsp_id = DSPId.apple def __init__(self, client: AppleMusicClient) -> None: self._client = client def close(self) -> None: self._client.close() def refresh_token(self, refresh_token: SecretStr) -> TokenResult: return TokenResult(access_token=refresh_token) def get_profile(self, access_token: SecretStr) -> dict[str, Any]: # noqa: ARG002 raise DSPResourceUnsupportedError("Apple Music has no user profile") def get_top_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: # noqa: ARG002 raise DSPResourceUnsupportedError("Apple Music has no top artists endpoint") def get_top_tracks(self, access_token: SecretStr) -> list[dict[str, Any]]: # noqa: ARG002 raise DSPResourceUnsupportedError("Apple Music has no top tracks endpoint") def get_recently_played( self, access_token: SecretStr, *, after: int | None = None, # noqa: ARG002 — Apple paginates by offset ) -> list[dict[str, Any]]: return exclude( self._client.get_recently_played_tracks(access_token, max_items=30), keys=_EXCLUDE_KEYS, recursive=True, ) def get_playlists(self, access_token: SecretStr) -> list[dict[str, Any]]: return exclude( self._client.get_library_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_library_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_library_songs(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, ) def get_followed_artists(self, access_token: SecretStr) -> list[dict[str, Any]]: # Apple has no "followed" concept — library artists are the nearest equivalent. return exclude( self._client.get_library_artists(access_token, max_items=100), keys=_EXCLUDE_KEYS, recursive=True, )