rin.kagamine
23672387e5
The Lyria API returns a candidates[].content.parts[] array where: - parts[0] = text metadata (e.g. '<instrumental>') - parts[1] = inlineData with the actual audio payload Previous code only checked parts[0] for inlineData, which always failed with 'No audio data in response'. Fixed by iterating all parts to find the inlineData entry. Successfully generated first real audio clip (lo-fi hip hop, 30s, 728KB MP3). Also includes: .env.example, requirements updates, docker-compose fixes, frontend vite config, logger setup.
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
import httpx
|
|
import base64
|
|
import os
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger("lyria_client")
|
|
|
|
LYRIA_API_KEY = os.getenv("LYRIA_API_KEY", "")
|
|
LYRIA_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
|
|
|
|
CLIPS_DIR = Path(__file__).parent.parent / "data" / "clips"
|
|
CLIPS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
async def generate_clip(prompt: str, model: str = "lyria-3-clip-preview", duration: int = 30) -> dict:
|
|
"""Generate a music clip using Google Lyria API.
|
|
|
|
Returns dict with:
|
|
- success: bool
|
|
- file_path: str (if success)
|
|
- file_size: int (if success)
|
|
- error: str (if failed)
|
|
"""
|
|
if not LYRIA_API_KEY:
|
|
return {"success": False, "error": "LYRIA_API_KEY not configured"}
|
|
|
|
url = f"{LYRIA_BASE_URL}/models/{model}:generateContent?key={LYRIA_API_KEY}"
|
|
|
|
payload = {
|
|
"contents": [{
|
|
"role": "user",
|
|
"parts": [{
|
|
"text": f"{prompt} ({duration} seconds)"
|
|
}]
|
|
}],
|
|
"generationConfig": {
|
|
"responseModalities": ["AUDIO"]
|
|
}
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=120.0) as client:
|
|
try:
|
|
response = await client.post(url, json=payload)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# Extract base64 audio from response
|
|
# Lyria returns parts array: first part may be text (e.g. "<instrumental>"),
|
|
# audio is in inlineData of a subsequent part.
|
|
candidates = data.get("candidates", [])
|
|
if not candidates:
|
|
return {"success": False, "error": "No candidates in response"}
|
|
|
|
content = candidates[0].get("content", {})
|
|
parts = content.get("parts", [])
|
|
if not parts:
|
|
return {"success": False, "error": "No parts in response"}
|
|
|
|
# Find the part with inlineData (audio) — not necessarily parts[0]
|
|
audio_part = None
|
|
for part in parts:
|
|
if "inlineData" in part:
|
|
audio_part = part
|
|
break
|
|
|
|
if audio_part is None:
|
|
return {"success": False, "error": "No audio data in response (no inlineData part found)"}
|
|
|
|
inline_data = audio_part.get("inlineData", {})
|
|
mime_type = inline_data.get("mimeType", "")
|
|
audio_b64 = inline_data.get("data", "")
|
|
|
|
if not audio_b64:
|
|
return {"success": False, "error": "No audio data in response"}
|
|
|
|
# Decode and save
|
|
audio_bytes = base64.b64decode(audio_b64)
|
|
|
|
return {
|
|
"success": True,
|
|
"audio_bytes": audio_bytes,
|
|
"mime_type": mime_type
|
|
}
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
error_detail = ""
|
|
try:
|
|
error_detail = e.response.json().get("error", {}).get("message", str(e))
|
|
except:
|
|
error_detail = str(e)
|
|
return {"success": False, "error": f"Lyria API error: {error_detail}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Client error: {str(e)}"}
|
|
|
|
async def save_clip_file(clip_id: str, audio_bytes: bytes) -> dict:
|
|
"""Save audio bytes to disk. Returns file info."""
|
|
filename = f"{clip_id}.mp3"
|
|
file_path = CLIPS_DIR / filename
|
|
|
|
file_path.write_bytes(audio_bytes)
|
|
file_size = file_path.stat().st_size
|
|
|
|
return {
|
|
"filename": filename,
|
|
"file_path": str(file_path),
|
|
"file_size": file_size
|
|
}
|