"""Modal ASGI app for uploading audio to Suno Studio API""" import asyncio import random import logging import os import subprocess import tempfile import traceback import uuid from collections import defaultdict, deque from datetime import datetime from typing import Deque, Dict, List, Optional import httpx import modal from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel, validator logger = logging.getLogger(__name__) # Modal app configuration image = ( modal.Image.debian_slim(python_version="3.11") .apt_install("ffmpeg") # For video format conversion .pip_install( "fastapi[standard]>=0.100.0", "httpx>=0.24.0", ) ) app = modal.App( "flappy-dodo-audio-upload", image=image, secrets=[ modal.Secret.from_name("chi-staging-token"), # os.environ["SUNO_TOKEN"] ], ) SUNO_STUDIO_API_BASE_URL = "https://studio-api.staging.suno.com" LEADERBOARD_MAX_ENTRIES = 50 leaderboard_store: Dict[str, Deque[dict]] = defaultdict( lambda: deque(maxlen=LEADERBOARD_MAX_ENTRIES) ) ALLOWED_DIFFICULTIES = { "practice", "easy", "hard", "easy_60s", "hard_60s", } DEFAULT_DIFFICULTY = "practice" PRACTICE_TIPS = [ "Feather your jumps to hover and align with the widest part of each gate.", "Trigger jumps just before entering the gap to keep recovery time comfortable.", "Stack short taps instead of long presses to maintain steady altitude.", "Ride the lower half of the screen to give yourself reaction time for rising pipes.", ] ARTIST_SPOTLIGHTS = [ { "id": "mozart", "name": "Wolfgang Amadeus Mozart", "genres": ["classical"], "recommended_challenge": "mozart_variations", "suggested_difficulty": "practice", "tip": "Shape light arpeggios across the gapβ€”think elegant Viennese ballroom phrasing.", }, { "id": "elvis", "name": "Elvis Presley", "genres": ["rock", "pop"], "recommended_challenge": "elvis_stagecraft", "suggested_difficulty": "easy", "tip": "Bounce with a backbeat rhythm: short tap on beat one, longer hold on beat three.", }, { "id": "hendrix", "name": "Jimi Hendrix", "genres": ["psychedelic", "rock"], "recommended_challenge": "hendrix_feedback", "suggested_difficulty": "hard", "tip": "Blend glide inputs to mimic bending guitar stringsβ€”smooth transitions keep the solo alive.", }, ] ARTIST_INSIGHTS = [ { "id": "mozart", "artist": "Wolfgang Amadeus Mozart", "signature_ability": "Virtuoso Variations", "mentor_tip": "Chain arpeggios smoothly and let the melody breathe between phrases.", "genres": ["classical", "baroque"], "recommended_challenges": ["mozart_variations", "mozart_duet"], }, { "id": "elvis", "artist": "Elvis Presley", "signature_ability": "Hip Swing Hype", "mentor_tip": "Lean on the backbeat and let the swagger carry every phrase.", "genres": ["rock", "pop"], "recommended_challenges": ["elvis_stagecraft", "elvis_crossover"], }, { "id": "hendrix", "artist": "Jimi Hendrix", "signature_ability": "Electric Sky Dive", "mentor_tip": "Balance wild experimentation with soulful blues phrasing.", "genres": ["psychedelic", "rock"], "recommended_challenges": ["hendrix_feedback", "hendrix_live"], }, { "id": "beatles", "artist": "The Beatles", "signature_ability": "Studio Alchemy", "mentor_tip": "Layer harmonies fearlessly and blend acoustic warmth with sonic surprises.", "genres": ["rock", "psychedelic"], "recommended_challenges": ["beatles_merseybeat", "beatles_studio"], }, { "id": "queen", "artist": "Queen", "signature_ability": "Stadium Crescendo", "mentor_tip": "Stack choirs of vocals and punctuate every phrase with drama.", "genres": ["rock", "progressive"], "recommended_challenges": ["queen_anthem", "queen_opera"], }, { "id": "michael_jackson", "artist": "Michael Jackson", "signature_ability": "Moonwalk Momentum", "mentor_tip": "Tighten every rhythm hit and stack silky vocal harmonies.", "genres": ["pop", "r&b", "funk"], "recommended_challenges": ["mj_thriller", "mj_stage_show"], }, { "id": "madonna", "artist": "Madonna", "signature_ability": "Vogue Velocity", "mentor_tip": "Own the hook, lean into provocative lyrics, and keep the beat fashion-forward.", "genres": ["pop", "dance"], "recommended_challenges": ["madonna_club", "madonna_manifesto"], }, { "id": "miles_davis", "artist": "Miles Davis", "signature_ability": "Modal Mosaic", "mentor_tip": "Leave space in your phrasing and let the rhythm section breathe.", "genres": ["jazz", "fusion"], "recommended_challenges": ["miles_blue", "jazz_combo", "fusion_power_chords"], }, ] @app.function( cpu=1, memory=1024, timeout=300, # 5 minutes for full upload + processing min_containers=0, ) @modal.asgi_app() def api(): """Create FastAPI app for audio upload""" logging.basicConfig(level=logging.INFO) web_app = FastAPI(title="Flappy Dodo Audio Upload API") web_app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) class GenerateCoverRequest(BaseModel): cover_clip_id: str style: str = "marimba" title: Optional[str] = None make_instrumental: bool = True is_mumble: bool = False task: str = "cover" # "cover" or "sample_condition" class UploadAndCoverRequest(BaseModel): styles: List[str] = [ "marimba" ] # Can generate multiple covers with different styles class CreateHookRequest(BaseModel): clip_id: str title: str = "Flappy DODO Performance" start_clip_timestamp: float = 0.0 end_clip_timestamp: Optional[float] = None class LeaderboardEntry(BaseModel): player: str difficulty: str score: int duration: float mode: str = "keyboard" recorded_at: Optional[str] = None @validator("player") def validate_player(cls, value: str) -> str: value = (value or "").strip() return value or "Anonymous" @validator("difficulty") def validate_difficulty(cls, value: str) -> str: if not value: raise ValueError("difficulty is required") return value.strip().lower() @validator("score") def validate_score(cls, value: int) -> int: if value < 0: raise ValueError("score must be non-negative") return value @validator("duration") def validate_duration(cls, value: float) -> float: if value < 0: raise ValueError("duration must be non-negative") return value @validator("mode") def validate_mode(cls, value: str) -> str: value = (value or "keyboard").strip().lower() return value or "keyboard" def get_suno_token() -> str: token = os.environ.get("SUNO_TOKEN", "").strip() if not token: logger.error("SUNO_TOKEN environment variable missing or empty") raise HTTPException(status_code=500, detail="SUNO_TOKEN not configured") token_lower = token.lower() if token_lower in {"changeme", "placeholder", "demo", "your_suno_token_here"}: logger.error("SUNO_TOKEN environment variable uses a placeholder value") raise HTTPException( status_code=503, detail="SUNO_TOKEN configured with a placeholder value. Provide a valid token.", ) return token def parse_httpx_response(response: httpx.Response, context: str) -> dict: if response.status_code < 400: try: return response.json() except ValueError as exc: logger.error("%s returned invalid JSON: %s", context, exc) raise HTTPException(status_code=502, detail=f"{context} returned invalid JSON") try: payload = response.json() except ValueError: payload = {"detail": response.text} logger.error("%s failed (%s): %s", context, response.status_code, payload) message = payload.get("detail") or payload.get("error") or context raise HTTPException(status_code=response.status_code, detail=message) async def httpx_post(url: str, *, json: Optional[dict] = None, data=None, files=None, headers=None, timeout: float = 30.0, context: str = "request") -> dict: try: async with httpx.AsyncClient(timeout=timeout) as client: response = await client.post(url, json=json, data=data, files=files, headers=headers) except httpx.RequestError as exc: logger.error("%s failed during transport: %s", context, exc) raise HTTPException(status_code=502, detail=f"{context} request failed") return parse_httpx_response(response, context) async def httpx_get(url: str, *, headers=None, timeout: float = 30.0, context: str = "request") -> dict: try: async with httpx.AsyncClient(timeout=timeout) as client: response = await client.get(url, headers=headers) except httpx.RequestError as exc: logger.error("%s failed during transport: %s", context, exc) raise HTTPException(status_code=502, detail=f"{context} request failed") return parse_httpx_response(response, context) @web_app.get("/") async def home(): """Health check endpoint""" return {"status": "ok", "service": "flappy-dodo-audio-upload"} @web_app.post("/upload-audio") async def upload_audio(file: UploadFile = File(...)): """ Upload audio file to Suno Studio API and return the final clip. This endpoint: 1. Requests an upload slot from Suno Studio API 2. Uploads the file to S3 3. Marks upload as finished 4. Polls for processing completion 5. Initializes the clip 6. Returns the final clip metadata """ try: token = get_suno_token() # Get file extension filename = file.filename or "audio.mp3" extension = filename.rsplit(".", 1)[-1] if "." in filename else "mp3" logger.info(f"Starting upload for file: {filename}") # Read file content file_content = await file.read() logger.info(f"Read {len(file_content)} bytes from uploaded file") # Step 1: Request upload slot upload_data = await request_upload_slot(token, extension) upload_id = upload_data["id"] s3_url = upload_data["url"] s3_fields = upload_data["fields"] logger.info(f"Got upload slot: {upload_id}") # Step 2: Upload to S3 await upload_to_s3(s3_url, s3_fields, file_content) logger.info("Uploaded to S3 successfully") # Step 3: Mark upload finished await mark_upload_finished(token, upload_id, filename) logger.info("Marked upload as finished") # Step 4: Poll for completion final_upload_data = await poll_upload_status(token, upload_id) logger.info( f"Upload processing complete, status: {final_upload_data.get('status')}" ) # Step 5: Initialize clip clip_data = await initialize_clip(token, upload_id) clip_id = clip_data.get("clip_id") logger.info(f"Initialized clip: {clip_id}") # Step 6: Get final clip details final_clip = await get_clip_details(token, clip_id) logger.info("Retrieved final clip details") return JSONResponse( content={"success": True, "upload_id": upload_id, "clip": final_clip} ) except HTTPException: raise except Exception as e: tb = traceback.format_exc() logger.exception(f"Error uploading audio: {e}") return JSONResponse( status_code=500, content={"error": str(e), "traceback": tb} ) async def request_upload_slot(token: str, extension: str) -> dict: """Request an upload slot from Suno Studio API""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/audio/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload = {"extension": extension} return await httpx_post( url, json=payload, headers=headers, context="Request upload slot" ) async def upload_to_s3(s3_url: str, s3_fields: dict, file_content: bytes): """Upload file to S3 using presigned POST""" # Build multipart form data files = { "file": ( "audio.mp3", file_content, s3_fields.get("Content-Type", "audio/mpeg"), ) } data = s3_fields.copy() try: async with httpx.AsyncClient(timeout=120.0) as client: resp = await client.post(s3_url, data=data, files=files) except httpx.RequestError as exc: logger.error("Failed to upload to S3 (transport): %s", exc) raise HTTPException(status_code=502, detail="S3 upload failed during transport") if resp.status_code >= 400: detail = resp.text logger.error("Failed to upload to S3: %s - %s", resp.status_code, detail) raise HTTPException( status_code=resp.status_code, detail=f"S3 upload failed: {detail}", ) async def mark_upload_finished(token: str, upload_id: str, filename: str): """Mark the upload as finished""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/audio/{upload_id}/upload-finish/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload = {"upload_type": "file_upload", "upload_filename": filename} await httpx_post( url, json=payload, headers=headers, context="Mark upload finished" ) async def poll_upload_status( token: str, upload_id: str, max_attempts: int = 60, delay_seconds: int = 2 ) -> dict: """Poll upload status until complete""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/audio/{upload_id}/" headers = { "Authorization": f"Bearer {token}", } for attempt in range(max_attempts): data = await httpx_get( url, headers=headers, context="Poll upload status" ) status = data.get("status") logger.info( f"Upload status (attempt {attempt + 1}/{max_attempts}): {status}" ) if status == "complete": return data if status == "error": raise HTTPException( status_code=500, detail=f"Upload processing failed: {data.get('error_message', 'Unknown error')}", ) await asyncio.sleep(delay_seconds) raise HTTPException(status_code=408, detail="Upload processing timed out") async def initialize_clip(token: str, upload_id: str) -> dict: """Initialize the clip from the upload""" url = ( f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/audio/{upload_id}/initialize-clip/" ) headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload = {} return await httpx_post( url, json=payload, headers=headers, context="Initialize clip" ) async def get_clip_details(token: str, clip_id: str) -> dict: """Get full clip details""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/clip/{clip_id}" headers = { "Authorization": f"Bearer {token}", } return await httpx_get( url, headers=headers, context="Get clip details" ) @web_app.post("/generate-cover") async def generate_cover(request: GenerateCoverRequest): """ Generate a cover of an uploaded song. This endpoint: 1. Calls the Suno generate API with cover parameters 2. Polls the clips until they reach "streaming" or "complete" status 3. Returns the final clips """ try: token = get_suno_token() logger.info( f"Generating cover for clip: {request.cover_clip_id} with style: {request.style}" ) # Step 1: Generate cover generation_data = await generate_cover_with_suno( token=token, cover_clip_id=request.cover_clip_id, style=request.style, title=request.title, make_instrumental=request.make_instrumental, is_mumble=request.is_mumble, task=request.task, ) clips = generation_data.get("clips", []) if not clips: raise HTTPException( status_code=500, detail="No clips returned from generation" ) clip_ids = [clip["id"] for clip in clips] logger.info(f"Generated {len(clip_ids)} clips: {clip_ids}") # Step 2: Poll until clips are ready final_clips = await poll_clips_until_ready(token, clip_ids) logger.info("All clips ready") return JSONResponse(content={"success": True, "clips": final_clips}) except HTTPException: raise except Exception as e: tb = traceback.format_exc() logger.exception(f"Error generating cover: {e}") return JSONResponse( status_code=500, content={"error": str(e), "traceback": tb} ) async def generate_cover_with_suno( token: str, cover_clip_id: str, style: str, title: Optional[str] = None, make_instrumental: bool = True, is_mumble: bool = False, task: str = "cover", cover_clip_prompt: str = "", ) -> dict: """Generate a cover using Suno Studio API""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/generate/v2-web/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } # Use the clip ID as the title if not provided cover_title = title or f"Cover {cover_clip_id[:8]}" # Build payload based on task type payload = { "project_id": None, "token": None, "task": task, "generation_type": "TEXT", "title": cover_title, "tags": style, "negative_tags": "", "mv": "chirp-bluejay-k", "prompt": cover_clip_prompt, "make_instrumental": make_instrumental, "user_uploaded_images_b64": None, "metadata": { "web_client_pathname": "/create", "is_max_mode": False, "is_mumble": is_mumble, "create_mode": "custom", "user_tier": "e31d9840-9a85-4d04-905a-46183e8cc322", "create_session_token": str(uuid.uuid4()), "disable_volume_normalization": False, "forced_infer_config": {}, "is_remix": task == "cover", "can_control_sliders": [], }, "override_fields": [], "persona_id": None, "artist_clip_id": None, "artist_start_s": None, "artist_end_s": None, "continue_clip_id": None, "continued_aligned_prompt": None, "continue_at": None, "transaction_uuid": str(uuid.uuid4()), } # Add task-specific fields if task == "cover": payload["cover_clip_id"] = cover_clip_id elif task == "sample_condition": payload["sample_clip_ids"] = [cover_clip_id] payload["cover_clip_id"] = None else: payload["cover_clip_id"] = cover_clip_id return await httpx_post( url, json=payload, headers=headers, timeout=60.0, context="Generate cover request" ) async def poll_clips_until_ready( token: str, clip_ids: List[str], max_attempts: int = 60, delay_seconds: int = 2 ) -> List[dict]: """Poll clips until they reach streaming or complete status""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/feed/v2" headers = { "Authorization": f"Bearer {token}", } # Join clip IDs for the query parameter ids_param = ",".join(clip_ids) for attempt in range(max_attempts): data = await httpx_get( f"{url}?ids={ids_param}", headers=headers, context="Poll generated clips" ) clips = data.get("clips", []) if not clips: logger.warning( f"No clips returned (attempt {attempt + 1}/{max_attempts})" ) await asyncio.sleep(delay_seconds) continue statuses = [clip.get("status") for clip in clips] logger.info( f"Clip statuses (attempt {attempt + 1}/{max_attempts}): {statuses}" ) all_ready = all( status in ["streaming", "complete"] for status in statuses ) if all_ready: return clips for clip in clips: if clip.get("status") == "error": raise HTTPException( status_code=500, detail=f"Clip {clip['id']} failed: {clip.get('error_message', 'Unknown error')}", ) await asyncio.sleep(delay_seconds) raise HTTPException(status_code=408, detail="Cover generation timed out") @web_app.post("/upload-video-and-create-hook") async def upload_video_and_create_hook( file: UploadFile = File(...), clip_id: str = Form(...), title: str = Form("Flappy DODO Performance"), ): """ Upload gameplay video and create a hook. This endpoint: 1. Uploads the video to Suno (converts WebM to MP4 if needed) 2. Polls until video processing is complete 3. Creates a hook with the uploaded video (uses full clip and video durations) Args: file: Video file (webm/mp4) clip_id: The Suno clip ID to use for the hook audio title: Title for the hook Returns: { "success": true, "upload_id": "...", "video_s3_id": "...", "hook": {...} } """ try: token = get_suno_token() filename = file.filename or "gameplay.webm" original_extension = ( filename.rsplit(".", 1)[-1] if "." in filename else "webm" ) logger.info( f"Starting video upload and hook creation for clip {clip_id}, file: {filename}" ) # Read file content file_content = await file.read() logger.info( f"πŸ“Ή Read {len(file_content)} bytes ({len(file_content) / 1024 / 1024:.2f} MB) from video file" ) if len(file_content) == 0: raise HTTPException(status_code=400, detail="Video file is empty") # Convert WebM to MP4 for better compatibility if original_extension.lower() in ["webm", "mkv"]: logger.info( f"πŸ”„ Converting {original_extension} to MP4 for better compatibility..." ) file_content, filename = await convert_to_mp4( file_content, original_extension ) extension = "mp4" logger.info( f"βœ… Converted to MP4: {len(file_content)} bytes ({len(file_content) / 1024 / 1024:.2f} MB)" ) else: extension = original_extension # Step 1: Request video upload slot video_upload_data = await request_video_upload_slot(token, extension) upload_id = video_upload_data["id"] s3_url = video_upload_data["url"] s3_fields = video_upload_data["fields"] logger.info(f"Got video upload slot: {upload_id}") # Step 2: Upload to S3 await upload_video_to_s3(s3_url, s3_fields, file_content, extension) logger.info(f"Uploaded video to S3 successfully: {s3_url}") logger.info(f"Video S3 fields: {s3_fields.get('key', 'N/A')}") # Step 3: Mark upload finished await mark_video_upload_finished(token, upload_id, clip_id, filename) logger.info("Marked video upload as finished") # Step 4: Poll for video processing completion final_video_data = await poll_video_upload_status(token, upload_id) video_s3_id = final_video_data.get("s3_id") video_duration = final_video_data.get("duration", 30) # Get video duration logger.info(f"βœ… Video processing complete, video_s3_id: {video_s3_id}") logger.info(f"πŸ“Š Final video data: {final_video_data}") logger.info(f"πŸ“Š Video duration: {video_duration}s") # Step 5: Get clip details to determine duration clip_details = await get_clip_details(token, clip_id) clip_duration = clip_details.get("metadata", {}).get("duration", 30) logger.info(f"πŸ“Š Clip duration: {clip_duration}s") # Step 6: Create hook (using full durations: 0 to duration) hook_data = await create_hook_with_video( token=token, clip_id=clip_id, upload_id=upload_id, title=title, clip_duration=clip_duration, video_duration=video_duration, ) logger.info(f"βœ… Created hook: {hook_data.get('id')}") return JSONResponse( content={ "success": True, "upload_id": upload_id, "video_s3_id": video_s3_id, "hook": hook_data, } ) except HTTPException: raise except Exception as e: tb = traceback.format_exc() logger.exception(f"Error uploading video and creating hook: {e}") return JSONResponse( status_code=500, content={"error": str(e), "traceback": tb} ) async def request_video_upload_slot(token: str, extension: str) -> dict: """Request a video upload slot from Suno Studio API""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/video/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload = {"extension": extension} return await httpx_post( url, json=payload, headers=headers, context="Request video upload slot" ) async def upload_video_to_s3( s3_url: str, s3_fields: dict, file_content: bytes, extension: str ): """Upload video file to S3 using presigned POST""" logger.info(f"πŸ“€ Uploading {len(file_content)} bytes to S3") logger.info(f"πŸ“€ S3 URL: {s3_url}") logger.info(f"πŸ“€ Content-Type: video/{extension}") files = { "file": ( f"gameplay.{extension}", file_content, f"video/{extension}", ) } data = s3_fields.copy() try: async with httpx.AsyncClient(timeout=300.0) as client: # 5 min timeout for video resp = await client.post(s3_url, data=data, files=files) except httpx.RequestError as exc: logger.error("Failed to upload video to S3 (transport): %s", exc) raise HTTPException(status_code=502, detail="S3 video upload failed during transport") logger.info(f"πŸ“€ S3 upload response status: {resp.status_code}") if resp.status_code >= 400: detail = resp.text logger.error( "Failed to upload video to S3: %s - %s", resp.status_code, detail ) raise HTTPException( status_code=resp.status_code, detail=f"S3 video upload failed: {detail}", ) logger.info(f"βœ… Successfully uploaded {len(file_content)} bytes to S3") async def mark_video_upload_finished( token: str, upload_id: str, clip_id: str, filename: str = "gameplay.webm" ): """Mark the video upload as finished""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/video/{upload_id}/upload-finish/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload = { "clip_id": clip_id, "upload_type": "file_upload", "upload_filename": filename, "video_upload_type": "video_hook", } await httpx_post( url, json=payload, headers=headers, context="Mark video upload finished" ) async def poll_video_upload_status( token: str, upload_id: str, max_attempts: int = 120, delay_seconds: int = 2 ) -> dict: """Poll video upload status until complete (videos take longer to process)""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/uploads/video/{upload_id}/" headers = { "Authorization": f"Bearer {token}", } for attempt in range(max_attempts): data = await httpx_get( url, headers=headers, context="Poll video upload status" ) status = data.get("status") logger.info( f"πŸ“Š Video upload status (attempt {attempt + 1}/{max_attempts}): {status}" ) logger.info(f"πŸ“Š Full video status response: {data}") if status == "complete": logger.info(f"βœ… Video processing complete! Data: {data}") return data if status == "error": logger.error(f"❌ Video processing error: {data}") raise HTTPException( status_code=500, detail=f"Video upload processing failed: {data.get('error_message', 'Unknown error')}", ) await asyncio.sleep(delay_seconds) raise HTTPException(status_code=408, detail="Video upload processing timed out") async def convert_to_mp4( video_bytes: bytes, input_extension: str ) -> tuple[bytes, str]: """ Convert video to MP4 format using FFmpeg Returns: (converted_bytes, new_filename) """ try: # Create temporary files with tempfile.NamedTemporaryFile( suffix=f".{input_extension}", delete=False ) as input_file: input_path = input_file.name input_file.write(video_bytes) output_path = input_path.rsplit(".", 1)[0] + ".mp4" logger.info(f"πŸ”„ Running FFmpeg conversion: {input_path} -> {output_path}") # Run FFmpeg conversion # -i: input file # -c:v libx264: use H.264 codec # -preset fast: encoding speed # -crf 23: quality (lower = better, 23 is default) # -c:a aac: audio codec # -b:a 128k: audio bitrate # -movflags +faststart: optimize for web streaming process = subprocess.run( [ "ffmpeg", "-i", input_path, "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", "-y", # Overwrite output file output_path, ], capture_output=True, text=True, timeout=60, # 60 second timeout ) if process.returncode != 0: logger.error(f"FFmpeg error: {process.stderr}") raise Exception(f"FFmpeg conversion failed: {process.stderr}") # Read converted file with open(output_path, "rb") as f: converted_bytes = f.read() # Cleanup os.unlink(input_path) os.unlink(output_path) logger.info(f"βœ… FFmpeg conversion complete: {len(converted_bytes)} bytes") return converted_bytes, "gameplay.mp4" except Exception as e: logger.error(f"Error converting video: {e}") # Cleanup on error try: if os.path.exists(input_path): os.unlink(input_path) if os.path.exists(output_path): os.unlink(output_path) except Exception as cleanup_error: logger.warning(f"Error during cleanup: {cleanup_error}") raise HTTPException( status_code=500, detail=f"Video conversion failed: {str(e)}" ) async def create_hook_with_video( token: str, clip_id: str, upload_id: str, title: str, clip_duration: float, video_duration: float, ) -> dict: """Create a hook with the uploaded video""" url = f"{SUNO_STUDIO_API_BASE_URL}/api/video/hooks/create" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } # Video object for the uploaded gameplay video video_object = { "upload_id": upload_id, "source_start_time": 0.0, "source_end_time": video_duration, "volume": 0, # Mute the gameplay video audio } payload = { "clip_id": clip_id, "hook_publish": { "title": title, "allow_comments": True, "show_lyrics": False, }, "video_render": { "song": { "song_id": clip_id, "clip_start_time": 0.0, "clip_end_time": clip_duration, "volume": 100, # Full volume for the song }, "videos": [video_object], }, } logger.info(f"πŸ“€ Creating hook with payload: {payload}") result = await httpx_post( url, json=payload, headers=headers, timeout=60.0, context="Create hook with video" ) logger.info(f"βœ… Hook created successfully: {result}") return result @web_app.post("/upload-and-cover") async def upload_and_cover( file: UploadFile = File(...), styles: str = Form("marimba"), is_mumble: bool = Form(False), task: str = Form("cover"), ): """ Combined endpoint: Upload audio and generate covers in one call. Args: file: Audio file to upload styles: Style tags as a single string (e.g., "marimba,jazz,rock" or "marimba") Returns: { "success": true, "upload_id": "...", "original_clip": {...}, "cover_clips": [...] } """ try: token = get_suno_token() filename = file.filename or "audio.mp3" extension = filename.rsplit(".", 1)[-1] if "." in filename else "mp3" logger.info( f"Starting upload and cover generation for: {filename}, styles: {styles}" ) # Read file content file_content = await file.read() logger.info(f"Read {len(file_content)} bytes from uploaded file") # Step 1: Upload the audio upload_data = await request_upload_slot(token, extension) upload_id = upload_data["id"] s3_url = upload_data["url"] s3_fields = upload_data["fields"] logger.info(f"Got upload slot: {upload_id}") await upload_to_s3(s3_url, s3_fields, file_content) logger.info("Uploaded to S3 successfully") await mark_upload_finished(token, upload_id, filename) logger.info("Marked upload as finished") final_upload_data = await poll_upload_status(token, upload_id) logger.info( f"Upload processing complete, status: {final_upload_data.get('status')}" ) clip_data = await initialize_clip(token, upload_id) clip_id = clip_data.get("clip_id") logger.info(f"Initialized clip: {clip_id}") original_clip = await get_clip_details(token, clip_id) logger.info("Retrieved original clip details") # Step 2: Generate cover with the style string logger.info(f"Generating cover with styles: {styles}") generation_data = await generate_cover_with_suno( token=token, cover_clip_id=clip_id, cover_clip_prompt=original_clip.get("metadata", {}).get("prompt", ""), style=styles, title=f"{original_clip.get('title', 'Audio')} ({task.capitalize()})", make_instrumental=True, is_mumble=is_mumble, task=task, ) clips = generation_data.get("clips", []) if not clips: raise HTTPException( status_code=500, detail="No clips returned from generation" ) clip_ids = [clip["id"] for clip in clips] logger.info(f"Generated {len(clip_ids)} clips: {clip_ids}") # Poll until ready cover_clips = await poll_clips_until_ready(token, clip_ids) logger.info("Cover clips ready") return JSONResponse( content={ "success": True, "upload_id": upload_id, "original_clip": original_clip, "cover_clips": cover_clips, } ) except HTTPException: raise except Exception as e: tb = traceback.format_exc() logger.exception(f"Error in upload and cover: {e}") return JSONResponse( status_code=500, content={"error": str(e), "traceback": tb} ) @web_app.get("/leaderboard") async def get_leaderboard(difficulty: Optional[str] = None): """Return practice and challenge leaderboards.""" if difficulty: diff = difficulty.strip().lower() if diff not in ALLOWED_DIFFICULTIES: raise HTTPException( status_code=400, detail=f"Unsupported difficulty '{diff}'. Expected one of {sorted(ALLOWED_DIFFICULTIES)}.", ) entries = list(leaderboard_store.get(diff, [])) return {"difficulty": diff, "entries": entries} return { "difficulties": { diff: list(entries) for diff, entries in leaderboard_store.items() } } @web_app.post("/leaderboard/record") async def record_leaderboard(entry: LeaderboardEntry): """Record a leaderboard entry (in-memory, best-effort).""" try: payload = entry.dict() payload["player"] = (payload.get("player") or "Anonymous").strip()[:40] or "Anonymous" payload["mode"] = (payload.get("mode") or "keyboard").strip().lower()[:20] payload["difficulty"] = ( (payload.get("difficulty") or DEFAULT_DIFFICULTY).strip().lower() ) payload["score"] = max(0, int(payload.get("score") or 0)) payload["duration"] = max(0.0, float(payload.get("duration") or 0)) payload["recorded_at"] = payload.get("recorded_at") or datetime.utcnow().isoformat() difficulty = payload["difficulty"] if difficulty not in ALLOWED_DIFFICULTIES: raise HTTPException( status_code=400, detail=f"Unsupported difficulty '{difficulty}'. Expected one of {sorted(ALLOWED_DIFFICULTIES)}.", ) bucket = leaderboard_store[difficulty] bucket.append(payload) sorted_entries = sorted( bucket, key=lambda item: (-item["score"], item["duration"]), ) leaderboard_store[difficulty] = deque( sorted_entries[:LEADERBOARD_MAX_ENTRIES], maxlen=LEADERBOARD_MAX_ENTRIES, ) return { "success": True, "difficulty": difficulty, "entries": list(leaderboard_store[difficulty]), } except HTTPException: raise except Exception as exc: logger.exception("Failed to record leaderboard entry: %s", exc) raise HTTPException(status_code=500, detail="Unable to record leaderboard entry") from exc @web_app.get("/leaderboard/summary") async def leaderboard_summary(): """Return a compact summary of stored leaderboard progress.""" summary = [] for diff in sorted(ALLOWED_DIFFICULTIES): entries = list(leaderboard_store.get(diff, [])) if not entries: continue top_entry = max(entries, key=lambda item: item["score"]) summary.append( { "difficulty": diff, "entry_count": len(entries), "best_score": top_entry["score"], "best_duration": top_entry["duration"], "best_player": top_entry["player"], "updated_at": top_entry["recorded_at"], } ) return {"summary": summary, "difficulties": sorted(ALLOWED_DIFFICULTIES)} @web_app.get("/practice/hints") async def practice_hints(): """Provide quick practice tips for easy onboarding.""" return {"tips": PRACTICE_TIPS} @web_app.get("/practice/spotlight") async def practice_spotlight(genre: Optional[str] = None): """Surface a rotating artist spotlight to pair with practice runs.""" candidates = ARTIST_SPOTLIGHTS if genre: normalized = genre.strip().lower() filtered = [ entry for entry in ARTIST_SPOTLIGHTS if normalized == entry["id"] or normalized in entry.get("genres", []) ] if filtered: candidates = filtered if not candidates: raise HTTPException(status_code=404, detail="No spotlights available") spotlight = random.choice(candidates) return {"spotlight": spotlight, "total": len(ARTIST_SPOTLIGHTS)} @web_app.get("/practice/insights") async def practice_insights( musician_id: Optional[str] = None, genre: Optional[str] = None ): """Return lightweight practice insights tied to RPG musician data.""" entries = ARTIST_INSIGHTS if musician_id: normalized = musician_id.strip().lower() filtered = [ entry for entry in ARTIST_INSIGHTS if entry["id"] == normalized ] if not filtered: raise HTTPException( status_code=404, detail=f"No practice insight found for musician '{musician_id}'", ) entries = filtered elif genre: normalized = genre.strip().lower() filtered = [ entry for entry in ARTIST_INSIGHTS if normalized in {g.lower() for g in entry.get("genres", [])} ] entries = filtered or [] return { "insights": entries, "count": len(entries), "available": len(ARTIST_INSIGHTS), } return web_app