File size: 4,293 Bytes
7fa9d90 ae40e4a 16ae8b2 ae40e4a 7fa9d90 | 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 | """
Video Creator Router - API Endpoints
"""
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
import logging
from .schemas import (
CreateVideoRequest,
CreateVideoResponse,
VideoStatusResponse,
VideoListResponse,
VideoListItem
)
from .services.short_creator import ShortCreator
logger = logging.getLogger(__name__)
# This will be set when the module registers
short_creator: ShortCreator = None
def set_short_creator(creator: ShortCreator):
"""Set the global short creator instance"""
global short_creator
short_creator = creator
router = APIRouter()
@router.post("/short-video",
response_model=CreateVideoResponse,
status_code=201,
summary="Create a new video",
description="Create a new short video from text scenes. Returns a video ID to track progress."
)
async def create_short_video(request: CreateVideoRequest):
"""Create a new short video"""
try:
logger.info(f"Creating short video with {len(request.scenes)} scenes")
video_id = short_creator.add_to_queue(
request.scenes,
request.config
)
return CreateVideoResponse(videoId=video_id)
except Exception as e:
logger.error(f"Error creating video: {e}", exc_info=True)
raise HTTPException(status_code=400, detail=str(e))
@router.get("/short-video/{video_id}/status",
response_model=VideoStatusResponse,
summary="Get video status",
description="Check the processing status of a video (processing, ready, or failed)"
)
async def get_video_status(video_id: str):
"""Get the status of a video"""
status = short_creator.get_status(video_id)
return VideoStatusResponse(status=status)
@router.get("/short-video/{video_id}",
summary="Download video",
description="Download the generated video file (MP4 format)",
responses={
200: {"description": "Video file", "content": {"video/mp4": {}}},
404: {"description": "Video not found"}
}
)
async def get_video(video_id: str):
"""Download/stream a video - supports both local and cloud storage"""
from config import config
# Check if video is in cloud (has .cloud metadata file)
cloud_meta_path = config.videos_dir_path / f"{video_id}.cloud"
if cloud_meta_path.exists():
cloud_url = cloud_meta_path.read_text().strip()
if cloud_url:
# Ensure ?download=true for direct download
if "?download=true" not in cloud_url:
cloud_url = cloud_url + "?download=true"
from fastapi.responses import RedirectResponse
return RedirectResponse(url=cloud_url, status_code=302)
# Otherwise serve from local storage
video_path = short_creator.get_video_path(video_id)
if not video_path.exists():
raise HTTPException(status_code=404, detail="Video not found")
return FileResponse(
video_path,
media_type="video/mp4",
filename=f"{video_id}.mp4"
)
@router.get("/short-videos",
response_model=VideoListResponse,
summary="List all videos",
description="Get a list of all videos with their current status"
)
async def list_videos():
"""List all videos"""
videos = short_creator.list_all_videos()
return VideoListResponse(
videos=[VideoListItem(**v) for v in videos]
)
@router.delete("/short-video/{video_id}",
summary="Delete video",
description="Delete a video by its ID"
)
async def delete_video(video_id: str):
"""Delete a video"""
try:
short_creator.delete_video(video_id)
return {"success": True}
except Exception as e:
logger.error(f"Error deleting video: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/voices",
summary="List TTS voices",
description="Get all available text-to-speech voice options"
)
async def get_voices():
"""List available TTS voices"""
return short_creator.get_available_voices()
@router.get("/music-tags",
summary="List music moods",
description="Get all available background music mood options"
)
async def get_music_tags():
"""List available music moods"""
return short_creator.get_available_music_tags()
|