145 lines
4.2 KiB
Python
145 lines
4.2 KiB
Python
import requests
|
|
import base64
|
|
import re
|
|
import webbrowser
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
import urllib.parse
|
|
import threading
|
|
import time
|
|
|
|
CODE = ''
|
|
TOKEN = ''
|
|
|
|
# Define the HTTP request handler class
|
|
class RequestHandler(BaseHTTPRequestHandler):
|
|
|
|
# Handle GET requests
|
|
def do_GET(self):
|
|
global CODE
|
|
|
|
# Parse the URL and query parameters
|
|
parsed_url = urllib.parse.urlparse(self.path)
|
|
query_params = urllib.parse.parse_qs(parsed_url.query)
|
|
|
|
# Extract the authorization code from query parameters
|
|
if 'code' in query_params:
|
|
authorization_code = query_params['code'][0]
|
|
print(f"Authorization Code: {authorization_code}")
|
|
|
|
# You can now proceed with exchanging this authorization code for an access token
|
|
# Example: Call a function to exchange code for access token
|
|
#with open('code', 'w') as file:
|
|
# file.write(authorization_code)
|
|
|
|
# code = authorization_code
|
|
# print("Shutting down the callback server")
|
|
|
|
# Send response back to the browser
|
|
self.send_response(200)
|
|
self.send_header('Content-type', 'text/html')
|
|
self.end_headers()
|
|
self.wfile.write(b'<html><body>Authorization code received. You can close this window.</body></html>')
|
|
|
|
print("\n\nCode recieved!!!\n")
|
|
CODE = authorization_code
|
|
# self.server.shutdown()
|
|
|
|
else:
|
|
# Handle cases where code is not found
|
|
self.send_response(400)
|
|
self.send_header('Content-type', 'text/html')
|
|
self.end_headers()
|
|
self.wfile.write(b'<html><body>Authorization code not found.</body></html>')
|
|
|
|
|
|
# Set up and start the HTTP server
|
|
def startLoginCallbackServer():
|
|
server_address = ('localhost', 8888) # Use any free port you prefer
|
|
httpd = HTTPServer(server_address, RequestHandler)
|
|
print('Starting HTTP server on port 8080...')
|
|
httpd.serve_forever()
|
|
|
|
|
|
client_id = '3b4db45bb66b45e9a2532989c8034332'
|
|
client_secret = '4ced4e4575094b749834a60996680c6c'
|
|
|
|
|
|
def openLoginPage():
|
|
authorize_url = 'https://accounts.spotify.com/authorize'
|
|
|
|
params = {
|
|
'client_id': client_id,
|
|
'response_type': 'code',
|
|
'redirect_uri': 'http://localhost:8888',
|
|
'scope': 'user-read-private user-read-email', # Add scopes based on your needs
|
|
}
|
|
|
|
auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
|
|
webbrowser.open(auth_url)
|
|
|
|
|
|
def authenticateWepAPIAfterLogin(code):
|
|
global TOKEN
|
|
|
|
client_creds = f"{client_id}:{client_secret}"
|
|
client_creds_b64 = base64.b64encode(client_creds.encode())
|
|
|
|
token_url = 'https://accounts.spotify.com/api/token'
|
|
|
|
token_data = {
|
|
"grant_type": "authorization_code",
|
|
"client_secret": client_secret,
|
|
"client_id": client_id,
|
|
"code": code,
|
|
'redirect_uri': 'http://localhost:8888',
|
|
"scope": "user-read-private, playlist-read-private, playlist-read-collaborative",
|
|
}
|
|
|
|
token_headers = {
|
|
# "Authorization": f"Basic {client_creds_b64.decode()}"
|
|
"Content-Type" : "application/x-www-form-urlencoded"
|
|
}
|
|
|
|
print("\nAuthenticating WEB API\n")
|
|
|
|
token_response = requests.post(token_url, data=token_data)
|
|
|
|
if token_response.status_code not in range(200, 299):
|
|
print(f"Failed to retrieve access token: {token_response.status_code}")
|
|
print(token_response.json())
|
|
|
|
token = token_response.json()['access_token']
|
|
token_type = token_response.json()['token_type']
|
|
token_time = token_response.json()['expires_in']
|
|
|
|
print(token_response.json())
|
|
|
|
TOKEN = token
|
|
|
|
print(f"\nGot the token! - {TOKEN} \n")
|
|
|
|
|
|
|
|
def main():
|
|
thread1 = threading.Thread(target=startLoginCallbackServer)
|
|
thread1.start()
|
|
|
|
time.sleep(3)
|
|
|
|
openLoginPage()
|
|
|
|
time.sleep(3)
|
|
|
|
print("Got the code ", CODE)
|
|
|
|
authenticateWepAPIAfterLogin(CODE)
|
|
|
|
with open('token', 'w') as file:
|
|
file.write(TOKEN)
|
|
|
|
thread1.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|