-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathspotify_flask_app.py
More file actions
126 lines (102 loc) · 4.26 KB
/
spotify_flask_app.py
File metadata and controls
126 lines (102 loc) · 4.26 KB
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
from flask import Flask, request, redirect, jsonify
import random
import base64
import os
import string
import json
import requests
import urllib.parse
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
app = Flask(__name__)
# Access environment variables
client_id = os.getenv('SPOTIFY_CLIENT_ID')
client_secret = os.getenv('SPOTIFY_CLIENT_SECRET')
redirect_uri = 'http://localhost:3000/callback'
def generate_random_string(length):
"""
Generates a random string of the specified length.
This function is typically used to create a unique 'state' parameter
during the OAuth 2.0 authorization process to prevent cross-site request forgery (CSRF) attacks.
Parameters:
- length (int): The length of the random string to generate.
Returns:
- str: A randomly generated string.
"""
rand_Str = string.ascii_letters + string.digits
return ''.join(random.choice(rand_Str) for _ in range(length))
@app.route('/login')
def login():
state = generate_random_string(16)
scope = 'user-read-private user-read-email user-read-recently-played playlist-read-private playlist-read-private user-top-read user-library-read user-follow-read'
params = {
'response_type': 'code',
'client_id': client_id,
'scope': scope,
'redirect_uri': redirect_uri,
'state': state
}
redirect_url = 'https://accounts.spotify.com/authorize?' + urllib.parse.urlencode(params)
return redirect(redirect_url)
@app.route('/callback')
def callback():
code = request.args.get('code', None)
state = request.args.get('state', None)
if state is None:
return jsonify({'error': 'state_mismatch'}), 400
else:
auth_options = {
'url': 'https://accounts.spotify.com/api/token',
'data': {
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
},
'headers': {
'content-type': 'application/x-www-form-urlencoded',
'Authorization': 'Basic ' + base64.b64encode(f'{client_id}:{client_secret}'.encode('utf-8')).decode('utf-8')
}
}
response = requests.post(auth_options['url'], data=auth_options['data'], headers=auth_options['headers'])
token_info = response.json()
# Store the token_info for further analysis
with open('token_info.json', 'w') as json_file:
json_file.write(json.dumps(token_info, indent=4))
# Return a response
return jsonify({'message': 'Authentication successful'})
@app.route('/refresh_token')
def refresh_token():
refresh_token = request.args.get('refresh_token', None)
if refresh_token is None:
return jsonify({'error': 'missing_refresh_token'}), 400
auth_options = {
'url': 'https://accounts.spotify.com/api/token',
'data': {
'grant_type': 'refresh_token',
'refresh_token': refresh_token
},
'headers': {
'content-type': 'application/x-www-form-urlencoded',
'Authorization': 'Basic ' + base64.b64encode(f'{client_id}:{client_secret}'.encode('utf-8')).decode('utf-8')
}
}
response = requests.post(auth_options['url'], data=auth_options['data'], headers=auth_options['headers'])
token_info = response.json()
# Write the modified token_info back to the file
with open('token_info_refreshed.json', 'w') as json_file:
json.dump(token_info, json_file, indent=4)
# Load existing token_info from the file
with open('token_info.json', 'r') as json_file:
token_info = json.load(json_file)
# Load refreshed token_info from the file
with open('token_info_refreshed.json', 'r') as refreshed_json_file:
refreshed_token_info = json.load(refreshed_json_file)
# Update the original token_info with the refreshed access_token
token_info['access_token'] = refreshed_token_info['access_token']
# Write the modified token_info back to the file
with open('token_info.json', 'w') as json_file:
json.dump(token_info, json_file, indent=4)
return jsonify({'message': 'Token have been successfully refreshed'})
if __name__ == '__main__':
app.run(port=3000)