"""TTS backends, preview, style-variation, streaming, voice-design, OpenAI-compat proxy.""" from __future__ import annotations import asyncio import io import json import logging import os import re import threading import time import uuid from pathlib import Path from urllib.parse import quote import requests from fastapi import APIRouter, HTTPException, Request from fastapi.responses import Response, StreamingResponse from core.config import _load_settings, _clean_preview_backend, _preview_backend_base_url from core.constants import ( _VOICES_DIR_DEFAULT, _TTS_CONTAINER, _TTS_CONTAINERS_RAW, _routing_log_add, CONFIG_DIR, _VOICE_PEAK_DBFS, ) from core.routing import ( _load_tts_routes, _resolve_tts_route, _route_backend, _canonical_app_name, _request_app_name, _detect_text_language, _routing_log_request, ) from core.presets import _load_design_presets, _virtual_voice_id, _resolve_virtual_voice from core.voice import ( _find_voice_audio, _active_voices_dir, _voice_audio_files, _load_meta, _save_meta, _AUDIO_EXTS, _read_reference_text, _voice_health, ) from core.voice_index import indexed_voices, refresh_voice_index_background from core.validation import _validate_http_url from core.registry import _registry_put, TEMP_DIR from core.tts_helpers import ( _tts_request_audio, _open_tts_stream_response, _read_tts_stream_response, _iter_tts_stream_response, _preview_request_audio, _prepare_proxy_audio, _requested_response_format, _audio_ext_media, _apply_route_sounds, _route_sound_path, _voice_design_request_audio, _voice_design_voice_request_audio, _nvidia_clone_request_audio, _parse_voice_design_dialogue, _infer_voice_design_gender, _voice_design_dialogue_request_audio, _fishspeech_request_audio, ) from core.audio import _duration, _export_normalized_wav from core.docker_client import _docker_post logger = logging.getLogger("uvicorn.error") router = APIRouter() _SEED_FINDER_VOICE_LOCK = threading.Lock() # ── TTS backend helpers ─────────────────────────────────────────────────────── def _backend_port_label(url: str) -> str: from urllib.parse import urlsplit try: parts = urlsplit(url) if parts.port: return str(parts.port) except Exception: pass return "" def _backend_display_name(backend: str, url: str) -> str: names = { "voice_clone": "Voice Clone/Base (WAV File)", "voice_design": "Voice Design", "customvoice": "CustomVoice", "streaming": "Streaming (WAV File)", "nvidia_magpie": "NVIDIA Magpie TTS", "nvidia_zeroshot":"NVIDIA Magpie Zeroshot Clone", "nvidia_flow": "NVIDIA Magpie Flow Clone", "kokoro": "Kokoro FastAPI (82M)", "vibevoice": "VibeVoice TTS", "xtts": "XTTS v2", "fishspeech": "Fish Audio S2 Pro (Clone)", } port = _backend_port_label(url) return f"{port} {names.get(backend, backend)}" if port else names.get(backend, backend) def _backend_capabilities(backend: str) -> dict: caps = { "voice_clone": { "purpose": "Clone a speaker from a short WAV/reference clip.", "identity": "Strongest match to saved WAV voices.", "style": "Weak per-request style; instruct may be ignored.", "best_for": "Known voices, multilingual cloning, benchmarks, and reliable speaker identity.", "uses_wav": True, "style_aware": False, "true_streaming": False, "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "Premium clone", "ram": "~6 GB", }, "voice_design": { "purpose": "Create or reuse prompt-designed voices from natural-language descriptions.", "identity": "Prompt persona, not the selected WAV speaker unless you first export/clone it.", "style": "Strong style and emotion control through instruct text.", "best_for": "New characters, personas, dialogue, and designing reference WAVs to clone later.", "uses_wav": False, "style_aware": True, "true_streaming": False, "speed": "~0.4× GPU", "latency": "1–3 s", "quality": "Premium", "ram": "~5 GB", }, "customvoice": { "purpose": "Generate speech with the CustomVoice model voices.", "identity": "Uses CustomVoice speakers, not arbitrary active WAV voices unless trained/configured there.", "style": "Good per-request style and emotion control.", "best_for": "Controlled style with configured CustomVoice speakers.", "uses_wav": False, "style_aware": True, "true_streaming": False, "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "Premium", "ram": "~5 GB", }, "streaming": { "purpose": "Low-latency playback from saved WAV/reference voices.", "identity": "Same WAV voice identity path as Base.", "style": "Weak per-request style in the current streaming server.", "best_for": "Long text, assistants, Open WebUI/SillyTavern playback that can start before completion.", "uses_wav": True, "style_aware": False, "true_streaming": True, "speed": "~0.1× GPU", "latency": "0.5–1 s", "quality": "Premium", "ram": "~5 GB", }, "nvidia_magpie": { "purpose": "Generate speech with NVIDIA Magpie fixed speaker voices.", "identity": "Uses Magpie speaker aliases such as sofia, aria, jason, leo, and john; it is not a WAV voice-cloning model.", "style": "Language and speaker are controlled by the backend voice config; per-request style text is usually ignored.", "best_for": "Fast local NVIDIA TTS voices and OpenAI-compatible assistant playback.", "uses_wav": False, "style_aware": False, "true_streaming": False, "speed": "~0.05× GPU", "latency": "0.3–0.8 s", "quality": "High", "ram": "~10 GB", }, "nvidia_zeroshot": { "purpose": "Clone a saved library voice through NVIDIA Magpie TTS Zeroshot NIM.", "identity": "Sends the selected WAV as audio_prompt; no prompt transcript is required.", "style": "Best with a clear 3-10 second prompt. Optional quality params can be configured in Settings.", "best_for": "Fast NVIDIA reference-audio cloning, streaming-class use cases, live agents, and games.", "uses_wav": True, "style_aware": False, "true_streaming": False, "speed": "~0.1× GPU", "latency": "0.5–1 s", "quality": "High clone", "ram": "~10 GB", }, "nvidia_flow": { "purpose": "Clone a saved library voice through NVIDIA Magpie TTS Flow NIM.", "identity": "Sends the selected WAV plus its exact saved reference transcript.", "style": "Offline high-fidelity clone path; prompt transcript must match the reference audio.", "best_for": "Studio-style dubbing, narration, and podcast-quality offline generation.", "uses_wav": True, "style_aware": False, "true_streaming": False, "speed": "~0.2× GPU", "latency": "1–2 s", "quality": "Studio", "ram": "~10 GB", }, "kokoro": { "purpose": "High-quality English TTS with Kokoro 82M model. OpenAI-compatible endpoint.", "identity": "Uses Kokoro built-in voices (af_bella, bf_emma, am_adam, …); no WAV cloning.", "style": "Voice selection via voice ID. Style instruction is not supported.", "best_for": "Fast, high-quality CPU TTS. Low RAM footprint. Easy local Docker setup.", "uses_wav": False, "style_aware": False, "true_streaming": False, "speed": "~0.1× CPU", "latency": "0.2–0.5 s", "quality": "High (82M)", "ram": "300 MB CPU", }, "vibevoice": { "purpose": "VibeVoice TTS service. Simple text-in, audio-out REST endpoint.", "identity": "Single built-in voice; no WAV cloning or voice selection.", "style": "Text only — no voice ID or style parameters.", "best_for": "Lightweight local TTS on port 8027. Minimal setup, fast response.", "uses_wav": False, "style_aware": False, "true_streaming": False, "speed": "fast", "latency": "0.2–1 s", "quality": "High", "ram": "~4 GB", }, "xtts": { "purpose": "XTTS v2 via xtts-api-server. OpenAI-compatible endpoint with speaker selection.", "identity": "Uses speakers registered in the XTTS server; not WAV voice cloning.", "style": "Speaker selected by voice ID. Style instruction not supported.", "best_for": "Local multi-speaker TTS with XTTS v2 model. Coqui/daswer123 docker setup.", "uses_wav": False, "style_aware": False, "true_streaming": False, "speed": "~0.3× GPU", "latency": "1–3 s", "quality": "High", "ram": "~5 GB", }, # Measured live against this exact checkpoint (fishaudio/s2-pro, rev 1de9996b): # inline [tag] emotion markers work well ONLY when no reference audio is sent. # With a reference (i.e. whenever cloning a library voice) S2-Pro prepends the # clip as an in-context example and imitates its prosody, which overwhelms the # tags: identical text scored a 10× loudness spread across emotions with no # reference, but only 1.2× with one — [screaming] vs [whisper] was inaudible. # Every lever was tried (canonical tags, free-form descriptions, repeated and # mid-sentence tags, temperature up to 1.0, 3s truncated references, # reference_id vs inline). So this backend is billed on what it actually does # well: rock-solid deterministic cloning, not per-line emotion. "fishspeech": { "purpose": "Clone a saved WAV voice with byte-identical, reproducible output.", "identity": "Excellent — clones the selected voice's reference WAV and is fully deterministic (same input always yields the same audio).", "style": "Weak while cloning: inline [tag] markers are overwhelmed by the reference clip's own delivery. Tone follows the reference, so the emotion of the reference recording is what you hear.", "best_for": "Consistent character voices and resumable bulk exports, where identical re-runs matter more than per-line emotion.", "uses_wav": True, "style_aware": False, "true_streaming": False, "speed": "~0.4× GPU", "latency": "1–4 s", "quality": "Premium", "ram": "~4 GB", }, } return caps.get(_clean_preview_backend(backend), {}) def _backend_health(url: str) -> bool: base = url.rstrip("/") for ep in ("/health", "/v1/health", "/v1/audio/list_voices"): try: r = requests.get(f"{base}{ep}", timeout=2) if r.status_code == 200: return True except Exception: continue return False def _backend_available(backend: str, voices: list, health: bool) -> bool: if _clean_preview_backend(backend) in {"nvidia_zeroshot", "nvidia_flow", "kokoro", "vibevoice", "fishspeech"}: return health return bool(voices) or health def _seed_finder_voices_json_path(settings: dict) -> Path: configured = Path(str(settings.get("tts_config_dir") or "/tts-config")) candidates = [ configured / "voices.json", Path("/tts-config/voices.json"), Path("/config/voices.json"), ] for path in candidates: if path.exists(): return path return candidates[0] def _seed_finder_backend_ref_audio(audio: Path, settings: dict) -> str: output_dir = _active_voices_dir(settings) try: rel = audio.resolve().relative_to(output_dir.resolve()) return "/voices/" + rel.as_posix() except Exception: pass return f"/voices/{audio.name}" def _seed_finder_voice_config(voice: str, voices: dict, settings: dict) -> dict: cfg = dict(voices.get(voice) or {}) if cfg.get("ref_audio"): return cfg scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) audio = _find_voice_audio(voice, scan_dir) if not audio: raise RuntimeError(f"reference WAV for voice {voice!r} not found") _has_ref, ref_text = _read_reference_text(audio) lang = { "DE": "German", "EN": "English", "GB": "English", "FR": "French", "ES": "Spanish", "IT": "Italian", "PT": "Portuguese", "NL": "Dutch", "PL": "Polish", "ZH": "Chinese", "JA": "Japanese", "KO": "Korean", }.get(voice.split("_", 1)[0].upper(), "Auto") return { "ref_audio": _seed_finder_backend_ref_audio(audio, settings), "language": lang, "chunk_size": 4, "ref_text": ref_text, } def _seed_finder_request_audio(text: str, voice: str, settings: dict, seed: int) -> tuple[bytes, str]: """Generate a Seed Finder sample from the reference WAV, not the voice's cached .pt. The Qwen3 VoiceClone server does not accept per-request ``seed`` on /v1/audio/speech; it reads the seed from voices.json. For Seed Finder we create a temporary voice entry that points at the selected reference WAV, omits the existing speaker embedding, sets the requested seed, generates once, then restores voices.json and removes the temporary .pt cache. """ voices_path = _seed_finder_voices_json_path(settings) if not voices_path.exists(): raise RuntimeError(f"TTS voices.json not found at {voices_path}") safe_voice = re.sub(r"[^A-Za-z0-9_\-.]+", "_", voice).strip("._-")[:60] or "voice" temp_voice = f"__seedfinder_{safe_voice}_{int(seed)}_{uuid.uuid4().hex[:8]}" temp_pt = Path("/tts-config/speakers") / f"{temp_voice}.pt" with _SEED_FINDER_VOICE_LOCK: original_text = voices_path.read_text(encoding="utf-8") try: voices = json.loads(original_text or "{}") if not isinstance(voices, dict): raise RuntimeError("voices.json is not an object") cfg = _seed_finder_voice_config(voice, voices, settings) cfg = dict(cfg) cfg.pop("speaker_embeddings", None) cfg.pop("speaker embeddings", None) cfg["seed"] = int(seed) voices[temp_voice] = cfg voices_path.write_text(json.dumps(voices, indent=2, ensure_ascii=False), encoding="utf-8") os.utime(voices_path, None) time.sleep(0.35) # let the backend hot-reload voices.json before resolve_voice() return _preview_request_audio(text, temp_voice, settings, "", "voice_clone") finally: try: voices_path.write_text(original_text, encoding="utf-8") os.utime(voices_path, None) except Exception as exc: logger.warning("Failed to restore voices.json after Seed Finder sample: %s", exc) try: if temp_pt.exists(): temp_pt.unlink() except Exception: pass _TTS_VOICE_ENDPOINTS = ("/v1/audio/voices", "/v1/audio/list_voices", "/v1/models", "/speakers", "/voices") _KOKORO_BUILTIN_VOICES = [ "af", "af_bella", "af_nicole", "af_sarah", "af_sky", "bf_emma", "bf_isabella", "am_adam", "am_michael", "bm_george", "bm_lewis", ] def _voice_ids_from_payload(payload) -> list: if isinstance(payload, list): return payload if not isinstance(payload, dict): return [] for key in ("data", "voices", "speakers"): value = payload.get(key) if isinstance(value, list): if key == "data": return [m.get("id", m) if isinstance(m, dict) else m for m in value] return value grouped = [] for value in payload.values(): if isinstance(value, list): grouped.extend(value) elif isinstance(value, dict): nested = _voice_ids_from_payload(value) if nested: grouped.extend(nested) return grouped def _active_library_voice_options(settings: dict) -> list[dict]: try: indexed = [] for voice in indexed_voices(settings): if voice.get("enabled", True) is False: continue indexed.append({ "id": voice.get("id", ""), "name": voice.get("name") or voice.get("id", ""), "duration": voice.get("duration"), "has_ref": bool(voice.get("has_ref")), "has_transcript": bool(voice.get("transcript")), "seed": voice.get("seed"), }) return [v for v in indexed if v["id"]] except Exception: pass active_dir = _active_voices_dir(settings) voices = [] seen = set() if not active_dir.exists(): return voices from core.voice import _read_reference_text for audio in sorted(_voice_audio_files(active_dir), key=lambda p: p.stem.lower()): if audio.stem in seen: continue seen.add(audio.stem) meta = _load_meta(audio) if meta.get("enabled", True) is False: continue has_ref, transcript = _read_reference_text(audio) try: dur = round(_duration(audio), 2) except Exception: dur = None voices.append({ "id": audio.stem, "name": audio.stem, "duration": dur, "has_ref": has_ref, "has_transcript": bool(transcript), "seed": meta.get("seed"), }) return voices def _fetch_backend_voices(settings: dict, backend: str) -> list: backend = _clean_preview_backend(backend) # Fish-Speech clones from the app's own saved WAV library (see # _fishspeech_request_audio, which resolves the voice via _find_voice_audio) and # exposes no voice-listing endpoint of its own. Without it in this set the # probe below found nothing and reported voice_count 0, so the Engines page # listed it with no selectable voices even while the server was healthy. if backend in {"voice_clone", "streaming", "nvidia_zeroshot", "nvidia_flow", "fishspeech"}: return _active_library_voice_options(settings) tts_url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/") key = (settings.get("voice_design_api_key") if backend == "voice_design" else settings.get("tts_api_key")) or "" tts_hdrs = {"Authorization": f"Bearer {key.strip()}"} if key.strip() else {} for ep in _TTS_VOICE_ENDPOINTS: try: r = requests.get(f"{tts_url}{ep}", headers=tts_hdrs, timeout=5) if r.status_code == 200: voices = _voice_ids_from_payload(r.json()) if voices: return voices except Exception: continue if backend == "kokoro": return _KOKORO_BUILTIN_VOICES return [] # ── TTS restart helpers ─────────────────────────────────────────────────────── def _clear_tts_restart_flags(settings: dict | None = None) -> int: cleared = 0 settings = settings or _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) if scan_dir.exists(): seen: set[str] = set() for audio in _voice_audio_files(scan_dir): if audio.stem in seen: continue seen.add(audio.stem) meta = _load_meta(audio) if meta.pop("needs_tts_restart", None) is not None: _save_meta(audio, meta) cleared += 1 if cleared: refresh_voice_index_background(settings, force=True) return cleared def _tts_container_names() -> list[str]: multi = os.environ.get("TTS_CONTAINER_NAMES", _TTS_CONTAINERS_RAW).strip() if multi: return [c.strip() for c in multi.split(",") if c.strip()] single = os.environ.get("TTS_CONTAINER_NAME", _TTS_CONTAINER).strip() return [single] if single else [] # ── Stream session store ────────────────────────────────────────────────────── _TTS_STREAM_SESSION_TTL = 15 * 60 _tts_stream_sessions: dict[str, dict] = {} def _purge_tts_stream_sessions() -> None: now = time.time() expired = [sid for sid, item in _tts_stream_sessions.items() if now - item.get("created", 0) > _TTS_STREAM_SESSION_TTL] for sid in expired: _tts_stream_sessions.pop(sid, None) # ── Routes ──────────────────────────────────────────────────────────────────── @router.get("/api/tts-voices") async def tts_voices(backend: str = "voice_clone"): return _fetch_backend_voices(_load_settings(), backend) @router.get("/api/tts-backends") async def tts_backends(): settings = _load_settings() items = [] for backend in ("voice_clone", "voice_design", "customvoice", "streaming", "fishspeech", "kokoro", "vibevoice", "xtts", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow"): url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/") voices = _fetch_backend_voices(settings, backend) health = _backend_health(url) available = _backend_available(backend, voices, health) items.append({ "id": backend, "label": _backend_display_name(backend, url), "url": url, "port": _backend_port_label(url), "available": available, "voice_count": len(voices) if isinstance(voices, list) else 0, **_backend_capabilities(backend), }) return {"backends": items} @router.post("/api/tts/restart") async def restart_tts_container(): containers = _tts_container_names() if not containers: raise HTTPException(400, "No TTS container names configured (set TTS_CONTAINER_NAMES in docker-compose.yml)") results = [] errors = [] for container in containers: path = f"/containers/{quote(container, safe='')}/restart?t=10" try: # _docker_post is a raw blocking socket call (core/docker_client.py) # and this loop can span two containers x up to 10s each — run off # the event loop thread, or every other request on this server # (including a plain GET /api/characters) hangs for the whole # restart instead of just this one call. Confirmed live: the # Studio Voices tab's own character fetch silently failed and # rendered "No characters yet" while a restart triggered elsewhere # was still in flight. code, raw = await asyncio.to_thread(_docker_post, path) if code not in (204, 304): detail = raw.split("\r\n\r\n", 1)[-1].strip() or f"HTTP {code}" errors.append(f"{container}: {detail}") else: results.append(container) except PermissionError: raise HTTPException(502, "No permission to access /var/run/docker.sock — is the socket mounted in docker-compose.yml?") except Exception as e: errors.append(f"{container}: {e}") cleared = 0 try: cleared = _clear_tts_restart_flags() except Exception as e: logger.warning("Could not clear TTS restart flags: %s", e) if errors and not results: raise HTTPException(502, "; ".join(errors)) return { "ok": True, "restarted": results, "errors": errors, "cleared_restart_flags": cleared, } @router.get("/api/tts/restart-info") async def tts_restart_info(): containers = _tts_container_names() sock_ok = Path(os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock")).exists() return {"containers": containers, "socket_available": sock_ok} @router.post("/api/tts/restart-flags/clear") async def clear_tts_restart_flags_endpoint(): try: cleared = _clear_tts_restart_flags() except Exception as e: raise HTTPException(500, f"Could not clear TTS restart flags: {e}") return {"ok": True, "cleared_restart_flags": cleared} @router.get("/api/tts-stream-health") async def tts_stream_health(): from core.constants import _TTS_STREAM_DEFAULT settings = _load_settings() try: stream_url = settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT base_url = _validate_http_url(stream_url, allow_private=True).rstrip("/") resp = await asyncio.to_thread(requests.get, f"{base_url}/health", timeout=3) return {"ok": resp.ok, "status_code": resp.status_code, "url": base_url} except Exception as e: return {"ok": False, "error": str(e)} @router.post("/api/tts-stream-session") async def tts_stream_session(request: Request): data = await request.json() text = str(data.get("text", "")).strip() voice = str(data.get("voice", "")).strip() if not voice: raise HTTPException(400, "voice is required") if not text: raise HTTPException(400, "text is required") _purge_tts_stream_sessions() sid = uuid.uuid4().hex instruct = str(data.get("instruct") or data.get("style_instruction") or "").strip() _tts_stream_sessions[sid] = {"created": time.time(), "text": text, "voice": voice, "instruct": instruct} return {"ok": True, "url": f"/api/tts-stream-session/{sid}"} @router.get("/api/tts-stream-session/{sid}") async def tts_stream_playback(sid: str): _purge_tts_stream_sessions() item = _tts_stream_sessions.pop(sid, None) if item is None: raise HTTPException(404, "stream session expired or not found") settings = _load_settings() try: resp = await asyncio.to_thread(_open_tts_stream_response, item["text"], item["voice"], settings, item.get("instruct", "")) except Exception as e: raise HTTPException(502, f"TTS stream error: {e}") media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav" return StreamingResponse( _iter_tts_stream_response(resp), media_type=media_type, headers={ "Cache-Control": "no-store", "X-Accel-Buffering": "no", "Content-Disposition": f"inline; filename=\"{item['voice']}_stream.wav\"", }, ) @router.post("/api/tts-preview") async def tts_preview(request: Request): data = await request.json() text: str = data["text"] voice: str = data["voice"] response_format = _requested_response_format(data) instruct = str(data.get("instruct") or data.get("style_instruction") or "") backend = _clean_preview_backend(data.get("backend", "voice_clone")) settings = _load_settings() # Per-request generation overrides (seed / temperature / top_p). When provided, # force them onto the stability params for this backend so _apply_tts_extra_params # forwards them to the engine (backends that reject them fall back gracefully). _overrides = {} for _k in ("seed", "temperature", "top_p"): _v = data.get(_k) if _v is not None and _v != "": try: _overrides[_k] = int(_v) if _k == "seed" else float(_v) except (TypeError, ValueError): pass _speed_v = data.get("speed") if _speed_v is not None and _speed_v != "": try: _speed_f = float(_speed_v) if _speed_f != 1.0: _overrides["speed"] = max(0.1, min(4.0, _speed_f)) except (TypeError, ValueError): pass if _overrides: from core.config import _tts_extra_params settings = dict(settings) eff = dict(_tts_extra_params(settings, backend) or {}) eff.update(_overrides) settings["tts_stability_enabled"] = True by_backend = dict(settings.get("tts_extra_params_by_backend") or {}) by_backend[backend] = eff settings["tts_extra_params_by_backend"] = by_backend if data.get("apply_persona"): scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) wav = _find_voice_audio(voice, scan_dir) persona = _load_meta(wav).get("persona", "") if wav else "" if not persona: # Confirmed live: checking "Apply character persona" on a voice with # no persona text saved (the common case — persona is a manually- # typed field on the Voice Inspector page, never auto-filled) used # to silently do nothing, which looked indistinguishable from the # feature being broken. Fail loud instead so the user knows to set # a persona first, rather than "why doesn't this work." raise HTTPException(400, f"Voice '{voice}' has no character persona saved — set one on the Voice Inspector page first, or uncheck 'Apply character persona.'") llm_url = (settings.get("llm_url") or "http://localhost:11434/v1").rstrip("/") llm_model = settings.get("llm_model") or "" try: from routes.conversation import _rewrite_with_persona_sync text = await asyncio.to_thread(_rewrite_with_persona_sync, text, persona, llm_url, llm_model) except Exception as e: raise HTTPException(502, f"Persona rewrite failed: {e}") try: if backend == "voice_clone" and data.get("seed_finder") and "seed" in _overrides: audio, media_type = await asyncio.to_thread( _seed_finder_request_audio, text, voice, settings, int(_overrides["seed"]) ) else: audio, media_type = await asyncio.to_thread(_preview_request_audio, text, voice, settings, instruct, backend) except Exception as e: raise HTTPException(502, f"TTS error: {e}") audio, media_type, ext, duration_sec, clipped = _prepare_proxy_audio(audio, media_type, response_format) return Response( content=audio, media_type=media_type, headers={ "Content-Disposition": f'inline; filename="{voice}_preview.{ext}"', "X-TTS-Audio-Duration": f"{duration_sec:.3f}" if duration_sec is not None else "", "X-TTS-Audio-Clipped": "true" if clipped else "false", }, ) @router.post("/api/tts-style-variation") async def tts_style_variation(request: Request): data = await request.json() source_voice = str(data.get("source_voice") or data.get("voice") or "").strip() new_voice_id = str(data.get("voice_id") or data.get("new_voice_id") or "").strip() text = str(data.get("text") or data.get("transcript") or "").strip() instruct = str(data.get("instruct") or data.get("style_instruction") or "").strip() backend = _clean_preview_backend(data.get("backend", "customvoice")) if not source_voice: raise HTTPException(400, "source_voice is required") if not new_voice_id: raise HTTPException(400, "new voice id is required") if not re.match(r"^[A-Za-z0-9_\-\.]+$", new_voice_id): raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .") if not text: raise HTTPException(400, "text/transcript is required") if not instruct: raise HTTPException(400, "style instruction is required") settings = _load_settings() try: audio, media_type = await asyncio.to_thread(_preview_request_audio, text, source_voice, settings, instruct, backend) audio, media_type, _ext, duration, _clipped = _prepare_proxy_audio(audio, media_type, "wav") except Exception as e: raise HTTPException(502, f"Style variation synthesis failed: {e}") tmp = TEMP_DIR / f"{uuid.uuid4().hex}_style_variation.wav" tmp.write_bytes(audio) out_dir = _active_voices_dir(settings) out_dir.mkdir(parents=True, exist_ok=True) wav_dest = out_dir / f"{new_voice_id}.wav" txt_dest = out_dir / f"{new_voice_id}.reference.txt" from core.voice import _remove_audio_variants _remove_audio_variants(out_dir, new_voice_id) loudness = _export_normalized_wav(tmp, wav_dest) txt_dest.write_text(text, encoding="utf-8") meta = _load_meta(wav_dest) meta.update({ "enabled": True, "loudness": loudness, "source_voice": source_voice, "style_instruction": instruct, "style_backend": backend, "note": f"Style variation of {source_voice}: {instruct[:180]}", "needs_tts_restart": True, }) _save_meta(wav_dest, meta) return { "ok": True, "voice_id": new_voice_id, "source_voice": source_voice, "backend": backend, "wav": str(wav_dest), "txt": str(txt_dest), "duration": duration, "loudness": loudness, "needs_tts_restart": True, } @router.post("/api/voice-design") async def voice_design(request: Request): data = await request.json() instruct: str = data.get("instruct", "").strip() sample_text: str = data.get("sample_text", "Hello! This is a voice design sample.").strip() language: str = data.get("language", "Auto") gender: str = data.get("gender", "") dialogue = bool(data.get("dialogue")) if not instruct: raise HTTPException(400, "instruct (voice description) is required") settings = _load_settings() try: if dialogue: parsed = _parse_voice_design_dialogue(instruct, sample_text) if not parsed: raise RuntimeError("dialogue mode needs speaker profiles and Speaker: text turns") audio, _media_type = await asyncio.to_thread( _voice_design_request_audio, instruct, sample_text, language, settings, "", ) else: audio, _media_type = await asyncio.to_thread( _voice_design_request_audio, instruct, sample_text, language, settings, gender, ) except Exception as e: raise HTTPException(502, f"Voice design error: {e}") audio, _media_type, _ext, _duration_sec, _clipped = _prepare_proxy_audio(audio, "audio/wav", "wav") tmp = TEMP_DIR / f"{uuid.uuid4().hex}_designed.wav" tmp.write_bytes(audio) fid = uuid.uuid4().hex _registry_put(fid, tmp) return {"id": fid, "duration": _duration(tmp)} @router.post("/api/tts-route-test") async def tts_route_test(request: Request): data = await request.json() text = str(data.get("input") or data.get("text") or "").strip() voice = str(data.get("voice") or "default").strip() app_name = _canonical_app_name(str(data.get("app") or data.get("client") or "Open WebUI").strip()) explicit_lang = str(data.get("language") or data.get("lang") or "").strip() if not text: raise HTTPException(400, "input is required") routed_voice, route = _resolve_tts_route(app_name, voice, text, explicit_lang) backend = _route_backend(route, routed_voice) settings = _load_settings() scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) routed_audio = None if backend in {"voice_design", "nvidia_magpie"} else _find_voice_audio(routed_voice, scan_dir) sound_status = {} for key in ("before_sound", "after_sound"): raw = str((route or {}).get(key, "")) if raw: try: path = _route_sound_path(settings, raw) sound_status[key] = {"ok": True, "path": str(path)} except Exception as e: sound_status[key] = {"ok": False, "error": str(e), "path": raw} result = { "app": app_name, "requested_voice": voice, "routed_voice": routed_voice, "backend": backend, "detected_language": str(route.get("detected_language", _detect_text_language(text))) if route else _detect_text_language(text), "matched": bool(route), "route": route, "voice_health": _voice_health(routed_audio) if routed_audio else None, "sounds": sound_status, } _routing_log_add( kind="test", status="matched" if route else "no_match", app=app_name, requested_voice=voice, routed_voice=routed_voice, backend=backend, language=result["detected_language"], matched=bool(route), route_id=str((route or {}).get("id", "")), text_preview=text[:160], sounds=sound_status, ) return result # ── OpenAI-compatible models/voices proxy ───────────────────────────────────── @router.get("/v1/models") async def openai_models_proxy(): now = 1686935002 data = [] seen = set() for name in sorted(_load_design_presets()): seen.add(_virtual_voice_id(name)) data.append({ "id": _virtual_voice_id(name), "object": "model", "created": now, "owned_by": "voice-design", }) for rule in _load_tts_routes(): alias = str(rule.get("input_voice", "")).strip() if rule.get("enabled", True) and alias and alias != "*" and alias not in seen: seen.add(alias) data.append({ "id": alias, "object": "model", "created": now, "owned_by": f"route:{rule.get('app', '*')}", }) try: voices = await tts_voices() for item in voices: if isinstance(item, str): voice_id = item elif isinstance(item, dict): voice_id = item.get("id") or item.get("voice") else: voice_id = str(item) if item is not None else "" if voice_id and voice_id not in seen: seen.add(str(voice_id)) data.append({ "id": str(voice_id), "object": "model", "created": now, "owned_by": "qwen", }) except Exception: pass return {"object": "list", "data": data} @router.get("/v1/audio/models") async def openai_audio_models_proxy(): return await openai_models_proxy() @router.api_route("/v1/audio/voices", methods=["GET", "POST"]) async def openai_audio_voices_proxy(): models = await openai_models_proxy() return [m["id"] for m in models["data"]] # ── OpenAI-compatible speech proxy ─────────────────────────────────────────── @router.post("/v1/audio/speech") async def openai_speech_proxy(request: Request): data = await request.json() text = str(data.get("input") or data.get("text") or "").strip() voice = str(data.get("voice") or data.get("model") or "").strip() response_format = _requested_response_format(data) request_app = _canonical_app_name(str(data.get("app") or data.get("client") or "").strip()) if (data.get("app") or data.get("client")) else _request_app_name(request) original_voice = voice if not text: _routing_log_request( request, status="error", app=request_app, requested_voice=original_voice, routed_voice=voice, backend="", route=None, response_format=response_format, text=text, error="input is required", ) raise HTTPException(400, "input is required") settings = _load_settings() if not voice: fallback = next((v["id"] for v in _active_library_voice_options(settings)), "") if not fallback: _routing_log_request( request, status="error", app=request_app, requested_voice=original_voice, routed_voice=voice, backend="", route=None, response_format=response_format, text=text, error="voice is required", ) raise HTTPException(400, "voice is required") voice = fallback explicit_lang = str(data.get("language") or data.get("lang") or "").strip() voice, route = _resolve_tts_route(request_app, voice, text, explicit_lang) backend = _route_backend(route, voice) style_instruction = str(data.get("instruct") or data.get("style_instruction") or "") virtual = _resolve_virtual_voice(voice) def _route_speed(r): try: return float((r or {}).get("speed", 1.0) or 1.0) except (TypeError, ValueError): return 1.0 route_has_sounds = bool((route or {}).get("before_sound") or (route or {}).get("after_sound")) route_has_speed = abs(_route_speed(route) - 1.0) > 0.01 if backend == "streaming" and not virtual and response_format == "wav" and not route_has_sounds and not route_has_speed: try: resp = await asyncio.to_thread(_open_tts_stream_response, text, voice, settings, style_instruction) except Exception as e: _routing_log_request( request, status="error", app=request_app, requested_voice=original_voice, routed_voice=voice, backend=backend, route=route, response_format=response_format, text=text, error=f"TTS stream proxy error: {e}", ) raise HTTPException(502, f"TTS stream proxy error: {e}") media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav" _routing_log_request( request, status="streaming", app=request_app, requested_voice=original_voice, routed_voice=voice, backend=backend, route=route, response_format=response_format, text=text, media_type=media_type, sounds=[], ) return StreamingResponse( _iter_tts_stream_response(resp), media_type=media_type, headers={ "Cache-Control": "no-store", "X-Accel-Buffering": "no", "Content-Disposition": f'inline; filename="{voice}_speech.wav"', "X-TTS-Voice-Requested": original_voice, "X-TTS-Voice-Routed": voice, "X-TTS-Route-Backend": backend, "X-TTS-Route-Language": str(route.get("detected_language", "")) if route else "", "X-TTS-Response-Format": response_format, "X-TTS-Audio-Clipped": "false", "X-TTS-Route-Sounds": "", }, ) try: if virtual: _name, preset = virtual instruct = str(preset.get("description", "")).strip() if not instruct: raise RuntimeError(f"Virtual voice '{voice}' has no description") audio, media_type = await asyncio.to_thread( _voice_design_request_audio, instruct, text, str(preset.get("language", "Auto")), settings, str(preset.get("gender", "")), ) elif backend == "voice_design": audio, media_type = await asyncio.to_thread( _voice_design_voice_request_audio, voice, text, settings, style_instruction, str(data.get("language") or "Auto"), ) elif backend == "streaming": resp = await asyncio.to_thread(_open_tts_stream_response, text, voice, settings, style_instruction) audio, media_type = await asyncio.to_thread(_read_tts_stream_response, resp) if not audio: raise RuntimeError("backend returned empty audio") elif backend == "nvidia_magpie": audio, media_type = await asyncio.to_thread( _tts_request_audio, text, voice, settings, style_instruction, _preview_backend_base_url(settings, "nvidia_magpie"), settings.get("tts_api_key", ""), "nvidia_magpie", "nvidia_magpie", ) elif backend in {"nvidia_zeroshot", "nvidia_flow"}: audio, media_type = await asyncio.to_thread( _nvidia_clone_request_audio, text, voice, settings, "flow" if backend == "nvidia_flow" else "zeroshot", str(data.get("audio_prompt_transcript") or ""), str(data.get("language") or ""), ) elif backend == "fishspeech": audio, media_type = await asyncio.to_thread( _fishspeech_request_audio, text, voice, settings, style_instruction, str(data.get("language") or ""), ) else: audio, media_type = await asyncio.to_thread( _tts_request_audio, text, voice, settings, style_instruction, ) except Exception as e: _routing_log_request( request, status="error", app=request_app, requested_voice=original_voice, routed_voice=voice, backend=backend, route=route, response_format=response_format, text=text, error=f"TTS proxy error: {e}", ) raise HTTPException(502, f"TTS proxy error: {e}") try: audio, media_type, applied_sounds = _apply_route_sounds(audio, media_type, route, settings) except Exception as e: _routing_log_request( request, status="error", app=request_app, requested_voice=original_voice, routed_voice=voice, backend=backend, route=route, response_format=response_format, text=text, error=f"TTS route sound error: {e}", ) raise HTTPException(502, f"TTS route sound error: {e}") audio, media_type, ext, duration, clipped = _prepare_proxy_audio(audio, media_type, response_format) logger.info( "TTS proxy app=%s voice=%s routed=%s backend=%s lang=%s format=%s bytes=%s duration=%s clipped=%s sounds=%s", request_app, original_voice, voice, backend, str(route.get("detected_language", "")) if route else "", response_format, len(audio), f"{duration:.2f}" if duration is not None else "?", clipped, ",".join(applied_sounds) if applied_sounds else "-", ) _routing_log_request( request, status="ok" if route else "no_match", app=request_app, requested_voice=original_voice, routed_voice=voice, backend=backend, route=route, response_format=response_format, text=text, bytes=len(audio), duration=duration, clipped=clipped, media_type=media_type, sounds=applied_sounds, ) return Response( content=audio, media_type=media_type, headers={ "Content-Disposition": f'inline; filename="{voice}_speech.{ext}"', "X-TTS-Voice-Requested": original_voice, "X-TTS-Voice-Routed": voice, "X-TTS-Route-Backend": backend, "X-TTS-Route-Language": str(route.get("detected_language", "")) if route else "", "X-TTS-Response-Format": response_format, "X-TTS-Audio-Duration": f"{duration:.3f}" if duration is not None else "", "X-TTS-Audio-Clipped": "true" if clipped else "false", "X-TTS-Route-Sounds": ",".join(applied_sounds), }, ) @router.post("/v1") async def openai_speech_proxy_v1_shortcut(request: Request): return await openai_speech_proxy(request) @router.post("/api/tts-voice-seed") async def tts_voice_seed(request: Request): """Proxy to the faster-qwen3-tts /voice-seed endpoint. Writes or clears the seed for a voice in the TTS server's voices.json. Body: {"voice": "EN_F_NatashaNeural", "seed": 7} (seed: null to remove) """ data = await request.json() settings = _load_settings() tts_base = _preview_backend_base_url(settings, "voice_clone").rstrip("/") try: resp = requests.post( f"{tts_base}/voice-seed", json=data, timeout=10, ) resp.raise_for_status() voice_name = data.get("voice") seed = data.get("seed") if voice_name: from core.voice import _find_voice_audio, _load_meta, _save_meta scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)) audio = _find_voice_audio(voice_name, scan_dir) if audio: meta = _load_meta(audio) if seed is None: meta.pop("seed", None) else: meta["seed"] = seed _save_meta(audio, meta) return resp.json() except requests.exceptions.ConnectionError: raise HTTPException(502, "Could not reach TTS server") except requests.exceptions.HTTPError as e: detail = "" try: detail = e.response.json().get("detail", "") except Exception: pass raise HTTPException(e.response.status_code, detail or str(e)) @router.get("/api/seed-samples/{voice_name}") async def list_seed_samples(voice_name: str): """List pre-generated seed sample numbers for a voice (batch script output).""" settings = _load_settings() tts_base = _preview_backend_base_url(settings, "voice_clone").rstrip("/") try: resp = requests.get(f"{tts_base}/seed-samples/{voice_name}", timeout=5) resp.raise_for_status() return resp.json() except requests.exceptions.ConnectionError: raise HTTPException(502, "Could not reach TTS server") except requests.exceptions.HTTPError as e: raise HTTPException(e.response.status_code, str(e)) @router.get("/api/seed-sample/{voice_name}/{seed}") async def get_seed_sample(voice_name: str, seed: int): """Serve a pre-generated seed WAV file (batch script output).""" settings = _load_settings() tts_base = _preview_backend_base_url(settings, "voice_clone").rstrip("/") try: resp = requests.get(f"{tts_base}/seed-sample/{voice_name}/{seed}", timeout=30, stream=True) resp.raise_for_status() return Response(content=resp.content, media_type="audio/wav") except requests.exceptions.ConnectionError: raise HTTPException(502, "Could not reach TTS server") except requests.exceptions.HTTPError as e: raise HTTPException(e.response.status_code, str(e)) @router.post("/api/audio/apply-gain") async def apply_gain(request: Request, gain_db: float = 0.0): """Apply a FIXED gain to a single WAV clip — NOT the same as normalizing each clip independently to one target level. That was the first version of this fix, and it broke something real: normalizing every clip to the same -20dBFS average erases a voice's own internal loudness dynamics along with fixing the cross-voice mismatch — confirmed live, synthesizing the same line "neutral" vs with a whisper instruct: the whisper version came out LOUDER than neutral once each was independently pushed to the same target, exactly backwards. A whisper should measurably be quieter than a shout from the SAME voice; per-clip auto-normalization can't preserve that, only a fixed offset can. The caller instead looks up the voice's OWN already-computed reference gain (from Calc dB / voice-loudness metadata) ONCE and applies that same fixed number to every clip from that voice — shifting the whole voice's baseline level to match others (fixing the Narrator-vs-designed-voice gap) while leaving each line's own relative loudness — quiet vs shouted — exactly as the model produced it. """ from pydub import AudioSegment wav_bytes = await request.body() if not wav_bytes: raise HTTPException(400, "Empty request body") if not gain_db: return Response(content=wav_bytes, media_type="audio/wav") try: seg = AudioSegment.from_file(io.BytesIO(wav_bytes), format="wav") # Still peak-limited — a fixed gain derived from a short reference # clip could clip a louder line (e.g. an already-shouted one). peak = seg.max_dBFS if seg.max_dBFS != float("-inf") else None g = gain_db if peak is not None and peak + g > _VOICE_PEAK_DBFS: g = _VOICE_PEAK_DBFS - peak seg2 = seg.apply_gain(g) out = io.BytesIO() seg2.export(out, format="wav") return Response(content=out.getvalue(), media_type="audio/wav") except Exception as e: raise HTTPException(400, f"Could not apply gain: {e}") @router.post("/api/audio/encode-mp3") async def encode_mp3(request: Request): """Encode a raw WAV body into MP3 at an explicit bitrate. audiobookExport() used to concatenate independently-encoded per-line MP3 byte streams directly into one Blob — each clip carries its own frame/ID3 headers, so most players only decode the first one (confirmed live: an 85MB file that reported as 22s playable). The fix merges lossless WAV clips client-side (mergeWavBlobs, already correct) and sends the single merged WAV here for one real encode pass — also fixes the previous 32kbps default (ffmpeg/lame's unset-bitrate fallback, not a deliberate choice anywhere in this app) without pretending to add quality beyond the engine's native 24kHz mono output. """ wav_bytes = await request.body() if not wav_bytes: raise HTTPException(400, "Empty request body") try: from pydub import AudioSegment segment = AudioSegment.from_file(io.BytesIO(wav_bytes), format="wav") out = io.BytesIO() segment.export(out, format="mp3", bitrate="96k") return Response(content=out.getvalue(), media_type="audio/mpeg") except Exception as e: raise HTTPException(400, f"Could not encode audio: {e}") # ── Per-paragraph synthesized-audio cache ─────────────────────────────────── # # "Synth all" pre-synthesizes every line for instant playback, but only ever # kept the result in the browser tab's memory — closing the tab (or a crash, # or just a normal reload) threw all of it away, and every future playback # or export had to wait on the GPU again from scratch. This persists each # line's audio to disk, keyed by a hash of its own content (text + voice + # instruct/tone) rather than its position in the script — editing a # paragraph changes its hash, so the edited version simply never matches a # cached file and gets synthesized fresh, while an untouched paragraph # reuses its file instantly regardless of how the surrounding lines shifted. # The key is computed client-side (SHA-256 over the exact inputs that affect # the audio) and treated here as an opaque cache token — this endpoint never # needs to know what it means, only that the same key always means the same # audio. _LINE_AUDIO_DIR = CONFIG_DIR / "line_audio_cache" _LINE_AUDIO_KEY_RE = re.compile(r"^[a-f0-9]{16,64}$") def _line_audio_book_dir(book: str) -> Path: safe_book = re.sub(r"[^A-Za-z0-9_-]+", "_", book).strip("_")[:80] or "book" d = _LINE_AUDIO_DIR / safe_book d.mkdir(parents=True, exist_ok=True) return d @router.get("/api/line-audio/{book}/{key}") async def get_line_audio(book: str, key: str): if not _LINE_AUDIO_KEY_RE.match(key): raise HTTPException(400, "Invalid cache key") path = _line_audio_book_dir(book) / f"{key}.wav" if not path.exists(): raise HTTPException(404, "Not cached") return Response(content=path.read_bytes(), media_type="audio/wav") @router.post("/api/line-audio/{book}/check") async def check_line_audio(book: str, request: Request): """Bulk existence check — one request instead of one GET per line — so the Stage page can mark its "pre-synthesized" dots correctly right after a reload, instead of every dot looking unsynthesized just because rehState.synthCache (this browser tab's own memory) starts empty on every fresh page load even when the audio is sitting on disk already.""" data = await request.json() keys = data.get("keys") or [] if not isinstance(keys, list): raise HTTPException(400, "keys must be a list of cache keys") book_dir = _line_audio_book_dir(book) existing = [k for k in keys if isinstance(k, str) and _LINE_AUDIO_KEY_RE.match(k) and (book_dir / f"{k}.wav").exists()] return {"ok": True, "existing": existing} @router.post("/api/line-audio/{book}/prune") async def prune_line_audio(book: str, request: Request): """Delete cached files for this book that no longer match any current line — a paragraph's cache key is its own content hash, so editing it just makes the old file unreachable rather than actively removing it (nothing on the write path knows a "previous" key exists to delete). The client sends every key still valid for the CURRENT script; anything else on disk for this book is safe to remove. Registered BEFORE the generic POST /api/line-audio/{book}/{key} route below — FastAPI matches routes in declaration order, and {key} is just a plain path segment at the routing level (its regex validation only runs inside the handler, after routing already picked one), so a literal "prune" segment would otherwise always match that generic route first and this one would never be reached at all. """ data = await request.json() keep = data.get("keep") or [] if not isinstance(keep, list): raise HTTPException(400, "keep must be a list of cache keys") keep_set = {k for k in keep if isinstance(k, str) and _LINE_AUDIO_KEY_RE.match(k)} book_dir = _line_audio_book_dir(book) deleted = 0 for f in book_dir.glob("*.wav"): if f.stem not in keep_set: try: f.unlink() deleted += 1 except OSError: pass return {"ok": True, "deleted": deleted, "kept": len(keep_set)} @router.post("/api/line-audio/{book}/normalize") async def normalize_line_audio(book: str): """Loudness-normalize every cached clip for this book in place, to the same target used for voice reference files (_normalize_segment). Different TTS backends (Voice Clone vs Voice Design) apparently ship very different default output levels — confirmed live as the Narrator (cloned) sounding noticeably louder than designed-voice characters in a finished export, since nothing in the synthesis or merge pipeline ever leveled clips against each other. Running this against the EXISTING cache is far cheaper than resynthesizing the whole book: it's pure audio processing, no TTS calls, so a ~1600-line book normalizes in well under a minute instead of the hours a full resynth would take. A subsequent export then hits 100% cache and just needs to merge + encode. Registered before the generic {key} route below for the same routing reason as /check and /prune. """ from core.audio import _normalize_segment from pydub import AudioSegment book_dir = _line_audio_book_dir(book) normalized = 0 skipped = 0 errors = 0 for f in book_dir.glob("*.wav"): try: seg = AudioSegment.from_file(str(f), format="wav") seg2, info = _normalize_segment(seg) if abs(info.get("gain_db") or 0.0) < 0.1: skipped += 1 continue seg2.export(str(f), format="wav") normalized += 1 except Exception: errors += 1 return {"ok": True, "normalized": normalized, "skipped": skipped, "errors": errors} @router.post("/api/line-audio/{book}/{key}") async def put_line_audio(book: str, key: str, request: Request): if not _LINE_AUDIO_KEY_RE.match(key): raise HTTPException(400, "Invalid cache key") wav_bytes = await request.body() if not wav_bytes: raise HTTPException(400, "Empty request body") path = _line_audio_book_dir(book) / f"{key}.wav" path.write_bytes(wav_bytes) return {"ok": True} # ── Finished audiobook chapter exports ────────────────────────────────────── # # audiobookExport() already triggers a browser download per chapter, but # that only ever lands wherever the browser's download settings put it — # confirmed as a real gap: nothing in the app itself says where the files # went, and re-finding a chapter later means re-running the whole export. # This additionally saves the exact same file server-side so the app can # show a real download link (and the on-disk path) right after the export # finishes, and again any time later without resynthesizing anything. _AUDIOBOOK_EXPORT_DIR = CONFIG_DIR / "audiobook_exports" _EXPORT_FILENAME_RE = re.compile(r"^[^/\\]{1,200}$") # any single path segment, no traversal def _audiobook_export_book_dir(book: str) -> Path: safe_book = re.sub(r"[^A-Za-z0-9_-]+", "_", book).strip("_")[:80] or "book" d = _AUDIOBOOK_EXPORT_DIR / safe_book d.mkdir(parents=True, exist_ok=True) return d @router.get("/api/audiobook-export/{book}") async def list_audiobook_exports(book: str): """List previously-saved chapter exports for this book — lets the app show a "browse what's already been exported" view without re-running the export, and without any real filesystem access on the user's part. Registered before the generic GET .../{filename} route below for the same reason "zip" and "check"/"prune" are elsewhere in this file: a plain path segment matches ANY literal string at the routing level. """ book_dir = _audiobook_export_book_dir(book) files = sorted( ({"name": f.name, "size": f.stat().st_size} for f in book_dir.iterdir() if f.is_file()), key=lambda x: x["name"], ) return {"book": book, "files": files, "dir": str(book_dir)} @router.get("/api/audiobook-export/{book}/zip") async def zip_audiobook_exports(book: str): """Bundle every saved chapter for this book into one ZIP download — the "download everything at once" the per-file list doesn't offer.""" import zipfile book_dir = _audiobook_export_book_dir(book) files = [f for f in book_dir.iterdir() if f.is_file()] if not files: raise HTTPException(404, "No exported files for this book") buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: for f in files: zf.write(f, arcname=f.name) buf.seek(0) zip_name = re.sub(r"[^A-Za-z0-9_-]+", "_", book).strip("_")[:80] or "audiobook" return Response( content=buf.getvalue(), media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="{zip_name}.zip"'}, ) @router.get("/api/audiobook-export/{book}/{filename}") async def get_audiobook_export(book: str, filename: str): if not _EXPORT_FILENAME_RE.match(filename) or filename in (".", ".."): raise HTTPException(400, "Invalid filename") path = _audiobook_export_book_dir(book) / filename if not path.exists() or not path.is_file(): raise HTTPException(404, "Not found") media_type = "audio/mpeg" if path.suffix.lower() == ".mp3" else "audio/wav" return Response( content=path.read_bytes(), media_type=media_type, headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) @router.post("/api/audiobook-export/{book}/{filename}") async def put_audiobook_export(book: str, filename: str, request: Request): if not _EXPORT_FILENAME_RE.match(filename) or filename in (".", ".."): raise HTTPException(400, "Invalid filename") audio_bytes = await request.body() if not audio_bytes: raise HTTPException(400, "Empty request body") path = _audiobook_export_book_dir(book) / filename path.write_bytes(audio_bytes) return {"ok": True, "path": str(path)}