"""EUMORA REST API -- FastAPI backend for the frontend.""" import os import sys from contextlib import asynccontextmanager from pathlib import Path from typing import Optional from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from dotenv import load_dotenv load_dotenv() sys.path.insert(0, str(Path(__file__).parent)) # 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: str target_sarcasm_prior: float = 0.15 disable_prior_adjustment: bool = False model_config = {"str_max_length": 2000} # prevent oversized payloads class RecommendRequest(BaseModel): text: str limit: int = 10 genre: Optional[str] = None blend: bool = True max_popularity: int = 65 target_sarcasm_prior: float = 0.15 disable_prior_adjustment: bool = False # --------------------------------------------------------------------------- # App lifecycle -- load heavy models once at startup # --------------------------------------------------------------------------- _predictor = None _recommender = None @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 yield _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 # --------------------------------------------------------------------------- @app.get("/health") def health(): return {"status": "ok", "spotify": _recommender is not None} @app.post("/api/predict") def predict(req: PredictRequest): if not _predictor: raise HTTPException(503, "Model not loaded") _predictor.target_sarcasm_prior = None if req.disable_prior_adjustment else req.target_sarcasm_prior result = _predictor.predict(req.text) return { "emotion": result["emotion"], "confidence": result["confidence"], "probabilities": result["probabilities"], "music_context": result["music_context"], "explanation": result["explanation"], } @app.post("/api/recommend") def recommend(req: RecommendRequest): if not _predictor: raise HTTPException(503, "Model not loaded") _predictor.target_sarcasm_prior = None if req.disable_prior_adjustment else req.target_sarcasm_prior predict_result = _predictor.predict(req.text) 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: return _recommender.recommend( predict_result, limit=req.limit, genre_override=req.genre, blend=req.blend, raw_text=req.text, ) 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, } if __name__ == "__main__": import uvicorn uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True)