"""TTS Voice Creator - Clone and Design — FastAPI backend""" from __future__ import annotations import asyncio import base64 import glob import ipaddress import io import json import logging import os import re import shutil import socket import struct import subprocess import tempfile import time import uuid import wave from datetime import datetime, timezone from html import unescape from pathlib import Path from typing import AsyncGenerator from urllib.parse import quote, urljoin, urlparse, urlsplit import requests from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile from fastapi.responses import FileResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from pydub import AudioSegment logger = logging.getLogger("uvicorn.error") # ── Boot-time defaults ──────────────────────────────────────────────────────── _VOICES_DIR_DEFAULT = os.environ.get("VOICES_DIR", "/voices") _OUTPUT_DIR_DEFAULT = os.environ.get("OUTPUT_DIR", "/voices/active_voices") _WHISPER_DEFAULT = os.environ.get("WHISPER_URL", "http://host.docker.internal:8010") _TTS_DEFAULT = os.environ.get("TTS_URL", "http://host.docker.internal:8020") _TTS_STREAM_DEFAULT = os.environ.get("TTS_STREAM_URL", "http://host.docker.internal:8023") _CUSTOMVOICE_DEFAULT = os.environ.get("CUSTOMVOICE_URL", "http://host.docker.internal:8022") _VOICE_DESIGN_DEFAULT = os.environ.get("VOICE_DESIGN_URL", "http://host.docker.internal:8021") _NVIDIA_ROUTER_DEFAULT = os.environ.get("NVIDIA_SPEECH_ROUTER_URL", "http://host.docker.internal:8090") _NVIDIA_TTS_DEFAULT = os.environ.get("NVIDIA_MAGPIE_TTS_URL", "http://host.docker.internal:8091") _NVIDIA_ASR_DEFAULT = os.environ.get("NVIDIA_PARAKEET_ASR_URL", "http://host.docker.internal:8092") _NVIDIA_CLONE_DEFAULT = os.environ.get("NVIDIA_TTS_CLONE_URL", "http://host.docker.internal:8093") _NVIDIA_ZEROSHOT_DEFAULT = os.environ.get("NVIDIA_ZEROSHOT_TTS_URL", _NVIDIA_CLONE_DEFAULT) _NVIDIA_FLOW_DEFAULT = os.environ.get("NVIDIA_FLOW_TTS_URL", "http://host.docker.internal:8094") _FASTER_WHISPER_DEFAULT = os.environ.get("FASTER_WHISPER_URL", "http://host.docker.internal:8000") _WHISPER_CPP_DEFAULT = os.environ.get("WHISPER_CPP_URL", "http://host.docker.internal:8080") _GROQ_STT_ENDPOINT = "https://api.groq.com/openai/v1" _KOKORO_DEFAULT = os.environ.get("KOKORO_URL", "http://host.docker.internal:8880/v1") _VIBEVOICE_DEFAULT = os.environ.get("VIBEVOICE_URL", "http://192.168.178.8:8027") _XTTS_DEFAULT = os.environ.get("XTTS_URL", "http://host.docker.internal:8024") _TTS_CONTAINER = os.environ.get("TTS_CONTAINER_NAME", "faster-qwen3-tts") _TTS_CONTAINERS_RAW = os.environ.get("TTS_CONTAINER_NAMES", "") # comma-separated override _VOICE_DESIGN_MODEL = os.environ.get("VOICE_DESIGN_MODEL", "Qwen3-TTS-12Hz-1.7B-VoiceDesign") _VOICE_TARGET_DBFS = float(os.environ.get("VOICE_TARGET_DBFS", "-20.0")) _VOICE_PEAK_DBFS = float(os.environ.get("VOICE_PEAK_DBFS", "-1.0")) _MAX_UPLOAD_BYTES = int(os.environ.get("MAX_UPLOAD_MB", "1024")) * 1024 * 1024 _MAX_PICTURE_BYTES = int(os.environ.get("MAX_PICTURE_MB", "10")) * 1024 * 1024 _MAX_SOUND_BYTES = int(os.environ.get("MAX_SOUND_MB", "25")) * 1024 * 1024 _MAX_TTS_OUTPUT_SECONDS = float(os.environ.get("MAX_TTS_OUTPUT_SECONDS", "30")) _STT_REQUEST_TIMEOUT = float(os.environ.get("STT_REQUEST_TIMEOUT", os.environ.get("REQUEST_TIMEOUT", "900"))) _BENCHMARK_TEXT = os.environ.get("VOICE_BENCHMARK_TEXT", "This is a short realtime voice benchmark.") _BENCHMARK_SENTENCES = [ ("short", _BENCHMARK_TEXT), ("medium", "The quick brown fox jumps over the lazy dog near the river bank."), ( "long", "Artificial intelligence is transforming the way we interact with technology. " "From voice assistants to autonomous vehicles, machine learning models are becoming " "an integral part of everyday life.", ), ] _ALLOW_PRIVATE_DOWNLOADS = os.environ.get("ALLOW_PRIVATE_DOWNLOADS", "").lower() in {"1", "true", "yes"} def _default_config_dir() -> Path: env_dir = os.environ.get("CONFIG_DIR") if env_dir: return Path(env_dir) current = Path("/home/app/.config/tts-voice-creator") legacy = Path("/home/app/.config/voice-clone-factory") if legacy.exists(): legacy.mkdir(parents=True, exist_ok=True) if current.exists(): for name in ("settings.json", "voice_design_presets.json", "tts_routes.json"): src, dst = current / name, legacy / name if src.exists() and not dst.exists(): try: shutil.copy2(src, dst) except Exception as exc: logger.warning("Could not migrate config %s to mounted config dir: %s", name, exc) return legacy return current CONFIG_DIR = _default_config_dir() CONFIG_FILE = CONFIG_DIR / "settings.json" DESIGN_PRESETS_FILE = CONFIG_DIR / "voice_design_presets.json" TTS_ROUTES_FILE = CONFIG_DIR / "tts_routes.json" # ── Temp file registry ──────────────────────────────────────────────────────── TEMP_DIR = Path(tempfile.mkdtemp(prefix="vcf_")) _registry: dict[str, Path] = {} app = FastAPI(title="TTS Voice Creator - Clone and Design") # ── In-memory log buffer ─────────────────────────────────────────────────────── _LOG_BUFFER_MAX = 400 _log_buffer: list[dict] = [] class _BufferHandler(logging.Handler): def emit(self, record: logging.LogRecord) -> None: try: _log_buffer.insert(0, { "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(), "level": record.levelname, "name": record.name, "msg": record.getMessage(), }) del _log_buffer[_LOG_BUFFER_MAX:] except Exception: pass _buf_handler = _BufferHandler() _buf_handler.setLevel(logging.DEBUG) logging.getLogger().addHandler(_buf_handler) _ROUTING_LOG_MAX = int(os.environ.get("TTS_ROUTING_LOG_MAX", "120")) _routing_log: list[dict] = [] def _routing_log_add(**entry) -> None: item = { "ts": datetime.now(timezone.utc).isoformat(), **entry, } _routing_log.insert(0, item) del _routing_log[_ROUTING_LOG_MAX:] def _routing_log_request( request: Request, *, status: str, app: str, requested_voice: str, routed_voice: str, backend: str, route: dict | None, response_format: str, text: str, **extra, ) -> None: route = route or {} _routing_log_add( kind="proxy", status=status, app=app, requested_voice=requested_voice, routed_voice=routed_voice, backend=backend, language=str(route.get("detected_language", "")) or (_detect_text_language(text) if text else ""), matched=bool(route), route_id=str(route.get("id", "")), response_format=response_format, text_preview=text[:160], client=request.client.host if request.client else "", user_agent=str(request.headers.get("user-agent", ""))[:160], **extra, ) def _safe_child_path(root: Path, candidate: Path) -> Path: root_resolved = root.resolve() candidate_resolved = candidate.resolve() try: candidate_resolved.relative_to(root_resolved) except ValueError: raise HTTPException(403, "Access denied") return candidate_resolved def _validate_http_url(raw: str, *, allow_private: bool = True) -> str: raw = str(raw or "").strip() if not raw: raise HTTPException(400, "URL is required") parts = urlsplit(raw) if parts.scheme not in {"http", "https"}: raise HTTPException(400, "Only http:// and https:// URLs are allowed") if not parts.hostname: raise HTTPException(400, "URL must include a hostname") if parts.username or parts.password: raise HTTPException(400, "URLs with embedded credentials are not allowed") if allow_private: return raw try: infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80), type=socket.SOCK_STREAM) except socket.gaierror: raise HTTPException(400, "URL hostname could not be resolved") for info in infos: ip = ipaddress.ip_address(info[4][0]) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved: raise HTTPException(400, "Private, local, and reserved network URLs are not allowed for downloads") return raw def _copy_limited(src, dest, limit: int) -> int: total = 0 while True: chunk = src.read(1024 * 1024) if not chunk: break total += len(chunk) if total > limit: raise HTTPException(413, "Uploaded file is too large") dest.write(chunk) return total def _decode_chunked_bytes(data: bytes) -> bytes: result = bytearray() pos = 0 while pos < len(data): end = data.find(b"\r\n", pos) if end < 0: break try: size = int(data[pos:end].split(b";")[0].strip(), 16) except ValueError: break if size == 0: break pos = end + 2 result.extend(data[pos:pos + size]) pos += size + 2 return bytes(result) def _docker_get_json(path: str) -> tuple[int, dict | list | None]: sock_path = os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock") if not Path(sock_path).exists(): raise RuntimeError(f"Docker socket not found: {sock_path}") request = ( f"GET {path} HTTP/1.1\r\n" "Host: docker\r\n" "Connection: close\r\n\r\n" ).encode("utf-8") chunks: list[bytes] = [] with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(10) sock.connect(sock_path) sock.sendall(request) while True: data = sock.recv(65536) if not data: break chunks.append(data) raw = b"".join(chunks) header_end = raw.find(b"\r\n\r\n") if header_end < 0: raise RuntimeError("Invalid Docker HTTP response") header_str = raw[:header_end].decode("utf-8", errors="replace") body_bytes = raw[header_end + 4:] status_line = header_str.splitlines()[0] m = re.match(r"HTTP/\S+\s+(\d+)", status_line) if not m: raise RuntimeError(f"Invalid Docker status line: {status_line!r}") code = int(m.group(1)) if "transfer-encoding: chunked" in header_str.lower(): body_bytes = _decode_chunked_bytes(body_bytes) if not body_bytes.strip(): return code, None return code, json.loads(body_bytes) def _docker_post(path: str) -> tuple[int, str]: sock_path = os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock") if not Path(sock_path).exists(): raise RuntimeError(f"Docker socket not found: {sock_path}") request = ( f"POST {path} HTTP/1.1\r\n" "Host: docker\r\n" "Content-Length: 0\r\n" "Connection: close\r\n\r\n" ).encode("utf-8") chunks = [] with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(20) sock.connect(sock_path) sock.sendall(request) while True: data = sock.recv(8192) if not data: break chunks.append(data) raw = b"".join(chunks).decode("utf-8", errors="replace") status_line = raw.splitlines()[0] if raw else "" match = re.match(r"HTTP/\S+\s+(\d+)", status_line) if not match: raise RuntimeError("Invalid Docker API response") return int(match.group(1)), raw # ── Settings ────────────────────────────────────────────────────────────────── _SETTINGS_KEYS = { "whisper_url", "tts_url", "tts_stream_url", "tts_stream_mode", "tts_backend", "output_dir", "voices_scan_dir", "voice_design_url", "customvoice_url", "nvidia_router_url", "nvidia_tts_url", "nvidia_asr_url", "nvidia_clone_url", "nvidia_zeroshot_url", "nvidia_flow_url", "faster_whisper_url", "whisper_cpp_url", "groq_api_key", "kokoro_url", "vibevoice_url", "xtts_url", "whisper_api_key", "tts_api_key", "voice_design_api_key", "elevenlabs_api_key", "tts_stability_enabled", "tts_extra_params", "tts_extra_params_by_backend", # Captures settings "stt_language", "stt_preferred_backend", "auto_refine", "refine_model", "refine_fillers", "refine_repetitions", "refine_corrections", "refine_punctuation", "captures_default_voice", "client_voice_bindings", "llm_url", } _TTS_STABILITY_DEFAULT = {"temperature": 0.1, "top_p": 0.8, "seed": 0} _TTS_STABILITY_BY_BACKEND_DEFAULT = { "voice_clone": dict(_TTS_STABILITY_DEFAULT), "streaming": dict(_TTS_STABILITY_DEFAULT), "customvoice": dict(_TTS_STABILITY_DEFAULT), "voice_design": dict(_TTS_STABILITY_DEFAULT), "nvidia_magpie": {}, "nvidia_zeroshot": {}, "nvidia_flow": {}, "kokoro": {}, "vibevoice": {}, } _TTS_PAYLOAD_CORE_KEYS = {"model", "input", "voice", "response_format", "instruct", "language"} def _settings_bool(value, default: bool = True) -> bool: if value is None: return default if isinstance(value, bool): return value return str(value).strip().lower() not in {"0", "false", "off", "no"} def _clean_tts_param_dict(raw, fallback: dict | None = None) -> dict: if raw in (None, ""): raw = fallback or {} if isinstance(raw, str): try: raw = json.loads(raw) except Exception: raw = fallback or {} if not isinstance(raw, dict): return {} params = {} for key, value in raw.items(): if key in _TTS_PAYLOAD_CORE_KEYS or str(key).startswith("_"): continue if isinstance(value, (str, int, float, bool)) or value is None: params[str(key)] = value return params def _tts_extra_params(settings: dict, backend: str = "voice_clone") -> dict: if not _settings_bool(settings.get("tts_stability_enabled"), True): return {} backend = _clean_preview_backend(backend) if "_clean_preview_backend" in globals() else str(backend or "voice_clone") by_backend = settings.get("tts_extra_params_by_backend") if isinstance(by_backend, str): try: by_backend = json.loads(by_backend) except Exception: by_backend = None if isinstance(by_backend, dict): fallback = _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT) if backend in by_backend: return _clean_tts_param_dict(by_backend.get(backend), fallback) return _clean_tts_param_dict(fallback, fallback) return _clean_tts_param_dict(settings.get("tts_extra_params", _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT)), _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT)) def _apply_tts_extra_params(payload: dict, settings: dict, backend: str = "voice_clone") -> dict: params = _tts_extra_params(settings, backend) if params: payload.update(params) return payload def _strip_tts_extra_params(payload: dict) -> dict: return {k: v for k, v in payload.items() if k in _TTS_PAYLOAD_CORE_KEYS} def _post_tts_with_fallback(endpoint: str, payload: dict, headers: dict, **kwargs) -> requests.Response: resp = requests.post(endpoint, json=payload, headers=headers, **kwargs) fallback = _strip_tts_extra_params(payload) if resp.status_code in {400, 404, 415, 422} and fallback != payload: try: resp.close() except Exception: pass logger.warning("TTS backend rejected extra generation params; retrying without them") return requests.post(endpoint, json=fallback, headers=headers, **kwargs) return resp def _clean_preview_backend(value: str) -> str: key = re.sub(r"[^a-z0-9]+", "_", str(value or "voice_clone").lower()).strip("_") aliases = { "clone": "voice_clone", "base": "voice_clone", "voiceclone": "voice_clone", "voice_clone_base": "voice_clone", "stream": "streaming", "tts_streaming": "streaming", "custom": "customvoice", "custom_voice": "customvoice", "voice_design": "voice_design", "voicedesign": "voice_design", "design": "voice_design", "nvidia": "nvidia_magpie", "magpie": "nvidia_magpie", "nvidia_tts": "nvidia_magpie", "nvidia_magpie_tts": "nvidia_magpie", "nvidia_clone": "nvidia_zeroshot", "nvidia_zeroshot_tts": "nvidia_zeroshot", "magpie_zeroshot": "nvidia_zeroshot", "zeroshot": "nvidia_zeroshot", "zero_shot": "nvidia_zeroshot", "nvidia_flow_tts": "nvidia_flow", "magpie_flow": "nvidia_flow", "flow": "nvidia_flow", "kokoro_fastapi": "kokoro", "kokoro_tts": "kokoro", "kokoro_local": "kokoro", "vibevoice_service": "vibevoice", "vibe_voice": "vibevoice", "vibetts": "vibevoice", "xtts_v2": "xtts", "xtts2": "xtts", "coqui_xtts": "xtts", } key = aliases.get(key, key) return key if key in {"voice_clone", "streaming", "customvoice", "voice_design", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow", "kokoro", "vibevoice", "xtts"} else "voice_clone" def _preview_backend_base_url(settings: dict, backend: str) -> str: backend = _clean_preview_backend(backend) if backend == "streaming": return settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT if backend == "customvoice": return settings.get("customvoice_url") or _CUSTOMVOICE_DEFAULT if backend == "voice_design": return settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT if backend == "nvidia_magpie": return settings.get("nvidia_tts_url") or settings.get("nvidia_router_url") or _NVIDIA_TTS_DEFAULT if backend == "nvidia_zeroshot": return settings.get("nvidia_zeroshot_url") or settings.get("nvidia_clone_url") or settings.get("nvidia_router_url") or _NVIDIA_ZEROSHOT_DEFAULT if backend == "nvidia_flow": return settings.get("nvidia_flow_url") or settings.get("nvidia_clone_url") or settings.get("nvidia_router_url") or _NVIDIA_FLOW_DEFAULT if backend == "kokoro": return settings.get("kokoro_url") or _KOKORO_DEFAULT if backend == "vibevoice": return settings.get("vibevoice_url") or _VIBEVOICE_DEFAULT if backend == "xtts": return settings.get("xtts_url") or _XTTS_DEFAULT return settings.get("tts_url") or _TTS_DEFAULT def _normalize_settings(s: dict) -> dict: scan_dir = Path(s.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) output_dir = Path(s.get("output_dir", _OUTPUT_DIR_DEFAULT)) legacy_output_dirs = { Path("/voices/speakers"), scan_dir / "speakers", } if output_dir in legacy_output_dirs: s["output_dir"] = str(scan_dir / "active_voices") if s.get("tts_stream_mode") not in {"auto", "streaming", "buffered"}: s["tts_stream_mode"] = "auto" return s def _load_settings() -> dict: defaults = { "whisper_url": _WHISPER_DEFAULT, "tts_url": _TTS_DEFAULT, "tts_stream_url": _TTS_STREAM_DEFAULT, "tts_stream_mode": "auto", "tts_backend": "openai", "output_dir": _OUTPUT_DIR_DEFAULT, "voices_scan_dir": _VOICES_DIR_DEFAULT, "voice_design_url": _VOICE_DESIGN_DEFAULT, "customvoice_url": _CUSTOMVOICE_DEFAULT, "nvidia_router_url": _NVIDIA_ROUTER_DEFAULT, "nvidia_tts_url": _NVIDIA_TTS_DEFAULT, "nvidia_asr_url": _NVIDIA_ASR_DEFAULT, "nvidia_clone_url": _NVIDIA_CLONE_DEFAULT, "nvidia_zeroshot_url": _NVIDIA_ZEROSHOT_DEFAULT, "nvidia_flow_url": _NVIDIA_FLOW_DEFAULT, "faster_whisper_url": _FASTER_WHISPER_DEFAULT, "whisper_cpp_url": _WHISPER_CPP_DEFAULT, "groq_api_key": "", "kokoro_url": _KOKORO_DEFAULT, "vibevoice_url": _VIBEVOICE_DEFAULT, "xtts_url": _XTTS_DEFAULT, "whisper_api_key": "", "tts_api_key": "", "voice_design_api_key": "", "tts_stability_enabled": True, "tts_extra_params": _TTS_STABILITY_DEFAULT, "tts_extra_params_by_backend": _TTS_STABILITY_BY_BACKEND_DEFAULT, # Captures "stt_language": "", "stt_preferred_backend": "", "auto_refine": "off", "refine_model": "", "refine_fillers": True, "refine_repetitions": True, "refine_corrections": True, "refine_punctuation": True, "captures_default_voice": "", "client_voice_bindings": {}, "llm_url": "http://localhost:11434/v1", } if CONFIG_FILE.exists(): try: saved = json.loads(CONFIG_FILE.read_text()) defaults.update({k: v for k, v in saved.items() if k in _SETTINGS_KEYS}) except Exception: pass return _normalize_settings(defaults) def _save_settings(s: dict) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text(json.dumps(s, indent=2)) @app.get("/api/settings") async def get_settings(): return _load_settings() @app.post("/api/settings") async def post_settings(request: Request): data = await request.json() s = _load_settings() s.update({k: v for k, v in data.items() if k in _SETTINGS_KEYS}) s = _normalize_settings(s) _save_settings(s) return {"ok": True} # ── TTS routing rules ───────────────────────────────────────────────────────── _ROUTE_LANGS = {"AUTO", "*", "EN", "DE", "FR", "ES", "IT", "PT", "NL", "PL"} _ROUTE_BACKENDS = {"voice_clone", "streaming", "voice_design", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow"} def _clean_route_token(value: str, default: str = "*") -> str: value = str(value or "").strip() if not value: return default value = re.sub(r"[^A-Za-z0-9_\-\.\* ]+", "_", value) return value[:80] or default def _clean_route_sound(value: str) -> str: value = str(value or "").strip() if not value: return "" value = re.sub(r"[^A-Za-z0-9_\-\.\* /]+", "_", value) return value[:240] def _clean_route_backend(value: str) -> str: key = re.sub(r"[^a-z0-9]+", "_", str(value or "voice_clone").lower()).strip("_") aliases = { "clone": "voice_clone", "base": "voice_clone", "tts": "voice_clone", "standard": "voice_clone", "voiceclone": "voice_clone", "voice_clone_base": "voice_clone", "stream": "streaming", "tts_streaming": "streaming", "voice_design": "voice_design", "voicedesign": "voice_design", "design": "voice_design", "nvidia": "nvidia_magpie", "magpie": "nvidia_magpie", "nvidia_tts": "nvidia_magpie", "nvidia_magpie_tts": "nvidia_magpie", "nvidia_clone": "nvidia_zeroshot", "nvidia_zeroshot_tts": "nvidia_zeroshot", "magpie_zeroshot": "nvidia_zeroshot", "zeroshot": "nvidia_zeroshot", "zero_shot": "nvidia_zeroshot", "nvidia_flow_tts": "nvidia_flow", "magpie_flow": "nvidia_flow", "flow": "nvidia_flow", } key = aliases.get(key, key) return key if key in _ROUTE_BACKENDS else "voice_clone" def _route_backend(route: dict | None, voice: str = "") -> str: if _resolve_virtual_voice(voice): return "voice_design" return _clean_route_backend((route or {}).get("backend", "voice_clone")) def _normalize_route(rule: dict, idx: int = 0) -> dict: lang = str(rule.get("language", "*") or "*").strip().upper() if lang not in _ROUTE_LANGS: lang = "*" output_voice = _clean_route_token(rule.get("output_voice", ""), "") return { "id": _clean_route_token(rule.get("id", f"route_{idx+1}"), f"route_{idx+1}"), "enabled": bool(rule.get("enabled", True)), "app": _clean_route_token(rule.get("app", "Open WebUI"), "Open WebUI"), "input_voice": _clean_route_token(rule.get("input_voice", "default"), "default"), "language": lang, "backend": _clean_route_backend(rule.get("backend", "voice_clone")), "output_voice": output_voice, "before_sound": _clean_route_sound(rule.get("before_sound", "")), "after_sound": _clean_route_sound(rule.get("after_sound", "")), } def _load_tts_routes() -> list[dict]: if not TTS_ROUTES_FILE.exists(): return [] try: raw = json.loads(TTS_ROUTES_FILE.read_text()) routes = raw.get("routes", raw) if isinstance(raw, dict) else raw if not isinstance(routes, list): return [] return [_normalize_route(r, i) for i, r in enumerate(routes) if isinstance(r, dict)] except Exception: return [] def _save_tts_routes(routes: list[dict]) -> None: TTS_ROUTES_FILE.parent.mkdir(parents=True, exist_ok=True) clean = [_normalize_route(r, i) for i, r in enumerate(routes)] TTS_ROUTES_FILE.write_text(json.dumps({"routes": clean}, indent=2)) def _app_key(value: str) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) def _canonical_app_name(value: str) -> str: key = _app_key(value) if key in {"openwebui", "openwebuiapp"}: return "Open WebUI" if key == "sillytavern": return "SillyTavern" if key in {"homeassistant", "ha"}: return "Home Assistant" return str(value or "").strip() or "Open WebUI" def _request_app_name(request: Request) -> str: explicit = request.headers.get("x-tts-app") or request.headers.get("x-client-app") if explicit: return _canonical_app_name(explicit) header = ( request.headers.get("x-openwebui-app") or request.headers.get("x-openwebui-user-name") or request.headers.get("referer") or request.headers.get("origin") or request.headers.get("user-agent") or "" ) h = _app_key(header) if "openwebui" in h: return "Open WebUI" if "sillytavern" in h: return "SillyTavern" if "homeassistant" in h: return "Home Assistant" return "Open WebUI" def _detect_text_language(text: str) -> str: low = f" {text.lower()} " if re.search(r"[äöüß]", low) or re.search(r"\b(und|der|die|das|ich|nicht|mit|ist|ein|eine|auf|für)\b", low): return "DE" if re.search(r"[éèêàçùœ]", low) or re.search(r"\b(et|le|la|les|des|une|avec|pour|est|pas|que)\b", low): return "FR" if re.search(r"[áéíóúñ¿¡]", low) or re.search(r"\b(el|la|los|las|una|con|para|que|pero|está|hola)\b", low): return "ES" if re.search(r"\b(il|lo|gli|una|con|per|che|ciao|grazie|sono|della)\b", low): return "IT" if re.search(r"[ãõç]", low) or re.search(r"\b(com|para|uma|que|não|está|obrigado)\b", low): return "PT" if re.search(r"\b(het|een|niet|met|voor|zijn|maar|dank|goede)\b", low): return "NL" if re.search(r"[ąćęłńóśźż]", low) or re.search(r"\b(jest|nie|tak|dla|oraz|dzień|dziękuję)\b", low): return "PL" return "EN" def _route_specificity(rule: dict, app: str, voice: str, lang: str) -> tuple[int, int, int, int] | None: if not rule.get("enabled", True) or not rule.get("output_voice"): return None r_app = str(rule.get("app", "*")) r_voice = str(rule.get("input_voice", "*")) r_lang = str(rule.get("language", "*")).upper() app_ok = r_app == "*" or _app_key(r_app) == _app_key(app) voice_ok = r_voice == "*" or r_voice.lower() == voice.lower() lang_ok = r_lang in {"*", "AUTO"} or r_lang == lang if not (app_ok and voice_ok and lang_ok): return None return ( 1 if r_app != "*" else 0, 1 if r_voice != "*" else 0, 1 if r_lang not in {"*", "AUTO"} else 0, 0, ) def _resolve_tts_route(app: str, voice: str, text: str) -> tuple[str, dict | None]: lang = _detect_text_language(text) best: tuple[tuple[int, int, int, int], dict] | None = None for idx, rule in enumerate(_load_tts_routes()): spec = _route_specificity(rule, app, voice, lang) if spec is None: continue spec = (spec[0], spec[1], spec[2], -idx) if best is None or spec > best[0]: best = (spec, rule) if not best: return voice, None routed = dict(best[1]) routed["detected_language"] = lang routed["requested_voice"] = voice routed["app"] = app return str(routed["output_voice"]), routed @app.get("/api/tts-routes") async def get_tts_routes(): return {"routes": _load_tts_routes()} @app.post("/api/tts-routes") async def post_tts_routes(request: Request): data = await request.json() routes = data.get("routes", data) if isinstance(data, dict) else data if not isinstance(routes, list): raise HTTPException(400, "routes must be a list") _save_tts_routes(routes) return {"ok": True, "routes": _load_tts_routes()} @app.get("/api/tts-routing-log") async def get_tts_routing_log(limit: int = 60): limit = max(1, min(int(limit or 60), _ROUTING_LOG_MAX)) return {"items": _routing_log[:limit], "max": _ROUTING_LOG_MAX} @app.delete("/api/tts-routing-log") async def clear_tts_routing_log(): _routing_log.clear() return {"ok": True, "items": []} @app.get("/api/logs") async def get_server_logs(limit: int = 200): limit = max(1, min(int(limit or 200), _LOG_BUFFER_MAX)) return {"items": _log_buffer[:limit], "max": _LOG_BUFFER_MAX} @app.delete("/api/logs") async def clear_server_logs(): _log_buffer.clear() return {"ok": True} # ── VoiceDesign virtual voice presets ───────────────────────────────────────── _DEFAULT_DESIGN_PRESETS = { "EN_M_Young_Energetic": { "description": "Young adult male voice, clear English, bright and energetic, moderately high pitch, quick but controlled speaking rate, confident and friendly, suitable for tutorials or streaming.", "sample_text": "Hey everyone, welcome back. Today we are going to move quickly, keep it clear, and make this setup feel easy.", "language": "English", "gender": "M", }, "EN_F_Warm_Narrator": { "description": "Adult female English narrator, warm and smooth, medium pitch, calm pace, gentle emotion, clear articulation, suited for audiobooks and voice assistant responses.", "sample_text": "The room grew quiet as the morning light touched the window, and for a moment everything felt simple and kind.", "language": "English", "gender": "F", }, "DE_M_Elderly_Documentary": { "description": "Aeltere maennliche deutsche Stimme, tief und resonant, langsam und gelassen, klar artikuliert, ruhig und dokumentarisch, mit serioeser und vertrauensvoller Praesenz.", "sample_text": "Seit vielen Jahren beobachten wir diesen Ort, seine Geschichte und die Menschen, die ihn mit Leben fuellen.", "language": "German", "gender": "M", }, "DE_F_Young_Friendly": { "description": "Junge weibliche deutsche Stimme, hell und freundlich, natuerliche Sprechgeschwindigkeit, klare Aussprache, leicht optimistisch und nahbar, passend fuer Assistenten und kurze Erklaerungen.", "sample_text": "Hallo, schoen dass du da bist. Ich zeige dir kurz, wie alles funktioniert, Schritt fuer Schritt.", "language": "German", "gender": "F", }, "EN_N_Old_Wise_Assistant": { "description": "Older neutral English voice, gentle and wise, slightly low pitch, slow measured pace, soothing tone, very clear pronunciation, calm personality for guidance and reflective narration.", "sample_text": "Take a slow breath. We will look at the facts carefully, choose the next step, and keep moving.", "language": "English", "gender": "N", }, } def _slug_voice_design_name(name: str) -> str: slug = re.sub(r"[^A-Za-z0-9_.-]+", "_", name.strip()).strip("._-") return slug or "VoiceDesign" def _load_design_presets() -> dict: presets = dict(_DEFAULT_DESIGN_PRESETS) if DESIGN_PRESETS_FILE.exists(): try: saved = json.loads(DESIGN_PRESETS_FILE.read_text()) if isinstance(saved, dict): for name, preset in saved.items(): if isinstance(preset, dict): presets[_slug_voice_design_name(str(name))] = { "description": str(preset.get("description", "")), "sample_text": str(preset.get("sample_text", preset.get("text", ""))), "language": str(preset.get("language", "Auto")), "gender": str(preset.get("gender", "N")), } except Exception: pass return presets def _save_design_presets(presets: dict) -> None: cleaned = {} for name, preset in presets.items(): if not isinstance(preset, dict): continue key = _slug_voice_design_name(str(name)) cleaned[key] = { "description": str(preset.get("description", "")), "sample_text": str(preset.get("sample_text", preset.get("text", ""))), "language": str(preset.get("language", "Auto")), "gender": str(preset.get("gender", "N")), } CONFIG_DIR.mkdir(parents=True, exist_ok=True) DESIGN_PRESETS_FILE.write_text(json.dumps(cleaned, indent=2)) @app.get("/api/voice-design-presets") async def get_voice_design_presets(): return _load_design_presets() @app.post("/api/voice-design-presets") async def post_voice_design_presets(request: Request): data = await request.json() if not isinstance(data, dict): raise HTTPException(400, "Expected a preset object") _save_design_presets(data) return {"ok": True, "presets": _load_design_presets()} # ── Audio helpers ───────────────────────────────────────────────────────────── def _to_wav_24k(src: Path) -> Path: out = TEMP_DIR / f"{src.stem}_24k.wav" seg = AudioSegment.from_file(str(src)) seg = seg.set_frame_rate(24000).set_channels(1).set_sample_width(2) seg.export(str(out), format="wav") return out def _trim(src: Path, start_s: float, end_s: float) -> Path: seg = AudioSegment.from_file(str(src)) trimmed = seg[int(start_s * 1000):int(end_s * 1000)] trimmed, _ = _normalize_segment(trimmed) out = TEMP_DIR / f"{uuid.uuid4().hex}_trimmed.wav" trimmed.export(str(out), format="wav") return out def _duration(path: Path) -> float: if path.suffix.lower() == ".wav": try: with wave.open(str(path), "rb") as wf: frames = wf.getnframes() rate = wf.getframerate() if rate: return frames / float(rate) except Exception: pass try: seg = AudioSegment.from_file(str(path)) return len(seg) / 1000.0 except Exception: pass probe = subprocess.run( [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(path), ], capture_output=True, text=True, timeout=10, ) if probe.returncode != 0: raise RuntimeError(probe.stderr.strip() or "ffprobe failed") return float(probe.stdout.strip()) def _normalize_segment(seg: AudioSegment, target_dbfs: float = _VOICE_TARGET_DBFS, peak_dbfs: float = _VOICE_PEAK_DBFS) -> tuple[AudioSegment, dict]: before_dbfs = seg.dBFS if seg.dBFS != float("-inf") else None before_peak = seg.max_dBFS if seg.max_dBFS != float("-inf") else None if before_dbfs is None or before_peak is None: return seg, {"before_dbfs": before_dbfs, "after_dbfs": before_dbfs, "gain_db": 0.0, "peak_dbfs": before_peak} gain = target_dbfs - before_dbfs if before_peak + gain > peak_dbfs: gain = peak_dbfs - before_peak normalized = seg.apply_gain(gain) after_dbfs = normalized.dBFS if normalized.dBFS != float("-inf") else None after_peak = normalized.max_dBFS if normalized.max_dBFS != float("-inf") else None return normalized, { "before_dbfs": round(before_dbfs, 2), "after_dbfs": round(after_dbfs, 2) if after_dbfs is not None else None, "gain_db": round(gain, 2), "peak_dbfs": round(after_peak, 2) if after_peak is not None else None, } def _export_normalized_wav(src: Path, dest: Path, target_dbfs: float = _VOICE_TARGET_DBFS) -> dict: seg = AudioSegment.from_file(str(src)) seg = seg.set_frame_rate(24000).set_channels(1).set_sample_width(2) seg, info = _normalize_segment(seg, target_dbfs=target_dbfs) seg.export(str(dest), format="wav") return info def _loudness_info(path: Path) -> dict: seg = AudioSegment.from_file(str(path)) dbfs = seg.dBFS if seg.dBFS != float("-inf") else None peak = seg.max_dBFS if seg.max_dBFS != float("-inf") else None return { "dbfs": round(dbfs, 2) if dbfs is not None else None, "peak_dbfs": round(peak, 2) if peak is not None else None, "target_dbfs": _VOICE_TARGET_DBFS, } def _auto_trim_bounds(path: Path) -> dict: seg = AudioSegment.from_file(str(path)).set_channels(1) dur_ms = len(seg) if dur_ms <= 20_000: return { "start": 0.0, "end": dur_ms / 1000.0, "duration": dur_ms / 1000.0, "reason": "Audio is already short enough.", } chunk_ms = 250 chunks = [] overall_db = seg.dBFS if seg.dBFS != float("-inf") else -60.0 speech_floor = max(overall_db - 18.0, -45.0) for pos in range(0, dur_ms, chunk_ms): ch = seg[pos:pos + chunk_ms] db = ch.dBFS if ch.dBFS != float("-inf") else -80.0 max_db = ch.max_dBFS if ch.max_dBFS != float("-inf") else -80.0 chunks.append({"db": db, "speech": db >= speech_floor, "clipped": max_db > -1.0}) def score_window(start_ms: int, length_ms: int) -> tuple[float, dict]: first = max(0, start_ms // chunk_ms) last = min(len(chunks), (start_ms + length_ms + chunk_ms - 1) // chunk_ms) win = chunks[first:last] if not win: return -9999.0, {} speech_ratio = sum(1 for c in win if c["speech"]) / len(win) silence_ratio = 1.0 - speech_ratio clip_ratio = sum(1 for c in win if c["clipped"]) / len(win) speech_dbs = [c["db"] for c in win if c["speech"]] avg_db = sum(speech_dbs) / len(speech_dbs) if speech_dbs else -80.0 variance = sum((x - avg_db) ** 2 for x in speech_dbs) / len(speech_dbs) if speech_dbs else 100.0 loudness_penalty = abs(avg_db - (-20.0)) * 1.7 steadiness_penalty = min(18.0, variance ** 0.5 * 1.4) duration_s = length_ms / 1000.0 duration_penalty = abs(duration_s - 12.0) * 0.9 score = ( speech_ratio * 100.0 - silence_ratio * 55.0 - clip_ratio * 85.0 - loudness_penalty - steadiness_penalty - duration_penalty ) return score, { "speech_ratio": speech_ratio, "silence_ratio": silence_ratio, "clip_ratio": clip_ratio, "avg_db": avg_db, } best = None window_lengths = [8_000, 10_000, 12_000, 15_000, 18_000] for length_ms in window_lengths: if length_ms > dur_ms: continue for start_ms in range(0, dur_ms - length_ms + 1, 500): score, metrics = score_window(start_ms, length_ms) if best is None or score > best["score"]: best = {"start_ms": start_ms, "length_ms": length_ms, "score": score, "metrics": metrics} if best is None: end_ms = min(dur_ms, 12_000) return {"start": 0.0, "end": end_ms / 1000.0, "duration": end_ms / 1000.0, "reason": "Using the beginning because no stable speech window was found."} start_s = best["start_ms"] / 1000.0 end_s = (best["start_ms"] + best["length_ms"]) / 1000.0 m = best["metrics"] return { "start": round(start_s, 2), "end": round(end_s, 2), "duration": round(end_s - start_s, 2), "score": round(best["score"], 2), "reason": ( f"Selected {end_s - start_s:.1f}s with " f"{m.get('speech_ratio', 0) * 100:.0f}% speech, " f"{m.get('silence_ratio', 0) * 100:.0f}% silence, " f"avg {m.get('avg_db', -80):.1f} dBFS." ), } # ── Voice meta helpers ──────────────────────────────────────────────────────── _PICTURE_EXTS = [".jpg", ".jpeg", ".png", ".webp"] _SOUND_ASSET_DIRS = {"sound", "sounds", "sfx", "effects", "sound_effects", "route_sounds"} _SOUND_ASSET_PREFIXES = ("computerbeep", "beep", "ding", "chime", "notification") _AUDIO_EXTS = [".wav", ".mp3", ".m4a", ".flac", ".ogg", ".opus"] _UPLOAD_EXTS = _AUDIO_EXTS + [".mp4", ".mkv", ".webm", ".mov", ".avi"] _AUDIO_MIME = { ".wav": "audio/wav", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".flac": "audio/flac", ".ogg": "audio/ogg", ".opus": "audio/ogg", } # Default flag (ISO country code) for each language code _LANG_FLAG_DEFAULT = { "EN": "GB", "DE": "DE", "ZH": "CN", "FR": "FR", "ES": "ES", "JA": "JP", "KO": "KR", "IT": "IT", "PT": "BR", "RU": "RU", "AR": "SA", "PL": "PL", "NL": "NL", "SV": "SE", "TR": "TR", "HI": "IN", } def _find_voice_audio(voice_id: str, scan_dir: Path) -> Path | None: for ext in _AUDIO_EXTS: for p in sorted(scan_dir.rglob(f"{voice_id}{ext}")): if not _is_internal_voice_file(p) and not _is_sound_asset_file(p): return p return None def _meta_path(wav: Path) -> Path: return wav.with_suffix(".meta.json") def _load_meta(wav: Path) -> dict: mp = _meta_path(wav) if mp.exists(): try: return json.loads(mp.read_text()) except Exception: pass # Auto-detect flag and gender from voice_id parts = wav.stem.split("_", 2) lang = parts[0].upper() if parts else "" gender = parts[1].upper() if len(parts) >= 2 and parts[1].upper() in ("F", "M", "N") else "" return { "note": "", "rating": 0, "flag": _LANG_FLAG_DEFAULT.get(lang, ""), "gender": gender, "enabled": True, } def _save_meta(wav: Path, meta: dict) -> None: _meta_path(wav).write_text(json.dumps(meta, indent=2)) def _picture_path(wav: Path) -> Path | None: for ext in _PICTURE_EXTS: p = wav.with_suffix(ext) if p.exists(): return p return None def _picture_mime(path: Path) -> str: return {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp"}.get(path.suffix.lower(), "image/jpeg") def _hidden_voices_dir(settings: dict) -> Path: return Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) / "hidden_voices" def _active_voices_dir(settings: dict) -> Path: return Path(settings.get("output_dir", _OUTPUT_DIR_DEFAULT)) def _voice_package_paths(audio: Path) -> list[Path]: parent = audio.parent paths = [] for sfx in _AUDIO_EXTS + [".reference.txt", ".meta.json"] + _PICTURE_EXTS: p = parent / f"{audio.stem}{sfx}" if p.exists(): paths.append(p) paths.extend( p for p in _backup_candidates(audio, _load_meta(audio)) if p.exists() and p.parent.resolve() == parent.resolve() ) return paths def _is_internal_voice_file(path: Path) -> bool: stem = path.stem.lower() name = path.name.lower() return ( stem.endswith(".original") or ".normalized.tmp" in stem or name.startswith(".") or name.endswith(".bak") ) def _is_sound_asset_file(path: Path) -> bool: stem = path.stem.lower() parent_names = {part.lower() for part in path.parts} return ( any(part in _SOUND_ASSET_DIRS for part in parent_names) or stem.startswith(_SOUND_ASSET_PREFIXES) ) def _voice_audio_files(root: Path): for ext in _AUDIO_EXTS: for p in root.rglob(f"*{ext}"): if not _is_internal_voice_file(p) and not _is_sound_asset_file(p): yield p def _backup_path(audio: Path) -> Path: return audio.with_name(f".{audio.stem}.original{audio.suffix}.bak") def _legacy_backup_path(audio: Path) -> Path: return audio.with_name(f"{audio.stem}.original{audio.suffix}") def _backup_candidates(audio: Path, meta: dict | None = None) -> list[Path]: candidates = [_backup_path(audio), _legacy_backup_path(audio)] if meta and meta.get("original_backup"): candidates.insert(0, Path(str(meta["original_backup"]))) seen: set[Path] = set() unique = [] for candidate in candidates: try: key = candidate.resolve() if candidate.exists() else candidate except Exception: key = candidate if key not in seen: seen.add(key) unique.append(candidate) return unique def _backup_audio_suffix(backup: Path, voice_id: str) -> str | None: name = backup.name for ext in _AUDIO_EXTS: if name == f".{voice_id}.original{ext}.bak" or name == f"{voice_id}.original{ext}": return ext return None def _remove_audio_variants(parent: Path, voice_id: str, keep: Path | None = None) -> None: keep_resolved = keep.resolve() if keep and keep.exists() else None for ext in _AUDIO_EXTS: p = parent / f"{voice_id}{ext}" if p.exists() and not _is_internal_voice_file(p) and (keep_resolved is None or p.resolve() != keep_resolved): p.unlink() def _remove_voice_package(audio: Path, keep: set[Path] | None = None) -> None: keep_resolved = {p.resolve() for p in (keep or set()) if p.exists()} for p in _voice_package_paths(audio): if p.exists() and p.resolve() not in keep_resolved: p.unlink() def _backup_original_voice(audio: Path) -> Path | None: if not audio.exists() or audio.suffix.lower() not in _AUDIO_EXTS: return None backup = _backup_path(audio) if not backup.exists(): legacy = _legacy_backup_path(audio) if legacy.exists(): shutil.move(str(legacy), str(backup)) else: shutil.copy2(str(audio), str(backup)) return backup def _voice_audio_from_request(data: dict, scan_dir: Path) -> Path | None: requested_path = data.get("path") if requested_path: p = _safe_child_path(scan_dir, Path(requested_path)) if not p.exists() or not p.is_file() or p.suffix.lower() not in _AUDIO_EXTS: raise HTTPException(404, "Voice file not found") return p voice_id = data.get("voice_id", "") if not voice_id: raise HTTPException(400, "voice_id or path is required") return _find_voice_audio(voice_id, scan_dir) def _move_voice_package(audio: Path, target_dir: Path) -> Path: if audio.parent.resolve() == target_dir.resolve(): return audio target_dir.mkdir(parents=True, exist_ok=True) paths = _voice_package_paths(audio) for src in paths: dest = target_dir / src.name if dest.exists() and dest.resolve() != src.resolve(): raise HTTPException(409, f"Target file already exists: {dest}") moved_audio = target_dir / audio.name for src in paths: dest = target_dir / src.name if dest.resolve() != src.resolve(): shutil.move(str(src), str(dest)) return moved_audio def _read_reference_text(audio: Path) -> tuple[bool, str]: ref = audio.with_suffix(".reference.txt") if not ref.exists(): return False, "" return True, ref.read_text(encoding="utf-8").strip() def _voice_entry(p: Path) -> dict: meta = _load_meta(p) parts = p.stem.split("_", 2) lang = parts[0].upper() if parts else "" try: duration = round(_duration(p), 2) except Exception: duration = None has_ref, transcript = _read_reference_text(p) loudness = meta.get("loudness", {}) benchmark = meta.get("benchmark", {}) if not meta.get("flag") and lang in _LANG_FLAG_DEFAULT: meta["flag"] = _LANG_FLAG_DEFAULT[lang] health = _voice_health(p, meta=meta, duration=duration, transcript=transcript) return { "id": p.stem, "path": str(p), "file_type": p.suffix.lower().lstrip("."), "duration": duration, "loudness": loudness, "benchmark": benchmark, "has_ref": has_ref, "transcript": transcript, "has_picture": _picture_path(p) is not None, "health": health, "lang": lang, **meta, } def _voice_health( p: Path, *, meta: dict | None = None, duration: float | None = None, transcript: str | None = None, ) -> dict: if transcript is None: _has_ref, transcript = _read_reference_text(p) if duration is None: try: duration = float(_duration(p)) except Exception: duration = 0.0 else: duration = float(duration) if meta is None: meta = _load_meta(p) words = re.findall(r"\b[\w'-]+\b", transcript, flags=re.UNICODE) word_count = len(words) words_per_sec = (word_count / duration) if duration > 0 else 0.0 warnings: list[str] = [] if not transcript: warnings.append("Missing reference transcript") if duration > 25: warnings.append("Reference audio is longer than the recommended 10-20 seconds") if duration > 0 and word_count and words_per_sec > 4.5: warnings.append("Reference transcript is too long for the audio; re-transcribe or shorten it") loudness = meta.get("loudness", {}) peak = loudness.get("peak_dbfs") if peak is not None and float(peak) > -0.1: warnings.append("Reference audio is clipping or too loud") return { "ok": not warnings, "warnings": warnings, "duration": round(duration, 2) if duration else None, "word_count": word_count, "words_per_sec": round(words_per_sec, 2) if words_per_sec else 0, "loudness": loudness, } def _benchmark_advice(audio: Path, elapsed_sec: float | None, audio_sec: float | None, clipped: bool = False, error: str = "") -> tuple[bool, list[str]]: advice: list[str] = [] try: ref_duration = float(_duration(audio)) except Exception: ref_duration = 0.0 name = audio.stem.lower() if ref_duration >= 40 or "privat_" in name or "privat-" in name: advice.append("Avoid for real-time assistants; long clone samples often benchmark slowly. Trim or remake as a 10-20 second voice.") elif ref_duration > 25: advice.append("Reference is longer than recommended. Trim to a clean 10-20 second sample.") if clipped: advice.append("Generated output hit the max-duration guard. Re-transcribe the reference exactly or remake this clone.") if elapsed_sec is not None and elapsed_sec > 12: advice.append("Slow synthesis. Prefer a shorter optimized voice for Open WebUI or Home Assistant.") if audio_sec and elapsed_sec: rtf = elapsed_sec / max(audio_sec, 0.01) if rtf > 2.0: advice.append("High real-time factor. Use a shorter reference, normalize volume, and remove silence/noise.") if error: advice.append("Benchmark failed. Check that Qwen3-TTS has rescanned this voice and that the reference files are valid.") realtime_ok = not error and not clipped and (elapsed_sec or 999) <= 8 and ref_duration <= 25 return realtime_ok, advice def _benchmark_summary(runs: list[dict]) -> dict: ok_runs = [r for r in runs if r.get("ok")] if not ok_runs: return {} def avg(key: str) -> float | None: vals = [float(r[key]) for r in ok_runs if r.get(key) is not None] return round(sum(vals) / len(vals), 3) if vals else None return { "avg_ttfa_ms": avg("ttfa_ms"), "avg_total_sec": avg("total_sec"), "avg_audio_sec": avg("audio_sec"), "avg_rtf": avg("rtf"), "avg_speed": avg("speed"), } def _benchmark_voice(audio: Path, settings: dict, sentences: list[tuple[str, str]]) -> dict: runs: list[dict] = [] for label, text in sentences: try: runs.append(_tts_benchmark_request(text, audio.stem, settings, label)) except Exception as e: runs.append({"ok": False, "label": label, "text": text, "error": str(e)}) ok_runs = [r for r in runs if r.get("ok")] summary = _benchmark_summary(runs) elapsed = summary.get("avg_total_sec") if summary else None audio_sec = summary.get("avg_audio_sec") if summary else None errors = [r.get("error", "Benchmark failed") for r in runs if not r.get("ok")] realtime_ok, advice = _benchmark_advice(audio, elapsed, audio_sec, error="; ".join(errors)) if summary.get("avg_rtf") is not None and summary["avg_rtf"] > 2: advice.append("RTF is above 2.0. This voice is likely too slow for real-time assistants.") return { "ok": bool(ok_runs) and not errors, "realtime_ok": realtime_ok and not errors, "elapsed_sec": round(elapsed, 2) if elapsed is not None else None, "audio_sec": round(audio_sec, 2) if audio_sec is not None else None, "rtf": round(summary.get("avg_rtf"), 2) if summary.get("avg_rtf") is not None else None, "speed": round(summary.get("avg_speed"), 2) if summary.get("avg_speed") is not None else None, "ttfa_ms": round(summary.get("avg_ttfa_ms"), 0) if summary.get("avg_ttfa_ms") is not None else None, "bytes": sum(int(r.get("bytes") or 0) for r in ok_runs), "clipped": False, "benchmarked_at": datetime.now(timezone.utc).isoformat(), "text": sentences[0][1] if len(sentences) == 1 else "short / medium / long", "runs": runs, "summary": summary, "advice": advice, **({"error": "; ".join(errors)} if errors else {}), } # ── Static files ────────────────────────────────────────────────────────────── STATIC_DIR = Path(__file__).parent / "static" STATIC_DIR.mkdir(exist_ok=True) @app.get("/") async def index(): return FileResponse( STATIC_DIR / "index.html", headers={"Cache-Control": "no-store, max-age=0"}, ) @app.get("/favicon.ico") async def favicon(): return Response(status_code=204) # ── Upload ──────────────────────────────────────────────────────────────────── @app.post("/api/upload") async def upload(file: UploadFile = File(...)): suffix = Path(file.filename or "audio").suffix.lower() or ".bin" if suffix not in _UPLOAD_EXTS: raise HTTPException(400, "Unsupported audio/video file type") dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}" with dest.open("wb") as f: _copy_limited(file.file, f, _MAX_UPLOAD_BYTES) try: wav = _to_wav_24k(dest) except Exception as e: raise HTTPException(400, f"Audio conversion failed: {e}") fid = uuid.uuid4().hex _registry[fid] = wav return {"id": fid, "duration": _duration(wav), "filename": file.filename} @app.get("/api/route-sounds") async def list_route_sounds(): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) sounds = [] seen = set() for dirname in sorted(_SOUND_ASSET_DIRS): root = scan_dir / dirname if not root.exists() or not root.is_dir(): continue for path in sorted(root.rglob("*"), key=lambda p: str(p).lower()): if not path.is_file() or path.suffix.lower() not in _AUDIO_EXTS: continue try: safe = _safe_child_path(scan_dir, path) rel = str(safe.relative_to(scan_dir)) except Exception: continue if rel in seen: continue seen.add(rel) try: size = path.stat().st_size except Exception: size = None sounds.append({ "path": rel, "name": path.name, "folder": str(path.parent.relative_to(scan_dir)), "duration": None, "size": size, "type": path.suffix.lower().lstrip("."), }) return {"sounds": sounds} @app.get("/api/route-sounds/file/{sound_path:path}") async def route_sound_file(sound_path: str): try: path = _route_sound_path(_load_settings(), sound_path) except Exception: raise HTTPException(404, "Sound not found") if path is None: raise HTTPException(404, "Sound not found") mime = _AUDIO_MIME.get(path.suffix.lower(), "audio/wav") return FileResponse(str(path), media_type=mime, filename=path.name) @app.post("/api/route-sounds/upload") async def upload_route_sound(file: UploadFile = File(...), name: str = Form("")): suffix = Path(file.filename or "sound").suffix.lower() or ".bin" if suffix not in _AUDIO_EXTS: raise HTTPException(400, "Unsupported sound file type") raw_name = Path(name or file.filename or "sound").stem.strip() safe_name = re.sub(r"[^A-Za-z0-9_\-.]+", "_", raw_name).strip("._-")[:80] or f"sound_{uuid.uuid4().hex[:8]}" settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) sounds_dir = scan_dir / "sounds" sounds_dir.mkdir(parents=True, exist_ok=True) target = sounds_dir / f"{safe_name}.wav" tmp = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}" with tmp.open("wb") as f: _copy_limited(file.file, f, _MAX_SOUND_BYTES) try: segment = AudioSegment.from_file(str(tmp)).set_channels(1).set_sample_width(2).set_frame_rate(24000) segment.export(str(target), format="wav") duration = len(segment) / 1000.0 except Exception as e: raise HTTPException(400, f"Sound conversion failed: {e}") finally: try: tmp.unlink(missing_ok=True) except Exception: pass return {"ok": True, "path": str(target.relative_to(scan_dir)), "filename": target.name, "duration": round(duration, 2)} def _source_import_url(value: str) -> str: raw = str(value or "").strip() validator = globals().get("_validate_download_url") if validator: return validator(raw) if not re.match(r"^https?://", raw, re.I): raise HTTPException(400, "Expected an http(s) URL") return raw def _source_download_limit(name: str, fallback_mb: int) -> int: value = globals().get(name) if isinstance(value, int): return value return int(os.environ.get(name.lstrip("_").replace("_BYTES", "_MB"), str(fallback_mb))) * 1024 * 1024 @app.post("/api/import-source-audio") async def import_source_audio(request: Request): data = await request.json() audio_url = _source_import_url(str(data.get("audio_url") or "")) filename = str(data.get("name") or Path(urlparse(audio_url).path).name or "source-audio").strip() suffix = Path(urlparse(audio_url).path).suffix.lower() if suffix not in _AUDIO_EXTS: suffix = ".bin" dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}" try: with requests.get(audio_url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True) as r: r.raise_for_status() total = 0 with dest.open("wb") as f: for chunk in r.iter_content(1024 * 1024): if not chunk: continue total += len(chunk) if total > _source_download_limit("_MAX_UPLOAD_BYTES", 1024): raise HTTPException(413, "Downloaded audio is too large") f.write(chunk) wav = _to_wav_24k(dest) except HTTPException: raise except Exception as e: raise HTTPException(400, f"Source audio import failed: {e}") fid = uuid.uuid4().hex _registry[fid] = wav return {"id": fid, "duration": _duration(wav), "filename": filename or dest.name, "audio_url": audio_url} # ── Audio proxy (CORS bypass for external sources) ──────────────────────────── _PROXY_AUDIO_DOMAINS: set[str] = { "drive.usercontent.google.com", "drive.google.com", "aiartes.com", "freesound.org", "lanceblairvo.com", "raw.githubusercontent.com", "sample-files.com", } @app.get("/api/proxy-audio") async def proxy_audio(url: str): parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise HTTPException(400, "Only http/https URLs are supported") domain = parsed.netloc.lower().lstrip("www.") if not any(domain == d or domain.endswith("." + d) for d in _PROXY_AUDIO_DOMAINS): raise HTTPException(403, f"Domain not in audio proxy allowlist: {parsed.netloc}") try: resp = requests.get(url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True) resp.raise_for_status() except Exception as e: raise HTTPException(502, f"Proxy fetch failed: {e}") content_type = resp.headers.get("content-type", "audio/mpeg") def _stream(): for chunk in resp.iter_content(65536): if chunk: yield chunk return StreamingResponse( _stream(), media_type=content_type, headers={"Cache-Control": "public, max-age=3600", "Access-Control-Allow-Origin": "*"}, ) # ── Quick voice import (download + save directly to library) ────────────────── @app.post("/api/quick-import-voice") async def quick_import_voice(request: Request): data = await request.json() audio_url = _source_import_url(str(data.get("audio_url") or "")) voice_id = re.sub(r"[^A-Za-z0-9_\-\.]", "_", str(data.get("voice_id") or "").strip())[:80] if not voice_id: raise HTTPException(400, "voice_id is required") transcript = str(data.get("transcript") or "").strip() suffix = Path(urlparse(audio_url).path).suffix.lower() if suffix not in _AUDIO_EXTS: suffix = ".bin" dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}" try: with requests.get(audio_url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True) as r: r.raise_for_status() total = 0 with dest.open("wb") as f: for chunk in r.iter_content(1024 * 1024): if not chunk: continue total += len(chunk) if total > _source_download_limit("_MAX_UPLOAD_BYTES", 1024): raise HTTPException(413, "Downloaded audio is too large") f.write(chunk) wav = _to_wav_24k(dest) except HTTPException: raise except Exception as e: raise HTTPException(400, f"Audio download failed: {e}") settings = _load_settings() out_dir = _active_voices_dir(settings) out_dir.mkdir(parents=True, exist_ok=True) final_id = voice_id if (out_dir / f"{final_id}.wav").exists(): for i in range(2, 1000): candidate = f"{voice_id}_{i}" if not (out_dir / f"{candidate}.wav").exists(): final_id = candidate break wav_dest = out_dir / f"{final_id}.wav" _remove_audio_variants(out_dir, final_id) loudness = _export_normalized_wav(wav, wav_dest) if transcript: (out_dir / f"{final_id}.reference.txt").write_text(transcript, encoding="utf-8") meta = _load_meta(wav_dest) meta["enabled"] = True meta["loudness"] = loudness _save_meta(wav_dest, meta) return {"voice_id": final_id, "loudness": loudness} # ── YouTube download (SSE) ──────────────────────────────────────────────────── @app.get("/api/download-yt") async def download_yt(url: str): url = _validate_http_url(url, allow_private=_ALLOW_PRIVATE_DOWNLOADS) out_path = TEMP_DIR / f"{uuid.uuid4().hex}.%(ext)s" async def event_stream() -> AsyncGenerator[str, None]: cmd = [ "yt-dlp", "--extract-audio", "--audio-format", "wav", "--audio-quality", "0", "--output", str(out_path), "--no-playlist", "--progress", "--newline", url, ] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) downloaded_path: Path | None = None assert proc.stdout is not None async for raw in proc.stdout: line = raw.decode(errors="replace").rstrip() if not line: continue if "Destination:" in line and "[ExtractAudio]" not in line: m = re.search(r"Destination:\s+(.+)$", line) if m: downloaded_path = Path(m.group(1).strip()) if line.startswith("[download]") or line.startswith("[ExtractAudio]"): pct_m = re.search(r"(\d+\.\d+)%", line) pct = pct_m.group(1) if pct_m else None yield f"data: {json.dumps({'msg': line, 'pct': pct})}\n\n" await proc.wait() if downloaded_path is None or not downloaded_path.exists(): matches = glob.glob(str(out_path).replace("%(ext)s", "*")) if matches: downloaded_path = Path(matches[0]) if downloaded_path is None or not downloaded_path.exists(): yield f"data: {json.dumps({'error': 'Download failed — no output file found'})}\n\n" return try: wav = _to_wav_24k(downloaded_path) except Exception as e: yield f"data: {json.dumps({'error': f'Conversion failed: {e}'})}\n\n" return fid = uuid.uuid4().hex _registry[fid] = wav yield f"data: {json.dumps({'done': True, 'id': fid, 'duration': _duration(wav)})}\n\n" return StreamingResponse(event_stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) # ── Audio serving ───────────────────────────────────────────────────────────── @app.get("/api/audio/{fid}") async def serve_audio(fid: str): path = _registry.get(fid) if path is None or not path.exists(): raise HTTPException(404, "File not found") return FileResponse(str(path), media_type="audio/wav") # ── Process (trim + convert) ────────────────────────────────────────────────── @app.post("/api/auto-trim") async def auto_trim(request: Request): data = await request.json() fid: str = data["id"] src = _registry.get(fid) if src is None or not src.exists(): raise HTTPException(404, "Source file not found") try: return _auto_trim_bounds(src) except Exception as e: raise HTTPException(400, f"Auto trim failed: {e}") @app.post("/api/process") async def process(request: Request): data = await request.json() fid: str = data["id"] start = float(data.get("start", 0)) end: float | None = data.get("end") src = _registry.get(fid) if src is None or not src.exists(): raise HTTPException(404, "Source file not found") dur = _duration(src) if end is None or end <= start: end = dur trimmed = _trim(src, start, end) nid = uuid.uuid4().hex _registry[nid] = trimmed return {"id": nid, "duration": _duration(trimmed)} # ── Transcribe ──────────────────────────────────────────────────────────────── _STT_BACKEND_ALIASES = { "": "configured", "default": "configured", "whisper": "configured", "configured_whisper": "configured", "configured_stt": "configured", "parakeet": "nvidia_parakeet", "nvidia": "nvidia_parakeet", "nvidia_asr": "nvidia_parakeet", "nvidia_parakeet_asr": "nvidia_parakeet", "router": "nvidia_router", "speech_router": "nvidia_router", "nvidia_speech_router": "nvidia_router", "faster_whisper_server": "faster_whisper", "faster-whisper": "faster_whisper", "ctranslate2": "faster_whisper", "faster_w": "faster_whisper", "whisper-cpp": "whisper_cpp", "whisper_cpp_server": "whisper_cpp", "cpp": "whisper_cpp", "groq": "groq_whisper", "groq_stt": "groq_whisper", "groq-whisper": "groq_whisper", } _STT_BACKEND_METRICS: dict[str, dict] = { "configured": {"speed": "GPU / CPU", "latency": "1–5 s", "quality": "large-v3", "ram": "3 GB VRAM"}, "faster_whisper": {"speed": "~70× RT · GPU", "latency": "0.5–2 s", "quality": "large-v3", "ram": "1.5 GB VRAM"}, "whisper_cpp": {"speed": "~8–15× RT · CPU", "latency": "1–5 s", "quality": "large-v3 Q5", "ram": "~1 GB RAM"}, "groq_whisper": {"speed": "fastest cloud", "latency": "0.5–1 s", "quality": "Whisper Turbo", "ram": "cloud · 0"}, "nvidia_parakeet":{"speed": "~200× RT · GPU", "latency": "<0.3 s", "quality": "Parakeet-TDT", "ram": "~3 GB"}, "nvidia_router": {"speed": "GPU routed", "latency": "~0.5 s", "quality": "varies", "ram": "~11 GB"}, } _STT_VALID_BACKENDS = {"configured", "nvidia_parakeet", "nvidia_router", "faster_whisper", "whisper_cpp", "groq_whisper"} def _clean_stt_backend(value: str) -> str: key = re.sub(r"[^a-z0-9]+", "_", str(value or "configured").lower()).strip("_") key = _STT_BACKEND_ALIASES.get(key, key) return key if key in _STT_VALID_BACKENDS else "configured" def _stt_backend_url(settings: dict, backend: str) -> str: backend = _clean_stt_backend(backend) if backend == "nvidia_parakeet": return settings.get("nvidia_asr_url") or _NVIDIA_ASR_DEFAULT if backend == "nvidia_router": return settings.get("nvidia_router_url") or _NVIDIA_ROUTER_DEFAULT if backend == "faster_whisper": return settings.get("faster_whisper_url") or _FASTER_WHISPER_DEFAULT if backend == "whisper_cpp": return settings.get("whisper_cpp_url") or _WHISPER_CPP_DEFAULT if backend == "groq_whisper": return _GROQ_STT_ENDPOINT return settings.get("whisper_url") or _WHISPER_DEFAULT def _stt_backend_model(backend: str) -> str: backend = _clean_stt_backend(backend) if backend in {"nvidia_parakeet", "nvidia_router"}: return "whisper-1" if backend == "whisper_cpp": return "whisper-1" if backend == "groq_whisper": return "whisper-large-v3-turbo" return "large-v3" def _stt_backend_label(backend: str, url: str) -> str: labels = { "configured": "Configured Whisper/STT", "nvidia_parakeet": "NVIDIA Parakeet ASR", "nvidia_router": "NVIDIA Speech Router", "faster_whisper": "faster-whisper (CTranslate2 GPU)", "whisper_cpp": "whisper.cpp (CPU/CUDA)", "groq_whisper": "Groq Whisper (cloud · free)", } port = _backend_port_label(url) label = labels.get(backend, backend) return f"{port} {label}" if port else label def _stt_backend_api_key(settings: dict, backend: str) -> str: if backend == "groq_whisper": return settings.get("groq_api_key", "").strip() return settings.get("whisper_api_key", "").strip() def _stt_backend_health(url: str) -> tuple[bool, list[str]]: base = _validate_http_url(url, allow_private=True).rstrip("/") models: list[str] = [] ok = False try: r = requests.get(f"{base}/health", timeout=2) ok = r.status_code == 200 except Exception: pass try: r = requests.get(f"{base}/v1/models", timeout=3) if r.status_code == 200: ok = True payload = r.json() data = payload.get("data", []) if isinstance(payload, dict) else [] for item in data: if isinstance(item, dict) and item.get("id"): models.append(str(item["id"])) elif isinstance(item, str): models.append(item) except Exception: pass return ok, models @app.get("/api/stt-backends") async def stt_backends(): settings = _load_settings() items = [] seen_urls: set[tuple[str, str]] = set() ordered = ("configured", "faster_whisper", "whisper_cpp", "groq_whisper", "nvidia_parakeet", "nvidia_router") for backend in ordered: raw_url = _stt_backend_url(settings, backend) url = _validate_http_url(raw_url, allow_private=True).rstrip("/") key = (backend, url) if key in seen_urls: continue seen_urls.add(key) api_key = _stt_backend_api_key(settings, backend) if backend == "groq_whisper": ok = bool(api_key) models: list[str] = ["whisper-large-v3-turbo", "whisper-large-v3", "distil-whisper-large-v3-en"] else: ok, models = _stt_backend_health(url) items.append({ "id": backend, "label": _stt_backend_label(backend, url), "url": url, "port": _backend_port_label(url), "available": ok, "model": _stt_backend_model(backend), "models": models, "metrics": _STT_BACKEND_METRICS.get(backend, {}), }) return {"backends": items} def _transcription_text_from_response(resp: requests.Response) -> str: try: payload = resp.json() if isinstance(payload, str): return payload.strip() if isinstance(payload, dict): for key in ("text", "transcript", "transcription"): if payload.get(key) is not None: return str(payload[key]).strip() except Exception: pass return resp.text.strip() def _transcribe_audio(src: Path, settings: dict, backend: str = "configured") -> tuple[str, str]: backend = _clean_stt_backend(backend) stt_url = _validate_http_url(_stt_backend_url(settings, backend), allow_private=True).rstrip("/") stt_key = _stt_backend_api_key(settings, backend) hdrs = {"Authorization": f"Bearer {stt_key}"} if stt_key else {} model = _stt_backend_model(backend) with src.open("rb") as f: resp = requests.post( f"{stt_url}/v1/audio/transcriptions", files={"file": ("audio.wav", f, "audio/wav")}, data={"model": model, "response_format": "text"}, headers=hdrs, timeout=_STT_REQUEST_TIMEOUT, ) if resp.status_code in {400, 404, 422} and model != "whisper-1": with src.open("rb") as f: resp = requests.post( f"{stt_url}/v1/audio/transcriptions", files={"file": ("audio.wav", f, "audio/wav")}, data={"model": "whisper-1", "response_format": "text"}, headers=hdrs, timeout=60, ) resp.raise_for_status() return _transcription_text_from_response(resp), backend @app.post("/api/transcribe") async def transcribe(request: Request): data = await request.json() fid: str = data["id"] src = _registry.get(fid) if src is None or not src.exists(): raise HTTPException(404, "Audio not found") settings = _load_settings() backend = _clean_stt_backend(str(data.get("backend") or data.get("stt_backend") or "configured")) try: text, used_backend = await asyncio.to_thread(_transcribe_audio, src, settings, backend) return {"text": text, "backend": used_backend} except Exception as e: raise HTTPException(502, f"STT error ({backend}): {e}") @app.post("/api/transcribe-bytes") async def transcribe_bytes( file: UploadFile = File(...), backend: str = Form("configured"), ): """Accept raw audio upload and return transcription directly (used by hotkey daemon).""" suffix = Path(file.filename or "audio.wav").suffix.lower() or ".wav" if suffix not in _AUDIO_EXTS: raise HTTPException(400, "Unsupported audio type") tmp = TEMP_DIR / f"{uuid.uuid4().hex}_daemon{suffix}" wav_tmp = tmp try: with tmp.open("wb") as f: _copy_limited(file.file, f, _MAX_UPLOAD_BYTES) if suffix != ".wav": wav_tmp = _to_wav_24k(tmp) settings = _load_settings() stt_backend = _clean_stt_backend(backend) text, used_backend = await asyncio.to_thread(_transcribe_audio, wav_tmp, settings, stt_backend) return {"text": text, "backend": used_backend} except HTTPException: raise except Exception as e: raise HTTPException(502, f"STT error: {e}") finally: for p in {tmp, wav_tmp}: try: p.unlink(missing_ok=True) except Exception: pass # ── Save voice ──────────────────────────────────────────────────────────────── @app.post("/api/save") async def save_voice(request: Request): data = await request.json() fid: str = data["id"] voice_id: str = data["voice_id"].strip() transcript: str = data.get("transcript", "").strip() if not voice_id: raise HTTPException(400, "Voice ID is required") if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id): raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .") src = _registry.get(fid) if src is None or not src.exists(): raise HTTPException(404, "Processed audio not found") settings = _load_settings() out_dir = _active_voices_dir(settings) out_dir.mkdir(parents=True, exist_ok=True) wav_dest = out_dir / f"{voice_id}.wav" txt_dest = out_dir / f"{voice_id}.reference.txt" _remove_audio_variants(out_dir, voice_id) loudness = _export_normalized_wav(src, wav_dest) txt_dest.write_text(transcript, encoding="utf-8") meta = _load_meta(wav_dest) meta["enabled"] = True meta["loudness"] = loudness _save_meta(wav_dest, meta) return {"voice_id": voice_id, "wav": str(wav_dest), "txt": str(txt_dest), "loudness": loudness} # ── Voice library ───────────────────────────────────────────────────────────── @app.post("/api/voice-load") @app.post("/api/voice/load") async def load_voice_for_edit(request: Request): data = await request.json() voice_id: str = data.get("voice_id", "") settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) audio = _voice_audio_from_request(data, scan_dir) if audio is None: raise HTTPException(404, f"Voice '{voice_id}' not found") try: wav = _to_wav_24k(audio) except Exception as e: raise HTTPException(400, f"Audio conversion failed: {e}") fid = uuid.uuid4().hex _registry[fid] = wav ref = audio.with_suffix(".reference.txt") return { "id": fid, "voice_id": voice_id or audio.stem, "path": str(audio), "duration": _duration(wav), "transcript": ref.read_text(encoding="utf-8").strip() if ref.exists() else "", "file_type": audio.suffix.lower().lstrip("."), } @app.post("/api/voice-replace") @app.post("/api/voice/replace") async def replace_voice_audio(request: Request): data = await request.json() fid: str = data["id"] voice_id: str = data["voice_id"].strip() transcript: str = data.get("transcript", "").strip() if not voice_id: raise HTTPException(400, "Voice ID is required") if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id): raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .") src = _registry.get(fid) if src is None or not src.exists(): raise HTTPException(404, "Processed audio not found") settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) current = _voice_audio_from_request(data, scan_dir) if current is None: raise HTTPException(404, f"Voice '{voice_id}' not found") parent = current.parent old_id = current.stem meta = _load_meta(current) wav_dest = parent / f"{voice_id}.wav" txt_dest = parent / f"{voice_id}.reference.txt" if voice_id != old_id: old_files = {p.resolve() for p in _voice_package_paths(current)} suffixes = _AUDIO_EXTS + [".reference.txt", ".meta.json"] + _PICTURE_EXTS for sfx in suffixes: target = parent / f"{voice_id}{sfx}" if target.exists() and target.resolve() not in old_files: raise HTTPException(409, f"Voice '{voice_id}' already exists") backup = _backup_original_voice(current) if voice_id == old_id else None _remove_audio_variants(parent, voice_id, keep=wav_dest) loudness = _export_normalized_wav(src, wav_dest) txt_dest.write_text(transcript, encoding="utf-8") meta["loudness"] = loudness if backup: meta["original_backup"] = str(backup) meta["needs_tts_restart"] = True _save_meta(wav_dest, meta) if voice_id != old_id: keep = {wav_dest, txt_dest, _meta_path(wav_dest)} for ext in _PICTURE_EXTS: old_pic = parent / f"{old_id}{ext}" if old_pic.exists(): new_pic = parent / f"{voice_id}{ext}" shutil.copy2(str(old_pic), str(new_pic)) keep.add(new_pic) _remove_voice_package(current, keep) return {"voice_id": voice_id, "wav": str(wav_dest), "txt": str(txt_dest), "duration": _duration(wav_dest), "file_type": "wav", "loudness": loudness, "backup": str(backup) if backup else None} @app.get("/api/voices") async def list_voices(): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) active_dir = _active_voices_dir(settings) hidden_dir = _hidden_voices_dir(settings) voices = [] if scan_dir.exists(): seen: set[str] = set() all_audio = list(_voice_audio_files(scan_dir)) def voice_sort_key(path: Path): try: path.relative_to(active_dir) folder_rank = 0 except ValueError: try: path.relative_to(hidden_dir) folder_rank = 1 except ValueError: folder_rank = 2 return (path.stem.lower(), folder_rank, str(path).lower()) for p in sorted(all_audio, key=voice_sort_key): if p.stem in seen: continue # prefer first extension found (wav beats mp3 etc.) seen.add(p.stem) entry = _voice_entry(p) try: p.relative_to(hidden_dir) entry["enabled"] = False except ValueError: try: p.relative_to(active_dir) entry["enabled"] = True except ValueError: pass voices.append(entry) return voices # ── Voice meta update ───────────────────────────────────────────────────────── @app.post("/api/voice/meta") async def update_voice_meta(request: Request): data = await request.json() voice_id: str = data["voice_id"] settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice_id, scan_dir) if wav is None: raise HTTPException(404, f"Voice '{voice_id}' not found") meta = _load_meta(wav) if "enabled" in data: enabled = bool(data["enabled"]) wav = _move_voice_package(wav, _active_voices_dir(settings) if enabled else _hidden_voices_dir(settings)) meta = _load_meta(wav) meta["enabled"] = enabled for field in ("note", "rating", "flag", "gender", "loudness", "persona"): if field in data: meta[field] = data[field] if "transcript" in data: meta["needs_tts_restart"] = True _save_meta(wav, meta) if "transcript" in data: wav.with_suffix(".reference.txt").write_text(data.get("transcript", "").strip(), encoding="utf-8") return {"ok": True, "path": str(wav), "enabled": meta.get("enabled", True), "transcript": wav.with_suffix(".reference.txt").read_text(encoding="utf-8").strip() if wav.with_suffix(".reference.txt").exists() else ""} @app.post("/api/voices/sync-folders") async def sync_voice_folders(): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) active_dir = _active_voices_dir(settings) hidden_dir = _hidden_voices_dir(settings) moved = {"active": 0, "hidden": 0} conflicts = [] if not scan_dir.exists(): return {"moved": moved, "conflicts": conflicts} seen: set[str] = set() all_audio = list(_voice_audio_files(scan_dir)) for p in sorted(all_audio, key=lambda x: x.stem.lower()): if p.stem in seen or not p.exists(): continue seen.add(p.stem) meta = _load_meta(p) enabled = meta.get("enabled") is not False target_dir = active_dir if enabled else hidden_dir try: new_audio = _move_voice_package(p, target_dir) meta = _load_meta(new_audio) meta["enabled"] = enabled _save_meta(new_audio, meta) moved["active" if enabled else "hidden"] += int(new_audio != p) except HTTPException as e: conflicts.append({"voice_id": p.stem, "detail": e.detail}) return {"moved": moved, "conflicts": conflicts, "active_dir": str(active_dir), "hidden_dir": str(hidden_dir)} @app.post("/api/voices/calculate-db") async def calculate_voice_db(): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) results = [] errors = [] if not scan_dir.exists(): return {"calculated": 0, "errors": errors, "voices": results} seen: set[str] = set() all_audio = list(_voice_audio_files(scan_dir)) for audio in sorted(all_audio, key=lambda p: p.stem.lower()): if audio.stem in seen: continue seen.add(audio.stem) try: loudness = _loudness_info(audio) meta = _load_meta(audio) meta["loudness"] = loudness _save_meta(audio, meta) results.append({"voice_id": audio.stem, "path": str(audio), "loudness": loudness}) except Exception as e: errors.append({"voice_id": audio.stem, "detail": str(e)}) return {"calculated": len(results), "errors": errors, "voices": results, "target_dbfs": _VOICE_TARGET_DBFS, "peak_dbfs": _VOICE_PEAK_DBFS} @app.post("/api/voices/benchmark") async def benchmark_voices(request: Request): data = await request.json() active_only = bool(data.get("active_only", True)) limit = int(data.get("limit") or 0) sample_text = str(data.get("text") or "").strip() sentences = [("sample", sample_text)] if sample_text else list(_BENCHMARK_SENTENCES) settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) active_dir = _active_voices_dir(settings) results = [] errors = [] if not scan_dir.exists(): return { "benchmarked": 0, "errors": errors, "voices": results, "text": sample_text or "short / medium / long", "sentences": [{"label": label, "text": sent_text} for label, sent_text in sentences], "active_only": active_only, "max_tts_output_seconds": _MAX_TTS_OUTPUT_SECONDS, } target_voice = str(data.get("voice_id") or "").strip() seen: set[str] = set() if target_voice: audio = _find_voice_audio(target_voice, scan_dir) if audio is None: raise HTTPException(404, f"Voice '{target_voice}' not found") all_audio = [audio] else: all_audio = list(_voice_audio_files(active_dir if active_only and active_dir.exists() else scan_dir)) for audio in sorted(all_audio, key=lambda p: p.stem.lower()): if audio.stem in seen: continue seen.add(audio.stem) if limit and len(results) >= limit: break benchmark = await asyncio.to_thread(_benchmark_voice, audio, settings, sentences) try: meta = _load_meta(audio) meta["benchmark"] = benchmark _save_meta(audio, meta) except Exception as e: errors.append({"voice_id": audio.stem, "detail": f"Could not save benchmark: {e}"}) item = {"voice_id": audio.stem, "path": str(audio), "benchmark": benchmark} results.append(item) if not benchmark.get("ok", False): errors.append({"voice_id": audio.stem, "detail": benchmark.get("error", "Benchmark failed")}) return { "benchmarked": len(results), "errors": errors, "voices": results, "text": sample_text or "short / medium / long", "sentences": [{"label": label, "text": sent_text} for label, sent_text in sentences], "active_only": active_only, "max_tts_output_seconds": _MAX_TTS_OUTPUT_SECONDS, } @app.post("/api/voices/normalize-active") async def normalize_active_voices(): settings = _load_settings() active_dir = _active_voices_dir(settings) if not active_dir.exists(): return {"normalized": 0, "skipped": 0, "errors": [], "target_dbfs": _VOICE_TARGET_DBFS} normalized = [] skipped = [] errors = [] for wav in sorted((p for p in active_dir.rglob("*.wav") if not _is_internal_voice_file(p)), key=lambda p: p.stem.lower()): try: tmp = wav.with_suffix(".normalized.tmp.wav") loudness = _export_normalized_wav(wav, tmp) shutil.move(str(tmp), str(wav)) meta = _load_meta(wav) meta["enabled"] = True meta["loudness"] = loudness _save_meta(wav, meta) normalized.append({"voice_id": wav.stem, **loudness}) except Exception as e: errors.append({"voice_id": wav.stem, "detail": str(e)}) try: tmp = wav.with_suffix(".normalized.tmp.wav") if tmp.exists(): tmp.unlink() except Exception: pass for ext in [e for e in _AUDIO_EXTS if e != ".wav"]: skipped.extend(str(p) for p in active_dir.rglob(f"*{ext}") if not _is_internal_voice_file(p)) return { "normalized": len(normalized), "skipped": len(skipped), "errors": errors, "target_dbfs": _VOICE_TARGET_DBFS, "peak_dbfs": _VOICE_PEAK_DBFS, "voices": normalized, } @app.post("/api/voice/normalize") async def normalize_voice(request: Request): data = await request.json() target_dbfs = float(data.get("target_dbfs", _VOICE_TARGET_DBFS)) settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) audio = _voice_audio_from_request(data, scan_dir) if audio is None: raise HTTPException(404, "Voice not found") if audio.suffix.lower() != ".wav": raise HTTPException(400, "Only WAV voices can be normalized in place") tmp = audio.with_suffix(".normalized.tmp.wav") try: loudness = _export_normalized_wav(audio, tmp, target_dbfs=target_dbfs) shutil.move(str(tmp), str(audio)) meta = _load_meta(audio) meta["loudness"] = loudness meta["needs_tts_restart"] = True _save_meta(audio, meta) return {"ok": True, "voice_id": audio.stem, "path": str(audio), "duration": _duration(audio), "file_type": "wav", "loudness": loudness} except Exception as e: if tmp.exists(): tmp.unlink() raise HTTPException(400, f"Normalize failed: {e}") @app.post("/api/voice/undo") async def undo_voice_edit(request: Request): data = await request.json() settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) current = _voice_audio_from_request(data, scan_dir) if current is None: raise HTTPException(404, "Voice not found") meta = _load_meta(current) backup = next((p for p in _backup_candidates(current, meta) if p.exists()), None) if backup is None: raise HTTPException(404, "No original backup found") original_suffix = _backup_audio_suffix(backup, current.stem) or current.suffix.lower() restored = current.with_suffix(original_suffix) shutil.copy2(str(backup), str(restored)) if restored.resolve() != current.resolve(): _remove_audio_variants(current.parent, current.stem, keep=restored) try: meta["loudness"] = _loudness_info(restored) except Exception: meta.pop("loudness", None) meta["needs_tts_restart"] = True _save_meta(restored, meta) return { "ok": True, "voice_id": restored.stem, "path": str(restored), "duration": _duration(restored), "file_type": restored.suffix.lower().lstrip("."), "loudness": meta.get("loudness", {}), } # ── Voice rename ────────────────────────────────────────────────────────────── @app.post("/api/voice/rename") async def rename_voice(request: Request): data = await request.json() old_id: str = data["old_id"] new_id: str = data["new_id"].strip() if not re.match(r"^[A-Za-z0-9_\-\.]+$", new_id): raise HTTPException(400, "Invalid voice ID characters") settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(old_id, scan_dir) if wav is None: raise HTTPException(404, f"Voice '{old_id}' not found") parent = wav.parent if any((parent / f"{new_id}{ext}").exists() for ext in _AUDIO_EXTS): raise HTTPException(409, f"Voice '{new_id}' already exists") meta = _load_meta(wav) backup = next((p for p in _backup_candidates(wav, meta) if p.exists()), None) new_audio = parent / f"{new_id}{wav.suffix.lower()}" new_backup = _backup_path(new_audio) if backup and backup.exists() and backup.resolve() != new_backup.resolve() and new_backup.exists(): raise HTTPException(409, f"Backup already exists for '{new_id}'") suffixes = _AUDIO_EXTS + [".reference.txt", ".meta.json"] + _PICTURE_EXTS for sfx in suffixes: src = parent / f"{old_id}{sfx}" if src.exists(): src.rename(parent / f"{new_id}{sfx}") if backup and backup.exists(): if backup.resolve() != new_backup.resolve(): backup.rename(new_backup) new_meta = _load_meta(new_audio) new_meta["original_backup"] = str(new_backup) _save_meta(new_audio, new_meta) return {"new_id": new_id, "path": str(new_audio), "file_type": new_audio.suffix.lower().lstrip(".")} # ── Voice delete ───────────────────────────────────────────────────────────── @app.delete("/api/voice/{voice_id}") async def delete_voice(voice_id: str): if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id): raise HTTPException(400, "Invalid voice ID") settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice_id, scan_dir) if wav is None: raise HTTPException(404, f"Voice '{voice_id}' not found") deleted = [] for f in _voice_package_paths(wav): if f.exists(): f.unlink() deleted.append(f.name) return {"deleted": deleted} # ── Voice picture upload ────────────────────────────────────────────────────── @app.post("/api/voice/picture") async def upload_picture(voice_id: str = Form(...), file: UploadFile = File(...)): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice_id, scan_dir) if wav is None: raise HTTPException(404, f"Voice '{voice_id}' not found") orig_suffix = Path(file.filename or "photo.jpg").suffix.lower() if orig_suffix not in _PICTURE_EXTS: orig_suffix = ".jpg" # Remove any existing picture first for ext in _PICTURE_EXTS: old = wav.with_suffix(ext) if old.exists(): old.unlink() dest = wav.with_suffix(orig_suffix) with dest.open("wb") as f: _copy_limited(file.file, f, _MAX_PICTURE_BYTES) return {"ok": True, "path": str(dest)} @app.post("/api/voice/picture-url") async def upload_picture_url(request: Request): data = await request.json() voice_id = str(data.get("voice_id") or "").strip() image_url = _source_import_url(str(data.get("image_url") or "")) if not voice_id or not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id): raise HTTPException(400, "Invalid voice ID") settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice_id, scan_dir) if wav is None: raise HTTPException(404, f"Voice '{voice_id}' not found") suffix = Path(urlparse(image_url).path).suffix.lower() if suffix not in _PICTURE_EXTS: suffix = ".jpg" dest = wav.with_suffix(suffix) try: with requests.get(image_url, headers=_VOICE_SOURCE_HEADERS, timeout=20, stream=True) as r: r.raise_for_status() content_type = (r.headers.get("content-type") or "").split(";", 1)[0].lower() if content_type and not content_type.startswith("image/"): raise HTTPException(400, "Image URL did not return an image") for ext in _PICTURE_EXTS: old = wav.with_suffix(ext) if old.exists(): old.unlink() total = 0 with dest.open("wb") as f: for chunk in r.iter_content(256 * 1024): if not chunk: continue total += len(chunk) if total > _source_download_limit("_MAX_PICTURE_BYTES", 10): raise HTTPException(413, "Image file is too large") f.write(chunk) except HTTPException: raise except Exception as e: raise HTTPException(400, f"Image import failed: {e}") return {"ok": True, "voice_id": voice_id, "path": str(dest)} # ── Voice picture serve ─────────────────────────────────────────────────────── @app.get("/api/voice/picture/{voice_id}") async def serve_picture(voice_id: str): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice_id, scan_dir) if wav is None: raise HTTPException(404, "Voice not found") pic = _picture_path(wav) if pic is None: raise HTTPException(404, "No picture") return FileResponse(str(pic), media_type=_picture_mime(pic)) # ── Serve voice WAV from library ────────────────────────────────────────────── @app.get("/api/voice-file") async def voice_file(path: str): settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) p = _safe_child_path(scan_dir, Path(path)) if not p.exists() or not p.is_file(): raise HTTPException(404, "File not found") if p.suffix.lower() not in _AUDIO_EXTS: raise HTTPException(404, "File not found") mime = _AUDIO_MIME.get(p.suffix.lower(), "audio/wav") return FileResponse( str(p), media_type=mime, headers={"Cache-Control": "no-store, max-age=0"}, ) # -- External voice source discovery ----------------------------------------- _VOICE_SOURCE_HEADERS = { "User-Agent": "TTS Voice Creator voice-source-scraper/1.0 (+local)", "Accept": "text/html,application/json,text/plain;q=0.9,*/*;q=0.8", } DEFAULT_VOICE_SOURCE_URLS = [ "https://aiartes.com/voiceai", "https://sample-files.com/downloads/audio/wav/voice-sample.wav", "https://freesound.org/people/Scott%20Simpson/", "https://lanceblairvo.com/raw-voiceover-samples/", "https://github.com/yaph/tts-samples/tree/main/mp3", "https://github.com/jim-schwoebel/voice_datasets", ] _DIRECT_AUDIO_RE = re.compile(r"\.(?:mp3|wav|ogg|flac|m4a|aac)(?:$|[?#])", re.I) _LANG_HINTS = { "english": "English", "german": "German", "deutsch": "German", "french": "French", "spanish": "Spanish", "italian": "Italian", "portuguese": "Portuguese", "dutch": "Dutch", "polish": "Polish", "russian": "Russian", "japanese": "Japanese", "korean": "Korean", "chinese": "Chinese", "arabic": "Arabic", "swedish": "Swedish", "turkish": "Turkish", "hindi": "Hindi", } def _source_get(url: str, timeout: int = 12) -> requests.Response: r = requests.get(url, headers=_VOICE_SOURCE_HEADERS, timeout=timeout) r.raise_for_status() return r def _plain_text(value: str) -> str: value = re.sub(r"<[^>]+>", " ", str(value or "")) value = unescape(value) return re.sub(r"\s+", " ", value).strip() def _voice_source_id(url: str) -> str: parsed = urlparse(url) base = (parsed.netloc + parsed.path).strip("/").lower() base = re.sub(r"[^a-z0-9]+", "-", base).strip("-") return base[:80] or "source" def _source_name_from_url(url: str) -> str: parsed = urlparse(url) host = parsed.netloc.replace("www.", "") tail = Path(parsed.path.rstrip("/")).stem.replace("-", " ").replace("_", " ").strip() return f"{host} / {tail}" if tail else host or url def _guess_language(*values: str) -> str: text = " ".join(str(v or "") for v in values).lower() for key, label in _LANG_HINTS.items(): if re.search(rf"\b{re.escape(key)}\b", text): return label code = re.search(r"(?:^|[^a-z])(en|de|fr|es|it|pt|nl|pl|ru|ja|ko|zh|ar|sv|tr|hi)(?:[^a-z]|$)", text) return {"en": "English", "de": "German", "fr": "French", "es": "Spanish", "it": "Italian", "pt": "Portuguese", "nl": "Dutch", "pl": "Polish", "ru": "Russian", "ja": "Japanese", "ko": "Korean", "zh": "Chinese", "ar": "Arabic", "sv": "Swedish", "tr": "Turkish", "hi": "Hindi"}.get(code.group(1), "Unknown") if code else "Unknown" def _guess_gender(*values: str) -> str: text = " ".join(str(v or "") for v in values).lower() if re.search(r"\b(female|woman|girl|fem|_f_|-f-)\b", text): return "Female" if re.search(r"\b(male|man|boy|masc|_m_|-m-)\b", text): return "Male" return "Unknown" def _source_item(source_id: str, source_name: str, name: str, kind: str, page_url: str, audio_url: str = "", image_url: str = "", category: str = "", description: str = "", file_type: str = "", language: str = "", gender: str = "") -> dict: if not file_type and audio_url: match = re.search(r"\.([A-Za-z0-9]+)(?:$|[?#])", audio_url) file_type = match.group(1).lower() if match else "audio" language = language or _guess_language(name, kind, category, description, page_url, audio_url) gender = gender or _guess_gender(name, kind, category, description, page_url, audio_url) return { "id": f"{source_id}:{uuid.uuid5(uuid.NAMESPACE_URL, page_url + audio_url + name + kind)}", "source_id": source_id, "source": source_name, "name": _plain_text(name)[:160], "kind": _plain_text(kind)[:80], "category": _plain_text(category)[:80], "description": _plain_text(description)[:420], "page_url": page_url, "audio_url": audio_url, "image_url": image_url, "file_type": file_type, "language": language, "gender": gender, "direct_audio": bool(audio_url), } def _source_result(source_id: str, name: str, homepage: str, description: str, items: list[dict]) -> dict: return { "id": source_id, "name": name, "homepage": homepage, "description": description, "items": items, "count": len(items), "direct_audio": sum(1 for item in items if item.get("direct_audio")), } def _direct_audio_source(url: str) -> dict: source_id = _voice_source_id(url) name = _source_name_from_url(url) item = _source_item( source_id, name, Path(urlparse(url).path).stem.replace("-", " ").replace("_", " ") or "Voice sample", "Direct audio file", url, audio_url=url, category="Direct audio", description="Direct audio URL from the editable source list.", ) return _source_result(source_id, name, url, "Single direct audio URL.", [item]) def _generic_audio_page_source(url: str) -> dict: source_id = _voice_source_id(url) source_name = _source_name_from_url(url) body = _source_get(url).text title = re.search(r"