89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
import base64
|
|
import datetime
|
|
import os
|
|
import requests
|
|
|
|
SPOTIFY_CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID")
|
|
SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET")
|
|
|
|
class SessionData():
|
|
__access_token: str
|
|
__refresh_token: str
|
|
__expires_in: int
|
|
__creation_date: datetime.datetime
|
|
|
|
def __init__(self, __access_token, __refresh_token, __expires_in, __creation_date = None):
|
|
self.__access_token = __access_token
|
|
self.__refresh_token = __refresh_token
|
|
self.__expires_in = __expires_in
|
|
self.__creation_date = __creation_date or datetime.datetime.now()
|
|
|
|
def nearly_expired(self, before=30):
|
|
delta_time = datetime.datetime.now() - self.__creation_date
|
|
return delta_time.seconds > self.__expires_in - before
|
|
|
|
def is_expired(self):
|
|
return self.nearly_expired(0)
|
|
|
|
def refresh(self):
|
|
token_url = "https://accounts.spotify.com/api/token"
|
|
headers = {
|
|
"Authorization": "Basic " + base64.encode(SPOTIFY_CLIENT_ID + ":" + SPOTIFY_CLIENT_SECRET),
|
|
"Content-Type": "application/json"
|
|
}
|
|
data = {
|
|
"refresh_token": self.__refresh_token,
|
|
"grant_type": "refresh_token"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(token_url, json=data, headers=headers)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
# Assumendo che la risposta contenga questi campi
|
|
self.__access_token = result["access_token"]
|
|
self.__refresh_token = result.get("refresh_token", self.__refresh_token)
|
|
self.__expires_in = result["expires_in"]
|
|
self.__creation_date = datetime.datetime.now()
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Errore durante il refresh del token: {e}")
|
|
raise e
|
|
|
|
@property
|
|
def access_tokens(self):
|
|
if self.nearly_expired():
|
|
self.refresh()
|
|
|
|
return self.__access_token
|
|
|
|
|
|
class SessionManager():
|
|
__current_session: SessionData
|
|
__instance: "SessionManager" = None
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls.__instance is None:
|
|
cls.__instance = super(SessionManager, cls).__new__(cls)
|
|
return cls.__instance
|
|
|
|
def __init__(self):
|
|
self.__current_session = None
|
|
|
|
@classmethod
|
|
def instance(cls):
|
|
if cls.__instance is None:
|
|
cls.__instance = SessionManager()
|
|
|
|
return cls.__instance
|
|
|
|
def set_session(self, __current_session):
|
|
if self.__current_session is not None:
|
|
return
|
|
|
|
self.__current_session = __current_session
|
|
|
|
def get_current_session(self):
|
|
return self.__current_session
|
|
|