backend/src/spotify.py

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


# Multiple mood-keyword sets per emotion — one is picked randomly each call
EMOTION_SEARCH_TERMS: dict = {
    "joy":      ["happy upbeat feel-good",      "cheerful bright danceable",   "fun positive energetic",      "joyful lively vibrant"],
    "love":     ["romantic love sweet",          "tender intimate soulful",     "passionate warm affectionate", "dreamy soft romantic"],
    "sadness":  ["sad melancholic emotional",    "heartbreak lonely blue",      "grief somber tearful",        "wistful slow mournful"],
    "anger":    ["angry intense aggressive",     "furious hard-hitting raw",    "rage rebellious fierce",      "heavy loud confrontational"],
    "fear":     ["dark atmospheric tense",       "eerie haunting unsettling",   "sinister cold suspenseful",   "gothic mysterious brooding"],
    "surprise": ["exciting energetic dynamic",   "unexpected bold eclectic",    "euphoric rush powerful",      "electrifying dramatic vivid"],
    "neutral":  ["chill relaxed smooth",         "ambient laid-back mellow",    "focus calm easy-listening",   "soft understated steady"],
    "sarcasm":  ["quirky witty alternative",     "sardonic indie offbeat",      "ironic playful subversive",   "dry wry unconventional"],
}


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)
        )
        self.mapper         = EmotionFeatureMapper()
        self.max_popularity = max_popularity

    def recommend(
        self,
        predict_result: dict,
        limit: int = 10,
        genre_override: Optional[str] = None,
        blend: bool = True,
        raw_text: Optional[str] = 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 query from emotion mood terms and genre seeds
        # Randomly pick a mood-term variant and a search offset for result variety
        term_options = EMOTION_SEARCH_TERMS.get(emotion, [""])
        mood_terms   = random.choice(term_options)
        genre_terms  = " ".join(seed_genres[:2])
        query        = f"{genre_terms} {mood_terms}".strip()
        offset       = random.randint(0, 3) * 10  # 0, 10, 20, or 30

        # NOTE: Spotify limits non-extended-access apps to limit=10 on /search.
        try:
            results = self.sp.search(q=query, type="track", limit=10, offset=offset)
        except Exception as exc:
            raise RuntimeError(f"Spotify API error: {exc}") from exc

        raw_tracks = results.get("tracks", {}).get("items", [])

        # 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

        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
        filtered = [t for t in raw_tracks if t.get("popularity", 101) <= self.max_popularity]
        if not filtered:
            filtered = sorted(raw_tracks, key=lambda t: t.get("popularity", 100))[: limit * 2]

        track_ids    = [t["id"] for t in filtered if t.get("id")]
        features_map = self._fetch_audio_features(track_ids)
        scored       = self._score_tracks(filtered, features_map, target_vector)
        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]

        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:
        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) -> list:
        scored = []
        n = len(tracks)
        for i, track in enumerate(tracks):
            feat = features_map.get(track.get("id"))
            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)))
                match_score = round(max(0.0, 100.0 - dist * 100.0), 1)
            else:
                # Fallback: use search-rank position (Spotify returns results by relevance).
                # Rank 0 = 95%, rank n-1 = 50%, spread linearly.
                match_score = round(95.0 - (i / max(n - 1, 1)) * 45.0, 1)

            images = (track.get("album") or {}).get("images", [])
            scored.append({
                "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,
                "audio_features": {
                    k: feat.get(k)
                    for k in ["valence", "energy", "danceability", "tempo", "mode", "acousticness", "speechiness"]
                } if feat else None,
            })
        scored.sort(key=lambda x: x["match_score"], reverse=True)
        return 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