94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
import info_imp
|
|
import json
|
|
import base64
|
|
import datetime
|
|
from urllib.parse import urlencode
|
|
import requests
|
|
|
|
client_id = info_imp.datinho['sp_client_id']
|
|
client_secret = info_imp.datinho['sp_client_secret']
|
|
|
|
class SpotifyAPI(object):
|
|
access_token = None
|
|
access_token_expires = datetime.datetime.now()
|
|
access_token_did_expire = True
|
|
client_id = None
|
|
client_secret = None
|
|
token_url = "https://accounts.spotify.com/api/token"
|
|
|
|
def __init__(self, client_id, client_secret, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
|
|
def get_client_credentials(self):
|
|
"""
|
|
Returns a base64 encoded string
|
|
"""
|
|
if client_secret == None or client_id == None:
|
|
raise Exception("You must set client_id and client_secret")
|
|
client_creds = f"{client_id}:{client_secret}"
|
|
client_creds_b64 = base64.b64encode(client_creds.encode())
|
|
return client_creds_b64.decode()
|
|
|
|
def get_token_headers(self):
|
|
client_creds_b64 = self.get_client_credentials()
|
|
return {
|
|
"Authorization": f"Basic {client_creds_b64}"
|
|
}
|
|
|
|
def get_token_data(self):
|
|
return {
|
|
"grant_type": "client_credentials"
|
|
}
|
|
|
|
def perform_auth(self):
|
|
token_url = self.token_url
|
|
token_data = self.get_token_data()
|
|
token_headers = self.get_token_headers()
|
|
r = requests.post(token_url, data=token_data, headers=token_headers)
|
|
if r.status_code not in range(200, 299):
|
|
raise Exception("Could not authenticate client.")
|
|
# return False
|
|
data = r.json()
|
|
now = datetime.datetime.now()
|
|
access_token = data['access_token']
|
|
expires_in = data['expires_in'] # seconds
|
|
expires = now + datetime.timedelta(seconds=expires_in)
|
|
self.access_token = access_token
|
|
self.access_token_expires = expires
|
|
self.access_token_did_expire = expires < now
|
|
return True
|
|
|
|
def get_access_token(self):
|
|
token = self.access_token
|
|
expires = self.access_token_expires
|
|
now = datetime.datetime.now()
|
|
if expires < now:
|
|
self.perform_auth()
|
|
return self.get_access_token()
|
|
elif token == None:
|
|
self.perform_auth()
|
|
return self.get_access_token()
|
|
return token
|
|
|
|
def get_resource_header(self):
|
|
access_token = self.get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}"
|
|
}
|
|
return headers
|
|
|
|
def getPlayTracks(self):
|
|
playlist_id = info_imp.datinho['sp_playlist_id']
|
|
endpoint = f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks"
|
|
access_token = self.get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/x-www-form-urlencoded"
|
|
}
|
|
r = requests.get(endpoint, headers=headers)
|
|
return r.json()
|
|
|
|
|