backend/clone.py

"""Voice-cloning helpers: stitch speaker audio + call ElevenLabs IVC/TTS/rename/delete.

Used by the clone-preview / clone-commit endpoints to:
 1. Build a reference WAV from a speaker's turns (or a custom range) of the
    source clip's audio.
 2. Instant-clone that into ElevenLabs (returns a temporary voice_id).
 3. Generate a short TTS sample with the new voice ("Hi, I'm <name>.") so the
    user can audition before committing.
 4. Either rename + persist (commit) or delete (cancel).
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import numpy as np
import requests
import soundfile as sf

ELEVENLABS_BASE = "https://api.elevenlabs.io/v1"


def stitch_speaker_turns(
    audio_path: Path,
    turns: list[dict[str, Any]],
    max_seconds: float = 25.0,
    gap_ms: int = 120,
) -> tuple[np.ndarray | None, int, float]:
    """Concatenate the speaker's turns (longest first) until reaching max_seconds.
    Returns (audio_array, sample_rate, total_duration_s). audio_array is None if no turns.
    """
    data, sr = sf.read(str(audio_path), dtype="float32", always_2d=False)
    if data.ndim > 1:
        data = data.mean(axis=1)

    gap = np.zeros(int(gap_ms / 1000 * sr), dtype="float32")

    turns_sorted = sorted(turns, key=lambda t: t["end"] - t["start"], reverse=True)

    chunks: list[np.ndarray] = []
    total = 0.0
    for t in turns_sorted:
        seg_dur = t["end"] - t["start"]
        if seg_dur < 0.3:
            continue
        if total + seg_dur > max_seconds:
            remaining = max_seconds - total
            if remaining < 1.0:
                break
            seg_dur = remaining
        i0 = max(0, int(t["start"] * sr))
        i1 = min(len(data), i0 + int(seg_dur * sr))
        if i1 <= i0:
            continue
        chunks.append(data[i0:i1])
        total += (i1 - i0) / sr
        if total >= max_seconds:
            break

    if not chunks:
        return None, sr, 0.0

    out_parts = []
    for i, c in enumerate(chunks):
        if i > 0:
            out_parts.append(gap)
        out_parts.append(c)
    return np.concatenate(out_parts), sr, total


def stitch_range(audio_path: Path, start_s: float, end_s: float) -> tuple[np.ndarray | None, int, float]:
    """Slice a custom [start_s, end_s] from the source audio."""
    data, sr = sf.read(str(audio_path), dtype="float32", always_2d=False)
    if data.ndim > 1:
        data = data.mean(axis=1)
    i0 = max(0, int(start_s * sr))
    i1 = min(len(data), int(end_s * sr))
    if i1 - i0 < int(0.5 * sr):
        return None, sr, 0.0
    return data[i0:i1], sr, (i1 - i0) / sr


def write_wav(audio: np.ndarray, sr: int, path: Path) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    sf.write(str(path), audio, sr, subtype="PCM_16")
    return path


def isolate_audio(api_key: str, wav_in: Path, wav_out: Path) -> Path:
    """Run ElevenLabs Voice Isolator on a WAV.
    Strips background music and noise; returns cleaned WAV path.
    """
    with wav_in.open("rb") as f:
        resp = requests.post(
            f"{ELEVENLABS_BASE}/audio-isolation",
            headers={"xi-api-key": api_key},
            files={"audio": (wav_in.name, f, "audio/wav")},
            timeout=180,
        )
    if resp.status_code != 200:
        raise RuntimeError(
            f"Voice Isolator failed: HTTP {resp.status_code} — {resp.text[:300]}"
        )
    # Response is MP3 — decode through ffmpeg to WAV.
    import subprocess
    import imageio_ffmpeg
    ffmpeg = imageio_ffmpeg.get_ffmpeg_exe()
    proc = subprocess.run(
        [ffmpeg, "-i", "pipe:0", "-ac", "1", "-ar", "44100",
         "-f", "wav", "pipe:1", "-loglevel", "error", "-y"],
        input=resp.content, capture_output=True, check=True,
    )
    wav_out.parent.mkdir(parents=True, exist_ok=True)
    wav_out.write_bytes(proc.stdout)
    return wav_out


def ivc_clone(api_key: str, name: str, wav_path: Path, description: str = "") -> str:
    """Instant Voice Clone. Returns the new eleven voice_id."""
    with wav_path.open("rb") as f:
        files = {"files": (wav_path.name, f, "audio/wav")}
        data = {"name": name}
        if description:
            data["description"] = description
        resp = requests.post(
            f"{ELEVENLABS_BASE}/voices/add",
            headers={"xi-api-key": api_key},
            files=files,
            data=data,
            timeout=120,
        )
    if resp.status_code != 200:
        raise RuntimeError(f"IVC failed: HTTP {resp.status_code} — {resp.text[:300]}")
    return resp.json()["voice_id"]


def tts_say(api_key: str, voice_id: str, text: str, out_wav: Path) -> Path:
    """Generate a short TTS sample with the cloned voice. Saves as PCM_16 WAV."""
    resp = requests.post(
        f"{ELEVENLABS_BASE}/text-to-speech/{voice_id}",
        params={"output_format": "pcm_44100"},
        headers={"xi-api-key": api_key},
        json={
            "text": text,
            "model_id": "eleven_multilingual_v2",
            "voice_settings": {"stability": 0.5, "similarity_boost": 0.85},
        },
        timeout=60,
    )
    if resp.status_code != 200:
        raise RuntimeError(f"TTS preview failed: HTTP {resp.status_code} — {resp.text[:300]}")
    samples = np.frombuffer(resp.content, dtype="<i2")
    out_wav.parent.mkdir(parents=True, exist_ok=True)
    sf.write(str(out_wav), samples, 44100, subtype="PCM_16")
    return out_wav


def rename_eleven_voice(api_key: str, voice_id: str, new_name: str) -> None:
    resp = requests.post(
        f"{ELEVENLABS_BASE}/voices/{voice_id}/edit",
        headers={"xi-api-key": api_key},
        data={"name": new_name},
        timeout=30,
    )
    if resp.status_code != 200:
        raise RuntimeError(f"Rename failed: HTTP {resp.status_code} — {resp.text[:300]}")


def delete_eleven_voice(api_key: str, voice_id: str) -> None:
    requests.delete(
        f"{ELEVENLABS_BASE}/voices/{voice_id}",
        headers={"xi-api-key": api_key},
        timeout=30,
    )