Spaces:
Running
Running
File size: 3,805 Bytes
1aa90a9 | 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 | from fastapi import APIRouter, HTTPException, Response
from fastapi.responses import FileResponse, StreamingResponse
import logging
from models.schemas import (
CreateVideoRequest,
CreateVideoResponse,
VideoStatusResponse,
VideoListResponse,
VideoListItem
)
from video_creator.short_creator import ShortCreator
logger = logging.getLogger(__name__)
# This will be set when the app starts
short_creator: ShortCreator = None
def set_short_creator(creator: ShortCreator):
"""Set the global short creator instance"""
global short_creator
short_creator = creator
router = APIRouter(prefix="/api")
@router.post("/short-video",
response_model=CreateVideoResponse,
status_code=201,
tags=["Videos"],
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,
tags=["Videos"],
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}",
tags=["Videos"],
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"""
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,
tags=["Videos"],
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}",
tags=["Videos"],
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",
tags=["Configuration"],
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",
tags=["Configuration"],
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()
|