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

try:
    import spaces
except ImportError:
    class _SpacesFallback:
        """Keep the shared backend importable outside Hugging Face Spaces."""

        @staticmethod
        def GPU(function=None, **_kwargs):
            def decorator(fn):
                return fn

            return decorator(function) if callable(function) else decorator

    spaces = _SpacesFallback()

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 gradio as gr
import requests

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

SERVICE_VERSION = os.environ.get("VOICECRAFT_SERVICE_VERSION", "3.1.0").strip()

app = gr.Server(
    debug=False,
    title="VoiceCraft TTS Server",
    description="VoiceCraft desktop service.",
    version=SERVICE_VERSION,
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
    enable_monitoring=False,
)
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)
)
DEFAULT_LICENSE_VALIDATION_URL = os.environ.get("DEFAULT_LICENSE_VALIDATION_URL", "").strip()
LICENSE_VALIDATION_URL = (
    os.environ.get("LICENSE_VALIDATION_URL", "").strip()
    or DEFAULT_LICENSE_VALIDATION_URL
)
AUTH_MODE = os.environ.get("VOICECRAFT_AUTH_MODE", "license").strip().lower()
if AUTH_MODE not in {"license", "layered", "api_secret"}:
    AUTH_MODE = "license"
MIN_CLIENT_VERSION = os.environ.get("MIN_CLIENT_VERSION", "3.0.0").strip()
MAX_CLONE_CHARACTERS = 60_000
CLONE_CHUNK_CHARACTERS = 1800
CLONE_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()
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,
        "clone_enabled": bool((item or {}).get("clone_enabled", (item or {}).get("clone", True))),
        "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."""
    if db is None:
        return []

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

    for source in (user_data or {}, lic or {}):
        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 value in source.get("assigned_tts_spaces") or []:
            url = _clean_hf_space_url(value) or str(value).strip()
            if url: assigned_tts_urls.add(url)
        for value in source.get("assigned_clone_spaces") or []:
            url = _clean_hf_space_url(value) or str(value).strip()
            if url: assigned_clone_urls.add(url)

    dedicated = []
    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. Never
            # return those endpoints to a customer during activation/status;
            # otherwise the desktop wastes time failing over from known-dead
            # Spaces. A newly added worker with no health result is still
            # eligible until the first health sweep.
            if item.get("last_health_ok") is False:
                continue
            payload = _space_public_payload(snapshot.id, item)
            if not payload:
                continue
            is_dedicated = (
                snapshot.id in assigned_ids
                or payload["url"] in assigned_urls
                or 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()
            )
            if is_dedicated:
                dedicated.append(payload)
            elif not item.get("assigned_uid") and not item.get("assigned_license_key") and not item.get("assigned_license"):
                auto_pool.append(payload)
    except Exception:
        return []

    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 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 _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 _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 verify_firebase_auth(request, is_clone: bool = False, requested_characters: int = 0):
    if not HAS_FIREBASE or not fb_auth:
        raise HTTPException(503, "Firebase authentication is not configured")
    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 = _init_firebase()
    if db is None:
        raise HTTPException(503, "Firebase database is unavailable")
    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
    if lic.get("expiry_date") and lic["expiry_date"].replace(tzinfo=timezone.utc) < 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")
    _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": {}}
        
    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,
    )
    _enforce_rate_limit(uid, is_clone)
    return {"uid": uid, "license_key": license_key, "license": lic}


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:
            db.collection("spaces").document(SPACE_ID).set({
                "active_jobs": active,
                "last_heartbeat": firestore.SERVER_TIMESTAMP,
            }, 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:
            db.collection("spaces").document(SPACE_ID).set({
                "active_jobs": active,
                "last_heartbeat": firestore.SERVER_TIMESTAMP,
            }, 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)
        batch.set(monthly_ref, {
            "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),
            reserved_field: firestore.Increment(-int(characters)),
            "generation_count": firestore.Increment(1),
            "updated_at": firestore.SERVER_TIMESTAMP,
        }, 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__)



# ══════════════════════════════════════════════════════════════════
# 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("✅ Piper binary ready")

        PIPER_READY = True
        print("✅ Piper TTS ready")
    except Exception as e:
        print(f"⚠️ Piper setup failed (non-critical): {e}")
        PIPER_READY = False

threading.Thread(target=setup_piper, daemon=True).start()


def download_piper_model(voice_code: str) -> tuple:
    """Model yoksa indir, path tuple dondur (onnx, json)"""
    if voice_code not in PIPER_VOICES:
        raise ValueError(f"Unknown Piper voice: {voice_code}")
    onnx_file, json_file = PIPER_VOICES[voice_code]
    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
    json_path  = os.path.join(PIPER_MODELS_DIR, json_file)

    import urllib.request
    # Build correct HF path: en/en_US/amy/medium/en_US-amy-medium.onnx
    parts = onnx_file.rsplit("-", 2)
    lang_code = parts[0]  # en_US
    voice     = parts[1]  # amy
    quality   = parts[2].replace(".onnx", "")  # medium
    lang_short = lang_code.split("_")[0]  # en
    hf_dir = f"{lang_short}/{lang_code}/{voice}/{quality}"

    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
        if not os.path.exists(fpath):
            url = f"{PIPER_BASE_URL}/{hf_dir}/{fname}"
            print(f"📥 Downloading Piper model: {fname}")
            try:
                urllib.request.urlretrieve(url, fpath)
            except Exception:
                url2 = f"{PIPER_BASE_URL}/{lang_short}/{lang_code}/{fname}"
                urllib.request.urlretrieve(url2, fpath)
    return onnx_path, json_path


def _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"📥 Downloading Silero v4: {url[:80]}...")
            urllib.request.urlretrieve(url, model_path)
            if os.path.getsize(model_path) > 100000:
                print(f"✅ Silero v4 downloaded ({os.path.getsize(model_path)//1024}KB)")
                return model_path
            os.remove(model_path)
        except Exception as e:
            print(f"⚠️ Download failed: {e}")
            try:
                os.remove(model_path)
            except Exception:
                pass
    return ""


def setup_silero():
    global SILERO_READY
    try:
        import torch
        model_path = download_silero_model()
        if model_path:
            model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
            SILERO_MODELS["ru"] = model
            SILERO_READY = True
            print("Silero TTS ready (v4 Russian - 5 speakers)")
        else:
            print("Silero TTS: model download failed")
    except Exception as e:
        print(f"[Silero] setup failed (non-critical): {e}")

threading.Thread(target=setup_silero, daemon=True).start()


def synthesize_silero(text: str, voice_code: str) -> bytes:
    """Silero TTS — code: silero:ru_xenia. v4 model via torch.package.
    Model lazily load hota hai (self-heal) agar startup thread fail hua ho."""
    import numpy as np, scipy.io.wavfile as wav
    try:
        import torch
    except ImportError:
        raise Exception("Silero TTS requires torch.")

    lang_speaker = voice_code.replace("silero:", "")
    lang = lang_speaker.split("_")[0]
    speaker = lang_speaker.split("_", 1)[1] if "_" in lang_speaker else lang_speaker

    # Validate speaker — only real v4 speakers allowed
    valid_speakers = set(SILERO_SPEAKERS_RU)
    if speaker not in valid_speakers:
        raise Exception(f"Invalid Silero speaker '{speaker}'. Valid: {', '.join(valid_speakers)}")

    # Model on-demand load (self-heals if startup thread failed / was slow)
    if lang not in SILERO_MODELS:
        with SILERO_LOAD_LOCK:
            if lang not in SILERO_MODELS:
                model_path = download_silero_model()
                if not model_path:
                    raise Exception("Silero v4 model could not be downloaded (check network / model URL).")
                model = torch.package.PackageImporter(model_path).load_pickle("tts_models", "model")
                SILERO_MODELS[lang] = model
                global SILERO_READY
                SILERO_READY = True

    model = SILERO_MODELS[lang]
    audio = model.apply_tts(text=text, speaker=speaker, sample_rate=SILERO_SAMPLE_RATE)
    audio_np = audio.numpy() if hasattr(audio, "numpy") else audio.cpu().detach().numpy()

    buf = io.BytesIO()
    wav.write(buf, SILERO_SAMPLE_RATE, (audio_np * 32767).astype(np.int16))
    buf.seek(0)
    return buf.read()


def _ffmpeg_concat_mp3(parts: list) -> bytes:
    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


threading.Thread(target=_warm_pocket_model, daemon=True).start()


def _reference_audio_key(reference_path: str) -> str:
    digest = hashlib.sha256()
    with open(reference_path, "rb") as reference_file:
        for chunk in iter(lambda: reference_file.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _normalize_clone_reference(reference_path: str) -> str:
    """Decode any accepted upload into a mono PCM WAV Pocket TTS can read."""
    if not reference_path or not os.path.exists(reference_path):
        raise ValueError("A reference audio file is required for voice cloning")
    ffmpeg = shutil.which("ffmpeg")
    if not ffmpeg:
        raise RuntimeError("ffmpeg is required to decode voice clone reference audio")
    fd, normalized_path = tempfile.mkstemp(suffix=".wav")
    os.close(fd)
    try:
        result = subprocess.run(
            [
                ffmpeg,
                "-y",
                "-hide_banner",
                "-loglevel",
                "error",
                "-i",
                reference_path,
                "-map",
                "0:a:0",
                "-t",
                "12",
                "-vn",
                "-sn",
                "-dn",
                "-acodec",
                "pcm_s16le",
                "-ac",
                "1",
                "-ar",
                "24000",
                "-f",
                "wav",
                normalized_path,
            ],
            capture_output=True,
            timeout=60,
            creationflags=CREATE_NO_WINDOW,
        )
        if result.returncode != 0 or os.path.getsize(normalized_path) <= 44:
            detail = result.stderr.decode("utf-8", errors="replace").strip()
            raise ValueError(
                "Reference audio could not be decoded. "
                f"Use a clear WAV, MP3, M4A, OGG, or FLAC file. {detail[:160]}"
            )
        
        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)


@app.api(name="voicecraft_clone_backend", api_visibility="private", concurrency_limit=1)
def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) -> bytes:
    if not reference_path or not os.path.exists(reference_path):
        raise ValueError("A reference audio file is required for voice cloning")
    model = _load_pocket_model()
    if not getattr(model, "has_voice_cloning", False):
        raise RuntimeError(_pocket_clone_auth_error())
    import numpy as np

    voice_state = _get_cached_voice_state(reference_key)
    audio_parts = []
    with _POCKET_INFER_LOCK:
        if voice_state is None:
            try:
                voice_state = model.get_state_for_audio_prompt(reference_path)
            except Exception as exc:
                message = str(exc)
                if "could not download the weights" in message.lower() or "voice cloning" in message.lower():
                    raise RuntimeError(_pocket_clone_auth_error()) from exc
                raise
            _cache_voice_state(reference_key, voice_state)
        chunks = split_text(text, max_chars=CLONE_CHUNK_CHARACTERS)
        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)})

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

    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)"
                )
            },
        )

    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 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(...)):
    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"]
    db = _init_firebase()
    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:
            existing_devices = list(existing.get("device_ids") or [])
            if not license_key:
                data["device_ids"] = firestore.ArrayUnion([device_id])
            elif device_id not in existing_devices:
                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))
                if len(existing_devices) >= max_devices:
                    raise HTTPException(403, "Maximum licensed devices reached")
                data["device_ids"] = firestore.ArrayUnion([device_id])
        user_ref.update(data)
        
    return {"status": "success", "uid": uid}

@app.post("/activate_license")
async def activate_license(request: Request, body: dict = Body(...)):
    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 = body.get("license_key")
    if not license_key:
        raise HTTPException(400, "license_key is required")
        
    db = _init_firebase()
    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")
        claimed_expiry = fresh.get("expiry_date")
        updates = {
            "status": "active",
            "assigned_uid": uid,
            "assigned_email": decoded.get("email", ""),
        }
        if not claimed_expiry:
            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:
            exp_dt = (
                claimed_expiry
                if getattr(claimed_expiry, "tzinfo", None)
                else claimed_expiry.replace(tzinfo=timezone.utc)
            )
            if exp_dt < datetime.now(timezone.utc):
                raise HTTPException(400, "License has expired")
        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,
        **credit_summary,
    }

@app.get("/license_status")
async def license_status(request: Request):
    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"]
    db = _init_firebase()
    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
    expiry_date = lic.get("expiry_date")
    is_expired = False
    days_remaining = 30
    if expiry_date:
        try:
            exp_dt = expiry_date if getattr(expiry_date, "tzinfo", None) else expiry_date.replace(tzinfo=timezone.utc)
            delta = exp_dt - datetime.now(timezone.utc)
            days_remaining = delta.days
            if delta.total_seconds() < 0:
                is_expired = True
        except Exception:
            pass

    if is_expired:
        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,
        "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())
    return {
        "status": "ok",
        "space_id": SPACE_ID,
        "firebase_configured": firebase_configured,
        "active_jobs": active_jobs,
        "max_concurrent_jobs": MAX_CONCURRENT_JOBS,
        "available": active_jobs < MAX_CONCURRENT_JOBS and bool(policy.get("space_enabled", True)) and not bool(policy.get("maintenance_mode", False)),
        "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}


GRADIO_DEFAULT_VOICES = {
    "edge": "en-US-AvaNeural",
    "piper": "piper:en_US-amy-medium",
    "silero": "silero:ru_xenia",
    "f5tts": "f5tts:v1_base",
}


def _write_audio_file(audio: bytes, suffix: str) -> str:
    fd, path = tempfile.mkstemp(suffix=suffix)
    os.close(fd)
    with open(path, "wb") as f:
        f.write(audio)
    return path


def _build_gradio_ui(gr):
    ui_engines = available_engines()
    css = """
    .gradio-container {
        max-width: 1160px !important;
        margin: auto !important;
        font-family: Inter, ui-sans-serif, system-ui, sans-serif;
    }
    .voicecraft-hero {
        border: 1px solid rgba(148, 163, 184, 0.24);
        border-radius: 18px;
        padding: 22px 24px;
        background: linear-gradient(135deg, rgba(15, 23, 42, 0.96), rgba(17, 24, 39, 0.92));
        color: white;
        box-shadow: 0 22px 60px rgba(15, 23, 42, 0.18);
    }
    .voicecraft-hero h1 {
        margin: 0 0 8px;
        font-size: 30px;
        line-height: 1.1;
        letter-spacing: 0;
    }
    .voicecraft-hero p {
        margin: 0;
        color: rgba(226, 232, 240, 0.88);
    }
    """

    async def generate_audio(engine, text, voice, rate, volume, pitch, speed, style, styledegree, reference_audio, token):
        if API_SECRET and not secrets.compare_digest((token or "").strip(), API_SECRET):
            raise gr.Error("Invalid API token")

        clean_text = (text or "").strip()
        if not clean_text:
            raise gr.Error("Text is empty")
        if len(clean_text) > 60000:
            raise gr.Error("Text is too long for one browser request. Use the desktop app for automatic long-script batching.")

        engine = (engine or "edge").strip().lower()
        if engine in CLONE_ENGINES and not clone_engines_enabled():
            raise gr.Error("Clone engines are disabled on this Space")
        voice = (voice or GRADIO_DEFAULT_VOICES.get(engine) or GRADIO_DEFAULT_VOICES["edge"]).strip()
        style = (style or "").strip() or None
        styledegree_value = str(styledegree) if styledegree is not None else None
        reference_path = reference_audio if isinstance(reference_audio, str) and reference_audio else None

        try:
            if engine == "edge":
                audio = await synthesize_edge(
                    clean_text,
                    voice,
                    rate=rate,
                    volume=volume,
                    pitch=pitch,
                    style=style,
                    styledegree=styledegree_value,
                )
                return _write_audio_file(audio, ".mp3"), "Ready - Edge audio generated."

            if engine == "piper":
                audio = await asyncio.to_thread(synthesize_piper, clean_text, voice, float(speed or 1.0))
                return _write_audio_file(audio, ".wav"), "Ready - Piper audio generated."

            if engine == "silero":
                audio = await asyncio.to_thread(synthesize_silero, clean_text, voice)
                return _write_audio_file(audio, ".wav"), "Ready - Silero audio generated."

            if engine == "f5tts":
                audio = await synthesize_f5tts(clean_text, reference_path=reference_path)
                return _write_audio_file(audio, ".wav"), "Ready - voice clone generated."

            raise gr.Error(f"Unknown engine: {engine}")

        except Exception as e:
            logging.error("Gradio synthesis failed: %s", e, exc_info=True)
            raise gr.Error(f"Synthesis failed: {str(e)[:220]}")

    def default_voice_for_engine(engine):
        return GRADIO_DEFAULT_VOICES.get((engine or "edge").lower(), GRADIO_DEFAULT_VOICES["edge"])

    with gr.Blocks(title="VoiceCraft TTS Server", css=css) as demo:
        gr.HTML(
            """
            <div class="voicecraft-hero">
                <h1>VoiceCraft TTS Server</h1>
                <p>FastAPI endpoints are live for the desktop app. This panel is only for quick browser testing.</p>
            </div>
            """
        )
        with gr.Row():
            with gr.Column(scale=3):
                text = gr.Textbox(
                    label="Script",
                    lines=10,
                    placeholder="Paste text here...",
                    value="Hello from VoiceCraft. Your deployment is ready for a quick audio test.",
                )
                with gr.Row():
                    engine = gr.Dropdown(
                        label="Engine",
                        choices=ui_engines,
                        value="edge",
                    )
                    voice = gr.Textbox(label="Voice code", value=GRADIO_DEFAULT_VOICES["edge"])
                reference_audio = gr.Audio(label="Reference audio for cloning", sources=["upload"], type="filepath")
                token = gr.Textbox(label="API token", type="password", placeholder="Required when API_SECRET is set")
            with gr.Column(scale=2):
                rate = gr.Textbox(label="Edge rate", value="+0%")
                volume = gr.Textbox(label="Edge volume", value="+0%")
                pitch = gr.Textbox(label="Edge pitch", value="+0Hz")
                style = gr.Textbox(label="Edge style", placeholder="cheerful, sad, whispering...")
                styledegree = gr.Slider(label="Style degree", minimum=0.0, maximum=2.0, value=1.0, step=0.1)
                speed = gr.Slider(label="Piper speed", minimum=0.65, maximum=1.35, value=1.0, step=0.05)
                run = gr.Button("Generate Audio", variant="primary")
        output_audio = gr.Audio(label="Output", type="filepath")
        status_box = gr.Markdown("Ready.")
        gr.Markdown("API paths: `/tts`, `/health`, `/status`, `/all_voices`.")

        engine.change(default_voice_for_engine, inputs=engine, outputs=voice)
        run.click(
            generate_audio,
            inputs=[engine, text, voice, rate, volume, pitch, speed, style, styledegree, reference_audio, token],
            outputs=[output_audio, status_box],
        )

    return demo


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


if __name__ == "__main__":
    port = int(os.environ.get("PORT") or os.environ.get("GRADIO_SERVER_PORT") or 7860)
    app.launch(
        server_name="0.0.0.0",
        server_port=port,
        share=False,
        show_error=False,
    )