#!/usr/bin/env python3 """ FastAPI Server for Content Generation Standalone service for LLM-based manifest generation No external dependencies - fully self-contained """ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Dict, Any, Optional import os import uvicorn # Import local content generator from content_gen import ContentGenerator # Initialize FastAPI app app = FastAPI( title="Content Generator API", description="LLM-based manifest generation for video content", version="1.0.0" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Hardcoded configuration CONFIG = { "scenes_per_video": 7, "llm_model": "moonshotai/kimi-k2.6:free", "resolution": [1080, 1920], "fps": 30 } # Initialize content generator content_gen = ContentGenerator(CONFIG) # Request/Response models class ManifestRequest(BaseModel): topic: str description: Optional[str] = None class SceneModel(BaseModel): idx: int label: str query: str class ManifestResponse(BaseModel): title: str scenes: list caption: str hashtags: list timestamp: str @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "service": "content-generator", "version": "1.0.0" } @app.post("/generate", response_model=ManifestResponse) async def generate_manifest(request: ManifestRequest): """ Generate a video manifest from a topic. Args: topic: The video topic/prompt description: Optional additional context Returns: Manifest with scenes, caption, and hashtags """ try: if not request.topic or len(request.topic.strip()) == 0: raise HTTPException(status_code=400, detail="Topic cannot be empty") print(f"[API] Generating manifest for: {request.topic}") manifest = content_gen.generate_manifest(request.topic) return ManifestResponse( title=manifest.get("title", ""), scenes=manifest.get("scenes", []), caption=manifest.get("caption", ""), hashtags=manifest.get("hashtags", []), timestamp=manifest.get("timestamp", "") ) except Exception as e: print(f"[API] ERROR: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/generate/batch") async def generate_multiple(requests_list: list[ManifestRequest]): """ Generate multiple manifests in batch. Args: requests_list: List of ManifestRequest objects Returns: List of generated manifests """ results = [] errors = [] for idx, req in enumerate(requests_list): try: manifest = content_gen.generate_manifest(req.topic) results.append({ "index": idx, "topic": req.topic, "manifest": manifest, "status": "success" }) except Exception as e: errors.append({ "index": idx, "topic": req.topic, "error": str(e), "status": "failed" }) return { "total": len(requests_list), "successful": len(results), "failed": len(errors), "results": results, "errors": errors } @app.get("/config") async def get_config(): """Get current configuration""" return { "scenes_per_video": CONFIG.get("scenes_per_video", 7), "llm_model": CONFIG.get("llm_model", ""), "resolution": CONFIG.get("resolution", [1080, 1920]), "fps": CONFIG.get("fps", 30) } if __name__ == "__main__": port = int(os.getenv("PORT", 8000)) host = os.getenv("HOST", "0.0.0.0") workers = int(os.getenv("WORKERS", 1)) print(f"[Server] Starting Content Generator API") print(f" Host: {host}") print(f" Port: {port}") print(f" Workers: {workers}") print(f" API Docs: http://{host}:{port}/docs") uvicorn.run( app, host=host, port=port, workers=workers, log_level="info", access_log=True, reload=False )