init

spotify.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+import base64
+import datetime
+from urllib.parse import urlencode
+
+import requests
+
+
+
+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
+        """
+        client_id = self.client_id
+        client_secret = self.client_secret
+        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()
+        print(token_url)
+        print(token_data)
+        print(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}",
+            "Accept": "application/json",
+            "Content-Type": "application/json"
+        }
+        return headers
+        
+        
+    def get_resource(self, lookup_id, resource_type='albums', version='v1'):
+        endpoint = f"https://api.spotify.com/{version}/{resource_type}/{lookup_id}"
+        headers = self.get_resource_header()
+        r = requests.get(endpoint, headers=headers)
+        if r.status_code not in range(200, 299):
+            return {}
+        return r.json()
+    
+    def get_album(self, _id):
+        return self.get_resource(_id, resource_type='albums')
+    
+    def get_artist(self, _id):
+        return self.get_resource(_id, resource_type='artists')
+    
+    def base_search(self, query_params): # type
+        headers = self.get_resource_header()
+        endpoint = "https://api.spotify.com/v1/search"
+        lookup_url = f"{endpoint}?{query_params}"
+        r = requests.get(lookup_url, headers=headers)
+        if r.status_code not in range(200, 299):  
+            return {}
+        return r.json()
+    
+    def get_devices(self): # type
+        headers = self.get_resource_header()
+        endpoint = "https://api.spotify.com/v1/me/player/devices"
+        lookup_url = f"{endpoint}"
+        print(lookup_url)
+        print(headers)
+        r = requests.get(lookup_url, headers=headers)
+        print(r.status_code)
+        if r.status_code not in range(200, 299):  
+            return {}
+        return r.json()
+    
+    def search(self, query=None, operator=None, operator_query=None, search_type='artist' ):
+        if query == None:
+            raise Exception("A query is required")
+        if isinstance(query, dict):
+            query = " ".join([f"{k}:{v}" for k,v in query.items()])
+        if operator != None and operator_query != None:
+            if operator.lower() == "or" or operator.lower() == "not":
+                operator = operator.upper()
+                if isinstance(operator_query, str):
+                    query = f"{query} {operator} {operator_query}"
+        query_params = urlencode({"q": query, "type": search_type.lower()})
+        print(query_params)
+        return self.base_search(query_params)
+
+
+spotify = SpotifyAPI("6a37229fc9a04878a608d519ea4b1d22","6569f6f876234093983920de2966ff84" )
+
+
+spotify.perform_auth()
+
+print(spotify.access_token)
+print(spotify.access_token_expires)
+
+print(spotify.get_resource_header())
+
+
+
+print(spotify.get_devices())