#!/usr/bin/env python3 """Private Qwen3-TTS-first gateway with a Piper fallback. The gateway implements the narrow /status and /tts protocol already consumed by the profile router. Request text is never logged or persisted. """ from __future__ import annotations import io import json import os import re import subprocess import threading import time import unicodedata import urllib.error import urllib.request import wave from array import array from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HOST = os.getenv("TTS_GATEWAY_HOST", "0.0.0.0") PORT = int(os.getenv("TTS_GATEWAY_PORT", "8085")) QWEN_TTS_URL = os.getenv("QWEN_TTS_URL", "http://qwen3-tts:8001").rstrip("/") QWEN_TTS_MODEL = os.getenv("QWEN_TTS_MODEL", "tts-1") QWEN_TTS_VOICE = os.getenv("QWEN_TTS_VOICE", "serena") QWEN_TTS_LANGUAGE = os.getenv("QWEN_TTS_LANGUAGE", "German") QWEN_TTS_TIMEOUT = float(os.getenv("QWEN_TTS_TIMEOUT", "120")) XTTS_URL = os.getenv("XTTS_URL", "http://xtts:80").rstrip("/") PIPER_URL = os.getenv("PIPER_URL", "http://piper:8085").rstrip("/") VOICE_ALIAS = os.getenv("TTS_VOICE_ALIAS", "alloy") XTTS_SPEAKER = os.getenv("XTTS_SPEAKER", "Annmarie Nele") DEFAULT_LANGUAGE = os.getenv("TTS_DEFAULT_LANGUAGE", "de") CODE_SWITCH_ENABLED = os.getenv("TTS_CODE_SWITCH_ENABLED", "false").lower() \ in {"1", "true", "yes", "on"} MAX_TEXT_CHARS = int(os.getenv("TTS_MAX_TEXT_CHARS", "8000")) MAX_REQUEST_BYTES = int(os.getenv("TTS_MAX_REQUEST_BYTES", "65536")) MAX_AUDIO_BYTES = int(os.getenv("TTS_MAX_AUDIO_BYTES", str(64 * 1024 * 1024))) XTTS_TIMEOUT = float(os.getenv("XTTS_TIMEOUT", "120")) PIPER_TIMEOUT = float(os.getenv("PIPER_TIMEOUT", "120")) QUEUE_TIMEOUT = float(os.getenv("XTTS_QUEUE_TIMEOUT", "15")) # XTTS loses natural prosody when a sentence is synthesized as many tiny # requests: every request starts a fresh utterance. Keep complete sentences # together and use this only as a safety ceiling for unusually long sentences. CHUNK_CHARS = int(os.getenv("XTTS_CHUNK_CHARS", "420")) TRIM_THRESHOLD = int(os.getenv("XTTS_TRIM_THRESHOLD", "90")) TRIM_PADDING_MS = int(os.getenv("XTTS_TRIM_PADDING_MS", "18")) CROSSFADE_MS = int(os.getenv("XTTS_CROSSFADE_MS", "8")) SENTENCE_PAUSE_MS = int(os.getenv("XTTS_SENTENCE_PAUSE_MS", "65")) INTERNAL_SILENCE_THRESHOLD = int(os.getenv("XTTS_INTERNAL_SILENCE_THRESHOLD", "512")) INTERNAL_SILENCE_TRIGGER_MS = int(os.getenv("XTTS_INTERNAL_SILENCE_TRIGGER_MS", "650")) INTERNAL_SILENCE_KEEP_MS = int(os.getenv("XTTS_INTERNAL_SILENCE_KEEP_MS", "220")) SYNTHESIS_LOCK = threading.Lock() STATE_LOCK = threading.Lock() SPEAKER_LOCK = threading.Lock() SPEAKER_CONDITIONING: dict | None = None STATE = { "last_backend": None, "xtts_failures": 0, "piper_fallbacks": 0, "last_error": None, } # Prefer full compounds to isolated terms. This keeps switches infrequent and # avoids making mixed-language speech sound like a sequence of separate clips. ENGLISH_TERMS = ( "unsupported image format", "premature end of JPEG", "incomplete scan", "Docker Containers", "Docker-Containers", "Docker Containern", "Docker-Containern", "Health Checks", "Health-Checks", "Restart Loops", "Restart-Loops", "False Positive", "Delivery Errors", "DeliveryErrors", "Ack Problem", "Ack-Problem", "I/O timeout", "Parity Check", "Parity-Check", "Disk disabled", "Disk invalid", "Home Assistant", "Open WebUI", "OpenWebUI", "Unraid Dashboard", "Unraid-Dashboard", "Docker Container", "Docker-Container", "Server Log", "Server-Log", "GitHub Repository", "GitHub Repo", "WireGuard Tunnel", "Cron Job", "Cronjob", "Home Server", "API Key", "Tool Calling", "Context Window", "Prompt Injection", "Unraid", "Docker", "Containern", "Containers", "Container", "Crashes", "healthy", "disabled", "invalid", "Dashboard", "Server", "Logs", "Log", "Matches", "up", "OpenAI", "GitHub", "WireGuard", "Linux", "Debian", "Frontend", "Backend", "Router", "Browser", "Web", "Token", "Prompt", "Context", "Model", "Image", "Tool", "Workflow", "Benchmark", "Streaming", "SSH", "MCP", "API", "CPU", "GPU", "VRAM", "RAM", "HTTP", "HTTPS", "HomeAssistant", "ESPHome", "iGotify", "Immich", "Vaultwarden", "UniFi", "UptimeKuma", "go2rtc", "Zigbee", "SONOFF", "eWeLink", "RTSP", "JPEG", "NVMe", "GiB", ) TERM_PATTERN = re.compile( r"(? tuple[bytes, str]: data = None headers = {} method = "GET" if payload is not None: data = json.dumps(payload, separators=(",", ":")).encode() headers["Content-Type"] = "application/json" method = "POST" request = urllib.request.Request( url, data=data, headers=headers, method=method) with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read(MAX_AUDIO_BYTES + 1) if len(body) > MAX_AUDIO_BYTES: raise RuntimeError("upstream audio response is too large") return body, response.headers.get_content_type() def _json(url: str, timeout: float = 10) -> dict | list: body, _ = _request(url, timeout=timeout) return json.loads(body) def _reachable(url: str, path: str, timeout: float = 2) -> bool: try: _request(f"{url}{path}", timeout=timeout) return True except (OSError, ValueError, RuntimeError, urllib.error.URLError): return False def _looks_english(text: str) -> bool: words = re.findall(r"[A-Za-zÀ-ÿ]+", text.lower()) if not words: return False german = sum(word in GERMAN_MARKERS for word in words) english = sum(word in ENGLISH_MARKERS for word in words) return english >= 2 and english > german * 1.5 and not re.search(r"[äöüß]", text.lower()) def segment_languages(text: str) -> list[tuple[str, str]]: """Return a compact German/English segment sequence.""" if _looks_english(text): return [("en", text)] if DEFAULT_LANGUAGE != "de": return [(DEFAULT_LANGUAGE, text)] if not CODE_SWITCH_ENABLED: return [("de", text)] segments: list[tuple[str, str]] = [] cursor = 0 for match in TERM_PATTERN.finditer(text): if match.start() > cursor: segments.append(("de", text[cursor:match.start()])) segments.append(("en", match.group(0))) cursor = match.end() if cursor < len(text): segments.append(("de", text[cursor:])) if not segments: return [("de", text)] merged: list[tuple[str, str]] = [] for language, part in segments: if not part: continue if merged and merged[-1][0] == language: previous_language, previous_text = merged[-1] merged[-1] = (previous_language, previous_text + part) else: merged.append((language, part)) return merged def _spoken_date(match: re.Match) -> str: day = int(match.group(1)) month = int(match.group(2)) year = match.group(3) month_name = GERMAN_MONTHS.get(month) if not month_name or not 1 <= day <= 31: return match.group(0) result = f"{day}. {month_name}" if year: result += f" {year}" return result def _spell_digits(value: str) -> str: return " ".join(GERMAN_DIGITS[digit] for digit in value) def _spoken_ipv4(match: re.Match) -> str: """Keep IPv4 octets intact while making the separators pronounceable.""" return " Punkt ".join(str(int(part)) for part in match.group(0).split(".")) def normalize_for_german_speech(text: str) -> str: """Turn common visual notation into unambiguous spoken German.""" # Run these before the date rule: otherwise 192.168.1.5 could be partly # interpreted as a visual date. text = re.sub( r"\b(?:\d{1,3}\.){3}\d{1,3}\b", _spoken_ipv4, text, ) text = re.sub( r"\b([01]?\d|2[0-3]):([0-5]\d)\b", lambda match: f"{int(match.group(1))} Uhr {int(match.group(2))}", text, ) text = re.sub( r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?[ \t]*/[ \t]*" r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?", r"Höchstwert \1 Grad, Tiefstwert \2 Grad", text, flags=re.IGNORECASE, ) text = re.sub( r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?", r"\1 Grad", text, flags=re.IGNORECASE, ) text = re.sub( r"\b([0-3]?\d)\.([01]?\d)\.(?:(\d{4})\b)?", _spoken_date, text, ) text = re.sub( r"\bPLZ\s+(\d{5})\b", lambda match: "Postleitzahl " + _spell_digits(match.group(1)), text, flags=re.IGNORECASE, ) text = re.sub( r"\((\d{5})\)", lambda match: "(Postleitzahl " + _spell_digits(match.group(1)) + ")", text, ) text = re.sub( r"https?://(?:www\.)?([^/\s)]+)(?:/[^\s)]*)?", lambda match: match.group(1), text, flags=re.IGNORECASE, ) text = re.sub( r"\b([A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9-]+)*)" r"\.(de|com|org|net|io|ai)\b", lambda match: ( match.group(1).replace(".", " Punkt ") + " Punkt " + ( "de" if match.group(2).lower() == "de" else " ".join(match.group(2).upper()) ) ), text, flags=re.IGNORECASE, ) text = re.sub( r"(-?\d+(?:[,.]\d+)?)\s*km\s*(?:/\s*)?h\b", r"\1 Kilometer pro Stunde", text, flags=re.IGNORECASE, ) text = re.sub(r"(\d)\s*%", r"\1 Prozent", text) text = re.sub(r"\b(\d+)\s*[×x]\s*", r"\1 mal ", text) return text def prepare_for_qwen_speech(text: str) -> str: """Normalize technical display text without destroying Qwen's prosody.""" text = re.sub(r"```.*?```", " Codeblock. ", text, flags=re.DOTALL) text = re.sub(r"`([^`]+)`", r"\1", text) text = re.sub(r"!\[([^]]*)\]\([^)]+\)", r"\1", text) text = re.sub(r"\[([^]]+)\]\([^)]+\)", r"\1", text) text = re.sub(r"(?m)^\s{0,3}#{1,6}\s*", "", text) text = re.sub(r"(?m)^\s*[-*+]\s+", "", text) text = text.replace("_", " ").replace("/", ", ") text = text.replace("→", ". ").replace("←", ". ") if DEFAULT_LANGUAGE == "de": text = normalize_for_german_speech(text) text = "".join( char for char in text if unicodedata.category(char) not in {"So", "Cs"} ) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\s*\n+\s*", ". ", text) return text.strip() def clean_for_speech(text: str) -> str: """Remove visual markup that makes long TTS output unstable or noisy.""" text = re.sub(r"```.*?```", " Code block. ", text, flags=re.DOTALL) text = re.sub(r"`([^`]+)`", r"\1", text) text = re.sub(r"!\[([^]]*)\]\([^)]+\)", r"\1", text) text = re.sub(r"\[([^]]+)\]\([^)]+\)", r"\1", text) text = re.sub(r"(?m)^\s{0,3}#{1,6}\s*", "", text) text = re.sub(r"(?m)^\s*[-*+]\s+", "", text) text = text.replace("→", ". ").replace("←", ". ") if DEFAULT_LANGUAGE == "de": text = normalize_for_german_speech(text) # XTTS occasionally hallucinates syllables when visual punctuation is # submitted literally or isolated at a chunk boundary. Preserve its pause # semantics, but never ask the model to pronounce the glyph itself. text = re.sub(r"(?<=\w)[\-‐‑‒–—−](?=\w)", " ", text) text = re.sub(r"\s+[\-‐‑‒–—−]\s+", ", ", text) text = re.sub(r"\s*[:;]+\s*", ", ", text) # XTTS can pronounce literal question/exclamation glyphs as short # nonsense syllables (for example "?" as "nau"). Retain a sentence # boundary for pacing, but never pass those glyphs to the model. text = re.sub(r"\s*[!?]+\s*", ". ", text) text = "".join( char for char in text if unicodedata.category(char) not in {"So", "Cs"} ) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\s*\n+\s*", ". ", text) text = re.sub(r",\s*\.", ".", text) text = re.sub(r"(?:,\s*){2,}", ", ", text) text = re.sub(r"(?:\.\s*){2,}", ". ", text) return text.strip() def _split_chunk(text: str, limit: int = CHUNK_CHARS) -> list[str]: """Return paragraph-sized XTTS requests with a conservative hard ceiling. Complete neighbouring sentences are combined while they fit. This avoids restarting the generative XTTS decoder after every short sentence, which can create invented tail syllables between sentences. Overlong sentences are split at clause boundaries first and at words only as a last resort. """ text = text.strip() if not text: return [] sentences = re.split(r"(?<=[.!?])\s+|\s+(?=\d+[.)]\s)", text) chunks: list[str] = [] current = "" for sentence in sentences: sentence = sentence.strip() if not sentence: continue if len(sentence) <= limit: candidate = f"{current} {sentence}".strip() if current and len(candidate) > limit: chunks.append(current) current = sentence else: current = candidate continue if current: chunks.append(current) current = "" # Retain commas in the preceding clause so XTTS can reproduce the # intended pause. Semicolons and colons were normalized earlier. clauses = re.split(r"(?<=,)\s+", sentence) long_current = "" for clause in clauses: clause = clause.strip() candidate = f"{long_current} {clause}".strip() if long_current and len(candidate) > limit: chunks.append(long_current) long_current = "" if len(clause) <= limit: long_current = f"{long_current} {clause}".strip() continue # A clause without a usable pause can still exceed the ceiling. for word in clause.split(): candidate = f"{long_current} {word}".strip() if long_current and len(candidate) > limit: chunks.append(long_current) long_current = word else: long_current = candidate if long_current: chunks.append(long_current) if current: chunks.append(current) return chunks def prepare_segments(text: str) -> list[tuple[str, str]]: """Prepare short, deterministic German/English XTTS requests.""" prepared: list[tuple[str, str]] = [] for language, segment in segment_languages(clean_for_speech(text)): if not re.search(r"\w", segment, flags=re.UNICODE): if prepared: previous_language, previous_text = prepared[-1] prepared[-1] = (previous_language, previous_text + segment.strip()) continue spoken = segment if language == "en": spoken = ENGLISH_PRONUNCIATIONS.get(segment.strip().lower(), segment) for chunk in _split_chunk(spoken): if not re.search(r"\w", chunk, flags=re.UNICODE): if prepared: previous_language, previous_text = prepared[-1] prepared[-1] = ( previous_language, previous_text.rstrip() + chunk.strip(), ) continue prepared.append((language, chunk)) return prepared def _speaker_conditioning() -> dict: global SPEAKER_CONDITIONING with SPEAKER_LOCK: if SPEAKER_CONDITIONING is not None: return SPEAKER_CONDITIONING speakers = _json(f"{XTTS_URL}/studio_speakers", XTTS_TIMEOUT) if not isinstance(speakers, dict) or XTTS_SPEAKER not in speakers: raise RuntimeError("configured XTTS speaker is unavailable") selected = speakers[XTTS_SPEAKER] SPEAKER_CONDITIONING = { "speaker_embedding": selected["speaker_embedding"], "gpt_cond_latent": selected["gpt_cond_latent"], } return SPEAKER_CONDITIONING def _stabilize_xtts_ending(text: str) -> str: """Give XTTS a reliable stop cue without changing the visible answer. XTTS v2 can continue with invented syllables after a terminal full stop, especially in German. A terminal semicolon is tokenized as a stronger, more reliable boundary while retaining neutral sentence intonation. """ spoken = text.rstrip() if spoken.endswith((".", "!", "?", ";", ":")): spoken = spoken[:-1].rstrip() return f"{spoken};" def _xtts_pcm(text: str, language: str, conditioning: dict) -> bytes: payload = { **conditioning, "text": _stabilize_xtts_ending(text), "language": language, "add_wav_header": True, "stream_chunk_size": "20", } audio, _ = _request(f"{XTTS_URL}/tts_stream", payload=payload, timeout=XTTS_TIMEOUT) if len(audio) < 44 or audio[:4] != b"RIFF" or audio[8:12] != b"WAVE": raise RuntimeError("XTTS returned invalid WAV data") return audio[44:] def _trim_pcm(pcm: bytes) -> bytes: """Trim generated edge silence while retaining a small safety padding.""" samples = array("h") samples.frombytes(pcm) if not samples: return pcm first = next((i for i, value in enumerate(samples) if abs(value) >= TRIM_THRESHOLD), 0) last = next((i for i in range(len(samples) - 1, -1, -1) if abs(samples[i]) >= TRIM_THRESHOLD), len(samples) - 1) padding = int(24000 * max(0, TRIM_PADDING_MS) / 1000) first = max(0, first - padding) last = min(len(samples) - 1, last + padding) return samples[first:last + 1].tobytes() def _fade_edge(pcm: bytes, *, fade_in: bool = False, fade_out: bool = False) -> bytes: samples = array("h") samples.frombytes(pcm) count = min(len(samples), int(24000 * max(0, CROSSFADE_MS) / 1000)) if count <= 1: return pcm if fade_in: for index in range(count): samples[index] = int(samples[index] * index / (count - 1)) if fade_out: start = len(samples) - count for index in range(count): samples[start + index] = int( samples[start + index] * (count - 1 - index) / (count - 1)) return samples.tobytes() def _compress_internal_silence(pcm: bytes) -> bytes: """Shorten XTTS silence hallucinations while preserving normal pauses.""" samples = array("h") samples.frombytes(pcm) if not samples: return pcm trigger = int(24000 * max(0, INTERNAL_SILENCE_TRIGGER_MS) / 1000) keep = int(24000 * max(0, INTERNAL_SILENCE_KEEP_MS) / 1000) if trigger <= 0 or keep >= trigger: return pcm output = array("h") index = 0 while index < len(samples): if abs(samples[index]) > INTERNAL_SILENCE_THRESHOLD: output.append(samples[index]) index += 1 continue end = index + 1 while end < len(samples) and abs(samples[end]) <= INTERNAL_SILENCE_THRESHOLD: end += 1 run = end - index if run >= trigger: output.extend(samples[index:index + keep]) else: output.extend(samples[index:end]) index = end return output.tobytes() def _join_pcm(parts: list[tuple[str, bytes]]) -> bytes: """Join clips without clicks; pause only at real sentence boundaries.""" if not parts: return b"" output = bytearray() sentence_silence = b"\x00\x00" * int( 24000 * max(0, SENTENCE_PAUSE_MS) / 1000) for index, (text, pcm) in enumerate(parts): pcm = _trim_pcm(pcm) previous_ends_sentence = index > 0 and bool( re.search(r"[.!?;:]\s*$", parts[index - 1][0])) if index == 0: output.extend(_fade_edge(pcm, fade_in=True)) elif previous_ends_sentence: if output: faded = _fade_edge(bytes(output), fade_out=True) output[:] = faded output.extend(sentence_silence) output.extend(_fade_edge(pcm, fade_in=True)) else: # Language switches inside a sentence get no artificial pause. # Small fades remove the discontinuity that otherwise sounds like # a high click or beep between independently generated clips. if output: faded = _fade_edge(bytes(output), fade_out=True) output[:] = faded output.extend(_fade_edge(pcm, fade_in=True)) return _compress_internal_silence(bytes(output)) def _wav(pcm: bytes) -> bytes: output = io.BytesIO() with wave.open(output, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(24000) wav_file.writeframes(pcm) return output.getvalue() def _convert(wav_bytes: bytes, output_format: str, speed: float) -> tuple[bytes, str]: if output_format == "wav" and speed == 1.0: return wav_bytes, "audio/wav" if output_format == "mp3": codec = ["-codec:a", "libmp3lame", "-b:a", "96k", "-f", "mp3"] content_type = "audio/mpeg" elif output_format == "pcm": # Hermes' OpenAI streaming TTS client expects headerless 24 kHz, # mono, signed 16-bit little-endian PCM chunks. codec = ["-ac", "1", "-ar", "24000", "-codec:a", "pcm_s16le", "-f", "s16le"] content_type = "application/octet-stream" else: codec = ["-codec:a", "pcm_s16le", "-f", "wav"] content_type = "audio/wav" command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "wav", "-i", "pipe:0"] if speed != 1.0: command.extend(["-filter:a", f"atempo={speed:.4f}"]) command.extend([*codec, "pipe:1"]) result = subprocess.run( command, input=wav_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, timeout=120) if result.returncode != 0 or not result.stdout: raise RuntimeError("audio conversion failed") return result.stdout, content_type def synthesize_xtts(text: str, output_format: str, speed: float) -> tuple[bytes, str]: conditioning = _speaker_conditioning() pcm_parts: list[tuple[str, bytes]] = [] for language, segment in prepare_segments(text): if not segment.strip(): continue pcm_parts.append((segment, _xtts_pcm(segment, language, conditioning))) if not pcm_parts: raise RuntimeError("no speech segments generated") return _convert(_wav(_join_pcm(pcm_parts)), output_format, speed) def synthesize_piper(text: str, output_format: str, speed: float) -> tuple[bytes, str]: upstream_format = "wav" if output_format == "pcm" else output_format audio, content_type = _request( f"{PIPER_URL}/tts", payload={"text": text, "voice": "alloy", "speed": speed, "format": upstream_format}, timeout=PIPER_TIMEOUT, ) return _convert(audio, "pcm", 1.0) if output_format == "pcm" else (audio, content_type) def synthesize_qwen(text: str, output_format: str, speed: float) -> tuple[bytes, str]: text = prepare_for_qwen_speech(text) upstream_format = "wav" if output_format == "pcm" else output_format audio, content_type = _request( f"{QWEN_TTS_URL}/v1/audio/speech", payload={"model": QWEN_TTS_MODEL, "input": text, "voice": QWEN_TTS_VOICE, "language": QWEN_TTS_LANGUAGE, "response_format": upstream_format, "speed": speed}, timeout=QWEN_TTS_TIMEOUT, ) return _convert(audio, "pcm", 1.0) if output_format == "pcm" else (audio, content_type) def synthesize(text: str, output_format: str, speed: float) -> tuple[bytes, str]: acquired = SYNTHESIS_LOCK.acquire(timeout=QUEUE_TIMEOUT) if acquired: try: audio = synthesize_qwen(text, output_format, speed) with STATE_LOCK: STATE["last_backend"] = "qwen3-tts-1.7b" STATE["last_error"] = None return audio except Exception as exc: # fallback must cover all Qwen failures with STATE_LOCK: STATE["xtts_failures"] += 1 STATE["last_error"] = type(exc).__name__ finally: SYNTHESIS_LOCK.release() else: with STATE_LOCK: STATE["xtts_failures"] += 1 STATE["last_error"] = "queue-timeout" audio = synthesize_piper(text, output_format, speed) with STATE_LOCK: STATE["last_backend"] = "piper" STATE["piper_fallbacks"] += 1 return audio class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt: str, *args: object) -> None: # Never log request URLs, bodies, synthesized text or speaker vectors. print(f"tts-gateway: {self.command} -> {args[1] if len(args) > 1 else '-'}") def send_bytes(self, status: int, body: bytes, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def send_json(self, status: int, payload: dict) -> None: self.send_bytes(status, json.dumps(payload, separators=(",", ":")).encode(), "application/json") def do_GET(self) -> None: # noqa: N802 if self.path != "/status": self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) return primary_ready = _reachable(QWEN_TTS_URL, "/health") fallback_ready = _reachable(PIPER_URL, "/status") with STATE_LOCK: state = dict(STATE) self.send_json( HTTPStatus.OK if fallback_ready else HTTPStatus.SERVICE_UNAVAILABLE, { "ready": fallback_ready, "engine": "qwen3-tts-with-piper-fallback", "model": "Qwen3-TTS-12Hz-1.7B-Base", "voices": [VOICE_ALIAS], "speaker": QWEN_TTS_VOICE, "primary_ready": primary_ready, "fallback_ready": fallback_ready, **state, }, ) def do_POST(self) -> None: # noqa: N802 if self.path != "/tts": self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: length = 0 if length <= 0 or length > MAX_REQUEST_BYTES: self.send_json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid request size"}) return try: request = json.loads(self.rfile.read(length)) text = request.get("text", "") voice = request.get("voice", VOICE_ALIAS) output_format = request.get("format", "mp3") speed = float(request.get("speed", 1.0)) except (json.JSONDecodeError, TypeError, ValueError): self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON request"}) return if not isinstance(text, str) or not text.strip() or len(text) > MAX_TEXT_CHARS: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid text"}) return if voice != VOICE_ALIAS: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "unknown voice"}) return if output_format not in {"wav", "mp3", "pcm"}: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "unsupported format"}) return if not 0.5 <= speed <= 2.0: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid speed"}) return started = time.monotonic() try: audio, content_type = synthesize(text.strip(), output_format, speed) except Exception as exc: with STATE_LOCK: STATE["last_error"] = type(exc).__name__ self.send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"error": "all local speech backends failed"}) return print(f"tts-gateway: synthesized via {STATE['last_backend']} in " f"{time.monotonic() - started:.2f}s") self.send_bytes(HTTPStatus.OK, audio, content_type) if __name__ == "__main__": print(f"TTS gateway ready on {HOST}:{PORT}; primary={QWEN_TTS_VOICE}; fallback=Piper") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()