"""Modal runner for generating shareable assets.""" import pathlib import modal import boto3 import warnings import tempfile import os import subprocess S3_BUCKET_NAME = "suno-data-uploads" S3_PREFIX = "studio/uploads" # Import the new shader rendering function from suno_utils.worker.music_video_gen.lib.render_video import render_video_with_audio from suno_utils.worker.music_video_gen.lib.audio_trimmer import download_and_trim_audio from suno_utils.worker.music_video_gen.lib.video_trimmer import download_and_trim_video from suno_utils.worker.schema import QueueItem DEPLOYMENT_TYPE = "dev" APP_NAME = f"hook-video-gen-{DEPLOYMENT_TYPE}" EVENTS_QUEUE_NAME = f"hook-video-gen-{DEPLOYMENT_TYPE}" events_queue = modal.Queue.from_name(EVENTS_QUEUE_NAME, create_if_missing=True) SECRETS = [ modal.Secret.from_name("studio-aws"), # For S3 upload modal.Secret.from_name("api-callback-token"), ] base_image = ( modal.Image.debian_slim(python_version="3.10") .apt_install( "curl", "unzip", ) .apt_install( "ffmpeg", ) .pip_install( "moderngl[headless]==5.8.2", # Specify version to ensure compatibility "skia-python==87.7", "requests", "tqdm", "numpy>=1.20.0", # Ensure numpy is installed for arrays "pydub", # For audio processing "scipy", # For FFT and other signal processing "Pillow", "boto3", ) .pip_install( "boto3", "skia-python", ) .pip_install_from_pyproject( str(pathlib.Path(__file__).parent.parent.parent.parent / "pyproject.toml"), ) .add_local_python_source("suno_utils", copy=False) ) app = modal.App(APP_NAME) # Filter warnings for specific messages warnings.filterwarnings("ignore", message=".*unknown element.*reset-dirs.*") @app.cls( secrets=SECRETS, cpu=1, memory=1024, timeout=360, scaledown_window=300, min_containers=0 if DEPLOYMENT_TYPE == "dev" else 1, # Scale down to 0 in dev max_containers=50, image=base_image, region="us-east", ) @modal.concurrent(max_inputs=4) class HookVideoGenerator: """ HookVideoGenerator which generate a video based on the schema """ def __init__(self): print("hook video generator initialized.") @modal.method() def create_hook_video(self, queue_item_json: str): """ Create a hook video based on the queue item. """ try: local_temp_dir = tempfile.mkdtemp(prefix="suno_render_job_") local_asset_dir = os.path.join(local_temp_dir, "assets") queue_item = QueueItem.model_validate_json(queue_item_json) # id = queue_item.id queue_item.notify_progress({"id": queue_item.id, "status": "started"}) metadata = queue_item.metadata hook_id = metadata["hook_id"] print("hook_id", hook_id) print("metadata", metadata) video_render = metadata["video_render"] if isinstance(video_render, str): import json video_render = json.loads(video_render) print("video_render", video_render) song_schema = video_render["song"] song_id = song_schema["song_id"] clip_start_time = song_schema["clip_start_time"] clip_end_time = song_schema["clip_end_time"] video = video_render["videos"][0] video_s3_id = video["s3_id"] video_source_start_time = video["source_start_time"] video_source_end_time = video["source_end_time"] # Initialize S3 client for downloading/uploading s3_client = boto3.client("s3") audio_s3_key = f"studio/uploads/{song_id}.mp3" video_url = f"https://cdn1.suno.ai/{video_s3_id}.mp4" # Use the original download_and_trim_audio function # which handles S3 download and trimming in one step audio_result = download_and_trim_audio( S3_BUCKET_NAME, audio_s3_key, clip_start_time, clip_end_time, local_asset_dir ) queue_item.notify_progress({"id": queue_item.id, "status": "generating video"}) print("audio_result", audio_result) video_dir = local_asset_dir + f"/{video_s3_id}.mp4" trimmed_video_path = download_and_trim_video( video_url, video_dir, video_source_start_time, video_source_end_time, ) generated_video_path = local_asset_dir + f"/generated_{video_s3_id}.mp4" song = video_render["song"] # default to 100 if not provided if "volume" not in song: song["volume"] = 100 video = video_render["videos"][0] if "volume" not in video: video["volume"] = 0 render_video_with_audio( audio_result["trimmed_audio_path"], trimmed_video_path, generated_video_path, clip_end_time - clip_start_time, video["volume"], song["volume"], ) ########### S3 Upload ########### # # Define S3 bucket and key bucket_name = "suno-data-uploads" s3_video_key = f"studio/uploads/hook_{hook_id}.mp4" s3_client.upload_file( generated_video_path, bucket_name, s3_video_key, ExtraArgs={"ContentType": "video/mp4"}, ) # get the video from generated_video_path video_duration = self.get_video_duration(generated_video_path) # ########## print(f"Successfully uploaded {generated_video_path} to s3://{bucket_name}/{s3_video_key}") queue_item.notify_progress( { "id": queue_item.id, "status": "finished", "video_url": f"https://cdn1.suno.ai/hook_{hook_id}.mp4", "video_s3_id": f"hook_{hook_id}", "video_duration": video_duration, } ) print("finished, video_duration", video_duration) except Exception as e: print("error", e) queue_item.notify_progress({"id": queue_item.id, "status": "error", "error": str(e)}) raise e def get_video_duration(self, video_path: str) -> float: """ Get the duration of a video file in seconds using ffprobe. Args: video_path (str): Path to the video file Returns: float: Duration of the video in seconds Raises: subprocess.CalledProcessError: If ffprobe fails to get the duration ValueError: If the duration cannot be parsed from ffprobe output """ try: duration = float( subprocess.check_output( [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", video_path, ] ).strip() ) return duration except subprocess.CalledProcessError as e: raise RuntimeError( f"Failed to get video duration: {e.stderr.decode() if e.stderr else str(e)}" ) except ValueError as e: raise ValueError(f"Could not parse duration from ffprobe output: {e}") @app.local_entrypoint() def main(): generator = HookVideoGenerator() # # test video with no audio dumped_json = { "song": { "song_id": "8511d294-ba7d-40b6-aef0-1e8dedb7543a", "clip_start_time": 0.0, "clip_end_time": 3, "lyric": None, "volume": 100, }, "text_sticker": None, "videos": [ { "upload_id": "00672e3c-5535-4b6c-84a0-b2432533a638", "s3_id": "video_upload_f84dc486-921b-412c-9c14-567f731fb609", "source_start_time": 0.0, "source_end_time": 4, "start_time": 0.0, "end_time": 0.0, "volume": 100, }, ], "images": None, "shader": None, } print("dumped_json", dumped_json) queue_item = ( QueueItem( id="test_render_hook_video", metadata={ "hook_id": "video_with_no_audio", "video_render": dumped_json, }, ) ).json() generator.create_hook_video.remote(queue_item) # test video with audio dumped_json = { "song": { "song_id": "8511d294-ba7d-40b6-aef0-1e8dedb7543a", "clip_start_time": 0.0, "clip_end_time": 15, "volume": 0, }, "text_sticker": None, "videos": [ { "upload_id": "7cce5134-5c3e-494f-9e68-92e58aafb800", "s3_id": "video_upload_7cce5134-5c3e-494f-9e68-92e58aafb800", "source_start_time": 0.0, "source_end_time": 15, "start_time": 0.0, "end_time": 0.0, "volume": 100, }, ], "images": None, "shader": None, } print("dumped_json", dumped_json) queue_item = ( QueueItem( id="test_render_hook_video", metadata={ "hook_id": "video_with_audio4", "video_render": dumped_json, }, ) ).json() generator.create_hook_video.remote(queue_item) print("done")