tts-voice-creator-clone-and-design-2/routes/tts.py

"""TTS backends, preview, style-variation, streaming, voice-design, OpenAI-compat proxy."""
from __future__ import annotations

import asyncio
import io
import logging
import os
import re
import time
import uuid
from pathlib import Path
from urllib.parse import quote

import requests
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import Response, StreamingResponse

from core.config import _load_settings, _clean_preview_backend, _preview_backend_base_url
from core.constants import (
    _VOICES_DIR_DEFAULT, _TTS_CONTAINER, _TTS_CONTAINERS_RAW,
    _routing_log_add,
)
from core.routing import (
    _load_tts_routes, _resolve_tts_route, _route_backend,
    _canonical_app_name, _request_app_name, _detect_text_language,
    _routing_log_request,
)
from core.presets import _load_design_presets, _virtual_voice_id, _resolve_virtual_voice
from core.voice import (
    _find_voice_audio, _active_voices_dir,
    _voice_audio_files, _load_meta, _save_meta,
    _AUDIO_EXTS,
    _voice_health,
)
from core.validation import _validate_http_url
from core.registry import _registry_put, TEMP_DIR
from core.tts_helpers import (
    _tts_request_audio, _open_tts_stream_response, _read_tts_stream_response,
    _iter_tts_stream_response, _preview_request_audio, _prepare_proxy_audio,
    _requested_response_format, _audio_ext_media, _apply_route_sounds,
    _route_sound_path, _voice_design_request_audio, _voice_design_voice_request_audio,
    _nvidia_clone_request_audio, _parse_voice_design_dialogue,
    _infer_voice_design_gender, _voice_design_dialogue_request_audio,
    _fishspeech_request_audio,
)
from core.audio import _duration, _export_normalized_wav
from core.docker_client import _docker_post

logger = logging.getLogger("uvicorn.error")

router = APIRouter()

# ── TTS backend helpers ───────────────────────────────────────────────────────

def _backend_port_label(url: str) -> str:
    from urllib.parse import urlsplit
    try:
        parts = urlsplit(url)
        if parts.port:
            return str(parts.port)
    except Exception:
        pass
    return ""


def _backend_display_name(backend: str, url: str) -> str:
    names = {
        "voice_clone":    "Voice Clone/Base (WAV File)",
        "voice_design":   "Voice Design",
        "customvoice":    "CustomVoice",
        "streaming":      "Streaming (WAV File)",
        "nvidia_magpie":  "NVIDIA Magpie TTS",
        "nvidia_zeroshot":"NVIDIA Magpie Zeroshot Clone",
        "nvidia_flow":    "NVIDIA Magpie Flow Clone",
        "kokoro":         "Kokoro FastAPI (82M)",
        "vibevoice":      "VibeVoice TTS",
        "xtts":           "XTTS v2",
        "fishspeech":     "Fish-Speech (Clone + Emotion)",
    }
    port = _backend_port_label(url)
    return f"{port} {names.get(backend, backend)}" if port else names.get(backend, backend)


def _backend_capabilities(backend: str) -> dict:
    caps = {
        "voice_clone": {
            "purpose": "Clone a speaker from a short WAV/reference clip.",
            "identity": "Strongest match to saved WAV voices.",
            "style": "Weak per-request style; instruct may be ignored.",
            "best_for": "Known voices, multilingual cloning, benchmarks, and reliable speaker identity.",
            "uses_wav": True, "style_aware": False, "true_streaming": False,
            "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "Premium clone", "ram": "~6 GB",
        },
        "voice_design": {
            "purpose": "Create or reuse prompt-designed voices from natural-language descriptions.",
            "identity": "Prompt persona, not the selected WAV speaker unless you first export/clone it.",
            "style": "Strong style and emotion control through instruct text.",
            "best_for": "New characters, personas, dialogue, and designing reference WAVs to clone later.",
            "uses_wav": False, "style_aware": True, "true_streaming": False,
            "speed": "~0.4× GPU", "latency": "1–3 s", "quality": "Premium", "ram": "~5 GB",
        },
        "customvoice": {
            "purpose": "Generate speech with the CustomVoice model voices.",
            "identity": "Uses CustomVoice speakers, not arbitrary active WAV voices unless trained/configured there.",
            "style": "Good per-request style and emotion control.",
            "best_for": "Controlled style with configured CustomVoice speakers.",
            "uses_wav": False, "style_aware": True, "true_streaming": False,
            "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "Premium", "ram": "~5 GB",
        },
        "streaming": {
            "purpose": "Low-latency playback from saved WAV/reference voices.",
            "identity": "Same WAV voice identity path as Base.",
            "style": "Weak per-request style in the current streaming server.",
            "best_for": "Long text, assistants, Open WebUI/SillyTavern playback that can start before completion.",
            "uses_wav": True, "style_aware": False, "true_streaming": True,
            "speed": "~0.1× GPU", "latency": "0.5–1 s", "quality": "Premium", "ram": "~5 GB",
        },
        "nvidia_magpie": {
            "purpose": "Generate speech with NVIDIA Magpie fixed speaker voices.",
            "identity": "Uses Magpie speaker aliases such as sofia, aria, jason, leo, and john; it is not a WAV voice-cloning model.",
            "style": "Language and speaker are controlled by the backend voice config; per-request style text is usually ignored.",
            "best_for": "Fast local NVIDIA TTS voices and OpenAI-compatible assistant playback.",
            "uses_wav": False, "style_aware": False, "true_streaming": False,
            "speed": "~0.05× GPU", "latency": "0.3–0.8 s", "quality": "High", "ram": "~10 GB",
        },
        "nvidia_zeroshot": {
            "purpose": "Clone a saved library voice through NVIDIA Magpie TTS Zeroshot NIM.",
            "identity": "Sends the selected WAV as audio_prompt; no prompt transcript is required.",
            "style": "Best with a clear 3-10 second prompt. Optional quality params can be configured in Settings.",
            "best_for": "Fast NVIDIA reference-audio cloning, streaming-class use cases, live agents, and games.",
            "uses_wav": True, "style_aware": False, "true_streaming": False,
            "speed": "~0.1× GPU", "latency": "0.5–1 s", "quality": "High clone", "ram": "~10 GB",
        },
        "nvidia_flow": {
            "purpose": "Clone a saved library voice through NVIDIA Magpie TTS Flow NIM.",
            "identity": "Sends the selected WAV plus its exact saved reference transcript.",
            "style": "Offline high-fidelity clone path; prompt transcript must match the reference audio.",
            "best_for": "Studio-style dubbing, narration, and podcast-quality offline generation.",
            "uses_wav": True, "style_aware": False, "true_streaming": False,
            "speed": "~0.2× GPU", "latency": "1–2 s", "quality": "Studio", "ram": "~10 GB",
        },
        "kokoro": {
            "purpose": "High-quality English TTS with Kokoro 82M model. OpenAI-compatible endpoint.",
            "identity": "Uses Kokoro built-in voices (af_bella, bf_emma, am_adam, …); no WAV cloning.",
            "style": "Voice selection via voice ID. Style instruction is not supported.",
            "best_for": "Fast, high-quality CPU TTS. Low RAM footprint. Easy local Docker setup.",
            "uses_wav": False, "style_aware": False, "true_streaming": False,
            "speed": "~0.1× CPU", "latency": "0.2–0.5 s", "quality": "High (82M)", "ram": "300 MB CPU",
        },
        "vibevoice": {
            "purpose": "VibeVoice TTS service. Simple text-in, audio-out REST endpoint.",
            "identity": "Single built-in voice; no WAV cloning or voice selection.",
            "style": "Text only — no voice ID or style parameters.",
            "best_for": "Lightweight local TTS on port 8027. Minimal setup, fast response.",
            "uses_wav": False, "style_aware": False, "true_streaming": False,
            "speed": "fast", "latency": "0.2–1 s", "quality": "High", "ram": "~4 GB",
        },
        "xtts": {
            "purpose": "XTTS v2 via xtts-api-server. OpenAI-compatible endpoint with speaker selection.",
            "identity": "Uses speakers registered in the XTTS server; not WAV voice cloning.",
            "style": "Speaker selected by voice ID. Style instruction not supported.",
            "best_for": "Local multi-speaker TTS with XTTS v2 model. Coqui/daswer123 docker setup.",
            "uses_wav": False, "style_aware": False, "true_streaming": False,
            "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "High", "ram": "~5 GB",
        },
        "fishspeech": {
            "purpose": "Clone a saved WAV voice AND control per-line tone via inline emotion markers.",
            "identity": "Clones the selected voice's reference WAV — consistent speaker identity across lines.",
            "style": "Strong: emotion/tone markers like (angry), (whispering), (excited) are honoured per request.",
            "best_for": "Consistent character voices that still react to tone changes — ideal for the Rehearser.",
            "uses_wav": True, "style_aware": True, "true_streaming": False,
            "speed": "~0.4× GPU", "latency": "1–4 s", "quality": "Premium", "ram": "~4 GB",
        },
    }
    return caps.get(_clean_preview_backend(backend), {})


def _backend_health(url: str) -> bool:
    base = url.rstrip("/")
    for ep in ("/health", "/v1/health", "/v1/audio/list_voices"):
        try:
            r = requests.get(f"{base}{ep}", timeout=2)
            if r.status_code == 200:
                return True
        except Exception:
            continue
    return False


def _backend_available(backend: str, voices: list, health: bool) -> bool:
    if _clean_preview_backend(backend) in {"nvidia_zeroshot", "nvidia_flow", "kokoro", "vibevoice", "fishspeech"}:
        return health
    return bool(voices) or health


_TTS_VOICE_ENDPOINTS = ("/v1/audio/voices", "/v1/audio/list_voices", "/v1/models", "/speakers", "/voices")
_KOKORO_BUILTIN_VOICES = [
    "af", "af_bella", "af_nicole", "af_sarah", "af_sky",
    "bf_emma", "bf_isabella", "am_adam", "am_michael",
    "bm_george", "bm_lewis",
]


def _voice_ids_from_payload(payload) -> list:
    if isinstance(payload, list):
        return payload
    if not isinstance(payload, dict):
        return []
    for key in ("data", "voices", "speakers"):
        value = payload.get(key)
        if isinstance(value, list):
            if key == "data":
                return [m.get("id", m) if isinstance(m, dict) else m for m in value]
            return value
    grouped = []
    for value in payload.values():
        if isinstance(value, list):
            grouped.extend(value)
        elif isinstance(value, dict):
            nested = _voice_ids_from_payload(value)
            if nested:
                grouped.extend(nested)
    return grouped


def _active_library_voice_options(settings: dict) -> list[dict]:
    active_dir = _active_voices_dir(settings)
    voices = []
    seen = set()
    if not active_dir.exists():
        return voices
    from core.voice import _read_reference_text
    for audio in sorted(_voice_audio_files(active_dir), key=lambda p: p.stem.lower()):
        if audio.stem in seen:
            continue
        seen.add(audio.stem)
        meta = _load_meta(audio)
        if meta.get("enabled", True) is False:
            continue
        has_ref, transcript = _read_reference_text(audio)
        try:
            dur = round(_duration(audio), 2)
        except Exception:
            dur = None
        voices.append({
            "id": audio.stem,
            "name": audio.stem,
            "duration": dur,
            "has_ref": has_ref,
            "has_transcript": bool(transcript),
        })
    return voices


def _fetch_backend_voices(settings: dict, backend: str) -> list:
    backend = _clean_preview_backend(backend)
    if backend in {"nvidia_zeroshot", "nvidia_flow"}:
        return _active_library_voice_options(settings)
    tts_url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/")
    key = (settings.get("voice_design_api_key") if backend == "voice_design" else settings.get("tts_api_key")) or ""
    tts_hdrs = {"Authorization": f"Bearer {key.strip()}"} if key.strip() else {}
    for ep in _TTS_VOICE_ENDPOINTS:
        try:
            r = requests.get(f"{tts_url}{ep}", headers=tts_hdrs, timeout=5)
            if r.status_code == 200:
                voices = _voice_ids_from_payload(r.json())
                if voices:
                    return voices
        except Exception:
            continue
    if backend == "kokoro":
        return _KOKORO_BUILTIN_VOICES
    return []


# ── TTS restart helpers ───────────────────────────────────────────────────────

def _clear_tts_restart_flags(settings: dict | None = None) -> int:
    cleared = 0
    settings = settings or _load_settings()
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
    if scan_dir.exists():
        seen: set[str] = set()
        for audio in _voice_audio_files(scan_dir):
            if audio.stem in seen:
                continue
            seen.add(audio.stem)
            meta = _load_meta(audio)
            if meta.pop("needs_tts_restart", None) is not None:
                _save_meta(audio, meta)
                cleared += 1
    return cleared


def _tts_container_names() -> list[str]:
    multi = os.environ.get("TTS_CONTAINER_NAMES", _TTS_CONTAINERS_RAW).strip()
    if multi:
        return [c.strip() for c in multi.split(",") if c.strip()]
    single = os.environ.get("TTS_CONTAINER_NAME", _TTS_CONTAINER).strip()
    return [single] if single else []


# ── Stream session store ──────────────────────────────────────────────────────

_TTS_STREAM_SESSION_TTL = 15 * 60
_tts_stream_sessions: dict[str, dict] = {}


def _purge_tts_stream_sessions() -> None:
    now = time.time()
    expired = [sid for sid, item in _tts_stream_sessions.items() if now - item.get("created", 0) > _TTS_STREAM_SESSION_TTL]
    for sid in expired:
        _tts_stream_sessions.pop(sid, None)


# ── Routes ────────────────────────────────────────────────────────────────────

@router.get("/api/tts-voices")
async def tts_voices(backend: str = "voice_clone"):
    return _fetch_backend_voices(_load_settings(), backend)


@router.get("/api/tts-backends")
async def tts_backends():
    settings = _load_settings()
    items = []
    for backend in ("voice_clone", "voice_design", "customvoice", "streaming", "fishspeech", "kokoro", "vibevoice", "xtts", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow"):
        url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/")
        voices = _fetch_backend_voices(settings, backend)
        health = _backend_health(url)
        available = _backend_available(backend, voices, health)
        items.append({
            "id": backend,
            "label": _backend_display_name(backend, url),
            "url": url,
            "port": _backend_port_label(url),
            "available": available,
            "voice_count": len(voices) if isinstance(voices, list) else 0,
            **_backend_capabilities(backend),
        })
    return {"backends": items}


@router.post("/api/tts/restart")
async def restart_tts_container():
    containers = _tts_container_names()
    if not containers:
        raise HTTPException(400, "No TTS container names configured (set TTS_CONTAINER_NAMES in docker-compose.yml)")

    results = []
    errors = []
    for container in containers:
        path = f"/containers/{quote(container, safe='')}/restart?t=10"
        try:
            code, raw = _docker_post(path)
            if code not in (204, 304):
                detail = raw.split("\r\n\r\n", 1)[-1].strip() or f"HTTP {code}"
                errors.append(f"{container}: {detail}")
            else:
                results.append(container)
        except PermissionError:
            raise HTTPException(502, "No permission to access /var/run/docker.sock — is the socket mounted in docker-compose.yml?")
        except Exception as e:
            errors.append(f"{container}: {e}")

    cleared = 0
    try:
        cleared = _clear_tts_restart_flags()
    except Exception as e:
        logger.warning("Could not clear TTS restart flags: %s", e)

    if errors and not results:
        raise HTTPException(502, "; ".join(errors))

    return {
        "ok": True,
        "restarted": results,
        "errors": errors,
        "cleared_restart_flags": cleared,
    }


@router.get("/api/tts/restart-info")
async def tts_restart_info():
    containers = _tts_container_names()
    sock_ok = Path(os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock")).exists()
    return {"containers": containers, "socket_available": sock_ok}


@router.post("/api/tts/restart-flags/clear")
async def clear_tts_restart_flags_endpoint():
    try:
        cleared = _clear_tts_restart_flags()
    except Exception as e:
        raise HTTPException(500, f"Could not clear TTS restart flags: {e}")
    return {"ok": True, "cleared_restart_flags": cleared}


@router.get("/api/tts-stream-health")
async def tts_stream_health():
    from core.constants import _TTS_STREAM_DEFAULT
    settings = _load_settings()
    try:
        stream_url = settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT
        base_url = _validate_http_url(stream_url, allow_private=True).rstrip("/")
        resp = await asyncio.to_thread(requests.get, f"{base_url}/health", timeout=3)
        return {"ok": resp.ok, "status_code": resp.status_code, "url": base_url}
    except Exception as e:
        return {"ok": False, "error": str(e)}


@router.post("/api/tts-stream-session")
async def tts_stream_session(request: Request):
    data = await request.json()
    text = str(data.get("text", "")).strip()
    voice = str(data.get("voice", "")).strip()
    if not voice:
        raise HTTPException(400, "voice is required")
    if not text:
        raise HTTPException(400, "text is required")
    _purge_tts_stream_sessions()
    sid = uuid.uuid4().hex
    instruct = str(data.get("instruct") or data.get("style_instruction") or "").strip()
    _tts_stream_sessions[sid] = {"created": time.time(), "text": text, "voice": voice, "instruct": instruct}
    return {"ok": True, "url": f"/api/tts-stream-session/{sid}"}


@router.get("/api/tts-stream-session/{sid}")
async def tts_stream_playback(sid: str):
    _purge_tts_stream_sessions()
    item = _tts_stream_sessions.pop(sid, None)
    if item is None:
        raise HTTPException(404, "stream session expired or not found")
    settings = _load_settings()
    try:
        resp = await asyncio.to_thread(_open_tts_stream_response, item["text"], item["voice"], settings, item.get("instruct", ""))
    except Exception as e:
        raise HTTPException(502, f"TTS stream error: {e}")
    media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
    return StreamingResponse(
        _iter_tts_stream_response(resp),
        media_type=media_type,
        headers={
            "Cache-Control": "no-store",
            "X-Accel-Buffering": "no",
            "Content-Disposition": f"inline; filename=\"{item['voice']}_stream.wav\"",
        },
    )


@router.post("/api/tts-preview")
async def tts_preview(request: Request):
    data = await request.json()
    text: str = data["text"]
    voice: str = data["voice"]
    response_format = _requested_response_format(data)
    instruct = str(data.get("instruct") or data.get("style_instruction") or "")
    backend = _clean_preview_backend(data.get("backend", "voice_clone"))
    settings = _load_settings()

    if data.get("apply_persona"):
        scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
        wav = _find_voice_audio(voice, scan_dir)
        if wav:
            persona = _load_meta(wav).get("persona", "")
            if persona:
                llm_url = (settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/")
                llm_model = settings.get("llm_model") or ""
                try:
                    from routes.conversation import _rewrite_with_persona_sync
                    text = await asyncio.to_thread(_rewrite_with_persona_sync, text, persona, llm_url, llm_model)
                except Exception as e:
                    raise HTTPException(502, f"Persona rewrite failed: {e}")

    try:
        audio, media_type = await asyncio.to_thread(_preview_request_audio, text, voice, settings, instruct, backend)
    except Exception as e:
        raise HTTPException(502, f"TTS error: {e}")
    audio, media_type, ext, _duration_sec, _clipped = _prepare_proxy_audio(audio, media_type, response_format)
    return Response(
        content=audio,
        media_type=media_type,
        headers={"Content-Disposition": f'inline; filename="{voice}_preview.{ext}"'},
    )


@router.post("/api/tts-style-variation")
async def tts_style_variation(request: Request):
    data = await request.json()
    source_voice = str(data.get("source_voice") or data.get("voice") or "").strip()
    new_voice_id = str(data.get("voice_id") or data.get("new_voice_id") or "").strip()
    text = str(data.get("text") or data.get("transcript") or "").strip()
    instruct = str(data.get("instruct") or data.get("style_instruction") or "").strip()
    backend = _clean_preview_backend(data.get("backend", "customvoice"))
    if not source_voice:
        raise HTTPException(400, "source_voice is required")
    if not new_voice_id:
        raise HTTPException(400, "new voice id is required")
    if not re.match(r"^[A-Za-z0-9_\-\.]+$", new_voice_id):
        raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .")
    if not text:
        raise HTTPException(400, "text/transcript is required")
    if not instruct:
        raise HTTPException(400, "style instruction is required")

    settings = _load_settings()
    try:
        audio, media_type = await asyncio.to_thread(_preview_request_audio, text, source_voice, settings, instruct, backend)
        audio, media_type, _ext, duration, _clipped = _prepare_proxy_audio(audio, media_type, "wav")
    except Exception as e:
        raise HTTPException(502, f"Style variation synthesis failed: {e}")

    tmp = TEMP_DIR / f"{uuid.uuid4().hex}_style_variation.wav"
    tmp.write_bytes(audio)
    out_dir = _active_voices_dir(settings)
    out_dir.mkdir(parents=True, exist_ok=True)
    wav_dest = out_dir / f"{new_voice_id}.wav"
    txt_dest = out_dir / f"{new_voice_id}.reference.txt"
    from core.voice import _remove_audio_variants
    _remove_audio_variants(out_dir, new_voice_id)
    loudness = _export_normalized_wav(tmp, wav_dest)
    txt_dest.write_text(text, encoding="utf-8")
    meta = _load_meta(wav_dest)
    meta.update({
        "enabled": True,
        "loudness": loudness,
        "source_voice": source_voice,
        "style_instruction": instruct,
        "style_backend": backend,
        "note": f"Style variation of {source_voice}: {instruct[:180]}",
        "needs_tts_restart": True,
    })
    _save_meta(wav_dest, meta)
    return {
        "ok": True,
        "voice_id": new_voice_id,
        "source_voice": source_voice,
        "backend": backend,
        "wav": str(wav_dest),
        "txt": str(txt_dest),
        "duration": duration,
        "loudness": loudness,
        "needs_tts_restart": True,
    }


@router.post("/api/voice-design")
async def voice_design(request: Request):
    data = await request.json()
    instruct: str = data.get("instruct", "").strip()
    sample_text: str = data.get("sample_text", "Hello! This is a voice design sample.").strip()
    language: str = data.get("language", "Auto")
    gender: str = data.get("gender", "")
    dialogue = bool(data.get("dialogue"))

    if not instruct:
        raise HTTPException(400, "instruct (voice description) is required")

    settings = _load_settings()
    try:
        if dialogue:
            parsed = _parse_voice_design_dialogue(instruct, sample_text)
            if not parsed:
                raise RuntimeError("dialogue mode needs speaker profiles and Speaker: text turns")
            audio, _media_type = await asyncio.to_thread(
                _voice_design_request_audio, instruct, sample_text, language, settings, "",
            )
        else:
            audio, _media_type = await asyncio.to_thread(
                _voice_design_request_audio, instruct, sample_text, language, settings, gender,
            )
    except Exception as e:
        raise HTTPException(502, f"Voice design error: {e}")

    audio, _media_type, _ext, _duration_sec, _clipped = _prepare_proxy_audio(audio, "audio/wav", "wav")
    tmp = TEMP_DIR / f"{uuid.uuid4().hex}_designed.wav"
    tmp.write_bytes(audio)
    fid = uuid.uuid4().hex
    _registry_put(fid, tmp)
    return {"id": fid, "duration": _duration(tmp)}


@router.post("/api/tts-route-test")
async def tts_route_test(request: Request):
    data = await request.json()
    text = str(data.get("input") or data.get("text") or "").strip()
    voice = str(data.get("voice") or "default").strip()
    app_name = _canonical_app_name(str(data.get("app") or data.get("client") or "Open WebUI").strip())
    if not text:
        raise HTTPException(400, "input is required")
    routed_voice, route = _resolve_tts_route(app_name, voice, text)
    backend = _route_backend(route, routed_voice)
    settings = _load_settings()
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
    routed_audio = None if backend in {"voice_design", "nvidia_magpie"} else _find_voice_audio(routed_voice, scan_dir)
    sound_status = {}
    for key in ("before_sound", "after_sound"):
        raw = str((route or {}).get(key, ""))
        if raw:
            try:
                path = _route_sound_path(settings, raw)
                sound_status[key] = {"ok": True, "path": str(path)}
            except Exception as e:
                sound_status[key] = {"ok": False, "error": str(e), "path": raw}
    result = {
        "app": app_name,
        "requested_voice": voice,
        "routed_voice": routed_voice,
        "backend": backend,
        "detected_language": str(route.get("detected_language", _detect_text_language(text))) if route else _detect_text_language(text),
        "matched": bool(route),
        "route": route,
        "voice_health": _voice_health(routed_audio) if routed_audio else None,
        "sounds": sound_status,
    }
    _routing_log_add(
        kind="test",
        status="matched" if route else "no_match",
        app=app_name,
        requested_voice=voice,
        routed_voice=routed_voice,
        backend=backend,
        language=result["detected_language"],
        matched=bool(route),
        route_id=str((route or {}).get("id", "")),
        text_preview=text[:160],
        sounds=sound_status,
    )
    return result


# ── OpenAI-compatible models/voices proxy ─────────────────────────────────────

@router.get("/v1/models")
async def openai_models_proxy():
    now = 1686935002
    data = []
    seen = set()
    for name in sorted(_load_design_presets()):
        seen.add(_virtual_voice_id(name))
        data.append({
            "id": _virtual_voice_id(name),
            "object": "model",
            "created": now,
            "owned_by": "voice-design",
        })

    for rule in _load_tts_routes():
        alias = str(rule.get("input_voice", "")).strip()
        if rule.get("enabled", True) and alias and alias != "*" and alias not in seen:
            seen.add(alias)
            data.append({
                "id": alias,
                "object": "model",
                "created": now,
                "owned_by": f"route:{rule.get('app', '*')}",
            })

    try:
        voices = await tts_voices()
        for item in voices:
            if isinstance(item, str):
                voice_id = item
            elif isinstance(item, dict):
                voice_id = item.get("id") or item.get("voice")
            else:
                voice_id = str(item) if item is not None else ""
            if voice_id and voice_id not in seen:
                seen.add(str(voice_id))
                data.append({
                    "id": str(voice_id),
                    "object": "model",
                    "created": now,
                    "owned_by": "qwen",
                })
    except Exception:
        pass
    return {"object": "list", "data": data}


@router.get("/v1/audio/models")
async def openai_audio_models_proxy():
    return await openai_models_proxy()


@router.get("/v1/audio/voices")
async def openai_audio_voices_proxy():
    models = await openai_models_proxy()
    return [m["id"] for m in models["data"]]


# ── OpenAI-compatible speech proxy ───────────────────────────────────────────

@router.post("/v1/audio/speech")
async def openai_speech_proxy(request: Request):
    data = await request.json()
    text = str(data.get("input") or data.get("text") or "").strip()
    voice = str(data.get("voice") or data.get("model") or "").strip()
    response_format = _requested_response_format(data)
    request_app = _canonical_app_name(str(data.get("app") or data.get("client") or "").strip()) if (data.get("app") or data.get("client")) else _request_app_name(request)
    original_voice = voice
    if not text:
        _routing_log_request(
            request, status="error", app=request_app, requested_voice=original_voice,
            routed_voice=voice, backend="", route=None, response_format=response_format,
            text=text, error="input is required",
        )
        raise HTTPException(400, "input is required")
    if not voice:
        _routing_log_request(
            request, status="error", app=request_app, requested_voice=original_voice,
            routed_voice=voice, backend="", route=None, response_format=response_format,
            text=text, error="voice is required",
        )
        raise HTTPException(400, "voice is required")

    settings = _load_settings()
    voice, route = _resolve_tts_route(request_app, voice, text)
    backend = _route_backend(route, voice)
    style_instruction = str(data.get("instruct") or data.get("style_instruction") or "")
    virtual = _resolve_virtual_voice(voice)
    route_has_sounds = bool((route or {}).get("before_sound") or (route or {}).get("after_sound"))

    if backend == "streaming" and not virtual and response_format == "wav" and not route_has_sounds:
        try:
            resp = await asyncio.to_thread(_open_tts_stream_response, text, voice, settings, style_instruction)
        except Exception as e:
            _routing_log_request(
                request, status="error", app=request_app, requested_voice=original_voice,
                routed_voice=voice, backend=backend, route=route, response_format=response_format,
                text=text, error=f"TTS stream proxy error: {e}",
            )
            raise HTTPException(502, f"TTS stream proxy error: {e}")
        media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
        _routing_log_request(
            request, status="streaming", app=request_app, requested_voice=original_voice,
            routed_voice=voice, backend=backend, route=route, response_format=response_format,
            text=text, media_type=media_type, sounds=[],
        )
        return StreamingResponse(
            _iter_tts_stream_response(resp),
            media_type=media_type,
            headers={
                "Cache-Control": "no-store",
                "X-Accel-Buffering": "no",
                "Content-Disposition": f'inline; filename="{voice}_speech.wav"',
                "X-TTS-Voice-Requested": original_voice,
                "X-TTS-Voice-Routed": voice,
                "X-TTS-Route-Backend": backend,
                "X-TTS-Route-Language": str(route.get("detected_language", "")) if route else "",
                "X-TTS-Response-Format": response_format,
                "X-TTS-Audio-Clipped": "false",
                "X-TTS-Route-Sounds": "",
            },
        )

    try:
        if virtual:
            _name, preset = virtual
            instruct = str(preset.get("description", "")).strip()
            if not instruct:
                raise RuntimeError(f"Virtual voice '{voice}' has no description")
            audio, media_type = await asyncio.to_thread(
                _voice_design_request_audio,
                instruct, text, str(preset.get("language", "Auto")), settings,
                str(preset.get("gender", "")),
            )
        elif backend == "voice_design":
            audio, media_type = await asyncio.to_thread(
                _voice_design_voice_request_audio,
                voice, text, settings, style_instruction, str(data.get("language") or "Auto"),
            )
        elif backend == "streaming":
            resp = await asyncio.to_thread(_open_tts_stream_response, text, voice, settings, style_instruction)
            audio, media_type = await asyncio.to_thread(_read_tts_stream_response, resp)
            if not audio:
                raise RuntimeError("backend returned empty audio")
        elif backend == "nvidia_magpie":
            audio, media_type = await asyncio.to_thread(
                _tts_request_audio, text, voice, settings, style_instruction,
                _preview_backend_base_url(settings, "nvidia_magpie"),
                settings.get("tts_api_key", ""), "nvidia_magpie", "nvidia_magpie",
            )
        elif backend in {"nvidia_zeroshot", "nvidia_flow"}:
            audio, media_type = await asyncio.to_thread(
                _nvidia_clone_request_audio, text, voice, settings,
                "flow" if backend == "nvidia_flow" else "zeroshot",
                str(data.get("audio_prompt_transcript") or ""),
                str(data.get("language") or ""),
            )
        elif backend == "fishspeech":
            audio, media_type = await asyncio.to_thread(
                _fishspeech_request_audio, text, voice, settings,
                style_instruction, str(data.get("language") or ""),
            )
        else:
            audio, media_type = await asyncio.to_thread(
                _tts_request_audio, text, voice, settings, style_instruction,
            )
    except Exception as e:
        _routing_log_request(
            request, status="error", app=request_app, requested_voice=original_voice,
            routed_voice=voice, backend=backend, route=route, response_format=response_format,
            text=text, error=f"TTS proxy error: {e}",
        )
        raise HTTPException(502, f"TTS proxy error: {e}")

    try:
        audio, media_type, applied_sounds = _apply_route_sounds(audio, media_type, route, settings)
    except Exception as e:
        _routing_log_request(
            request, status="error", app=request_app, requested_voice=original_voice,
            routed_voice=voice, backend=backend, route=route, response_format=response_format,
            text=text, error=f"TTS route sound error: {e}",
        )
        raise HTTPException(502, f"TTS route sound error: {e}")

    audio, media_type, ext, duration, clipped = _prepare_proxy_audio(audio, media_type, response_format)
    logger.info(
        "TTS proxy app=%s voice=%s routed=%s backend=%s lang=%s format=%s bytes=%s duration=%s clipped=%s sounds=%s",
        request_app, original_voice, voice, backend,
        str(route.get("detected_language", "")) if route else "",
        response_format, len(audio),
        f"{duration:.2f}" if duration is not None else "?",
        clipped, ",".join(applied_sounds) if applied_sounds else "-",
    )
    _routing_log_request(
        request, status="ok" if route else "no_match", app=request_app,
        requested_voice=original_voice, routed_voice=voice, backend=backend,
        route=route, response_format=response_format, text=text,
        bytes=len(audio), duration=duration, clipped=clipped,
        media_type=media_type, sounds=applied_sounds,
    )

    return Response(
        content=audio,
        media_type=media_type,
        headers={
            "Content-Disposition": f'inline; filename="{voice}_speech.{ext}"',
            "X-TTS-Voice-Requested": original_voice,
            "X-TTS-Voice-Routed": voice,
            "X-TTS-Route-Backend": backend,
            "X-TTS-Route-Language": str(route.get("detected_language", "")) if route else "",
            "X-TTS-Response-Format": response_format,
            "X-TTS-Audio-Duration": f"{duration:.3f}" if duration is not None else "",
            "X-TTS-Audio-Clipped": "true" if clipped else "false",
            "X-TTS-Route-Sounds": ",".join(applied_sounds),
        },
    )


@router.post("/v1")
async def openai_speech_proxy_v1_shortcut(request: Request):
    return await openai_speech_proxy(request)