from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, JSONResponse from pydantic import BaseModel import os from dotenv import load_dotenv import asyncio # Load environment variables FIRST load_dotenv() # ⚙️ DEPLOYMENT CONFIGURATION # FFmpeg Configuration - Tries to find FFmpeg automatically import shutil # Check if ffmpeg is already in system PATH if shutil.which('ffmpeg') is None: # If not in PATH, try custom path (update this for your system) # Common Windows path: r"C:\ffmpeg\bin" # Linux/Mac: Usually already in PATH, no need to set custom_ffmpeg_path = os.environ.get('FFMPEG_PATH', r"M:\Ap\ffmpeg-7.1.1-essentials_build\ffmpeg-7.1.1-essentials_build\bin") if os.path.exists(custom_ffmpeg_path): os.environ["PATH"] += os.pathsep + custom_ffmpeg_path print(f"✅ FFmpeg found at: {custom_ffmpeg_path}") else: print("⚠️ FFmpeg not found in system PATH") print(" Options:") print(" 1. Install FFmpeg and add to system PATH") print(" 2. Set FFMPEG_PATH environment variable") print(f" 3. Update custom_ffmpeg_path in main.py (currently: {custom_ffmpeg_path})") else: print("✅ FFmpeg found in system PATH") from teacher import generate_outline from compiler import generate_manim_code from runner import render_scene from narrator import generate_narration_audio app = FastAPI() # Get the directory of the current script (backend/) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) MEDIA_DIR = os.path.join(BASE_DIR, "media") STATIC_DIR = os.path.join(BASE_DIR, "static") # Root directory (parent of backend/) ROOT_DIR = os.path.dirname(BASE_DIR) # Ensure the media directory exists os.makedirs(MEDIA_DIR, exist_ok=True) app.mount("/media", StaticFiles(directory=MEDIA_DIR), name="media") # Serve static frontend files app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # Global Job Status job_status = { "stage": "idle", # idle, planning, coding, executing, success, failed "message": "System Ready", "video_path": None, "error": None } class PromptRequest(BaseModel): prompt: str async def process_video_generation(prompt: str): global job_status try: # 1. Planning job_status["stage"] = "planning" job_status["message"] = "Analyzing Prompt & Generating Outline..." outline = await generate_outline(prompt) # 2. Narrator: Generate per-step narration audio files steps = outline.get("steps", []) step_audio_paths = [] from narrator import generate_narration_audio, concatenate_audio_files, get_audio_duration total_audio_duration = 0.0 for idx, step in enumerate(steps): narration = step.get("narration", "") if narration: audio_filename = f"step_{idx+1}_narration.mp3" audio_path = generate_narration_audio(narration, filename=audio_filename) if audio_path: step_audio_paths.append(audio_path) duration = get_audio_duration(audio_path) total_audio_duration += duration print(f"✓ Step {idx+1} audio: {duration:.2f}s") else: step_audio_paths.append(None) else: step_audio_paths.append(None) # Combine all step audios into one file combined_audio_path = None if step_audio_paths and any(step_audio_paths): combined_audio_path = concatenate_audio_files(step_audio_paths, output_filename="combined_narration.mp3") if combined_audio_path: print(f"✓ Combined audio duration: {total_audio_duration:.2f}s") # 3. Coding job_status["stage"] = "coding" job_status["message"] = "Generating Manim Script..." code = await generate_manim_code(outline, step_audio_paths=step_audio_paths) # 4. Executing job_status["stage"] = "executing" job_status["message"] = "Rendering Animation Frames..." video_path = await render_scene(code) # 5. Merge audio with video if available if combined_audio_path: from narrator import merge_audio_video job_status["message"] = "Merging Audio with Video..." video_path = merge_audio_video(video_path, combined_audio_path) print(f"✓ Final video with audio: {video_path}") # Success relative_path = os.path.relpath(video_path, start=MEDIA_DIR).replace("\\", "/") job_status["stage"] = "success" job_status["message"] = "Render Complete!" job_status["video_path"] = relative_path except Exception as e: error_msg = str(e) print(f"Error generating video: {error_msg}") job_status["stage"] = "failed" job_status["message"] = "Process Failed" job_status["error"] = error_msg # Log error with open("error.log", "w") as f: f.write(error_msg) import traceback traceback.print_exc(file=f) @app.get("/") async def read_index(): return FileResponse(os.path.join(STATIC_DIR, 'index.html')) @app.get("/favicon.ico") async def get_favicon(): favicon_path = os.path.join(ROOT_DIR, 'favicon.ico') if os.path.exists(favicon_path): return FileResponse(favicon_path) raise HTTPException(status_code=404, detail="Favicon not found") @app.post("/generate") async def generate_video(request: PromptRequest, background_tasks: BackgroundTasks): global job_status # Reset status job_status = { "stage": "planning", "message": "Initializing...", "video_path": None, "error": None } # Start background task background_tasks.add_task(process_video_generation, request.prompt) return {"status": "started"} @app.get("/status") async def get_status(): return job_status @app.get("/video/{path:path}") async def get_video(path: str): video_path = os.path.join(MEDIA_DIR, path) if os.path.exists(video_path): return FileResponse(video_path) raise HTTPException(status_code=404, detail="Video not found") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)