AI-Profile-Router/dev/mock_tts_worker.py

#!/usr/bin/env python3
"""Mock-TTS-Worker für lokale Tests (gleiche HTTP-API wie tts_worker.py).

Erzeugt ein kurzes, leises WAV statt echten Audios.

API:
  GET  /status  -> {"status":"ok","ready":true,"voices":[...],...}
  POST /tts     -> {"text":"...","voice":"...","speed":1.0,"format":"wav"}
                   -> binäres Audio (WAV/MP3/FLAC/PCM)

Optionen (Umgebungsvariablen):
  MOCK_TTS_PORT    Port (Default 8082)
  MOCK_TTS_DELAY   Sekunden pro Synthese (Default 0.2)
  MOCK_TTS_LOG     Datei, in die die Requests geloggt werden (JSON-Zeilen)

Sonder-Texte:
  "FAIL"  -> Worker antwortet mit 500 (simulierter Fehler)
  "SLOW"  -> Worker schläft 5 s
"""

import io
import json
import os
import struct
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PORT = int(os.environ.get("MOCK_TTS_PORT", "8082"))
DELAY = float(os.environ.get("MOCK_TTS_DELAY", "0.2"))
LOG_FILE = os.environ.get("MOCK_TTS_LOG", "")
SAMPLE_RATE = 24000


def _make_wav(seconds: float = 0.1) -> bytes:
    """Erzeugt ein kurzes WAV (16-bit, mono, 24 kHz) mit Sinus-Ton."""
    import math
    n = int(SAMPLE_RATE * seconds)
    samples = b"".join(
        struct.pack("<h", int(8000 * math.sin(2 * math.pi * 440 * i / SAMPLE_RATE)))
        for i in range(n)
    )
    header = (
        b"RIFF" + struct.pack("<I", 36 + len(samples)) + b"WAVE"
        + b"fmt " + struct.pack("<IHHIIHH", 16, 1, 1, SAMPLE_RATE,
                                SAMPLE_RATE * 2, 2, 16)
        + b"data" + struct.pack("<I", len(samples))
    )
    return header + samples


def _log_request(req: dict) -> None:
    if not LOG_FILE:
        return
    try:
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(req) + "\n")
    except OSError:
        pass


class Handler(BaseHTTPRequestHandler):
    server_version = "MockKokoroTTS/1.0"
    timeout = 60

    def log_message(self, fmt, *args):  # noqa: N802
        pass  # leise

    def do_GET(self):  # noqa: N802
        if self.path.split("?", 1)[0] == "/status":
            self._send_json(200, {
                "status": "ok",
                "ready": True,
                "voices": ["martin", "victoria"],
                "default_voice": "martin",
                "load_errors": [],
                "sample_rate": SAMPLE_RATE,
                "uptime_seconds": 1.0,
                "total_requests": 0,
                "last_seconds": None,
                "last_voice": None,
                "last_error": None,
            })
        else:
            self._send_json(404, {"error": "not found"})

    def do_POST(self):  # noqa: N802
        if self.path.split("?", 1)[0] != "/tts":
            self._send_json(404, {"error": "not found"})
            return
        length = int(self.headers.get("Content-Length") or 0)
        try:
            req = json.loads(self.rfile.read(length))
        except ValueError:
            self._send_json(400, {"error": "ungültiges JSON"})
            return
        _log_request(req)
        text = req.get("text", "")
        if text == "SLOW":
            time.sleep(5.0)
        else:
            time.sleep(DELAY)
        if text == "FAIL":
            self._send_json(500, {"error": "simulierter TTS-Fehler"})
            return
        fmt = req.get("format", "wav")
        if fmt == "wav":
            data, ctype = _make_wav(), "audio/wav"
        elif fmt == "mp3":
            # Minimales MP3-Frame (silence) – reicht für Format-Tests.
            data = b"\xff\xfb\x90\x00" + b"\x00" * 417
            ctype = "audio/mpeg"
        elif fmt == "flac":
            data = b"fLaC" + b"\x00" * 100
            ctype = "audio/flac"
        else:  # pcm
            data = b"\x00" * 4800
            ctype = "application/octet-stream"
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(data)

    def _send_json(self, code: int, payload: dict) -> None:
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)


def main() -> None:
    server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    server.daemon_threads = True
    print(f"Mock-TTS-Worker auf Port {PORT}", file=sys.stderr)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()