152 lines
5.2 KiB
Python
152 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Mock-STT-Worker für lokale Tests.
|
|
|
|
Simuliert den Qwen3-ASR-Adapter:
|
|
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()
|