forked from simonbru/spotify-backup
-
Notifications
You must be signed in to change notification settings - Fork 0
/
auth_server.py
executable file
·190 lines (153 loc) · 5.3 KB
/
auth_server.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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
#!/usr/bin/env python3
import ast
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.request import HTTPError, Request, urlopen
from urllib.parse import urlencode, urlparse, parse_qs
from config import TOKEN_FILE as RELATIVE_TOKEN_FILE
from config import CLIENT_ID as SPOTIFY_CLIENT_ID, CLIENT_SECRET as SPOTIFY_CLIENT_SECRET
from config import REFRESHABLE_AUTH
SERVER_PORT = 8193
REDIRECT_URI = f'http://localhost:{SERVER_PORT}/auth'
TOKEN_FILE = Path(__file__).parent / RELATIVE_TOKEN_FILE
def create_request_handler():
shared_context = {
'access_token': None,
'code': None,
'error': None
}
class AuthRequestHandler(BaseHTTPRequestHandler):
_redirect_tpl = """
<html>
<script>
if (location.href.indexOf('#') != -1)
location.href = location.href.replace("#","?");
</script>
<h1>Redirecting...</h1>
</html>
"""
_success_tpl = """
<html>
<h1>Authorization successful</h1>
<p>You can close this page now</p>
</html>
"""
_error_tpl = """
<html>
<h1>Authorization error</h1>
<p>{error}</p>
</html>
"""
def do_GET(self):
qs = urlparse(self.path).query
qs_dict = parse_qs(qs)
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
if not qs_dict:
html = self._redirect_tpl
elif 'access_token' in qs_dict:
token = qs_dict['access_token'][0]
shared_context['access_token'] = token
html = self._success_tpl
elif 'code' in qs_dict:
token = qs_dict['code'][0]
shared_context['code'] = token
html = self._success_tpl
else:
error = qs_dict.get('error', ['unknown'])[0]
shared_context['error'] = error
html = self._error_tpl.format(error=error)
self.wfile.write(html.encode('utf8'))
self.wfile.flush()
return AuthRequestHandler, shared_context
class AuthorizationError(Exception):
pass
def listen_for_token(port):
request_handler, shared_context = create_request_handler()
httpd = HTTPServer(('localhost', port), request_handler)
while True:
httpd.handle_request()
if shared_context['access_token']:
return shared_context['access_token']
if shared_context['code']:
return shared_context['code']
elif shared_context['error']:
raise AuthorizationError(shared_context['error'])
def prompt_user_for_auth():
if REFRESHABLE_AUTH:
response_type = 'code'
else:
response_type = 'token'
url_params = urlencode({
'redirect_uri': REDIRECT_URI,
'response_type': response_type,
'client_id': SPOTIFY_CLIENT_ID,
'scope': 'playlist-read-private playlist-read-collaborative'
})
url = f'https://accounts.spotify.com/authorize?{url_params}'
print(
"",
"Please open the following URL in a Web Browser to continue:",
url,
sep='\n'
)
return listen_for_token(port=SERVER_PORT)
def request_refresh_token(token_params):
request_params = urlencode({
'redirect_uri': REDIRECT_URI,
'client_id': SPOTIFY_CLIENT_ID,
'client_secret': SPOTIFY_CLIENT_SECRET,
**token_params
}).encode()
url = f'https://accounts.spotify.com/api/token'
with urlopen(url, data=request_params) as response:
return ast.literal_eval(response.read().decode())
def activate_refresh_token(refresh_token):
token_params = {
'grant_type': 'authorization_code',
'code': refresh_token
}
content = request_refresh_token(token_params)
return content['access_token'], content['refresh_token']
def redeem_refresh_token(refresh_token):
token_params = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token
}
content = request_refresh_token(token_params)
return content['access_token']
def save_token_to_file(token):
token_path = Path(TOKEN_FILE).resolve()
token_path.touch(0o600, exist_ok=True)
token_path.write_text(token)
def get_token(restore_token=True, save_token=True):
if restore_token and TOKEN_FILE.resolve().exists():
token = Path(TOKEN_FILE).read_text().strip()
if REFRESHABLE_AUTH:
try:
token = redeem_refresh_token(token)
except HTTPError:
# This will fail if the token wasn't a refresh token
pass
# Check that the token is valid
req = Request(
'https://api.spotify.com/v1/me',
headers={'Authorization': f'Bearer {token}'}
)
try:
urlopen(req)
except HTTPError:
pass
else:
return token
token = prompt_user_for_auth()
if REFRESHABLE_AUTH:
token, savable_token = activate_refresh_token(token)
else:
savable_token = token
if save_token:
save_token_to_file(savable_token)
return token
if __name__ == '__main__':
print(get_token())