kithara-studio/backend/src/kithara_client.py

import httpx
import base64
import os
import logging
from pathlib import Path

logger = logging.getLogger("kithara_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()
            
            # Check for safety/copyright filter blocks first
            # Google's filter can reject prompts and return finish_reason: "SAFETY"
            # or blockReason with a message about copyrighted material
            block_reason = data.get("blockReason", "")
            if block_reason:
                block_message = data.get("blockReasonMessage", "")
                logger.warning(f"Google Lyria API blocked: {block_reason} — {block_message}")
                return {"success": False, "error": f"Prompt was filtered. Try describing the sound using mood, instruments, and tempo instead of genre names. ({block_message or block_reason})"}
            
            # Extract base64 audio from response
            # Google 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": "Prompt was filtered. Try describing the sound using mood, instruments, and tempo instead of genre names."}
            
            # Check finish_reason for blocks
            finish_reason = candidates[0].get("finishReason", "")
            if finish_reason == "SAFETY":
                return {"success": False, "error": "Prompt was filtered by safety controls. Try rephrasing — describe the sound using mood, instruments, and tempo instead of specific genre or artist names."}
            
            content = candidates[0].get("content", {})
            parts = content.get("parts", [])
            if not parts:
                return {"success": False, "error": "No audio was generated. Try rephrasing your prompt — describe the sound using mood, instruments, and tempo."}
            
            # 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:
                # No audio in any part — likely filtered
                return {"success": False, "error": "No audio was generated. Try rephrasing your prompt — describe the sound using mood, instruments, and tempo instead of specific genre names."}
            
            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"Google 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
    }