"""FastAPI server for AgentCut - AI Director Video Production Pipeline.""" import logging import os import json import shutil import time import uuid import asyncio from typing import AsyncGenerator from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from backend.pipeline import run_pipeline, PipelineEvent from backend.config import OUTPUT_DIR, validate_config logging.basicConfig( level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("agentcut") validate_config() app = FastAPI(title="AgentCut", description="AI Director - Multi-Agent Video Production") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # In-memory job storage jobs: dict = {} class CreateVideoRequest(BaseModel): prompt: str = Field(..., min_length=1, max_length=1000) duration: int = Field(default=30, ge=18, le=30) num_shots: int = Field(default=3, ge=1, le=5) include_music: bool = True class JobStatus(BaseModel): job_id: str status: str # "pending", "running", "completed", "failed" progress: float events: list output_path: str = None error: str = None @app.post("/api/create") async def create_video(req: CreateVideoRequest): """Start a new video production job.""" job_id = str(uuid.uuid4())[:8] logger.info("Job %s: created (prompt=%r, duration=%d, shots=%d, music=%s)", job_id, req.prompt[:60], req.duration, req.num_shots, req.include_music) jobs[job_id] = { "status": "pending", "progress": 0.0, "events": [], "output_path": None, "error": None, "created_at": time.time(), } # Run pipeline in background asyncio.get_event_loop().run_in_executor( None, _run_job, job_id, req.prompt, req.duration, req.num_shots, req.include_music, ) return {"job_id": job_id} def _run_job(job_id: str, prompt: str, duration: int, num_shots: int, include_music: bool): """Execute the pipeline in a background thread.""" jobs[job_id]["status"] = "running" start_time = time.time() logger.info("Job %s: pipeline started", job_id) def on_event(event: PipelineEvent): jobs[job_id]["progress"] = event.progress jobs[job_id]["events"].append({ "stage": event.stage, "message": event.message, "progress": event.progress, "data": event.data, "timestamp": event.timestamp, }) try: result = run_pipeline( prompt, duration, num_shots, include_music, on_event=on_event, job_id=job_id, ) jobs[job_id]["status"] = "completed" jobs[job_id]["progress"] = 1.0 jobs[job_id]["output_path"] = result["output_path"] elapsed = time.time() - start_time logger.info("Job %s: completed in %.1fs, output=%s", job_id, elapsed, result["output_path"]) except Exception as e: jobs[job_id]["status"] = "failed" jobs[job_id]["error"] = str(e) elapsed = time.time() - start_time logger.error("Job %s: failed after %.1fs: %s", job_id, elapsed, e, exc_info=True) @app.get("/api/status/{job_id}") async def get_status(job_id: str): """Get current job status and progress.""" if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") return jobs[job_id] @app.get("/api/stream/{job_id}") async def stream_events(job_id: str): """SSE stream for real-time progress updates.""" if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") async def event_generator() -> AsyncGenerator[str, None]: last_idx = 0 while True: job = jobs[job_id] events = job["events"] # Send new events while last_idx < len(events): event = events[last_idx] yield f"data: {json.dumps(event)}\n\n" last_idx += 1 if job["status"] in ("completed", "failed"): final = { "stage": "done", "message": "Pipeline finished" if job["status"] == "completed" else f"Error: {job['error']}", "progress": job["progress"], "data": {"status": job["status"], "output_path": job.get("output_path")}, } yield f"data: {json.dumps(final)}\n\n" break await asyncio.sleep(1) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, ) @app.get("/api/download/{job_id}") async def download_video(job_id: str): """Download the final video.""" if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") job = jobs[job_id] if job["status"] != "completed" or not job["output_path"]: raise HTTPException(status_code=400, detail="Video not ready") return FileResponse( job["output_path"], media_type="video/mp4", filename="agentcut_output.mp4", ) @app.get("/api/health") async def health(): return {"status": "ok", "service": "AgentCut"} # Job cleanup: remove jobs older than 1 hour every 10 minutes async def _cleanup_old_jobs(): while True: await asyncio.sleep(600) # 10 minutes cutoff = time.time() - 3600 # 1 hour stale = [jid for jid, j in jobs.items() if j.get("created_at", 0) < cutoff and j["status"] in ("completed", "failed")] for jid in stale: # Remove output directory job_dir = os.path.join(OUTPUT_DIR, f"job_{jid}") if os.path.isdir(job_dir): shutil.rmtree(job_dir, ignore_errors=True) del jobs[jid] if stale: logger.info("Cleaned up %d old jobs: %s", len(stale), stale) @app.on_event("startup") async def startup(): asyncio.create_task(_cleanup_old_jobs()) # Serve frontend static files (must be last - catches all unmatched routes) _frontend_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "frontend") if os.path.isdir(_frontend_dir): app.mount("/", StaticFiles(directory=_frontend_dir, html=True), name="frontend") if __name__ == "__main__": import uvicorn uvicorn.run("backend.main:app", host="0.0.0.0", port=8000, reload=True)