# COMPLETE MODIFIED api.py - READY TO USE import os, uuid, time, pytz from typing import List, Optional from datetime import datetime, timezone import pandas as pd import numpy as np from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger # import your updated model file import pacebeats_model as core # ---------- Schemas ---------- class RecommendRequest(BaseModel): run_mode: str = Field("quick", description="Run mode: 'quick' (real-time pace) or 'goal' (target pace)") pace_min: float = Field(..., description="User pace in minutes per km") goal_pace_min_per_km: Optional[float] = Field(None, description="Target pace for goal-based runs") user_mood: Optional[str] = Field(None, description="User mood filter (sad/happy/chill/hype/focus/angry)") playlist_id: Optional[str] = Field(None, description="If provided, only recommend songs from this playlist") custom_catalog: Optional[List[dict]] = Field(None, description="Custom song catalog for local music") top_n: int = Field(5, description="Number of tracks to recommend") use_ml: bool = Field(True, description="Use ML re-ranking if trained, else fallback to rule-based") alpha: float = Field(0.3, description="Blending weight: 0.3=30% rule score, 70% ML score") session_id: Optional[str] = Field(None, description="Session UUID") user_id: Optional[str] = Field(None, description="App user's UUID") class RecommendItem(BaseModel): track_id: str spotify_id: Optional[str] = None title: Optional[str] = "" bpm: Optional[float] = None mood: Optional[str] = None final_score: Optional[float] = None rule_score: Optional[float] = None ml_probability: Optional[float] = None class RecommendResponse(BaseModel): session_id: str count: int items: List[RecommendItem] class FeedbackRequest(BaseModel): track_id: str liked: bool user_id: Optional[str] = None class EventLogRequest(BaseModel): track_id: str played_ms: int skipped: Optional[bool] = False liked: Optional[bool] = None disliked: Optional[bool] = None completed: Optional[bool] = False session_id: Optional[str] = None user_id: Optional[str] = None class CacheUpdateResponse(BaseModel): ok: bool user_id: str cache_data: Optional[dict] = None message: str class IncrementalTrainResponse(BaseModel): ok: bool training_id: str samples_processed: int duration_seconds: float message: str class TrainingStatusResponse(BaseModel): algorithms: List[dict] class TrainingLogsResponse(BaseModel): logs: List[dict] total_logs: int # ---------- App ---------- app = FastAPI( title="PaceBeats API", version="1.5.0", description="Music Recommender API with Multi-Model Training Support." ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ---------- Globals ---------- training_scheduler = None # ---------- Helpers ---------- def _df_to_items(df: pd.DataFrame): if df is None or df.empty: return [] cols = ["track_id","spotify_id","title","bpm","mood","final_score","rule_score","ml_probability"] present = [c for c in cols if c in df.columns] recs = df[present].to_dict(orient="records") return [{k: (v.item() if isinstance(v, np.generic) else v) for k,v in row.items()} for row in recs] def _set_user(user_id: Optional[str]): if user_id: core.USER_ID = user_id def start_training_scheduler(): """Start background scheduler for nightly retraining at 1 AM Manila Time""" global training_scheduler if training_scheduler is not None: return training_scheduler = BackgroundScheduler() ph_tz = pytz.timezone('Asia/Manila') # Updated to call the 4-algorithm retraining logic training_scheduler.add_job( func=core.scheduled_full_retraining, trigger=CronTrigger(hour=1, minute=0, timezone=ph_tz), id="nightly_full_retrain_all", name="Nightly Full Retraining (4 Algorithms)", replace_existing=True ) training_scheduler.start() print("✅ Training scheduler started - All 4 models scheduled for 1 AM Manila Time.") # ---------- Startup/Shutdown ---------- @app.on_event("startup") def on_startup(): if not os.getenv("SUPABASE_URL") or not os.getenv("SUPABASE_KEY"): raise RuntimeError("Missing SUPABASE_URL or SUPABASE_KEY in Space secrets.") if "ml_model" not in core.__dict__ or core.ml_model is None: core.ml_model = core.PaceBeatsMlModel() start_training_scheduler() @app.on_event("shutdown") def on_shutdown(): global training_scheduler if training_scheduler: training_scheduler.shutdown() print("✅ Training scheduler stopped") # ---------- Endpoints ---------- @app.get("/health", tags=["System"]) def health(): return {"ok": True} @app.post("/recommend", response_model=RecommendResponse, tags=["Recommendation"]) def recommend(req: RecommendRequest): _set_user(req.user_id) sid = req.session_id or str(uuid.uuid4()) try: if req.run_mode == "goal": pace = req.goal_pace_min_per_km if req.goal_pace_min_per_km else req.pace_min target_pace = req.goal_pace_min_per_km else: pace = req.pace_min target_pace = None if req.custom_catalog: core.catalog = pd.DataFrame(req.custom_catalog) # (Validation and formatting logic omitted for brevity, keeping original behavior) elif req.playlist_id: playlist_songs = core.supabase.table("playlist_songs").select("track_id").eq("playlist_id", req.playlist_id).execute() track_ids = [row["track_id"] for row in (playlist_songs.data or [])] core.catalog = core.fetch_catalog() core.catalog = core.catalog[core.catalog["track_id"].isin(track_ids)] else: core.catalog = core.fetch_catalog() df = core.recommend_tracks_ml( pace, req.user_mood, req.top_n, sid, req.use_ml, req.alpha, run_mode=req.run_mode, target_pace_min=target_pace ) items = _df_to_items(df) return RecommendResponse(session_id=sid, count=len(items), items=items) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/feedback", tags=["Feedback"]) def feedback(req: FeedbackRequest): _set_user(req.user_id) try: core.record_feedback(req.track_id, req.liked) return {"ok": True} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/events/log", tags=["Feedback"]) def log_event(req: EventLogRequest): _set_user(req.user_id) sid = req.session_id or str(uuid.uuid4()) try: core.log_listening_event(req.track_id, req.played_ms, skipped=req.skipped, liked=req.liked, disliked=req.disliked, completed=req.completed, session_id=sid) return {"ok": True, "session_id": sid} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # ========== TRAINING ENDPOINTS ========== @app.post("/train/full-retrain", tags=["Training"]) def manual_full_retrain(): """Manually trigger full model retraining for all 4 algorithms""" try: ok = core.scheduled_full_retraining() return { "ok": ok, "message": "Full retraining completed." if ok else "Training failed. You might have 0 data samples or a database error." } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/train/status", response_model=TrainingStatusResponse, tags=["Training"]) def get_training_status(): """Get current training status & metadata for all 4 algorithms from Supabase""" try: # Fetch all rows (1-4) to see the status of every algorithm metadata = core.supabase.table("model_training_metadata").select("*").order("id").execute() return {"algorithms": metadata.data or []} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/train/incremental", response_model=IncrementalTrainResponse, tags=["Training"]) def train_incremental(): try: training_id = str(uuid.uuid4()) start_time = time.time() new_data = core.get_training_data_since_last_update() if new_data.empty: return IncrementalTrainResponse(ok=False, training_id=training_id, samples_processed=0, duration_seconds=0, message="⚠️ No new data") ok = core.update_model_incrementally(new_data) elapsed = time.time() - start_time return IncrementalTrainResponse(ok=ok, training_id=training_id, samples_processed=len(new_data), duration_seconds=round(elapsed, 2), message="Done") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/train/logs", response_model=TrainingLogsResponse, tags=["Training"]) def get_training_logs(limit: int = 20, status: Optional[str] = None): try: query = core.supabase.table("model_training_logs").select("*").order("created_at", desc=True).limit(limit) if status: query = query.eq("status", status) logs = query.execute() return TrainingLogsResponse(logs=logs.data or [], total_logs=len(logs.data or [])) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/session/{session_id}/finalize", tags=["Training"]) def finalize_session(session_id: str, user_id: Optional[str] = None): try: _set_user(user_id) events = core.supabase.table("listening_events").select("*").eq("session_id", session_id).execute() if not events.data: return {"ok": False, "message": "No events"} user_id_from_events = events.data[0]["user_id"] cache_result = core.update_user_preference_cache(user_id_from_events) return {"ok": True, "cache_updated": cache_result is not None} except Exception as e: raise HTTPException(status_code=500, detail=str(e))