# 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!
"""