Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/agbocsardi/dss-group-assignment.
Something went wrong. Try again.
16 kB · 414 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415# Lovely resource# https://developer.spotify.com/documentation/web-api/reference/#/operations/remove-tracks-playlistimport credentialsimport requestsimport jsonimport base64import random
# For interacting with the Flask serverimport os, signal, time, pathlibfrom selenium import webdriverfrom selenium.webdriver.support.wait import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.common.by import By
class API: def __init__(self): self.ACCESS_TOKEN = None self.refreshToken() # self.getPlaylists()
def refreshToken(self): try: self.ImplicitGrantFlow() except: if self.ACCESS_TOKEN == None: self.ClientCredentialsFlow() print("Client Credential Flow. If you wish to alter user playlists, look up how to run selenium with the Firefox driver.")
def RunServer(self): # cwd = os.getcwd() filepath = str(pathlib.Path(__file__).resolve().parent) print('filepath', filepath) implicit_path = filepath.split(sep="/")[:-1] implicit_path = "/".join(implicit_path) + "/implicit" print('implicit path', implicit_path) self.SERVER = os.popen(f"cd {implicit_path} && flask run") # Inform the scraper that the server is running self.ON = True
def Login(self): while not self.ON: time.sleep(1)
options = webdriver.FirefoxOptions() # Hides the window options.add_argument('headless')
# Open firefox driver = webdriver.Firefox(options=options) # Go to website driver.get('http://127.0.0.1:5000/') # Click login btn = driver.find_element(By.XPATH, '//a[@href="/auth"]') driver.execute_script("arguments[0].click();", btn)
# Log in email = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.CSS_SELECTOR,"input[id='login-username']"))) password = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.CSS_SELECTOR,"input[id='login-password']")))
email.clear() password.clear()
email.send_keys(credentials.EMAIL) password.send_keys(credentials.PASSWORD)
WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.CSS_SELECTOR,"button[id='login-button']"))).click()
try: WebDriverWait(driver, 2).until(EC.element_to_be_clickable((By.CSS_SELECTOR,"button[class='Button-qlcn5g-0 jWBSO']"))).click() except: print()
# Get auth token url = driver.current_url def find_between(s, start, end): return (s.split(start))[1].split(end)[0] self.ACCESS_TOKEN = find_between(url, 'access_token=', '&token_type=')
# Set it # Close site driver.quit() if self.ON: self.ON = False # Very hacky way to exit, because it doesn't actually work lol. Right now behavior works properly because the ACCESS_TOKEN is set, and the refreshToken checks for it before running Client Credential Flow. self.SERVER.send_signal(signal.CTRL_C_EVENT)
def ImplicitGrantFlow(self): self.RunServer() self.Login()
def ClientCredentialsFlow(self): # The data to be sent url = "https://accounts.spotify.com/api/token"
headers= { "Authorization": f"Basic {base64.b64encode(bytes(credentials.CLIENT_ID + ':' + credentials.CLIENT_SECRET, 'ISO-8859-1')).decode('ascii')}", "Content-Type": "application/x-www-form-urlencoded" }
body= { "grant_type": "client_credentials" # TODO need to use this scope and another one of the 3 available workflows in order to create playlists # "scope": "playlist-modify-public playlist-read-private playlist-modify-private" }
# Send the request response = requests.post(url=url, data=body, headers=headers)
# Update the access token self.ACCESS_TOKEN = json.loads(response.text)["access_token"] print (response.reason)
def getPlaylists(self, user=credentials.USER_ID, offset=0, limit=20): """Get the public playlists of the specified user (default is the test user)
user: user ID from Spotify
offset: offset where the list begins, default is 0, meaning it starts at the beginning
limit: how many items to return """ url = f"https://api.spotify.com/v1/users/{user}/playlists?offset={offset}&limit={limit}"
payload={} headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}' }
response = requests.request("GET", url, headers=headers, data=payload) response = json.loads(response.text) self.playlists = response["items"]
return self.playlists
def getAvailableGenres(self): """List of genre names that can be used in the Spotify API""" url = "https://api.spotify.com/v1/recommendations/available-genre-seeds" headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Content-Type': 'application/json' } response = requests.get(url, headers=headers) response = response.json() return response
def getRecommendations(self, seed_artists= '', seed_genres= 'str', seed_tracks= '', limit=10, try_number=1, min_acousticness= None, max_acousticness= None, min_danceability= None, max_danceability= None, min_duration_ms= None, max_duration_ms= None, min_energy= None, max_energy= None, min_instrumentalness= None, max_instrumentalness= None, min_key= None, max_key= None, min_liveness= None, max_liveness= None, min_loudness= None, max_loudness= None, min_speechiness= None, max_speechiness= None, min_tempo= None, max_tempo= None, min_time_signature= None, max_time_signature= None, min_valence= None, max_valence= None): """Get recommendations based on characteristics, NOT based on user! Max 5 seed values total! len(seed_artists) + len(seed_genres) + len(seed_tracks) <= 5. That's a hard limit. Inputs: seed_artists (str): comma-separated list; artist Spotify IDs (unique string at the end of the Spotify URI) seed_genres (str): comma-separated list; genres such as "classical,country" seed_tracks (str): comma-separated list; song Spotify IDs (unique string at the end of the Spotify URI) limit (int): number of songs to return try_number (int): number of attempts to get recommendations min_acousticness (int): optional input max_acousticness (int): optional input min_danceability (int): optional input max_danceability (int): optional input min_duration_ms (int): optional input max_duration_ms (int): optional input min_energy (int): optional input max_energy (int): optional input min_instrumentalness (int): optional input max_instrumentalness (int): optional input min_key (int): optional input max_key (int): optional input min_liveness (int): optional input max_liveness (int): optional input min_loudness (int): optional input max_loudness (int): optional input min_speechiness (int): optional input max_speechiness (int): optional input min_tempo (int): optional input max_tempo (int): optional input min_time_signature (int): optional input max_time_signature (int): optional input min_valence (int): optional input max_valence (int): optional input """ # Error handling # TODO break up the seeds, split by column non_empty_seeds = [] if seed_artists != '': non_empty_seeds.append(seed_artists) if seed_genres != '': non_empty_seeds.append(seed_genres) if seed_tracks != '': non_empty_seeds.append(seed_tracks)
seed_length = len([seed.split(',') for seed in non_empty_seeds]) # seed_length = len(seed_artists.split(',')) + len(seed_genres.split(',')) + len(seed_tracks.split(',')) if seed_length > 5: raise Exception(f"No more than 5 seeds TOTAL allowed! {seed_length} seeds are in the current input.\nlen(seed_artists) + len(seed_genres) + len(seed_tracks) must be less than 5.\nCurrently receiving: seed_artists={seed_artists}, seed_genres={seed_genres}, seed_tracks={seed_tracks}") elif seed_length < 1: raise Exception(f"You need at least 1 seed! {seed_length} seeds are in the current input.\nlen(seed_artists) + len(seed_genres) + len(seed_tracks) must be at least 1.")
# Jeez that's a lot of inputs! input_dict = { "seed_artists": seed_artists, "seed_genres": seed_genres, "seed_tracks": seed_tracks, "limit": limit, "min_acousticness": min_acousticness, "max_acousticness": max_acousticness, "min_danceability": min_danceability, "max_danceability": max_danceability, "min_duration_ms": min_duration_ms, "max_duration_ms": max_duration_ms, "min_energy": min_energy, "max_energy": max_energy, "min_instrumentalness": min_instrumentalness, "max_instrumentalness": max_instrumentalness, "min_key": min_key, "max_key": max_key, "min_liveness": min_liveness, "max_liveness": max_liveness, "min_loudness": min_loudness, "max_loudness": max_loudness, "min_speechiness": min_speechiness, "max_speechiness": max_speechiness, "min_tempo": min_tempo, "max_tempo": max_tempo, "min_time_signature": min_time_signature, "max_time_signature": max_time_signature, "min_valence": min_valence, "max_valence": max_valence }
# Remove any None values, sending those to the API may cause weird behavior payload_dict = dict() for key, value in input_dict.items(): # Add it to the dictionary if value != None: payload_dict[key] = value
# Prepare query url = "https://api.spotify.com/v1/recommendations?"
query = url for key, value in payload_dict.items(): query = f"{query}{key}={value}&" query = query[:-1] # removes the last '&'
headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Accept': 'application/json', 'Content-Type': 'application/json' }
# Sometimes this just doesn't work, rather than fixing, I'm putting a bandage on it try: # Send the request response = requests.get(query, headers=headers) response = response.json() tracks = [track['uri'] for track in response['tracks']] # Grab just the song ID, not the URI tracks = [track.split(':')[2] for track in tracks] return [self.getTrackFeatures(track) for track in tracks] except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError): # If there's an error, try this recursively time.sleep(2**try_number + random.random()*0.01) # exponential backoff return self.getRecommendations(seed_artists, seed_genres, seed_tracks, limit, try_number=try_number+1, min_acousticness=min_acousticness, max_acousticness=max_acousticness, min_danceability=min_danceability, max_danceability=max_danceability, min_duration_ms=min_duration_ms, max_duration_ms=max_duration_ms, min_energy=min_energy, max_energy=max_energy, min_instrumentalness=min_instrumentalness, max_instrumentalness=max_instrumentalness, min_key=min_key, max_key=max_key, min_liveness=min_liveness, max_liveness=max_liveness, min_loudness=min_loudness, max_loudness=max_loudness, min_speechiness=min_speechiness, max_speechiness=max_speechiness, min_tempo=min_tempo, max_tempo=max_tempo, min_time_signature=min_time_signature, max_time_signature=max_time_signature, min_valence=min_valence, max_valence=max_valence)
def getTrackFeatures(self, song_id): url = f"https://api.spotify.com/v1/audio-features/{song_id}"
headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Content-Type': 'application/json' }
# Send the request response = requests.get(url, headers=headers) response = response.json() return response
def generatePlaylistNames(self, users: list): """These are just the ids given in the data.
Args: users (list): list of unique user id values.
Returns: a list of strings. """ # Use user ID as base # users = pd.read_csv("participant data/survey data/msi_response.csv") # users = users["user_id"].unique() playlist_names = [] for user in users: # To be used as the diverse playlist playlist_names.append(user+'_d') # To be used as the non-diverse playlist playlist_names.append(user+'_n')
return playlist_names
def createPlaylist(self, playlist_name: str, user= credentials.USER_ID): url = f"https://api.spotify.com/v1/users/{user}/playlists"
payload = json.dumps({ "name": playlist_name, "public": "true" })
headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Content-Type': 'application/json' }
response = requests.request("POST", url, headers=headers, data=payload)
print(response.text) if not "collaborative" in response.text: print(f"Creation of {playlist_name} failed.")
def createAllPlaylistsForAllUsers(self, user_ids: list, login_user=credentials.USER_ID): """The user refers to the user account where the playlists will be created. """ # Get names of playlists to generate new_names = self.generatePlaylistNames(user_ids) # existing_playlists = self.getPlaylists(limit=len(new_names)) existing_playlists = self.getPlaylists(user=login_user)
# Only create playlists if no playlists with that name exists prior playlist_names = [] for i in range(len(existing_playlists)): playlist_names.append(existing_playlists[i]['name']) creatables = [name for name in new_names if name not in playlist_names]
# # TODO REMOVE! The following line is for testing purposes only # creatables = ['python test 2'] # Requests for the creation of the playlists url = f"https://api.spotify.com/v1/users/{login_user}/playlists"
for new_name in creatables: payload = json.dumps({ "name": new_name, "public": "true" })
headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Content-Type': 'application/json' }
response = requests.request("POST", url, headers=headers, data=payload)
if not "collaborative" in response.text: print(f"Creation of {new_name} failed.") print(response.text) print("The playlists are created!")
def getPlaylistIdFromName(self, playlist_name: str, user= credentials.USER_ID, limit= 1): # Get the playlists of the user first if user != credentials.USER_ID: playlists= self.getPlaylists(user=user, limit=limit) else: playlists = self.playlists # Create dictionary with names as keys and URIs as values for item in playlists: if item["name"] == playlist_name: print(f"Playlist {playlist_name} found!") playlist_id = item["uri"].split(":")[2] return playlist_id
print (f"ERROR: No playlist with the name {playlist_name} found for user {user}.")
# TODO test this def populatePlaylist(self, playlist_id: str, song_uris: list, user=credentials.USER_ID): # Test add to list url = f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks"
# # Convert song uris to the right format, if only the id is given # uris = [] # for uri in song_uris: # uris.append(f"spotify:track:{uri}")
payload = json.dumps({ "uris": song_uris }) headers = { 'Authorization': f'Bearer {self.ACCESS_TOKEN}', 'Content-Type': 'application/json' }
response = requests.request("POST", url, headers=headers, data=payload)
print(response.text)