import base64 import re import time import httpx from sqlalchemy.orm import Session from app.services import credential_service TOKEN_URL = "https://accounts.spotify.com/api/token" API_BASE = "https://api.spotify.com/v1" # Module-level cache: one alembic process, one Spotify app registration -- # a single shared token is fine (single-user app, no per-request identity). # Keyed by mode too, so connecting/disconnecting an account (switching # between a user token and a client-credentials token) can't serve a stale # token minted under the other grant type. _token_cache: dict = {"token": None, "expires_at": 0.0, "mode": None} _PLAYLIST_ID_RE = re.compile(r"playlist/([A-Za-z0-9]+)") def _extract_playlist_id(playlist_url: str) -> str: m = _PLAYLIST_ID_RE.search(playlist_url) if not m: raise ValueError(f"not a Spotify playlist URL: {playlist_url}") return m.group(1) def _get_token(db: Session) -> str: now = time.time() creds = credential_service.get_scope(db, "spotify") client_id = creds.get("client_id") client_secret = creds.get("client_secret") if not client_id or not client_secret: raise RuntimeError("Spotify credentials not configured (settings/credentials)") refresh_token = creds.get("refresh_token") # A user token (from /connect/spotify) can read playlists on newly created # Spotify apps; client-credentials can't (Spotify blocks GET /items for # new apps without user auth). Prefer the user token whenever one is # connected, falling back to client-credentials for grandfathered apps. mode = "user" if refresh_token else "client_credentials" if ( _token_cache["token"] and _token_cache["mode"] == mode and _token_cache["expires_at"] > now + 30 ): return _token_cache["token"] basic = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() grant_data = ( {"grant_type": "refresh_token", "refresh_token": refresh_token} if refresh_token else {"grant_type": "client_credentials"} ) resp = httpx.post( TOKEN_URL, data=grant_data, headers={"Authorization": f"Basic {basic}"}, timeout=15, ) resp.raise_for_status() data = resp.json() _token_cache["token"] = data["access_token"] _token_cache["expires_at"] = now + data["expires_in"] _token_cache["mode"] = mode return _token_cache["token"] def get_playlist_tracks(db: Session, playlist_url: str) -> list[dict]: """Return [{"artist": ..., "title": ..., "isrc": ... | None}, ...] for every track in the playlist, paginated 100/page.""" token = _get_token(db) playlist_id = _extract_playlist_id(playlist_url) headers = {"Authorization": f"Bearer {token}"} tracks = [] # /items, not /tracks: Spotify removed GET /playlists/{id}/tracks in its # February 2026 API changes in favor of /items (same response shape). New # apps are 403'd on the old endpoint; grandfathered apps still tolerate it # for now, but /items is the correct, future-proof one. url = f"{API_BASE}/playlists/{playlist_id}/items" params = {"limit": 100, "fields": "items(track(name,artists(name),external_ids)),next"} while url: resp = httpx.get(url, params=params, headers=headers, timeout=15) resp.raise_for_status() data = resp.json() for item in data.get("items", []): track = item.get("track") if not track: continue # local files / removed tracks show up as null artists = track.get("artists") or [] tracks.append( { "artist": artists[0]["name"] if artists else "", "title": track.get("name", ""), "isrc": (track.get("external_ids") or {}).get("isrc"), } ) url = data.get("next") params = None # `next` is a full URL with its own query string return tracks