Upload 6 files

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.10-slim
+FROM python:3.10-slim-bookworm
 
 # System tools
 RUN apt-get update && apt-get install -y \
@@ -7,16 +7,23 @@ RUN apt-get update && apt-get install -y \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
+ENV ENABLE_CLONE_ENGINES=1
+ENV PIP_DISABLE_PIP_VERSION_CHECK=1
 
-# Install Python deps (includes torch==2.1.2 — needed for Silero v4 PackageImporter,
-# removed in torch 2.3+. PyPI torch wheels are CPU-only by default.)
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+# Docker profile runs the CPU voice-clone backend.
+COPY requirements*.txt ./
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir --index-url https://download.pytorch.org/whl/cpu "torch==2.5.1+cpu" && \
+    if [ -f requirements-docker.txt ]; then \
+        pip install --no-cache-dir -r requirements-docker.txt; \
+    else \
+        pip install --no-cache-dir -r requirements.txt; \
+    fi
 
 # Copy app
 COPY app.py .
 
-# HuggingFace Spaces runs on port 7860
+# Hugging Face Spaces runs on port 7860
 EXPOSE 7860
 
 CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.docker.md

@@ -0,0 +1,39 @@
+---
+title: VoiceCraft CPU Clone Server
+emoji: 🎙️
+colorFrom: gray
+colorTo: blue
+sdk: docker
+app_port: 7860
+---
+
+# VoiceCraft CPU Clone Server
+
+Use this README content for CPU/Docker Spaces.
+
+Settings:
+
+- `VOICECRAFT_AUTH_MODE=license`
+- `LICENSE_VALIDATION_URL` - deployed Google Apps Script URL
+- `ENABLE_CLONE_ENGINES=1`
+- `HF_TOKEN` - Hugging Face read token from an account that accepted the Pocket TTS model terms.
+
+`API_SECRET` is legacy compatibility only and should not be distributed in desktop clients. Use `VOICECRAFT_AUTH_MODE=layered` only during migration, then rotate/remove the old shared secret.
+
+Voice clone access:
+
+1. Open `https://huggingface.co/kyutai/pocket-tts`.
+2. Accept the model terms/access form.
+3. Create a read token at `https://huggingface.co/settings/tokens`.
+4. Add it to this Space as `HF_TOKEN`.
+5. Restart or rebuild the Space.
+
+In the Google Sheet row for this Space:
+
+- `AppLink` = this Docker Space URL
+- `VoiceClone` = `active`, `enabled`, `on`, or `true`
+- Configure `DeviceID` and `MaxDevices` for device binding.
+
+This Space serves `/tts`, `/health`, `/status`, and `/all_voices`. `/tts` validates the license and registered device server-side; clone access also requires the row's `VoiceClone` entitlement. Voice cloning runs on CPU through Pocket TTS and does not use ZeroGPU quota.
+
+Voice Clone accepts long scripts from the desktop app. For best speed, keep each generation under 10,000 characters; longer scripts are batched automatically and joined into one WAV.

README.md

@@ -1,10 +1,69 @@
----
-title: '3'
-emoji: 📊
-colorFrom: blue
-colorTo: gray
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: VoiceCraft CPU Clone Server
+emoji: 🎙️
+colorFrom: gray
+colorTo: blue
+sdk: docker
+app_port: 7860
+startup_duration_timeout: 1h
+---
+
+# VoiceCraft CPU Clone Server
+
+Deploy the contents of this `HF` folder at the root of your Hugging Face Space.
+
+This profile is CPU-first:
+
+- Normal TTS: Edge, Piper, Anime presets
+- Voice clone: Pocket TTS CPU backend
+- GPU/ZeroGPU: not required
+
+## Required Space settings
+
+Set these in Space settings:
+
+- `VOICECRAFT_AUTH_MODE=license` - requires a live active license and registered device on every `/tts` request.
+- `LICENSE_VALIDATION_URL` - the deployed Google Apps Script URL. The current production URL is the code default, but setting it explicitly is recommended.
+- `ENABLE_CLONE_ENGINES=1` - keeps voice clone enabled.
+- `HF_TOKEN` - Hugging Face read token from an account that has accepted the Pocket TTS model terms.
+
+`API_SECRET` is legacy compatibility only. Do not put a shared API secret in distributed desktop apps. After every supported desktop build sends `X-License-Key` and `X-Device-ID`, remove the sheet `ApiSecret` value and rotate/remove the old Space `API_SECRET`.
+
+Optional modes:
+
+- `VOICECRAFT_AUTH_MODE=layered` requires both the live license/device and `API_SECRET`.
+- `VOICECRAFT_AUTH_MODE=api_secret` is legacy-only and is not recommended for public distribution.
+
+Rate limits default to 60 TTS requests/minute and 8 clone requests/minute per license/device. Override with `TTS_REQUESTS_PER_MINUTE` and `CLONE_REQUESTS_PER_MINUTE`.
+
+## Voice clone access
+
+Pocket TTS voice cloning uses gated weights. Before using custom reference audio:
+
+1. Open `https://huggingface.co/kyutai/pocket-tts` while logged in.
+2. Accept the model terms/access form.
+3. Create a read token at `https://huggingface.co/settings/tokens`.
+4. Add that token to this Space as the secret `HF_TOKEN`.
+5. Restart or rebuild the Space.
+
+## Google Sheet row
+
+For a clone-enabled Space:
+
+- `AppLink` = this Space URL
+- `VoiceClone` = `active`, `enabled`, `on`, or `true`
+- `DeviceID` and `MaxDevices` should be enabled for device binding.
+
+For a TTS-only Space:
+
+- `VoiceClone` = `inactive`, `disabled`, `off`, or blank
+
+## Notes
+
+The desktop app still sends the clone engine as `f5tts` for compatibility, but this backend runs Pocket TTS on CPU. The public voice list only shows `Voice Clone`, so the model name is not shown in the desktop GUI.
+
+First startup can be slower while model files download and load. Repeated use with the same reference audio is faster because the backend caches the reference voice state.
+
+Voice Clone accepts long scripts from the desktop app. For best speed, keep each generation under 10,000 characters; longer scripts are batched automatically and joined into one WAV.
+
+Authorization decisions are enforced by the backend. Hiding or unlocking a desktop UI control is never treated as authorization.

app.py

@@ -1,646 +1,1369 @@
-import os, io, asyncio, tempfile, threading, re, subprocess, shutil, logging, secrets
-from fastapi import FastAPI, Form, Request, HTTPException
-from fastapi.responses import StreamingResponse, JSONResponse
-from fastapi.middleware.gzip import GZipMiddleware
-
-app = FastAPI()
-app.add_middleware(GZipMiddleware, minimum_size=1000)
-
-# ══════════════════════════════════════════════════════════════════
-# SECURITY — Token check
-# Set API_SECRET environment variable in HuggingFace Space settings
-# ══════════════════════════════════════════════════════════════════
-API_SECRET = os.environ.get("API_SECRET", "")
-
-def verify_token(request: Request):
-    if not API_SECRET:
-        logging.warning("⚠️ API_SECRET not set — rejecting request")
-        raise HTTPException(status_code=503, detail="Server not configured")
-    token = request.headers.get("X-API-Token", "")
-    if not secrets.compare_digest(token, API_SECRET):
-        raise HTTPException(status_code=403, detail="Unauthorized")
-
-# ══════════════════════════════════════════════════════════════════
-# PIPER TTS SETUP — Auto download on first run
-# ══════════════════════════════════════════════════════════════════
-PIPER_DIR = "/tmp/piper"
-PIPER_BIN = os.path.join(PIPER_DIR, "piper")
-PIPER_MODELS_DIR = "/tmp/piper_models"
-PIPER_READY = False
-
-PIPER_VOICES = {
-    # English
-    "piper:en_US-amy-medium":        ("en_US-amy-medium.onnx",         "en_US-amy-medium.onnx.json"),
-    "piper:en_US-joe-medium":        ("en_US-joe-medium.onnx",         "en_US-joe-medium.onnx.json"),
-    "piper:en_US-lessac-medium":     ("en_US-lessac-medium.onnx",      "en_US-lessac-medium.onnx.json"),
-    "piper:en_US-ryan-high":         ("en_US-ryan-high.onnx",          "en_US-ryan-high.onnx.json"),
-    "piper:en_GB-alan-medium":       ("en_GB-alan-medium.onnx",        "en_GB-alan-medium.onnx.json"),
-    "piper:en_GB-alba-medium":       ("en_GB-alba-medium.onnx",        "en_GB-alba-medium.onnx.json"),
-    # Urdu / Hindi / Arabic
-    "piper:ur_PK-fasih-medium":      ("ur_PK-fasih-medium.onnx",       "ur_PK-fasih-medium.onnx.json"),
-    "piper:hi_IN-pratham-medium":     ("hi_IN-pratham-medium.onnx",     "hi_IN-pratham-medium.onnx.json"),
-    "piper:ar_JO-kareem-medium":     ("ar_JO-kareem-medium.onnx",      "ar_JO-kareem-medium.onnx.json"),
-    # Other languages
-    "piper:de_DE-thorsten-medium":   ("de_DE-thorsten-medium.onnx",    "de_DE-thorsten-medium.onnx.json"),
-    "piper:fr_FR-upmc-medium":       ("fr_FR-upmc-medium.onnx",        "fr_FR-upmc-medium.onnx.json"),
-    "piper:ru_RU-irina-medium":      ("ru_RU-irina-medium.onnx",       "ru_RU-irina-medium.onnx.json"),
-    "piper:tr_TR-dfki-medium":       ("tr_TR-dfki-medium.onnx",        "tr_TR-dfki-medium.onnx.json"),
-    "piper:pt_BR-faber-medium":      ("pt_BR-faber-medium.onnx",       "pt_BR-faber-medium.onnx.json"),
-    "piper:nl_NL-mls-medium":        ("nl_NL-mls-medium.onnx",         "nl_NL-mls-medium.onnx.json"),
-}
-
-PIPER_BASE_URL = "https://huggingface.co/rhasspy/piper-voices/resolve/main"
-
-def setup_piper():
-    global PIPER_READY
-    try:
-        import platform
-        os.makedirs(PIPER_DIR, exist_ok=True)
-        os.makedirs(PIPER_MODELS_DIR, exist_ok=True)
-
-        system = platform.system().lower()
-        arch   = platform.machine().lower()
-
-        if system == "linux" and "x86" in arch:
-            piper_url = "https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_x86_64.tar.gz"
-        elif system == "linux" and "aarch" in arch:
-            piper_url = "https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_aarch64.tar.gz"
-        else:
-            print(f"⚠️ Piper: unsupported platform {system}/{arch}, Piper disabled")
-            return
-
-        if not os.path.exists(PIPER_BIN):
-            print("📥 Piper binary indiriliyor...")
-            import urllib.request
-            tar_path = "/tmp/piper.tar.gz"
-            urllib.request.urlretrieve(piper_url, tar_path)
-            import tarfile
-            with tarfile.open(tar_path, "r:gz") as tf:
-                tf.extractall("/tmp/piper_extract")
-            extracted = "/tmp/piper_extract/piper"
-            if os.path.isdir(extracted):
-                for item in os.listdir(extracted):
-                    shutil.move(os.path.join(extracted, item), os.path.join(PIPER_DIR, item))
-            else:
-                shutil.move(extracted, PIPER_BIN)
-            os.chmod(PIPER_BIN, 0o755)
-            print("✅ Piper binary ready")
-
-        PIPER_READY = True
-        print("✅ Piper TTS ready")
-    except Exception as e:
-        print(f"⚠️ Piper setup failed (non-critical): {e}")
-        PIPER_READY = False
-
-threading.Thread(target=setup_piper, daemon=True).start()
-
-
-def download_piper_model(voice_code: str) -> tuple:
-    """Model yoksa indir, path tuple dondur (onnx, json)"""
-    if voice_code not in PIPER_VOICES:
-        raise ValueError(f"Unknown Piper voice: {voice_code}")
-    onnx_file, json_file = PIPER_VOICES[voice_code]
-    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
-    json_path  = os.path.join(PIPER_MODELS_DIR, json_file)
-
-    import urllib.request
-    # Build correct HF path: en/en_US/amy/medium/en_US-amy-medium.onnx
-    parts = onnx_file.rsplit("-", 2)
-    lang_code = parts[0]  # en_US
-    voice     = parts[1]  # amy
-    quality   = parts[2].replace(".onnx", "")  # medium
-    lang_short = lang_code.split("_")[0]  # en
-    hf_dir = f"{lang_short}/{lang_code}/{voice}/{quality}"
-
-    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
-        if not os.path.exists(fpath):
-            url = f"{PIPER_BASE_URL}/{hf_dir}/{fname}"
-            print(f"📥 Downloading Piper model: {fname}")
-            try:
-                urllib.request.urlretrieve(url, fpath)
-            except Exception:
-                url2 = f"{PIPER_BASE_URL}/{lang_short}/{lang_code}/{fname}"
-                urllib.request.urlretrieve(url2, fpath)
-    return onnx_path, json_path
-
-
-def download_piper_dynamic(voice_code: str) -> tuple:
-    """Dynamic Piper voice download — code format: piper:ar_JO-kareem-low"""
-    model_name = voice_code.replace("piper:", "")  # ar_JO-kareem-low
-    # Prevent path traversal
-    if ".." in model_name or "/" in model_name or "\\" in model_name:
-        raise ValueError(f"Invalid voice code: {voice_code}")
-    onnx_file = f"{model_name}.onnx"
-    json_file = f"{model_name}.onnx.json"
-    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
-    json_path = os.path.join(PIPER_MODELS_DIR, json_file)
-
-    if os.path.exists(onnx_path) and os.path.exists(json_path):
-        return onnx_path, json_path
-
-    import urllib.request
-    # Model format: lang_code-voice_name-quality (e.g., ar_JO-kareem-low)
-    # HF path: ar/ar_JO/kareem/low/ar_JO-kareem-low.onnx
-    dash_parts = model_name.rsplit("-", 2)  # Split from right: ["ar_JO", "kareem", "low"]
-    if len(dash_parts) >= 3:
-        lang_code = dash_parts[0]    # ar_JO
-        voice = dash_parts[1]        # kareem
-        quality = dash_parts[2]      # low
-        lang = lang_code.split("_")[0]  # ar
-        hf_path = f"{lang}/{lang_code}/{voice}/{quality}"
-    else:
-        hf_path = f"{model_name}/{model_name}"
-
-    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
-        if not os.path.exists(fpath):
-            url = f"{PIPER_BASE_URL}/{hf_path}/{fname}"
-            print(f"📥 Downloading dynamic Piper model: {fname}")
-            try:
-                urllib.request.urlretrieve(url, fpath)
-            except Exception as e:
-                print(f"⚠️ Dynamic Piper download failed: {e}")
-                raise
-    return onnx_path, json_path
-
-
-def _piper_synth_chunk(text: str, onnx_path: str, json_path: str, length_scale: float) -> bytes:
-    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as out_f:
-        out_path = out_f.name
-    try:
-        cmd = [
-            PIPER_BIN,
-            "--model", onnx_path,
-            "--config", json_path,
-            "--output_file", out_path,
-            "--length_scale", str(round(length_scale, 2)),
-        ]
-        result = subprocess.run(
-            cmd,
-            input=text.encode("utf-8"),
-            capture_output=True,
-            timeout=120,
-        )
-        if result.returncode != 0:
-            raise Exception(f"Piper error: {result.stderr.decode()[:200]}")
-        with open(out_path, "rb") as f:
-            return f.read()
-    finally:
-        if os.path.exists(out_path):
-            os.unlink(out_path)
-
-
-def _ffmpeg_concat_wav(parts: list) -> bytes:
-    """Merge multiple WAV byte chunks using ffmpeg concat (same codec)."""
-    if len(parts) == 1:
-        return parts[0]
-    tmp_files, concat_list, out_path = [], None, None
-    try:
-        for data in parts:
-            fd, path = tempfile.mkstemp(suffix=".wav")
-            os.close(fd)
-            with open(path, "wb") as f:
-                f.write(data)
-            tmp_files.append(path)
-        fd, concat_list = tempfile.mkstemp(suffix=".txt")
-        os.close(fd)
-        with open(concat_list, "w") as f:
-            for p in tmp_files:
-                f.write(f"file '{p}'\n")
-        fd, out_path = tempfile.mkstemp(suffix=".wav")
-        os.close(fd)
-        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
-                        "-f", "concat", "-safe", "0", "-i", concat_list,
-                        "-c", "copy", out_path], check=True, timeout=60)
-        with open(out_path, "rb") as f:
-            return f.read()
-    except Exception:
-        return b"".join(parts)
-    finally:
-        for p in tmp_files:
-            try: os.unlink(p)
-            except Exception: pass
-        for x in (concat_list, out_path):
-            if x:
-                try: os.unlink(x)
-                except Exception: pass
-
-
-def synthesize_piper(text: str, voice_code: str, speed: float = 1.0) -> bytes:
-    if not PIPER_READY:
-        raise Exception("Piper is not available on this system")
-    # Pehle PIPER_VOICES dict mein check karo, nahi mila to dynamic download
-    if voice_code in PIPER_VOICES:
-        onnx_path, json_path = download_piper_model(voice_code)
-    else:
-        # Dynamic voice — direct HuggingFace se download
-        onnx_path, json_path = download_piper_dynamic(voice_code)
-    length_scale = 1.0 / max(0.25, min(4.0, speed))
-    # Lambi text ko chunk karo (Piper stdin limit + timeout avoid karne ke liye)
-    if len(text) > 1400:
-        chunks = split_text(text, max_chars=1400)
-    else:
-        chunks = [text]
-    if len(chunks) == 1:
-        return _piper_synth_chunk(chunks[0], onnx_path, json_path, length_scale)
-    parts = []
-    for ch in chunks:
-        if ch.strip():
-            parts.append(_piper_synth_chunk(ch, onnx_path, json_path, length_scale))
-    if not parts:
-        raise Exception("Piper: no audio generated")
-    return _ffmpeg_concat_wav(parts)
-
-
-def split_text(text: str, max_chars: int = 1400) -> list:
-    """Text ko chunklara bol"""
-    text = text.strip()
-    if not text:
-        return []
-    sentence_re = re.compile(
-        r'(?:(?<=[.!?\u0964\u06D4\u061F\u2026])\s+)|(?<=[\u3002\uff01\uff1f])'
-    )
-    chunks, current = [], ""
-    for para in re.split(r'\n+', text):
-        para = para.strip()
-        if not para:
-            if current:
-                chunks.append(current)
-                current = ""
-            continue
-        for sentence in sentence_re.split(para):
-            sentence = sentence.strip()
-            if not sentence:
-                continue
-            if len(sentence) > max_chars:
-                words, buf = sentence.split(), ""
-                for word in words:
-                    add = (" " if buf else "") + word
-                    if len(buf) + len(add) <= max_chars:
-                        buf += add
-                    else:
-                        if buf:
-                            chunks.append(buf)
-                        buf = word
-                if buf:
-                    chunks.append(buf)
-            elif len(current) + len(sentence) + 1 <= max_chars:
-                current = (current + " " + sentence).strip()
-            else:
-                if current:
-                    chunks.append(current)
-                current = sentence
-        if current:
-            chunks.append(current)
-            current = ""
-    if current:
-        chunks.append(current)
-    return [c for c in chunks if c.strip()]
-
-
-# ══════════════════════════════════════════════════════════════════
-# SILERO TTS — v4 model (48kHz, 5 Russian speakers)
-# Loaded via torch.package.PackageImporter (requires torch < 2.3)
-# ══════════════════════════════════════════════════════════════════
-SILERO_READY = False
-SILERO_MODELS_DIR = "/tmp/silero_models"
-SILERO_SAMPLE_RATE = 48000
-SILERO_MODELS = {}
-SILERO_LOAD_LOCK = threading.Lock()
-
-SILERO_MODEL_URLS = [
-    "https://models.silero.ai/models/tts/ru/v4_ru.pt",
-    "https://huggingface.co/Derur/silero-models/resolve/main/tts/ru/ru_v4/v4_ru.pt",
-]
-
-SILERO_SPEAKERS_RU = [
-    "aidar", "baya", "kseniya", "xenia", "eugene",
-]
-
-
-def download_silero_model() -> str:
-    """Download Silero v4 Russian model. Returns path on success."""
-    import urllib.request
-    os.makedirs(SILERO_MODELS_DIR, exist_ok=True)
-    model_path = os.path.join(SILERO_MODELS_DIR, "v4_ru.pt")
-    if os.path.exists(model_path) and os.path.getsize(model_path) > 100000:
-        return model_path
-    for url in SILERO_MODEL_URLS:
-        try:
-            print(f"📥 Downloading Silero v4: {url[:80]}...")
-            urllib.request.urlretrieve(url, model_path)
-            if os.path.getsize(model_path) > 100000:
-                print(f"✅ Silero v4 downloaded ({os.path.getsize(model_path)//1024}KB)")
-                return model_path
-            os.remove(model_path)
-        except Exception as e:
-            print(f"⚠️ Download failed: {e}")
-            try:
-                os.remove(model_path)
-            except Exception:
-                pass
-    return ""
-
-
-def setup_silero():
-    global SILERO_READY
-    try:
-        import torch
-        model_path = download_silero_model()
-        if model_path:
-            model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
-            SILERO_MODELS["ru"] = model
-            SILERO_READY = True
-            print("✅ Silero TTS ready (v4 Russian — 5 speakers)")
-        else:
-            print("❌ Silero TTS: model download failed")
-    except Exception as e:
-        print(f"❌ Silero setup failed: {e}")
-
-threading.Thread(target=setup_silero, daemon=True).start()
-
-
-def synthesize_silero(text: str, voice_code: str) -> bytes:
-    """Silero TTS — code: silero:ru_xenia. v4 model via torch.package.
-    Model lazily load hota hai (self-heal) agar startup thread fail hua ho."""
-    import numpy as np, scipy.io.wavfile as wav
-    try:
-        import torch
-    except ImportError:
-        raise Exception("Silero TTS requires torch. Install: pip install torch==2.1.2")
-
-    lang_speaker = voice_code.replace("silero:", "")
-    lang = lang_speaker.split("_")[0]
-    speaker = lang_speaker.split("_", 1)[1] if "_" in lang_speaker else lang_speaker
-
-    # Validate speaker — only real v4 speakers allowed
-    valid_speakers = set(SILERO_SPEAKERS_RU)
-    if speaker not in valid_speakers:
-        raise Exception(f"Invalid Silero speaker '{speaker}'. Valid: {', '.join(valid_speakers)}")
-
-    # Model on-demand load (self-heals if startup thread failed / was slow)
-    if lang not in SILERO_MODELS:
-        with SILERO_LOAD_LOCK:
-            if lang not in SILERO_MODELS:
-                model_path = download_silero_model()
-                if not model_path:
-                    raise Exception("Silero v4 model could not be downloaded (check network / model URL).")
-                model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
-                SILERO_MODELS[lang] = model
-                global SILERO_READY
-                SILERO_READY = True
-
-    model = SILERO_MODELS[lang]
-    audio = model.apply_tts(text=text, speaker=speaker, sample_rate=SILERO_SAMPLE_RATE)
-    audio_np = audio.numpy() if hasattr(audio, "numpy") else audio.cpu().detach().numpy()
-
-    buf = io.BytesIO()
-    wav.write(buf, SILERO_SAMPLE_RATE, (audio_np * 32767).astype(np.int16))
-    buf.seek(0)
-    return buf.read()
-
-
-def _ffmpeg_concat_mp3(parts: list) -> bytes:
-    """Merge multiple MP3 byte chunks using ffmpeg concat."""
-    if len(parts) == 1:
-        return parts[0]
-    tmp_files = []
-    concat_list = None
-    out_path = None
-    try:
-        for data in parts:
-            fd, path = tempfile.mkstemp(suffix=".mp3")
-            os.close(fd)
-            with open(path, "wb") as f:
-                f.write(data)
-            tmp_files.append(path)
-        fd, concat_list = tempfile.mkstemp(suffix=".txt")
-        os.close(fd)
-        with open(concat_list, "w") as f:
-            for p in tmp_files:
-                f.write(f"file '{p}'\n")
-        fd, out_path = tempfile.mkstemp(suffix=".mp3")
-        os.close(fd)
-        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
-                        "-f", "concat", "-safe", "0", "-i", concat_list,
-                        "-c", "copy", out_path], check=True, timeout=60)
-        with open(out_path, "rb") as f:
-            return f.read()
-    except Exception:
-        return b"".join(parts)
-    finally:
-        for p in tmp_files:
-            try: os.unlink(p)
-            except Exception: pass
-        if concat_list:
-            try: os.unlink(concat_list)
-            except Exception: pass
-        if out_path:
-            try: os.unlink(out_path)
-            except Exception: pass
-
-
-async def _edge_synth_chunk(chunk: str, voice: str, kwargs: dict) -> bytes:
-    """One chunk ki audio lao, 3 baar retry karo. Fail par b"" return."""
-    import edge_tts
-    for attempt in range(3):
-        data = bytearray()
-        try:
-            comm = edge_tts.Communicate(chunk, voice, **kwargs)
-            async for packet in comm.stream():
-                if packet["type"] == "audio" and packet.get("data"):
-                    data.extend(packet["data"])
-            if data:
-                return bytes(data)
-        except Exception:
-            if attempt == 2:
-                return b""
-            await asyncio.sleep(1 + attempt)
-    return b""
-
-
-async def synthesize_edge(
-    text: str,
-    voice: str,
-    rate: str = "+0%",
-    volume: str = "+0%",
-    pitch: str = "+0Hz",
-    style: str = None,
-    styledegree: str = None,
-) -> bytes:
-    import edge_tts
-    chunks = split_text(text)
-    audio_parts = []
-    kwargs = {"rate": rate, "volume": volume, "pitch": pitch}
-    if style and style != "Default" and style != "General":
-        kwargs["style"] = style
-    if styledegree is not None:
-        try:
-            sd = float(styledegree)
-            if 0.0 <= sd <= 2.0:
-                kwargs["styledegree"] = styledegree
-        except Exception:
-            pass
-    # Style drop karne wala safe set (non-EN voices kuch styles reject karti hain)
-    safe_kwargs = {k: v for k, v in kwargs.items() if k not in ("style", "styledegree")}
-    for chunk in chunks:
-        audio = await _edge_synth_chunk(chunk, voice, kwargs)
-        # Agar style ki wajah se fail hua ho, style hata kar dobara try karo
-        if not audio and kwargs.get("style"):
-            audio = await _edge_synth_chunk(chunk, voice, safe_kwargs)
-        audio_parts.append(audio if audio else b"")
-    if len(audio_parts) == 1:
-        return audio_parts[0]
-    return _ffmpeg_concat_mp3(audio_parts)
-
-
-@app.get("/")
-def root():
-    return {"status": "VoiceCraft TTS Server OK", "engines": ["edge", "piper", "silero"]}
-
-
-@app.get("/health")
-@app.head("/health")
-def health():
-    return {
-        "status": "ok",
-        "piper_ready": PIPER_READY,
-        "silero_ready": SILERO_READY,
-        "silero_speakers": SILERO_SPEAKERS_RU,
-        "engines": ["edge", "piper", "silero"],
-    }
-
-
-@app.get("/all_voices")
-async def all_voices_list():
-    """All voices — Edge (400+) + Piper (900+) + Silero (5 RU)"""
-    result = {"edge": {}, "piper": {}, "silero": {}}
-
-    # Edge TTS — 400+ voices (complete list, clean naming)
-    try:
-        import edge_tts
-        voices = await edge_tts.list_voices()
-        for v in voices:
-            short = v.get("ShortName", "")
-            friendly = v.get("FriendlyName", "") or short
-            name = friendly
-            for remove in ["Microsoft Server Speech Text to Speech Voice", "Microsoft", "Online", "(Natural)", "(Neural)", "(Standard)", "(Multilingual)", "(Expressive)"]:
-                name = name.replace(remove, "")
-            if "," in name:
-                name = name.split(",")[-1]
-            # Clean dash pattern: " - " or "  -  " → single " - "
-            name = re.sub(r'\s*-\s*', ' - ', name)
-            # Collapse all whitespace to single space
-            name = re.sub(r'\s+', ' ', name)
-            name = name.strip(" -").strip()
-            region = short.split("-")[0] + "-" + short.split("-")[1] if "-" in short else ""
-            result["edge"][f"{name} [{region}]"] = short
-    except Exception as e:
-        print(f"Edge voices error: {e}")
-
-    # Piper TTS — all voices (clean naming, no engine hints)
-    try:
-        import urllib.request, json as _json
-        api_url = "https://huggingface.co/api/models/rhasspy/piper-voices"
-        req = urllib.request.Request(api_url, headers={"User-Agent": "VoiceCraft/2.0"})
-        with urllib.request.urlopen(req, timeout=60) as resp:
-            data = _json.loads(resp.read())
-        siblings = data.get("siblings", [])
-        for s in siblings:
-            rfn = s.get("rfilename", s.get("rfn", ""))
-            if not rfn.endswith(".onnx") or ".json" in rfn or "samples" in rfn:
-                continue
-            parts = rfn.split("/")
-            if len(parts) < 2:
-                continue
-            model_name = parts[-1].replace(".onnx", "")
-            dash_parts = model_name.rsplit("-", 2)
-            if len(dash_parts) >= 3:
-                lang_code = dash_parts[0].replace("_", "-")
-                voice = dash_parts[1].replace("_", " ").title()
-                quality = dash_parts[2]
-                if quality in ("low", "x_low"):
-                    continue
-                quality_map = {"high": " +", "medium": "", "low": " -", "x_low": " --"}
-                qs = quality_map.get(quality, " -")
-                display = f"{voice} [{lang_code}]{qs}"
-            else:
-                display = model_name.replace("_", " ").title()
-            full_code = f"piper:{model_name}"
-            result["piper"][display] = full_code
-    except Exception as e:
-        print(f"Piper dynamic fetch error: {e}")
-        for key in PIPER_VOICES:
-            result["piper"][key] = key  # voice code as string, not tuple
-
-    # Silero v4 — Russian (official v4_ru speakers)
-    silero_ru_speakers = {
-        "aidar": "Aidar", "baya": "Baya", "kseniya": "Kseniya",
-        "xenia": "Xenia", "eugene": "Eugene",
-    }
-    for speaker_code, display_name in silero_ru_speakers.items():
-        result["silero"][f"{display_name} \u2022 Russian [RU]"] = f"silero:ru_{speaker_code}"
-
-    total = len(result["edge"]) + len(result["piper"]) + len(result.get("silero", {}))
-    return {"voices": result, "total": total}
-
-
-@app.post("/tts")
-async def tts_endpoint(
-    request: Request,
-    engine:  str = Form(...),   # "edge" | "piper" | "silero"
-    text:    str = Form(...),
-    voice:   str = Form("en-US-AvaNeural"),   # edge voice code OR piper/silero code
-    rate:    str = Form("+0%"),               # edge only
-    volume:  str = Form("+0%"),               # edge only
-    pitch:   str = Form("+0Hz"),              # edge only
-    speed:   float = Form(1.0),              # piper only
-    style:   str = Form(None),               # edge style (emotion)
-    styledegree: str = Form(None),           # edge style degree 0-2
-):
-    verify_token(request)
-
-    if not text or not text.strip():
-        return JSONResponse(status_code=400, content={"error": "Text is empty"})
-
-    text = text.strip()
-
-    # Abuse / timeout guard — lambi text Chapter Mode se bhejo
-    if len(text) > 60000:
-        return JSONResponse(
-            status_code=413,
-            content={"error": "Text too long (max 60000 chars). Use Chapter Mode for longer text."},
-        )
-
-    try:
-        if engine == "edge":
-            audio = await synthesize_edge(text, voice, rate=rate, volume=volume, pitch=pitch, style=style, styledegree=styledegree)
-            media = "audio/mpeg"
-            fname = "tts_edge.mp3"
-
-        elif engine == "piper":
-            # Sync + subprocess/torch → run off the event loop so the server
-            # stays responsive and doesn't time out under load.
-            audio = await asyncio.to_thread(synthesize_piper, text, voice, speed)
-            media = "audio/wav"
-            fname = "tts_piper.wav"
-
-        elif engine == "silero":
-            audio = await asyncio.to_thread(synthesize_silero, text, voice)
-            media = "audio/wav"
-            fname = "tts_silero.wav"
-
-        else:
-            return JSONResponse(status_code=400, content={"error": f"Unknown engine: {engine}"})
-
-        return StreamingResponse(
-            io.BytesIO(audio),
-            media_type=media,
-            headers={"Content-Disposition": f"attachment; filename={fname}"},
-        )
-
-    except Exception as e:
-        logging.error(f"TTS error: {e}", exc_info=True)
-        return JSONResponse(
-            status_code=500,
-            content={"error": f"Synthesis failed: {str(e)[:240]}"},
-        )
+import os, io, asyncio, tempfile, threading, re, subprocess, shutil, logging, secrets, sys, platform, math, hashlib, time
+
+try:
+    import spaces
+except ImportError:
+    class _SpacesFallback:
+        """Keep the shared backend importable outside Hugging Face Spaces."""
+
+        @staticmethod
+        def GPU(function=None, **_kwargs):
+            def decorator(fn):
+                return fn
+
+            return decorator(function) if callable(function) else decorator
+
+    spaces = _SpacesFallback()
+
+import gradio as gr
+import requests
+
+# Hide console window on Windows
+CREATE_NO_WINDOW = 0x08000000 if sys.platform == "win32" else 0
+from fastapi import Form, Request, HTTPException, UploadFile, File
+from fastapi.responses import StreamingResponse, JSONResponse
+from fastapi.middleware.gzip import GZipMiddleware
+
+app = gr.Server(
+    title="VoiceCraft TTS Server",
+    description="VoiceCraft desktop API with CPU voice cloning support.",
+    version="2.0",
+)
+app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+
+# ══════════════════════════════════════════════════════════════════
+# SECURITY — server-side authorization
+# Production mode validates each license/device with Google Apps Script.
+# ══════════════════════════════════════════════════════════════════
+_RAW_API_SECRET = os.environ.get("API_SECRET", "").strip()
+if _RAW_API_SECRET.startswith("hf_") and not (
+    os.environ.get("HF_TOKEN")
+    or os.environ.get("HUGGINGFACE_HUB_TOKEN")
+    or os.environ.get("HUGGING_FACE_HUB_TOKEN")
+):
+    os.environ["HF_TOKEN"] = _RAW_API_SECRET
+
+API_SECRET = (
+    os.environ.get("VOICECRAFT_API_SECRET", "").strip()
+    or os.environ.get("APP_API_SECRET", "").strip()
+    or ("" if _RAW_API_SECRET.startswith("hf_") else _RAW_API_SECRET)
+)
+DEFAULT_LICENSE_VALIDATION_URL = (
+    "https://script.google.com/macros/s/"
+    "AKfycbx6KWT18JHUevX9zXhzsOg40_Ek7wNDhpTIesQ63Lm6aPNCoxhujQL8z_Ll9Dt6-cQ/exec"
+)
+LICENSE_VALIDATION_URL = (
+    os.environ.get("LICENSE_VALIDATION_URL", "").strip()
+    or DEFAULT_LICENSE_VALIDATION_URL
+)
+AUTH_MODE = os.environ.get("VOICECRAFT_AUTH_MODE", "license").strip().lower()
+if AUTH_MODE not in {"license", "layered", "api_secret"}:
+    AUTH_MODE = "license"
+MIN_CLIENT_VERSION = os.environ.get("MIN_CLIENT_VERSION", "3.0.0").strip()
+MAX_CLONE_CHARACTERS = 60_000
+CLONE_CHUNK_CHARACTERS = 1800
+CLONE_ENGINE_PREFIXES = ("f5tts:",)
+CLONE_ENGINES = ["f5tts"]
+CLONE_BACKEND_NAME = "pocket-tts-cpu"
+BASE_ENGINES = ["edge", "piper", "silero", "anime"]
+
+
+def _env_flag(name: str, default=None):
+    raw = os.environ.get(name)
+    if raw is None or raw == "":
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on", "active", "enabled"}
+
+
+def clone_engines_enabled() -> bool:
+    explicit = _env_flag("ENABLE_CLONE_ENGINES", None)
+    if explicit is not None:
+        return explicit
+    # CPU clone is now the default backend. Set ENABLE_CLONE_ENGINES=0 for TTS-only Spaces.
+    return True
+
+
+def available_engines():
+    return BASE_ENGINES + (CLONE_ENGINES if clone_engines_enabled() else [])
+
+
+def verify_token(request: Request):
+    if not API_SECRET:
+        logging.warning("⚠️ API_SECRET not set — rejecting request")
+        raise HTTPException(status_code=503, detail="API_SECRET missing in Hugging Face Space secrets")
+    token = request.headers.get("X-API-Token", "").strip()
+    if not secrets.compare_digest(token, API_SECRET):
+        raise HTTPException(status_code=403, detail="Unauthorized")
+
+
+_LICENSE_CACHE = {}
+_LICENSE_CACHE_LOCK = threading.Lock()
+_LICENSE_RATE_STATE = {}
+_LICENSE_RATE_LOCK = threading.Lock()
+
+
+def _bounded_env_int(name: str, default: int, minimum: int, maximum: int) -> int:
+    try:
+        return max(minimum, min(maximum, int(os.environ.get(name, default))))
+    except (TypeError, ValueError):
+        return default
+
+
+def _license_cache_key(license_key: str, device_id: str) -> str:
+    raw = f"{license_key}\0{device_id}".encode("utf-8")
+    return hashlib.sha256(raw).hexdigest()
+
+
+def _license_flag(value) -> bool:
+    return str(value or "").strip().lower() in {
+        "active", "enabled", "enable", "on", "true", "yes", "1"
+    }
+
+
+def _version_tuple(value: str) -> tuple[int, ...]:
+    parts = re.findall(r"\d+", str(value or ""))
+    numbers = [int(part) for part in parts[:4]]
+    return tuple((numbers + [0, 0, 0, 0])[:4])
+
+
+def _validate_license_sync(license_key: str, device_id: str) -> dict:
+    cache_key = _license_cache_key(license_key, device_id)
+    now = time.monotonic()
+    with _LICENSE_CACHE_LOCK:
+        cached = _LICENSE_CACHE.get(cache_key)
+        if cached and cached["expires_at"] > now:
+            return cached["payload"]
+    try:
+        response = requests.get(
+            LICENSE_VALIDATION_URL,
+            params={
+                "action": "check",
+                "license_key": license_key,
+                "pc_user": "VoiceCraftDesktop",
+                "device_id": device_id,
+            },
+            headers={"User-Agent": "VoiceCraft-Server/3.0"},
+            timeout=(6, 20),
+        )
+        response.raise_for_status()
+        payload = response.json()
+    except Exception as exc:
+        logging.warning("License validation service unavailable: %s", exc.__class__.__name__)
+        raise HTTPException(status_code=503, detail="License validation service unavailable")
+    status = str(payload.get("status", "") or "").strip().lower()
+    if status not in {"active", "trial"}:
+        message = str(payload.get("message", "") or "").strip()
+        if status == "device_limit":
+            detail = message or "Maximum device limit reached for this license"
+        elif status:
+            detail = message or f"License is {status}"
+        else:
+            detail = "License is not active"
+        raise HTTPException(status_code=403, detail=detail)
+    ttl = _bounded_env_int("LICENSE_CACHE_TTL_SECONDS", 45, 5, 300)
+    with _LICENSE_CACHE_LOCK:
+        _LICENSE_CACHE[cache_key] = {
+            "payload": payload,
+            "expires_at": now + ttl,
+        }
+        if len(_LICENSE_CACHE) > 2048:
+            expired = [
+                key for key, value in _LICENSE_CACHE.items()
+                if value["expires_at"] <= now
+            ]
+            for key in expired:
+                _LICENSE_CACHE.pop(key, None)
+    return payload
+
+
+def _enforce_rate_limit(license_key: str, device_id: str, is_clone: bool):
+    identity = _license_cache_key(license_key, device_id)
+    now = time.monotonic()
+    window = 60.0
+    limit = _bounded_env_int(
+        "CLONE_REQUESTS_PER_MINUTE" if is_clone else "TTS_REQUESTS_PER_MINUTE",
+        8 if is_clone else 60,
+        1,
+        600,
+    )
+    state_key = f"{identity}:{'clone' if is_clone else 'tts'}"
+    with _LICENSE_RATE_LOCK:
+        requests_in_window = [
+            timestamp
+            for timestamp in _LICENSE_RATE_STATE.get(state_key, [])
+            if now - timestamp < window
+        ]
+        if len(requests_in_window) >= limit:
+            raise HTTPException(status_code=429, detail="Request limit reached; try again shortly")
+        requests_in_window.append(now)
+        _LICENSE_RATE_STATE[state_key] = requests_in_window
+
+
+async def authorize_request(request: Request, engine: str) -> dict:
+    if AUTH_MODE in {"api_secret", "layered"}:
+        verify_token(request)
+    if AUTH_MODE == "api_secret":
+        return {}
+    license_key = request.headers.get("X-License-Key", "").strip()
+    device_id = request.headers.get("X-Device-ID", "").strip()
+    client_version = request.headers.get("X-Client-Version", "").strip()
+    if _version_tuple(client_version) < _version_tuple(MIN_CLIENT_VERSION):
+        raise HTTPException(status_code=426, detail="VoiceCraft update required")
+    if not license_key or not device_id:
+        raise HTTPException(status_code=401, detail="License credentials are required")
+    if len(license_key) > 256 or len(device_id) > 128:
+        raise HTTPException(status_code=400, detail="Invalid license credentials")
+    payload = await asyncio.to_thread(_validate_license_sync, license_key, device_id)
+    is_clone = engine in CLONE_ENGINES
+    if is_clone and not _license_flag(payload.get("voiceClone")):
+        raise HTTPException(status_code=403, detail="Voice cloning is not enabled for this license")
+    _enforce_rate_limit(license_key, device_id, is_clone)
+    return payload
+
+# ══════════════════════════════════════════════════════════════════
+# PIPER TTS SETUP — Auto download on first run
+# ══════════════════════════════════════════════════════════════════
+PIPER_DIR = "/tmp/piper"
+PIPER_BIN = os.path.join(PIPER_DIR, "piper")
+PIPER_MODELS_DIR = "/tmp/piper_models"
+PIPER_READY = False
+
+PIPER_VOICES = {
+    # English
+    "piper:en_US-amy-medium":        ("en_US-amy-medium.onnx",         "en_US-amy-medium.onnx.json"),
+    "piper:en_US-joe-medium":        ("en_US-joe-medium.onnx",         "en_US-joe-medium.onnx.json"),
+    "piper:en_US-lessac-medium":     ("en_US-lessac-medium.onnx",      "en_US-lessac-medium.onnx.json"),
+    "piper:en_US-ryan-high":         ("en_US-ryan-high.onnx",          "en_US-ryan-high.onnx.json"),
+    "piper:en_GB-alan-medium":       ("en_GB-alan-medium.onnx",        "en_GB-alan-medium.onnx.json"),
+    "piper:en_GB-alba-medium":       ("en_GB-alba-medium.onnx",        "en_GB-alba-medium.onnx.json"),
+    # Urdu / Hindi / Arabic
+    "piper:ur_PK-fasih-medium":      ("ur_PK-fasih-medium.onnx",       "ur_PK-fasih-medium.onnx.json"),
+    "piper:hi_IN-pratham-medium":     ("hi_IN-pratham-medium.onnx",     "hi_IN-pratham-medium.onnx.json"),
+    "piper:ar_JO-kareem-medium":     ("ar_JO-kareem-medium.onnx",      "ar_JO-kareem-medium.onnx.json"),
+    # Other languages
+    "piper:de_DE-thorsten-medium":   ("de_DE-thorsten-medium.onnx",    "de_DE-thorsten-medium.onnx.json"),
+    "piper:fr_FR-upmc-medium":       ("fr_FR-upmc-medium.onnx",        "fr_FR-upmc-medium.onnx.json"),
+    "piper:ru_RU-irina-medium":      ("ru_RU-irina-medium.onnx",       "ru_RU-irina-medium.onnx.json"),
+    "piper:tr_TR-dfki-medium":       ("tr_TR-dfki-medium.onnx",        "tr_TR-dfki-medium.onnx.json"),
+    "piper:pt_BR-faber-medium":      ("pt_BR-faber-medium.onnx",       "pt_BR-faber-medium.onnx.json"),
+    "piper:nl_NL-mls-medium":        ("nl_NL-mls-medium.onnx",         "nl_NL-mls-medium.onnx.json"),
+}
+
+PIPER_BASE_URL = "https://huggingface.co/rhasspy/piper-voices/resolve/main"
+
+def setup_piper():
+    global PIPER_READY
+    try:
+        import platform
+        os.makedirs(PIPER_DIR, exist_ok=True)
+        os.makedirs(PIPER_MODELS_DIR, exist_ok=True)
+
+        system = platform.system().lower()
+        arch   = platform.machine().lower()
+
+        if system == "linux" and "x86" in arch:
+            piper_url = "https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_x86_64.tar.gz"
+        elif system == "linux" and "aarch" in arch:
+            piper_url = "https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_aarch64.tar.gz"
+        else:
+            print(f"⚠️ Piper: unsupported platform {system}/{arch}, Piper disabled")
+            return
+
+        if not os.path.exists(PIPER_BIN):
+            print("📥 Piper binary indiriliyor...")
+            import urllib.request
+            tar_path = "/tmp/piper.tar.gz"
+            urllib.request.urlretrieve(piper_url, tar_path)
+            import tarfile
+            with tarfile.open(tar_path, "r:gz") as tf:
+                tf.extractall("/tmp/piper_extract")
+            extracted = "/tmp/piper_extract/piper"
+            if os.path.isdir(extracted):
+                for item in os.listdir(extracted):
+                    shutil.move(os.path.join(extracted, item), os.path.join(PIPER_DIR, item))
+            else:
+                shutil.move(extracted, PIPER_BIN)
+            os.chmod(PIPER_BIN, 0o755)
+            print("✅ Piper binary ready")
+
+        PIPER_READY = True
+        print("✅ Piper TTS ready")
+    except Exception as e:
+        print(f"⚠️ Piper setup failed (non-critical): {e}")
+        PIPER_READY = False
+
+threading.Thread(target=setup_piper, daemon=True).start()
+
+
+def download_piper_model(voice_code: str) -> tuple:
+    """Model yoksa indir, path tuple dondur (onnx, json)"""
+    if voice_code not in PIPER_VOICES:
+        raise ValueError(f"Unknown Piper voice: {voice_code}")
+    onnx_file, json_file = PIPER_VOICES[voice_code]
+    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
+    json_path  = os.path.join(PIPER_MODELS_DIR, json_file)
+
+    import urllib.request
+    # Build correct HF path: en/en_US/amy/medium/en_US-amy-medium.onnx
+    parts = onnx_file.rsplit("-", 2)
+    lang_code = parts[0]  # en_US
+    voice     = parts[1]  # amy
+    quality   = parts[2].replace(".onnx", "")  # medium
+    lang_short = lang_code.split("_")[0]  # en
+    hf_dir = f"{lang_short}/{lang_code}/{voice}/{quality}"
+
+    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
+        if not os.path.exists(fpath):
+            url = f"{PIPER_BASE_URL}/{hf_dir}/{fname}"
+            print(f"📥 Downloading Piper model: {fname}")
+            try:
+                urllib.request.urlretrieve(url, fpath)
+            except Exception:
+                url2 = f"{PIPER_BASE_URL}/{lang_short}/{lang_code}/{fname}"
+                urllib.request.urlretrieve(url2, fpath)
+    return onnx_path, json_path
+
+
+def download_piper_dynamic(voice_code: str) -> tuple:
+    """Dynamic Piper voice download — code format: piper:ar_JO-kareem-low"""
+    model_name = voice_code.replace("piper:", "")  # ar_JO-kareem-low
+    # Prevent path traversal
+    if ".." in model_name or "/" in model_name or "\\" in model_name:
+        raise ValueError(f"Invalid voice code: {voice_code}")
+    onnx_file = f"{model_name}.onnx"
+    json_file = f"{model_name}.onnx.json"
+    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
+    json_path = os.path.join(PIPER_MODELS_DIR, json_file)
+
+    if os.path.exists(onnx_path) and os.path.exists(json_path):
+        return onnx_path, json_path
+
+    import urllib.request
+    # Model format: lang_code-voice_name-quality (e.g., ar_JO-kareem-low)
+    # HF path: ar/ar_JO/kareem/low/ar_JO-kareem-low.onnx
+    dash_parts = model_name.rsplit("-", 2)  # Split from right: ["ar_JO", "kareem", "low"]
+    if len(dash_parts) >= 3:
+        lang_code = dash_parts[0]    # ar_JO
+        voice = dash_parts[1]        # kareem
+        quality = dash_parts[2]      # low
+        lang = lang_code.split("_")[0]  # ar
+        hf_path = f"{lang}/{lang_code}/{voice}/{quality}"
+    else:
+        hf_path = f"{model_name}/{model_name}"
+
+    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
+        if not os.path.exists(fpath):
+            url = f"{PIPER_BASE_URL}/{hf_path}/{fname}"
+            print(f"📥 Downloading dynamic Piper model: {fname}")
+            try:
+                urllib.request.urlretrieve(url, fpath)
+            except Exception as e:
+                print(f"⚠️ Dynamic Piper download failed: {e}")
+                raise
+    return onnx_path, json_path
+
+
+def _piper_synth_chunk(text: str, onnx_path: str, json_path: str, length_scale: float) -> bytes:
+    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as out_f:
+        out_path = out_f.name
+    try:
+        cmd = [
+            PIPER_BIN,
+            "--model", onnx_path,
+            "--config", json_path,
+            "--output_file", out_path,
+            "--length_scale", str(round(length_scale, 2)),
+        ]
+        result = subprocess.run(
+            cmd,
+            input=text.encode("utf-8"),
+            capture_output=True,
+            timeout=120,
+            creationflags=CREATE_NO_WINDOW,
+        )
+        if result.returncode != 0:
+            raise Exception(f"Piper error: {result.stderr.decode()[:200]}")
+        with open(out_path, "rb") as f:
+            return f.read()
+    finally:
+        if os.path.exists(out_path):
+            os.unlink(out_path)
+
+
+def _ffmpeg_concat_wav(parts: list) -> bytes:
+    """Merge multiple WAV byte chunks using ffmpeg concat (same codec)."""
+    if len(parts) == 1:
+        return parts[0]
+    tmp_files, concat_list, out_path = [], None, None
+    try:
+        for data in parts:
+            fd, path = tempfile.mkstemp(suffix=".wav")
+            os.close(fd)
+            with open(path, "wb") as f:
+                f.write(data)
+            tmp_files.append(path)
+        fd, concat_list = tempfile.mkstemp(suffix=".txt")
+        os.close(fd)
+        with open(concat_list, "w") as f:
+            for p in tmp_files:
+                f.write(f"file '{p}'\n")
+        fd, out_path = tempfile.mkstemp(suffix=".wav")
+        os.close(fd)
+        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
+                        "-f", "concat", "-safe", "0", "-i", concat_list,
+                        "-c", "copy", out_path], check=True, timeout=60, creationflags=CREATE_NO_WINDOW)
+        with open(out_path, "rb") as f:
+            return f.read()
+    except Exception:
+        return b"".join(parts)
+    finally:
+        for p in tmp_files:
+            try: os.unlink(p)
+            except Exception: pass
+        for x in (concat_list, out_path):
+            if x:
+                try: os.unlink(x)
+                except Exception: pass
+
+
+def synthesize_piper(text: str, voice_code: str, speed: float = 1.0) -> bytes:
+    if not PIPER_READY:
+        raise Exception("Piper is not available on this system")
+    # Pehle PIPER_VOICES dict mein check karo, nahi mila to dynamic download
+    if voice_code in PIPER_VOICES:
+        onnx_path, json_path = download_piper_model(voice_code)
+    else:
+        # Dynamic voice — direct HuggingFace se download
+        onnx_path, json_path = download_piper_dynamic(voice_code)
+    length_scale = 1.0 / max(0.25, min(4.0, speed))
+    # Lambi text ko chunk karo (Piper stdin limit + timeout avoid karne ke liye)
+    if len(text) > 1400:
+        chunks = split_text(text, max_chars=1400)
+    else:
+        chunks = [text]
+    if len(chunks) == 1:
+        return _piper_synth_chunk(chunks[0], onnx_path, json_path, length_scale)
+    parts = []
+    for ch in chunks:
+        if ch.strip():
+            parts.append(_piper_synth_chunk(ch, onnx_path, json_path, length_scale))
+    if not parts:
+        raise Exception("Piper: no audio generated")
+    return _ffmpeg_concat_wav(parts)
+
+
+def split_text(text: str, max_chars: int = 1400) -> list:
+    """Text ko chunklara bol"""
+    text = text.strip()
+    if not text:
+        return []
+    sentence_re = re.compile(
+        r'(?:(?<=[.!?\u0964\u06D4\u061F\u2026])\s+)|(?<=[\u3002\uff01\uff1f])'
+    )
+    chunks, current = [], ""
+    for para in re.split(r'\n+', text):
+        para = para.strip()
+        if not para:
+            if current:
+                chunks.append(current)
+                current = ""
+            continue
+        for sentence in sentence_re.split(para):
+            sentence = sentence.strip()
+            if not sentence:
+                continue
+            if len(sentence) > max_chars:
+                words, buf = sentence.split(), ""
+                for word in words:
+                    add = (" " if buf else "") + word
+                    if len(buf) + len(add) <= max_chars:
+                        buf += add
+                    else:
+                        if buf:
+                            chunks.append(buf)
+                        buf = word
+                if buf:
+                    chunks.append(buf)
+            elif len(current) + len(sentence) + 1 <= max_chars:
+                current = (current + " " + sentence).strip()
+            else:
+                if current:
+                    chunks.append(current)
+                current = sentence
+        if current:
+            chunks.append(current)
+            current = ""
+    if current:
+        chunks.append(current)
+    return [c for c in chunks if c.strip()]
+
+
+# ══════════════════════════════════════════════════════════════════
+# SILERO TTS — v4 model (48kHz, 5 Russian speakers)
+# Loaded via torch.package.PackageImporter (requires torch < 2.3)
+# ══════════════════════════════════════════════════════════════════
+SILERO_READY = False
+SILERO_MODELS_DIR = "/tmp/silero_models"
+SILERO_SAMPLE_RATE = 48000
+SILERO_MODELS = {}
+SILERO_LOAD_LOCK = threading.Lock()
+
+SILERO_MODEL_URLS = [
+    "https://models.silero.ai/models/tts/ru/v4_ru.pt",
+    "https://huggingface.co/Derur/silero-models/resolve/main/tts/ru/ru_v4/v4_ru.pt",
+]
+
+SILERO_SPEAKERS_RU = [
+    "aidar", "baya", "kseniya", "xenia", "eugene",
+]
+
+
+def download_silero_model() -> str:
+    """Download Silero v4 Russian model. Returns path on success."""
+    import urllib.request
+    os.makedirs(SILERO_MODELS_DIR, exist_ok=True)
+    model_path = os.path.join(SILERO_MODELS_DIR, "v4_ru.pt")
+    if os.path.exists(model_path) and os.path.getsize(model_path) > 100000:
+        return model_path
+    for url in SILERO_MODEL_URLS:
+        try:
+            print(f"📥 Downloading Silero v4: {url[:80]}...")
+            urllib.request.urlretrieve(url, model_path)
+            if os.path.getsize(model_path) > 100000:
+                print(f"✅ Silero v4 downloaded ({os.path.getsize(model_path)//1024}KB)")
+                return model_path
+            os.remove(model_path)
+        except Exception as e:
+            print(f"⚠️ Download failed: {e}")
+            try:
+                os.remove(model_path)
+            except Exception:
+                pass
+    return ""
+
+
+def setup_silero():
+    global SILERO_READY
+    try:
+        import torch
+        model_path = download_silero_model()
+        if model_path:
+            model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
+            SILERO_MODELS["ru"] = model
+            SILERO_READY = True
+            print("✅ Silero TTS ready (v4 Russian — 5 speakers)")
+        else:
+            print("❌ Silero TTS: model download failed")
+    except Exception as e:
+        print(f"❌ Silero setup failed: {e}")
+
+threading.Thread(target=setup_silero, daemon=True).start()
+
+
+def synthesize_silero(text: str, voice_code: str) -> bytes:
+    """Silero TTS — code: silero:ru_xenia. v4 model via torch.package.
+    Model lazily load hota hai (self-heal) agar startup thread fail hua ho."""
+    import numpy as np, scipy.io.wavfile as wav
+    try:
+        import torch
+    except ImportError:
+        raise Exception("Silero TTS requires torch.")
+
+    lang_speaker = voice_code.replace("silero:", "")
+    lang = lang_speaker.split("_")[0]
+    speaker = lang_speaker.split("_", 1)[1] if "_" in lang_speaker else lang_speaker
+
+    # Validate speaker — only real v4 speakers allowed
+    valid_speakers = set(SILERO_SPEAKERS_RU)
+    if speaker not in valid_speakers:
+        raise Exception(f"Invalid Silero speaker '{speaker}'. Valid: {', '.join(valid_speakers)}")
+
+    # Model on-demand load (self-heals if startup thread failed / was slow)
+    if lang not in SILERO_MODELS:
+        with SILERO_LOAD_LOCK:
+            if lang not in SILERO_MODELS:
+                model_path = download_silero_model()
+                if not model_path:
+                    raise Exception("Silero v4 model could not be downloaded (check network / model URL).")
+                model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
+                SILERO_MODELS[lang] = model
+                global SILERO_READY
+                SILERO_READY = True
+
+    model = SILERO_MODELS[lang]
+    audio = model.apply_tts(text=text, speaker=speaker, sample_rate=SILERO_SAMPLE_RATE)
+    audio_np = audio.numpy() if hasattr(audio, "numpy") else audio.cpu().detach().numpy()
+
+    buf = io.BytesIO()
+    wav.write(buf, SILERO_SAMPLE_RATE, (audio_np * 32767).astype(np.int16))
+    buf.seek(0)
+    return buf.read()
+
+
+def _ffmpeg_concat_mp3(parts: list) -> bytes:
+    """Merge multiple MP3 byte chunks using ffmpeg concat."""
+    if len(parts) == 1:
+        return parts[0]
+    tmp_files = []
+    concat_list = None
+    out_path = None
+    try:
+        for data in parts:
+            fd, path = tempfile.mkstemp(suffix=".mp3")
+            os.close(fd)
+            with open(path, "wb") as f:
+                f.write(data)
+            tmp_files.append(path)
+        fd, concat_list = tempfile.mkstemp(suffix=".txt")
+        os.close(fd)
+        with open(concat_list, "w") as f:
+            for p in tmp_files:
+                f.write(f"file '{p}'\n")
+        fd, out_path = tempfile.mkstemp(suffix=".mp3")
+        os.close(fd)
+        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
+                        "-f", "concat", "-safe", "0", "-i", concat_list,
+                        "-c", "copy", out_path], check=True, timeout=60, creationflags=CREATE_NO_WINDOW)
+        with open(out_path, "rb") as f:
+            return f.read()
+    except Exception:
+        return b"".join(parts)
+    finally:
+        for p in tmp_files:
+            try: os.unlink(p)
+            except Exception: pass
+        if concat_list:
+            try: os.unlink(concat_list)
+            except Exception: pass
+        if out_path:
+            try: os.unlink(out_path)
+            except Exception: pass
+
+
+async def _edge_synth_chunk(chunk: str, voice: str, kwargs: dict) -> bytes:
+    """One chunk ki audio lao, 3 baar retry karo. Fail par b"" return."""
+    import edge_tts
+    for attempt in range(3):
+        data = bytearray()
+        try:
+            comm = edge_tts.Communicate(chunk, voice, **kwargs)
+            async for packet in comm.stream():
+                if packet["type"] == "audio" and packet.get("data"):
+                    data.extend(packet["data"])
+            if data:
+                return bytes(data)
+        except Exception:
+            if attempt == 2:
+                return b""
+            await asyncio.sleep(1 + attempt)
+    return b""
+
+
+async def synthesize_edge(
+    text: str,
+    voice: str,
+    rate: str = "+0%",
+    volume: str = "+0%",
+    pitch: str = "+0Hz",
+    style: str = None,
+    styledegree: str = None,
+) -> bytes:
+    import edge_tts
+    chunks = split_text(text)
+    audio_parts = []
+    kwargs = {"rate": rate, "volume": volume, "pitch": pitch}
+    if style and style != "Default" and style != "General":
+        kwargs["style"] = style
+    if styledegree is not None:
+        try:
+            sd = float(styledegree)
+            if 0.0 <= sd <= 2.0:
+                kwargs["styledegree"] = styledegree
+        except Exception:
+            pass
+    # Style drop karne wala safe set (non-EN voices kuch styles reject karti hain)
+    safe_kwargs = {k: v for k, v in kwargs.items() if k not in ("style", "styledegree")}
+    for chunk in chunks:
+        audio = await _edge_synth_chunk(chunk, voice, kwargs)
+        # Agar style ki wajah se fail hua ho, style hata kar dobara try karo
+        if not audio and kwargs.get("style"):
+            audio = await _edge_synth_chunk(chunk, voice, safe_kwargs)
+        audio_parts.append(audio if audio else b"")
+    if len(audio_parts) == 1:
+        return audio_parts[0]
+    return _ffmpeg_concat_mp3(audio_parts)
+# ═════════════════════════════════════════════════════════════════════════
+# VOICE CLONING ENGINES - FREE MODELS (Task 4/5)
+# ═════════════════════════════════════════════════════════════════════════
+
+CLONE_MODELS = {
+    "f5tts:v1_base": ("f5tts", "multilingual", "Voice Clone"),
+}
+
+_POCKET_MODEL = None
+_POCKET_MODEL_ERROR = None
+_POCKET_MODEL_LOCK = threading.Lock()
+_POCKET_INFER_LOCK = threading.Lock()
+_POCKET_RESULT_CACHE = {}
+_POCKET_RESULT_CACHE_ORDER = []
+_POCKET_RESULT_CACHE_LOCK = threading.Lock()
+_POCKET_RESULT_CACHE_LIMIT = 8
+_POCKET_RESULT_CACHE_MAX_BYTES = 16 * 1024 * 1024
+_POCKET_VOICE_STATE_CACHE = {}
+_POCKET_VOICE_STATE_ORDER = []
+_POCKET_VOICE_STATE_LOCK = threading.Lock()
+_POCKET_VOICE_STATE_LIMIT = 6
+
+
+def _hf_token_configured() -> bool:
+    return bool(
+        os.environ.get("HF_TOKEN")
+        or os.environ.get("HUGGINGFACE_HUB_TOKEN")
+        or os.environ.get("HUGGING_FACE_HUB_TOKEN")
+    )
+
+
+def _pocket_clone_auth_error() -> str:
+    return (
+        "Voice cloning weights are gated. Accept the Hugging Face model terms, "
+        "add a read token as HF_TOKEN in Space secrets, then restart/rebuild the Space."
+    )
+
+
+def _pocket_clone_ready() -> bool:
+    return bool(_POCKET_MODEL is not None and getattr(_POCKET_MODEL, "has_voice_cloning", False))
+
+
+def _load_pocket_model():
+    global _POCKET_MODEL, _POCKET_MODEL_ERROR
+    if _POCKET_MODEL is not None:
+        return _POCKET_MODEL
+    with _POCKET_MODEL_LOCK:
+        if _POCKET_MODEL is not None:
+            return _POCKET_MODEL
+        try:
+            from pocket_tts import TTSModel
+            _POCKET_MODEL = TTSModel.load_model()
+            _POCKET_MODEL_ERROR = None
+            if getattr(_POCKET_MODEL, "has_voice_cloning", False):
+                print("Pocket TTS CPU voice clone ready")
+            else:
+                print("Pocket TTS loaded without voice cloning weights. HF_TOKEN/model access required.")
+        except Exception as exc:
+            _POCKET_MODEL_ERROR = str(exc)
+            logging.exception("Pocket TTS startup failed")
+            raise RuntimeError(_POCKET_MODEL_ERROR or "Voice clone model is not ready")
+    return _POCKET_MODEL
+
+
+def _warm_pocket_model():
+    if clone_engines_enabled():
+        try:
+            _load_pocket_model()
+        except Exception:
+            pass
+
+
+threading.Thread(target=_warm_pocket_model, daemon=True).start()
+
+
+def _reference_audio_key(reference_path: str) -> str:
+    digest = hashlib.sha256()
+    with open(reference_path, "rb") as reference_file:
+        for chunk in iter(lambda: reference_file.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _normalize_clone_reference(reference_path: str) -> str:
+    """Decode any accepted upload into a mono PCM WAV Pocket TTS can read."""
+    if not reference_path or not os.path.exists(reference_path):
+        raise ValueError("A reference audio file is required for voice cloning")
+    ffmpeg = shutil.which("ffmpeg")
+    if not ffmpeg:
+        raise RuntimeError("ffmpeg is required to decode voice clone reference audio")
+    fd, normalized_path = tempfile.mkstemp(suffix=".wav")
+    os.close(fd)
+    try:
+        result = subprocess.run(
+            [
+                ffmpeg,
+                "-y",
+                "-hide_banner",
+                "-loglevel",
+                "error",
+                "-i",
+                reference_path,
+                "-map",
+                "0:a:0",
+                "-t",
+                "12",
+                "-vn",
+                "-sn",
+                "-dn",
+                "-acodec",
+                "pcm_s16le",
+                "-ac",
+                "1",
+                "-ar",
+                "24000",
+                "-f",
+                "wav",
+                normalized_path,
+            ],
+            capture_output=True,
+            timeout=60,
+            creationflags=CREATE_NO_WINDOW,
+        )
+        if result.returncode != 0 or os.path.getsize(normalized_path) <= 44:
+            detail = result.stderr.decode("utf-8", errors="replace").strip()
+            raise ValueError(
+                "Reference audio could not be decoded. "
+                f"Use a clear WAV, MP3, M4A, OGG, or FLAC file. {detail[:160]}"
+            )
+        return normalized_path
+    except Exception:
+        try:
+            os.unlink(normalized_path)
+        except OSError:
+            pass
+        raise
+
+
+def _clone_request_key(text: str, reference_key: str) -> str:
+    payload = f"{reference_key}\0{text or ''}".encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _get_cached_clone(cache_key: str):
+    with _POCKET_RESULT_CACHE_LOCK:
+        return _POCKET_RESULT_CACHE.get(cache_key)
+
+
+def _cache_clone(cache_key: str, audio: bytes):
+    if len(audio) > _POCKET_RESULT_CACHE_MAX_BYTES:
+        return
+    with _POCKET_RESULT_CACHE_LOCK:
+        if cache_key in _POCKET_RESULT_CACHE:
+            _POCKET_RESULT_CACHE_ORDER.remove(cache_key)
+        _POCKET_RESULT_CACHE[cache_key] = audio
+        _POCKET_RESULT_CACHE_ORDER.append(cache_key)
+        while len(_POCKET_RESULT_CACHE_ORDER) > _POCKET_RESULT_CACHE_LIMIT:
+            oldest = _POCKET_RESULT_CACHE_ORDER.pop(0)
+            _POCKET_RESULT_CACHE.pop(oldest, None)
+
+
+def _get_cached_voice_state(reference_key: str):
+    with _POCKET_VOICE_STATE_LOCK:
+        return _POCKET_VOICE_STATE_CACHE.get(reference_key)
+
+
+def _cache_voice_state(reference_key: str, voice_state):
+    with _POCKET_VOICE_STATE_LOCK:
+        if reference_key in _POCKET_VOICE_STATE_CACHE:
+            _POCKET_VOICE_STATE_ORDER.remove(reference_key)
+        _POCKET_VOICE_STATE_CACHE[reference_key] = voice_state
+        _POCKET_VOICE_STATE_ORDER.append(reference_key)
+        while len(_POCKET_VOICE_STATE_ORDER) > _POCKET_VOICE_STATE_LIMIT:
+            oldest = _POCKET_VOICE_STATE_ORDER.pop(0)
+            _POCKET_VOICE_STATE_CACHE.pop(oldest, None)
+
+
+@app.api(name="voicecraft_clone_backend", api_visibility="private", concurrency_limit=1)
+def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) -> bytes:
+    if not reference_path or not os.path.exists(reference_path):
+        raise ValueError("A reference audio file is required for voice cloning")
+    model = _load_pocket_model()
+    if not getattr(model, "has_voice_cloning", False):
+        raise RuntimeError(_pocket_clone_auth_error())
+    import numpy as np
+
+    voice_state = _get_cached_voice_state(reference_key)
+    audio_parts = []
+    with _POCKET_INFER_LOCK:
+        if voice_state is None:
+            try:
+                voice_state = model.get_state_for_audio_prompt(reference_path)
+            except Exception as exc:
+                message = str(exc)
+                if "could not download the weights" in message.lower() or "voice cloning" in message.lower():
+                    raise RuntimeError(_pocket_clone_auth_error()) from exc
+                raise
+            _cache_voice_state(reference_key, voice_state)
+        chunks = split_text(text, max_chars=CLONE_CHUNK_CHARACTERS)
+        for chunk in chunks:
+            audio = model.generate_audio(voice_state, chunk)
+            if hasattr(audio, "detach"):
+                audio = audio.detach().cpu().numpy()
+            audio_np = np.asarray(audio).squeeze().reshape(-1)
+            if audio_np.size:
+                audio_parts.append(audio_np)
+    if not audio_parts:
+        raise RuntimeError("Voice clone model did not produce audio")
+    if len(audio_parts) == 1:
+        audio_np = audio_parts[0]
+    else:
+        silence = np.zeros(int(model.sample_rate * 0.12), dtype=audio_parts[0].dtype)
+        joined = []
+        for index, part in enumerate(audio_parts):
+            if index:
+                joined.append(silence)
+            joined.append(part)
+        audio_np = np.concatenate(joined)
+    import scipy.io.wavfile as wav
+    if audio_np.size == 0:
+        raise RuntimeError("Voice clone model did not produce audio")
+    buf = io.BytesIO()
+    wav.write(buf, int(model.sample_rate), audio_np)
+    return buf.getvalue()
+
+
+async def synthesize_f5tts(text: str, reference_path: str = None) -> bytes:
+    if not reference_path or not os.path.exists(reference_path):
+        raise ValueError("A reference audio file is required for voice cloning")
+    reference_key = await asyncio.to_thread(_reference_audio_key, reference_path)
+    cache_key = _clone_request_key(text, reference_key)
+    cached = _get_cached_clone(cache_key)
+    if cached is not None:
+        return cached
+    if _get_cached_voice_state(reference_key) is not None:
+        # Same reference voice is already encoded. Skip ffmpeg normalization for repeated
+        # desktop batches/generations; audio quality is unchanged because voice_state is reused.
+        audio = await asyncio.to_thread(
+            _run_pocket_clone_cpu, text, reference_path, reference_key
+        )
+    else:
+        normalized_path = await asyncio.to_thread(_normalize_clone_reference, reference_path)
+        try:
+            audio = await asyncio.to_thread(
+                _run_pocket_clone_cpu, text, normalized_path, reference_key
+            )
+        finally:
+            try:
+                os.unlink(normalized_path)
+            except OSError:
+                pass
+    _cache_clone(cache_key, audio)
+    return audio
+
+async def synthesize_anime(text: str, voice_code: str, style: str = None) -> bytes:
+    """Anime-style TTS using reliable Edge voices until a custom anime backend is configured."""
+    voice_map = {
+        "anime:en_whisper": ("en-US-AvaMultilingualNeural", "whispering", "+8%", "+4%", "+4Hz"),
+        "anime:ja_edge": ("ja-JP-NanamiNeural", "cheerful", "+10%", "+5%", "+5Hz"),
+        "anime:zh_piper": ("zh-CN-XiaoxiaoNeural", "cheerful", "+10%", "+5%", "+5Hz"),
+    }
+    edge_voice, fallback_style, rate, volume, pitch = voice_map.get(
+        voice_code,
+        ("en-US-AvaMultilingualNeural", "cheerful", "+10%", "+5%", "+5Hz"),
+    )
+    return await synthesize_edge(
+        text,
+        edge_voice,
+        rate=rate,
+        volume=volume,
+        pitch=pitch,
+        style=style or fallback_style,
+    )
+
+@app.post("/tts")
+async def tts_endpoint(
+    request: Request,
+    engine:  str = Form(...),   # "edge" | "piper" | "silero" | "anime" | "f5tts"
+    text:    str = Form(...),
+    voice:   str = Form("en-US-AvaNeural"),   # edge voice code OR piper/silero code
+    rate:    str = Form("+0%"),               # edge only
+    volume:  str = Form("+0%"),               # edge only
+    pitch:   str = Form("+0Hz"),              # edge only
+    speed:   float = Form(1.0),              # piper only
+    style:   str = Form(None),               # edge style (emotion)
+    styledegree: str = Form(None),           # edge style degree 0-2
+    voice_cloning: bool = Form(False),       # voice cloning toggle
+    reference_audio: UploadFile = File(None),
+):
+    if not text or not text.strip():
+        return JSONResponse(status_code=400, content={"error": "Text is empty"})
+
+    text = text.strip()
+
+    # Abuse / timeout guard — lambi text Chapter Mode se bhejo
+    if len(text) > 60000:
+        return JSONResponse(
+            status_code=413,
+            content={"error": "Text too long (max 60000 chars). Use Chapter Mode for longer text."},
+        )
+
+    engine = engine.strip().lower()
+    await authorize_request(request, engine)
+
+    if engine in CLONE_ENGINES and not clone_engines_enabled():
+        return JSONResponse(
+            status_code=403,
+            content={"error": "Clone engines are disabled on this Space"},
+        )
+    if engine in CLONE_ENGINES and len(text) > MAX_CLONE_CHARACTERS:
+        return JSONResponse(
+            status_code=413,
+            content={
+                "error": (
+                    "Clone text is too long "
+                    f"(max {MAX_CLONE_CHARACTERS} characters per request)"
+                )
+            },
+        )
+
+    reference_path = None
+    try:
+        if reference_audio is not None and reference_audio.filename:
+            suffix = os.path.splitext(reference_audio.filename)[1].lower() or ".wav"
+            if suffix not in (".wav", ".mp3", ".m4a", ".ogg", ".flac"):
+                return JSONResponse(status_code=400, content={"error": "Unsupported reference audio format"})
+            fd_ref, reference_path = tempfile.mkstemp(suffix=suffix)
+            os.close(fd_ref)
+            data = await reference_audio.read(25 * 1024 * 1024 + 1)
+            if not data:
+                return JSONResponse(status_code=400, content={"error": "Reference audio is empty"})
+            if len(data) > 25 * 1024 * 1024:
+                return JSONResponse(status_code=413, content={"error": "Reference audio is too large (max 25 MB)"})
+            with open(reference_path, "wb") as f:
+                f.write(data)
+            voice_cloning = True
+
+        if engine == "edge":
+            audio = await synthesize_edge(text, voice, rate=rate, volume=volume, pitch=pitch, style=style, styledegree=styledegree)
+            media = "audio/mpeg"
+            fname = "tts_edge.mp3"
+
+        elif engine == "piper":
+            # Sync + subprocess/torch → run off the event loop so the server
+            # stays responsive and doesn't time out under load.
+            audio = await asyncio.to_thread(synthesize_piper, text, voice, speed)
+            media = "audio/wav"
+            fname = "tts_piper.wav"
+
+        elif engine == "silero":
+            audio = await asyncio.to_thread(synthesize_silero, text, voice)
+            media = "audio/wav"
+            fname = "tts_silero.wav"
+
+        elif engine == "f5tts":
+            audio = await synthesize_f5tts(text, reference_path=reference_path)
+            media = "audio/wav"
+            fname = "tts_f5_clone.wav"
+
+        elif engine == "anime":
+            audio = await synthesize_anime(text, voice, style)
+            media = "audio/mpeg"
+            fname = "tts_anime.mp3"
+
+        else:
+            return JSONResponse(status_code=400, content={"error": f"Unknown engine: {engine}"})
+
+        return StreamingResponse(
+            io.BytesIO(audio),
+            media_type=media,
+            headers={"Content-Disposition": f"attachment; filename={fname}"},
+        )
+
+    except Exception as e:
+        logging.error(f"TTS error: {e}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={"error": f"Synthesis failed: {str(e)[:240]}"},
+        )
+    finally:
+        try:
+            if reference_path and os.path.exists(reference_path):
+                os.unlink(reference_path)
+        except Exception:
+            pass
+
+
+@app.get("/status")
+def status():
+    return {
+        "status": "VoiceCraft TTS Server OK",
+        "engines": available_engines(),
+        "clone_enabled": clone_engines_enabled(),
+    }
+
+
+@app.get("/health")
+@app.head("/health")
+def health():
+    return {
+        "status": "ok",
+        "piper_ready": PIPER_READY,
+        "silero_ready": SILERO_READY,
+        "silero_speakers": SILERO_SPEAKERS_RU,
+        "engines": available_engines(),
+        "clone_enabled": clone_engines_enabled(),
+        "zerogpu_enabled": False,
+        "clone_backend": CLONE_BACKEND_NAME,
+        "auth_mode": AUTH_MODE,
+        "api_auth_required": AUTH_MODE in {"api_secret", "layered"},
+        "license_auth_required": AUTH_MODE in {"license", "layered"},
+        "clone_model_ready": _pocket_clone_ready(),
+        "clone_auth_configured": _hf_token_configured(),
+        "clone_setup_required": clone_engines_enabled() and not _pocket_clone_ready(),
+        "clone_reference_cache_entries": len(_POCKET_VOICE_STATE_CACHE),
+    }
+
+
+@app.get("/all_voices")
+async def all_voices_list():
+    """All voices - Edge, Piper, Silero, Anime styles, and CPU voice cloning."""
+    result = {"edge": {}, "piper": {}, "silero": {}}
+
+    # Edge TTS — 400+ voices (complete list, clean naming)
+    try:
+        import edge_tts
+        voices = await edge_tts.list_voices()
+        for v in voices:
+            short = v.get("ShortName", "")
+            friendly = v.get("FriendlyName", "") or short
+            name = friendly
+            for remove in ["Microsoft Server Speech Text to Speech Voice", "Microsoft", "Online", "(Natural)", "(Neural)", "(Standard)", "(Multilingual)", "(Expressive)"]:
+                name = name.replace(remove, "")
+            if "," in name:
+                name = name.split(",")[-1]
+            # Clean dash pattern: " - " or "  -  " → single " - "
+            name = re.sub(r'\s*-\s*', ' - ', name)
+            # Collapse all whitespace to single space
+            name = re.sub(r'\s+', ' ', name)
+            name = name.strip(" -").strip()
+            region = short.split("-")[0] + "-" + short.split("-")[1] if "-" in short else ""
+            result["edge"][f"{name} [{region}]"] = short
+    except Exception as e:
+        print(f"Edge voices error: {e}")
+
+    # Piper TTS — all voices (clean naming, no engine hints)
+    try:
+        import urllib.request, json as _json
+        api_url = "https://huggingface.co/api/models/rhasspy/piper-voices"
+        req = urllib.request.Request(api_url, headers={"User-Agent": "VoiceCraft/2.0"})
+        with urllib.request.urlopen(req, timeout=60) as resp:
+            data = _json.loads(resp.read())
+        siblings = data.get("siblings", [])
+        for s in siblings:
+            rfn = s.get("rfilename", s.get("rfn", ""))
+            if not rfn.endswith(".onnx") or ".json" in rfn or "samples" in rfn:
+                continue
+            parts = rfn.split("/")
+            if len(parts) < 2:
+                continue
+            model_name = parts[-1].replace(".onnx", "")
+            dash_parts = model_name.rsplit("-", 2)
+            if len(dash_parts) >= 3:
+                lang_code = dash_parts[0].replace("_", "-")
+                voice = dash_parts[1].replace("_", " ").title()
+                quality = dash_parts[2]
+                if quality in ("low", "x_low"):
+                    continue
+                quality_map = {"high": " +", "medium": "", "low": " -", "x_low": " --"}
+                qs = quality_map.get(quality, " -")
+                display = f"{voice} [{lang_code}]{qs}"
+            else:
+                display = model_name.replace("_", " ").title()
+            full_code = f"piper:{model_name}"
+            result["piper"][display] = full_code
+    except Exception as e:
+        print(f"Piper dynamic fetch error: {e}")
+        for key in PIPER_VOICES:
+            result["piper"][key] = key  # voice code as string, not tuple
+
+    # Silero v4 — Russian (official v4_ru speakers)
+    silero_ru_speakers = {
+        "aidar": "Aidar", "baya": "Baya", "kseniya": "Kseniya",
+        "xenia": "Xenia", "eugene": "Eugene",
+    }
+    for speaker_code, display_name in silero_ru_speakers.items():
+        result["silero"][f"{display_name} \u2022 Russian [RU]"] = f"silero:ru_{speaker_code}"
+
+    # Anime styles are lightweight Edge-based presets and remain CPU-safe.
+    anime_voices = {
+        "Anime Whisper [en]": "anime:en_whisper",
+        "Anime Edge [ja]": "anime:ja_edge",
+        "Anime Piper [zh]": "anime:zh_piper",
+    }
+    result["anime"] = anime_voices
+
+    if not clone_engines_enabled():
+        total = sum(len(group) for group in result.values())
+        return {"voices": result, "total": total, "clone_enabled": False}
+
+    result["f5tts"] = {"Voice Clone": "f5tts:v1_base"}
+    total = sum(len(group) for group in result.values())
+    return {"voices": result, "total": total, "clone_enabled": True}
+
+
+GRADIO_DEFAULT_VOICES = {
+    "edge": "en-US-AvaNeural",
+    "piper": "piper:en_US-amy-medium",
+    "silero": "silero:ru_xenia",
+    "anime": "anime:en_whisper",
+    "f5tts": "f5tts:v1_base",
+}
+
+
+def _write_audio_file(audio: bytes, suffix: str) -> str:
+    fd, path = tempfile.mkstemp(suffix=suffix)
+    os.close(fd)
+    with open(path, "wb") as f:
+        f.write(audio)
+    return path
+
+
+def _build_gradio_ui(gr):
+    ui_engines = available_engines()
+    css = """
+    .gradio-container {
+        max-width: 1160px !important;
+        margin: auto !important;
+        font-family: Inter, ui-sans-serif, system-ui, sans-serif;
+    }
+    .voicecraft-hero {
+        border: 1px solid rgba(148, 163, 184, 0.24);
+        border-radius: 18px;
+        padding: 22px 24px;
+        background: linear-gradient(135deg, rgba(15, 23, 42, 0.96), rgba(17, 24, 39, 0.92));
+        color: white;
+        box-shadow: 0 22px 60px rgba(15, 23, 42, 0.18);
+    }
+    .voicecraft-hero h1 {
+        margin: 0 0 8px;
+        font-size: 30px;
+        line-height: 1.1;
+        letter-spacing: 0;
+    }
+    .voicecraft-hero p {
+        margin: 0;
+        color: rgba(226, 232, 240, 0.88);
+    }
+    """
+
+    async def generate_audio(engine, text, voice, rate, volume, pitch, speed, style, styledegree, reference_audio, token):
+        if not API_SECRET or not secrets.compare_digest((token or "").strip(), API_SECRET):
+            raise gr.Error("Invalid API token")
+
+        clean_text = (text or "").strip()
+        if not clean_text:
+            raise gr.Error("Text is empty")
+        if len(clean_text) > 60000:
+            raise gr.Error("Text is too long for one browser request. Use the desktop app for automatic long-script batching.")
+
+        engine = (engine or "edge").strip().lower()
+        if engine in CLONE_ENGINES and not clone_engines_enabled():
+            raise gr.Error("Clone engines are disabled on this Space")
+        voice = (voice or GRADIO_DEFAULT_VOICES.get(engine) or GRADIO_DEFAULT_VOICES["edge"]).strip()
+        style = (style or "").strip() or None
+        styledegree_value = str(styledegree) if styledegree is not None else None
+        reference_path = reference_audio if isinstance(reference_audio, str) and reference_audio else None
+
+        try:
+            if engine == "edge":
+                audio = await synthesize_edge(
+                    clean_text,
+                    voice,
+                    rate=rate,
+                    volume=volume,
+                    pitch=pitch,
+                    style=style,
+                    styledegree=styledegree_value,
+                )
+                return _write_audio_file(audio, ".mp3"), "Ready - Edge audio generated."
+
+            if engine == "piper":
+                audio = await asyncio.to_thread(synthesize_piper, clean_text, voice, float(speed or 1.0))
+                return _write_audio_file(audio, ".wav"), "Ready - Piper audio generated."
+
+            if engine == "silero":
+                audio = await asyncio.to_thread(synthesize_silero, clean_text, voice)
+                return _write_audio_file(audio, ".wav"), "Ready - Silero audio generated."
+
+            if engine == "anime":
+                audio = await synthesize_anime(clean_text, voice, style)
+                return _write_audio_file(audio, ".mp3"), "Ready - anime-style audio generated."
+
+            if engine == "f5tts":
+                audio = await synthesize_f5tts(clean_text, reference_path=reference_path)
+                return _write_audio_file(audio, ".wav"), "Ready - voice clone generated."
+
+            raise gr.Error(f"Unknown engine: {engine}")
+
+        except Exception as e:
+            logging.error("Gradio synthesis failed: %s", e, exc_info=True)
+            raise gr.Error(f"Synthesis failed: {str(e)[:220]}")
+
+    def default_voice_for_engine(engine):
+        return GRADIO_DEFAULT_VOICES.get((engine or "edge").lower(), GRADIO_DEFAULT_VOICES["edge"])
+
+    with gr.Blocks(title="VoiceCraft TTS Server", css=css) as demo:
+        gr.HTML(
+            """
+            <div class="voicecraft-hero">
+                <h1>VoiceCraft TTS Server</h1>
+                <p>FastAPI endpoints are live for the desktop app. This panel is only for quick browser testing.</p>
+            </div>
+            """
+        )
+        with gr.Row():
+            with gr.Column(scale=3):
+                text = gr.Textbox(
+                    label="Script",
+                    lines=10,
+                    placeholder="Paste text here...",
+                    value="Hello from VoiceCraft. Your deployment is ready for a quick audio test.",
+                )
+                with gr.Row():
+                    engine = gr.Dropdown(
+                        label="Engine",
+                        choices=ui_engines,
+                        value="edge",
+                    )
+                    voice = gr.Textbox(label="Voice code", value=GRADIO_DEFAULT_VOICES["edge"])
+                reference_audio = gr.Audio(label="Reference audio for cloning", sources=["upload"], type="filepath")
+                token = gr.Textbox(label="API token", type="password", placeholder="Required when API_SECRET is set")
+            with gr.Column(scale=2):
+                rate = gr.Textbox(label="Edge rate", value="+0%")
+                volume = gr.Textbox(label="Edge volume", value="+0%")
+                pitch = gr.Textbox(label="Edge pitch", value="+0Hz")
+                style = gr.Textbox(label="Edge style", placeholder="cheerful, sad, whispering...")
+                styledegree = gr.Slider(label="Style degree", minimum=0.0, maximum=2.0, value=1.0, step=0.1)
+                speed = gr.Slider(label="Piper speed", minimum=0.65, maximum=1.35, value=1.0, step=0.05)
+                run = gr.Button("Generate Audio", variant="primary")
+        output_audio = gr.Audio(label="Output", type="filepath")
+        status_box = gr.Markdown("Ready.")
+        gr.Markdown("API paths: `/tts`, `/health`, `/status`, `/all_voices`.")
+
+        engine.change(default_voice_for_engine, inputs=engine, outputs=voice)
+        run.click(
+            generate_audio,
+            inputs=[engine, text, voice, rate, volume, pitch, speed, style, styledegree, reference_audio, token],
+            outputs=[output_audio, status_box],
+        )
+
+    return demo
+
+
+@app.get("/")
+def root():
+    return {
+        "name": "VoiceCraft TTS Server",
+        "status": "running",
+        "clone_enabled": clone_engines_enabled(),
+        "zerogpu_enabled": False,
+        "clone_backend": CLONE_BACKEND_NAME,
+        "endpoints": ["/tts", "/health", "/status", "/all_voices"],
+    }
+
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT") or os.environ.get("GRADIO_SERVER_PORT") or 7860)
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=port,
+        share=False,
+        show_error=True,
+    )

requirements-docker.txt

@@ -0,0 +1,13 @@
+fastapi
+uvicorn[standard]
+gradio==6.20.0
+edge-tts
+python-multipart
+requests
+numpy>=2.0,<3
+scipy>=1.15.3
+soundfile>=0.12,<1
+pocket-tts>=2.1.0
+
+# Docker profile is now CPU clone first. Silero v4 may be unavailable because
+# Pocket TTS requires a newer PyTorch runtime.

requirements.txt

@@ -1,8 +1,10 @@
 fastapi
-uvicorn
+uvicorn[standard]
+gradio==6.20.0
 edge-tts
 python-multipart
 requests
-numpy<2
-scipy
-torch==2.1.2
+numpy>=2.0,<3
+scipy>=1.15.3
+soundfile>=0.12,<1
+pocket-tts>=2.1.0