main.py

"""
Voice Cloning API
-----------------
Free stack: Whisper (STT) · MzansiLM (SA translate + story gen)
            · Google Translate (Hindi / Chinese / Tamil)
            · MMS TTS (speech synthesis — 13 languages)
            · OpenVoice v2 (voice tone transfer)

Profile persistence
  Profiles are stored locally in data/profiles/ and synced to a private
  HuggingFace Hub dataset repo (HF_PROFILES_DATASET) so they survive
  Space restarts.  Set HF_TOKEN and HF_PROFILES_DATASET env vars (Space
  secrets) to enable persistence.  The app works without them — profiles
  just won't survive a restart.
"""

from __future__ import annotations

import io
import json
import os
import re
import shutil
import subprocess
import tempfile
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Literal

import soundfile as sf
import torch
import whisper
import numpy as np
from deep_translator import GoogleTranslator
from dotenv import load_dotenv
from fastapi import Depends, FastAPI, File, Form, HTTPException, Response, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel, Field

from mms_tts import synthesize as mms_synthesize
from neural_tts import supports as neural_tts_supports
from neural_tts import synthesize as neural_tts_synthesize
from mzansi_lm import (
    generate as mzansi_generate,
    load_model as load_mzansi,
    translate as mzansi_translate,
)
from openvoice_transfer import extract_se, load_converter, transfer_voice
from security import AuthenticatedUser, require_user
from supabase_store import SupabaseStoreError, beta_store

load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), ".env"))

# ── Env / config ───────────────────────────────────────────────────────────────
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("hugging_face_token", "")
HF_PROFILES_DATASET = os.environ.get("HF_PROFILES_DATASET", "")
PROVIDER_NAME = "hybrid_neural"
_PRODUCTION_WEB_ORIGIN = "https://lingo-world.expo.app"
_LOCAL_DEVELOPMENT_ORIGINS = {
    "http://localhost:8081",
    "http://localhost:19006",
    "http://localhost:3000",
}
_configured_origins = {
    origin.strip().rstrip("/")
    for origin in os.environ.get("ALLOWED_ORIGINS", "").split(",")
    if origin.strip() and origin.strip() != "*"
}
ALLOWED_ORIGINS = sorted(
    {_PRODUCTION_WEB_ORIGIN, *_LOCAL_DEVELOPMENT_ORIGINS}
    | {
        origin
        for origin in _configured_origins
        if origin == _PRODUCTION_WEB_ORIGIN or origin in _LOCAL_DEVELOPMENT_ORIGINS
    }
)
# EAS Hosting's production alias and immutable preview aliases only. This is
# intentionally not configurable from the Space to prevent a stale `*` value
# from silently reopening browser access to private voice endpoints.
ALLOWED_ORIGIN_REGEX = r"^https://(?:[a-z0-9-]+--)?lingo-world(?:--[a-z0-9-]+)?\.expo\.app$"
_ALLOWED_ORIGIN_PATTERN = re.compile(ALLOWED_ORIGIN_REGEX)


def _is_allowed_browser_origin(origin: str) -> bool:
    """Reject unknown browser callers even if an upstream proxy reflects CORS."""
    normalized = origin.strip().rstrip("/")
    return normalized in ALLOWED_ORIGINS or bool(_ALLOWED_ORIGIN_PATTERN.fullmatch(normalized))

# ── Language config ────────────────────────────────────────────────────────────
# All 11 official South African languages — MzansiLM translates + MMS speaks
LANGUAGE_CATALOG = [
    {"code": "af", "label": "Afrikaans", "group": "South African"},
    {"code": "zul", "label": "Zulu", "group": "South African"},
    {"code": "xho", "label": "Xhosa", "group": "South African"},
    {"code": "nso", "label": "Sepedi", "group": "South African"},
    {"code": "sot", "label": "Sesotho", "group": "South African"},
    {"code": "tsn", "label": "Setswana", "group": "South African"},
    {"code": "ssw", "label": "Swati", "group": "South African"},
    {"code": "tso", "label": "Tsonga", "group": "South African"},
    {"code": "ven", "label": "Tshivenda", "group": "South African"},
    {"code": "nbl", "label": "Ndebele", "group": "South African"},
    {"code": "hi", "label": "Hindi", "group": "International"},
    {"code": "zh", "label": "Mandarin", "group": "International"},
    {"code": "ta", "label": "Tamil", "group": "International"},
]

SA_LANGS: set[str] = {
    language["code"]
    for language in LANGUAGE_CATALOG
    if language["group"] == "South African"
}

# Google translation is materially more reliable than the small base LM for
# pronunciation-sensitive read-aloud text. Codes follow Google's API.
GOOGLE_TRANSLATE_MAP: dict[str, str] = {
    "af": "af",
    "zul": "zu",
    "xho": "xh",
    "nso": "nso",
    "sot": "st",
    "tsn": "tn",
    "ssw": "ss",
    "tso": "ts",
    "ven": "ve",
    "nbl": "nr",
    "hi": "hi",
    "zh": "zh-CN",
    "ta": "ta",
}

ALL_LANGUAGES: set[str] = {language["code"] for language in LANGUAGE_CATALOG}
MAX_READ_ALOUD_CHARS = 400


def story_safety_violation(text: str) -> str | None:
    """Catch narrow, clearly high-risk requests before translation or synthesis.

    This deliberately avoids a broad keyword blacklist. User reporting and human
    review remain available for context-dependent output that this guard cannot
    reliably classify.
    """
    normalized = " ".join(text.lower().split())
    if (
        re.search(r"\b(child|minor|kid|baby)\b", normalized)
        and re.search(r"\b(sex|sexual|naked|nude)\b", normalized)
    ):
        return "sexual_content_involving_a_minor"
    if (
        re.search(r"\b(pretend to be|impersonate|sound like)\b", normalized)
        and re.search(r"\b(bank|password|pin|money|account)\b", normalized)
    ):
        return "fraud_or_deceptive_impersonation"
    if (
        re.search(r"\b(how to|exactly how|steps to)\b", normalized)
        and re.search(r"\b(hurt myself|kill myself|suicide|self-harm)\b", normalized)
    ):
        return "self_harm_instructions"
    return None

# ── Directories ────────────────────────────────────────────────────────────────
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
PROFILE_DIR = os.path.join(DATA_DIR, "profiles")
# This directory sits below PROFILE_DIR so the existing private-dataset
# snapshot restore also restores saved generations after a Space restart.
OUTPUT_DIR = os.path.join(PROFILE_DIR, "generations")
CORPUS_DIR = os.path.join(PROFILE_DIR, "corpora")
os.makedirs(PROFILE_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(CORPUS_DIR, exist_ok=True)

PROFILES_INDEX = os.path.join(PROFILE_DIR, "index.json")
GENERATIONS_INDEX = os.path.join(OUTPUT_DIR, "index.json")
REPORTS_INDEX = os.path.join(PROFILE_DIR, "content-reports.json")
MAX_PROFILE_BYTES = 25 * 1024 * 1024
MIN_PROFILE_BYTES = 1024
MAX_TRANSCRIBE_BYTES = 15 * 1024 * 1024
MIN_PROFILE_SECONDS = 3.0
MAX_PROFILE_SECONDS = 90.0
MAX_PROFILE_NAME_CHARS = 80
QUALITY_RECOMMENDED_SECONDS = 60.0
MIN_STUDIO_CORPUS_SECONDS = 30 * 60.0
MAX_STUDIO_CORPUS_SECONDS = 180 * 60.0
MIN_CORPUS_SAMPLE_SECONDS = 10.0
MAX_CORPUS_SAMPLE_SECONDS = 30 * 60.0
MAX_CORPUS_SAMPLE_BYTES = 250 * 1024 * 1024

_PROFILE_LOCK = threading.RLock()
_GENERATION_LOCK = threading.RLock()
_PROFILE_SYNC_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="profile-sync")
_USER_GENERATION_LOCKS: dict[str, threading.Lock] = {}
_USER_GENERATION_LOCKS_GUARD = threading.Lock()

# ── Global state ───────────────────────────────────────────────────────────────
whisper_model = None
MODEL_STATUS = {
    "whisper_ready": False,
    "openvoice_ready": False,
    "mzansi_ready": False,
}


# ── HuggingFace Hub helpers ────────────────────────────────────────────────────

def _hub_api():
    """Return an HfApi instance if credentials are configured, else None."""
    if not HF_TOKEN or not HF_PROFILES_DATASET:
        return None
    try:
        from huggingface_hub import HfApi
        return HfApi(token=HF_TOKEN)
    except Exception as exc:
        print(f"[HF Hub] Could not create API client: {exc}")
        return None


def _sync_profile_upload(
    profile_id: str,
    wav_bytes: bytes,
    embedding_bytes: bytes,
    index_bytes: bytes,
) -> None:
    """Atomically persist a ready profile and index snapshot to HF Hub."""
    api = _hub_api()
    if api is None:
        return
    try:
        from huggingface_hub import CommitOperationAdd

        api.create_commit(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            commit_message=f"Save voice profile {profile_id}",
            operations=[
                CommitOperationAdd(path_in_repo=f"{profile_id}.wav", path_or_fileobj=wav_bytes),
                CommitOperationAdd(path_in_repo=f"{profile_id}.pt", path_or_fileobj=embedding_bytes),
                CommitOperationAdd(path_in_repo="index.json", path_or_fileobj=index_bytes),
            ],
        )
    except Exception as exc:
        print(f"[HF Hub] Profile upload failed ({profile_id}): {exc}")


def _sync_profile_delete(
    profile_id: str,
    index_bytes: bytes,
    corpus_id: str | None = None,
) -> None:
    """Persist a deletion after earlier queued profile operations finish."""
    api = _hub_api()
    if api is None:
        return
    try:
        from huggingface_hub import CommitOperationAdd, CommitOperationDelete

        remote_files = set(api.list_repo_files(HF_PROFILES_DATASET, repo_type="dataset"))
        operations = [
            CommitOperationDelete(path_in_repo=filename)
            for filename in (f"{profile_id}.wav", f"{profile_id}.pt")
            if filename in remote_files
        ]
        if corpus_id:
            operations.extend(
                CommitOperationDelete(path_in_repo=filename)
                for filename in remote_files
                if filename.startswith(f"corpora/{corpus_id}/")
            )
        operations.append(CommitOperationAdd(path_in_repo="index.json", path_or_fileobj=index_bytes))
        api.create_commit(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            commit_message=f"Delete voice profile {profile_id}",
            operations=operations,
        )
    except Exception as exc:
        print(f"[HF Hub] Profile delete failed ({profile_id}): {exc}")


def _sync_generation_upload(
    generation_id: str,
    wav_bytes: bytes,
    index_bytes: bytes,
) -> None:
    api = _hub_api()
    if api is None:
        return
    try:
        from huggingface_hub import CommitOperationAdd

        api.create_commit(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            commit_message=f"Save generation {generation_id}",
            operations=[
                CommitOperationAdd(
                    path_in_repo=f"generations/{generation_id}.wav",
                    path_or_fileobj=wav_bytes,
                ),
                CommitOperationAdd(
                    path_in_repo="generations/index.json",
                    path_or_fileobj=index_bytes,
                ),
            ],
        )
    except Exception as exc:
        print(f"[HF Hub] Generation upload failed ({generation_id}): {exc}")


def _sync_generation_delete(generation_id: str, index_bytes: bytes) -> None:
    api = _hub_api()
    if api is None:
        return
    try:
        from huggingface_hub import CommitOperationAdd, CommitOperationDelete

        remote_path = f"generations/{generation_id}.wav"
        remote_files = set(api.list_repo_files(HF_PROFILES_DATASET, repo_type="dataset"))
        operations = []
        if remote_path in remote_files:
            operations.append(CommitOperationDelete(path_in_repo=remote_path))
        operations.append(
            CommitOperationAdd(
                path_in_repo="generations/index.json",
                path_or_fileobj=index_bytes,
            )
        )
        api.create_commit(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            commit_message=f"Delete generation {generation_id}",
            operations=operations,
        )
    except Exception as exc:
        print(f"[HF Hub] Generation delete failed ({generation_id}): {exc}")


def _sync_corpus_sample(
    corpus_id: str,
    sample_id: str,
    wav_bytes: bytes,
    manifest_bytes: bytes,
) -> None:
    api = _hub_api()
    if api is None:
        return
    try:
        from huggingface_hub import CommitOperationAdd

        api.create_commit(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            commit_message=f"Add studio corpus sample {sample_id}",
            operations=[
                CommitOperationAdd(
                    path_in_repo=f"corpora/{corpus_id}/{sample_id}.wav",
                    path_or_fileobj=wav_bytes,
                ),
                CommitOperationAdd(
                    path_in_repo=f"corpora/{corpus_id}/manifest.json",
                    path_or_fileobj=manifest_bytes,
                ),
            ],
        )
    except Exception as exc:
        print(f"[HF Hub] Corpus sample upload failed ({sample_id}): {exc}")


def _enqueue_profile_upload(
    profile_id: str,
    wav_bytes: bytes,
    embedding_bytes: bytes,
    index_bytes: bytes,
) -> None:
    _PROFILE_SYNC_EXECUTOR.submit(
        _sync_profile_upload,
        profile_id,
        wav_bytes,
        embedding_bytes,
        index_bytes,
    )


def _enqueue_profile_delete(
    profile_id: str,
    index_bytes: bytes,
    corpus_id: str | None = None,
) -> None:
    _PROFILE_SYNC_EXECUTOR.submit(_sync_profile_delete, profile_id, index_bytes, corpus_id)


def _enqueue_generation_upload(
    generation_id: str,
    wav_bytes: bytes,
    index_bytes: bytes,
) -> None:
    _PROFILE_SYNC_EXECUTOR.submit(
        _sync_generation_upload,
        generation_id,
        wav_bytes,
        index_bytes,
    )


def _enqueue_generation_delete(generation_id: str, index_bytes: bytes) -> None:
    _PROFILE_SYNC_EXECUTOR.submit(_sync_generation_delete, generation_id, index_bytes)


def _enqueue_corpus_sample(
    corpus_id: str,
    sample_id: str,
    wav_bytes: bytes,
    manifest_bytes: bytes,
) -> None:
    _PROFILE_SYNC_EXECUTOR.submit(
        _sync_corpus_sample,
        corpus_id,
        sample_id,
        wav_bytes,
        manifest_bytes,
    )


def _restore_profiles_from_hub() -> None:
    """Download all profiles from HF Hub on startup (Space restart recovery)."""
    api = _hub_api()
    if api is None:
        print("[HF Hub] Not configured — skipping profile restore.")
        return
    try:
        from huggingface_hub import snapshot_download
        snapshot_download(
            repo_id=HF_PROFILES_DATASET,
            repo_type="dataset",
            local_dir=PROFILE_DIR,
            token=HF_TOKEN,
            ignore_patterns=["*.gitattributes", "README.md", ".gitattributes"],
        )
        print("[HF Hub] Profiles restored.")
    except Exception as exc:
        print(f"[HF Hub] Profile restore failed: {exc}")


def _generation_lock_for(user_id: str) -> threading.Lock:
    with _USER_GENERATION_LOCKS_GUARD:
        return _USER_GENERATION_LOCKS.setdefault(user_id, threading.Lock())


def require_generation_slot(
    user: AuthenticatedUser = Depends(require_user),
):
    lock = _generation_lock_for(user.id)
    if not lock.acquire(blocking=False):
        raise HTTPException(
            status_code=429,
            detail="A voice generation is already running for this account",
        )
    try:
        yield user
    finally:
        lock.release()


def _belongs_to_user(record: dict, user_id: str) -> bool:
    return record.get("user_id") == user_id


def _client_record(record: dict) -> dict:
    hidden = {"user_id", "wav_path", "embedding_path", "storage_path"}
    return {key: value for key, value in record.items() if key not in hidden}


def _profiles_for_user(user_id: str) -> list[dict]:
    if beta_store.enabled:
        try:
            return beta_store.list_profiles(user_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    return [profile for profile in load_index() if _belongs_to_user(profile, user_id)]


def _profile_for_user(user_id: str, profile_id: str) -> dict | None:
    if beta_store.enabled:
        try:
            return beta_store.get_profile(user_id, profile_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    return next(
        (
            profile
            for profile in load_index()
            if profile.get("id") == profile_id and _belongs_to_user(profile, user_id)
        ),
        None,
    )


def _corpus_for_user(user_id: str, corpus_id: str) -> dict | None:
    if beta_store.enabled:
        try:
            corpus = beta_store.get_corpus(user_id, corpus_id)
            if corpus:
                _save_corpus(corpus)
            return corpus
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    corpus = _load_corpus(corpus_id)
    return corpus if corpus and _belongs_to_user(corpus, user_id) else None


def _generations_for_user(user_id: str) -> list[dict]:
    if beta_store.enabled:
        try:
            return beta_store.list_generations(user_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    return [
        generation
        for generation in load_generations()
        if _belongs_to_user(generation, user_id)
    ]


# ── Profile index helpers ──────────────────────────────────────────────────────

def load_index() -> list:
    with _PROFILE_LOCK:
        if not os.path.exists(PROFILES_INDEX):
            return []
        try:
            with open(PROFILES_INDEX, encoding="utf-8") as f:
                profiles = json.load(f)
        except (OSError, json.JSONDecodeError) as exc:
            print(f"[Profiles] Could not read index: {exc}")
            return []
        return profiles if isinstance(profiles, list) else []


def save_index(profiles: list) -> None:
    with _PROFILE_LOCK:
        os.makedirs(os.path.dirname(PROFILES_INDEX), exist_ok=True)
        fd, temp_path = tempfile.mkstemp(
            prefix="profiles-",
            suffix=".json.tmp",
            dir=os.path.dirname(PROFILES_INDEX),
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(profiles, f, indent=2)
                f.flush()
                os.fsync(f.fileno())
            os.replace(temp_path, PROFILES_INDEX)
        finally:
            if os.path.exists(temp_path):
                os.unlink(temp_path)


def _index_bytes(profiles: list) -> bytes:
    return json.dumps(profiles, indent=2).encode("utf-8")


def load_generations() -> list:
    with _GENERATION_LOCK:
        if not os.path.exists(GENERATIONS_INDEX):
            return []
        try:
            with open(GENERATIONS_INDEX, encoding="utf-8") as index_file:
                generations = json.load(index_file)
        except (OSError, json.JSONDecodeError) as exc:
            print(f"[Generations] Could not read index: {exc}")
            return []
        return generations if isinstance(generations, list) else []


def save_generations(generations: list) -> None:
    with _GENERATION_LOCK:
        os.makedirs(OUTPUT_DIR, exist_ok=True)
        fd, temp_path = tempfile.mkstemp(
            prefix="generations-",
            suffix=".json.tmp",
            dir=OUTPUT_DIR,
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as index_file:
                json.dump(generations, index_file, indent=2, ensure_ascii=False)
                index_file.flush()
                os.fsync(index_file.fileno())
            os.replace(temp_path, GENERATIONS_INDEX)
        finally:
            if os.path.exists(temp_path):
                os.unlink(temp_path)


def load_content_reports() -> list[dict]:
    try:
        with open(REPORTS_INDEX, "r", encoding="utf-8") as report_file:
            reports = json.load(report_file)
    except (FileNotFoundError, json.JSONDecodeError):
        return []
    return reports if isinstance(reports, list) else []


def save_content_reports(reports: list[dict]) -> None:
    reports_dir = os.path.dirname(REPORTS_INDEX)
    os.makedirs(reports_dir, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=reports_dir,
        prefix="content-reports-",
        suffix=".json",
        delete=False,
    ) as report_file:
        json.dump(reports, report_file, indent=2, ensure_ascii=False)
        temp_path = report_file.name
    os.replace(temp_path, REPORTS_INDEX)


def _generation_path(generation_id: str) -> str:
    return os.path.join(OUTPUT_DIR, f"{generation_id}.wav")


def _language_label(language: str) -> str:
    entry = next(item for item in LANGUAGE_CATALOG if item["code"] == language)
    return entry["label"]


def _translate_text(text: str, language: str) -> tuple[str, str]:
    target_code = GOOGLE_TRANSLATE_MAP[language]
    translated = GoogleTranslator(source="en", target=target_code).translate(text)
    translated = (translated or "").strip()
    if not translated:
        raise RuntimeError("Translation produced no speech text")
    return translated, "google"


def _profile_path(profile_id: str, extension: str) -> str:
    return os.path.join(PROFILE_DIR, f"{profile_id}{extension}")


def _corpus_path(corpus_id: str) -> str:
    return os.path.join(CORPUS_DIR, corpus_id)


def _corpus_manifest_path(corpus_id: str) -> str:
    return os.path.join(_corpus_path(corpus_id), "manifest.json")


def _load_corpus(corpus_id: str) -> dict | None:
    path = _corpus_manifest_path(corpus_id)
    if not os.path.exists(path):
        return None
    try:
        with open(path, encoding="utf-8") as corpus_file:
            corpus = json.load(corpus_file)
    except (OSError, json.JSONDecodeError):
        return None
    return corpus if isinstance(corpus, dict) else None


def _save_corpus(corpus: dict) -> None:
    corpus_dir = _corpus_path(corpus["id"])
    os.makedirs(corpus_dir, exist_ok=True)
    fd, temp_path = tempfile.mkstemp(prefix="manifest-", suffix=".tmp", dir=corpus_dir)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as corpus_file:
            json.dump(corpus, corpus_file, indent=2, ensure_ascii=False)
            corpus_file.flush()
            os.fsync(corpus_file.fileno())
        os.replace(temp_path, _corpus_manifest_path(corpus["id"]))
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)


def _clean_profile_files(profile_id: str) -> None:
    for extension in (".wav", ".pt"):
        path = _profile_path(profile_id, extension)
        if os.path.exists(path):
            os.unlink(path)


def _normalise_profile_name(name: str) -> str:
    cleaned = " ".join((name or "").split())
    return (cleaned or "My Voice")[:MAX_PROFILE_NAME_CHARS]


def _safe_audio_suffix(filename: str) -> str:
    suffix = os.path.splitext(filename or "")[1].lower()
    return suffix if suffix in {".wav", ".webm", ".ogg", ".mp3", ".m4a", ".mp4", ".aac"} else ".audio"


def _analyse_profile_audio(path: str, duration: float) -> dict:
    """Return transparent, non-destructive diagnostics for a voice reference."""
    samples, sample_rate = sf.read(path, dtype="float32", always_2d=False)
    samples = np.asarray(samples, dtype=np.float32).reshape(-1)
    if not len(samples):
        raise ValueError("Recording contains no audio samples")

    peak = float(np.max(np.abs(samples)))
    rms = float(np.sqrt(np.mean(np.square(samples))))
    peak_dbfs = 20 * np.log10(max(peak, 1e-8))
    rms_dbfs = 20 * np.log10(max(rms, 1e-8))
    clipped_fraction = float(np.mean(np.abs(samples) >= 0.99))

    # 40 ms windows make this robust to normal phonetic pauses while still
    # identifying recordings that are mostly silence.
    window = max(1, int(sample_rate * 0.04))
    usable = samples[: len(samples) - (len(samples) % window)]
    if len(usable):
        window_rms = np.sqrt(np.mean(np.square(usable.reshape(-1, window)), axis=1))
        active_ratio = float(np.mean(window_rms >= max(rms * 0.2, 0.003)))
    else:
        active_ratio = 0.0

    feedback = []
    if duration < QUALITY_RECOMMENDED_SECONDS:
        feedback.append("A 60–90 second reference will improve instant voice matching.")
    if rms_dbfs < -35:
        feedback.append("The recording is quiet. Move closer to the microphone and record in a quieter room.")
    if peak_dbfs > -1 or clipped_fraction > 0.001:
        feedback.append("The recording is clipping. Speak a little softer or move farther from the microphone.")
    if active_ratio < 0.45:
        feedback.append("The recording contains long silent sections. Keep a steady, natural reading pace.")

    grade = "good" if not feedback else "needs_improvement"
    return {
        "grade": grade,
        "duration_seconds": round(duration, 1),
        "sample_rate_hz": sample_rate,
        "peak_dbfs": round(peak_dbfs, 1),
        "rms_dbfs": round(rms_dbfs, 1),
        "active_ratio": round(active_ratio, 2),
        "feedback": feedback,
    }


def _normalise_profile_audio(
    audio_bytes: bytes,
    filename: str,
    destination: str,
    min_seconds: float = MIN_PROFILE_SECONDS,
    max_seconds: float = MAX_PROFILE_SECONDS,
) -> float:
    """Convert browser/mobile audio into a high-resolution mono PCM WAV."""
    suffix = _safe_audio_suffix(filename)

    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as source_file:
        source_file.write(audio_bytes)
        source_path = source_file.name

    try:
        result = subprocess.run(
            [
                "ffmpeg", "-y", "-v", "error", "-i", source_path,
                "-ac", "1", "-ar", "24000", "-c:a", "pcm_s16le", destination,
            ],
            capture_output=True,
            text=True,
            timeout=90,
            check=False,
        )
        if result.returncode != 0 or not os.path.exists(destination):
            detail = result.stderr.strip() or "unsupported or damaged audio"
            raise ValueError(f"Could not decode recording: {detail}")

        info = sf.info(destination)
        duration = float(info.duration)
        if duration < min_seconds:
            raise ValueError(
                f"Recording is too short ({duration:.1f}s). Record at least {min_seconds:.0f} seconds."
            )
        if duration > max_seconds:
            limit_label = (
                f"{max_seconds:.0f} seconds"
                if max_seconds < 120
                else f"{max_seconds / 60:.0f} minutes"
            )
            raise ValueError(
                f"Recording is too long ({duration:.1f}s). Keep it under {limit_label}."
            )
        return duration
    finally:
        try:
            os.unlink(source_path)
        except OSError:
            pass


# ── Lifespan (startup / shutdown) ─────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    global whisper_model

    print("=== Startup ===")

    print("Loading Whisper base …")
    whisper_model = whisper.load_model("base")
    MODEL_STATUS["whisper_ready"] = True
    print("Whisper ready.")

    ckpt_dir = os.path.join(os.path.dirname(__file__), "checkpoints_v2", "converter")
    print(f"Loading OpenVoice v2 converter from {ckpt_dir} …")
    load_converter(ckpt_dir)
    MODEL_STATUS["openvoice_ready"] = True
    print("OpenVoice v2 ready.")

    print("Loading MzansiLM …")
    load_mzansi()
    MODEL_STATUS["mzansi_ready"] = True
    print("MzansiLM ready.")

    print("Restoring profiles from HF Hub …")
    _restore_profiles_from_hub()

    print("=== API ready ===")
    yield


# ── Pydantic request bodies ────────────────────────────────────────────────────

class GenerateRequest(BaseModel):
    seed: str
    language: str = "zul"
    max_tokens: int = 200


class TranslateRequest(BaseModel):
    text: str
    language: str


class AudioGenerationRequest(BaseModel):
    text: str
    language: str
    profile_id: str = ""
    voice_mode: str = "cloned"
    translate_text: bool = True


class ContentReportRequest(BaseModel):
    reason: Literal[
        "scary_or_upsetting",
        "unsafe_or_inappropriate",
        "wrong_voice_or_language",
        "other",
    ]
    details: str = Field(default="", max_length=500)



# ── App ────────────────────────────────────────────────────────────────────────

app = FastAPI(
    title="Lingo World API",
    version="1.0.0",
    description="Multilingual neural TTS with optional OpenVoice tone transfer",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_origin_regex=ALLOWED_ORIGIN_REGEX,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=[
        "X-Generation-Id",
        "X-Generation-Time-Ms",
        "X-Translation-Provider",
        "X-TTS-Provider",
    ],
)


@app.middleware("http")
async def reject_untrusted_browser_origins(request, call_next):
    origin = request.headers.get("origin")
    if origin and not _is_allowed_browser_origin(origin):
        return JSONResponse(
            status_code=403,
            content={"detail": "This browser origin is not allowed"},
        )
    return await call_next(request)


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

@app.get("/")
def root():
    return {
        "status": "ok",
        "provider": PROVIDER_NAME,
        "message": "Voice Cloning API — native neural TTS + OpenVoice v2",
        "languages": sorted(ALL_LANGUAGES),
    }


@app.get("/health")
def health():
    return {
        "status": "ready" if all(MODEL_STATUS.values()) else "starting",
        **MODEL_STATUS,
    }


@app.get("/languages")
def languages():
    return LANGUAGE_CATALOG


@app.post("/v1/translate")
def translate(
    body: TranslateRequest,
    user: AuthenticatedUser = Depends(require_user),
):
    text = body.text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="'text' must not be empty")
    if body.language not in ALL_LANGUAGES:
        raise HTTPException(status_code=400, detail="Unsupported language")
    try:
        translated, provider = _translate_text(text, body.language)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"Translation failed: {exc}") from exc
    return {
        "translated": translated,
        "language": body.language,
        "provider": provider,
    }


@app.post("/v1/profiles")
async def upload_profile(
    file: UploadFile = File(...),
    name: str = Form(default="My Voice"),
    consent: bool = Form(default=False),
    user: AuthenticatedUser = Depends(require_user),
):
    """
    Upload a voice sample (~3–90 s) in a browser or mobile audio format.
    Normalises it to WAV, extracts a speaker embedding, and stores both locally
    and on HF Hub for persistence across Space restarts.
    """
    if not consent:
        raise HTTPException(
            status_code=403,
            detail="Confirm that you own this voice and consent to creating a reusable voice profile.",
        )

    audio_bytes = await file.read(MAX_PROFILE_BYTES + 1)
    if len(audio_bytes) < MIN_PROFILE_BYTES:
        raise HTTPException(status_code=400, detail="The recording is empty or too small")
    if len(audio_bytes) > MAX_PROFILE_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"Recording is larger than {MAX_PROFILE_BYTES // (1024 * 1024)} MB",
        )

    profile_id = str(uuid.uuid4())
    profile_name = _normalise_profile_name(name)
    wav_path = _profile_path(profile_id, ".wav")
    se_path = _profile_path(profile_id, ".pt")

    # 1. Decode browser/mobile recording and validate duration.
    try:
        duration = _normalise_profile_audio(audio_bytes, file.filename or "recording", wav_path)

        quality = _analyse_profile_audio(wav_path, duration)

        # 2. Extract and save the reusable speaker embedding.
        se = extract_se(wav_path)
        torch.save(se, se_path)
    except ValueError as exc:
        _clean_profile_files(profile_id)
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        _clean_profile_files(profile_id)
        raise HTTPException(
            status_code=500,
            detail=f"Speaker embedding extraction failed: {exc}",
        ) from exc

    # 3. Update local index
    try:
        with _PROFILE_LOCK:
            profiles = load_index()
            profile = {
                "id": profile_id,
                "user_id": user.id,
                "name": profile_name,
                "profile_type": "instant",
                "quality": quality,
                "consented_at": datetime.now(timezone.utc).isoformat(),
                "created_at": datetime.now(timezone.utc).isoformat(),
            }
            profiles.append(profile)
            save_index(profiles)
    except Exception as exc:
        _clean_profile_files(profile_id)
        raise HTTPException(status_code=500, detail=f"Could not save profile index: {exc}") from exc

    # 4. Persist new beta data in the caller's private Supabase namespace.
    if beta_store.enabled:
        try:
            beta_store.save_profile(user.id, profile, wav_path, se_path)
        except SupabaseStoreError as exc:
            with _PROFILE_LOCK:
                save_index([item for item in profiles if item.get("id") != profile_id])
            _clean_profile_files(profile_id)
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        # Local-development fallback only. The legacy HF dataset remains
        # supported for existing installations but is not used by production.
        with open(wav_path, "rb") as wav_file, open(se_path, "rb") as embedding_file:
            _enqueue_profile_upload(
                profile_id,
                wav_file.read(),
                embedding_file.read(),
                _index_bytes(profiles),
            )

    return {
        "profile_id": profile_id,
        "name": profile_name,
        "duration_seconds": round(duration, 1),
        "quality": quality,
        "sync_status": (
            "supabase"
            if beta_store.enabled
            else ("queued" if _hub_api() is not None else "local_only")
        ),
    }


def _studio_corpus_response(corpus: dict) -> dict:
    total_seconds = float(corpus.get("total_seconds", 0.0))
    return {
        **_client_record(corpus),
        "total_minutes": round(total_seconds / 60, 1),
        "minimum_minutes": int(MIN_STUDIO_CORPUS_SECONDS / 60),
        "maximum_minutes": int(MAX_STUDIO_CORPUS_SECONDS / 60),
        "ready_to_finalize": total_seconds >= MIN_STUDIO_CORPUS_SECONDS,
        "progress_percent": round(
            min(100.0, total_seconds / MIN_STUDIO_CORPUS_SECONDS * 100),
            1,
        ),
    }


@app.post("/v1/studio-corpora")
def create_studio_corpus(
    name: str = Form(default="Studio Voice"),
    consent: bool = Form(default=False),
    user: AuthenticatedUser = Depends(require_user),
):
    """Start a resumable 30–180 minute high-fidelity voice corpus."""
    if not consent:
        raise HTTPException(
            status_code=403,
            detail="Confirm that you own this voice and consent to corpus training and storage.",
        )
    corpus_id = str(uuid.uuid4())
    corpus = {
        "id": corpus_id,
        "user_id": user.id,
        "name": _normalise_profile_name(name),
        "status": "collecting",
        "samples": [],
        "total_seconds": 0.0,
        "consented_at": datetime.now(timezone.utc).isoformat(),
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    with _PROFILE_LOCK:
        _save_corpus(corpus)
    if beta_store.enabled:
        try:
            beta_store.save_corpus(user.id, corpus)
        except SupabaseStoreError as exc:
            shutil.rmtree(_corpus_path(corpus_id), ignore_errors=True)
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    return _studio_corpus_response(corpus)


@app.get("/v1/studio-corpora/{corpus_id}")
def studio_corpus_status(
    corpus_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    corpus = _corpus_for_user(user.id, corpus_id)
    if corpus is None:
        raise HTTPException(status_code=404, detail="Studio corpus not found")
    return _studio_corpus_response(corpus)


@app.post("/v1/studio-corpora/{corpus_id}/samples")
async def add_studio_corpus_sample(
    corpus_id: str,
    file: UploadFile = File(...),
    user: AuthenticatedUser = Depends(require_user),
):
    """Normalise, evaluate, and append one recording to a studio corpus."""
    with _PROFILE_LOCK:
        corpus = _corpus_for_user(user.id, corpus_id)
        if corpus is None:
            raise HTTPException(status_code=404, detail="Studio corpus not found")
        if corpus.get("status") != "collecting":
            raise HTTPException(status_code=409, detail="Studio corpus is already finalized")

    audio_bytes = await file.read(MAX_CORPUS_SAMPLE_BYTES + 1)
    if len(audio_bytes) < MIN_PROFILE_BYTES:
        raise HTTPException(status_code=400, detail="The recording is empty or too small")
    if len(audio_bytes) > MAX_CORPUS_SAMPLE_BYTES:
        raise HTTPException(status_code=413, detail="Studio sample is larger than 250 MB")

    sample_id = str(uuid.uuid4())
    sample_path = os.path.join(_corpus_path(corpus_id), f"{sample_id}.wav")
    try:
        duration = _normalise_profile_audio(
            audio_bytes,
            file.filename or "studio-sample",
            sample_path,
            min_seconds=MIN_CORPUS_SAMPLE_SECONDS,
            max_seconds=MAX_CORPUS_SAMPLE_SECONDS,
        )
        quality = _analyse_profile_audio(sample_path, duration)
    except ValueError as exc:
        if os.path.exists(sample_path):
            os.unlink(sample_path)
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        if os.path.exists(sample_path):
            os.unlink(sample_path)
        raise HTTPException(status_code=500, detail=f"Could not process corpus sample: {exc}") from exc

    with _PROFILE_LOCK:
        corpus = _corpus_for_user(user.id, corpus_id)
        if corpus is None or corpus.get("status") != "collecting":
            os.unlink(sample_path)
            raise HTTPException(status_code=409, detail="Studio corpus is no longer accepting samples")
        new_total = float(corpus.get("total_seconds", 0.0)) + duration
        if new_total > MAX_STUDIO_CORPUS_SECONDS:
            os.unlink(sample_path)
            raise HTTPException(
                status_code=400,
                detail="This sample would take the corpus beyond the 180-minute maximum.",
            )
        corpus["samples"].append({
            "id": sample_id,
            "filename": file.filename or "studio-sample",
            "duration_seconds": round(duration, 1),
            "quality": quality,
        })
        corpus["total_seconds"] = round(new_total, 1)
        corpus["updated_at"] = datetime.now(timezone.utc).isoformat()
        _save_corpus(corpus)

    if beta_store.enabled:
        try:
            beta_store.save_corpus_sample(
                user.id,
                corpus_id,
                corpus["samples"][-1],
                sample_path,
                corpus,
            )
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        with open(sample_path, "rb") as sample_file:
            _enqueue_corpus_sample(
                corpus_id,
                sample_id,
                sample_file.read(),
                json.dumps(corpus, indent=2, ensure_ascii=False).encode("utf-8"),
            )
    return _studio_corpus_response(corpus)


@app.post("/v1/studio-corpora/{corpus_id}/finalize")
def finalize_studio_corpus(
    corpus_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    """Build one duration-weighted OpenVoice identity from the full corpus."""
    with _PROFILE_LOCK:
        corpus = _corpus_for_user(user.id, corpus_id)
        if corpus is None:
            raise HTTPException(status_code=404, detail="Studio corpus not found")
        if corpus.get("status") == "ready" and corpus.get("profile_id"):
            return {**_studio_corpus_response(corpus), "profile_id": corpus["profile_id"]}
        if float(corpus.get("total_seconds", 0.0)) < MIN_STUDIO_CORPUS_SECONDS:
            remaining = (MIN_STUDIO_CORPUS_SECONDS - float(corpus.get("total_seconds", 0.0))) / 60
            raise HTTPException(
                status_code=400,
                detail=f"Add at least {remaining:.1f} more minutes before finalizing.",
            )
        samples = list(corpus.get("samples", []))

    embeddings = []
    weights = []
    try:
        if beta_store.enabled:
            beta_store.restore_corpus_samples(corpus, _corpus_path(corpus_id))
        for sample in samples:
            sample_path = os.path.join(_corpus_path(corpus_id), f"{sample['id']}.wav")
            if not os.path.exists(sample_path):
                raise RuntimeError(f"Corpus sample is missing: {sample['filename']}")
            embeddings.append(extract_se(sample_path))
            # Cap one file's influence so varied sessions contribute to identity.
            weights.append(min(float(sample["duration_seconds"]), 300.0))

        total_weight = sum(weights)
        aggregate_se = embeddings[0] * weights[0]
        for embedding, weight in zip(embeddings[1:], weights[1:]):
            aggregate_se = aggregate_se + embedding * weight
        aggregate_se = aggregate_se / total_weight
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Studio profile extraction failed: {exc}") from exc

    profile_id = str(uuid.uuid4())
    se_path = _profile_path(profile_id, ".pt")
    wav_path = _profile_path(profile_id, ".wav")
    representative = max(samples, key=lambda item: float(item["duration_seconds"]))
    representative_path = os.path.join(_corpus_path(corpus_id), f"{representative['id']}.wav")
    torch.save(aggregate_se, se_path)
    shutil.copyfile(representative_path, wav_path)

    profile = {
        "id": profile_id,
        "user_id": user.id,
        "name": corpus["name"],
        "profile_type": "studio_corpus",
        "corpus_id": corpus_id,
        "corpus_minutes": round(float(corpus["total_seconds"]) / 60, 1),
        "sample_count": len(samples),
        "consented_at": corpus["consented_at"],
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    with _PROFILE_LOCK:
        profiles = load_index()
        profiles.append(profile)
        save_index(profiles)
        corpus["status"] = "ready"
        corpus["profile_id"] = profile_id
        corpus["finalized_at"] = datetime.now(timezone.utc).isoformat()
        _save_corpus(corpus)

    if beta_store.enabled:
        try:
            beta_store.save_profile(user.id, profile, wav_path, se_path)
            beta_store.save_corpus(user.id, corpus)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        with open(wav_path, "rb") as wav_file, open(se_path, "rb") as embedding_file:
            _enqueue_profile_upload(
                profile_id,
                wav_file.read(),
                embedding_file.read(),
                _index_bytes(profiles),
            )
    return {**_studio_corpus_response(corpus), "profile_id": profile_id}


@app.delete("/v1/studio-corpora/{corpus_id}")
def delete_studio_corpus(
    corpus_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    corpus = _corpus_for_user(user.id, corpus_id)
    if corpus is None:
        raise HTTPException(status_code=404, detail="Studio corpus not found")
    if corpus.get("status") == "ready":
        raise HTTPException(
            status_code=409,
            detail="Delete the finalized voice profile before removing its retained corpus.",
        )
    if beta_store.enabled:
        try:
            beta_store.delete_corpus(user.id, corpus_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    shutil.rmtree(_corpus_path(corpus_id), ignore_errors=True)
    return {"deleted": corpus_id}


@app.get("/v1/profiles")
def list_profiles(user: AuthenticatedUser = Depends(require_user)):
    return [_client_record(profile) for profile in _profiles_for_user(user.id)]


@app.delete("/v1/profiles/{profile_id}")
def delete_profile(
    profile_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    with _PROFILE_LOCK:
        profiles = load_index()
        entry = _profile_for_user(user.id, profile_id)
        if not entry:
            raise HTTPException(status_code=404, detail="Profile not found")

        remaining = [
            profile
            for profile in profiles
            if not (
                profile.get("id") == profile_id
                and _belongs_to_user(profile, user.id)
            )
        ]
        save_index(remaining)
        _clean_profile_files(profile_id)
        corpus_id = entry.get("corpus_id")
        if corpus_id and os.path.isdir(_corpus_path(corpus_id)):
            shutil.rmtree(_corpus_path(corpus_id))

    if beta_store.enabled:
        try:
            beta_store.delete_profile(user.id, profile_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    elif entry.get("corpus_id"):
        _enqueue_profile_delete(profile_id, _index_bytes(remaining), entry["corpus_id"])
    else:
        _enqueue_profile_delete(profile_id, _index_bytes(remaining))
    return {"deleted": profile_id}


@app.post("/v1/transcriptions")
async def transcribe(
    file: UploadFile = File(...),
    user: AuthenticatedUser = Depends(require_user),
):
    """Transcribe English speech to text using Whisper."""
    if whisper_model is None:
        raise HTTPException(status_code=503, detail="Whisper not ready")

    audio_bytes = await file.read(MAX_TRANSCRIBE_BYTES + 1)
    if len(audio_bytes) < MIN_PROFILE_BYTES:
        raise HTTPException(status_code=400, detail="The story recording is empty or too small")
    if len(audio_bytes) > MAX_TRANSCRIBE_BYTES:
        raise HTTPException(status_code=413, detail="The story recording is too large")

    with tempfile.NamedTemporaryFile(
        suffix=_safe_audio_suffix(file.filename or "recording"),
        delete=False,
    ) as tmp:
        tmp.write(audio_bytes)
        tmp_path = tmp.name
    try:
        try:
            result = whisper_model.transcribe(tmp_path, language="en")
        except Exception as exc:
            raise HTTPException(status_code=400, detail=f"Could not decode story recording: {exc}") from exc
        return {"text": result["text"].strip()}
    finally:
        os.unlink(tmp_path)


async def generate_audio(
    text: str,
    language: str,
    profile_id: str = "",
    voice_mode: str = "cloned",
    translate_text: bool = True,
    user: AuthenticatedUser = Depends(require_generation_slot),
):
    """
    Translate text, synthesize with the best language-native source, optionally
    apply the user's cloned timbre, and retain the finished generation.

    language: af | zul | xho | nso | sot | ssw | tsn | tso | ven | nbl
              hi | zh | ta
    """
    if language not in ALL_LANGUAGES:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported language '{language}'. Supported: {sorted(ALL_LANGUAGES)}",
        )
    if voice_mode not in {"natural", "cloned"}:
        raise HTTPException(status_code=400, detail="voice_mode must be 'natural' or 'cloned'")

    text = text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="Story text is required")
    if len(text) > MAX_READ_ALOUD_CHARS:
        raise HTTPException(
            status_code=400,
            detail=f"Story text must be {MAX_READ_ALOUD_CHARS} characters or fewer",
        )
    if story_safety_violation(text):
        raise HTTPException(
            status_code=400,
            detail="This story needs a grown-up's help because it may be unsafe or inappropriate.",
        )

    started_at = time.perf_counter()

    target_se = None
    if voice_mode == "cloned":
        profile = _profile_for_user(user.id, profile_id)
        if not profile:
            raise HTTPException(status_code=404, detail="Voice profile not found")
        se_path = _profile_path(profile_id, ".pt")
        if beta_store.enabled:
            try:
                beta_store.restore_profile_files(
                    profile,
                    _profile_path(profile_id, ".wav"),
                    se_path,
                )
            except SupabaseStoreError as exc:
                raise HTTPException(status_code=503, detail=str(exc)) from exc
        if not os.path.exists(se_path):
            raise HTTPException(status_code=404, detail="Voice profile audio is unavailable")
        target_se = torch.load(se_path, weights_only=True)

    # ── Step 1: Translate ─────────────────────────────────────────────────────
    if translate_text:
        try:
            translated, translation_provider = _translate_text(text, language)
        except Exception as exc:
            raise HTTPException(status_code=502, detail=f"Translation failed: {exc}") from exc
    else:
        translated = text
        translation_provider = "reviewed"

    translated = translated.strip()
    if not translated:
        raise HTTPException(status_code=500, detail="Translation produced no speech text")

    # ── Step 2: MMS TTS → WAV ─────────────────────────────────────────────────
    try:
        if neural_tts_supports(language):
            try:
                source_wav_bytes = await neural_tts_synthesize(translated, language)
                tts_provider = "native_neural"
            except Exception as neural_exc:
                print(f"[Neural TTS] Falling back for {language}: {neural_exc}")
                source_wav_bytes = mms_synthesize(translated, language)
                tts_provider = "local_fallback"
        else:
            source_wav_bytes = mms_synthesize(translated, language)
            tts_provider = "mms" if language in {"tso", "hi", "ta"} else "local_fallback"
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Speech synthesis failed: {exc}") from exc

    # ── Step 3: OpenVoice v2 voice transfer ───────────────────────────────────
    if voice_mode == "cloned":
        try:
            output_wav_bytes = transfer_voice(source_wav_bytes, target_se, tau=0.22)
        except Exception as exc:
            raise HTTPException(status_code=500, detail=f"Voice transfer failed: {exc}") from exc
    else:
        output_wav_bytes = source_wav_bytes

    generation_id = str(uuid.uuid4())
    output_path = _generation_path(generation_id)
    with open(output_path, "wb") as output_file:
        output_file.write(output_wav_bytes)

    profile_name = "Natural voice"
    if profile_id:
        profile = _profile_for_user(user.id, profile_id)
        if profile:
            profile_name = profile.get("name", "My Voice")
    generation = {
        "id": generation_id,
        "user_id": user.id,
        "profile_id": profile_id or None,
        "profile_name": profile_name,
        "language": language,
        "language_label": _language_label(language),
        "voice_mode": voice_mode,
        "tts_provider": tts_provider,
        "translation_provider": translation_provider,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    with _GENERATION_LOCK:
        generations = load_generations()
        generations.insert(0, generation)
        save_generations(generations)
    if beta_store.enabled:
        try:
            beta_store.save_generation(user.id, generation, output_wav_bytes)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        _enqueue_generation_upload(
            generation_id,
            output_wav_bytes,
            json.dumps(generations, indent=2, ensure_ascii=False).encode("utf-8"),
        )

    return StreamingResponse(
        io.BytesIO(output_wav_bytes),
        media_type="audio/wav",
        headers={
            "Content-Disposition": "inline; filename=output.wav",
            "X-Generation-Id": generation_id,
            "X-Generation-Time-Ms": str(round((time.perf_counter() - started_at) * 1000)),
            "X-Translation-Provider": translation_provider,
            "X-TTS-Provider": tts_provider,
        },
    )


# ── MzansiLM endpoints ────────────────────────────────────────────────────────

@app.post("/v1/generations")
async def create_generation(
    body: AudioGenerationRequest,
    user: AuthenticatedUser = Depends(require_generation_slot),
):
    return await generate_audio(
        text=body.text,
        language=body.language,
        profile_id=body.profile_id,
        voice_mode=body.voice_mode,
        translate_text=body.translate_text,
        user=user,
    )


@app.get("/v1/generations")
def list_generations(
    profile_id: str | None = None,
    user: AuthenticatedUser = Depends(require_user),
):
    generations = _generations_for_user(user.id)
    if profile_id:
        generations = [item for item in generations if item.get("profile_id") == profile_id]
    return [_client_record(generation) for generation in generations]


@app.get("/v1/generations/{generation_id}")
def generation_audio(
    generation_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    if beta_store.enabled:
        try:
            audio_bytes = beta_store.generation_audio(user.id, generation_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        if audio_bytes is None:
            raise HTTPException(status_code=404, detail="Saved generation not found")
        return StreamingResponse(
            io.BytesIO(audio_bytes),
            media_type="audio/wav",
            headers={
                "Content-Disposition": f'inline; filename="{generation_id}.wav"',
            },
        )
    generation = next(
        (
            item
            for item in load_generations()
            if item.get("id") == generation_id and _belongs_to_user(item, user.id)
        ),
        None,
    )
    path = _generation_path(generation_id)
    if generation is None or not os.path.exists(path):
        raise HTTPException(status_code=404, detail="Saved generation not found")
    return FileResponse(path, media_type="audio/wav", filename=f"{generation_id}.wav")


@app.post("/v1/generations/{generation_id}/reports", status_code=201)
def report_generation(
    generation_id: str,
    body: ContentReportRequest,
    user: AuthenticatedUser = Depends(require_user),
):
    if beta_store.enabled:
        try:
            generation = beta_store.get_generation(user.id, generation_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        generation = next(
            (
                item
                for item in load_generations()
                if item.get("id") == generation_id and _belongs_to_user(item, user.id)
            ),
            None,
        )
    if generation is None:
        raise HTTPException(status_code=404, detail="Saved generation not found")

    report = {
        "id": str(uuid.uuid4()),
        "user_id": user.id,
        "generation_id": generation_id,
        "reason": body.reason,
        "details": body.details.strip(),
        "status": "received",
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    if beta_store.enabled:
        try:
            beta_store.save_content_report(user.id, report)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    else:
        with _GENERATION_LOCK:
            reports = load_content_reports()
            reports.append(report)
            save_content_reports(reports)
    return {"id": report["id"], "status": report["status"]}


@app.delete("/v1/generations/{generation_id}")
def delete_generation(
    generation_id: str,
    user: AuthenticatedUser = Depends(require_user),
):
    if beta_store.enabled:
        try:
            deleted = beta_store.delete_generation(user.id, generation_id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        if not deleted:
            raise HTTPException(status_code=404, detail="Saved generation not found")
        return {"deleted": generation_id}
    with _GENERATION_LOCK:
        generations = load_generations()
        if not any(
            item.get("id") == generation_id and _belongs_to_user(item, user.id)
            for item in generations
        ):
            raise HTTPException(status_code=404, detail="Saved generation not found")
        remaining = [
            item
            for item in generations
            if not (
                item.get("id") == generation_id
                and _belongs_to_user(item, user.id)
            )
        ]
        save_generations(remaining)
        path = _generation_path(generation_id)
        if os.path.exists(path):
            os.unlink(path)
    _enqueue_generation_delete(
        generation_id,
        json.dumps(remaining, indent=2, ensure_ascii=False).encode("utf-8"),
    )
    return {"deleted": generation_id}


@app.post("/v1/story-drafts")
async def mzansi_gen(
    body: GenerateRequest,
    user: AuthenticatedUser = Depends(require_user),
):
    """
    Generate a story continuation from a seed in the target SA language.

    If the seed is English and the target is an SA language, the seed is
    first translated into that language so the model continues in-language.
    """
    if not body.seed.strip():
        raise HTTPException(status_code=400, detail="'seed' must not be empty")
    if body.language not in SA_LANGS:
        raise HTTPException(
            status_code=400,
            detail=f"MzansiLM only generates SA languages: {sorted(SA_LANGS)}",
        )

    seed = body.seed.strip()

    # Translate English seed into the target SA language so the model generates
    # in the right language (MzansiLM is a base model, not multilingual chat).
    if body.language in SA_LANGS:
        try:
            seed = mzansi_translate(seed, body.language)
        except Exception:
            pass  # Fall back to English seed; model will try to continue it

    try:
        story = mzansi_generate(seed, max_new_tokens=min(body.max_tokens, 400))
        return {"text": story, "language": body.language}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Generation failed: {exc}")


@app.post("/v1/mzansi/translate")
async def mzansi_trans(
    body: TranslateRequest,
    user: AuthenticatedUser = Depends(require_user),
):
    """
    Translate English text into a South African language using MzansiLM.
    Only SA language codes are accepted (use Google Translate for hi/zh/ta).
    """
    if not body.text.strip():
        raise HTTPException(status_code=400, detail="'text' must not be empty")
    if body.language not in SA_LANGS:
        raise HTTPException(
            status_code=400,
            detail=f"MzansiLM only translates to SA languages: {sorted(SA_LANGS)}",
        )
    try:
        translated = mzansi_translate(body.text.strip(), body.language)
        return {"translated": translated, "language": body.language}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Translation failed: {exc}")


@app.delete("/v1/account", status_code=204)
def delete_account(
    user: AuthenticatedUser = Depends(require_user),
):
    if beta_store.enabled:
        try:
            beta_store.delete_account(user.id)
        except SupabaseStoreError as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
    with _PROFILE_LOCK:
        owned_profiles = [
            profile for profile in load_index() if _belongs_to_user(profile, user.id)
        ]
        save_index(
            [profile for profile in load_index() if not _belongs_to_user(profile, user.id)]
        )
        for profile in owned_profiles:
            _clean_profile_files(profile["id"])
            corpus_id = profile.get("corpus_id")
            if corpus_id:
                shutil.rmtree(_corpus_path(corpus_id), ignore_errors=True)
    with _GENERATION_LOCK:
        save_content_reports(
            [
                report
                for report in load_content_reports()
                if not _belongs_to_user(report, user.id)
            ]
        )
        owned_generations = [
            generation
            for generation in load_generations()
            if _belongs_to_user(generation, user.id)
        ]
        save_generations(
            [
                generation
                for generation in load_generations()
                if not _belongs_to_user(generation, user.id)
            ]
        )
        for generation in owned_generations:
            path = _generation_path(generation["id"])
            if os.path.exists(path):
                os.unlink(path)
    return Response(status_code=204)