Spaces:
Running
Running
File size: 3,691 Bytes
cb6fbf5 4ae04e1 cb6fbf5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """
api/main.py — FastAPI backend for feelin'
"""
from __future__ import annotations
from fastapi import FastAPI, BackgroundTasks, HTTPException
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from data.db import init_db, get_leaderboard, get_latest_podcast, get_stats ,SEG_DIR, POD_DIR
from agents.graph import run_post_pipeline
BASE_DIR = Path(__file__).parent.parent
app = FastAPI(title="feelin' API", version="1.0")
# Serve static podcast/segment files
static_dir = BASE_DIR / "static"
static_dir.mkdir(parents=True, exist_ok=True)
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
# Init DB on startup
@app.on_event("startup")
def on_startup():
init_db()
# ── Models ──────────────────────────────────────────────────────────────────
class PostRequest(BaseModel):
text: str
class PodcastRequest(BaseModel):
category: str
# ── Routes ──────────────────────────────────────────────────────────────────
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/post")
def submit_post(req: PostRequest):
if not req.text or len(req.text.strip()) < 10:
raise HTTPException(status_code=400, detail="Post too short.")
result = run_post_pipeline(req.text.strip())
return result
@app.get("/leaderboard/{category}")
def leaderboard_category(category: str, limit: int = 10):
valid = ["hilarious", "tragic", "unhinged", "awkward", "chaotic"]
if category not in valid:
raise HTTPException(status_code=400, detail="Invalid category.")
posts = get_leaderboard(category, limit=limit)
return {"category": category, "posts": posts}
@app.get("/leaderboard")
def leaderboard_all():
result = {}
for cat in ["hilarious", "tragic", "unhinged", "awkward", "chaotic"]:
result[cat] = get_leaderboard(cat, limit=5)
return result
@app.post("/podcast/generate")
def generate_podcast_endpoint(req: PodcastRequest, background_tasks: BackgroundTasks):
valid = ["hilarious", "tragic", "unhinged", "awkward", "chaotic"]
if req.category not in valid:
raise HTTPException(status_code=400, detail="Invalid category.")
def _safe_generate(cat: str):
try:
from podcast.builder import generate_podcast
generate_podcast(cat)
except Exception as e:
print(f"[Podcast background error] {e}")
background_tasks.add_task(_safe_generate, req.category)
return {"status": "generating", "category": req.category}
@app.get("/podcast/{category}")
def get_podcast(category: str):
valid = ["hilarious", "tragic", "unhinged", "awkward", "chaotic"]
if category not in valid:
raise HTTPException(status_code=400, detail="Invalid category.")
pod = get_latest_podcast(category)
if not pod:
return {"status": "not_ready", "message": "No episode generated yet. Click ⚡ to generate."}
audio_path = pod.get("audio_path", "")
audio_url = f"/{audio_path}" if audio_path else None
return {
"status": "ready",
"category": category,
"audio_url": audio_url,
"script": pod.get("script", ""),
"post_count": pod.get("post_count", 0),
"created_at": pod.get("created_at", ""),
}
@app.get("/stats")
def stats():
return get_stats()
|