Spaces:
Running
Running
| """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 | |