File size: 3,594 Bytes
9cee612 c50e12a 9cee612 9d54dfd 9cee612 f20025d 9cee612 f61c891 9cee612 f61c891 9cee612 f61c891 9cee612 fc8ec65 ee25523 fc8ec65 b3fc79f 503d4ac b3fc79f fcf62f5 b3fc79f fc8ec65 b3fc79f ee25523 fc8ec65 503d4ac 9cee612 b8b8c8c 1d300b0 b8b8c8c 9cee612 ee25523 fc8ec65 f61c891 503d4ac f61c891 fc8ec65 f61c891 fc8ec65 503d4ac 9cee612 |
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 |
import os
import uuid
import json
import logging
import time
import asyncio
from typing import Dict, Optional
from src.config import get_config_value
from google_src import ai_studio_sdk
from runwayml.generate_video import generate_video_runway
from src.logger_config import logger
from google_src.gcs_utils import upload_file_to_gcs
async def generate_video_process(prompt: str, duration: int, image_input: str = None, **kwargs) -> Dict:
"""
Orchestrate video generation:
1. Check if TEST_AUTOMATION is on -> return mock data.
2. Check USE_GEMIMI_VIDEO -> use ai_studio_sdk.
3. Check USE_GROK_VIDEO -> use Grok.
4. Check USE_FAL_VIDEO (or default) -> use Fal.ai.
5. Else -> use RunwayML.
Args:
prompt: Text prompt
duration: Duration in seconds
image_input: Optional image URL for img2video
**kwargs: Provider-specific args (aspect_ratio, negative_prompt, etc.)
"""
# 0. Get Config
video_type = get_config_value("video_generation_type", "runway").lower()
# 1. Grok Video
if video_type == "grok":
try:
from src.grok_src.grok_video_generator import GrokVideoGenerator
logger.debug("Using Grok SDK for video generation...")
generator = GrokVideoGenerator()
# Forward negative_prompt and system_prompt so they get merged into the Grok prompt
return generator.generate_video(
prompt,
duration=duration,
image_url=image_input,
negative_prompt=kwargs.get("negative_prompt", ""),
system_prompt=kwargs.get("system_prompt", ""),
)
except Exception as e:
logger.error(f"Grok video generation failed: {e}")
raise # Raise to fail visibly
# 2. Gemini / Veo
if video_type == "gemini" or video_type == "veo":
logger.debug("Using Gemini SDK for video generation...")
model_name = "veo3.1_fast" # implied default from context
output_path = f'/tmp/video_{duration}_{model_name}_{uuid.uuid4().hex[:8]}.mp4'
ai_studio_sdk.generate_video(prompt, output_path, image_input)
# Upload to GCS to get a URL to match expectations
upload_result = upload_file_to_gcs(
output_path,
folder="temp_folder_during_processing_delete_after_processing"
)
video_url = upload_result.get('url')
video_result = {
"local_path": output_path,
"video_url": video_url,
"task_id": None,
"duration": duration,
"prompt": prompt,
"status": "success",
"created_at": None,
"model": model_name,
}
return video_result
# 3. Fal.ai (Default if configured or explicitly requested)
if video_type == "fal":
try:
from src.fal_ai import generate_video_fal
logger.debug("Using Fal.ai for video generation...")
# Pass kwargs like negative_prompt, aspect_ratio
return await generate_video_fal(prompt, duration=float(duration), **kwargs)
except Exception as e:
logger.error(f"Fal.ai video generation failed: {e}")
if video_type == "fal" or use_fal: # If explicitly requested, raise
raise
# If implicit, maybe fall through to Runway?
# 4. RunwayML (Final Fallback)
logger.debug("Using RunwayML for video generation...")
return await generate_video_runway(prompt, duration, image_input)
|