MusicAnalyser/backend/app/services/spotify_client.py

import os
import base64
import time
import httpx
from fastapi import HTTPException
from typing import List, Dict, Any

SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
SPOTIFY_API_BASE = "https://api.spotify.com/v1"


class SpotifyClient:
    def __init__(self, client_id: str, client_secret: str, refresh_token: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.access_token = None
        self.token_expires_at = 0

    async def get_access_token(self):
        """Returns a valid access token, refreshing if necessary."""
        if self.access_token and time.time() < self.token_expires_at:
            return self.access_token

        print("Refreshing Spotify Access Token...")
        async with httpx.AsyncClient() as client:
            auth_str = f"{self.client_id}:{self.client_secret}"
            b64_auth = base64.b64encode(auth_str.encode()).decode()

            response = await client.post(
                SPOTIFY_TOKEN_URL,
                data={
                    "grant_type": "refresh_token",
                    "refresh_token": self.refresh_token,
                },
                headers={"Authorization": f"Basic {b64_auth}"},
            )

            if response.status_code != 200:
                print(f"Failed to refresh token: {response.text}")
                raise Exception("Could not refresh Spotify token")

            data = response.json()
            self.access_token = data["access_token"]
            # expires_in is usually 3600 seconds. buffer by 60s
            self.token_expires_at = time.time() + data["expires_in"] - 60
            return self.access_token

    async def get_recently_played(self, limit=50):
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/me/player/recently-played",
                params={"limit": limit},
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code != 200:
                print(f"Error fetching recently played: {response.text}")
                return []

            return response.json().get("items", [])

    async def get_track(self, track_id: str):
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/tracks/{track_id}",
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code != 200:
                return None
            return response.json()

    async def get_artists(self, artist_ids: List[str]) -> List[Dict[str, Any]]:
        """
        Fetches artist details (including genres) for a list of artist IDs.
        Spotify allows up to 50 IDs per request.
        """
        if not artist_ids:
            return []

        token = await self.get_access_token()
        ids_param = ",".join(artist_ids)

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/artists",
                params={"ids": ids_param},
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code != 200:
                print(f"Error fetching artists: {response.text}")
                return []

            return response.json().get("artists", [])

    async def get_currently_playing(self) -> Dict[str, Any] | None:
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/me/player/currently-playing",
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code == 204:
                return None
            if response.status_code != 200:
                print(f"Error fetching currently playing: {response.text}")
                return None
            return response.json()

    async def get_current_user_id(self) -> str:
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/me",
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code != 200:
                raise Exception(f"Failed to get user profile: {response.text}")
            return response.json().get("id")

    async def create_playlist(
        self, user_id: str, name: str, description: str = "", public: bool = False
    ) -> Dict[str, Any]:
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{SPOTIFY_API_BASE}/users/{user_id}/playlists",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
                json={"name": name, "description": description, "public": public},
            )
            if response.status_code not in [200, 201]:
                raise Exception(f"Failed to create playlist: {response.text}")
            return response.json()

    async def update_playlist_details(
        self, playlist_id: str, name: str = None, description: str = None
    ) -> bool:
        token = await self.get_access_token()
        data = {}
        if name:
            data["name"] = name
        if description:
            data["description"] = description

        if not data:
            return True

        async with httpx.AsyncClient() as client:
            response = await client.put(
                f"{SPOTIFY_API_BASE}/playlists/{playlist_id}",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
                json=data,
            )
            return response.status_code == 200

    async def replace_playlist_tracks(
        self, playlist_id: str, track_uris: List[str]
    ) -> bool:
        token = await self.get_access_token()
        async with httpx.AsyncClient() as client:
            response = await client.put(
                f"{SPOTIFY_API_BASE}/playlists/{playlist_id}/tracks",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
                json={"uris": track_uris[:100]},
            )
            return response.status_code in [200, 201]

    async def get_recommendations(
        self,
        seed_tracks: List[str] = None,
        seed_artists: List[str] = None,
        seed_genres: List[str] = None,
        limit: int = 20,
        target_energy: float = None,
        target_valence: float = None,
    ) -> List[Dict[str, Any]]:
        token = await self.get_access_token()
        params = {"limit": limit}

        if seed_tracks:
            params["seed_tracks"] = ",".join(seed_tracks[:5])
        if seed_artists:
            params["seed_artists"] = ",".join(seed_artists[:5])
        if seed_genres:
            params["seed_genres"] = ",".join(seed_genres[:5])
        if target_energy is not None:
            params["target_energy"] = target_energy
        if target_valence is not None:
            params["target_valence"] = target_valence

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SPOTIFY_API_BASE}/recommendations",
                params=params,
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code != 200:
                print(f"Error fetching recommendations: {response.text}")
                return []
            return response.json().get("tracks", [])