Logpy-AgentOne/pipe/transcribe.py

"""Spracherkennung: Audiodatei -> Transkript (lokal, kein Cloud-Aufruf).

Zwei Backends:
  - "whispercpp": nutzt das CLI `whisper-cli` mit einem vorhandenen ggml-Modell
                  (Apple-GPU-beschleunigt, kein Modell-Download).
  - "faster":     faster-whisper (CTranslate2); lädt sein Modell bei Bedarf.

Beliebige Eingabeformate/Abtastraten werden vor der Erkennung per ffmpeg auf
16 kHz mono normalisiert — passt auch für 8-kHz-Telefonaufnahmen.
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path

from . import config


@dataclass
class Transkript:
    text: str
    sprache: str
    sprache_wahrscheinlichkeit: float
    dauer: float
    segmente: list[dict] = field(default_factory=list)


# Bekannte Whisper-Halluzinationen: bei Stille/Rauschen (z.B. Anrufer legt
# gleich nach dem Signalton auf) erfindet Whisper haeufig auswendig gelernte
# Untertitel-Textbausteine aus den Trainingsdaten, statt "kein Text" zu
# liefern - "Untertitelung des ZDF, 2020" trat bei uns mehrfach auf identisch
# kurzen (1,7-4s) Aufnahmen auf und wurde anschliessend von der Kategorisierung
# als echtes Anliegen fehlinterpretiert. whisper.cpp's eingebauter
# no-speech-Schwellwert (--no-speech-thold, Default 0.6) faengt das nicht
# zuverlaessig ab. Statt einer echten VAD (zusaetzliches Modell noetig) reicht
# hier ein Blocklist-Abgleich gegen die dokumentiert haeufigsten Phrasen.
_BEKANNTE_HALLUZINATIONEN = [
    re.compile(p, re.IGNORECASE) for p in (
        r"untertitel(ung)?\s+(des|im auftrag des|von)\s+(zdf|wdr|ndr|swr|br|mdr|hr|rbb)",
        r"copyright\s+(wdr|zdf|ndr|swr|br)\s*\d{4}",
        r"untertitel\s+von\s+stephanie\s+geiges",
        r"vielen\s+dank\s+f(ü|u)rs?\s+zuschauen",
        r"amara\.org",
    )
]


def _ist_bekannte_halluzination(text: str) -> bool:
    return any(m.search(text) for m in _BEKANNTE_HALLUZINATIONEN)


def _letzte_zeile(ausgabe: str | None) -> str:
    """Letzte nicht-leere Zeile einer Fehlerausgabe (für knappe Meldungen)."""
    zeilen = [z for z in (ausgabe or "").strip().splitlines() if z.strip()]
    return zeilen[-1].strip() if zeilen else "keine Fehlerausgabe"


def _nach_16k_mono(audio: Path) -> Path:
    """Normalisiert beliebiges Audio auf 16 kHz mono WAV (Tempdatei)."""
    fd, pfad = tempfile.mkstemp(suffix=".wav", prefix="stt_")
    os.close(fd)
    ziel = Path(pfad)
    try:
        subprocess.run(
            ["ffmpeg", "-y", "-i", str(audio), "-ar", "16000", "-ac", "1", str(ziel)],
            check=True, capture_output=True, text=True,
        )
    except FileNotFoundError as fehler:
        ziel.unlink(missing_ok=True)
        raise RuntimeError("ffmpeg wurde nicht gefunden (brew install ffmpeg).") from fehler
    except subprocess.CalledProcessError as fehler:
        ziel.unlink(missing_ok=True)
        raise RuntimeError(
            f"ffmpeg konnte {audio.name} nicht lesen: {_letzte_zeile(fehler.stderr)}"
        ) from fehler
    return ziel


def transkribiere(audio: str | Path) -> Transkript:
    """Wandelt eine Audiodatei in Text um. FileNotFoundError, wenn sie fehlt."""
    pfad = Path(audio)
    if not pfad.is_file():
        raise FileNotFoundError(f"Audiodatei nicht gefunden: {pfad}")

    if config.STT_BACKEND == "faster":
        return _mit_faster_whisper(pfad)
    return _mit_whispercpp(pfad)


# --- Backend: whisper.cpp -----------------------------------------------------

def _mit_whispercpp(audio: Path) -> Transkript:
    wav = _nach_16k_mono(audio)
    fd, pfad = tempfile.mkstemp(suffix="", prefix="stt_out_")
    os.close(fd)
    ausgabe_basis = Path(pfad)
    try:
        cmd = [
            config.WHISPERCPP_BIN,
            "-m", config.WHISPERCPP_MODELL,
            "-f", str(wav),
            "-l", config.WHISPER_SPRACHE,
            "-t", str(config.WHISPER_THREADS),
            "-oj", "-of", str(ausgabe_basis),
            "-np",
        ]
        if config.WHISPER_PROMPT:
            cmd += ["--prompt", config.WHISPER_PROMPT]
        try:
            subprocess.run(cmd, check=True, capture_output=True, text=True)
        except FileNotFoundError as fehler:
            raise RuntimeError(
                f"Spracherkennung '{config.WHISPERCPP_BIN}' nicht gefunden "
                f"(brew install whisper-cpp, oder STT_BACKEND=faster setzen)."
            ) from fehler
        except subprocess.CalledProcessError as fehler:
            raise RuntimeError(
                f"whisper-cli brach ab: {_letzte_zeile(fehler.stderr)}"
            ) from fehler
        roh = json.loads(Path(f"{ausgabe_basis}.json").read_text(encoding="utf-8"))
    finally:
        wav.unlink(missing_ok=True)
        Path(f"{ausgabe_basis}.json").unlink(missing_ok=True)
        ausgabe_basis.unlink(missing_ok=True)

    segmente = []
    for s in roh.get("transcription", []):
        text = (s.get("text") or "").strip()
        if text:
            off = s.get("offsets", {})
            segmente.append({
                "start": round(off.get("from", 0) / 1000, 2),
                "ende": round(off.get("to", 0) / 1000, 2),
                "text": text,
            })
    text_ganz = " ".join(s["text"] for s in segmente).strip()
    sprache = roh.get("result", {}).get("language", config.WHISPER_SPRACHE)
    dauer = segmente[-1]["ende"] if segmente else 0.0
    if _ist_bekannte_halluzination(text_ganz):
        text_ganz, segmente = "", []
    return Transkript(text=text_ganz, sprache=sprache,
                      sprache_wahrscheinlichkeit=1.0, dauer=dauer, segmente=segmente)


# --- Backend: faster-whisper --------------------------------------------------

@lru_cache(maxsize=1)
def _faster_modell():
    from faster_whisper import WhisperModel
    return WhisperModel(config.WHISPER_MODELL, device="cpu",
                        compute_type=config.WHISPER_COMPUTE)


def _mit_faster_whisper(audio: Path) -> Transkript:
    sprache = None if config.WHISPER_SPRACHE == "auto" else config.WHISPER_SPRACHE
    segmente, info = _faster_modell().transcribe(str(audio), language=sprache, vad_filter=True)
    teile, seg_liste = [], []
    for s in segmente:
        t = s.text.strip()
        if t:
            teile.append(t)
            seg_liste.append({"start": round(s.start, 2), "ende": round(s.end, 2), "text": t})
    text_ganz = " ".join(teile).strip()
    if _ist_bekannte_halluzination(text_ganz):
        text_ganz, seg_liste = "", []
    return Transkript(
        text=text_ganz, sprache=info.language,
        sprache_wahrscheinlichkeit=round(info.language_probability, 3),
        dauer=round(info.duration, 2), segmente=seg_liste,
    )