Spaces:

VivDubs
/

eumora-api

Running

GitHub Action
deploy 3d50547
037f340 2 days ago
24.5 kB
"""EUMORA REST API -- FastAPI backend for the frontend."""

import os
import sys
import threading
import time as _time
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeout
from contextlib import asynccontextmanager, contextmanager
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from dotenv import load_dotenv

load_dotenv()
sys.path.insert(0, str(Path(__file__).parent))

# Global socket timeout covers ALL network calls including Spotify's OAuth token
# requests which bypass spotipy's requests_timeout parameter.
import socket as _socket
_socket.setdefaulttimeout(10)

# CORS: comma-separated origins, or "*" for development
# e.g. ALLOWED_ORIGINS=https://yourapp.vercel.app,https://yourapp.com
_raw_origins = os.environ.get("ALLOWED_ORIGINS", "*")
ALLOWED_ORIGINS = [o.strip() for o in _raw_origins.split(",")] if _raw_origins != "*" else ["*"]


# ---------------------------------------------------------------------------
# Request models
# ---------------------------------------------------------------------------

class PredictRequest(BaseModel):
    text: Optional[str] = None
    # raw=True: plain softmax as the HF router returns it, for the Vercel
    # emotion route's fallback when the router is unavailable.
    raw: bool = False
    reference_url: Optional[str] = None
    target_sarcasm_prior: float = Field(0.15, gt=0.0, lt=1.0)
    disable_prior_adjustment: bool = False

    model_config = {"str_max_length": 2000}  # prevent oversized payloads


class RecommendRequest(BaseModel):
    text: Optional[str] = None
    reference_url: Optional[str] = None
    limit: int = 10
    genre: Optional[str] = None
    blend: bool = True
    max_popularity: int = 65
    target_sarcasm_prior: float = Field(0.15, gt=0.0, lt=1.0)
    disable_prior_adjustment: bool = False


class ScoreRequest(BaseModel):
    title: str
    artist: str
    user_probabilities: dict
    # Optional: a preview the client already has (iTunes-sourced tracks carry
    # one). Saves a search; only allow-listed preview hosts are fetched.
    preview_url: Optional[str] = None


# ---------------------------------------------------------------------------
# App lifecycle -- load heavy models once at startup
# ---------------------------------------------------------------------------

_predictor   = None
_recommender = None

# Acoustic analysis (VGGish + valence/arousal head). Loaded in the background
# so a slow weight load can never delay startup or /health, and optional: with
# it unavailable, /api/score behaves exactly as it did before audio existed.
_acoustic = None
_acoustic_status = "disabled"
# Four-modality VAD fusion (Objective 8): a 32 KB JSON of linear models, loaded
# with the app. When present and approved by its ship rule, it produces the
# match score; otherwise scoring falls back to the 50/50 lyrics + audio mix.
_fusion = None
_audio_cache: "OrderedDict[tuple, dict]" = OrderedDict()
_audio_lock = threading.Lock()
# Light threads that wait on the acoustic worker processes (src/acoustic_worker)
# and fill the cache. The compute runs in niced processes, and only when no
# lyric work is in flight (below): on a Space capped at 2 CPUs, DeBERTa is the
# scoring bottleneck, and the sound must never slow it down.
_audio_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="audio")
AUDIO_CACHE_SIZE = 2000
# Audio is waited for only until this long after the request arrived, so it
# can never make a response slower than the lyric path already is. Analysis
# that misses the deadline keeps running, is cached, and the response says
# `audio_pending` so the client can ask again once, cheaply.
AUDIO_BUDGET_S = 7.0
NO_PREVIEW_TTL_S = 6 * 3600    # the catalogue changes slowly
ERROR_TTL_S = 300              # rate limits and timeouts clear quickly

# Audio yields to lyrics. The Space runs under a 2-CPU cgroup quota
# (cpu.max 200000/100000), where process priorities inside the container do
# little, so the API schedules it itself: an analysis starts only when no
# request is in its lyric phase, or after AUDIO_GATE_MAX_S under continuous
# load. Measured with audio held back this way, a burst of 10 new tracks
# returns exactly as fast as lyrics alone. A lone request still gets its sound
# within the budget: its own lyric phase ends in a second or two.
_lyric_active = 0
_lyric_idle = threading.Condition()
AUDIO_GATE_MAX_S = 30.0


@contextmanager
def _lyric_phase():
    global _lyric_active
    with _lyric_idle:
        _lyric_active += 1
    try:
        yield
    finally:
        with _lyric_idle:
            _lyric_active -= 1
            if _lyric_active == 0:
                _lyric_idle.notify_all()


# Lyric distributions per track: up to five DeBERTa passes each, and the same
# for every listener, so a repeat request (or the client's audio retry) costs
# nothing. Tracks with no lyrics are remembered briefly to skip re-fetching.
_lyric_cache: "OrderedDict[tuple, dict]" = OrderedDict()
_lyric_lock = threading.Lock()
LYRIC_CACHE_SIZE = 4000
NO_LYRICS_TTL_S = 600


def _load_acoustic(attempts: int = 4, backoff_s: float = 30.0):
    """Load VGGish and the head, retrying: the weights may need downloading if
    the image was built without them, and a network blip should not leave the
    Space lyric-only until its next restart."""
    global _acoustic, _acoustic_status
    for attempt in range(1, attempts + 1):
        _acoustic_status = "loading"
        try:
            from src.acoustic_worker import ProcessAnalyzer
            _acoustic = ProcessAnalyzer(workers=2, nice=10)
            _acoustic_status = "ready"
            print("Acoustic analyzer ready.")
            return
        except Exception as e:  # never fatal: lyric scoring still works
            _acoustic_status = f"error: {type(e).__name__}: {str(e)[:160]}"
            print(f"Warning: acoustic load attempt {attempt}/{attempts} failed -- {_acoustic_status}")
            if attempt < attempts:
                _time.sleep(backoff_s * attempt)


def _track_key(title: str, artist: str) -> tuple:
    return (title.strip().lower(), artist.strip().lower())


def _lyric_probs_for(title: str, artist: str) -> tuple:
    """(lyric distribution or None, whether lyrics were found, whether they
    read as English), cached."""
    key = _track_key(title, artist)
    now = _time.time()
    with _lyric_lock:
        hit = _lyric_cache.get(key)
        if hit and (hit["probs"] is not None or now < hit["retry_after"]):
            _lyric_cache.move_to_end(key)
            return hit["probs"], hit["found"], hit.get("english", True)

    lyrics = None
    lyrics_fetcher = getattr(_recommender, "lyrics_fetcher", None) if _recommender else None
    if lyrics_fetcher:
        try:
            lyrics_raw = lyrics_fetcher.get_lyrics(title, artist)
            if lyrics_raw:
                lyrics = lyrics_raw[:3000].strip()
        except Exception as e:
            print(f"Lyrics fetch failed for '{title}' by '{artist}': {e}")

    probs, english = None, True
    if lyrics:
        english = _lyrics_language_ok(lyrics)
        try:
            probs = _chunked_lyric_probs(_predictor, lyrics)
        except Exception as e:
            print(f"Chunked scoring failed for '{title}': {e}")
            return None, True, english  # transient: do not cache
    with _lyric_lock:
        _lyric_cache[key] = {"probs": probs, "found": bool(lyrics), "english": english,
                             "retry_after": now + NO_LYRICS_TTL_S}
        while len(_lyric_cache) > LYRIC_CACHE_SIZE:
            _lyric_cache.popitem(last=False)
    return probs, bool(lyrics), english


def _audio_for(title: str, artist: str, preview_url: Optional[str]) -> Optional[dict]:
    """Cached acoustic analysis of a track's preview, or None."""
    key = _track_key(title, artist)
    now = _time.time()
    with _audio_lock:
        hit = _audio_cache.get(key)
        if hit and (hit["value"] is not None or now < hit["retry_after"]):
            _audio_cache.move_to_end(key)
            return hit["value"]
    with _lyric_idle:
        _lyric_idle.wait_for(lambda: _lyric_active == 0, timeout=AUDIO_GATE_MAX_S)
    try:
        value = _acoustic.analyze_track(title, artist, preview_url)
        retry_after = _time.time() + NO_PREVIEW_TTL_S
    except Exception as e:
        print(f"Audio analysis failed for '{title}' by '{artist}': {type(e).__name__}: {e}")
        value, retry_after = None, _time.time() + ERROR_TTL_S
    with _audio_lock:
        _audio_cache[key] = {"value": value, "retry_after": retry_after}
        while len(_audio_cache) > AUDIO_CACHE_SIZE:
            _audio_cache.popitem(last=False)
    return value


@asynccontextmanager
async def lifespan(app: FastAPI):
    global _predictor, _recommender

    from src.predict import EmotionPredictor
    from src.spotify import SpotifyRecommender

    print("Loading emotion model...")
    _predictor = EmotionPredictor(enable_viz=False)
    print("Emotion model ready.")

    try:
        _recommender = SpotifyRecommender()
        print("Spotify recommender ready.")
    except (ValueError, ImportError) as e:
        print(f"Warning: Spotify disabled -- {e}")
        _recommender = None

    if os.environ.get("EUMORA_AUDIO", "1") != "0":
        threading.Thread(target=_load_acoustic, name="acoustic-load", daemon=True).start()

    global _fusion
    try:
        from src.fusion import FUSION_PATH, FusionModel
        if FUSION_PATH.exists() and os.environ.get("EUMORA_FUSION", "1") != "0":
            candidate = FusionModel.load()
            _fusion = candidate if candidate.spec.get("ship_as_score") else None
            print(f"VAD fusion {'active' if _fusion else 'present but not approved to score'}.")
    except Exception as e:
        print(f"Warning: VAD fusion disabled -- {type(e).__name__}: {e}")
        _fusion = None

    yield

    # Stop the acoustic worker processes with the server; left running, they
    # outlive it as orphans.
    pool = getattr(_acoustic, "pool", None)
    if pool is not None:
        pool.shutdown(wait=False, cancel_futures=True)
    _predictor   = None
    _recommender = None


app = FastAPI(title="EUMORA API", version="1.0.0", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_methods=["POST", "GET"],
    allow_headers=["*"],
)


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------

def _cpu_info() -> dict:
    """What the container can actually use; os.cpu_count() reports the host."""
    info = {"cpu_count": os.cpu_count()}
    try:
        info["affinity"] = len(os.sched_getaffinity(0))
    except AttributeError:
        pass
    try:
        info["cgroup_cpu_max"] = Path("/sys/fs/cgroup/cpu.max").read_text().strip()
    except OSError:
        pass
    try:
        import torch
        info["torch_threads"] = torch.get_num_threads()
    except Exception:
        pass
    return info


@app.get("/")
def root():
    """What this service is, for anyone who opens the Space page itself."""
    return {
        "service": "EUMORA API",
        "app": "https://eumora.vercel.app",
        "status": "ok" if _predictor else "starting",
        "endpoints": {
            "GET /health": "model, audio and fusion status",
            "POST /api/score": "match a track to a listener's emotion (lyrics + sound, calibrated VAD)",
            "POST /api/predict": "emotion distribution for a text (raw=true: plain softmax)",
        },
        "acoustic": _acoustic_status,
        "fusion": "active" if _fusion is not None else "off",
    }


@app.get("/health")
def health():
    import time as _t
    genius_ok = (
        _recommender is not None
        and _recommender.lyrics_fetcher is not None
        and _recommender.lyrics_fetcher.genius is not None
    )
    on_hf = bool(os.environ.get("SPACE_ID") or os.environ.get("SPACES_ZERO_GPU"))

    # Quick DeBERTa warmup timing (first call cold, this measures it)
    deberta_ms = None
    if _predictor:
        t0 = _t.monotonic()
        _predictor.predict("test", create_chart=False)
        deberta_ms = round((_t.monotonic() - t0) * 1000)

    return {
        "status": "ok",
        "version": os.environ.get("COMMIT_SHA", "dea7be4"),
        "on_hf": on_hf,
        "spotify": _recommender is not None,
        "genius": genius_ok,
        "deberta_warmup_ms": deberta_ms,
        "acoustic": _acoustic_status,
        "fusion": "active" if _fusion is not None else "off",
        "cpu": _cpu_info(),
    }


@app.get("/health/genius")
def health_genius():
    """Attempt a live Genius search to verify the token and connectivity."""
    import os
    token = os.environ.get("GENIUS_ACCESS_TOKEN")
    if not token:
        return {"status": "error", "reason": "GENIUS_ACCESS_TOKEN env var not set"}

    genius_client = _recommender.lyrics_fetcher.genius if _recommender and _recommender.lyrics_fetcher else None
    if not genius_client:
        init_err = getattr(getattr(_recommender, "lyrics_fetcher", None), "init_error", None)
        return {"status": "error", "reason": f"Genius client is None — init error: {init_err or 'unknown'}"}

    genius_result = {}
    try:
        song = genius_client.search_song("Happy", "Pharrell Williams")
        if song and song.lyrics:
            genius_result = {"status": "ok", "test_track": song.title, "lyrics_length": len(song.lyrics)}
        else:
            genius_result = {"status": "error", "reason": "search_song returned None or empty"}
    except Exception as e:
        genius_result = {"status": "error", "reason": f"{type(e).__name__}: {str(e)[:200]}"}

    # Test MusixMatch lyrics fallback
    import json, os, urllib.parse, urllib.request
    mm_key = os.environ.get("MUSIXMATCH_API_KEY")
    mm_result = {}
    if mm_key:
        try:
            params = urllib.parse.urlencode({"q_track": "Happy", "q_artist": "Pharrell Williams", "apikey": mm_key, "format": "json"})
            req = urllib.request.Request(f"https://api.musixmatch.com/ws/1.1/matcher.lyrics.get?{params}", headers={"User-Agent": "eumora/1.0"})
            with urllib.request.urlopen(req, timeout=8) as r:
                data = json.loads(r.read())
            hdr = data.get("message", {}).get("header", {})
            body = data.get("message", {}).get("body", {})
            lyrics_body = body.get("lyrics", {}).get("lyrics_body", "")
            mm_result = {"status_code": hdr.get("status_code"), "lyrics_preview": lyrics_body[:100] if lyrics_body else None}
        except Exception as e:
            mm_result = {"error": f"{type(e).__name__}: {e}"}
    else:
        mm_result = {"error": "MUSIXMATCH_API_KEY not set"}

    return {"genius": genius_result, "musixmatch": mm_result}


@app.get("/health/spotify")
def health_spotify():
    """Test that Spotify search actually returns tracks."""
    if not _recommender:
        return {"status": "error", "reason": "recommender not loaded"}
    try:
        res = _recommender.sp.search(q="happy pop upbeat", type="track", limit=5, market="US")
        items = res.get("tracks", {}).get("items", [])
        return {
            "status": "ok" if items else "empty",
            "track_count": len(items),
            "sample": [{"name": t["name"], "artist": t["artists"][0]["name"]} for t in items[:3]],
        }
    except Exception as e:
        return {"status": "error", "reason": f"{type(e).__name__}: {str(e)[:200]}"}


@app.post("/api/predict")
def predict(req: PredictRequest):
    if not _predictor:
        raise HTTPException(503, "Model not loaded")
    if not req.text and not req.reference_url:
        raise HTTPException(400, "Must provide text or reference_url")

    if req.raw:
        if not req.text:
            raise HTTPException(400, "raw prediction needs text")
        probs = _predictor.predict_raw(req.text)
        labels = sorted(({"label": k, "score": round(v, 6)} for k, v in probs.items()),
                        key=lambda x: x["score"], reverse=True)
        return {"labels": labels, "probabilities": {k: round(v, 6) for k, v in probs.items()}}

    prior = None if req.disable_prior_adjustment else req.target_sarcasm_prior

    text_result = None
    if req.text:
        text_result = _predictor.predict(req.text, target_sarcasm_prior=prior)

    ref_probs = None
    if req.reference_url and _recommender:
        try:
            ref_probs = _recommender.get_reference_emotion(req.reference_url, _predictor)
        except Exception as e:
            print(f"Reference URL failed in /api/predict: {e}")

    if text_result and ref_probs:
        t_probs = text_result.get("probabilities", {})
        merged = {k: (t_probs.get(k, 0.0) + ref_probs.get(k, 0.0)) / 2 for k in set(t_probs) | set(ref_probs)}
        top = max(merged, key=merged.get)
        return {
            "emotion":       top,
            "confidence":    merged[top],
            "probabilities": merged,
            "music_context": text_result.get("music_context", {}),
            "explanation":   text_result.get("explanation", ""),
        }
    elif ref_probs:
        top = max(ref_probs, key=ref_probs.get)
        return {
            "emotion":       top,
            "confidence":    ref_probs[top],
            "probabilities": ref_probs,
            "music_context": _predictor._build_music_context(top) if hasattr(_predictor, "_build_music_context") else {},
            "explanation":   "Emotion derived from the reference track's lyrics.",
        }
    elif text_result:
        return {
            "emotion":       text_result["emotion"],
            "confidence":    text_result["confidence"],
            "probabilities": text_result["probabilities"],
            "music_context": text_result["music_context"],
            "explanation":   text_result["explanation"],
        }
    else:
        raise HTTPException(503, "Could not extract emotion from text or reference track")


@app.post("/api/recommend")
def recommend(req: RecommendRequest):
    if not _predictor:
        raise HTTPException(503, "Model not loaded")

    if not req.text and not req.reference_url:
        raise HTTPException(400, "Must provide text or reference_url")

    prior = None if req.disable_prior_adjustment else req.target_sarcasm_prior

    predict_result = None
    if req.text:
        predict_result = _predictor.predict(req.text, target_sarcasm_prior=prior)
        
    ref_probs = None
    if req.reference_url and _recommender:
        ref_probs = _recommender.get_reference_emotion(req.reference_url, _predictor)

    if not predict_result and not ref_probs:
        raise HTTPException(400, "Could not extract emotion from text or reference track.")

    # Merge or assign
    if predict_result and ref_probs:
        t_probs = predict_result.get("probabilities", {})
        merged = {k: (t_probs.get(k, 0.0) + ref_probs.get(k, 0.0)) / 2 for k in set(t_probs) | set(ref_probs)}
        top = max(merged, key=merged.get)
        
        predict_result["emotion"] = top
        predict_result["confidence"] = merged[top]
        predict_result["probabilities"] = merged
        
    elif ref_probs:
        top = max(ref_probs, key=ref_probs.get)
        predict_result = {
            "emotion": top,
            "confidence": ref_probs[top],
            "probabilities": ref_probs,
            "explanation": "Derived entirely from the reference track's lyrics.",
            "music_context": _predictor._build_music_context(top) if hasattr(_predictor, '_build_music_context') else {}
        }
        
    # Standard recommendation flow
    if not _recommender:
        return {
            "emotion":            predict_result["emotion"],
            "confidence":         predict_result["confidence"],
            "probabilities":      predict_result.get("probabilities", {}),
            "explanation":        predict_result.get("explanation", ""),
            "music_context":      predict_result.get("music_context", {}),
            "targets_used":       None,
            "tracks":             [],
            "spotify_unavailable": True,
        }

    try:
        import time as _t
        _t0 = _t.monotonic()
        result = _recommender.recommend(
            predict_result,
            limit=req.limit,
            genre_override=req.genre,
            blend=req.blend,
            raw_text=req.text,
            predictor=_predictor,
        )
        result["_timing_s"] = round(_t.monotonic() - _t0, 2)
        return result
    except RuntimeError as exc:
        err_str = str(exc)
        print(f"Spotify recommend error: {err_str}")
        return {
            "emotion":             predict_result["emotion"],
            "confidence":          predict_result["confidence"],
            "probabilities":       predict_result.get("probabilities", {}),
            "explanation":         predict_result.get("explanation", ""),
            "music_context":       predict_result.get("music_context", {}),
            "targets_used":        None,
            "tracks":              [],
            "spotify_unavailable": True,
        }


# The scoring helpers live in src/scoring.py so the local benchmark can run the
# exact code this endpoint runs. Old names kept for anything importing them.
from src.scoring import (  # noqa: E402
    chunked_lyric_probs as _chunked_lyric_probs,
    combine as _combine,
    cosine as _cosine,
    emotion_match as _emotion_match,
    fused_response as _fused_response,
    lyrics_language_ok as _lyrics_language_ok,
    score_lyrics as _score_lyrics,
    score_probs as _score_probs,
)


@app.post("/api/score")
def score_track(req: ScoreRequest):
    """Score a track against the user's emotion by its lyrics and its sound.
    Lyrics: fetched, scored by stanza with DeBERTa, compared with the user's
    emotion vector. Audio: the 30 s preview, embedded with VGGish and placed in
    valence/arousal space. The two run in parallel and are fused 50/50; either
    alone is used when the other is unavailable (see scoring.combine).
    """
    if not _predictor:
        raise HTTPException(503, "Model not loaded")

    started = _time.monotonic()
    # Enter the lyric phase before submitting the audio, so the audio sees
    # this request's lyric work and waits for it rather than racing it.
    with _lyric_phase():
        audio_future = (_audio_pool.submit(_audio_for, req.title, req.artist, req.preview_url)
                        if _acoustic is not None else None)
        probs, lyrics_found, lyrics_english = _lyric_probs_for(req.title, req.artist)
    lyric_result = _score_probs(probs, req.user_probabilities) if probs else None

    audio, audio_match, pending = None, None, False
    if audio_future is not None:
        remaining = max(0.0, AUDIO_BUDGET_S - (_time.monotonic() - started))
        try:
            audio = audio_future.result(timeout=remaining)
        except FutureTimeout:
            pending = True  # still running; it will be cached when it finishes
        except Exception as e:
            print(f"Audio analysis failed for '{req.title}': {type(e).__name__}")
        if audio:
            audio_match = _acoustic.head.match(req.user_probabilities,
                                               (audio["valence"], audio["arousal"]))

    result = None
    if _fusion is not None:
        try:
            result = _fused_response(_fusion, req.user_probabilities, probs, audio, lyrics_english)
        except Exception as e:  # never let the new path take scoring down
            print(f"VAD fusion failed for '{req.title}': {type(e).__name__}: {e}")
            result = None
    if result is None:
        view = None
        if audio:
            view = {k: audio.get(k) for k in ("valence", "arousal", "sound", "descriptors", "preview_url")}
        result = _combine(lyric_result, view, audio_match)
    if result is None:
        result = {"match_score": None, "reason": "predict_failed" if lyrics_found else "no_lyrics"}
    if pending:
        result["audio_pending"] = True
    return result


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True)