import modal import aiohttp import asyncio import time import uuid from pydub import AudioSegment from typing import Any, List, Optional from pydantic import BaseModel STARTER_CLIP = "/Users/zoeshleifer/Downloads/Untitled.mp3" import random RAND_TAG_LIST = [ 'acoustic', 'aggressive', 'anthemic', 'atmospheric', 'bouncy', 'chill', 'dark', 'dreamy', 'electronic', 'emotional', 'epic', 'experimental', 'futuristic', 'groovy', 'heartfelt', 'infectious', 'melodic', 'mellow', 'powerful', 'psychedelic', 'romantic', 'smooth', 'syncopated', 'uplifting', 'afrobeat', 'anime', 'ballad', 'bedroom pop', 'bluegrass', 'blues', 'classical', 'country', 'cumbia', 'dance', 'delta blues', 'electropop', 'disco', 'drum and bass', 'edm', 'emo', 'folk', 'funk', 'future bass', 'gospel', 'grunge', 'grime', 'hip hop', 'house', 'indie', 'j-pop', 'jazz', 'k-pop', 'kids music', 'metal', 'new jack swing', 'new wave', 'opera', 'punk', 'raga', 'rap', 'reggae', 'reggaeton', 'rock', 'rumba', 'salsa', 'samba', 'sertanejo', 'soul', 'synthpop', 'swing', 'synthwave', 'techno', 'trap', 'uk garage' ]; env_name = "dev" async def call_modal_orchestrator(model_name: str, queue_item): f = getattr(modal.Cls.lookup(f"orchestrator-{env_name}", "ChatGptStub")(), "moderate_and_generate") fn = await f.spawn.aio(queue_item.model_dump_json()) return fn class QueueItem(BaseModel): id: str # clip ID prompt_audio: Optional[str] # S3 ID of audio to extend from prompt_text: Optional[str] # lyrics/prompt text metadata: Any # All the extend metadata callback_url: Optional[str] # Where to post results model_name: Optional[str] # e.g., "chirp-v4" title: Optional[str] # Clip title ids: Optional[List[str]] # List of clip IDs if batch_all=True def create_extend_metadata(continue_clip, params): metadata = continue_clip.metadata.copy() keys_to_remove = [ "duration", "refund_credits", "experiment", "param_experiment", "experiment_version", ] for key in keys_to_remove: metadata.pop(key, None) metadata.update( { "continued_from_prompt": continue_clip.metadata.get("prompt"), "continued_aligned_prompt": params.get("continued_aligned_prompt", ""), "history": continue_clip.metadata.get("history", []) + [ { "id": continue_clip.s3_id or str(continue_clip.id), "continue_at": params["continue_at"], # timestamp in seconds "type": continue_clip.metadata.get("type", "gen"), "source": "web", } ], "continue_at": params["continue_at"], "task": "extend", # or "upload_extend" "edited_clip_id": str(continue_clip.id), "type": "gen", "source": "web", "prompt": params["prompt"], "tags": params.get("tags"), "gpt_description_prompt": params.get("gpt_description_prompt"), "user_id": params["user_id"], } ) return metadata async def poll_for_mp3(s3_url): start_time = time.time() attempt_count = 0 async with aiohttp.ClientSession() as session: while True: attempt_count += 1 elapsed_time = time.time() - start_time try: headers = {"Range": "bytes=0-1023"} # Just get first 1KB to test async with session.get(s3_url, headers=headers) as response: if response.status in [200, 206]: total_elapsed = time.time() - start_time print(f"MP3 file found at {s3_url}") print( f"Time elapsed: {total_elapsed:.1f} seconds ({attempt_count} attempts)" ) # Now download the full file async with session.get(s3_url) as download_response: assert download_response.status == 200 content = await download_response.read() # Generate filename based on timestamp timestamp = int(time.time()) original_filename = f"extended_song_{timestamp}.mp3" # Save original file temporarily with open(original_filename, "wb") as f: f.write(content) return original_filename except Exception as e: print( f"Error polling URL (attempt {attempt_count}, {elapsed_time:.1f}s elapsed): {e}" ) await asyncio.sleep(5) SAMPLE_RATE = 24000 MIN_LENGTH = 90 MIN_VOLUME = 0.5 from pydub import AudioSegment def has_sustained_silence(audio, min_length_s=90, silence_threshold_dbfs=-50.0, silence_duration_ms=1000, chunk_ms=100): """ Return True if sustained silence (below threshold) occurs before MIN_LENGTH seconds. """ if len(audio) < min_length_s * 1000: return True num_chunks = silence_duration_ms // chunk_ms silent_chunks = 0 for t in range(0, len(audio), chunk_ms): if t > min_length_s * 1000: return False # Song is long enough chunk = audio[t:t + chunk_ms] if chunk.dBFS < silence_threshold_dbfs: silent_chunks += 1 if silent_chunks >= num_chunks: return True else: silent_chunks = 0 async def extend_clip_minimal(continue_clip, extend_params): """Minimal extend implementation with correct batch_size""" # print(f"DEBUG: Starting extend_clip_minimal") request_id = str(uuid.uuid4()) # print(f"DEBUG: Request ID: {request_id}") # Generate 2 clip IDs (Suno always generates 2) clip_ids = [str(uuid.uuid4()), str(uuid.uuid4())] # print(f"DEBUG: Generated clip IDs: {clip_ids}") # 1. Create metadata # print(f"DEBUG: Creating metadata...") metadata = create_extend_metadata(continue_clip, extend_params) print(f"DEBUG: Metadata created: {metadata}") # 2. Create QueueItem with 2 IDs # print(f"DEBUG: Creating QueueItem...") queue_item = QueueItem( id=request_id, # Request ID prompt_audio=f"{continue_clip.s3_id}.mp3", prompt_text=extend_params["prompt"], metadata=metadata, callback_url="http://platypus.han-mahi.ts.net", model_name="chirp-v4", title=extend_params.get("title", "Extended Song"), ids=clip_ids, # ← Must be 2 UUIDs ) # print(f"DEBUG: QueueItem created: {queue_item.model_dump()}") # 3. Call Modal # print(f"DEBUG: About to call modal orchestrator...") result = await call_modal_orchestrator("chirp-v4", queue_item) # print(f"DEBUG: Modal orchestrator returned: {result}") return clip_ids, result # Return both IDs to track both clips def extend(tags, neg_tags, continue_clip, continue_at = 90): print(f"DEBUG: Starting extend with tags: {tags}") # Create extend parameters with more complete configuration extend_params = { "continue_at": continue_at, # Continue at 90 seconds "prompt": "", "user_id": "sunoforever", "control_sliders": { "style_weight": 0.5, "weirdness_constraint": 0.5, "audio_strength": 0.5, }, "make_instrumental": 1, "tags": tags, # Include both fixed and random tags "negative_tags": "repetitive, loop"+neg_tags, "title": " ".join(tags), "continued_aligned_prompt": "", "gpt_description_prompt": " ".join(tags), } result = asyncio.run(extend_clip_minimal(continue_clip, extend_params)) # Extract S3 URL from result and poll for MP3 if result and len(result) > 0: s3_url_1 = f"https://cdn1.suno.ai/{result[0][0]}.mp3" s3_url_2 = f"https://cdn1.suno.ai/{result[0][1]}.mp3" downloaded_file_1 = asyncio.run(poll_for_mp3(s3_url_1)) downloaded_file_2 = asyncio.run(poll_for_mp3(s3_url_2)) audio_1 = AudioSegment.from_mp3(downloaded_file_1).set_channels(1) audio_2 = AudioSegment.from_mp3(downloaded_file_2).set_channels(1) if downloaded_file_1 and downloaded_file_2: silence_time_1 = has_sustained_silence(audio_1) silence_time_2 = has_sustained_silence(audio_2) if silence_time_1 and silence_time_2: return STARTER_CLIP, continue_clip.s3_id if silence_time_1: print(f"DEBUG: returning second clip") return downloaded_file_2, result[0][1] if silence_time_2: print(f"DEBUG: returning first clip") return downloaded_file_1, result[0][0] elif len(audio_1) > len(audio_2): print(f"DEBUG: returning first clip") return downloaded_file_1, result[0][0] else: print(f"DEBUG: returning second clip") return downloaded_file_2, result[0][1] else: return STARTER_CLIP, continue_clip.s3_id else: return STARTER_CLIP, continue_clip.s3_id