"""Hook video generation worker integration""" import json import logging import time from typing import Any, Dict from pipeline.constants import HOOK_VIDEO_GEN_WORKER_PATH, SUNO_UTILS_PATH logger = logging.getLogger(__name__) class HookVideoWorker: """Client for the Modal hook video generation worker""" def __init__(self, deployment: str = "dev"): self.deployment = deployment self.app_name = f"hook-video-gen-{deployment}" def generate_video( self, input_s3_key: str, hook_data: Dict[str, Any] ) -> Dict[str, Any]: """Run the hook_video_gen_worker.py worker using actual hook data Args: input_s3_key: S3 key of the input video hook_data: Hook data from database Returns: Dictionary with generation results """ logger.info("Running hook video generation worker...") hook_id = f"test-reprocess-{hook_data.get('hook_id')}-{int(time.time())}" # Use the actual render_schema from VideoHookMetadata render_schema = hook_data.get("render_schema") if isinstance(render_schema, str): render_schema = json.loads(render_schema) # If no render_schema, use defaults if not render_schema: logger.error("No render_schema found, cannot generate hook video") raise ValueError("No render_schema found") else: # Update the s3_id in the render_schema to use the existing video # Use the original video from the database if the upload worker didn't produce output if render_schema.get("videos") and len(render_schema["videos"]) > 0: # Keep the original s3_id from the database unless we have a new one original_video_s3_id = render_schema["videos"][0].get("s3_id", "") # Only update if we have a new processed video if "test-upload" in input_s3_key: render_schema["videos"][0]["s3_id"] = input_s3_key.replace( "studio/uploads/", "" ).replace(".mp4", "") render_schema["videos"][0]["upload_id"] = input_s3_key.replace( "studio/uploads/", "" ).replace(".mp4", "") else: # Use the original video s3_id from the database logger.info(f"Using original video s3_id: {original_video_s3_id}") logger.info( f"Using render_schema config: {json.dumps(render_schema, indent=2)}" ) # Create queue item for the worker matching expected format output_prefix = "tests/hook-transcode-tests" queue_item = { "id": hook_id, "metadata": { "hook_id": hook_id, "platform": "test", "video_render": json.dumps(render_schema), "is_reprocess": True, "skip_audio_polling": True, # Skip polling for test "s3_folder": output_prefix, }, } queue_item_json = json.dumps(queue_item) try: import os import select import subprocess import tempfile # Run the Modal function locally from the glockenspiel/suno_utils directory logger.info(f"Running Modal function locally from {SUNO_UTILS_PATH}") # Create a temporary file for the queue item JSON with tempfile.NamedTemporaryFile( mode="w", suffix=".json", delete=False ) as f: f.write(queue_item_json) queue_item_file = f.name try: # Run the Modal function using uv run, passing queue item via stdin cmd = ["uv", "run", "modal", "run", HOOK_VIDEO_GEN_WORKER_PATH] logger.info(f"Running command: {' '.join(cmd)}...") # Run Modal with real-time output streaming with open(queue_item_file, "r") as f: process = subprocess.Popen( cmd, cwd=SUNO_UTILS_PATH, stdin=f, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, # Line buffered universal_newlines=True, ) # Collect output while streaming to logger stdout_lines = [] stderr_lines = [] # Read stdout and stderr in real-time while True: # Check if process is done if process.poll() is not None: break # Use select to read available data from stdout/stderr ready, _, _ = select.select( [process.stdout, process.stderr], [], [], 0.1 ) for stream in ready: if stream == process.stdout: line = stream.readline() if line: line = line.rstrip() stdout_lines.append(line) # Log Modal output with [HOOK-GEN] prefix logger.info(f"[HOOK-GEN] {line}") elif stream == process.stderr: line = stream.readline() if line: line = line.rstrip() stderr_lines.append(line) # Log Modal errors/warnings if "ERROR" in line or "error" in line: logger.error(f"[HOOK-GEN] {line}") elif "WARNING" in line or "warning" in line: logger.warning(f"[HOOK-GEN] {line}") else: logger.info(f"[HOOK-GEN STDERR] {line}") # Read any remaining output remaining_stdout = process.stdout.read() remaining_stderr = process.stderr.read() if remaining_stdout: for line in remaining_stdout.strip().split("\n"): if line: stdout_lines.append(line) logger.info(f"[HOOK-GEN] {line}") if remaining_stderr: for line in remaining_stderr.strip().split("\n"): if line: stderr_lines.append(line) logger.info(f"[HOOK-GEN STDERR] {line}") # Wait for process to complete and get return code return_code = process.wait() if return_code != 0: raise subprocess.CalledProcessError( return_code, cmd, output="\n".join(stdout_lines), stderr="\n".join(stderr_lines), ) logger.info("Hook video generation complete") # Parse the output to extract the result result_data = None for line in reversed(stdout_lines): if line.startswith("Result:"): # Extract JSON from "Result: {...}" line try: result_json = line.replace("Result:", "").strip() result_data = json.loads(result_json) break except json.JSONDecodeError: pass else: # Try parsing the line as JSON directly try: result_data = json.loads(line) break except json.JSONDecodeError: continue if not result_data: # If no JSON found, log a warning but don't include raw output in results logger.warning("No JSON result found in Modal output") result_data = { "status": "completed", "message": "Processing completed but no structured result returned", } else: logger.info("Successfully parsed result from Modal output") output_s3_key = f"{output_prefix}/hook_{hook_id}.mp4" return { "hook_id": hook_id, "output_s3_key": output_s3_key, "result": result_data, } finally: # Clean up temporary file if os.path.exists(queue_item_file): os.unlink(queue_item_file) except subprocess.CalledProcessError as e: logger.error(f"Error running hook video gen worker: {e}") logger.error(f"stdout: {e.stdout}") logger.error(f"stderr: {e.stderr}") raise except Exception as e: logger.error(f"Error running hook video gen worker: {e}") raise