-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
118 lines (101 loc) · 4.19 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import requests
import base64
import datetime
from urllib.parse import urlencode
client_id = 'your_client_id'
client_secret = 'your_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):
"""
Return a base64 encoded string
"""
client_id = self.client_id
client_secret = self.client_secret
if client_secret == None or client_id == None:
raise Exception("Need to indicate 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("Client is not authenticated!")
data = r.json()
now = datetime.datetime.now()
access_token = data['access_token']
expires_in = data['expires_in']
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 get_resource(self, loockup_id, resource_type='albums', version='v1'):
endpoint = f"https://api.spotify.com/{version}/{resource_type}/{loockup_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):
headers = self.get_resource_header()
endpoint = 'https://api.spotify.com/v1/search'
looup_url = f"{endpoint}?{query_params}"
r = requests.get(looup_url, headers=headers)
if r.status_code not in range(200, 299):
return {}
return r.json()
def search(self, operator=None, operator_query=None, 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()})
return self.base_search(query_params)