File size: 4,298 Bytes
9834af3 db23a91 9834af3 db23a91 9834af3 db23a91 9834af3 | 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 | """
Fact Image Router
FastAPI endpoints for fact-image video generation
"""
import logging
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, HTTPException, Depends
from fastapi.responses import FileResponse, RedirectResponse
from .schemas import (
FactImageRequest,
FactImageResponse,
FactImageStatus,
JobStatus
)
from .services.fact_creator import FactCreator
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/fact-image", tags=["Fact Image"])
# Will be set during app startup
fact_creator: Optional[FactCreator] = None
def get_fact_creator() -> FactCreator:
"""Dependency to get FactCreator instance"""
if fact_creator is None:
raise HTTPException(status_code=503, detail="Service not initialized")
return fact_creator
@router.post("/", response_model=FactImageResponse)
async def create_fact_image(
request: FactImageRequest,
creator: FactCreator = Depends(get_fact_creator)
):
"""
Create a new fact-image video.
- **model**: Image generation model (nvidia, cloudflare, pexels)
- **image_prompt**: Prompt for background image
- **fact_heading**: Optional heading text (e.g., 'Psychological Hack')
- **heading_background**: Heading background style config
- **fact_text**: The fact/quote to overlay on the image
- **duration**: Video duration in seconds (4-7)
"""
logger.info(f"New fact-image request: model={request.model}, heading='{request.fact_heading}', duration={request.duration}s")
# Convert heading_background to dict if present
heading_bg_dict = None
if request.heading_background:
heading_bg_dict = {
"enabled": request.heading_background.enabled,
"color": request.heading_background.color,
"padding": request.heading_background.padding,
"corner_radius": request.heading_background.corner_radius
}
job_id = creator.add_to_queue(
model=request.model,
image_prompt=request.image_prompt,
fact_text=request.fact_text,
duration=request.duration,
fact_heading=request.fact_heading,
heading_background=heading_bg_dict
)
return FactImageResponse(
job_id=job_id,
status="processing",
status_url=f"/api/fact-image/{job_id}/status",
download_url=f"/api/fact-image/{job_id}"
)
@router.get("/{job_id}/status", response_model=FactImageStatus)
async def get_status(
job_id: str,
creator: FactCreator = Depends(get_fact_creator)
):
"""Get the status of a fact-image job"""
status = creator.get_status(job_id)
return FactImageStatus(**status)
@router.get("/{job_id}")
async def download_video(
job_id: str,
creator: FactCreator = Depends(get_fact_creator)
):
"""
Download the generated fact-image video.
- If cloud-stored: redirects to HF Hub URL
- If local: returns the MP4 file
"""
# Check for cloud storage first
cloud_file = creator.config.videos_dir_path / f"{job_id}.cloud"
if cloud_file.exists():
cloud_url = cloud_file.read_text().strip()
# Ensure download parameter
if "?download=true" not in cloud_url:
cloud_url = f"{cloud_url}?download=true"
return RedirectResponse(url=cloud_url)
# Check for local file
video_path = creator.get_video_path(job_id)
if video_path and video_path.exists():
return FileResponse(
path=str(video_path),
media_type="video/mp4",
filename=f"{job_id}.mp4"
)
raise HTTPException(status_code=404, detail="Video not found")
@router.delete("/{job_id}")
async def delete_video(
job_id: str,
creator: FactCreator = Depends(get_fact_creator)
):
"""Delete a fact-image video"""
# Delete from jobs dict
if job_id in creator.jobs:
del creator.jobs[job_id]
# Delete video file
video_path = creator.get_video_path(job_id)
if video_path and video_path.exists():
video_path.unlink()
# Delete cloud metadata
cloud_file = creator.config.videos_dir_path / f"{job_id}.cloud"
if cloud_file.exists():
cloud_file.unlink()
return {"message": "Deleted", "job_id": job_id}
|