AI-Profile-Router/dev/mock_stt_worker.py

#!/usr/bin/env python3
"""Mock-STT-Worker für lokale Tests.

Simuliert den Whisper-STT-Worker:
  GET  /status      → ready: true
  POST /transcribe  → liefert festes Transkript

Konfiguration:
  MOCK_STT_PORT     Port (Default: 18083)
  MOCK_STT_DELAY    Verzögerung in Sekunden (Default: 0.1)
  MOCK_STT_LOG      JSONL-Log für Requests (optional)
"""

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

PORT = int(os.environ.get("MOCK_STT_PORT", "18083"))
DELAY = float(os.environ.get("MOCK_STT_DELAY", "0.1"))
LOG_FILE = os.environ.get("MOCK_STT_LOG", "")


class MockSTTHandler(BaseHTTPRequestHandler):
    server_version = "MockSTT/1.0"

    def log_message(self, fmt, *args):
        pass

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

    def _read_body(self) -> bytes:
        length = int(self.headers.get("Content-Length", 0))
        return self.rfile.read(length) if length > 0 else b""

    def do_GET(self):
        if self.path == "/status":
            self._send_json(200, {
                "ready": True,
                "model": "mock-whisper-large-v3-turbo",
                "model_exists": True,
                "whisper_cli_exists": True,
                "threads": 8,
                "language": "de",
                "ffmpeg_exists": True,
            })
        else:
            self._send_json(404, {"error": "nicht gefunden"})

    def do_POST(self):
        if self.path != "/transcribe":
            self._send_json(404, {"error": "nicht gefunden"})
            return

        data = self._read_body()
        content_type = self.headers.get("Content-Type", "")

        # Multipart parsen (einfach)
        filename = ""
        language = None
        prompt = None
        temperature = None
        file_data = b""

        if "multipart/form-data" in content_type:
            # Boundary extrahieren
            boundary = None
            for part in content_type.split(";"):
                part = part.strip()
                if part.startswith("boundary="):
                    boundary = part[len("boundary="):]
                    break
            if boundary:
                boundary_bytes = boundary.encode("utf-8")
                parts = data.split(b"--" + boundary_bytes)
                for part in parts:
                    if part in (b"", b"--", b"--\r\n", b"\r\n"):
                        continue
                    if b"\r\n\r\n" not in part:
                        continue
                    header_part, body_part = part.split(b"\r\n\r\n", 1)
                    if body_part.endswith(b"\r\n"):
                        body_part = body_part[:-2]
                    header_text = header_part.decode("utf-8", errors="replace")
                    for line in header_text.split("\r\n"):
                        if "name=" in line and "filename=" in line:
                            for kv in line.split(";"):
                                kv = kv.strip()
                                if kv.startswith("filename="):
                                    filename = kv[len("filename="):].strip('"')
                            file_data = body_part
                        elif "name=" in line:
                            name = line.split("name=")[1].strip().strip('"')
                            if name == "language":
                                language = body_part.decode("utf-8", errors="replace")
                            elif name == "prompt":
                                prompt = body_part.decode("utf-8", errors="replace")
                            elif name == "temperature":
                                temperature = body_part.decode("utf-8", errors="replace")

        # Log
        if LOG_FILE:
            entry = {
                "timestamp": time.time(),
                "filename": filename,
                "file_size": len(file_data),
                "language": language,
                "prompt": prompt,
                "temperature": temperature,
            }
            with open(LOG_FILE, "a") as f:
                f.write(json.dumps(entry) + "\n")

        time.sleep(DELAY)

        # Simuliertes Transkript
        text = "Hallo, dies ist ein Test der deutschen Spracherkennung."
        if language == "de":
            text = "Hallo, dies ist ein Test der deutschen Spracherkennung."
        elif language == "en":
            text = "Hello, this is a test of English speech recognition."

        self._send_json(200, {
            "text": text,
            "language": language or "de",
            "duration_ms": int(DELAY * 1000),
            "audio_duration_ms": 7300,
        })


def main():
    print(f"Mock-STT-Worker lauscht auf Port {PORT}", flush=True)
    server = ThreadingHTTPServer(("127.0.0.1", PORT), MockSTTHandler)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()