# Standard Library Imports import os import random import re from urllib.parse import urlparse, parse_qs # Third-Party Imports import gradio as gr import lyricsgenius import requests import spotipy from bs4 import BeautifulSoup from dotenv import load_dotenv from fuzzywuzzy import fuzz from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity from spotipy.exceptions import SpotifyException from requests.exceptions import Timeout # Local Application/Library Specific Imports from langchain.schema import HumanMessage, SystemMessage from final_msgs import AUTH_HEADER, DISCLAIMER, LOCAL_INSTALL, NEED_SPOTIFY load_dotenv() ### ### ### Global Settings ### ### ### DEBUG_MODE = True # set to False to disable print statements def debug_print(*args, **kwargs): if DEBUG_MODE: print(*args, **kwargs) REDIRECT_URI = "https://huggingface.co/sjw" # TODO: switch to personal website # as required by the functions SCOPE = ['user-library-read', 'user-read-playback-state', 'user-modify-playback-state', 'playlist-modify-public', 'user-top-read'] # for play_genre_by_name_and_mood, play_artist_by_name_and_mood, and recommend_tracks() MOOD_SETTINGS = { "happy": {"max_instrumentalness": 0.001, "min_valence": 0.6}, "sad": {"max_danceability": 0.65, "max_valence": 0.4}, "energetic": {"min_tempo": 120, "min_danceability": 0.75}, "calm": {"max_energy": 0.65, "max_tempo": 130} } # for play_genre_by_name_and_mood NUM_ARTISTS = 20 # number of artists to retrieve from user's top artists; function accepts max 50 TIME_RANGE = "medium_term" # the time frame in which affinities are computed valid-values; short_term, medium_term, long_term NUM_TRACKS = 10 # number of tracks to return; also used by recommend_tracks() MAX_ARTISTS = 4 # recommendations() accepts a maximum of 5 seeds; in this case, genre will always be 1/5 # for play_artist_by_name_and_mood() NUM_ALBUMS = 20 # number of albums to retrieve at a maximum; function accepts max 20 MAX_TRACKS = 10 # number of tracks to randomly select from artist's albums ### ### ### Other Globals ### ### ### # NOTE: extremely important; ensures user isolation SP_STATE = gr.State() DEVICE_ID_STATE = gr.State() # for get_genre_by_name() # created states to avoid using global variables when possible GENRE_LIST = gr.State() GENRE_EMBEDDINGS = gr.State() AUTH_MSG = "Spotify client not initialized. Authenticate Spotify first." # for explain_track() GENIUS_TOKEN = os.getenv("GENIUS_ACCESS_TOKEN") # for play_playlist_by_name() and get_user_mood() # popular smaller/faster BERT; 6 layers as opposed to 12/24 MODEL = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') os.environ["TOKENIZERS_PARALLELISM"] = "false" # satisfies warning # for get_user_mood() MOOD_LIST = ["happy", "sad", "energetic", "calm"] MOOD_EMBEDDINGS = MODEL.encode(MOOD_LIST) # adjectives for playlist names THEMES = ["Epic", "Hypnotic", "Dreamy", "Legendary", "Majestic", "Enchanting", "Ethereal", "Super Lit", "Harmonious", "Heroic"] ### ### ### User Authentication ### ### ### with gr.Blocks() as auth_page: gr.Markdown(AUTH_HEADER) with gr.Row(): client_id = gr.Textbox(placeholder="5. Paste Spotify Client ID here, then click the button below", container=False, text_align="center") generate_link = gr.Button("6. Get Authentication Link") display_link = gr.Markdown() url = gr.Textbox(placeholder="7. Paste entire URL here, then click the button below", container=False, text_align="center") authorize_url = gr.Button("8. Authorize URL") auth_result = gr.Markdown() def spotify_auth(client_id, url=None): """ Authenticate Spotify with the provided client_id and url. """ if url: parsed_url = urlparse(url) fragment = parsed_url.fragment access_token = parse_qs(fragment)['access_token'][0] # NOTE: creating distinct Spotify states for each user sp = spotipy.Spotify(auth=access_token) SP_STATE.value = sp device_id = SP_STATE.value.devices()['devices'][0]['id'] DEVICE_ID_STATE.value = device_id # TODO: this is overkill; should probably just hardcode the genres GENRE_LIST.value = SP_STATE.value.recommendation_genre_seeds()["genres"] GENRE_EMBEDDINGS.value = MODEL.encode(GENRE_LIST.value) debug_print(SP_STATE.value) debug_print(DEVICE_ID_STATE.value) #return access_token # proof of distinct user sessions return """ Authentication Success. """ else: auth_url = ( f"https://accounts.spotify.com/authorize?response_type=token&client_id={client_id}" f"&scope={'%20'.join(SCOPE)}&redirect_uri={REDIRECT_URI}" ) return ("""Authorize by clicking here and copy the 'entire URL' you are redirected to""") generate_link.click(spotify_auth, inputs=[client_id], outputs=display_link) authorize_url.click(spotify_auth, inputs=[client_id, url], outputs=auth_result) with gr.Accordion(label="Local Installation 💻", open=False): gr.Markdown(LOCAL_INSTALL) with gr.Accordion(label="Don't Have Spotify 🫴?", open=False): gr.Markdown(NEED_SPOTIFY) with gr.Accordion(label="Security & Privacy 🛡️", open=False): gr.Markdown(DISCLAIMER) ### ### ### Basic Functions ### ### ### def find_track_by_name(track_name): """ Finds the Spotify track URI given the track name. """ if SP_STATE.value is None: return f"{AUTH_MSG}" results = SP_STATE.value.search(q=track_name, type='track') track_uri = results['tracks']['items'][0]['uri'] return track_uri def play_track_by_name(track_name): """ Plays a track given its name. Uses the above function. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" track_uri = find_track_by_name(track_name) track_name = SP_STATE.value.track(track_uri)["name"] artist_name = SP_STATE.value.track(track_uri)['artists'][0]['name'] try: SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, uris=[track_uri]) return f"♫ Now playing {track_name} by {artist_name} ♫" except SpotifyException as e: return f"An error occurred with Spotify: {e}. \n\n**Remember to wake up Spotify.**" except Exception as e: return f"An unexpected error occurred: {e}." def queue_track_by_name(track_name): """ Queues track given its name. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" track_uri = find_track_by_name(track_name) track_name = SP_STATE.value.track(track_uri)["name"] SP_STATE.value.add_to_queue(uri=track_uri, device_id=DEVICE_ID_STATE.value) return f"♫ Added {track_name} to your queue ♫" def pause_track(): """ Pauses the current playback. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" SP_STATE.value.pause_playback(device_id=DEVICE_ID_STATE.value) return "♫ Playback paused ♫" def resume_track(): """ Resumes the current playback. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value) return "♫ Playback started ♫" def skip_track(): """ Skips the current playback. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" SP_STATE.value.next_track(device_id=DEVICE_ID_STATE.value) return "♫ Skipped to your next track ♫" ### ### ### More Elaborate Functions ### ### ### def play_album_by_name_and_artist(album_name, artist_name): """ Plays an album given its name and the artist. context_uri (provide a context_uri to start playback of an album, artist, or playlist) expects a string. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" results = SP_STATE.value.search(q=f'{album_name} {artist_name}', type='album') album_id = results['albums']['items'][0]['id'] album_info = SP_STATE.value.album(album_id) album_name = album_info['name'] artist_name = album_info['artists'][0]['name'] try: SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, context_uri=f'spotify:album:{album_id}') return f"♫ Now playing {album_name} by {artist_name} ♫" except spotipy.SpotifyException as e: return f"An error occurred with Spotify: {e}. \n\n**Remember to wake up Spotify.**" except Timeout: return f"An unexpected error occurred: {e}." def play_playlist_by_name(playlist_name): """ Plays an existing playlist in the user's library given its name. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" playlists = SP_STATE.value.current_user_playlists() playlist_dict = {playlist['name']: (playlist['id'], playlist['owner']['display_name']) for playlist in playlists['items']} playlist_names = [key for key in playlist_dict.keys()] # defined inside to capture user-specific playlists playlist_name_embeddings = MODEL.encode(playlist_names) user_playlist_embedding = MODEL.encode([playlist_name]) # compares (embedded) given name to (embedded) playlist library and outputs the closest match similarity_scores = cosine_similarity(user_playlist_embedding, playlist_name_embeddings) most_similar_index = similarity_scores.argmax() playlist_name = playlist_names[most_similar_index] try: playlist_id, creator_name = playlist_dict[playlist_name] SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, context_uri=f'spotify:playlist:{playlist_id}') return f'♫ Now playing {playlist_name} by {creator_name} ♫' except: return "Unable to find playlist. Please try again." def get_track_info(): """ Harvests information for explain_track() using Genius' API and basic webscraping. """ if SP_STATE.value is None: return f"{AUTH_MSG}" current_track_item = SP_STATE.value.current_user_playing_track()['item'] track_name = current_track_item['name'] artist_name = current_track_item['artists'][0]['name'] album_name = current_track_item['album']['name'] release_date = current_track_item['album']['release_date'] basic_info = { 'track_name': track_name, 'artist_name': artist_name, 'album_name': album_name, 'release_date': release_date, } # define inside to avoid user conflicts (simultaneously query Genius) genius = lyricsgenius.Genius(GENIUS_TOKEN) # removing feature information from song titles to avoid scewing search track_name = re.split(' \(with | \(feat\. ', track_name)[0] result = genius.search_song(track_name, artist_name) # if no Genius page exists if result is not None and hasattr(result, 'artist'): genius_artist = result.artist.lower().replace(" ", "") spotify_artist = artist_name.lower().replace(" ", "") debug_print(spotify_artist) debug_print(genius_artist) if spotify_artist not in genius_artist: return basic_info, None, None, None else: genius_artist = None return basic_info, None, None, None # if Genius page exists lyrics = result.lyrics url = result.url response = requests.get(url) # parsing the webpage and locating 'About' section soup = BeautifulSoup(response.text, 'html.parser') # universal 'About' section element across all Genius song lyrics pages about_section = soup.select_one('div[class^="RichText__Container-oz284w-0"]') # if no 'About' section exists if not about_section: return basic_info, None, lyrics, url # if 'About' section exists else: about_section = about_section.get_text(separator='\n') return basic_info, about_section, lyrics, url def explain_track(): """ Displays track information in an organized, informational, and compelling manner. Uses the above function. """ # defined inside to avoid circular importing from final_agent import LLM_STATE basic_info, about_section, lyrics, url = get_track_info() debug_print(basic_info, about_section, lyrics, url) if lyrics: # if Genius page exists system_message_content = """ Your task is to create an engaging summary for a track using the available details about the track and its lyrics. If there's insufficient or no additional information besides the lyrics, craft the entire summary based solely on the lyrical content." """ human_message_content = f"{about_section}\n\n{lyrics}" messages = [ SystemMessage(content=system_message_content), HumanMessage(content=human_message_content) ] ai_response = LLM_STATE.value(messages).content summary = f""" **Name:** {basic_info["track_name"]} **Artist:** {basic_info["artist_name"]} **Album:** {basic_info["album_name"]} **Release:** {basic_info["release_date"]} **About:** {ai_response} Click here for more information on Genius! """ return summary else: # if no Genius page exists url = "https://genius.com/Genius-how-to-add-songs-to-genius-annotated" summary = f""" **Name:** {basic_info["track_name"]} **Artist:** {basic_info["artist_name"]} **Album:** {basic_info["album_name"]} **Release:** {basic_info["release_date"]} **About:** Unfortunately, this track has not been uploaded to Genius.com Be the first to change that! """ return summary ### ### ### Genre + Mood ### ### ### def get_user_mood(user_mood): """ Categorizes the user's mood as either 'happy', 'sad', 'energetic', or 'calm'. Uses same cosine similarity/embedding concepts as with determining playlist names. """ if user_mood.lower() in MOOD_LIST: user_mood = user_mood.lower() return user_mood else: user_mood_embedding = MODEL.encode([user_mood.lower()]) similarity_scores = cosine_similarity(user_mood_embedding, MOOD_EMBEDDINGS) most_similar_index = similarity_scores.argmax() user_mood = MOOD_LIST[most_similar_index] return user_mood def get_genre_by_name(genre_name): """ Matches user's desired genre to closest (most similar) existing genre in the list of genres. recommendations() only accepts genres from this list. """ if genre_name.lower() in GENRE_LIST.value: genre_name = genre_name.lower() return genre_name else: genre_name_embedding = MODEL.encode([genre_name.lower()]) similarity_scores = cosine_similarity(genre_name_embedding, GENRE_EMBEDDINGS.value) most_similar_index = similarity_scores.argmax() genre_name = GENRE_LIST.value[most_similar_index] return genre_name def is_genre_match(genre1, genre2, threshold=75): """ Determines if two genres are semantically similar. token_set_ratio() - for quantifying semantic similarity - and threshold of 75 (out of 100) were were arbitrarily determined through basic testing. """ score = fuzz.token_set_ratio(genre1, genre2) debug_print(score) return score >= threshold def create_track_list_str(track_uris): """ Creates an organized list of track names. Used in final return statements by functions below. """ if SP_STATE.value is None: return f"{AUTH_MSG}" track_details = SP_STATE.value.tracks(track_uris) track_names_with_artists = [f"{track['name']} by {track['artists'][0]['name']}" for track in track_details['tracks']] track_list_str = "
".join(track_names_with_artists) return track_list_str def play_genre_by_name_and_mood(genre_name, user_mood): """ 1. Retrieves user's desired genre and current mood. 2. Matches genre and mood to existing options. 3. Gets 4 of user's top artists that align with genre. 4. Conducts personalized recommendations() search. 5. Plays selected track, clears the queue, and adds the rest to the now-empty queue. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" genre_name = get_genre_by_name(genre_name) user_mood = get_user_mood(user_mood).lower() debug_print(genre_name) debug_print(user_mood) # increased personalization user_top_artists = SP_STATE.value.current_user_top_artists(limit=NUM_ARTISTS, time_range=TIME_RANGE) matching_artists_ids = [] for artist in user_top_artists['items']: debug_print(artist['genres']) for artist_genre in artist['genres']: if is_genre_match(genre_name, artist_genre): matching_artists_ids.append(artist['id']) break # don't waste time cycling artist genres after match if len(matching_artists_ids) == MAX_ARTISTS: break if not matching_artists_ids: matching_artists_ids = None else: artist_names = [artist['name'] for artist in SP_STATE.value.artists(matching_artists_ids)['artists']] debug_print(artist_names) debug_print(matching_artists_ids) recommendations = SP_STATE.value.recommendations( # accepts maximum {genre + artists} = 5 seeds seed_artists=matching_artists_ids, seed_genres=[genre_name], seed_tracks=None, limit=NUM_TRACKS, # number of tracks to return country=None, **MOOD_SETTINGS[user_mood]) # maps to mood settings dictionary track_uris = [track['uri'] for track in recommendations['tracks']] track_list_str = create_track_list_str(track_uris) SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, uris=track_uris) return f""" **♫ Now Playing:** {genre_name} ♫ **Selected Tracks:** {track_list_str} """ ### ### ### Artist + Mood ### ### ### def play_artist_by_name_and_mood(artist_name, user_mood): """ Plays tracks (randomly selected) by a given artist that matches the user's mood. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" user_mood = get_user_mood(user_mood).lower() debug_print(user_mood) # retrieving and shuffling all artist's tracks first_name = artist_name.split(',')[0].strip() results = SP_STATE.value.search(q=first_name, type='artist') artist_id = results['artists']['items'][0]['id'] # most recent albums retrieved first artist_albums = SP_STATE.value.artist_albums(artist_id, album_type='album', limit=NUM_ALBUMS) artist_tracks = [] for album in artist_albums['items']: album_tracks = SP_STATE.value.album_tracks(album['id'])['items'] artist_tracks.extend(album_tracks) random.shuffle(artist_tracks) # filtering until we find enough (MAX_TRACKS) tracks that match user's mood selected_tracks = [] for track in artist_tracks: if len(selected_tracks) == MAX_TRACKS: break features = SP_STATE.value.audio_features([track['id']])[0] mood_criteria = MOOD_SETTINGS[user_mood] match = True for criteria, threshold in mood_criteria.items(): if "min_" in criteria and features[criteria[4:]] < threshold: match = False break elif "max_" in criteria and features[criteria[4:]] > threshold: match = False break if match: debug_print(f"{features}\n{mood_criteria}\n\n") selected_tracks.append(track) track_names = [track['name'] for track in selected_tracks] track_list_str = "
".join(track_names) # using HTML line breaks for each track name debug_print(track_list_str) track_uris = [track['uri'] for track in selected_tracks] SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, uris=track_uris) return f""" **♫ Now Playing:** {artist_name} ♫ **Selected Tracks:** {track_list_str} """ ### ### ### Recommendations ### ### ### def recommend_tracks(genre_name=None, artist_name=None, track_name=None, user_mood=None): """ 1. Retrieves user's preferences based on artist_name, track_name, genre_name, and/or user_mood. 2. Uses these parameters to conduct personalized recommendations() search. 3. Returns the track URIs of (NUM_TRACKS) recommendation tracks. """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" user_mood = get_user_mood(user_mood).lower() if user_mood else None debug_print(user_mood) seed_genre, seed_artist, seed_track = None, None, None if genre_name: first_name = genre_name.split(',')[0].strip() genre_name = get_genre_by_name(first_name) seed_genre = [genre_name] debug_print(seed_genre) if artist_name: first_name = artist_name.split(',')[0].strip() # if user provides multiple artists, use the first results = SP_STATE.value.search(q='artist:' + first_name, type='artist') seed_artist = [results['artists']['items'][0]['id']] if track_name: first_name = track_name.split(',')[0].strip() results = SP_STATE.value.search(q='track:' + first_name, type='track') seed_track = [results['tracks']['items'][0]['id']] # if user requests recommendations without specifying anything but their mood # this is because recommendations() requires at least one seed if seed_genre is None and seed_artist is None and seed_track is None: raise ValueError("At least one genre, artist, or track must be provided.") recommendations = SP_STATE.value.recommendations( # passing in 3 seeds seed_artists=seed_artist, seed_genres=seed_genre, seed_tracks=seed_track, limit=NUM_TRACKS, country=None, **MOOD_SETTINGS[user_mood] if user_mood else {}) track_uris = [track['uri'] for track in recommendations['tracks']] return track_uris def play_recommended_tracks(genre_name=None, artist_name=None, track_name=None, user_mood=None): """ Plays the track_uris returned by recommend_tracks(). """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" try: track_uris = recommend_tracks(genre_name, artist_name, track_name, user_mood) track_list_str = create_track_list_str(track_uris) SP_STATE.value.start_playback(device_id=DEVICE_ID_STATE.value, uris=track_uris) return f""" **♫ Now Playing Recommendations Based On:** {', '.join(filter(None, [genre_name, artist_name, track_name, "Your Mood"]))} ♫ **Selected Tracks:** {track_list_str} """ except ValueError as e: return str(e) def create_playlist_from_recommendations(genre_name=None, artist_name=None, track_name=None, user_mood=None): """ Creates a playlist from recommend_tracks(). """ if SP_STATE.value is None or DEVICE_ID_STATE.value is None: return f"{AUTH_MSG}" user = SP_STATE.value.current_user() user_id = user['id'] user_name = user["display_name"] playlists = SP_STATE.value.current_user_playlists() playlist_names = [playlist['name'] for playlist in playlists["items"]] chosen_theme = random.choice(THEMES) playlist_name = f"{user_name}'s {chosen_theme} Playlist" # ensuring the use of new adjective each time while playlist_name in playlist_names: chosen_theme = random.choice(THEMES) playlist_name = f"{user_name}'s {chosen_theme} Playlist" playlist_description=f"Apollo AI's personalized playlist for {user_name}. Get yours here: (add link)." # TODO: add link to project new_playlist = SP_STATE.value.user_playlist_create(user=user_id, name=playlist_name, public=True, collaborative=False, description=playlist_description) track_uris = recommend_tracks(genre_name, artist_name, track_name, user_mood) track_list_str = create_track_list_str(track_uris) SP_STATE.value.user_playlist_add_tracks(user=user_id, playlist_id=new_playlist['id'], tracks=track_uris, position=None) playlist_url = f"https://open.spotify.com/playlist/{new_playlist['id']}" return f""" ♫ Created *{playlist_name}* Based On: {', '.join(filter(None, [genre_name, artist_name, track_name, 'Your Mood']))} ♫ **Selected Tracks:** {track_list_str} Click here to listen to the playlist on Spotify! """