"""Video upload worker integration""" import json import logging import time from typing import Any, Dict from pipeline.constants import MODAL_RUNNER_VIDEO_UPLOAD_PATH, SUNO_UTILS_PATH logger = logging.getLogger(__name__) class VideoUploadWorker: """Client for the Modal video upload worker""" def __init__(self, deployment: str = "dev"): self.deployment = deployment self.app_name = f"upload-video-{deployment}" def process_video(self, s3_key: str, hook_data: Dict[str, Any]) -> Dict[str, Any]: """Run the modal_runner_video_upload.py worker locally Args: s3_key: S3 key of the input video hook_data: Hook data from database Returns: Dictionary with processing results """ import os import subprocess logger.info("Running video upload worker locally...") upload_id = ( f"test-upload-{hook_data.get('hook_id', 'unknown')}-{int(time.time())}" ) # Create queue item for the worker - matching the expected format queue_item = { "id": upload_id, "metadata": { "upload_key": s3_key, "video_upload_type": "video_hook", "require_video_sprite": True, "fail_task_on_moderation": False, "skip_moderation": True, "skip_downsampling": True, "is_video_cover": False, "is_reprocess": True, # This is a reprocess since we're testing existing content }, "title": f"Test Video Upload for Hook {hook_data.get('hook_id')}", } queue_item_json = json.dumps(queue_item) try: # Run the Modal function locally from the glockenspiel/suno_utils directory logger.info(f"Running Modal function locally from {SUNO_UTILS_PATH}") # Create a Python script to run the Modal function with the queue item runner_script = f""" import json import sys sys.path.insert(0, "{SUNO_UTILS_PATH}") from suno_utils.worker.video_processing.modal_runner_video_upload import VideoUploadStub # Create the stub instance and run the method stub = VideoUploadStub() queue_item_json = '''{queue_item_json}''' result = stub.process_uploaded_video(queue_item_json) print(json.dumps(result)) """ # Write the runner script to a temporary file import tempfile with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(runner_script) temp_script_path = f.name try: # Write queue item JSON to a temporary file to pass as stdin queue_item_file = temp_script_path.replace(".py", "_queue.json") with open(queue_item_file, "w") as f: f.write(queue_item_json) # Run the Modal function using uv run, passing queue item via stdin cmd = ["uv", "run", "modal", "run", MODAL_RUNNER_VIDEO_UPLOAD_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 import select 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 [MODAL] prefix logger.info(f"[MODAL] {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"[MODAL] {line}") elif "WARNING" in line or "warning" in line: logger.warning(f"[MODAL] {line}") else: logger.info(f"[MODAL 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"[MODAL] {line}") if remaining_stderr: for line in remaining_stderr.strip().split("\n"): if line: stderr_lines.append(line) logger.info(f"[MODAL 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("Video upload processing 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") # Extract the S3 key from the result output_key = f"studio/uploads/video_upload_{upload_id}.mp4" return { "result": result_data, "output_s3_key": output_key, "upload_id": upload_id, } finally: # Clean up temporary files if os.path.exists(temp_script_path): os.unlink(temp_script_path) if "queue_item_file" in locals() and os.path.exists(queue_item_file): os.unlink(queue_item_file) except subprocess.CalledProcessError as e: logger.error(f"Error running video upload worker: {e}") logger.error(f"stdout: {e.stdout}") logger.error(f"stderr: {e.stderr}") raise except Exception as e: logger.error(f"Error running video upload worker: {e}") raise