app.py

import os, io, asyncio, tempfile, threading, re, subprocess, shutil, logging, secrets, sys, platform, hashlib, time, json
from contextlib import asynccontextmanager, nullcontext

try:
    import firebase_admin
    from firebase_admin import auth as fb_auth, firestore, credentials
    HAS_FIREBASE = True
except Exception:
    HAS_FIREBASE = False
    firebase_admin = None
    fb_auth = None
    firestore = None
    credentials = None


def _apply_space_config_secret():
    """Allow one protected HF secret to populate the usual env settings."""
    raw = os.environ.get("VOICECRAFT_SPACE_CONFIG_JSON", "").strip()
    if not raw:
        return
    try:
        config = json.loads(raw)
        if not isinstance(config, dict):
            raise ValueError("VOICECRAFT_SPACE_CONFIG_JSON must be a JSON object")
    except Exception as exc:
        print("Space config JSON notice:", exc)
        return

    def set_default(env_name, *keys, transform=str):
        if os.environ.get(env_name):
            return
        for key in keys:
            if key in config and config[key] not in (None, ""):
                os.environ[env_name] = transform(config[key])
                return

    firebase_json = (
        config.get("firebase_service_account_json")
        or config.get("firebase_admin_json")
        or config.get("firebase_service_account")
        or config.get("service_account")
    )
    if firebase_json and not os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON"):
        os.environ["FIREBASE_SERVICE_ACCOUNT_JSON"] = (
            firebase_json if isinstance(firebase_json, str) else json.dumps(firebase_json)
        )

    set_default("VOICECRAFT_SPACE_ID", "space_id", "voicecraft_space_id")
    set_default("VOICECRAFT_AUTH_MODE", "auth_mode", "voicecraft_auth_mode")
    set_default("VOICECRAFT_MAX_CONCURRENT_JOBS", "max_concurrent_jobs", "voicecraft_max_concurrent_jobs")
    set_default("VOICECRAFT_SERVICE_VERSION", "service_version", "voicecraft_service_version")
    set_default("MIN_CLIENT_VERSION", "min_client_version", "minimum_version")
    set_default("ENABLE_CLONE_ENGINES", "enable_clone_engines", "clone_enabled", transform=lambda value: "true" if bool(value) else "false")
    set_default("HF_TOKEN", "hf_token", "huggingface_token")
    set_default("VOICECRAFT_API_SECRET", "api_secret", "voicecraft_api_secret")


_apply_space_config_secret()

_fb_app = None
_firestore_db = None
def _init_firebase():
    global _fb_app, _firestore_db
    if not HAS_FIREBASE:
        return None
    if _fb_app is None:
        try:
            credential_json = os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON", "").strip()
            cred_path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', '').strip()
            if credential_json:
                cred = credentials.Certificate(json.loads(credential_json))
                _fb_app = firebase_admin.initialize_app(cred)
                _firestore_db = firestore.client()
            elif cred_path and os.path.exists(cred_path):
                cred = credentials.Certificate(cred_path)
                _fb_app = firebase_admin.initialize_app(cred)
                _firestore_db = firestore.client()
        except Exception as e:
            print("Firebase init notice:", e)
            return None
    return _firestore_db

import requests

# Hide console window on Windows
CREATE_NO_WINDOW = 0x08000000 if sys.platform == "win32" else 0
from fastapi import FastAPI, Form, Request, HTTPException, UploadFile, File
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.gzip import GZipMiddleware

SERVICE_VERSION = os.environ.get("VOICECRAFT_SERVICE_VERSION", "3.1.2").strip()
_BACKGROUND_WARMUP_LOCK = threading.Lock()
_BACKGROUND_WARMUP_STARTED = False


@asynccontextmanager
async def _service_lifespan(_app):
    """Start model warm-up only when the ASGI service actually starts.

    The old build launched download/model threads while merely importing this
    module. That made tests, health tooling, and multi-worker process startup
    unpredictable. A single guarded startup pass is enough for each process.
    """
    global _BACKGROUND_WARMUP_STARTED
    disabled = os.environ.get("VOICECRAFT_DISABLE_WARMUP", "").strip().lower() in {
        "1", "true", "yes", "on"
    }
    if not disabled:
        with _BACKGROUND_WARMUP_LOCK:
            should_start = not _BACKGROUND_WARMUP_STARTED
            _BACKGROUND_WARMUP_STARTED = True
        if should_start:
            threading.Thread(target=setup_piper, daemon=True, name="voicecraft-piper-warmup").start()
            threading.Thread(target=setup_silero, daemon=True, name="voicecraft-silero-warmup").start()
            if clone_engines_enabled():
                threading.Thread(target=_warm_pocket_model, daemon=True, name="voicecraft-clone-warmup").start()
    yield


app = FastAPI(
    debug=False,
    title="VoiceCraft TTS Server",
    description="VoiceCraft desktop service.",
    version=SERVICE_VERSION,
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
    lifespan=_service_lifespan,
)
app.add_middleware(GZipMiddleware, minimum_size=1000)


# -----------------------------------------------------------------------------
# SECURITY — server-side Firebase authorization
# -----------------------------------------------------------------------------
_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)
)
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_RETRY_CHUNK_CHARACTERS = 900
CLONE_ENGINE_PREFIXES = ("f5tts:",)
CLONE_ENGINES = ["f5tts"]
CLONE_BACKEND_NAME = "pocket-tts-cpu"
BASE_ENGINES = ["edge", "piper", "silero"]

def _derive_space_id():
    explicit = os.environ.get("VOICECRAFT_SPACE_ID", "").strip()
    hf_space = os.environ.get("SPACE_ID", "").strip()
    if explicit.lower() in {"auto", "hf", "huggingface"}:
        explicit = ""
    selected = explicit or hf_space or os.environ.get("SPACE_HOST", "").strip()
    selected = selected.rsplit("/", 1)[-1].strip()
    selected = re.sub(r"[^0-9A-Za-z_-]+", "-", selected).strip("-_")
    return selected[:80] or "unregistered"

SPACE_ID = _derive_space_id()


def _derive_self_url():
    """Best-effort public URL of this Space, for access checks by URL."""
    explicit = os.environ.get("VOICECRAFT_SPACE_URL", "").strip()
    if explicit:
        return explicit
    host = os.environ.get("SPACE_HOST", "").strip()
    if host:
        return host if host.startswith("http") else f"https://{host}"
    repo = os.environ.get("SPACE_ID", "").strip()
    if "/" in repo:
        owner, name = repo.split("/", 1)
        slug = f"{owner}-{name}".replace("_", "-").replace(".", "-").lower()
        return f"https://{slug}.hf.space"
    return ""


SELF_SPACE_URL = _derive_self_url()
MAX_CONCURRENT_JOBS = max(1, min(int(os.environ.get("VOICECRAFT_MAX_CONCURRENT_JOBS", "1")), 20))
_SPACE_CAPACITY = threading.BoundedSemaphore(MAX_CONCURRENT_JOBS)
_SPACE_STATE_LOCK = threading.Lock()
_ACTIVE_JOBS = 0
_POLICY_LOCK = threading.Lock()
_POLICY_CACHE = {"loaded_at": 0.0, "maintenance_mode": False, "minimum_version": MIN_CLIENT_VERSION}


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 clone_backend_available() -> bool:
    if not clone_engines_enabled():
        return False
    if _POCKET_MODEL_ERROR or _POCKET_GENERATION_ERROR:
        return False
    if _POCKET_MODEL is not None:
        return bool(getattr(_POCKET_MODEL, "has_voice_cloning", False))
    return _hf_token_configured()


def clone_backend_status() -> dict:
    if not clone_engines_enabled():
        return {"enabled": False, "ready": False, "message": "disabled"}
    if _POCKET_GENERATION_ERROR:
        return {"enabled": False, "ready": False, "message": "generation_failed"}
    if _POCKET_MODEL is not None:
        ready = bool(getattr(_POCKET_MODEL, "has_voice_cloning", False))
        return {
            "enabled": ready,
            "ready": ready,
            "message": "ready" if ready else "model_loaded_without_clone_weights",
        }
    if _POCKET_MODEL_ERROR:
        return {"enabled": False, "ready": False, "message": "model_startup_failed"}
    if not _hf_token_configured():
        return {"enabled": False, "ready": False, "message": "hf_token_required"}
    return {"enabled": True, "ready": False, "message": "loading"}


def available_engines():
    return BASE_ENGINES + (CLONE_ENGINES if clone_backend_available() else [])


def get_runtime_policy():
    now = time.monotonic()
    with _POLICY_LOCK:
        if now - float(_POLICY_CACHE.get("loaded_at", 0.0)) < 30:
            return dict(_POLICY_CACHE)
    policy = {
        "loaded_at": now,
        "maintenance_mode": False,
        "minimum_version": MIN_CLIENT_VERSION,
        "space_enabled": True,
        "space_clone_enabled": clone_backend_available(),
    }
    db = _init_firebase()
    if db is not None:
        try:
            runtime_doc = db.collection("public_config").document("runtime").get()
            if runtime_doc.exists:
                runtime = runtime_doc.to_dict() or {}
                policy["maintenance_mode"] = bool(runtime.get("maintenance_mode", False))
                policy["minimum_version"] = str(runtime.get("minimum_version") or MIN_CLIENT_VERSION)
            if SPACE_ID != "unregistered":
                space_doc = db.collection("spaces").document(SPACE_ID).get()
                if space_doc.exists:
                    space_data = space_doc.to_dict() or {}
                    policy["space_enabled"] = bool(space_data.get("enabled", False))
                    policy["space_clone_enabled"] = bool(space_data.get("clone_enabled", False)) and clone_backend_available()
        except Exception:
            pass
    with _POLICY_LOCK:
        _POLICY_CACHE.clear()
        _POLICY_CACHE.update(policy)
    return dict(policy)


def _clean_hf_space_url(value):
    raw = str(value or "").strip().rstrip("/")
    if not raw.startswith("https://"):
        return ""
    host = raw.split("/", 3)[2].lower()
    if host == "hf.space" or host.endswith(".hf.space") or host == "huggingface.co" or host.endswith(".huggingface.co"):
        return raw
    return ""


def _space_public_payload(snapshot_id, item):
    url = _clean_hf_space_url((item or {}).get("url"))
    if not url:
        return None
    return {
        "id": str(snapshot_id or "")[:80],
        "name": str((item or {}).get("name") or snapshot_id or "")[:80],
        "url": url,
        # Default False, matching get_runtime_policy() and the admin panel's
        # space_row(). It used to default True here, so a Space document with
        # no clone_enabled field advertised cloning to the desktop and then
        # refused every clone job with a 403 from the policy check.
        "clone_enabled": bool((item or {}).get("clone_enabled", (item or {}).get("clone", False))),
        "priority": int((item or {}).get("priority", 100) or 100),
        "max_concurrent_jobs": max(1, int((item or {}).get("max_concurrent_jobs", 1) or 1)),
    }


def _license_space_pool(db, uid, user_data, license_key, lic):
    """Return only the Spaces this signed-in user may see/use.

    Access is decided by ``space_access_mode`` on the user document:

      pool          assigned Spaces first, then the shared public pool
      assigned_only ONLY the Spaces the admin ticked for this user
      blocked       nothing at all

    Private Spaces (``visibility == "private"``) never enter the shared pool;
    they are handed out exclusively to the users they were assigned to.
    """
    if db is None:
        return []

    assigned_ids = set()
    assigned_urls = set()
    assigned_tts_ids = set()
    assigned_tts_urls = set()
    assigned_clone_ids = set()
    assigned_clone_urls = set()

    mode = ""
    for source in (user_data or {}, lic or {}):
        if not mode:
            mode = str(source.get("space_access_mode") or "").strip().lower()
        for key in ("space_ids", "clone_space_ids", "dedicated_space_ids", "assigned_space_ids"):
            for value in source.get(key) or []:
                if value:
                    assigned_ids.add(str(value).strip())
        for key in ("spaces", "clone_spaces", "dedicated_spaces", "assigned_spaces"):
            for value in source.get(key) or []:
                if isinstance(value, dict):
                    payload = _space_public_payload(value.get("id") or value.get("name"), value)
                    if payload:
                        assigned_urls.add(payload["url"])
                else:
                    url = _clean_hf_space_url(value)
                    if url:
                        assigned_urls.add(url)
                    else:
                        assigned_ids.add(str(value).strip())
        for key in ("assigned_tts_space_ids",):
            for value in source.get(key) or []:
                if value:
                    assigned_tts_ids.add(str(value).strip())
        for key in ("assigned_clone_space_ids",):
            for value in source.get(key) or []:
                if value:
                    assigned_clone_ids.add(str(value).strip())
        for value in source.get("assigned_tts_spaces") or []:
            url = _clean_hf_space_url(value)
            if url:
                assigned_tts_urls.add(url)
            elif value:
                assigned_tts_ids.add(str(value).strip())
        for value in source.get("assigned_clone_spaces") or []:
            url = _clean_hf_space_url(value)
            if url:
                assigned_clone_urls.add(url)
            elif value:
                assigned_clone_ids.add(str(value).strip())

    if mode not in {"pool", "assigned_only", "blocked"}:
        mode = "pool"
    if mode == "blocked":
        return []

    assigned_ids |= assigned_tts_ids | assigned_clone_ids
    assigned_urls |= assigned_tts_urls | assigned_clone_urls

    dedicated = []
    dedicated_unhealthy = []
    auto_pool = []
    try:
        for snapshot in db.collection("spaces").where("enabled", "==", True).stream():
            item = snapshot.to_dict() or {}
            # Health synchronization marks paused/crashed workers false. Those
            # endpoints are kept out of the shared pool so the desktop doesn't
            # waste time failing over from known-dead Spaces. A dedicated
            # Space is different: for an assigned_only customer it may be the
            # only route they have, and a stale flag from one failed probe
            # would otherwise take them fully offline. Those are held back as
            # a last resort instead of being dropped.
            space_unhealthy = item.get("last_health_ok") is False
            payload = _space_public_payload(snapshot.id, item)
            if not payload:
                continue

            visibility = str(item.get("visibility") or "").strip().lower()
            if visibility not in {"public", "private"}:
                visibility = "private" if (
                    item.get("assigned_uid") or item.get("assigned_license_key")
                ) else "public"

            has_scoped_assignments = bool(
                assigned_tts_ids or assigned_tts_urls
                or assigned_clone_ids or assigned_clone_urls
            )
            legacy_direct_match = (
                str(item.get("assigned_uid") or "").strip() == str(uid or "").strip()
                or str(
                    item.get("assigned_license_key") or item.get("assigned_license") or ""
                ).strip() == str(license_key or "").strip()
            )
            is_dedicated = (
                snapshot.id in assigned_ids
                or payload["url"] in assigned_urls
                or (legacy_direct_match and not has_scoped_assignments)
            )

            if is_dedicated:
                # Per-Space entitlements: a worker assigned for TTS only must
                # not accept clone jobs from this user, and vice versa.
                tts_ok = True
                clone_ok = bool(payload["clone_enabled"])
                if assigned_tts_ids or assigned_tts_urls or assigned_clone_ids or assigned_clone_urls:
                    tts_ok = (
                        snapshot.id in assigned_tts_ids
                        or payload["url"] in assigned_tts_urls
                    )
                    clone_ok = clone_ok and (
                        snapshot.id in assigned_clone_ids
                        or payload["url"] in assigned_clone_urls
                    )
                    if not tts_ok and not clone_ok and not has_scoped_assignments:
                        # Only present through a legacy combined list.
                        tts_ok = True
                payload = dict(payload, tts_enabled=tts_ok, clone_enabled=clone_ok, dedicated=True)
                (dedicated_unhealthy if space_unhealthy else dedicated).append(payload)
            elif (
                not space_unhealthy
                and visibility == "public"
                and not item.get("assigned_uid")
                and not item.get("assigned_license_key")
                and not item.get("assigned_license")
            ):
                auto_pool.append(dict(payload, tts_enabled=True, dedicated=False))
    except Exception:
        return []

    if mode == "assigned_only":
        auto_pool = []

    if not dedicated and not auto_pool and dedicated_unhealthy:
        # Better to hand back a Space that failed its last probe than to hand
        # back nothing: the desktop re-checks /health itself before using one.
        dedicated = dedicated_unhealthy

    dedicated.sort(key=lambda item: (item["priority"], item["name"].lower()))
    auto_pool.sort(key=lambda item: (item["priority"], item["name"].lower()))
    seen_urls = set()
    rows = []
    for item in dedicated + auto_pool:
        if item["url"] not in seen_urls:
            rows.append(item)
            seen_urls.add(item["url"])
    return rows


def _space_pool_is_exclusive(user_data, lic):
    """True when the desktop must replace its cached pool, not merge into it."""
    for source in (user_data or {}, lic or {}):
        mode = str(source.get("space_access_mode") or "").strip().lower()
        if mode in {"assigned_only", "blocked"}:
            return True
        if mode == "pool":
            return False
    return False


def verify_token(request: Request):
    if not API_SECRET:
        logging.warning("API_SECRET missing in Hugging Face Space secrets")
        raise HTTPException(status_code=503, detail="Service unavailable")
    token = request.headers.get("X-API-Token", "").strip()
    if not secrets.compare_digest(token, API_SECRET):
        raise HTTPException(status_code=403, detail="Unauthorized")


_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 _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 _coerce_utc_datetime(value):
    """Parse Firestore timestamps and ISO strings into aware UTC datetimes."""
    from datetime import date, datetime, timezone

    if value in (None, ""):
        return None
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, date):
        parsed = datetime(value.year, value.month, value.day)
    elif isinstance(value, str):
        raw = value.strip()
        if not raw:
            return None
        try:
            parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
        except ValueError:
            return None
    else:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _require_firebase_db():
    """Return a usable Firebase client or a truthful 503 response."""
    if not HAS_FIREBASE or fb_auth is None or firestore is None:
        raise HTTPException(503, "Firebase authentication is not configured")
    db = _init_firebase()
    if db is None:
        raise HTTPException(503, "Firebase database is unavailable")
    return db


def _reserve_monthly_usage(db, uid, license_key, lic, is_clone, requested_characters):
    from datetime import datetime, timezone

    requested_characters = max(0, int(requested_characters or 0))
    if not requested_characters:
        return
    month_id = datetime.now(timezone.utc).strftime("%Y%m")
    usage_ref = db.collection("usage_monthly").document(f"{uid}_{month_id}")
    used_field = "clone_characters" if is_clone else "tts_characters"
    reserved_field = "clone_reserved" if is_clone else "tts_reserved"
    limit = int(
        lic.get("monthly_clone_characters" if is_clone else "monthly_characters")
        or 0
    )
    transaction = db.transaction()

    @firestore.transactional
    def reserve(transaction):
        snapshot = usage_ref.get(transaction=transaction)
        usage = snapshot.to_dict() if snapshot.exists else {}
        used = max(0, int(usage.get(used_field) or 0))
        reserved = max(0, int(usage.get(reserved_field) or 0))
        if limit and used + reserved + requested_characters > limit:
            label = "voice-clone" if is_clone else "TTS character"
            raise HTTPException(
                status_code=403,
                detail=f"Monthly {label} allowance has been reached",
            )
        transaction.set(
            usage_ref,
            {
                "uid": uid,
                "license_key": license_key,
                "period": month_id,
                reserved_field: reserved + requested_characters,
                "updated_at": firestore.SERVER_TIMESTAMP,
            },
            merge=True,
        )

    try:
        reserve(transaction)
    except HTTPException:
        raise
    except Exception as exc:
        logging.warning("Usage reservation failed: %s", exc.__class__.__name__)
        raise HTTPException(503, "Usage allowance could not be verified")


def _release_monthly_reservation(uid, is_clone, characters):
    """Give a reservation back when the request never reaches record_usage.

    ``_reserve_monthly_usage`` books characters against the customer's monthly
    allowance before generation starts, and ``record_usage`` returns them in a
    ``finally`` block. Anything that rejects the request *between* those two
    points (rate limiting, a per-request size check) would otherwise leak the
    reservation permanently, slowly eating the customer's paid allowance until
    an admin resets it by hand.
    """
    characters = max(0, int(characters or 0))
    if not characters or not uid:
        return
    from datetime import datetime, timezone

    db = _init_firebase()
    if db is None:
        return
    reserved_field = "clone_reserved" if is_clone else "tts_reserved"
    month_id = datetime.now(timezone.utc).strftime("%Y%m")
    usage_ref = db.collection("usage_monthly").document(f"{uid}_{month_id}")
    transaction = db.transaction()

    @firestore.transactional
    def release(transaction):
        snapshot = usage_ref.get(transaction=transaction)
        usage = snapshot.to_dict() if snapshot.exists else {}
        current = max(0, int(usage.get(reserved_field) or 0))
        if current <= 0:
            return
        transaction.set(
            usage_ref,
            {
                reserved_field: max(0, current - characters),
                "updated_at": firestore.SERVER_TIMESTAMP,
            },
            merge=True,
        )

    try:
        release(transaction)
    except Exception as exc:
        logging.warning("Reservation release failed: %s", exc.__class__.__name__)


def _monthly_credit_summary(db, uid, lic):
    """Return the current UTC month's enforced credit balance for the client."""
    from datetime import datetime, timezone

    month_id = datetime.now(timezone.utc).strftime("%Y%m")
    snapshot = db.collection("usage_monthly").document(f"{uid}_{month_id}").get()
    usage = snapshot.to_dict() if snapshot.exists else {}

    def balance(limit_field, used_field, reserved_field):
        limit = max(0, int(lic.get(limit_field) or 0))
        used = max(0, int(usage.get(used_field) or 0))
        reserved = max(0, int(usage.get(reserved_field) or 0))
        unlimited = limit == 0
        remaining = None if unlimited else max(0, limit - used - reserved)
        return limit, used, reserved, remaining, unlimited

    tts = balance("monthly_characters", "tts_characters", "tts_reserved")
    clone = balance(
        "monthly_clone_characters", "clone_characters", "clone_reserved"
    )
    return {
        "usage_period": month_id,
        "monthly_characters": tts[0],
        "tts_characters_used": tts[1],
        "tts_characters_reserved": tts[2],
        "tts_characters_remaining": tts[3],
        "tts_unlimited": tts[4],
        "monthly_clone_characters": clone[0],
        "clone_characters_used": clone[1],
        "clone_characters_reserved": clone[2],
        "clone_characters_remaining": clone[3],
        "clone_unlimited": clone[4],
    }


def _register_device_tx(db, user_ref, device_id, max_devices):
    """Transactionally register a device against the user's device list."""
    max_devices = max(1, int(max_devices or 1))

    @firestore.transactional
    def register(transaction):
        snapshot = user_ref.get(transaction=transaction)
        current = snapshot.to_dict() if snapshot.exists else {}
        ids = list(current.get("device_ids") or [])
        if device_id not in ids:
            if len(ids) >= max_devices:
                raise HTTPException(403, "This device is not registered for the license")
            ids.append(device_id)
            transaction.update(user_ref, {"device_ids": ids})
        return ids

    try:
        return register(db.transaction())
    except HTTPException:
        raise
    except Exception as exc:
        logging.warning("Device registration failed: %s", exc.__class__.__name__)
        raise HTTPException(503, "Device registration could not be verified")


def _collect_space_assignments(user_data, lic, is_clone: bool):
    """Return request-scoped and all assignment references for one account."""
    sources = (user_data or {}, lic or {})
    scoped_id_key = "assigned_clone_space_ids" if is_clone else "assigned_tts_space_ids"
    scoped_url_key = "assigned_clone_spaces" if is_clone else "assigned_tts_spaces"
    scoped_keys = (
        "assigned_tts_space_ids", "assigned_clone_space_ids",
        "assigned_tts_spaces", "assigned_clone_spaces",
    )
    has_scoped_lists = any(key in source for source in sources for key in scoped_keys)

    request_ids, request_urls = set(), set()
    all_ids, all_urls = set(), set()
    for source in sources:
        for key in ("assigned_tts_space_ids", "assigned_clone_space_ids", "assigned_space_ids"):
            for value in source.get(key) or []:
                if value:
                    all_ids.add(str(value).strip())
        for key in ("assigned_tts_spaces", "assigned_clone_spaces", "assigned_spaces"):
            for value in source.get(key) or []:
                cleaned = _clean_hf_space_url(value)
                if cleaned:
                    all_urls.add(cleaned)

        id_keys = (scoped_id_key,) if has_scoped_lists else (scoped_id_key, "assigned_space_ids")
        url_keys = (scoped_url_key,) if has_scoped_lists else (scoped_url_key, "assigned_spaces")
        for key in id_keys:
            for value in source.get(key) or []:
                if value:
                    request_ids.add(str(value).strip())
        for key in url_keys:
            for value in source.get(key) or []:
                cleaned = _clean_hf_space_url(value)
                if cleaned:
                    request_urls.add(cleaned)

    return request_ids, request_urls, all_ids, all_urls, has_scoped_lists


def _enforce_space_access(uid, user_data, lic, is_clone: bool, db=None, license_key=""):
    """Enforce blocked, scoped, assigned-only, and private-worker access.

    The old implementation only checked assignments when a user was in
    ``assigned_only`` mode. A pool-mode customer who learned a private worker
    URL could therefore call that worker directly, and a TTS-only assignment
    could be used for cloning. Public pool workers remain available normally;
    private/dedicated workers and scoped grants are now authoritative on every
    request.
    """
    mode = ""
    for source in (user_data or {}, lic or {}):
        candidate = str(source.get("space_access_mode") or "").strip().lower()
        if candidate:
            mode = candidate
            break
    if mode not in {"pool", "assigned_only", "blocked"}:
        mode = "pool"
    if mode == "blocked":
        raise HTTPException(
            403,
            "Server access for this account has been disabled. Please contact support.",
        )

    request_ids, request_urls, all_ids, all_urls, has_scoped_lists = _collect_space_assignments(
        user_data, lic, is_clone
    )
    own_url = _clean_hf_space_url(SELF_SPACE_URL)
    assigned_for_request = SPACE_ID in request_ids or bool(own_url and own_url in request_urls)
    listed_in_any_scope = SPACE_ID in all_ids or bool(own_url and own_url in all_urls)

    # If this account explicitly lists the worker for one service only, the
    # other service must not be able to use the same URL even in pool mode.
    if has_scoped_lists and listed_in_any_scope and not assigned_for_request:
        service = "voice cloning" if is_clone else "text-to-speech"
        raise HTTPException(403, f"This server is not assigned for {service} on your account")

    space_data = {}
    if SPACE_ID != "unregistered":
        db = db or _init_firebase()
        if db is None:
            raise HTTPException(503, "Server access policy is unavailable")
        try:
            snapshot = db.collection("spaces").document(SPACE_ID).get()
            if snapshot.exists:
                space_data = snapshot.to_dict() or {}
        except HTTPException:
            raise
        except Exception as exc:
            logging.warning("Space access lookup failed: %s", exc.__class__.__name__)
            raise HTTPException(503, "Server access policy could not be verified")

    visibility = str(space_data.get("visibility") or "").strip().lower()
    directly_assigned_uid = str(space_data.get("assigned_uid") or "").strip()
    directly_assigned_license = str(
        space_data.get("assigned_license_key") or space_data.get("assigned_license") or ""
    ).strip()
    legacy_direct_match = bool(
        (directly_assigned_uid and directly_assigned_uid == str(uid or "").strip())
        or (
            directly_assigned_license
            and directly_assigned_license == str(license_key or user_data.get("license_key") or "").strip()
        )
    )
    is_reserved_worker = visibility == "private" or bool(
        directly_assigned_uid or directly_assigned_license
    )

    if is_reserved_worker:
        allowed = assigned_for_request or (legacy_direct_match and not has_scoped_lists)
        if not allowed:
            raise HTTPException(403, "This private server is not assigned to your account")

    if mode == "assigned_only" and not (
        assigned_for_request or (legacy_direct_match and not has_scoped_lists)
    ):
        raise HTTPException(
            403,
            "This server is not assigned to your account. Restart VoiceCraft to "
            "refresh your assigned servers, or contact support.",
        )


def verify_firebase_auth(request, is_clone: bool = False, requested_characters: int = 0):
    _require_firebase_db()
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing authentication token")
    token = auth_header.split("Bearer ", 1)[1]
    try:
        decoded = fb_auth.verify_id_token(token)
    except Exception:
        raise HTTPException(401, "Invalid or expired authentication token")
    uid = decoded["uid"]
    db = _require_firebase_db()
    user_doc = db.collection("users").document(uid).get()
    if not user_doc.exists:
        raise HTTPException(403, "User account not found")
    user_data = user_doc.to_dict() or {}
    if user_data.get("is_blocked"):
        raise HTTPException(403, "Account has been blocked")
    device_id = str(request.headers.get("X-Device-ID", "")).strip()[:128]
    if not device_id:
        raise HTTPException(403, "Device identification is required")
    license_key = user_data.get("license_key")
    if not license_key:
        raise HTTPException(403, "No active license. Activate a license key from Account settings")

    license_doc = db.collection("licenses").document(license_key).get()
    if not license_doc.exists:
        raise HTTPException(403, "License key not found")
    lic = license_doc.to_dict()
    if lic.get("status") not in ("active",):
        raise HTTPException(403, f"License is {lic.get('status', 'invalid')}")
    max_devs = max(1, int(lic.get("max_devices") or 1))
    device_ids = _register_device_tx(db, user_doc.reference, device_id, max_devs)
    if len(device_ids) > max_devs:
        raise HTTPException(403, "Maximum licensed devices exceeded")
    
    from datetime import datetime, timezone
    raw_expiry = lic.get("expiry_date")
    expiry = _coerce_utc_datetime(raw_expiry)
    if raw_expiry not in (None, "") and expiry is None:
        raise HTTPException(403, "License expiry is invalid. Please contact support")
    if expiry and expiry < datetime.now(timezone.utc):
        raise HTTPException(403, "License has expired")

    if is_clone and (
        not lic.get("voice_clone")
        or not user_data.get("voice_clone_enabled", lic.get("voice_clone", False))
    ):
        raise HTTPException(403, "Voice cloning not included in your plan")

    # Server-side access enforcement. The desktop is told which Spaces it may
    # use, but a customer who saved an old URL could still call this worker
    # directly, so the worker checks for itself on every request.
    _enforce_space_access(uid, user_data, lic, is_clone, db=db, license_key=license_key)
    _reserve_monthly_usage(
        db,
        uid,
        license_key,
        lic,
        is_clone,
        requested_characters,
    )
    return uid, license_key, lic, user_data

def _enforce_rate_limit(uid: str, is_clone: bool):
    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"{uid}:{'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, requested_characters: int = 0) -> dict:
    policy = await asyncio.to_thread(get_runtime_policy)
    if policy.get("maintenance_mode"):
        raise HTTPException(status_code=503, detail="VoiceCraft is temporarily under maintenance")
    if not policy.get("space_enabled", True):
        raise HTTPException(status_code=503, detail="This worker has been disabled")
    if engine in CLONE_ENGINES and not policy.get("space_clone_enabled", False):
        raise HTTPException(status_code=403, detail="Voice cloning is not available on this server")
    client_version = request.headers.get("X-Client-Version", "").strip()
    minimum_version = str(policy.get("minimum_version") or MIN_CLIENT_VERSION)
    if client_version and _version_tuple(client_version) < _version_tuple(minimum_version):
        raise HTTPException(status_code=426, detail="VoiceCraft update required")

    if AUTH_MODE in {"api_secret", "layered"}:
        verify_token(request)
    if AUTH_MODE == "api_secret":
        return {
            "uid": "api_secret_system",
            "license_key": "api_secret_key",
            "license": {},
            "usage_reserved": False,
        }
        
    if not client_version:
        raise HTTPException(status_code=426, detail="VoiceCraft client version is required")
        
    is_clone = engine in CLONE_ENGINES
    uid, license_key, lic, _user = await asyncio.to_thread(
        verify_firebase_auth,
        request,
        is_clone,
        requested_characters,
    )
    try:
        _enforce_rate_limit(uid, is_clone)
    except Exception:
        # The allowance was already booked inside verify_firebase_auth; a
        # throttled request must not cost the customer any characters.
        await asyncio.to_thread(
            _release_monthly_reservation, uid, is_clone, requested_characters
        )
        raise
    return {
        "uid": uid,
        "license_key": license_key,
        "license": lic,
        "usage_reserved": bool(requested_characters),
    }


def acquire_worker_slot():
    global _ACTIVE_JOBS
    if not _SPACE_CAPACITY.acquire(blocking=False):
        raise HTTPException(status_code=503, detail="Worker is busy; try another Space")
    with _SPACE_STATE_LOCK:
        _ACTIVE_JOBS += 1
        active = _ACTIVE_JOBS
    db = _init_firebase()
    if db is not None and SPACE_ID != "unregistered":
        try:
            _policy = get_runtime_policy()
            db.collection("spaces").document(SPACE_ID).set({
                "active_jobs": active,
                "last_heartbeat": firestore.SERVER_TIMESTAMP,
                "enabled": bool(_policy.get("space_enabled", True)),
                "clone_enabled": bool(_policy.get("space_clone_enabled", False)),
            }, merge=True)
        except Exception:
            pass


def release_worker_slot():
    global _ACTIVE_JOBS
    with _SPACE_STATE_LOCK:
        _ACTIVE_JOBS = max(0, _ACTIVE_JOBS - 1)
        active = _ACTIVE_JOBS
    _SPACE_CAPACITY.release()
    db = _init_firebase()
    if db is not None and SPACE_ID != "unregistered":
        try:
            _policy = get_runtime_policy()
            db.collection("spaces").document(SPACE_ID).set({
                "active_jobs": active,
                "last_heartbeat": firestore.SERVER_TIMESTAMP,
                "enabled": bool(_policy.get("space_enabled", True)),
                "clone_enabled": bool(_policy.get("space_clone_enabled", False)),
            }, merge=True)
        except Exception:
            pass



def record_usage(auth_context, engine, characters, status, duration_ms):
    uid = str((auth_context or {}).get("uid") or "")
    if not uid:
        return
    db = _init_firebase()
    if db is None:
        return
    is_clone = engine in CLONE_ENGINES
    event = {
        "uid": uid,
        "license_key": str(auth_context.get("license_key") or ""),
        "space_id": SPACE_ID,
        "engine": engine,
        "characters": int(characters),
        "is_clone": is_clone,
        "status": status,
        "duration_ms": int(duration_ms),
        "created_at": firestore.SERVER_TIMESTAMP,
    }
    try:
        batch = db.batch()
        event_ref = db.collection("usage_events").document()
        user_ref = db.collection("users").document(uid)
        space_ref = db.collection("spaces").document(SPACE_ID)
        from datetime import datetime, timezone
        month_id = datetime.now(timezone.utc).strftime("%Y%m")
        monthly_ref = db.collection("usage_monthly").document(f"{uid}_{month_id}")
        batch.set(event_ref, event)
        billed_characters = int(characters) if status == "success" else 0
        reserved_field = "clone_reserved" if is_clone else "tts_reserved"
        user_updates = {
            "generation_count": firestore.Increment(1),
            "last_generation_at": firestore.SERVER_TIMESTAMP,
            "tts_characters": firestore.Increment(0 if is_clone else billed_characters),
            "clone_characters": firestore.Increment(billed_characters if is_clone else 0),
        }
        batch.set(user_ref, user_updates, merge=True)
        monthly_updates = {
            "uid": uid,
            "period": month_id,
            "tts_characters": firestore.Increment(0 if is_clone else billed_characters),
            "clone_characters": firestore.Increment(billed_characters if is_clone else 0),
            "generation_count": firestore.Increment(1),
            "updated_at": firestore.SERVER_TIMESTAMP,
        }
        if auth_context.get("usage_reserved"):
            monthly_updates[reserved_field] = firestore.Increment(-int(characters))
        batch.set(monthly_ref, monthly_updates, merge=True)
        if SPACE_ID != "unregistered":
            batch.set(space_ref, {
                "total_requests": firestore.Increment(1),
                "total_characters": firestore.Increment(billed_characters),
                "last_heartbeat": firestore.SERVER_TIMESTAMP,
            }, merge=True)
        batch.commit()
    except Exception as exc:
        logging.warning("Usage logging failed: %s", exc.__class__.__name__)
        # Firestore batches are atomic. If the batch failed, release the
        # pre-authorized allowance so a transient logging fault cannot consume
        # customer credits permanently.
        if auth_context.get("usage_reserved"):
            _release_monthly_reservation(uid, is_clone, characters)



# -----------------------------------------------------------------------------
# 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] downloading binary...")
            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("[OK] Piper binary ready")

        PIPER_READY = True
        print("[OK] Piper TTS ready")
    except Exception as e:
        print(f"[WARN] Piper setup failed (non-critical): {e}")
        PIPER_READY = False



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"[DOWNLOAD] 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 _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 _numpy_concat(parts: list, is_mp3: bool) -> bytes:
    if len(parts) == 1:
        return parts[0]
    import numpy as np
    import soundfile as sf
    import subprocess, tempfile, os

    audio_arrays = []
    samplerate = 24000

    for data in parts:
        fd, in_path = tempfile.mkstemp(suffix=".mp3" if is_mp3 else ".wav")
        os.close(fd)
        with open(in_path, "wb") as f:
            f.write(data)
        
        wav_path = in_path
        if is_mp3:
            fd, wav_path = tempfile.mkstemp(suffix=".wav")
            os.close(fd)
            subprocess.run(["ffmpeg", "-y", "-i", in_path, wav_path], check=True, capture_output=True, timeout=60)
            
        try:
            arr, sr = sf.read(wav_path)
            audio_arrays.append(arr)
            samplerate = sr
        except Exception:
            pass
            
        try: os.unlink(in_path)
        except: pass
        if is_mp3:
            try: os.unlink(wav_path)
            except: pass

    if not audio_arrays:
        return b""

    result = audio_arrays[0]
    fade_len = int(samplerate * 0.075)
    silence_len = int(samplerate * 0.15)
    silence = np.zeros(silence_len, dtype=result.dtype)

    for next_arr in audio_arrays[1:]:
        result = np.concatenate([result, silence])
        if len(result) > fade_len and len(next_arr) > fade_len:
            fade_out = np.linspace(1.0, 0.0, fade_len)
            fade_in = np.linspace(0.0, 1.0, fade_len)
            
            if len(result.shape) > 1:
                fade_out = fade_out[:, np.newaxis]
                fade_in = fade_in[:, np.newaxis]
                
            overlap_result = result[-fade_len:] * fade_out
            overlap_next = next_arr[:fade_len] * fade_in
            
            result[-fade_len:] = overlap_result + overlap_next
            result = np.concatenate([result, next_arr[fade_len:]])
        else:
            result = np.concatenate([result, next_arr])

    fd, out_wav = tempfile.mkstemp(suffix=".wav")
    os.close(fd)
    sf.write(out_wav, result, samplerate)

    if is_mp3:
        fd, out_mp3 = tempfile.mkstemp(suffix=".mp3")
        os.close(fd)
        subprocess.run(["ffmpeg", "-y", "-i", out_wav, out_mp3], check=True, capture_output=True, timeout=60)
        with open(out_mp3, "rb") as f:
            res = f.read()
        os.unlink(out_mp3)
        os.unlink(out_wav)
        return res
    else:
        with open(out_wav, "rb") as f:
            res = f.read()
        os.unlink(out_wav)
        return res

def _ffmpeg_concat_wav(parts: list) -> bytes:
    return _numpy_concat(parts, False)


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")
    # Only explicitly tested models are permitted.
    if voice_code not in PIPER_VOICES:
        raise ValueError("Unsupported Piper voice")
    onnx_path, json_path = download_piper_model(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 sanitize_text(text: str) -> str:
    import re
    text = re.sub(r'[\u200B-\u200D\uFEFF]', '', text)
    text = re.sub(r'[\x00-\x08\x0b-\x1f\x7f]', '', text)
    text = re.sub(r'[ \t]+', ' ', text)
    text = re.sub(r'\n\s*\n+', '\n\n', text)
    return text.strip()


_EDGE_RATE_RE = re.compile(r'^[+-]?(?:\d+(?:\.\d+)?%|\d+(?:\.\d+)?x)$')
_EDGE_VOLUME_RE = re.compile(r'^[+-]?\d+(?:\.\d+)?%$')
_EDGE_PITCH_RE = re.compile(r'^[+-]?\d+(?:\.\d+)?(?:Hz|st)?$')


def validate_edge_prosody(rate: str, volume: str, pitch: str):
    """Reject anything that could break out of the generated SSML prosody tag."""
    if not _EDGE_RATE_RE.match(str(rate or "").strip()):
        raise ValueError("Invalid rate. Use a value like +0%, -25% or 1.2x.")
    if not _EDGE_VOLUME_RE.match(str(volume or "").strip()):
        raise ValueError("Invalid volume. Use a value like +0% or -50%.")
    if not _EDGE_PITCH_RE.match(str(pitch or "").strip()):
        raise ValueError("Invalid pitch. Use a value like +0Hz or -2st.")

def split_text(text: str, max_chars: int = 1400) -> list:
    """Text ko chunklara bol"""
    text = sanitize_text(text)
    if not text:
        return []
    sentence_re = re.compile(
        r'(?<=[.!?\u0964\u06D4\u061F\u2026\u3002\uff01\uff1f])\s+(?=\S)'
    )
    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:
                    if len(word) > max_chars:
                        if buf:
                            chunks.append(buf)
                            buf = ""
                        for sub_idx in range(0, len(word), max_chars):
                            chunks.append(word[sub_idx:sub_idx + max_chars])
                        continue
                    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"[DOWNLOAD] Downloading Silero v4: {url[:80]}...")
            urllib.request.urlretrieve(url, model_path)
            if os.path.getsize(model_path) > 100000:
                print(f"[OK] Silero v4 downloaded ({os.path.getsize(model_path)//1024}KB)")
                return model_path
            os.remove(model_path)
        except Exception as e:
            print(f"[WARN] 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 (non-critical): {e}")



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:
    return _numpy_concat(parts, True)


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
    import asyncio
    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:
                if voice != "en-US-AriaNeural":
                    try:
                        comm = edge_tts.Communicate(chunk, "en-US-AriaNeural", **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:
                        pass
                return b""
            await asyncio.sleep(2)
    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_GENERATION_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
_POCKET_TORCH_PATCHED = False


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 _patch_torch_for_pocket_tts():
    """Pocket TTS mutates a streaming KV cache; no_grad keeps tensors mutable."""
    global _POCKET_TORCH_PATCHED
    if _POCKET_TORCH_PATCHED:
        return
    try:
        import torch
        torch.inference_mode = torch.no_grad
        _POCKET_TORCH_PATCHED = True
        logging.info("Pocket TTS torch compatibility patch enabled")
    except Exception as exc:
        logging.warning("Pocket TTS torch patch skipped: %s", exc.__class__.__name__)


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:
            _patch_torch_for_pocket_tts()
            from pocket_tts import TTSModel
            _POCKET_MODEL = TTSModel.load_model()
            # Pocket TTS is inference-only in this service. Disabling gradient
            # tracking reduces CPU work and memory without changing synthesis.
            try:
                import torch
                torch.set_grad_enabled(False)
            except Exception:
                pass
            _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




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]}"
            )
        
        try:
            import soundfile as sf
            import numpy as np
            data, sr = sf.read(normalized_path)
            if len(data.shape) > 1:
                data = data.mean(axis=1)
            
            sr = sr or 24000
            duration = len(data) / float(sr)
            if len(data) == 0 or duration < 0.3:
                raise ValueError("Reference audio snippet is too short. Please upload a voice sample of at least 1-3 seconds.")
            
            # Auto-trim if longer than 12s, auto-tile if shorter than 3s
            if duration > 12.0:
                data = data[:int(12.0 * sr)]
            elif duration < 3.0:
                repeats = int(np.ceil(3.0 / duration))
                data = np.tile(data, repeats)[:int(3.0 * sr)]
            
            rms = np.sqrt(np.mean(data**2))
            if rms < 0.0005:
                raise ValueError("Reference audio is mostly silence. Please record a clearer voice sample.")
            
            target_rms = 10 ** (-20 / 20)
            data = data * (target_rms / (rms + 1e-9))
            
            threshold = 10 ** (-40 / 20)
            mask = np.abs(data) > threshold
            if np.any(mask):
                start = np.argmax(mask)
                end = len(mask) - np.argmax(mask[::-1])
                data = data[start:end]
            
            sf.write(normalized_path, data, sr)
        except ValueError:
            raise
        except Exception:
            pass
        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:
        audio = _POCKET_RESULT_CACHE.get(cache_key)
        if audio is not None:
            try:
                _POCKET_RESULT_CACHE_ORDER.remove(cache_key)
            except ValueError:
                pass
            _POCKET_RESULT_CACHE_ORDER.append(cache_key)
        return audio


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:
        voice_state = _POCKET_VOICE_STATE_CACHE.get(reference_key)
        if voice_state is not None:
            try:
                _POCKET_VOICE_STATE_ORDER.remove(reference_key)
            except ValueError:
                pass
            _POCKET_VOICE_STATE_ORDER.append(reference_key)
        return voice_state


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)


def _estimate_min_clone_seconds(text: str) -> float:
    words = len(re.findall(r"\S+", str(text or "")))
    # A very conservative floor; natural speech can be fast, but a much shorter
    # result usually means the model stopped early and skipped words.
    return max(0.45, min(20.0, words * 0.105))


def _generate_pocket_chunk(model, voice_state, chunk: str):
    import numpy as np

    audio = model.generate_audio(voice_state, chunk)
    if hasattr(audio, "detach"):
        audio = audio.detach().cpu().numpy()
    return np.asarray(audio).squeeze().reshape(-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)
        try:
            import torch
            inference_context = torch.no_grad()
        except Exception:
            inference_context = nullcontext()
        with inference_context:
            for chunk in chunks:
                audio_np = _generate_pocket_chunk(model, voice_state, chunk)
                duration = audio_np.size / float(getattr(model, "sample_rate", 24000) or 24000)
                if (
                    audio_np.size
                    and len(chunk) > CLONE_RETRY_CHUNK_CHARACTERS
                    and duration < _estimate_min_clone_seconds(chunk)
                ):
                    logging.warning("Clone chunk looked truncated; retrying with smaller chunks")
                    smaller_parts = []
                    for sub_chunk in split_text(chunk, max_chars=CLONE_RETRY_CHUNK_CHARACTERS):
                        sub_audio = _generate_pocket_chunk(model, voice_state, sub_chunk)
                        if sub_audio.size:
                            smaller_parts.append(sub_audio)
                    if smaller_parts:
                        silence = np.zeros(int(model.sample_rate * 0.08), dtype=smaller_parts[0].dtype)
                        joined = []
                        for index, part in enumerate(smaller_parts):
                            if index:
                                joined.append(silence)
                            joined.append(part)
                        audio_np = np.concatenate(joined)
                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:
    global _POCKET_GENERATION_ERROR
    if not reference_path or not os.path.exists(reference_path):
        raise ValueError("A reference audio file is required for voice cloning")
    try:
        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)
        _POCKET_GENERATION_ERROR = None
        return audio
    except Exception as exc:
        _POCKET_GENERATION_ERROR = exc.__class__.__name__
        logging.exception("Voice clone generation failed")
        raise RuntimeError("Voice cloning failed on this server. Please contact support.") from exc

def post_process_bytes(audio: bytes, media_type: str) -> bytes:
    """Apply gentle, smooth speech cleanup without block-gate clicks.

    A single ffmpeg pass is faster than decoding into Python/SciPy, walking
    every 20 ms window, writing a WAV, and encoding again. The conservative
    denoise amount removes low-level hiss while preserving voice character.
    """
    suffix = ".mp3" if "mpeg" in media_type else ".wav"
    fd, in_path = tempfile.mkstemp(suffix=suffix)
    os.close(fd)
    with open(in_path, "wb") as f:
        f.write(audio)
    fd, out_path = tempfile.mkstemp(suffix=suffix)
    os.close(fd)
    try:
        ffmpeg = shutil.which("ffmpeg")
        if not ffmpeg:
            return audio
        filters = (
            "highpass=f=55,"
            "afftdn=nr=4:nf=-55:tn=1,"
            "equalizer=f=8000:width_type=h:width=1200:g=1.2,"
            "alimiter=limit=0.891:attack=5:release=50"
        )
        cmd = [
            ffmpeg, "-y", "-hide_banner", "-loglevel", "error",
            "-i", in_path, "-vn", "-af", filters,
        ]
        if suffix == ".mp3":
            # Explicit high-quality encoding avoids ffmpeg's lower default
            # bitrate and prevents an avoidable quality drop.
            cmd += ["-codec:a", "libmp3lame", "-b:a", "320k", out_path]
        else:
            cmd += ["-acodec", "pcm_s16le", out_path]
        result = subprocess.run(
            cmd,
            capture_output=True,
            timeout=300,
            creationflags=CREATE_NO_WINDOW,
        )
        if result.returncode != 0 or not os.path.exists(out_path) or os.path.getsize(out_path) <= 44:
            logging.warning("Audio cleanup failed with ffmpeg exit code %s", result.returncode)
            return audio
        with open(out_path, "rb") as output_file:
            return output_file.read()
    except Exception as exc:
        logging.warning("Audio cleanup failed: %s", exc.__class__.__name__)
        return audio
    finally:
        for p in (in_path, out_path):
            if os.path.exists(p):
                try:
                    os.unlink(p)
                except OSError:
                    pass

@app.post("/tts")
async def tts_endpoint(
    request: Request,
    engine:  str = Form(...),   # "edge" | "piper" | "silero" | "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
    post_process: bool = Form(True),         # audio post-processing toggle
    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()
    if engine not in available_engines():
        return JSONResponse(status_code=400, content={"error": "Unsupported TTS engine"})

    has_reference = bool(reference_audio is not None and reference_audio.filename)
    if has_reference and engine not in CLONE_ENGINES:
        return JSONResponse(
            status_code=400,
            content={"error": "Reference audio is only accepted for voice cloning"},
        )
    if engine in CLONE_ENGINES and not has_reference:
        return JSONResponse(
            status_code=400,
            content={"error": "Voice cloning requires reference audio"},
        )
    if engine == "piper" and voice not in PIPER_VOICES:
        return JSONResponse(status_code=400, content={"error": "Unsupported Piper voice"})
    if engine == "silero" and not str(voice).startswith("silero:ru_"):
        return JSONResponse(status_code=400, content={"error": "Unsupported Silero voice"})
    if engine == "edge" and not re.match(r"^[a-z]{2,3}-[A-Z]{2}-.+Neural$", str(voice)):
        return JSONResponse(status_code=400, content={"error": "Unsupported Edge voice"})
    if engine == "edge":
        try:
            validate_edge_prosody(rate, volume, pitch)
        except ValueError as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})

    # These two checks run BEFORE authorize_request on purpose. Authorization
    # books the requested characters against the monthly allowance, and only
    # the try/finally further down gives them back. Rejecting here afterwards
    # would leak the reservation and silently bill the customer for a request
    # that never ran.
    if engine in CLONE_ENGINES and not clone_backend_available():
        return JSONResponse(
            status_code=403,
            content={"error": "Voice cloning is not available on this server"},
        )
    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)"
                )
            },
        )

    auth_context = await authorize_request(request, engine, len(text))

    reference_path = None
    started_at = time.monotonic()
    generation_status = "error"
    slot_acquired = False
    try:
        acquire_worker_slot()
        slot_acquired = True
        if has_reference:
            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"

        else:
            return JSONResponse(status_code=400, content={"error": f"Unknown engine: {engine}"})

        # Preserve clone identity and improve speed: Pocket TTS already emits
        # clean PCM. Re-filtering cloned audio can subtly change timbre and
        # adds a full ffmpeg pass. Normal TTS still receives gentle cleanup.
        if post_process and engine not in CLONE_ENGINES:
            audio = await asyncio.to_thread(post_process_bytes, audio, media)

        if not audio or len(audio) == 0:
            return JSONResponse(
                status_code=502,
                content={"error": "Audio generation failed or produced empty audio. Please retry."},
            )

        generation_status = "success"
        return StreamingResponse(
            io.BytesIO(audio),
            media_type=media,
            headers={"Content-Disposition": f"attachment; filename={fname}"},
        )

    except HTTPException:
        # "Worker is busy; try another Space" (503) is raised by
        # acquire_worker_slot. Swallowing it into a generic 500 stopped the
        # desktop client from failing over to the next Space in the pool.
        raise
    except Exception as e:
        logging.error(f"TTS error: {e}", exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"error": "Audio generation failed. Please retry shortly."},
        )
    finally:
        await asyncio.to_thread(
            record_usage,
            auth_context,
            engine,
            len(text),
            generation_status,
            int((time.monotonic() - started_at) * 1000),
        )
        if slot_acquired:
            release_worker_slot()
        try:
            if reference_path and os.path.exists(reference_path):
                os.unlink(reference_path)
        except Exception:
            pass



from fastapi import Body

@app.post("/register_user")
async def register_user(request: Request, body: dict = Body(...)):
    db = _require_firebase_db()
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing authentication token")
    token = auth_header.split("Bearer ", 1)[1]
    try:
        decoded = fb_auth.verify_id_token(token)
    except Exception:
        raise HTTPException(401, "Invalid token")
    
    uid = decoded["uid"]
    user_ref = db.collection("users").document(uid)
    user_doc = user_ref.get()
    
    data = {
        "email": decoded.get("email", ""),
        "display_name": body.get("display_name", ""),
        "updated_at": firestore.SERVER_TIMESTAMP
    }
    device_id = str(body.get("device_id", "")).strip()[:128]
    
    if not user_doc.exists:
        data["created_at"] = firestore.SERVER_TIMESTAMP
        data["is_blocked"] = False
        data["device_ids"] = [device_id] if device_id else []
        user_ref.set(data)
    else:
        existing = user_doc.to_dict() or {}
        license_key = existing.get("license_key")
        if device_id:
            if not license_key:
                # Keep only the current pre-activation device. Otherwise an
                # unlicensed account can accumulate stale IDs and later fail
                # activation against a one-device license.
                data["device_ids"] = [device_id]
            else:
                license_doc = db.collection("licenses").document(license_key).get()
                lic = license_doc.to_dict() if license_doc.exists else {}
                max_devices = max(1, int(lic.get("max_devices") or 1))
                _register_device_tx(db, user_ref, device_id, max_devices)
        user_ref.update(data)
        
    return {"status": "success", "uid": uid}

@app.post("/activate_license")
async def activate_license(request: Request, body: dict = Body(...)):
    db = _require_firebase_db()
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing token")
    token = auth_header.split("Bearer ", 1)[1]
    try:
        decoded = fb_auth.verify_id_token(token)
    except Exception:
        raise HTTPException(401, "Invalid token")
    
    uid = decoded["uid"]
    license_key = str(body.get("license_key") or "").strip().upper()
    if not license_key:
        raise HTTPException(400, "license_key is required")
    if not re.fullmatch(r"[A-Z0-9][A-Z0-9_-]{5,79}", license_key):
        raise HTTPException(400, "License key format is invalid")

    license_ref = db.collection("licenses").document(license_key)
    license_doc = license_ref.get()
    
    if not license_doc.exists:
        raise HTTPException(404, "License key not found")
        
    lic = license_doc.to_dict()
    current_status = lic.get("status")
    if current_status not in ("unused", "unclaimed", "active"):
        raise HTTPException(400, f"License status is {current_status}")
        
    if lic.get("assigned_uid") and lic.get("assigned_uid") != uid:
        raise HTTPException(400, "License is already assigned to another user")
        
    from datetime import datetime, timedelta, timezone
    
    duration_days = lic.get("duration_days", 30)
    expiry_date = lic.get("expiry_date")
    device_id = str(request.headers.get("X-Device-ID", "")).strip()[:128]
    user_ref = db.collection("users").document(uid)
    @firestore.transactional
    def claim_license(transaction):
        fresh_snapshot = license_ref.get(transaction=transaction)
        if not fresh_snapshot.exists:
            raise HTTPException(404, "License key not found")
        fresh = fresh_snapshot.to_dict() or {}
        fresh_status = fresh.get("status")
        if fresh_status not in ("unused", "unclaimed", "active"):
            raise HTTPException(400, f"License status is {fresh_status}")
        assigned_uid = fresh.get("assigned_uid")
        if assigned_uid and assigned_uid != uid:
            raise HTTPException(400, "License is already assigned to another user")
        raw_claimed_expiry = fresh.get("expiry_date")
        updates = {
            "status": "active",
            "assigned_uid": uid,
            "assigned_email": decoded.get("email", ""),
        }
        if raw_claimed_expiry in (None, ""):
            claimed_expiry = datetime.now(timezone.utc) + timedelta(
                days=int(fresh.get("duration_days") or duration_days)
            )
            updates["activated_at"] = firestore.SERVER_TIMESTAMP
            updates["expiry_date"] = claimed_expiry
        else:
            claimed_expiry = _coerce_utc_datetime(raw_claimed_expiry)
            if claimed_expiry is None:
                raise HTTPException(400, "License expiry is invalid. Please contact support")
            if claimed_expiry < datetime.now(timezone.utc):
                raise HTTPException(400, "License has expired")
            # Normalize imported ISO strings back to a real Firestore timestamp.
            if isinstance(raw_claimed_expiry, str):
                updates["expiry_date"] = claimed_expiry
        fresh_user_snapshot = user_ref.get(transaction=transaction)
        fresh_user = fresh_user_snapshot.to_dict() if fresh_user_snapshot.exists else {}
        device_ids = list(fresh_user.get("device_ids") or [])
        max_devices = max(1, int(fresh.get("max_devices") or 1))
        if len(device_ids) > max_devices:
            raise HTTPException(403, "Maximum licensed devices reached")
        if device_id and device_id not in device_ids:
            if len(device_ids) >= max_devices:
                raise HTTPException(403, "Maximum licensed devices reached")
            device_ids.append(device_id)
        user_clone_setting = (
            bool(fresh_user.get("voice_clone_enabled"))
            if "voice_clone_enabled" in fresh_user
            else bool(fresh.get("voice_clone", False))
        )
        transaction.update(license_ref, updates)
        transaction.set(user_ref, {
            "license_key": license_key,
            "license_status": "active",
            "license_expiry": claimed_expiry,
            "voice_clone_enabled": user_clone_setting,
            "device_ids": device_ids,
            "updated_at": firestore.SERVER_TIMESTAMP,
        }, merge=True)
        return claimed_expiry, user_clone_setting, fresh, dict(fresh_user, device_ids=device_ids)

    expiry_date, user_clone_setting, lic, user_data = claim_license(db.transaction())
    
    # User-specific pool: dedicated Spaces first, otherwise public auto pool.
    space_pool = _license_space_pool(db, uid, user_data, license_key, lic)
    credit_summary = _monthly_credit_summary(db, uid, lic)

    return {
        "status": "success",
        "plan_name": lic.get("plan_name", "Standard"),
        "expiry_date": str(expiry_date) if expiry_date else None,
        "voice_clone_enabled": user_clone_setting and bool(lic.get("voice_clone", False)),
        "app_link": lic.get("app_link") or lic.get("appLink", ""),
        "space_pool": space_pool,
        "space_pool_exclusive": _space_pool_is_exclusive(user_data, lic),
        **credit_summary,
    }

@app.get("/license_status")
async def license_status(request: Request):
    db = _require_firebase_db()
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing token")
    token = auth_header.split("Bearer ", 1)[1]
    try:
        decoded = fb_auth.verify_id_token(token)
    except Exception:
        raise HTTPException(401, "Invalid token")
        
    uid = decoded["uid"]
    user_doc = db.collection("users").document(uid).get()
    
    if not user_doc.exists:
        return {"has_license": False, "status": "none"}
        
    user_data = user_doc.to_dict()
    if user_data.get("is_blocked"):
        return {"has_license": False, "status": "blocked"}

    license_key = user_data.get("license_key")
    if not license_key:
        return {"has_license": False, "status": "none"}
        
    license_doc = db.collection("licenses").document(license_key).get()
    if not license_doc.exists:
        return {"has_license": False, "status": "none"}
        
    lic = license_doc.to_dict()
    lic_status = lic.get("status", "active")
    if lic_status != "active":
        return {"has_license": False, "status": lic_status}

    from datetime import datetime, timezone
    raw_expiry_date = lic.get("expiry_date")
    expiry_date = _coerce_utc_datetime(raw_expiry_date)
    if raw_expiry_date not in (None, "") and expiry_date is None:
        return {"has_license": False, "status": "invalid_expiry", "is_expired": False}
    days_remaining = 30
    if expiry_date:
        delta = expiry_date - datetime.now(timezone.utc)
        days_remaining = delta.days
        if delta.total_seconds() < 0:
            return {"has_license": False, "status": "expired", "is_expired": True}

    # User-specific pool: dedicated Spaces first, otherwise public auto pool.
    space_pool = _license_space_pool(db, uid, user_data, license_key, lic)
    credit_summary = _monthly_credit_summary(db, uid, lic)

    return {
        "has_license": True,
        "status": "active",
        "license_key": license_key,
        "plan_name": lic.get("plan_name", "Pro"),
        "expiry_date": str(expiry_date) if expiry_date else None,
        "days_remaining": max(0, days_remaining),
        "voice_clone_enabled": bool(user_data.get("voice_clone_enabled", lic.get("voice_clone", False))) and bool(lic.get("voice_clone", False)),
        "app_link": lic.get("app_link") or lic.get("appLink", ""),
        "space_pool": space_pool,
        "space_pool_exclusive": _space_pool_is_exclusive(user_data, lic),
        "is_expired": False,
        **credit_summary,
    }

@app.get("/status")
def status():
    return {"status": "ok"}


@app.get("/health")
@app.head("/health")
def health():
    with _SPACE_STATE_LOCK:
        active_jobs = _ACTIVE_JOBS
    policy = get_runtime_policy()
    clone_status = clone_backend_status()
    firebase_configured = bool(HAS_FIREBASE and _init_firebase())
    if AUTH_MODE == "api_secret":
        auth_ready = bool(API_SECRET)
    elif AUTH_MODE == "layered":
        auth_ready = firebase_configured and bool(API_SECRET)
    else:
        auth_ready = firebase_configured
    worker_available = (
        auth_ready
        and active_jobs < MAX_CONCURRENT_JOBS
        and bool(policy.get("space_enabled", True))
        and not bool(policy.get("maintenance_mode", False))
    )
    return {
        "status": "ok" if auth_ready else "degraded",
        "space_id": SPACE_ID,
        "firebase_configured": firebase_configured,
        "auth_ready": auth_ready,
        "active_jobs": active_jobs,
        "max_concurrent_jobs": MAX_CONCURRENT_JOBS,
        "available": worker_available,
        "clone_enabled": bool(policy.get("space_clone_enabled", False)),
        "clone_ready": bool(clone_status.get("ready", False)),
        "clone_status": clone_status.get("message", "unknown"),
        "engines": BASE_ENGINES + (CLONE_ENGINES if policy.get("space_clone_enabled", False) else []),
        "maintenance_mode": bool(policy.get("maintenance_mode", False)),
        "version": SERVICE_VERSION,
    }


@app.get("/all_voices")
async def all_voices_list(request: Request):
    """All configured Edge, Piper, Silero, and CPU voice-cloning voices."""
    await authorize_request(request, "edge")
    result = {"edge": {}, "piper": {}, "silero": {}}

    # Current Edge catalog. Desktop defaults to a smaller Featured view.
    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}")

    # Curated Piper catalog only; arbitrary repository models are not exposed.
    piper_display_names = {
        "piper:en_US-amy-medium": "Amy [en-US]",
        "piper:en_US-joe-medium": "Joe [en-US]",
        "piper:en_US-lessac-medium": "Lessac HQ [en-US]",
        "piper:en_US-ryan-high": "Ryan HQ [en-US]",
        "piper:en_GB-alan-medium": "Alan [en-GB]",
        "piper:en_GB-alba-medium": "Alba [en-GB]",
        "piper:ur_PK-fasih-medium": "Fasih [ur-PK]",
        "piper:ar_JO-kareem-medium": "Kareem [ar-JO]",
        "piper:hi_IN-pratham-medium": "Pratham [hi-IN]",
        "piper:de_DE-thorsten-medium": "Thorsten [de-DE]",
        "piper:ru_RU-irina-medium": "Irina [ru-RU]",
        "piper:fr_FR-upmc-medium": "UPMC HQ [fr-FR]",
        "piper:pt_BR-faber-medium": "Faber [pt-BR]",
        "piper:tr_TR-dfki-medium": "Dfki [tr-TR]",
        "piper:nl_NL-mls-medium": "MLS [nl-NL]",
    }
    for code in PIPER_VOICES:
        result["piper"][piper_display_names[code]] = code

    # 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}"

    if not clone_backend_available():
        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}


@app.get("/")
def root():
    return {"name": "VoiceCraft Service", "status": "ok"}


if __name__ == "__main__":
    import uvicorn

    port = int(os.environ.get("PORT") or os.environ.get("GRADIO_SERVER_PORT") or 7860)
    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")