""" 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}