| import uvicorn
|
| from fastapi import FastAPI, File, UploadFile, Form, HTTPException
|
| from fastapi.responses import FileResponse
|
| from fastapi.staticfiles import StaticFiles
|
| from typing import Optional
|
| import services.vertex_service as vertex_service
|
| import services.video_service as video_service
|
| import os
|
|
|
| app = FastAPI()
|
|
|
| @app.post("/api/generate_video")
|
| async def generate_video(
|
| api_key: str = Form(...),
|
| prompt: str = Form(...),
|
| negative_prompt: Optional[str] = Form(None),
|
| model: str = Form(...),
|
| aspect_ratio: str = Form(...),
|
| duration: int = Form(...),
|
| resolution: str = Form(...),
|
| sample_count: int = Form(...),
|
| seed: Optional[int] = Form(None),
|
| person_generation: str = Form(...),
|
| generate_audio: bool = Form(...),
|
| enhance_prompt: bool = Form(...),
|
| image: Optional[UploadFile] = File(None),
|
| video: Optional[UploadFile] = File(None),
|
| ):
|
| try:
|
| project_id = await vertex_service.get_project_id(api_key)
|
|
|
| params = {
|
| "prompt": prompt,
|
| "negativePrompt": negative_prompt,
|
| "model": model,
|
| "aspectRatio": aspect_ratio,
|
| "durationSeconds": duration,
|
| "resolution": resolution,
|
| "sampleCount": sample_count,
|
| "seed": seed,
|
| "personGeneration": person_generation,
|
| "generateAudio": generate_audio,
|
| "enhancePrompt": enhance_prompt,
|
| }
|
|
|
| if image:
|
| params["image"] = await image.read()
|
| params["image_mime_type"] = image.content_type
|
|
|
| if video:
|
| params["video"] = await video.read()
|
| params["video_mime_type"] = video.content_type
|
|
|
| model_id = params["model"]
|
| operation_name = await vertex_service.start_video_generation(project_id, api_key, params)
|
| video_data = await vertex_service.poll_video_status(project_id, model_id, operation_name, api_key)
|
|
|
| video_path = await video_service.save_video(video_data)
|
|
|
| return {"video_url": f"/api/video/{os.path.basename(video_path)}"}
|
| except Exception as e:
|
| raise HTTPException(status_code=400, detail=str(e))
|
|
|
| @app.get("/api/video/{video_id}")
|
| async def get_video(video_id: str):
|
| video_path = os.path.join(video_service.VIDEO_DIR, video_id)
|
| if os.path.exists(video_path):
|
| return FileResponse(video_path)
|
| raise HTTPException(status_code=404, detail="Video not found")
|
|
|
| app.mount("/", StaticFiles(directory="../frontend", html=True), name="static")
|
|
|
| if __name__ == "__main__":
|
| uvicorn.run(app, host="0.0.0.0", port=8000) |
