spotify_client.py•7.8 kB
import base64
from urllib.parse import urlencode
from typing import List
import requests
class SpotifyClient:
def __init__(self, client_id: str, client_secret: str, redirect_uri: str):
self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
self.base_url = "https://api.spotify.com/v1"
def build_auth_url(self, state: str):
params = {
"client_id": self.client_id,
"response_type": "code",
"redirect_uri": self.redirect_uri,
"scope": "playlist-read-private playlist-read-collaborative playlist-modify-private playlist-modify-public user-read-email user-read-private",
"state": state
}
return f"https://accounts.spotify.com/authorize?{urlencode(params)}"
def get_access_token(self, code: str):
authOptions = {
"data": {
"code": code,
"redirect_uri": self.redirect_uri,
"grant_type": 'authorization_code'
},
"headers": {
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Basic {base64.b64encode(f'{self.client_id}:{self.client_secret}'.encode()).decode()}"
}
}
response = requests.post("https://accounts.spotify.com/api/token", **authOptions)
return response.json() if response.status_code == 200 else None
def create_playlist_with_tracks(self, access_token: str, user_id: str, name: str, track_uris: List[str], description: str = "") -> dict:
"""
Creates a private playlist and adds the specified tracks to it.
Args:
access_token: Spotify access token
user_id: Spotify user ID
name: Name of the playlist
track_uris: List of Spotify track URIs to add
description: Optional playlist description
Returns:
The created playlist data if successful, None otherwise
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# Create the playlist
playlist_data = {
"name": name,
"description": description,
"public": False
}
response = requests.post(
f"{self.base_url}/users/{user_id}/playlists",
headers=headers,
json=playlist_data
)
if response.status_code != 201:
return None
playlist = response.json()
# Add tracks to the playlist
tracks_data = {
"uris": track_uris
}
response = requests.post(
f"{self.base_url}/playlists/{playlist['id']}/tracks",
headers=headers,
json=tracks_data
)
return playlist if response.status_code == 201 else None
def get_track_uri(self, access_token: str, artist: str, song_name: str) -> str:
"""
Search for a track on Spotify and return its URI.
Args:
access_token: Spotify access token
artist: Name of the artist
song_name: Name of the song
Returns:
Spotify track URI if found, None otherwise
"""
headers = {
"Authorization": f"Bearer {access_token}"
}
# Create search query combining artist and track name
query = f"artist:{artist} track:{song_name}"
params = {
"q": query,
"type": "track",
"limit": 1 # We only need the best match
}
response = requests.get(
f"{self.base_url}/search",
headers=headers,
params=params
)
if response.status_code != 200:
return None
results = response.json()
# Check if we got any tracks back
if not results["tracks"]["items"]:
return None
# Return the URI of the first (best matching) track
return results["tracks"]["items"][0]["uri"]
def refresh_access_token(self, refresh_token: str) -> dict:
"""
Get a new access token using a refresh token.
Args:
refresh_token: The refresh token from a previous authentication
Returns:
Dict containing the new access token and refresh token if successful, None otherwise
"""
auth_options = {
"data": {
"grant_type": "refresh_token",
"refresh_token": refresh_token
},
"headers": {
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Basic {base64.b64encode(f'{self.client_id}:{self.client_secret}'.encode()).decode()}"
}
}
response = requests.post("https://accounts.spotify.com/api/token", **auth_options)
return response.json() if response.status_code == 200 else None
def update_playlist_details(self, access_token: str, playlist_id: str, name: str = None, description: str = None, public: bool = None) -> dict:
"""
Updates an existing playlist's details.
Args:
access_token: Spotify access token
playlist_id: ID of the playlist to update
name: New name for the playlist (optional)
description: New description for the playlist (optional)
public: New public/private status for the playlist (optional)
Returns:
Dictionary containing playlist details if successful, None otherwise
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
playlist_data = {}
if name is not None:
playlist_data["name"] = name
if description is not None:
playlist_data["description"] = description
if public is not None:
playlist_data["public"] = public
response = requests.put(
f"{self.base_url}/playlists/{playlist_id}",
headers=headers,
json=playlist_data
)
if response.status_code == 200:
return {
"id": playlist_id,
"name": name if name is not None else "",
"description": description if description is not None else "",
"public": public if public is not None else False,
"tracks": {"total": 0},
"type": "playlist",
"uri": f"spotify:playlist:{playlist_id}"
}
return None
def replace_playlist_tracks(self, access_token: str, playlist_id: str, track_uris: List[str]) -> bool:
"""
Replaces all tracks in a playlist with new ones.
Args:
access_token: Spotify access token
playlist_id: ID of the playlist to update
track_uris: New list of track URIs to replace existing tracks
Returns:
True if successful, False otherwise
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
tracks_data = {
"uris": track_uris
}
response = requests.put(
f"{self.base_url}/playlists/{playlist_id}/tracks",
headers=headers,
json=tracks_data
)
return response.status_code == 200