app.py

# app.py
import os
import io
import base64
import tempfile
from typing import List, Dict, Tuple, Iterable, Union
from elevenlabs.core.api_error import ApiError


import gradio as gr
from elevenlabs import ElevenLabs

API_KEY_ENV = "ELEVENLABS_API_KEY"

# Voices you want to hide (by NAME, case-Sensitive)
HIDDEN_VOICES = {
    "Ruby Conv",
    "Scouse Woman",
    "Scouse Woman 2",
    "Scouse Woman 3",
    "Scouse Woman 4",
    "Scouse Woman 5",
    "Lara Old",
    "Valilu Old",
    "BLANK SLOT",
    "BLANK SLOT",
    "BLANK SLOT"
}

_client = None  # ElevenLabs client cache


# ---------- helpers ----------
def get_client() -> ElevenLabs:
    global _client
    if _client is None:
        api_key = os.getenv(API_KEY_ENV)
        if not api_key:
            raise RuntimeError(
                f"Missing {API_KEY_ENV}. In Spaces, add it under Settings → Variables as a secret."
            )
        _client = ElevenLabs(api_key=api_key)
    return _client


def _norm(s: str) -> str:
    return s.strip().lower()


def _collect_bytes(chunks_or_bytes: Union[bytes, bytearray, Iterable[bytes]]) -> bytes:
    """ElevenLabs may return bytes or a generator of bytes; normalize to bytes."""
    if isinstance(chunks_or_bytes, (bytes, bytearray)):
        return bytes(chunks_or_bytes)
    buf = io.BytesIO()
    for chunk in chunks_or_bytes:
        if not chunk:
            continue
        if isinstance(chunk, (bytes, bytearray, memoryview)):
            buf.write(bytes(chunk))
    return buf.getvalue()


def _write_temp_mp3(data: bytes) -> str:
    fd, path = tempfile.mkstemp(suffix=".mp3")
    with os.fdopen(fd, "wb") as f:
        f.write(data)
    return path


def _list_all_voices():
    """
    Retrieve all voices robustly:
    - request up to the API max page_size (100)
    - follow next_cursor / has_more if present
    - dedupe by voice_id
    """
    client = get_client()
    seen_ids = set()
    items = []

    cursor = None
    for _ in range(50):  # safety cap
        kwargs = {"include_total_count": False, "page_size": 100}  # <= max allowed
        if cursor:
            kwargs["cursor"] = cursor

        try:
            res = client.voices.search(**kwargs)
        except TypeError:
            # SDK doesn't support these kwargs → single fetch fallback
            res = client.voices.search(include_total_count=False)

        voices = getattr(res, "voices", res) or []

        new_count = 0
        for v in voices:
            vid = getattr(v, "voice_id", None)
            if not vid or vid in seen_ids:
                continue
            seen_ids.add(vid)
            items.append(v)
            new_count += 1

        print(f"[voices] page got={len(voices)} new={new_count} total_unique={len(items)}")

        # Pagination signals (if API/SDK exposes them)
        cursor = getattr(res, "next_cursor", None)
        has_more = bool(getattr(res, "has_more", False))

        # Stop when we have no new items, or there's no indication of more pages
        if new_count == 0:
            break
        if not cursor and not has_more:
            break

    print(f"[voices] done: total_unique={len(items)}")
    return items


def fetch_voices() -> List[Tuple[str, str]]:
    """
    Return (label, value) = (name, voice_id) tuples for the dropdown.
    Only show voices that appear on "My Voices":
      - your generated/cloned voices (owner or category)
      - community/library voices you've saved (favorited)
    Excludes plain premades you haven't saved.
    """
    try:
        voices = _list_all_voices()
        hidden_norm = {_norm(n) for n in HIDDEN_VOICES}
        entries: List[Tuple[str, str]] = []

        for v in voices:
            name = getattr(v, "name", "Voice")
            vid = getattr(v, "voice_id", None)
            if not vid:
                continue

            # Skip hidden names
            if _norm(name) in hidden_norm:
                continue

            category = getattr(v, "category", None)                 # 'premade' | 'generated' | 'cloned' | None
            is_owner = bool(getattr(v, "is_owner", False))          # True if you own the voice
            favorited = getattr(v, "favorited_at_unix", None) is not None  # Saved/added to your library

            # Keep if it's yours OR saved OR clearly user-type (generated/cloned)
            keep = (
                is_owner
                or favorited
                or (category in {"generated", "cloned"})
            )

            # But drop plain premade voices you haven't saved
            if category == "premade" and not favorited:
                keep = False

            if not keep:
                continue

            entries.append((name, vid))

        # Sort nicely
        entries.sort(key=lambda x: _norm(x[0]))

        # Fallback so UI still works if nothing matches
        if not entries:
            return [("Rachel", "21m00Tcm4TlvDq8ikWAM")]

        return entries

    except Exception:
        return [("Rachel", "21m00Tcm4TlvDq8ikWAM")]


def refresh_voices():
    # IMPORTANT: return the same shape as initial choices (tuples)
    return gr.update(choices=fetch_voices(), value=None)


# ---------- TTS ----------
def synthesize(text: str, voice_id: str, voice_id_override: str) -> tuple[str, str]:
    """
    Returns (html_player, mp3_filepath).
    - html_player: <audio> tag with base64 data URL → always plays inline
    - mp3_filepath: temp file path so users can download the mp3
    """
    if not text or not text.strip():
        raise gr.Error("Please enter some text.")

    client = get_client()

    # Prefer manual override if provided; otherwise take dropdown value (voice_id)
    use_voice_id = voice_id_override.strip() if voice_id_override.strip() else voice_id

    # Locked defaults
    model_id = "eleven_v3"
    output_format = "mp3_44100_192"

    try:
        audio_stream_or_bytes = client.text_to_speech.convert(
            text=text,
            voice_id=use_voice_id,
            model_id=model_id,
            output_format=output_format,
        )
    except Exception as e:
        raise gr.Error(f"TTS request failed: {e}")

    audio_bytes = _collect_bytes(audio_stream_or_bytes)

    # Inline preview via base64 data URL (robust when embedded in wrapper Spaces)
    b64 = base64.b64encode(audio_bytes).decode("ascii")
    data_url = f"data:audio/mpeg;base64,{b64}"
    html_player = f"""
    <audio controls style="width: 100%;">
        <source src="{data_url}" type="audio/mpeg">
        Your browser does not support the audio element.
    </audio>
    """

    # Downloadable MP3
    mp3_path = _write_temp_mp3(audio_bytes)

    return html_player, mp3_path


# ---------- UI ----------
with gr.Blocks(
    title="Voices",
    css="""
    /* Taller, scrollable dropdown popup */
    #voice_dd [role="listbox"] { max-height: 380px !important; overflow-y: auto !important; }
    """
) as demo:
    gr.Markdown(
        "## Voices TTS\n"
        "Type anything, pick one of **your voices**, and generate downloadable audio."
    )

    text = gr.Textbox(
        label="Text",
        placeholder="Type your script here… (tags like [laughs], [giggles] work)",
        lines=6,
    )

    with gr.Row():
        voice_dropdown = gr.Dropdown(
            label="Voice",
            choices=fetch_voices(),        # list of (name, voice_id)
            interactive=True,
            elem_id="voice_dd",
        )
        refresh_btn = gr.Button("Refresh voices")

    voice_id_override = gr.Textbox(
        label="Override voice ID (optional)",
        placeholder="Paste a voice_id to bypass the dropdown (useful if voices_read permission is missing).",
    )

    # data-only endpoint for the public wrapper to fetch voices
    _voices_json = gr.JSON(visible=False)
    gr.Button(visible=False).click(
        lambda: fetch_voices(), inputs=None, outputs=_voices_json, api_name="list_voices"
    )

    gen = gr.Button("Generate audio")

    # Inline player (HTML) + file download
    preview = gr.HTML(label="Preview (inline player)")
    dl = gr.File(label="Download MP3")

    gen.click(synthesize, inputs=[text, voice_dropdown, voice_id_override], outputs=[preview, dl])
    refresh_btn.click(refresh_voices, inputs=None, outputs=voice_dropdown)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)))