#!/usr/bin/env python3 """ STT-Worker – langlebiger Whisper-Transkriptions-Service (CPU-only). Liest Audio-Dateien (WAV, MP3, OGG, FLAC, WebM/Opus via ffmpeg), transkribiert sie mit einem dauerhaft geladenen whisper.cpp-Server (mit whisper-cli als Rückfallweg) und liefert JSON-Text. Konfiguration über Umgebungsvariablen: WHISPER_HOST Bind-Adresse (Default: 127.0.0.1) WHISPER_PORT Port (Default: 8083) WHISPER_CLI Pfad zu whisper-cli (Default: /opt/mike-ai/whisper.cpp/build-cpu/bin/whisper-cli) WHISPER_MODEL Pfad zum ggml-Modell (Default: /opt/mike-ai/models/whisper/ggml-large-v3-turbo.bin) WHISPER_THREADS Anzahl CPU-Threads (Default: 8) WHISPER_LANGUAGE Standard-Sprache (Default: de) FFMPEG_BIN Pfad zu ffmpeg (Default: /usr/bin/ffmpeg) LOG_LEVEL Logging-Level (Default: INFO) Endpunkte: GET /status → Health + Konfiguration POST /transcribe → Audio-Datei transkribieren (multipart/form-data oder raw body) """ import json import logging import os import subprocess import sys import tempfile import time import uuid import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # --------------------------------------------------------------------------- # Konfiguration # --------------------------------------------------------------------------- HOST = os.environ.get("WHISPER_HOST", "127.0.0.1") PORT = int(os.environ.get("WHISPER_PORT", "8083")) WHISPER_CLI = os.environ.get( "WHISPER_CLI", "/opt/mike-ai/whisper.cpp/build-cpu/bin/whisper-cli", ) WHISPER_MODEL = os.environ.get( "WHISPER_MODEL", "/opt/mike-ai/models/whisper/ggml-large-v3-turbo.bin", ) WHISPER_THREADS = int(os.environ.get("WHISPER_THREADS", "8")) WHISPER_LANGUAGE = os.environ.get("WHISPER_LANGUAGE", "de") WHISPER_SERVER_URL = os.environ.get("WHISPER_SERVER_URL", "").rstrip("/") FFMPEG_BIN = os.environ.get("FFMPEG_BIN", "/usr/bin/ffmpeg") LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") # Audio-Formate, die whisper.cpp nativ unterstützt NATIVE_FORMATS = {".wav", ".mp3", ".ogg", ".flac"} # Formate, die ffmpeg-Konvertierung benötigen CONVERT_FORMATS = {".webm", ".m4a", ".aac", ".opus", ".wma", ".amr", ".mka"} logging.basicConfig( level=getattr(logging, LOG_LEVEL.upper(), logging.INFO), format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout, ) log = logging.getLogger("stt-worker") # --------------------------------------------------------------------------- # Audio-Konvertierung # --------------------------------------------------------------------------- def _detect_format(filename: str) -> str: """Erkennt das Dateiformat anhand der Endung.""" ext = os.path.splitext(filename)[1].lower() return ext def _convert_to_wav(input_path: str, output_path: str) -> None: """Konvertiert Audio per ffmpeg zu 16 kHz mono WAV (s16).""" cmd = [ FFMPEG_BIN, "-y", "-i", input_path, "-ar", "16000", "-ac", "1", "-sample_fmt", "s16", "-c:a", "pcm_s16le", output_path, ] proc = subprocess.run( cmd, capture_output=True, text=True, timeout=30, ) if proc.returncode != 0: raise RuntimeError(f"ffmpeg-Fehler: {proc.stderr[-500:]}") def _prepare_audio(data: bytes, filename: str) -> str: """ Bereitet Audio-Datei für whisper-cli vor. Liefert Pfad zu einer WAV-Datei (16 kHz mono s16). """ ext = _detect_format(filename) if ext in NATIVE_FORMATS: # Nativ unterstützt – direkt verwenden tmp = tempfile.NamedTemporaryFile( suffix=ext, prefix="stt_", delete=False ) tmp.write(data) tmp.close() return tmp.name if ext in CONVERT_FORMATS: # ffmpeg-Konvertierung nötig tmp_in = tempfile.NamedTemporaryFile( suffix=ext, prefix="stt_in_", delete=False ) tmp_in.write(data) tmp_in.close() tmp_out = tempfile.NamedTemporaryFile( suffix=".wav", prefix="stt_out_", delete=False ) tmp_out.close() _convert_to_wav(tmp_in.name, tmp_out.name) os.unlink(tmp_in.name) return tmp_out.name # Unbekanntes Format – versuchen, es als WAV zu behandeln tmp = tempfile.NamedTemporaryFile( suffix=".wav", prefix="stt_", delete=False ) tmp.write(data) tmp.close() return tmp.name # --------------------------------------------------------------------------- # Transkription # --------------------------------------------------------------------------- def transcribe( audio_path: str, language: str | None = None, prompt: str | None = None, temperature: float | None = None, ) -> dict: """ Führt die Transkription bevorzugt über den persistenten whisper.cpp- Server aus. Dadurch wird das Modell nicht pro Aufnahme neu geladen. Liefert dict mit 'text' und Metadaten. """ lang = language or WHISPER_LANGUAGE if lang == "auto": lang = "auto" if WHISPER_SERVER_URL: return _transcribe_via_server( audio_path, language=lang, prompt=prompt, temperature=temperature, ) out_prefix = f"/tmp/stt_{uuid.uuid4().hex[:12]}" out_json = out_prefix + ".json" cmd = [ WHISPER_CLI, "-m", WHISPER_MODEL, "-f", audio_path, "-l", lang, "-t", str(WHISPER_THREADS), "-oj", "-of", out_prefix, "-np", ] if prompt: cmd.extend(["--prompt", prompt]) if temperature is not None: cmd.extend(["-tp", str(temperature)]) t0 = time.monotonic() proc = subprocess.run( cmd, capture_output=True, text=True, timeout=300, ) elapsed = time.monotonic() - t0 if proc.returncode != 0: raise RuntimeError( f"whisper-cli-Fehler (rc={proc.returncode}): " f"{proc.stderr[-500:]}" ) # JSON-Output lesen result = {"text": "", "language": lang, "duration_ms": int(elapsed * 1000)} if os.path.exists(out_json): with open(out_json, "r", encoding="utf-8") as f: jdata = json.load(f) # whisper.cpp JSON-Format: # {"transcription": [{"text": "...", "offsets": {"from": 0, "to": 1000}}], # "result": {"language": "de"}, ...} transcription = jdata.get("transcription", []) if isinstance(transcription, list): texts = [t.get("text", "") for t in transcription if isinstance(t, dict)] result["text"] = " ".join(texts).strip() # Audio-Dauer aus letztem Segment if transcription and isinstance(transcription[-1], dict): offsets = transcription[-1].get("offsets", {}) if offsets: result["audio_duration_ms"] = offsets.get("to", 0) elif isinstance(transcription, str): result["text"] = transcription.strip() # Sprache aus result.language if "result" in jdata and isinstance(jdata["result"], dict): if "language" in jdata["result"]: result["language"] = jdata["result"]["language"] elif "language" in jdata: result["language"] = jdata["language"] os.unlink(out_json) # Aufräumen for suffix in (".wav", ".mp3", ".ogg", ".flac", ".json"): p = out_prefix + suffix if os.path.exists(p): os.unlink(p) log.info( "Transkription: %d ms, %d Zeichen, Sprache=%s", result["duration_ms"], len(result["text"]), result["language"], ) return result def _transcribe_via_server( audio_path: str, language: str, prompt: str | None = None, temperature: float | None = None, ) -> dict: """Sendet eine Aufnahme an den bereits geladenen whisper-server.""" boundary = f"----athena-whisper-{uuid.uuid4().hex}" chunks: list[bytes] = [] def add_field(name: str, value: str) -> None: chunks.extend([ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(), value.encode("utf-8"), b"\r\n", ]) with open(audio_path, "rb") as f: audio = f.read() chunks.extend([ f"--{boundary}\r\n".encode(), b'Content-Disposition: form-data; name="file"; filename="audio.wav"\r\n', b"Content-Type: audio/wav\r\n\r\n", audio, b"\r\n", ]) add_field("response_format", "json") add_field("language", language) if prompt: add_field("prompt", prompt) if temperature is not None: add_field("temperature", str(temperature)) chunks.append(f"--{boundary}--\r\n".encode()) request = urllib.request.Request( f"{WHISPER_SERVER_URL}/inference", data=b"".join(chunks), headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, method="POST", ) t0 = time.monotonic() with urllib.request.urlopen(request, timeout=300) as response: payload = json.loads(response.read().decode("utf-8")) elapsed = time.monotonic() - t0 text = payload.get("text", "") if isinstance(payload, dict) else "" result = { "text": text.strip(), "language": language, "duration_ms": int(elapsed * 1000), "engine": "whisper-server", } log.info( "Transkription (persistent): %d ms, %d Zeichen, Sprache=%s", result["duration_ms"], len(result["text"]), language, ) return result # --------------------------------------------------------------------------- # HTTP-Handler # --------------------------------------------------------------------------- class STTHandler(BaseHTTPRequestHandler): server_version = "STTWorker/1.0" def log_message(self, fmt, *args): log.info("%s %s", self.address_string(), fmt % args) 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 _parse_multipart(self, data: bytes, content_type: str) -> tuple[bytes, str, dict]: """ Parst multipart/form-data. Liefert (file_data, filename, form_fields). """ # Boundary extrahieren boundary = None for part in content_type.split(";"): part = part.strip() if part.startswith("boundary="): boundary = part[len("boundary="):] break if not boundary: raise ValueError("Kein Boundary in Content-Type") boundary_bytes = boundary.encode("utf-8") file_data = b"" filename = "" fields = {} # Multipart parsen parts = data.split(b"--" + boundary_bytes) for part in parts: if part in (b"", b"--", b"--\r\n", b"\r\n"): continue # Header und Body trennen if b"\r\n\r\n" not in part: continue header_part, body_part = part.split(b"\r\n\r\n", 1) # Trailing CRLF entfernen 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: # Datei-Feld 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: # Text-Feld name = line.split("name=")[1].strip().strip('"') fields[name] = body_part.decode("utf-8", errors="replace") return file_data, filename, fields def do_GET(self): if self.path == "/status": model_ok = os.path.isfile(WHISPER_MODEL) cli_ok = os.path.isfile(WHISPER_CLI) self._send_json(200, { "ready": model_ok and cli_ok, "model": WHISPER_MODEL, "model_exists": model_ok, "whisper_cli": WHISPER_CLI, "whisper_cli_exists": cli_ok, "threads": WHISPER_THREADS, "language": WHISPER_LANGUAGE, "server_url": WHISPER_SERVER_URL or None, "persistent_model": bool(WHISPER_SERVER_URL), "ffmpeg": FFMPEG_BIN, "ffmpeg_exists": os.path.isfile(FFMPEG_BIN), }) else: self._send_json(404, {"error": "nicht gefunden"}) def do_POST(self): if self.path != "/transcribe": self._send_json(404, {"error": "nicht gefunden"}) return content_type = self.headers.get("Content-Type", "") try: if "multipart/form-data" in content_type: data = self._read_body() file_data, filename, fields = self._parse_multipart( data, content_type ) if not file_data: self._send_json(400, {"error": "Keine Datei im Request"}) return language = fields.get("language") prompt = fields.get("prompt") temperature = fields.get("temperature") if temperature: temperature = float(temperature) else: # Raw body (direkte Audio-Daten) file_data = self._read_body() filename = self.headers.get("X-Filename", "audio.wav") language = self.headers.get("X-Language") prompt = self.headers.get("X-Prompt") temperature = self.headers.get("X-Temperature") if temperature: temperature = float(temperature) if not file_data: self._send_json(400, {"error": "Leerer Request-Body"}) return # Audio vorbereiten audio_path = _prepare_audio(file_data, filename) try: result = transcribe( audio_path, language=language, prompt=prompt, temperature=temperature, ) finally: os.unlink(audio_path) self._send_json(200, result) except ValueError as e: self._send_json(400, {"error": str(e)}) except subprocess.TimeoutExpired: self._send_json(504, {"error": "Transkription-Timeout"}) except Exception as e: log.exception("Transkriptions-Fehler") self._send_json(500, {"error": str(e)}) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): log.info( "STT-Worker startet: host=%s port=%d model=%s threads=%d lang=%s", HOST, PORT, WHISPER_MODEL, WHISPER_THREADS, WHISPER_LANGUAGE, ) if not os.path.isfile(WHISPER_MODEL): log.warning("Modell nicht gefunden: %s", WHISPER_MODEL) if not os.path.isfile(WHISPER_CLI): log.warning("whisper-cli nicht gefunden: %s", WHISPER_CLI) server = ThreadingHTTPServer((HOST, PORT), STTHandler) log.info("STT-Worker lauscht auf %s:%d", HOST, PORT) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()