| from fastapi import FastAPI, UploadFile, File, Form |
|
|
| from fastapi.responses import FileResponse, JSONResponse |
| from datetime import datetime |
| from typing import List |
|
|
| from extract_metadata import get_metadata, get_scenes_metadata |
| from grok_analyze import create_caps_with_grok, get_story_with_grok |
| from create_captions import create_caps, add_captions_to_video |
| from transcripts_editing import get_dialog |
| from video_editing_ffmpeg import concatenate_videos |
|
|
| import shutil |
| import os |
| import uuid |
|
|
| app = FastAPI(title="Video Storyline Generator API") |
|
|
| |
| UPLOAD_DIR = "uploads" |
| os.makedirs(UPLOAD_DIR, exist_ok=True) |
|
|
| @app.post("/generatestoryline") |
| async def generate_storyline( |
| videos: List[UploadFile] = File(...), |
| date: str = Form(...) |
| ): |
| """ |
| Generate a storyline from uploaded videos and a date. |
| Returns a selected video filename and a list of generated captions. |
| """ |
| try: |
| |
| try: |
| input_date = datetime.strptime(date, "%Y-%m-%d") |
| except ValueError: |
| return JSONResponse( |
| status_code=400, |
| content={"error": "Invalid date format. Use YYYY-MM-DD"} |
| ) |
|
|
| if input_date: |
| dialog= get_dialog(date) |
| |
| saved_videos = [] |
| directory=f"{uuid.uuid4()}" |
| if not os.path.exists(os.path.join(UPLOAD_DIR,directory)): |
| os.makedirs(os.path.join(UPLOAD_DIR,directory)) |
| for video in videos: |
| print(video.filename) |
| |
| file_extension = video.filename.split(".")[-1] |
| file_name = f"{uuid.uuid4()}.{file_extension}" |
|
|
| |
|
|
| file_path = os.path.join(UPLOAD_DIR,directory, file_name) |
| |
| with open(file_path, "wb") as buffer: |
| shutil.copyfileobj(video.file, buffer) |
|
|
| saved_videos.append(file_path) |
|
|
| print("HERE") |
| metadata, filename= get_metadata(saved_videos) |
| print(metadata) |
|
|
| metadata=get_scenes_metadata(metadata) |
|
|
| steps=get_story_with_grok(metadata,"",dialog) |
| print("here are the steps", steps) |
|
|
| caps=create_caps(steps) |
| print("the caps:",caps) |
| final_video=concatenate_videos(steps) |
|
|
| final_video=add_captions_to_video(final_video,False,caps) |
| |
|
|
| return FileResponse( |
| path=final_video, |
| media_type="video/mp4", |
| filename="final_storyline_video.mp4" |
| ) |
|
|
| except Exception as e: |
| return JSONResponse( |
| status_code=500, |
| content={"error": f"An error occurred: {str(e)}"} |
| ) |
|
|
| @app.on_event("shutdown") |
| async def cleanup(): |
| """Clean up uploaded files on shutdown""" |
| for file in os.listdir(UPLOAD_DIR): |
| file_path = os.path.join(UPLOAD_DIR, file) |
| try: |
| if os.path.isfile(file_path): |
| os.unlink(file_path) |
| except Exception: |
| pass |