""" backend/suno_utils.py – staging‑aware Suno helper ================================================= Handles Suno’s occasionally inconsistent *staging* API, with generous retry/back‑off, robust schema handling, and **very chatty** debug output so you can see exactly what every endpoint returns. Environment variables --------------------- * **SUNO_TOKEN** – Bearer token (required) * **SUNO_DEBUG=1** – prints every poll/response * **SUNO_INIT_TIMEOUT** – seconds to wait for clip init (default 180) * **SUNO_RENDER_TIMEOUT** – seconds to wait for wav render (default 360) * **SUNO_INIT_POLL** – poll interval for init (default 2) * **SUNO_RENDER_POLL** – poll interval for render (default 2) Public API ---------- `audio_to_suno_song(local_wav: Path, prompt: str, out_wav: Path) -> str` Uploads *local_wav*, generates a continuation from *prompt*, downloads the rendered WAV to *out_wav*, and returns the generated **clip_id**. """ from __future__ import annotations import json import os import random import time from pathlib import Path from typing import Final, Optional import requests from tqdm import tqdm # --------------------------------------------------------------------------- # configuration # --------------------------------------------------------------------------- API_BASE: Final = "https://studio-api.staging.suno.com" INIT_TIMEOUT: Final = int(os.getenv("SUNO_INIT_TIMEOUT", "180")) RENDER_TIMEOUT: Final = int(os.getenv("SUNO_RENDER_TIMEOUT", "360")) INIT_POLL: Final = int(os.getenv("SUNO_INIT_POLL", "2")) RENDER_POLL: Final = int(os.getenv("SUNO_RENDER_POLL", "2")) DEBUG: Final = bool(int(os.getenv("SUNO_DEBUG", "0"))) # --------------------------------------------------------------------------- # helpers – logging # --------------------------------------------------------------------------- def _dbg(msg: str): if DEBUG: print(f"[suno_utils] {msg}") # --------------------------------------------------------------------------- # low‑level helpers # --------------------------------------------------------------------------- def _headers() -> dict[str, str]: tok = os.getenv("SUNO_TOKEN") if not tok: raise RuntimeError("SUNO_TOKEN env var missing – see README") return {"Authorization": f"Bearer {tok}", "Content-Type": "application/json"} def _post_retry(url: str, max_attempts: int = 5, **kwargs) -> requests.Response: """POST with exponential back‑off on 5xx.""" for attempt in range(max_attempts): r = requests.post(url, **kwargs) if r.status_code < 500: return r wait = 2 * (attempt + 1) + random.random() _dbg(f"POST {url} → {r.status_code}; retry in {wait:.1f}s") time.sleep(wait) r.raise_for_status() def _get_retry(url: str, max_attempts: int = 5, **kwargs) -> requests.Response: """POST with exponential back‑off on 5xx.""" for attempt in range(max_attempts): r = requests.get(url, **kwargs) if r.status_code < 500: return r wait = 2 * (attempt + 1) + random.random() _dbg(f"POST {url} → {r.status_code}; retry in {wait:.1f}s") time.sleep(wait) r.raise_for_status() # --------------------------------------------------------------------------- # upload helpers # --------------------------------------------------------------------------- # ─────────────────────────── upload helpers ──────────────────────────────── def _upload_prod(j: dict, wav: Path) -> str: """Upload to production bucket via presigned PUT.""" upload_id, presigned = j["upload_id"], j["upload_url"] with wav.open("rb") as fd: requests.put( presigned, data=fd, headers={"Content-Type": "audio/wav"} ).raise_for_status() _dbg(f"[prod] PUT ✓ id={upload_id}") return upload_id def _upload_staging(j: dict, wav: Path) -> str: """Upload to staging bucket via multipart POST.""" upload_id, url, fields = j["id"], j["url"], j["fields"] ctype = fields.get("Content-Type", "audio/mpeg") _dbg(f"[staging] POST → {url} key={fields.get('key')} " f"({wav.stat().st_size/2**20:.2f} MiB, {ctype})") with wav.open("rb") as fd: files = {"file": (wav.name, fd, ctype)} requests.post(url, data=fields, files=files).raise_for_status() _dbg(f"[staging] POST ✓ id={upload_id}") return upload_id def upload_audio(local_wav: Path) -> str: """End‑to‑end upload → clip initialisation. Returns the new clip_id.""" meta = {"filename": local_wav.name, "content_type": "audio/wav"} up = _post_retry(f"{API_BASE}/api/uploads/audio", json=meta, headers=_headers()).json() if "upload_id" in up: # production flow upload_id = _upload_prod(up, local_wav) elif "id" in up and "url" in up: # staging flow upload_id = _upload_staging(up, local_wav) else: raise RuntimeError(f"Unknown upload schema: {json.dumps(up)[:200]}") finish_payload = { "upload_type": "audio", "upload_filename": local_wav.name, "upload_key": up.get("fields", {}).get( "key", f"raw_uploads/{upload_id}.mp3" ), } requests.post( f"{API_BASE}/api/uploads/audio/{upload_id}/upload-finish", json=finish_payload, headers=_headers() ).raise_for_status() _dbg("upload step 1.4 → upload‑finish ✓") # wait until the backend has actually processed the binary _wait_for_file(upload_id) # now we can create a clip from it return _initialise_clip(upload_id) def get_aligned_lyrics(clip_id: str, post=True) -> str: """Convert raw upload → clip and return its id.""" if not post: init = {"detail": "blah"} i =0 while init.get("detail", False) and i < 50: i += 1 time.sleep(0.2) init = _get_retry(f"{API_BASE}/api/gen/{clip_id}/aligned_lyrics/", headers=_headers()).json() init = {"aligned_words": init["data"][0]} else: init = {"detail": "blah"} i =0 while init.get("detail", False) and i < 3: i += 1 init = _post_retry(f"{API_BASE}/api/gen/{clip_id}/aligned_lyrics/v2/", json={}, headers=_headers()).json() return init # raise TimeoutError(f"Clip init timed out; last={json.dumps(last)[:200]}") def _initialise_clip(upload_id: str) -> str: """Convert raw upload → clip and return its id.""" init = _post_retry(f"{API_BASE}/api/uploads/audio/{upload_id}/initialize-clip", json={}, headers=_headers()) # fast path if init.ok and init.text.strip(): try: data = init.json() clip_id = data.get("clip_id") or ( data.get("id") if data.get("status") == "complete" else None ) if clip_id: return clip_id except json.JSONDecodeError: pass # poll path status_url = f"{API_BASE}/api/uploads/audio/{upload_id}" start = time.time() last: Optional[dict] = None while time.time() - start < INIT_TIMEOUT: res = requests.get(status_url, headers=_headers()) res.raise_for_status() data = last = res.json() _dbg(f"init poll {int(time.time()-start):3}s → {data.get('status')}") clip_id = data.get("clip_id") or ( data.get("id") if data.get("status") == "complete" else None ) if clip_id: return clip_id if data.get("status") == "error": raise RuntimeError(f"Upload processing failed: {data}") time.sleep(INIT_POLL) raise TimeoutError(f"Clip init timed out; last={json.dumps(last)[:200]}") # ───────────────── wait until S3->Suno ingestion is finished ──────────────── def _wait_for_file(upload_id: str) -> None: """Poll /uploads/audio/{id} until the binary is really ingested.""" status_url = f"{API_BASE}/api/uploads/audio/{upload_id}" start = time.time() while time.time() - start < INIT_TIMEOUT: j = requests.get(status_url, headers=_headers()).json() # success on either flag done = j.get("is_file_uploaded") is True or j.get("status") == "complete" _dbg(f"file poll {int(time.time()-start):3}s → {done:<5} " f"status={j.get('status')}") if done: return time.sleep(INIT_POLL) raise TimeoutError("Binary never ingested (status never became complete)") # --------------------------------------------------------------------------- # generation helpers # --------------------------------------------------------------------------- def _wait_for_generation(gen_id: str) -> dict: """Poll `/api/generate/requests` until the generation finishes and **log every poll**.""" url = f"{API_BASE}/api/generate/requests?ids={gen_id}" start = time.time() while time.time() - start < 300: res = requests.get(url, headers=_headers()) res.raise_for_status() data = res.json() if DEBUG and data: _dbg(f"gen poll {int(time.time()-start):3}s → {data[0].get('status', '?')}") if data and data[0].get("status") in {"complete", "streaming", "error", "failed"}: return data[0] time.sleep(5) raise TimeoutError("Generation timed out") def generate_song(prompt: str, reference_clip: str | None = None, is_cover: bool = False, audio_weight: int = 100) -> str: if is_cover and reference_clip: # For covers/remixes, use TEXT generation with special parameters payload = { "prompt": "", # Empty prompt for instrumental covers "generation_type": "TEXT", "tags": prompt, # Style/genre tags go here "mv": "chirp-v4.5", "task": "cover", "cover_clip_id": reference_clip, "metadata": { "control_sliders": { "audio_weight": audio_weight / 100.0, # Convert to 0-1 scale "style_weight": 0.5, "weirdness_constraint": 0.3 }, "is_remix": True } } _dbg(f"Generating cover with payload: {json.dumps(payload, indent=2)}") else: # Regular generation or continuation payload = {"prompt": prompt, "generation_type": "AUDIO", "mv": "chirp-v4.5"} if reference_clip: payload["continuation_clip_id"] = reference_clip _dbg(f"Generating song with payload: {json.dumps(payload, indent=2)}") res = _post_retry(f"{API_BASE}/api/generate/v2", json=payload, headers=_headers()) res.raise_for_status() gen_id = res.json()["id"] _dbg(f"Generation started with ID: {gen_id}") result = _wait_for_generation(gen_id) if not result["clips"]: raise RuntimeError("No clips generated") return result["clips"][0]["id"] # --------------------------------------------------------------------------- # render helpers # --------------------------------------------------------------------------- def _resolve_wav_url(clip_id: str) -> str: wav_api = f"{API_BASE}/api/gen/{clip_id}/wav_file" start = time.time() last_payload: Optional[str] = None poll_count = 0 _dbg(f"Starting WAV resolution for clip: {clip_id}") while time.time() - start < RENDER_TIMEOUT: poll_count += 1 res = requests.get(wav_api, headers=_headers(), allow_redirects=False) status, ctype = res.status_code, res.headers.get("Content-Type", "") preview = "" if ctype.startswith("application/json") or ctype.startswith("text"): preview = " " + res.text[:200].replace("\n", " ") _dbg(f"wav poll #{poll_count} [{int(time.time()-start):3}s] status={status} ctype={ctype}{preview}") # finished asset – S3 redirect if status == 302 and "Location" in res.headers: _dbg(f"Got S3 redirect after {poll_count} polls") return res.headers["Location"] # JSON payload with direct URL if status == 200 and ctype.startswith("application/json"): try: data = res.json() last_payload = json.dumps(data)[:300] _dbg(f"Got JSON response: {last_payload}") for k in ("url", "download_url", "wav_url", "location"): if k in data and data[k]: _dbg(f"Found URL in key '{k}': {data[k]}") return data[k] except json.JSONDecodeError: _dbg(f"Failed to parse JSON response") pass # If status is 200 but content-type is audio, the WAV is ready if status == 200 and ctype.startswith("audio"): _dbg(f"Got direct audio response after {poll_count} polls") # Return the API URL itself, it's serving the audio directly return wav_api time.sleep(RENDER_POLL) raise TimeoutError(f"WAV render timed out after {poll_count} polls; last payload={last_payload}") # --------------------------------------------------------------------------- # download helper # --------------------------------------------------------------------------- def _download_wav(src_url: str, dst: Path): """Stream *src_url* to *dst* with auth & progress bar.""" headers: dict[str, str] = {} if API_BASE in src_url: # signed endpoint – pass token headers["Authorization"] = _headers()["Authorization"] _dbg(f"Downloading from: {src_url}") with requests.get(src_url, stream=True, headers=headers) as r: r.raise_for_status() content_type = r.headers.get("Content-Type", "") _dbg(f"Download response: status={r.status_code}, content-type={content_type}") if not content_type.startswith("audio"): # Sometimes Suno returns text/html for audio files, let's be more lenient if r.status_code == 200 and len(r.content) > 1000: _dbg(f"Got non-audio content-type but proceeding anyway (size={len(r.content)})") else: raise RuntimeError( f"Expected audio but got {content_type} from {src_url}" ) total = int(r.headers.get("Content-Length", 0)) _dbg(f"Starting download of {total/1024/1024:.2f} MB") with tqdm.wrapattr(dst.open("wb"), "write", total=total, desc="Downloading") as fd: for chunk in r.iter_content(chunk_size=8192): fd.write(chunk) _dbg(f"Download complete: {dst}") # --------------------------------------------------------------------------- # public convenience helper # --------------------------------------------------------------------------- def audio_to_suno_song(local_wav: Path, prompt: str, out_wav: Path) -> str: """ Upload *local_wav*, generate a continuation, download the rendered WAV to *out_wav*, and return the generated clip-id. """ # Use the step-by-step approach which works better from suno_step_by_step_utils import upload_audio_step_by_step, StepLogger # First upload the audio properly clip_id, _ = upload_audio_step_by_step(local_wav) _dbg(f"reference clip → {clip_id}") # Generate continuation (not cover) gen_payload = { "prompt": prompt, "generation_type": "AUDIO", "mv": "chirp-v4.5", "continuation_clip_id": clip_id } _dbg(f"Generating continuation with payload: {json.dumps(gen_payload, indent=2)}") res = _post_retry(f"{API_BASE}/api/generate/v2", json=gen_payload, headers=_headers()) res.raise_for_status() gen_id = res.json()["id"] _dbg(f"Generation started with ID: {gen_id}") result = _wait_for_generation(gen_id) if not result["clips"]: raise RuntimeError("No clips generated") gen_clip = result["clips"][0]["id"] _dbg(f"generated clip → {gen_clip}") wav_url = _resolve_wav_url(gen_clip) _dbg(f"wav url → {wav_url}") _download_wav(wav_url, out_wav) return clip_id, gen_clip def audio_to_suno_cover(local_wav: Path, prompt: str, out_wav: Path, audio_weight: int = 100) -> str: """ Upload *local_wav*, generate a cover with the given prompt and audio weight, download the rendered WAV to *out_wav*, and return the generated clip-id. Default audio weight is 100% to maintain original audio characteristics. """ # Use the step-by-step approach which works better from suno_step_by_step_utils import upload_and_remix_step_by_step # Use the working remix approach result, _ = upload_and_remix_step_by_step( local_wav, prompt, audio_weight=audio_weight / 100.0 # Convert to 0-1 scale ) # Get the first generated clip if not result["generated_clips"]: raise RuntimeError("No clips generated") gen_clip = result["generated_clips"][0]["id"] _dbg(f"generated cover clip → {gen_clip}") # Download the WAV wav_url = _resolve_wav_url(gen_clip) _dbg(f"cover wav url → {wav_url}") _download_wav(wav_url, out_wav) return gen_clip if __name__ == "__main__": import os env_vars = dict(REPLICATE_API_TOKEN="r8_HGsdsEyz339nplinKh9Ex5Se2mxM8Sc1lbdu4", WHISPER_MODEL="villesau/whisper-timestamped:c5b122b7e513b1b5a6ef849891c538869b77cc932cbd0f8203e11d3b357553b8", SUNO_TOKEN="ae6cb0480e2e493cbab61a7324030ec5", SUNO_DEBUG="1") for k, v in env_vars.items(): os.environ[k] = v full_v2("https://www.tiktok.com/@ethan.stee1e/video/7529957148734786838?kref=vGWRdzLJnjYf&kuid=31b6d78a-c086-437f-9ab0-df56eaf7fd8b")