"""Spotify integration -- emotion-to-feature mapping and track recommendations.""" import os import json import math import random import urllib.request import urllib.parse from typing import Optional from .logger import tracker # Search terms per emotion — use genre/scene/artist-style descriptors rather than # pure mood adjectives, because mood adjectives also appear in stock music titles. EMOTION_SEARCH_TERMS: dict = { "joy": ["indie pop anthems", "disco funk dance", "britpop power pop", "afrobeat summer"], "love": ["neo soul r&b", "soft rock romance", "indie folk love", "soul ballads"], "sadness": ["slowcore sadcore", "emo indie heartbreak", "folk singer songwriter", "shoegaze dream"], "anger": ["hardcore punk thrash", "metal rage", "post hardcore noise", "industrial rock"], "fear": ["witch house darkwave", "post punk goth", "cold wave minimal", "drone doom dark"], "surprise": ["art pop experimental", "synth pop new wave", "glitch indie eclectic", "progressive indie"], "neutral": ["lo fi indie bedroom", "chillwave dream pop", "ambient indie folk", "slacker indie"], "sarcasm": ["art rock wry", "indie new wave quirky","post punk ironic", "college rock offbeat"], } EMOTION_TARGETS: dict = { "joy": {"valence": 0.85, "energy": 0.82, "danceability": 0.80, "tempo": 128, "mode": 1, "genres": ["pop", "dance", "happy"]}, "love": {"valence": 0.75, "energy": 0.52, "danceability": 0.62, "tempo": 96, "mode": 1, "genres": ["romance", "r-n-b", "soul"]}, "sadness": {"valence": 0.18, "energy": 0.28, "danceability": 0.32, "tempo": 72, "mode": 0, "genres": ["sad", "indie", "acoustic"]}, "anger": {"valence": 0.18, "energy": 0.88, "danceability": 0.58, "tempo": 148, "mode": 0, "genres": ["metal", "hard-rock", "punk"]}, "fear": {"valence": 0.22, "energy": 0.52, "danceability": 0.38, "tempo": 88, "mode": 0, "genres": ["ambient", "goth", "trip-hop"]}, "surprise": {"valence": 0.62, "energy": 0.80, "danceability": 0.70, "tempo": 138, "mode": 1, "genres": ["pop", "electronic", "edm"]}, "neutral": {"valence": 0.50, "energy": 0.38, "danceability": 0.50, "tempo": 100, "mode": 1, "genres": ["chill", "study", "indie-pop"]}, "sarcasm": {"valence": 0.45, "energy": 0.60, "danceability": 0.55, "tempo": 110, "mode": 0, "genres": ["alternative", "indie", "emo"]}, } class EmotionFeatureMapper: """Converts emotion classifier output to Spotify audio feature targets. Blends the top-2 emotions weighted by their probabilities. Uses confidence to set tight/loose constraint windows around targets. """ def map(self, probabilities: dict, confidence: float) -> dict: """Map emotion probabilities to Spotify target parameters. Args: probabilities: {emotion: probability} from EmotionPredictor confidence: Top emotion confidence (0-1) Returns: Dict with target_* params, min/max_valence window, seed_genres, and _target_vector (internal, stripped before Spotify API call). """ sorted_emotions = sorted(probabilities.items(), key=lambda x: x[1], reverse=True) top1_name, top1_prob = sorted_emotions[0] top2_name, top2_prob = sorted_emotions[1] if len(sorted_emotions) > 1 else (top1_name, 0.0) t1 = EMOTION_TARGETS.get(top1_name, EMOTION_TARGETS["neutral"]) t2 = EMOTION_TARGETS.get(top2_name, EMOTION_TARGETS["neutral"]) total = top1_prob + top2_prob w1 = top1_prob / total if total > 0 else 1.0 w2 = top2_prob / total if total > 0 else 0.0 bv = round(w1 * t1["valence"] + w2 * t2["valence"], 3) be = round(w1 * t1["energy"] + w2 * t2["energy"], 3) bd = round(w1 * t1["danceability"] + w2 * t2["danceability"], 3) bt = round(w1 * t1["tempo"] + w2 * t2["tempo"]) bm = round(w1 * t1["mode"] + w2 * t2["mode"]) window = 0.08 if confidence > 0.75 else (0.15 if confidence > 0.45 else 0.25) return { "target_valence": bv, "target_energy": be, "target_danceability": bd, "target_tempo": bt, "target_mode": bm, "min_valence": round(max(0.0, bv - window), 3), "max_valence": round(min(1.0, bv + window), 3), "seed_genres": t1["genres"][:2], "_target_vector": [bv, be, bd], } class SpotifyRecommender: """Fetches song recommendations from Spotify based on emotion analysis. Credentials are read from env vars: SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET Anti-popularity bias: max_popularity (default 65) avoids chart-toppers. """ def __init__( self, client_id: Optional[str] = None, client_secret: Optional[str] = None, max_popularity: int = 65, ): try: import spotipy from spotipy.oauth2 import SpotifyClientCredentials except ImportError: raise ImportError( "spotipy is required.\n" "Install: pip install spotipy>=2.23.0" ) _id = client_id or os.environ.get("SPOTIFY_CLIENT_ID") _secret = client_secret or os.environ.get("SPOTIFY_CLIENT_SECRET") if not _id or not _secret: raise ValueError( "Spotify credentials not found.\n" "Copy .env.example -> .env and fill in SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET." ) self.sp = spotipy.Spotify( auth_manager=SpotifyClientCredentials(client_id=_id, client_secret=_secret), requests_timeout=8, ) self.mapper = EmotionFeatureMapper() self.max_popularity = max_popularity try: from .genius import GeniusLyricsFetcher self.lyrics_fetcher = GeniusLyricsFetcher() except ImportError: self.lyrics_fetcher = None def get_reference_emotion(self, url: str, predictor) -> Optional[dict]: """Extract the emotional probability distribution of a reference track via its lyrics.""" import re match = re.search(r"track/([a-zA-Z0-9]+)", url) if not match: print("⚠️ Invalid Spotify URL.") return None track_id = match.group(1) try: track = self.sp.track(track_id) title = track.get("name") artists = track.get("artists", []) # Use primary artist only — multi-artist strings (e.g. "Gotye, Kimbra") break Genius search artist = artists[0]["name"] if artists else "" # Check cache from .cache import cache if cache: cached = cache.get_score(title, artist) if cached: return cached # Fetch lyrics lyrics = self.lyrics_fetcher.get_lyrics(title, artist) if not lyrics: print(f"⚠️ No lyrics for '{title}' — trying audio features.") probs = self._audio_features_to_probs(track_id) if probs: return probs # Last resort: run emotion model on track title + artist print(f"⚠️ Audio features unavailable — inferring from track title.") pred = predictor.predict(f"{title} {artist}", create_chart=False) return pred.get("probabilities") or None # Predict emotion pred = predictor.predict(lyrics[:3000], create_chart=False) probs = pred.get("probabilities", {}) if cache and probs: cache.set_score(title, artist, probs) return probs except Exception as e: print(f"⚠️ Failed to extract emotion for reference track: {e}") return None def _audio_features_to_probs(self, track_id: str) -> Optional[dict]: """Approximate emotion probabilities from a track's Spotify audio features.""" try: feats = self.sp.audio_features([track_id]) if not feats or feats[0] is None: return None f = feats[0] v = f.get("valence", 0.5) e = f.get("energy", 0.5) d = f.get("danceability", 0.5) # Russell's Circumplex: valence = pleasantness, energy = arousal raw = { "joy": v * e * d * 2.0, "love": v * (1 - e) * 1.2, "sadness": (1 - v) * (1 - e) * 1.5, "anger": (1 - v) * e * 1.4, "fear": (1 - v) * (0.3 + e * 0.4), "surprise": v * e * (1 - d) * 1.3, "neutral": (1 - abs(v - 0.5) * 2) * (1 - abs(e - 0.5) * 2) * 0.8, "sarcasm": 0.03, } total = sum(raw.values()) or 1.0 return {k: round(val / total, 4) for k, val in raw.items()} except Exception as ex: print(f"⚠️ Audio-feature fallback failed: {ex}") return None def recommend( self, predict_result: dict, limit: int = 10, genre_override: Optional[str] = None, blend: bool = True, raw_text: Optional[str] = None, predictor = None, ) -> dict: """Recommend songs based on an EmotionPredictor result. Args: predict_result: dict from EmotionPredictor.predict() limit: Number of tracks to return (default 10) genre_override: Override seed genres with a single genre string blend: If False, use only top-1 emotion (no blending) Returns: dict with emotion summary, targets used, and ranked track list """ emotion = predict_result["emotion"] confidence = predict_result["confidence"] probabilities = predict_result["probabilities"] if not blend: top = max(probabilities, key=probabilities.get) probabilities = {top: 1.0} targets = self.mapper.map(probabilities, confidence) target_vector = targets.pop("_target_vector") seed_genres = [genre_override] if genre_override else targets.pop("seed_genres") api_params = {k: v for k, v in targets.items() if k.startswith(("target_", "min_", "max_"))} # Build search queries — 4 natural-language variants per emotion. # Threads cycle through all variants so results come from different queries. import time as _time t0 = _time.monotonic() term_options = EMOTION_SEARCH_TERMS.get(emotion, ["pop music"]) queries = term_options base_offset = random.randint(0, 2) * 10 # small randomisation per call # NOTE: Spotify limits non-extended-access apps to limit=10 on /search. # We fetch 100 tracks by distributing 10 threads across the 4 query variants. from concurrent.futures import ThreadPoolExecutor, as_completed raw_tracks = [] def fetch_page(thread_idx): q = queries[thread_idx % len(queries)] page_off = (thread_idx // len(queries)) * 10 + base_offset res = self.sp.search(q=q, type="track", limit=10, offset=page_off, market="US") return res.get("tracks", {}).get("items", []) errors = [] with ThreadPoolExecutor(max_workers=10) as ex: futures = [ex.submit(fetch_page, i) for i in range(10)] for future in as_completed(futures): try: raw_tracks.extend(future.result()) except Exception as exc: errors.append(str(exc)) print(f"⏱️ search done in {_time.monotonic()-t0:.1f}s ({len(raw_tracks)} raw tracks, {len(errors)} errors)") if errors: print(f"⚠️ {len(errors)}/10 search threads failed: {errors[0][:120]}") if not raw_tracks and errors: raise RuntimeError(f"All Spotify search threads failed: {errors[0]}") # Spotify search results often omit or zero-out the popularity field. # Fetch full track objects in one batched call to get real popularity values. if raw_tracks: try: ids = [t["id"] for t in raw_tracks if t.get("id")] pop_map: dict = {} for i in range(0, len(ids), 50): full = self.sp.tracks(ids[i:i+50]) for ft in (full.get("tracks") or []): if ft and ft.get("id") and ft.get("popularity") is not None: pop_map[ft["id"]] = ft["popularity"] for t in raw_tracks: if t.get("id") in pop_map: t["popularity"] = pop_map[t["id"]] except Exception as e: print(f"⚠️ Popularity fetch failed: {e}") # Prefer tracks that have album art — niche searches often return library # tracks with empty image arrays; filter them out unless it empties the pool with_art = [t for t in raw_tracks if (t.get("album") or {}).get("images")] if with_art: raw_tracks = with_art # Remove compilation/playlist/stock-music entries (not real songs) _COMP_PATTERNS = ("top hits", "top songs", "best of", "greatest hits", "various artists", "now that's", "playlist", "soundtrack to", "royalty free", "royalty-free", "background music", "meditation", "relaxation", "sleep music", "study music", "focus music", "instrumental", "lofi beats", "lo-fi beats") _STOCK_TITLE_WORDS = {"upbeat", "energetic", "happy", "uplifting", "positive", "relaxing", "calm", "peaceful", "motivational", "inspiring", "cheerful", "joyful", "soothing", "ambient", "emotional", "epic", "trailer", "cinematic", "orchestral", "piano", "violin", "guitar", "beats", "tunes", "music", "songs", "nature", "bubbling", "trickling", "flowing", "water", "rain", "waves", "forest", "sleep", "meditation", "eerie", "spooky", "atmospheric", "atmosphere", "mysterious", "horror", "scary", "terrifying", "zombie", "haunted"} def _is_stock(track): name = track.get("name", "").lower() # Artist name is a keyword list (comma-separated moods/genres = library music) artists = track.get("artists", []) artist_name = artists[0].get("name", "") if artists else "" if artist_name.count(",") >= 1: return True # Artist or title pattern match if any(p in name or p in artist_name.lower() for p in _COMP_PATTERNS): return True # Keyword-stuffed title: 3+ stock adjectives in a short name words = set(name.split()) if len(words) <= 7 and len(words & _STOCK_TITLE_WORDS) >= 3: return True return False real_tracks = [t for t in raw_tracks if not _is_stock(t)] if real_tracks: raw_tracks = real_tracks # Prevent title-matched genre-word garbage (e.g. tracks literally named "Alternative") genre_words = [g.lower() for g in seed_genres] title_filtered = [] for t in raw_tracks: title_lower = t.get("name", "").lower() if not any(g in title_lower for g in genre_words): title_filtered.append(t) # Only apply the title filter if it leaves us with at least some tracks if len(title_filtered) >= limit: raw_tracks = title_filtered # Deduplicate by track ID AND by (artist, normalised-name) to collapse # "Song (Single Version)", "Song (Remastered)", "Song (Club Mix)" etc. import re as _re _VER_RE = _re.compile( r"\s*[\(\[](single|album|radio|remaster|remastered|live|acoustic|" r"club mix|remix|extended|instrumental|demo|version|edit|mix)[^\)\]]*[\)\]]", _re.I, ) seen_ids: set = set() seen_name_artist: set = set() deduped = [] for t in raw_tracks: tid = t.get("id") if not tid or tid in seen_ids: continue seen_ids.add(tid) name_norm = _VER_RE.sub("", t.get("name", "")).strip().lower() artist_key = t.get("artists", [{}])[0].get("name", "").lower() na_key = (name_norm, artist_key) if na_key in seen_name_artist: continue seen_name_artist.add(na_key) deduped.append(t) raw_tracks = deduped if not raw_tracks: return { "emotion": emotion, "confidence": confidence, "probabilities": predict_result.get("probabilities", {}), "explanation": predict_result.get("explanation", ""), "targets_used": {**api_params, "seed_genres": seed_genres}, "tracks": [], } # Popularity filter — prefer hidden gems but exclude truly obscure stock tracks # Floor of 15 removes zero-coverage tracks; ceiling keeps it non-mainstream def _pop(t): p = t.get("popularity") return p if isinstance(p, int) else 0 MIN_POP = 20 filtered = [t for t in raw_tracks if MIN_POP <= _pop(t) <= self.max_popularity] if not filtered: # Widen: accept pop>=10 before giving up on the ceiling entirely filtered = [t for t in raw_tracks if 10 <= _pop(t) <= self.max_popularity] if not filtered: # Last resort: just sort by popularity descending and take the best ones filtered = sorted(raw_tracks, key=_pop, reverse=True)[: limit * 2] # Cap the pool fed to _score_tracks — lyrics.ovh calls are per-track and # HF Spaces has a ~60s request timeout, so keep this tight. MAX_TO_SCORE = 20 filtered = filtered[:MAX_TO_SCORE] t1 = _time.monotonic() track_ids = [t["id"] for t in filtered if t.get("id")] features_map = self._fetch_audio_features(track_ids) print(f"⏱️ audio_features done in {_time.monotonic()-t1:.1f}s ({len(filtered)} tracks to score)") t2 = _time.monotonic() scored = self._score_tracks(filtered, features_map, target_vector, predict_result.get("probabilities", {}), predictor) print(f"⏱️ scoring done in {_time.monotonic()-t2:.1f}s") diverse = self._diversity_filter(scored, limit=limit, max_per_artist=2) # If input looks like lyrics, pin the source song at position 0 pinned: list = [] if raw_text: src = self._find_source_track(raw_text) if src: src_url = src.get("external_urls", {}).get("spotify") diverse = [t for t in diverse if t.get("spotify_url") != src_url] images = src.get("album", {}).get("images", []) pinned = [{ "name": src["name"], "artist": ", ".join(a["name"] for a in src.get("artists", [])), "album": src.get("album", {}).get("name", ""), "album_art": images[0]["url"] if images else None, "spotify_url": src_url, "preview_url": src.get("preview_url"), "popularity": src.get("popularity"), "match_score": 100.0, "audio_features": None, "is_source": True, }] tracks = (pinned + diverse)[:limit] try: tracker.log_request( user_text=raw_text or "No text provided", emotion=emotion, confidence=confidence, initial_100=raw_tracks, scored_ranking=scored, final_top_10=tracks ) except Exception as e: print(f"⚠️ Failed to log to TrackerDB: {e}") return { "emotion": emotion, "confidence": confidence, "probabilities": predict_result.get("probabilities", {}), "explanation": predict_result.get("explanation", ""), "music_context": predict_result.get("music_context", {}), "targets_used": {**api_params, "seed_genres": seed_genres}, "tracks": tracks, } def _find_source_track(self, raw_text: str) -> Optional[dict]: """Identify the song the input text is from. Primary: Musixmatch lyrics search (same method Spotify uses internally), then looks the result up on Spotify. Fallback: heuristic title-word matching against a direct Spotify search. """ mm_key = os.environ.get("MUSIXMATCH_API_KEY") if mm_key: track = self._musixmatch_lyrics_search(raw_text, mm_key) if track: return track # Fallback: Spotify search + title-word / 5-word-window heuristics try: results = self.sp.search(q=raw_text[:150], type="track", limit=5) candidates = results.get("tracks", {}).get("items", []) or [] except Exception: return None tokens = [w.lower().strip(".,!?\"'-") for w in raw_text.split() if w.strip(".,!?\"'-")] input_set = set(tokens) windows = [" ".join(tokens[i:i + 5]) for i in range(len(tokens) - 4)] for track in candidates: title_words = [ w.lower().strip(".,!?\"'-") for w in track.get("name", "").split() if len(w.strip(".,!?\"'-")) >= 2 ] if len(title_words) >= 2 and all(w in input_set for w in title_words): return track haystack = " ".join([ track.get("name", "").lower(), *[a["name"].lower() for a in track.get("artists", [])], ]) if windows and any(window in haystack for window in windows): return track return None def _musixmatch_lyrics_search(self, raw_text: str, api_key: str) -> Optional[dict]: """Query Musixmatch for a song matching the lyrics, then look it up on Spotify.""" params = urllib.parse.urlencode({ "q_lyrics": raw_text[:200], "apikey": api_key, "page_size": 3, "s_track_rating": "desc", "format": "json", }) try: req = urllib.request.Request( f"https://api.musixmatch.com/ws/1.1/track.search?{params}", headers={"User-Agent": "eumora/1.0"}, ) with urllib.request.urlopen(req, timeout=5) as r: data = json.loads(r.read()) track_list = ( data.get("message", {}) .get("body", {}) .get("track_list", []) ) if not track_list: return None top = track_list[0]["track"] title = top["track_name"] artist = top["artist_name"] except Exception: return None # Look up the identified song on Spotify try: results = self.sp.search( q=f'track:"{title}" artist:"{artist}"', type="track", limit=1, ) items = results.get("tracks", {}).get("items") or [] return items[0] if items else None except Exception: return None def _fetch_audio_features(self, track_ids: list) -> dict: # audio_features endpoint is deprecated — skip it on HF Spaces to save time on_hf = bool(os.environ.get("SPACE_ID") or os.environ.get("SPACES_ZERO_GPU")) if on_hf: return {} features_map: dict = {} try: for i in range(0, len(track_ids), 100): for feat in (self.sp.audio_features(track_ids[i : i + 100]) or []): if feat and feat.get("id"): features_map[feat["id"]] = feat except Exception: pass # audio-features may be deprecated; degrade gracefully return features_map def _score_tracks(self, tracks: list, features_map: dict, target_vector: list, user_probs: dict = None, predictor = None) -> list: scored = [None] * len(tracks) n = len(tracks) import threading from concurrent.futures import ThreadPoolExecutor, as_completed try: from .cache import cache except ImportError: cache = None # On HF Spaces (free CPU), DeBERTa inference takes 20-30s each and causes timeouts. # Disable lyrics scoring there — popularity fallback only. _on_hf = bool(os.environ.get("SPACE_ID") or os.environ.get("SPACES_ZERO_GPU")) _deberta_budget = [0 if _on_hf else 3] _deberta_lock = threading.Lock() def process_track(i, track): feat = features_map.get(track.get("id")) audio_match = None if feat: tv = [feat.get("valence", 0.5), feat.get("energy", 0.5), feat.get("danceability", 0.5)] dist = math.sqrt(sum((a - b) ** 2 for a, b in zip(target_vector, tv))) audio_match = round(max(0.0, 100.0 - dist * 100.0), 1) # --- NEW LYRICS SCORING --- lyrics_match = None lyric_probs = None if predictor and self.lyrics_fetcher and user_probs: title = track.get("name") track_artists = track.get("artists", []) # Primary artist for Genius search (full string breaks search for feat. tracks) search_artist = track_artists[0]["name"] if track_artists else "" cache_key_artist = ", ".join(a["name"] for a in track_artists) # Check cache if cache: lyric_probs = cache.get_score(title, cache_key_artist) if not lyric_probs: # Pre-reserve a DeBERTa slot before fetching lyrics. # This limits both the lyrics network calls AND the inference # to at most _deberta_budget tracks — prevents 20 concurrent # cloudscraper/lyrics.ovh calls from piling up. reserved = False with _deberta_lock: if _deberta_budget[0] > 0: _deberta_budget[0] -= 1 reserved = True if reserved: lyrics = self.lyrics_fetcher.get_lyrics(title, search_artist) if lyrics: try: lyric_pred = predictor.predict(lyrics[:3000], create_chart=False) lyric_probs = lyric_pred.get("probabilities", {}) if cache: cache.set_score(title, cache_key_artist, lyric_probs) except Exception as e: print(f"⚠️ Failed to predict lyrics emotion for {title}: {e}") else: # No lyrics found — refund the slot so other tracks can try with _deberta_lock: _deberta_budget[0] += 1 if lyric_probs: # Calculate Euclidean distance between user_probs and lyric_probs dist_sq = 0.0 for em in user_probs.keys(): u_p = user_probs.get(em, 0.0) l_p = lyric_probs.get(em, 0.0) dist_sq += (u_p - l_p) ** 2 lyric_dist = math.sqrt(dist_sq) # max dist is ~1.414, normalize to 0-100 lyrics_match = round(max(0.0, 100.0 - (lyric_dist / 1.414) * 100.0), 1) # Combine scores or fallback if lyrics_match is not None: match_score = lyrics_match elif audio_match is not None: match_score = audio_match else: # Fallback: use popularity as a proxy when no lyric/audio data is available. # Scale Spotify popularity (0-100) to a 20-44 range so lyrics-scored # tracks always rank above unscored ones (lyrics_match starts at ~50+). pop = track.get("popularity") or 30 match_score = round(20.0 + (pop / 100.0) * 24.0, 1) images = (track.get("album") or {}).get("images", []) return i, { "name": track["name"], "artist": ", ".join(a["name"] for a in track.get("artists", [])), "album": track.get("album", {}).get("name", ""), "album_art": images[0]["url"] if images else None, "spotify_url": track.get("external_urls", {}).get("spotify"), "preview_url": track.get("preview_url"), "popularity": track.get("popularity"), "match_score": match_score, "lyrics_match_score": lyrics_match, "lyrical_emotion": lyric_probs, "audio_features": { k: feat.get(k) for k in ["valence", "energy", "danceability", "tempo", "mode", "acousticness", "speechiness"] } if feat else None, } with ThreadPoolExecutor(max_workers=20) as executor: futures = {executor.submit(process_track, i, track): i for i, track in enumerate(tracks)} for future in as_completed(futures): idx, result = future.result() scored[idx] = result final_scored = [s for s in scored if s is not None] final_scored.sort(key=lambda x: x["match_score"], reverse=True) return final_scored def _diversity_filter(self, scored: list, limit: int, max_per_artist: int) -> list: seen: dict = {} result: list = [] for track in scored: if seen.get(track["artist"], 0) < max_per_artist: result.append(track) seen[track["artist"]] = seen.get(track["artist"], 0) + 1 if len(result) >= limit: break return result