Spaces:
Sleeping
Sleeping
| 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 | |
| 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() | |
| 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() | |
| 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)) | |
| 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 | |
| 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 | |
| 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} | |
| 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) | |
| 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, | |
| } | |
| 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, | |
| } | |
| def status(): | |
| return {"status": "ok"} | |
| 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, | |
| } | |
| 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} | |
| 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") | |