import base64 from collections.abc import Callable from typing import Any, NotRequired, TypedDict import httpx from fansifter_common.httpclient import HTTPClient, HTTPClientError from pydantic import SecretStr, TypeAdapter # OAuth lives on Login with Amazon; the Music Web API on its own host. The Music # API is in closed beta — base path and some connection field names below are the # documented shapes and may need confirming against live beta responses. TOKEN_URL = "https://api.amazon.com/auth/o2/token" API_BASE = "https://api.music.amazon.dev/v1" TIMEOUT = 10.0 # --------------------------------------------------------------------------- # Response types # --------------------------------------------------------------------------- class AmazonMusicTokenResponse(TypedDict): access_token: SecretStr refresh_token: NotRequired[SecretStr | None] token_type: NotRequired[str | None] expires_in: NotRequired[int | None] class AmazonMusicPageInfo(TypedDict): hasNextPage: NotRequired[bool | None] token: NotRequired[str | None] class AmazonMusicEdge(TypedDict): node: NotRequired[dict[str, Any] | None] playedAt: NotRequired[str | None] class AmazonMusicConnection(TypedDict): pageInfo: NotRequired[AmazonMusicPageInfo | None] edges: NotRequired[list[AmazonMusicEdge] | None] _TOKEN_RESPONSE_ADAPTER = TypeAdapter(AmazonMusicTokenResponse) # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class AmazonMusicError(Exception): """Base exception for Amazon Music client errors.""" class AmazonMusicTokenError(AmazonMusicError): """Raised when an OAuth token refresh fails (transient).""" class AmazonMusicTokenRevokedError(AmazonMusicTokenError): """Raised when the refresh token is permanently invalid (invalid_grant).""" class AmazonMusicHTTPError(HTTPClientError): """HTTPClientError that appends Amazon Music's JSON error message to the text.""" @property def detail(self) -> str | None: if self.response is None: return None try: body = self.response.json() except ValueError, AttributeError: return None error = body.get("error") if isinstance(body, dict) else None if isinstance(error, dict): return error.get("message") if isinstance(error, str): return error return body.get("message") if isinstance(body, dict) else None def __str__(self) -> str: return f"{self.message}: {self.detail}" if self.detail else self.message # --------------------------------------------------------------------------- # Client # --------------------------------------------------------------------------- _REVOKED_TOKEN_ERRORS = {"invalid_grant", "unauthorized_client"} _SAFE_TOKEN_ERRORS = _REVOKED_TOKEN_ERRORS | { "invalid_client", "invalid_request", "unsupported_grant_type", } def _sanitize_token_error(exc: httpx.HTTPStatusError) -> tuple[str, bool]: """Returns (message, is_revoked).""" try: body = exc.response.json() error_type = body.get("error", "") if error_type in _SAFE_TOKEN_ERRORS: msg = body.get("error_description") or error_type return msg, error_type in _REVOKED_TOKEN_ERRORS except Exception: pass return f"HTTP {exc.response.status_code}", False class AmazonMusicClient(HTTPClient): exception_class = AmazonMusicHTTPError def __init__( self, client_id: str, client_secret: SecretStr, profile_id: str ) -> None: super().__init__(client_options={"timeout": TIMEOUT}) self._client_id = client_id self._client_secret = client_secret self._profile_id = profile_id self.start() def _basic_auth(self) -> str: raw = f"{self._client_id}:{self._client_secret.get_secret_value()}" return base64.b64encode(raw.encode()).decode() def refresh_token(self, refresh_token: SecretStr) -> AmazonMusicTokenResponse: try: response = self.client.post( TOKEN_URL, data={ "grant_type": "refresh_token", "refresh_token": refresh_token.get_secret_value(), }, headers={ "Authorization": f"Basic {self._basic_auth()}", "Content-Type": "application/x-www-form-urlencoded", }, ) response.raise_for_status() except httpx.HTTPStatusError as exc: msg, is_revoked = _sanitize_token_error(exc) if is_revoked: raise AmazonMusicTokenRevokedError(msg) from exc raise AmazonMusicTokenError(msg) from exc except httpx.HTTPError as exc: raise AmazonMusicTokenError(str(exc)) from exc return _TOKEN_RESPONSE_ADAPTER.validate_json(response.content) def _auth_headers(self, access_token: SecretStr) -> dict[str, str]: # The single unwrap point — access_token stays a SecretStr everywhere else. return { "Authorization": f"Bearer {access_token.get_secret_value()}", "x-api-key": self._profile_id, } def _paginate_cursor( self, request_page: Callable[[str | None, int], AmazonMusicConnection], *, max_items: int, edge_to_item: Callable[[AmazonMusicEdge], dict[str, Any] | None] = ( lambda edge: edge.get("node") ), ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] cursor: str | None = None while len(items) < max_items: page = request_page(cursor, min(100, max_items - len(items))) for edge in page.get("edges") or []: item = edge_to_item(edge) if item is not None: items.append(item) page_info = page.get("pageInfo") or {} cursor = page_info.get("token") if not page_info.get("hasNextPage") or not cursor: break return items[:max_items] def get_current_user_profile(self, access_token: SecretStr) -> dict[str, Any]: raw = self.request( "GET", f"{API_BASE}/me", headers=self._auth_headers(access_token), type=dict[str, Any], ) return raw.get("data", {}).get("user", raw) def get_current_user_recently_played( self, access_token: SecretStr, *, max_items: int = 60, ) -> list[dict[str, Any]]: # Amazon paginates by opaque cursor only — no after-timestamp like Spotify. def request_page(cursor: str | None, limit: int) -> AmazonMusicConnection: params: dict[str, str | int] = {"limit": limit} if cursor is not None: params["cursor"] = cursor raw = self.request( "GET", f"{API_BASE}/player/recentlyPlayed", params=params, headers=self._auth_headers(access_token), type=dict[str, Any], ) user = raw.get("data", {}).get("user", {}) return user.get("recentTrackPlayback", {}) return self._paginate_cursor( request_page, max_items=max_items, edge_to_item=lambda edge: ( {**node, "playedAt": edge.get("playedAt")} if (node := edge.get("node")) is not None else None ), ) def get_current_user_playlists( self, access_token: SecretStr, *, max_items: int = 100 ) -> list[dict[str, Any]]: def request_page(cursor: str | None, limit: int) -> AmazonMusicConnection: params: dict[str, str | int] = {"limit": limit} if cursor is not None: params["cursor"] = cursor raw = self.request( "GET", f"{API_BASE}/me/playlists", params=params, headers=self._auth_headers(access_token), type=dict[str, Any], ) return raw.get("data", {}).get("user", {}).get("playlists", {}) return self._paginate_cursor(request_page, max_items=max_items) def get_current_user_followed_artists( self, access_token: SecretStr, *, max_items: int = 100 ) -> list[dict[str, Any]]: def request_page(cursor: str | None, limit: int) -> AmazonMusicConnection: params: dict[str, str | int] = {"limit": limit} if cursor is not None: params["cursor"] = cursor raw = self.request( "GET", f"{API_BASE}/me/followed/artists", params=params, headers=self._auth_headers(access_token), type=dict[str, Any], ) return raw.get("data", {}).get("user", {}).get("followedArtists", {}) return self._paginate_cursor(request_page, max_items=max_items)