#!/usr/bin/env python3 """Minimal Sonilo text-to-music example. Writes output.m4a only when the API returns real audio bytes or a real audio URL. Does not create placeholder files after API errors or insufficient balance. Status handling: 401 is auth_invalid, 403 is forbidden, 402 payment_required is insufficient_balance, 404 means check the https://api.sonilo.com/v1 base URL, and 429 retries with Retry-After/backoff. """ import base64 import json import os import random import sys import time import urllib.error import urllib.request import uuid from pathlib import Path API_BASE = os.getenv("SONILO_API_BASE", "https://api.sonilo.com/v1").rstrip("/") API_KEY = os.getenv("SONILO_API_KEY") USER_AGENT = os.getenv("SONILO_USER_AGENT", "SoniloExample/1.0 (+https://sonilo.com/docs)") OUTPUT_FILE = Path(os.getenv("OUTPUT_FILE", "output.m4a")) def require_api_key() -> str: if not API_KEY: raise SystemExit("SONILO_API_KEY is required") return API_KEY def retry_delay(attempt: int, retry_after: str | None) -> float: if retry_after: try: return max(0.0, float(retry_after)) except ValueError: pass return min(30.0, (2 ** attempt) + random.random()) def http_error_message(exc: urllib.error.HTTPError) -> str: body = exc.read().decode("utf-8", errors="replace") try: parsed = json.loads(body) return parsed.get("message") or parsed.get("error") or body except json.JSONDecodeError: return body or exc.reason def build_multipart(fields: dict[str, str]) -> tuple[bytes, str]: boundary = f"----sonilo-example-{uuid.uuid4().hex}" body = bytearray() for name, value in fields.items(): body.extend(f"--{boundary}\r\n".encode()) body.extend(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()) body.extend(str(value).encode()) body.extend(b"\r\n") body.extend(f"--{boundary}--\r\n".encode()) return bytes(body), f"multipart/form-data; boundary={boundary}" def open_text_to_music_stream(): fields = { "prompt": os.getenv( "SONILO_PROMPT", "30 seconds of upbeat instrumental background music for a clean product launch video; no vocals", ), "duration": os.getenv("SONILO_DURATION", "30"), "mode": "stream", "output_format": "m4a", } body, content_type = build_multipart(fields) for attempt in range(4): request = urllib.request.Request( f"{API_BASE}/text-to-music", data=body, method="POST", headers={ "Authorization": f"Bearer {require_api_key()}", "Content-Type": content_type, "Accept": "application/x-ndjson", "User-Agent": USER_AGENT, }, ) try: return urllib.request.urlopen(request, timeout=180) except urllib.error.HTTPError as exc: message = http_error_message(exc) if exc.code == 401: raise SystemExit(f"auth_invalid: check SONILO_API_KEY and Authorization header: {message}") if exc.code == 403: raise SystemExit(f"forbidden: API key is valid but this account cannot access the requested service or workspace: {message}") if exc.code == 402: raise SystemExit(f"insufficient_balance: {message}") if exc.code == 429 and attempt < 3: time.sleep(retry_delay(attempt, exc.headers.get("Retry-After"))) continue if exc.code == 404: raise SystemExit(f"not_found: check API_BASE uses https://api.sonilo.com/v1, not a docs URL: {message}") raise SystemExit(f"Sonilo text-to-music failed with HTTP {exc.code}: {message}") raise SystemExit("Sonilo text-to-music failed after retries") def decode_audio_chunk(value: object) -> bytes: if isinstance(value, dict): for key in ("chunk", "data", "audio", "audio_data", "base64", "content"): decoded = decode_audio_chunk(value.get(key)) if decoded: return decoded if not isinstance(value, str): return b"" payload = value.strip() if payload.startswith("data:"): payload = payload.split(",", 1)[-1] try: return base64.b64decode(payload + "=" * (-len(payload) % 4)) except Exception: return b"" def find_audio_url(value: object) -> str | None: if isinstance(value, str) and value.startswith(("http://", "https://")): return value if isinstance(value, dict): for key in ("audio_url", "output_url", "url", "download_url", "audio", "music", "result"): found = find_audio_url(value.get(key)) if found: return found for nested in value.values(): found = find_audio_url(nested) if found: return found if isinstance(value, list): for nested in value: found = find_audio_url(nested) if found: return found return None def download_audio(url: str) -> bytes: request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(request, timeout=180) as response: data = response.read() if not data: raise SystemExit("Sonilo returned an empty audio download") return data def main() -> None: chunks: list[bytes] = [] final_url: str | None = None with open_text_to_music_stream() as response: for raw_line in response: line = raw_line.decode("utf-8", errors="replace").strip() if not line: continue event = json.loads(line) event_type = str(event.get("type") or event.get("event") or event.get("status") or "").lower() if event_type in {"error", "failed"}: raise SystemExit(f"Sonilo generation failed: {event}") chunk = decode_audio_chunk(event) if chunk: chunks.append(chunk) possible_url = find_audio_url(event) if possible_url: final_url = possible_url if event_type in {"complete", "completed", "succeeded", "success"}: break audio = b"".join(chunks) if chunks else (download_audio(final_url) if final_url else b"") if not audio: raise SystemExit("Sonilo stream completed without audio bytes or an audio URL") OUTPUT_FILE.write_bytes(audio) size = OUTPUT_FILE.stat().st_size if size <= 0: raise SystemExit("Output file is empty") print(f"saved {OUTPUT_FILE} ({size} bytes)") if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit("interrupted")