alembic/app/services/spotify_client.py

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