Spaces:
Sleeping
Sleeping
File size: 4,614 Bytes
3577c5c | 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | #!/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
)
|