326 lines
No EOL
12 KiB
Python
326 lines
No EOL
12 KiB
Python
import os
|
|
import time
|
|
import uuid
|
|
from google import genai
|
|
from google.genai import types
|
|
from google.auth.exceptions import DefaultCredentialsError
|
|
from utils.auth import get_google_credentials
|
|
from utils.storage import download_blob, delete_blob
|
|
from config import Config
|
|
|
|
# In-memory storage for job status (in production, use Redis or database)
|
|
job_status = {}
|
|
|
|
def generate_video_async(
|
|
prompt: str,
|
|
model_name: str = None,
|
|
video_length_sec: int = 8,
|
|
aspect_ratio: str = "16:9",
|
|
num_videos: int = 1,
|
|
person_generation: str = "dont_allow"
|
|
) -> str:
|
|
"""
|
|
Start video generation asynchronously and return job ID.
|
|
"""
|
|
job_id = str(uuid.uuid4())
|
|
|
|
# Initialize job status
|
|
job_status[job_id] = {
|
|
'status': 'starting',
|
|
'progress': 0,
|
|
'message': 'Initializing video generation...',
|
|
'video_path': None,
|
|
'gcs_uri': None,
|
|
'error': None
|
|
}
|
|
|
|
try:
|
|
# Set up authentication
|
|
credentials = get_google_credentials()
|
|
|
|
# Configure the GenAI client for Vertex AI
|
|
if credentials:
|
|
client = genai.Client(
|
|
vertexai=True,
|
|
project=Config.PROJECT_ID,
|
|
location=Config.REGION,
|
|
credentials=credentials
|
|
)
|
|
else:
|
|
client = genai.Client(
|
|
vertexai=True,
|
|
project=Config.PROJECT_ID,
|
|
location=Config.REGION
|
|
)
|
|
|
|
# Define a unique prefix for this generation job in GCS
|
|
timestamp = int(time.time())
|
|
safe_prompt = prompt[:30].replace(' ', '_').replace('/', '_').replace('\\', '_')
|
|
output_gcs_uri = f"gs://{Config.OUTPUT_GCS_BUCKET_NAME}/veo_outputs/{safe_prompt}_{timestamp}/"
|
|
|
|
job_status[job_id].update({
|
|
'status': 'generating',
|
|
'progress': 10,
|
|
'message': 'Starting video generation...',
|
|
'gcs_uri': output_gcs_uri
|
|
})
|
|
|
|
# DEBUG: Log environment and SDK info
|
|
print("=== ENVIRONMENT DEBUG ===")
|
|
import sys
|
|
print(f"Python version: {sys.version}")
|
|
try:
|
|
import google.genai
|
|
print(f"google-genai version: {google.genai.__version__}")
|
|
except AttributeError:
|
|
print("google-genai version not available")
|
|
print(f"Model ID: {model_name or Config.MODEL_ID}")
|
|
print(f"Output GCS URI: {output_gcs_uri}")
|
|
print("=== END ENVIRONMENT DEBUG ===")
|
|
|
|
# Start the video generation operation using the updated SDK
|
|
print("Submitting video generation request...")
|
|
operation = client.models.generate_videos(
|
|
model=model_name or Config.MODEL_ID,
|
|
prompt=prompt,
|
|
config=types.GenerateVideosConfig(
|
|
person_generation=person_generation,
|
|
aspect_ratio=aspect_ratio,
|
|
output_gcs_uri=output_gcs_uri,
|
|
),
|
|
)
|
|
|
|
print(f"Video generation operation started: {operation.name}")
|
|
print("Waiting for operation to complete... This can take several minutes.")
|
|
|
|
job_status[job_id].update({
|
|
'status': 'processing',
|
|
'progress': 20,
|
|
'message': 'Video generation in progress...',
|
|
'operation_name': operation.name
|
|
})
|
|
|
|
# Poll for completion in a separate thread would be better for production
|
|
# For now, we'll do synchronous polling
|
|
while not operation.done:
|
|
print("Still processing...")
|
|
time.sleep(30) # Check every 30 seconds
|
|
operation = client.operations.get(operation)
|
|
|
|
# Update progress (approximate)
|
|
current_progress = min(80, job_status[job_id]['progress'] + 10)
|
|
job_status[job_id].update({
|
|
'progress': current_progress,
|
|
'message': 'Still processing video...'
|
|
})
|
|
|
|
if operation.error:
|
|
error_message = f"Video generation failed: {operation.error}"
|
|
print(error_message)
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': str(operation.error)
|
|
})
|
|
return job_id
|
|
|
|
# DEBUG: Log operation response details
|
|
print("=== OPERATION RESPONSE DEBUG ===")
|
|
print(f"Operation completed successfully: {operation.done}")
|
|
print(f"Operation response: {operation.response}")
|
|
print(f"Response type: {type(operation.response)}")
|
|
print(f"Response attributes: {[attr for attr in dir(operation.response) if not attr.startswith('_')]}")
|
|
|
|
if operation.response:
|
|
print(f"Response has data: True")
|
|
print(f"Response string representation: {str(operation.response)}")
|
|
|
|
# Check for different possible attribute names
|
|
possible_video_attrs = ['generated_videos', 'videos', 'video_uris', 'results', 'outputs', 'data']
|
|
for attr in possible_video_attrs:
|
|
if hasattr(operation.response, attr):
|
|
attr_value = getattr(operation.response, attr)
|
|
print(f"Found attribute '{attr}': {attr_value} (type: {type(attr_value)})")
|
|
else:
|
|
print(f"Attribute '{attr}' not found")
|
|
|
|
# Try to access as dict if it's dict-like
|
|
try:
|
|
if hasattr(operation.response, 'keys'):
|
|
print(f"Response keys: {list(operation.response.keys())}")
|
|
for key, value in operation.response.items():
|
|
print(f" {key}: {value} (type: {type(value)})")
|
|
except Exception as e:
|
|
print(f"Could not iterate as dict: {e}")
|
|
|
|
else:
|
|
print(f"Operation response is None or empty")
|
|
print("=== END DEBUG ===")
|
|
|
|
# Check if we have generated videos in the response
|
|
if not operation.response or not hasattr(operation.response, 'generated_videos') or not operation.response.generated_videos:
|
|
# Check if content was filtered by safety settings
|
|
if (hasattr(operation.response, 'rai_media_filtered_count') and
|
|
operation.response.rai_media_filtered_count > 0 and
|
|
hasattr(operation.response, 'rai_media_filtered_reasons')):
|
|
|
|
# Content was filtered - show the actual Google error message
|
|
filtered_reasons = operation.response.rai_media_filtered_reasons
|
|
error_message = f"Video generation blocked by content safety filters:\n\n{filtered_reasons[0] if filtered_reasons else 'Content filtered by safety settings'}"
|
|
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': 'Content filtered by safety settings'
|
|
})
|
|
else:
|
|
# Technical issue - no videos found
|
|
error_message = f"Operation completed, but no video URIs found in the response. See debug logs for response structure details."
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': error_message
|
|
})
|
|
|
|
return job_id
|
|
|
|
# Get the first generated video
|
|
first_video = operation.response.generated_videos[0]
|
|
|
|
# DEBUG: Log video object details
|
|
print("=== VIDEO OBJECT DEBUG ===")
|
|
print(f"First video object: {first_video}")
|
|
print(f"First video type: {type(first_video)}")
|
|
print(f"First video attributes: {[attr for attr in dir(first_video) if not attr.startswith('_')]}")
|
|
|
|
if hasattr(first_video, 'video'):
|
|
print(f"Video attribute exists: {first_video.video}")
|
|
print(f"Video attribute type: {type(first_video.video)}")
|
|
print(f"Video attributes: {[attr for attr in dir(first_video.video) if not attr.startswith('_')]}")
|
|
else:
|
|
print("No 'video' attribute found")
|
|
|
|
# Check for alternative URI locations
|
|
possible_uri_paths = [
|
|
('video', 'uri'),
|
|
('uri',),
|
|
('url',),
|
|
('gcs_uri',),
|
|
('output_uri',),
|
|
('file_uri',)
|
|
]
|
|
|
|
video_uri = None
|
|
for path in possible_uri_paths:
|
|
try:
|
|
obj = first_video
|
|
for attr in path:
|
|
obj = getattr(obj, attr)
|
|
video_uri = obj
|
|
print(f"Found URI at path {' -> '.join(path)}: {video_uri}")
|
|
break
|
|
except AttributeError:
|
|
print(f"Path {' -> '.join(path)} not found")
|
|
print("=== END VIDEO DEBUG ===")
|
|
|
|
if not hasattr(first_video, 'video') or not hasattr(first_video.video, 'uri'):
|
|
error_message = f"Video information in response does not contain video URI at expected path. See debug logs for actual structure."
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': error_message
|
|
})
|
|
return job_id
|
|
|
|
gcs_video_uri = first_video.video.uri
|
|
print(f"Video generated successfully. GCS URI: {gcs_video_uri}")
|
|
|
|
job_status[job_id].update({
|
|
'status': 'downloading',
|
|
'progress': 90,
|
|
'message': 'Downloading video...',
|
|
'gcs_video_uri': gcs_video_uri
|
|
})
|
|
|
|
# Parse GCS URI to get bucket name and blob name
|
|
if not gcs_video_uri.startswith("gs://"):
|
|
error_message = f"Invalid GCS URI received: {gcs_video_uri}"
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': error_message
|
|
})
|
|
return job_id
|
|
|
|
path_parts = gcs_video_uri.replace("gs://", "").split("/", 1)
|
|
source_bucket_name = path_parts[0]
|
|
source_blob_name = path_parts[1]
|
|
|
|
# Construct local file name
|
|
os.makedirs(Config.TEMP_DOWNLOAD_PATH, exist_ok=True)
|
|
video_filename = os.path.join(Config.TEMP_DOWNLOAD_PATH, f"{job_id}_{os.path.basename(source_blob_name)}")
|
|
if not video_filename.lower().endswith((".mp4", ".webm", ".avi", ".mov")):
|
|
video_filename += ".mp4"
|
|
|
|
# Download the video
|
|
local_video_path = download_blob(source_bucket_name, source_blob_name, video_filename)
|
|
|
|
job_status[job_id].update({
|
|
'status': 'completed',
|
|
'progress': 100,
|
|
'message': 'Video generation completed successfully!',
|
|
'video_path': local_video_path,
|
|
'source_bucket_name': source_bucket_name,
|
|
'source_blob_name': source_blob_name
|
|
})
|
|
|
|
except Exception as e:
|
|
error_message = f"An error occurred during video generation: {str(e)}"
|
|
job_status[job_id].update({
|
|
'status': 'failed',
|
|
'progress': 0,
|
|
'message': error_message,
|
|
'error': str(e)
|
|
})
|
|
|
|
return job_id
|
|
|
|
def get_job_status(job_id: str) -> dict:
|
|
"""Get the status of a video generation job."""
|
|
return job_status.get(job_id, {'status': 'not_found', 'message': 'Job not found'})
|
|
|
|
def cleanup_job_files(job_id: str) -> bool:
|
|
"""Clean up local and GCS files for a job."""
|
|
job = job_status.get(job_id)
|
|
if not job:
|
|
return False
|
|
|
|
success = True
|
|
|
|
# Delete local file
|
|
if job.get('video_path') and os.path.exists(job['video_path']):
|
|
try:
|
|
os.remove(job['video_path'])
|
|
print(f"Deleted local file: {job['video_path']}")
|
|
except Exception as e:
|
|
print(f"Error deleting local file: {e}")
|
|
success = False
|
|
|
|
# Delete GCS file
|
|
if job.get('source_bucket_name') and job.get('source_blob_name'):
|
|
try:
|
|
delete_blob(job['source_bucket_name'], job['source_blob_name'])
|
|
except Exception as e:
|
|
print(f"Error deleting GCS file: {e}")
|
|
success = False
|
|
|
|
# Remove job from memory
|
|
if success:
|
|
del job_status[job_id]
|
|
|
|
return success |