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", [])