app.py

"""API-Backend der Sprecher-Analyse (HF Space auf ZeroGPU).

Die eigentliche App (PWA) läuft als eigener Static Space und spricht die
hier registrierten API-Endpunkte an. Jeder Endpunkt verlangt den
Zugangsschlüssel (Secret APP_PASS); erst nach der Prüfung wird GPU-Zeit
verbraucht.
"""
import hmac
import json
import os
import re
import secrets
import subprocess
import tempfile
import threading
import time
import traceback
import unicodedata
import math
from datetime import datetime
from zoneinfo import ZoneInfo

import gradio as gr
import spaces
import torch
import torchaudio.functional as AF
import yaml
from torchaudio.pipelines import MMS_FA as FA_BUNDLE
from huggingface_hub import HfApi, hf_hub_download
from pyannote.audio import Pipeline
from pyannote.audio.core.io import Audio
from pyannote.core import Segment
from transformers import pipeline as hf_pipeline

from kern import (FA_ANTEIL, FA_BLOCK, FA_FENSTER, FARBEN, FENSTER,
                  GLOSSAR_DATEI, GLOSSAR_MAX, GLOSSAR_PROMPT,
                  _begriffe_zaehlen, _decode_optionen, _fenstergrenzen,
                  _ist_halluzination, _konfidenz_markieren, _korrekturziele,
                  _korrigieren, abgleich, aufnahme_urteil,
                  KORREKTUR_MIN_ZAEHLER, stimmen_zuordnen,
                  stimmen_auffrischen,
                  _frontmatter, _kontext_bauen, _markdown, _namen_farben,
                  _rangfolge, _romanisieren, _sprecher_bei, _zeit,
                  _zusammenfuegen, glossar_lesen, glossar_schreiben,
                  restzeit, tempo, _woerter_zaehlen,
                  KONTINGENT_S, kontingent_buchen,
                  kontingent_korrigieren, kontingent_stand,
                  pfad_erlaubt)

HF_TOKEN = os.environ["HF_TOKEN"]
APP_KEY = os.environ["APP_PASS"]
DATEN_REPO = os.environ.get("DATEN_REPO", "Monstermango/diarization-results")
FRONTEND = os.environ.get("FRONTEND_URL",
                          "https://monstermango-sprecher-analyse.static.hf.space")
ZEITZONE = ZoneInfo("Europe/Berlin")

api = HfApi(token=HF_TOKEN)
api.create_repo(DATEN_REPO, repo_type="dataset", private=True, exist_ok=True)

diarizer = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-community-1", token=HF_TOKEN
)
diarizer.to(torch.device("cuda"))

# Auf Deutsch nachtrainiertes large-v3: bei deutschen Gesprächsaufnahmen
# deutlich treffsicherer als das allgemeine Modell. Fällt bei anderen
# Sprachen auf das Original zurück.
ASR_MODELLE = {
    "german": "primeline/whisper-large-v3-german",
    "*": "openai/whisper-large-v3",
}
_asr_cache = {}


def _asr_holen(sprache):
    name = ASR_MODELLE.get(sprache or "", ASR_MODELLE["*"])
    if name not in _asr_cache:
        # SDPA rechnet die Aufmerksamkeit über PyTorchs eigene Kerne statt
        # über die Referenzschleife — dieselben Ausgaben, spürbar
        # schneller. Fällt zurück, falls die Fassung es nicht kennt.
        gemeinsam = dict(model=name, torch_dtype=torch.float16, device="cuda")
        try:
            _asr_cache[name] = hf_pipeline(
                "automatic-speech-recognition",
                model_kwargs={"attn_implementation": "sdpa"}, **gemeinsam)
        except Exception:
            traceback.print_exc()
            _asr_cache[name] = hf_pipeline(
                "automatic-speech-recognition", **gemeinsam)
    return _asr_cache[name]


asr = _asr_holen("german")

loader = Audio(sample_rate=16000, mono="downmix")

# Modell zum Einpassen eines mitgelieferten Transkripts
FA_GERAET = torch.device("cuda")
FA_DICT = FA_BUNDLE.get_dict(star="*")
FA_STERN = FA_DICT["*"]
fa_modell = FA_BUNDLE.get_model(with_star=True).to(FA_GERAET).eval()


_fehlversuche = {"anzahl": 0, "zeit": 0.0}


def _pruefe(key):
    """Vergleicht laufzeitkonstant und bremst nach Fehlversuchen.

    Der Space ist öffentlich erreichbar; ohne Bremse ließe sich der
    Schlüssel automatisiert durchprobieren.
    """
    gueltig = bool(key) and hmac.compare_digest(str(key), APP_KEY)
    if gueltig:
        _fehlversuche["anzahl"] = 0
        return True
    jetzt = time.time()
    if jetzt - _fehlversuche["zeit"] > 300:
        _fehlversuche["anzahl"] = 0
    _fehlversuche["anzahl"] += 1
    _fehlversuche["zeit"] = jetzt
    time.sleep(min(5.0, 0.25 * _fehlversuche["anzahl"]))
    return False


# Lange Aufnahmen werden in Fenster zerlegt, damit jeder GPU-Aufruf innerhalb
# des ZeroGPU-Limits bleibt. Die Diarization läuft trotzdem über die gesamte
# Datei, sonst wären die Sprecher zwischen den Fenstern nicht dieselben.


KONTO_DATEI = "gpu_konto.json"
GPU_BUDGET = float(os.environ.get("GPU_BUDGET_S") or KONTINGENT_S)
_konto = {"daten": None, "schmutzig": False}


def _konto_laden():
    if _konto["daten"] is None:
        daten = {}
        try:
            pfad = hf_hub_download(DATEN_REPO, KONTO_DATEI, repo_type="dataset",
                                   token=HF_TOKEN, force_download=True)
            with open(pfad, encoding="utf-8") as fh:
                daten = json.load(fh)
        except Exception:
            pass
        _konto["daten"] = daten
    return _konto["daten"]


def _konto_sichern():
    if not _konto["schmutzig"]:
        return
    try:
        api.upload_file(
            path_or_fileobj=json.dumps(_konto["daten"]).encode("utf-8"),
            path_in_repo=KONTO_DATEI, repo_id=DATEN_REPO, repo_type="dataset",
            commit_message="GPU-Konto aktualisiert")
        _konto["schmutzig"] = False
    except Exception:
        traceback.print_exc()


# Obergrenze je Aufruf. Die Wartezeit in der Schlange kostet kein
# Kontingent, ist von aussen aber nicht von der Rechenzeit zu trennen —
# und weiter als bis zur Reservierung kann ein Aufruf nie laufen, weil er
# dort abgebrochen wird. Also deckeln statt masslos überschätzen.
GPU_DECKEL_S = 200.0


def bucht_gpu(fn):
    """Zählt die Zeit über einem GPU-Aufruf.

    Muss OBERHALB von @spaces.GPU stehen. Darunter liefe die Zählung im
    Kindprozess, den der Dekorator intern aufspannt — die Buchung wäre mit
    dessen Ende verloren. Genau so war es zuerst gebaut, und das Konto
    blieb nach einem echten Lauf auf null stehen.

    Gezählt wird damit Wartezeit mit. Deshalb der Deckel, deshalb die
    Kennzeichnung als Schätzung, und deshalb schlägt eine echte
    Kontingent-Meldung diese Rechnung jederzeit.
    """
    def huelle(*a, **kw):
        t0 = time.time()
        try:
            return fn(*a, **kw)
        finally:
            gebraucht = min(time.time() - t0, GPU_DECKEL_S)
            _konto["daten"] = kontingent_buchen(_konto_laden(), gebraucht,
                                                time.time())
            _konto["schmutzig"] = True
    huelle.__name__ = fn.__name__
    return huelle


def _diar_dauer(audio_path, num_speakers):
    try:
        laenge = loader.get_duration(audio_path)
    except Exception:
        laenge = 1800
    return int(min(120, max(30, 20 + laenge / 40)))


@bucht_gpu
@spaces.GPU(duration=_diar_dauer)
def _diarize_gpu(audio_path, num_speakers):
    kwargs = {}
    if num_speakers and int(num_speakers) > 0:
        kwargs["num_speakers"] = int(num_speakers)

    # Der Stimmabdruck fällt beim Clustern ohnehin an. Aufgehoben erkennt
    # die App dieselbe Person beim nächsten Mal wieder, statt sie erneut
    # als „Sprecher 2“ zu begrüßen. Nicht jede Fassung gibt ihn heraus,
    # deshalb der Rückfall auf den gewöhnlichen Aufruf.
    einbettungen, output = None, None
    try:
        output, einbettungen = diarizer(audio_path, return_embeddings=True,
                                        **kwargs)
    except (TypeError, ValueError):
        output = diarizer(audio_path, **kwargs)
    dia = getattr(output, "speaker_diarization", output)

    turns = [{"start": round(t.start, 2), "ende": round(t.end, 2), "label": lb}
             for t, _, lb in dia.itertracks(yield_label=True)]
    stats = {lb: {"dauer": round(dia.label_duration(lb), 1),
                  "turns": len(dia.label_timeline(lb))}
             for lb in sorted(dia.labels())}
    overlaps = [{"start": round(seg.start, 2), "ende": round(seg.end, 2),
                 "wer": sorted(lb for lb in dia.labels()
                               if dia.label_timeline(lb).crop(seg))}
                for seg in dia.get_overlap()]

    abdruecke = {}
    if einbettungen is not None:
        try:
            for lb, vektor in zip(sorted(dia.labels()), einbettungen):
                werte = [float(x) for x in vektor]
                if any(werte):
                    abdruecke[lb] = werte
        except (TypeError, ValueError):
            abdruecke = {}
    return turns, stats, overlaps, abdruecke


@bucht_gpu
@spaces.GPU(duration=30)
def _sprache_gpu(audio_path, ab):
    """Erkennt die Sprache einmalig anhand von 30 s ab der ersten Sprechstelle."""
    wellenform, sr = loader.crop(audio_path, Segment(ab, ab + 30.0), mode="pad")
    eingabe = {"array": wellenform.squeeze(0).numpy(), "sampling_rate": sr}
    try:
        res = asr(eingabe, return_timestamps=True, return_language=True)
        for c in res.get("chunks", []):
            if c.get("language"):
                return c["language"]
    except (TypeError, ValueError):
        pass
    return None


def _worte_einpassen(audio_path, text, a, b, gesamt):
    """Zerlegt einen erkannten Abschnitt in einzeln verortete Wörter.

    Text und Audio des Abschnitts entsprechen einander exakt, deshalb ist
    das Alignment hier besonders zuverlässig. Nur so lässt sich der Text
    später wortweise dem richtigen Sprecher zuordnen — satzweise Zuordnung
    verteilt Wortwechsel systematisch falsch.
    """
    worte = text.split()
    if not worte:
        return []
    von, bis = max(0.0, a - 0.2), min(gesamt, b + 0.4)
    treffer = []
    if bis - von > 0.1:
        try:
            treffer = _align_fenster(audio_path, worte, von, bis)
        except Exception:
            treffer = []
    if len(treffer) < len(worte) * 0.6:
        # Notfalls gleichmäßig verteilen, damit kein Text verloren geht
        schritt = (b - a) / max(len(worte), 1)
        # Gleichmäßig verteilt heißt: nicht wirklich verortet. Das ist
        # genau der Fall, in dem man dem Wort nicht trauen sollte.
        return [{"start": a + i * schritt, "ende": a + (i + 1) * schritt,
                 "text": w, "sprache": None, "wert": 0.0}
                for i, w in enumerate(worte)]
    return [{"start": s, "ende": e, "text": worte[nr], "sprache": None,
             "wert": wert} for nr, s, e, wert in treffer]


def _pegel_messen(wav_pfad, turns=None):
    """Misst Sprachpegel, Rauschteppich und Übersteuerung.

    Was nicht im Signal steckt, holt kein Modell zurück. Diese Grenze war
    bisher unsichtbar: der Nutzer sah nur ein schlechtes Ergebnis und
    konnte nicht wissen, dass es an der Aufnahme lag. Liegen bereits
    Sprecherzeiten vor, wird Sprache gegen Nichtsprache gemessen — sonst
    ersatzweise das laute gegen das leise Zehntel.
    """
    import numpy as np
    try:
        welle, sr = loader({"audio": wav_pfad})
        x = welle.squeeze(0).numpy().astype("float32")
    except Exception:
        return None
    if x.size < 1000:
        return None

    fenster = max(1, int(sr * 0.05))
    n = x.size // fenster
    if n < 4:
        return None
    rahmen = x[:n * fenster].reshape(n, fenster)
    energie = np.sqrt((rahmen ** 2).mean(axis=1)) + 1e-9
    db = 20.0 * np.log10(energie)

    if turns:
        ist_sprache = np.zeros(n, dtype=bool)
        for t in turns:
            a = max(0, int(t["start"] * sr / fenster))
            b = min(n, int(t["ende"] * sr / fenster) + 1)
            ist_sprache[a:b] = True
        if ist_sprache.any() and (~ist_sprache).any():
            sprache_db = float(np.percentile(db[ist_sprache], 75))
            rausch_db = float(np.percentile(db[~ist_sprache], 50))
        else:
            turns = None
    if not turns:
        sprache_db = float(np.percentile(db, 90))
        rausch_db = float(np.percentile(db, 10))

    return {"sprache_db": sprache_db, "rausch_db": rausch_db,
            "uebersteuert": float((np.abs(x) > 0.995).mean())}


def _asr_dauer(audio_path, start, ende, sprache=None, kontext=""):
    # Grosszuegiger als die reine Rechenzeit: der Temperatur-Rückfall
    # verwirft entgleiste Abschnitte und dekodiert sie erneut, im
    # schlimmsten Fall fünfmal. Das Budget ist eine Reservierung, keine
    # Abrechnung — zu knapp bemessen bricht der Lauf mittendrin ab.
    return int(min(200, max(60, 40 + (ende - start) / 2)))


@bucht_gpu
@spaces.GPU(duration=_asr_dauer)
def _transkribiere_gpu(audio_path, start, ende, sprache=None, kontext=""):
    """Transkribiert ein Fenster am Stück — voller Kontext, beste Qualität.

    Die Sprache wird fest vorgegeben: sonst rät Whisper sie pro Abschnitt
    neu und kippt dabei gern mitten im Gespräch in eine englische
    Übersetzung.
    """
    modell = _asr_holen(sprache)
    wellenform, sr = loader.crop(audio_path, Segment(start, ende), mode="pad")
    eingabe = {"array": wellenform.squeeze(0).numpy(), "sampling_rate": sr}
    # Bewusst ohne prompt_ids — ein Vorwissen-Prompt bringt die
    # Langform-Dekodierung zum Verstummen.
    gk = _decode_optionen(sprache)
    try:
        res = modell(eingabe, return_timestamps=True, generate_kwargs=gk,
                     return_language=not sprache)
    except (TypeError, ValueError):
        res = modell(eingabe, return_timestamps=True, generate_kwargs=gk)

    sprache_erkannt = None
    worte = []
    for c in res.get("chunks", []):
        ts = c.get("timestamp") or (None, None)
        text = (c.get("text") or "").strip()
        if ts[0] is None or not text or _ist_halluzination(text):
            continue
        sprache_erkannt = sprache_erkannt or c.get("language")
        bis = ts[1] if ts[1] is not None else ts[0] + 30.0
        worte += _worte_einpassen(audio_path, text, float(ts[0]) + start,
                                  float(bis) + start, ende)
    if not worte:
        # Manche Modelle (etwa deutsche Feinabstimmungen) liefern keine
        # Zeitmarken. Dann den gesamten Text des Fensters selbst verorten.
        ganz = (res.get("text") or "").strip()
        alle = [] if _ist_halluzination(ganz) else ganz.split()
        if alle:
            rate = len(alle) / max(ende - start, 1.0)
            treffer, _, _ = _aligniere_bereich(audio_path, alle, start, ende,
                                               rate, schluss=True)
            worte = [{"start": a, "ende": b, "text": alle[nr],
                      "sprache": None, "wert": wert}
                     for nr, a, b, wert in treffer]

    for w in worte:
        w["sprache"] = sprache_erkannt or sprache
    return worte


# ---------- Eigenes Transkript einpassen (Forced Alignment) ----------
# Wird ein fertiges Transkript mitgeliefert (z. B. aus Apples Sprachmemos),
# muss nicht neu erkannt werden: der Text wird direkt auf die Tonspur gelegt.
# Das ist schneller als Whisper und übernimmt dessen bessere Textqualität.

def _align_fenster(audio_path, worte, start, ende):
    """Ordnet Wörter einem Audiofenster zu -> [(index, start_s, ende_s)].

    Es wird bewusst weniger Text übergeben, als das Fenster fassen kann;
    das überschüssige Audio fängt der Stern-Token am Ende auf. Andernfalls
    müsste der Aligner die Wörter zusammenstauchen und die Zeiten wären
    unbrauchbar.
    """
    tokens, idx = [], []
    for i, w in enumerate(worte):
        t = [FA_DICT[c] for c in _romanisieren(w) if c in FA_DICT]
        if t:
            tokens.append(t)
            idx.append(i)
    if not tokens:
        return []

    wellenform, sr = loader.crop(audio_path, Segment(start, ende), mode="pad")
    welle = wellenform.to(FA_GERAET)
    with torch.inference_mode():
        emission, _ = fa_modell(welle)
    # Stern nur am Ende: das Fenster beginnt exakt dort, wo die zuletzt
    # zugeordneten Wörter endeten — ein führender Stern würde den Aligner
    # verleiten, den Anfang zu überspringen.
    ziel = torch.tensor([[t for wort in tokens for t in wort] + [FA_STERN]],
                        dtype=torch.int32, device=FA_GERAET)
    pfad, punkte = AF.forced_align(emission, ziel, blank=0)
    spans = [s for s in AF.merge_tokens(pfad[0], punkte[0]) if s.token != FA_STERN]

    verhaeltnis = welle.size(1) / emission.size(1) / sr
    ergebnis, pos = [], 0
    for wort_nr, tok in zip(idx, tokens):
        teil = spans[pos:pos + len(tok)]
        pos += len(tok)
        if teil:
            # Der Anpassungswert sagt, wie gut das Wort im Audio
            # wiedergefunden wurde. Bisher weggeworfen — damit sah ein
            # wackliges Wort so sicher aus wie ein zweifelsfreies.
            wert = sum(sp.score for sp in teil) / len(teil)
            ergebnis.append((wort_nr, start + teil[0].start * verhaeltnis,
                             start + teil[-1].end * verhaeltnis, float(wert)))
    return ergebnis


def _aligniere_bereich(audio_path, worte, start, ende, rate, schluss):
    """Verteilt eine Wortliste über einen Zeitbereich, Fenster für Fenster.

    `schluss` gibt an, ob das Bereichsende zugleich das Ende des Materials
    ist — nur dann muss der verbleibende Text vollständig untergebracht
    werden.
    """
    treffer_gesamt, zeit, wi = [], start, 0
    while wi < len(worte) and zeit < ende - 0.5:
        fenster_ende = min(ende, zeit + FA_FENSTER)
        if schluss and fenster_ende >= ende - 0.5:
            nehmen = len(worte) - wi          # Rest muss vollständig hinein
        else:
            nehmen = max(3, int((fenster_ende - zeit) * rate * FA_ANTEIL))
        treffer = _align_fenster(audio_path, worte[wi:wi + nehmen],
                                 zeit, fenster_ende)
        if not treffer:
            zeit = fenster_ende
            continue
        for nr, a, b, wert in treffer:
            treffer_gesamt.append((wi + nr, a, b, wert))
        wi += treffer[-1][0] + 1
        zeit = max(treffer[-1][2], zeit + 1.0)
    return treffer_gesamt, zeit, wi


def _fa_dauer(audio_path, worte, start, block_ende, rate, gesamt):
    return int(min(200, max(30, 20 + (block_ende - start) / 10)))


@bucht_gpu
@spaces.GPU(duration=_fa_dauer)
def _align_gpu(audio_path, worte, start, block_ende, rate, gesamt):
    """Passt Wörter fortlaufend in einen Audioblock ein.

    Mehrere Fenster je GPU-Aufruf, damit lange Aufnahmen nicht in sehr
    viele Einzelanfragen zerfallen.
    """
    return _aligniere_bereich(audio_path, worte, start, block_ende, rate,
                              schluss=block_ende >= gesamt - 0.5)


def _align_schritt(s):
    """Passt den nächsten Audioblock des Transkripts ein. -> weiter?"""
    worte, gesamt = s["worte"], s["dauer"]
    if s["fa_wort"] >= len(worte) or s["fa_zeit"] >= gesamt - 0.5:
        return False

    block_ende = min(gesamt, s["fa_zeit"] + FA_BLOCK)
    rate = len(worte) / max(gesamt, 1.0)
    treffer, neue_zeit, verbraucht = _align_gpu(
        s["wav"], worte[s["fa_wort"]:], s["fa_zeit"], block_ende, rate, gesamt)

    for nr, a, b, wert in treffer:
        s["chunks"].append({"start": a, "ende": b, "wert": wert,
                            "text": worte[s["fa_wort"] + nr], "sprache": None})
    s["fa_wort"] += verbraucht
    s["fa_zeit"] = max(neue_zeit, s["fa_zeit"] + 1.0)
    return s["fa_wort"] < len(worte) and s["fa_zeit"] < gesamt - 0.5


# ---------- Sitzungen ----------
# Lange Aufnahmen werden über mehrere HTTP-Anfragen verarbeitet: ein einzelner
# Aufruf würde länger dauern als das ZeroGPU-Token einer Anfrage gültig ist.

SITZUNGEN = {}
SITZUNG_TTL = 3 * 3600


def _dateien_loeschen(s):
    """Entfernt Audio vom Server — Originalupload wie umgewandelte Fassung.

    Gesprächsaufnahmen sind das Sensibelste, was durch dieses System läuft;
    sie dürfen nicht länger liegen bleiben als für die Analyse nötig.
    """
    for schluessel in ("wav", "original"):
        pfad = s.get(schluessel)
        if not pfad:
            continue
        try:
            os.unlink(pfad)
        except OSError:
            pass
        s[schluessel] = None


def _aufraeumen():
    jetzt = time.time()
    for sid in [s for s, v in SITZUNGEN.items()
                if jetzt - v["zeit"] > SITZUNG_TTL]:
        _dateien_loeschen(SITZUNGEN[sid])
        SITZUNGEN.pop(sid, None)


def _aufraeum_schleife():
    """Räumt auch dann auf, wenn keine neue Aufnahme mehr hochgeladen wird."""
    while True:
        time.sleep(600)
        try:
            _aufraeumen()
        except Exception:
            traceback.print_exc()


threading.Thread(target=_aufraeum_schleife, daemon=True).start()


def _sitzung(sid):
    s = SITZUNGEN.get(sid)
    if s:
        s["zeit"] = time.time()
    return s


# ---------- Markdown-Ablage ----------

def _speichern(md_text, zeitpunkt, pfad=None):
    # Mit vorhandenem Pfad wird ersetzt statt danebengelegt: der
    # Hintergrundlauf sichert bereits ein Ergebnis, und nachträglich
    # benannte Sprecher sollen dieselbe Aufnahme aktualisieren, nicht
    # eine zweite Datei erzeugen.
    pfad = pfad if (pfad and pfad_erlaubt(pfad)) \
        else f"aufnahmen/{zeitpunkt:%Y-%m-%d_%H%M%S}.md"
    api.upload_file(path_or_fileobj=md_text.encode("utf-8"), path_in_repo=pfad,
                    repo_id=DATEN_REPO, repo_type="dataset",
                    commit_message=f"Analyse {pfad}")
    return pfad


# ---------- Glossar ----------
# Wiederkehrende Begriffe und Sprechernamen liegen im selben privaten
# Dataset. Sie werden bei jeder Analyse als Vorwissen an die Erkennung
# gegeben, damit Eigennamen und Fachbegriffe über Aufnahmen hinweg besser
# getroffen werden.

_glossar_cache = {"zeit": 0.0, "daten": None}
GLOSSAR_TTL = 60.0

# Häufige groß geschriebene Wörter, die als Vorschlag nichts bringen


STIMMEN_DATEI = "stimmen.json"
_stimmen_cache = {"zeit": 0.0, "daten": None}


def _stimmen_laden(frisch=False):
    """Gespeicherte Stimmabdrücke je benanntem Sprecher."""
    jetzt = time.time()
    if (not frisch and _stimmen_cache["daten"] is not None
            and jetzt - _stimmen_cache["zeit"] < GLOSSAR_TTL):
        return _stimmen_cache["daten"]
    daten = {}
    try:
        pfad = hf_hub_download(DATEN_REPO, STIMMEN_DATEI, repo_type="dataset",
                               token=HF_TOKEN, force_download=frisch)
        with open(pfad, encoding="utf-8") as fh:
            daten = json.load(fh)
    except Exception:
        pass
    _stimmen_cache.update(zeit=jetzt, daten=daten)
    return daten


def _stimmen_sichern(daten):
    api.upload_file(path_or_fileobj=json.dumps(daten).encode("utf-8"),
                    path_in_repo=STIMMEN_DATEI, repo_id=DATEN_REPO,
                    repo_type="dataset",
                    commit_message="Stimmabdrücke aktualisiert")
    _stimmen_cache.update(zeit=time.time(), daten=daten)


def _glossar_laden(frisch=False):
    """Begriffe mit Häufigkeit, absteigend sortiert."""
    jetzt = time.time()
    if (not frisch and _glossar_cache["daten"] is not None
            and jetzt - _glossar_cache["zeit"] < GLOSSAR_TTL):
        return _glossar_cache["daten"]
    zaehler, sprecher, festen = {}, [], {}
    try:
        pfad = hf_hub_download(DATEN_REPO, GLOSSAR_DATEI, repo_type="dataset",
                               token=HF_TOKEN, force_download=frisch)
        with open(pfad, encoding="utf-8") as fh:
            zaehler, sprecher, festen = glossar_lesen(fh.read())
    except Exception:
        pass
    daten = {"zaehler": zaehler, "sprecher": sprecher, "festen": festen,
             "begriffe": _rangfolge(zaehler)}
    _glossar_cache.update(zeit=jetzt, daten=daten)
    return daten


def _glossar_speichern(zaehler, sprecher, festen=None):
    text, geordnet = glossar_schreiben(zaehler, sprecher, festen)
    api.upload_file(path_or_fileobj=text.encode("utf-8"),
                    path_in_repo=GLOSSAR_DATEI, repo_id=DATEN_REPO,
                    repo_type="dataset", commit_message="Glossar aktualisiert")
    behalten = {b: zaehler[b] for b in geordnet}
    _glossar_cache.update(zeit=time.time(),
                          daten={"zaehler": behalten, "sprecher": sprecher,
                                 "festen": festen or {},
                                 "begriffe": geordnet})


def _glossar_fortschreiben(segmente, sprechernamen):
    """Zählt Begriffe des neuen Transkripts mit — ohne Zutun des Nutzers.

    Gibt zusätzlich zurück, was neu hinzugekommen ist: nur so kann die App
    nachvollziehbar machen, dass und was das Glossar gelernt hat.
    """
    try:
        g = _glossar_laden(frisch=True)
        zaehler = dict(g["zaehler"])
        neu = []
        for wort, anzahl in _begriffe_zaehlen(segmente).items():
            if wort not in zaehler:
                neu.append(wort)
            zaehler[wort] = zaehler.get(wort, 0) + anzahl
        sprecher = list(g["sprecher"])
        for name in sprechernamen:
            if name and name not in sprecher:
                sprecher.append(name)
                neu.append(name)
        _glossar_speichern(zaehler, sprecher, g.get("festen"))
        return _glossar_laden(), neu
    except Exception:
        traceback.print_exc()
        return _glossar_laden(), []


def glossar_api(key):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    g = _glossar_laden(frisch=True)
    return {"ok": True, **g}


def glossar_speichern_api(key, begriffe_text, sprecher_text):
    """Manuelles Nachbearbeiten; die Reihenfolge legt den Rang fest."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}

    def zerlegen(t):
        raus = []
        for w in re.split(r"[\n,;]+", t or ""):
            w = re.sub(r"\s*\(\d+\)$", "", w.strip(" -\t")).strip()
            if w and w not in raus:
                raus.append(w)
        return raus[:GLOSSAR_MAX]

    begriffe, sprecher = zerlegen(begriffe_text), zerlegen(sprecher_text)
    vorher = _glossar_laden(frisch=True)
    alt = vorher["zaehler"]
    # Vorhandene Häufigkeiten behalten, neue oben einsortieren
    hoechster = max(alt.values(), default=1)
    zaehler = {}
    for rang, b in enumerate(begriffe):
        zaehler[b] = alt.get(b, max(2, hoechster - rang))
    try:
        _glossar_speichern(zaehler, sprecher, vorher.get("festen"))
    except Exception as e:
        return {"ok": False, "fehler": f"Speichern fehlgeschlagen: {e}"}
    g = _glossar_laden()
    return {"ok": True, "begriffe": g["begriffe"], "sprecher": g["sprecher"],
            "zaehler": g["zaehler"]}


# ---------- API-Endpunkte ----------

def _nach_wav(pfad):
    """Konvertiert nach 16-kHz-Mono-WAV und gleicht die Lautstärke an.

    Aufnahmen mit Abstand zum Mikrofon sind oft sehr leise und schwanken
    stark; die Pegelangleichung verbessert Sprechererkennung und
    Transkription spürbar.
    """
    ziel = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name
    # Ohne dynaudnorm. Die dynamische Normalisierung zieht leise Stellen
    # hoch — in Sprechpausen also den Rauschteppich, und auf angehobenes
    # Rauschen antwortet Whisper mit erfundenem Text. Sie hebelt zugleich
    # no_speech_threshold aus, das Stille an ihrem Pegel erkennt. Bleiben
    # Hochpass gegen Trittschall und loudnorm für einen gleichmäßigen
    # Pegel; Whisper ist auf unbearbeitetem Alltagsmaterial trainiert und
    # braucht mehr nicht.
    filter_kette = "highpass=f=60,loudnorm=I=-18:LRA=11:TP=-2"
    try:
        subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", pfad,
                        "-af", filter_kette, "-ar", "16000", "-ac", "1", ziel],
                       check=True)
    except subprocess.CalledProcessError:
        subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", pfad,
                        "-ar", "16000", "-ac", "1", ziel], check=True)
    return ziel


# ---------- Hintergrundverarbeitung ----------
# Bisher war der Browser der Taktgeber: er rief Schritt für Schritt auf.
# Schläft das Telefon, steht die Verarbeitung. Der Space kann die Schritte
# aber selbst durchlaufen — nachgemessen laufen sie dabei auf dem
# PRO-Kontingent des Space-Besitzers, nicht auf dem anonymen.

AUFTRAEGE = {}
AUFTRAG_TTL = 6 * 3600


def _fehler_deuten(ex):
    """Übersetzt die Meldung in etwas, mit dem man handeln kann.

    „Expired token“ und „quota exceeded“ sehen im Rohtext ähnlich aus,
    verlangen aber Gegenteiliges: einmal ist etwas kaputt, einmal ist
    schlicht das Tagesbudget alle und Warten hilft.
    """
    t = f"{type(ex).__name__}: {ex}".lower()
    if "expired" in t and "token" in t:
        return ("Die GPU-Kennung ist mitten im Lauf abgelaufen. Das ist ein "
                "Fehler in der App, nicht bei dir — bitte melden.")
    if "quota" in t or "exceeded" in t or "runs limit" in t:
        # In der Meldung steht die amtliche Restzeit („30s left“). Die ist
        # mehr wert als die eigene Zählung, denn sie kennt auch, was
        # andere Spaces verbraucht haben.
        treffer = re.search(r"([\d.]+)\s*s(?:ec\w*)?\s*left", t)
        if treffer:
            try:
                _konto["daten"] = kontingent_korrigieren(
                    _konto_laden(), float(treffer.group(1)), time.time(),
                    GPU_BUDGET)
                _konto["schmutzig"] = True
                _konto_sichern()
            except Exception:
                traceback.print_exc()
        return ("Das GPU-Tageskontingent ist aufgebraucht. Es füllt sich 24 "
                "Stunden nach der ersten Nutzung wieder auf.")
    if "gpu task aborted" in t or "aborted" in t:
        return ("Die GPU-Zuteilung wurde abgebrochen — meist Andrang. "
                "Später noch einmal starten.")
    return f"{type(ex).__name__}: {ex}"


def _auftrag_lauf(jid, sid, num_speakers, sprache):
    """Fährt die komplette Kette durch, ohne dass jemand zusieht."""
    j = AUFTRAEGE[jid]
    # Wo die Zeit hingeht, gemessen statt vermutet — sonst optimiert man
    # beim nächsten Mal wieder die falsche Stelle.
    j["zeiten"] = {}
    _uhr = {"t": time.time(), "name": None}

    def takt(name):
        jetzt = time.time()
        if _uhr["name"]:
            j["zeiten"][_uhr["name"]] = round(
                j["zeiten"].get(_uhr["name"], 0.0) + jetzt - _uhr["t"], 1)
        _uhr.update(t=jetzt, name=name)
    try:
        takt("Audio vorbereiten"); j.update(schritt="Audio vorbereiten")
        s = _sitzung(sid)
        if not s:
            raise RuntimeError("Sitzung abgelaufen.")
        try:
            _audio_bereitstellen(s)
        except subprocess.CalledProcessError:
            raise RuntimeError("Audioformat konnte nicht gelesen werden.")

        takt("Sprecher erkennen"); j.update(schritt="Sprecher erkennen")
        r = diarisieren_api(APP_KEY, sid, num_speakers, sprache)
        if not r.get("ok"):
            raise RuntimeError(r.get("fehler", "Sprechererkennung fehlgeschlagen"))
        gesamt = max(1, int(r.get("abschnitte") or 1))
        rest, art = restzeit(time.time() - j["begonnen"], 0, gesamt,
                             (_sitzung(sid) or {}).get("dauer"))
        takt("Text erkennen")
        j.update(stand="laeuft", schritt="Text erkennen", von=0, bis=gesamt,
                 rest_s=rest, rest_art=art)

        i, getan = 0, 0
        while True:
            a = transkribieren_api(APP_KEY, sid, i)
            if not a.get("ok"):
                raise RuntimeError(a.get("fehler", "Transkription fehlgeschlagen"))
            getan += 1
            rest, art = restzeit(time.time() - j["begonnen"], getan, gesamt,
                                 (_sitzung(sid) or {}).get("dauer"))
            j.update(von=min(getan, gesamt), rest_s=rest, rest_art=art)
            if not a.get("weiter"):
                break
            # Beim Einpassen zählt das Backend selbst weiter, bei der
            # Fenster-Erkennung der Index.
            i = a.get("index", i) + 1 if "index" in a else i

        takt("Zusammenstellen"); j.update(schritt="Zusammenstellen")
        e = abschliessen_api(APP_KEY, sid)
        if not e.get("ok"):
            raise RuntimeError(e.get("fehler", "Abschluss fehlgeschlagen"))

        # Sofort sichern. Im alten Ablauf tippte der Nutzer nach dem
        # Ergebnis auf „Speichern“ — im Hintergrund tippt niemand, und
        # ohne das lag das Ergebnis nur im Arbeitsspeicher und war beim
        # Schließen der App verloren. Benannt wird mit dem, was schon
        # bekannt ist; Nachbenennen ersetzt später dieselbe Datei.
        takt("Sichern"); j.update(schritt="Sichern")
        try:
            namen = dict(e.get("vorschlag") or {})
            e["daten"]["tempo"] = tempo(
                _woerter_zaehlen(e["daten"].get("segmente")),
                time.time() - j["begonnen"], e["daten"].get("dauer"))
            md = _markdown(e["daten"], e.get("quelle") or "aufnahme",
                           datetime.fromisoformat(e["zeitpunkt"]), namen)
            e["pfad"] = _speichern(md, datetime.fromisoformat(e["zeitpunkt"]))
            e["markdown"] = md
        except Exception:
            traceback.print_exc()

        takt(None)
        # Die Zahlen gehören ins Ergebnis, nicht nur in den Auftrag: sonst
        # sind sie weg, sobald man die App einmal geschlossen hat.
        e["tempo"] = tempo(_woerter_zaehlen(e.get("daten", {}).get("segmente")),
                           time.time() - j["begonnen"],
                           (e.get("daten") or {}).get("dauer"))
        e["zeiten"] = dict(j["zeiten"])
        e["kontingent"] = kontingent_stand(_konto_laden(), time.time(),
                                           GPU_BUDGET)
        _konto_sichern()
        j.update(stand="fertig", ergebnis=e, zeit=time.time())
        _push_senden("Analyse fertig",
                     f"{len(e.get('daten', {}).get('stats', {}))} Sprecher, "
                     f"{_zeit(e.get('daten', {}).get('dauer', 0))} min")
    except Exception as ex:
        traceback.print_exc()
        j.update(stand="fehler", fehler=_fehler_deuten(ex),
                 roh=f"{type(ex).__name__}: {ex}"[:300],
                 schritt_beim_fehler=j.get("schritt"), zeit=time.time())
        _konto_sichern()
        _push_senden("Analyse fehlgeschlagen", j["fehler"][:120])


def starten_api(key, sid, num_speakers, sprache=None, request: gr.Request = None):
    """Stößt die Verarbeitung an und gibt sofort zurück."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    if not _sitzung(sid):
        return {"ok": False, "fehler": "Sitzung abgelaufen. Bitte erneut starten."}

    # Festhalten, womit gestartet wurde. ZeroGPU rechnet die GPU-Zeit dem
    # zu, dessen Kennung an der Anfrage hängt; ob sie anlag, soll nicht
    # Glaubenssache sein, sondern ablesbar.
    kopf = {}
    try:
        kopf = {k.lower(): v for k, v in dict(request.headers).items()}
    except Exception:
        pass
    mit_token = bool(
        (kopf.get("authorization") or "").lower().startswith("bearer hf_")
        or kopf.get("x-ip-token"))

    jid = secrets.token_urlsafe(12)
    AUFTRAEGE[jid] = {"stand": "laeuft", "schritt": "Sprecher erkennen",
                      "von": 0, "bis": 1, "sid": sid, "zeit": time.time(),
                      "begonnen": time.time(), "mit_token": mit_token}

    # Bewusst OHNE den Kontext dieser Anfrage. Der Zuordnungs-Token hängt
    # an der Anfrage und läuft nach wenigen Minuten ab; mitkopiert hält er
    # für einen kurzen Auftrag, nicht für eine Stunde Audio — dann bricht
    # der Lauf mittendrin mit „expired token“ ab. Ohne ihn läuft die
    # Arbeit unter der Kennung des Space selbst, die nicht verfällt. Das
    # ist nachgemessen: eine Reservierung über 200 s lief so durch, weit
    # über dem anonymen Tagesbudget von 120 s.
    threading.Thread(target=_auftrag_lauf,
                     args=(jid, sid, num_speakers, sprache),
                     daemon=True).start()
    return {"ok": True, "auftrag": jid, "mit_token": mit_token}


def auftrag_api(key, jid):
    """Fragt den Stand ab. Der Browser darf zwischendurch schlafen."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    j = AUFTRAEGE.get(jid)
    if not j:
        return {"ok": False, "fehler": "Auftrag unbekannt oder abgelaufen."}
    antwort = {k: v for k, v in j.items() if k != "ergebnis"}
    if j.get("stand") == "fertig":
        antwort["ergebnis"] = j["ergebnis"]
    return {"ok": True, **antwort}


# ---------- Benachrichtigung ----------
# Web Push funktioniert auf dem iPhone nur für eine zum Home-Bildschirm
# hinzugefügte App. Der Inhalt ist Ende-zu-Ende verschlüsselt: Apples
# Zustelldienst sieht nur Chiffrat, keine Gesprächsdaten.

PUSH_DATEI = "push.json"
_push_cache = {"daten": None}


def _push_zustand():
    if _push_cache["daten"] is None:
        daten = {"vapid_pem": None, "abos": []}
        try:
            pfad = hf_hub_download(DATEN_REPO, PUSH_DATEI, repo_type="dataset",
                                   token=HF_TOKEN, force_download=True)
            with open(pfad, encoding="utf-8") as fh:
                daten = json.load(fh)
        except Exception:
            pass
        if not daten.get("vapid_pem"):
            # Beim ersten Mal selbst erzeugen. Der private Schlüssel liegt
            # im privaten Datensatz — dieselbe Vertrauensgrenze wie die
            # Protokolle, und niemand muss ein Geheimnis von Hand setzen.
            from cryptography.hazmat.primitives.asymmetric import ec
            from cryptography.hazmat.primitives import serialization
            k = ec.generate_private_key(ec.SECP256R1())
            daten["vapid_pem"] = k.private_bytes(
                serialization.Encoding.PEM,
                serialization.PrivateFormat.PKCS8,
                serialization.NoEncryption()).decode()
            _push_sichern(daten)
        _push_cache["daten"] = daten
    return _push_cache["daten"]


def _push_sichern(daten):
    api.upload_file(path_or_fileobj=json.dumps(daten).encode("utf-8"),
                    path_in_repo=PUSH_DATEI, repo_id=DATEN_REPO,
                    repo_type="dataset", commit_message="Push aktualisiert")
    _push_cache["daten"] = daten


def _push_oeffentlich():
    from cryptography.hazmat.primitives import serialization
    k = serialization.load_pem_private_key(
        _push_zustand()["vapid_pem"].encode(), password=None)
    roh = k.public_key().public_bytes(
        serialization.Encoding.X962,
        serialization.PublicFormat.UncompressedPoint)
    import base64
    return base64.urlsafe_b64encode(roh).decode().rstrip("=")


_vapid_pfad = {"wert": None}


def _vapid_datei():
    """Schreibt den privaten Schlüssel einmalig als Datei.

    pywebpush nimmt für `vapid_private_key` einen Dateipfad oder
    base64-kodiertes DER — eine PEM-Zeichenkette weist es mit „Could not
    deserialize key data“ zurück. Der Dateipfad ist die dokumentierte und
    eindeutige Form; die Datei liegt nur im flüchtigen Space-Dateisystem.
    """
    if not _vapid_pfad["wert"] or not os.path.exists(_vapid_pfad["wert"]):
        pfad = os.path.join(tempfile.gettempdir(), "vapid_privat.pem")
        with open(pfad, "w", encoding="utf-8") as fh:
            fh.write(_push_zustand()["vapid_pem"])
        os.chmod(pfad, 0o600)
        _vapid_pfad["wert"] = pfad
    return _vapid_pfad["wert"]


def _dienst(abo):
    """Welcher Zustelldienst — hilft, Apple von anderen zu unterscheiden."""
    try:
        return (abo.get("endpoint", "").split("/")[2]) or "?"
    except Exception:
        return "?"


def _push_bereit():
    """Ist die Versandbibliothek überhaupt da? -> (ja, Grund)

    pywebpush zieht http-ece nach, das eine Erweiterung baut. Scheitert
    das beim Aufbau des Space, wäre Push still tot — der Fehler fiele
    erst auf, wenn eine erwartete Nachricht ausbleibt. Deshalb abfragbar.
    """
    try:
        import pywebpush  # noqa: F401
        return True, "bereit"
    except Exception as e:
        return False, f"{type(e).__name__}: {e}"


def _push_senden(titel, text):
    """Verschickt an alle angemeldeten Geräte. -> (zugestellt, Fehler)"""
    try:
        from pywebpush import webpush, WebPushException
    except Exception as e:
        return 0, [f"Versandbibliothek fehlt: {e}"]
    z = _push_zustand()
    uebrig, geaendert, zugestellt, fehler = [], False, 0, []
    for abo in z.get("abos", []):
        try:
            # Zwei Dinge, die beim ersten echten Versand aufgefallen sind:
            # ohne Zeitgrenze kann der Aufruf hängen, bis die HTTP-Anfrage
            # des Browsers aufgibt — dann sieht der Nutzer nur „An error
            # occurred“ und weiß nichts. Und der sub-Anspruch muss eine
            # erreichbare Adresse sein; erfundene .invalid-Domains weisen
            # Zustelldienste zurück.
            webpush(subscription_info=abo,
                    data=json.dumps({"titel": titel, "text": text}),
                    vapid_private_key=_vapid_datei(),
                    vapid_claims={"sub": FRONTEND},
                    timeout=10)
            uebrig.append(abo)
            zugestellt += 1
        except WebPushException as e:
            # 404/410 heißt: Gerät hat abgemeldet. Eintrag darf weg.
            antwort = getattr(e, "response", None)
            code = getattr(antwort, "status_code", None)
            rumpf = ""
            try:
                rumpf = (antwort.text or "")[:120]
            except Exception:
                pass
            if code in (404, 410):
                geaendert = True
            else:
                uebrig.append(abo)
            fehler.append(f"{_dienst(abo)} {code or '?'}: {rumpf or e}"[:200])
        except Exception as e:
            uebrig.append(abo)
            fehler.append(f"{_dienst(abo)} {type(e).__name__}: {e}"[:200])
    if geaendert:
        z["abos"] = uebrig
        try:
            _push_sichern(z)
        except Exception:
            traceback.print_exc()
    return zugestellt, fehler


def push_schluessel_api(key):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    try:
        return {"ok": True, "schluessel": _push_oeffentlich()}
    except Exception as e:
        return {"ok": False, "fehler": str(e)}


def push_stand_api(key):
    """Sagt ohne Netzverkehr, wie es um Push steht.

    Bewusst getrennt vom Versand: hängt der Versand, liefert diese
    Auskunft trotzdem sofort und man sieht, woran es nicht liegt.
    """
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    bereit, grund = _push_bereit()
    try:
        z = _push_zustand()
        return {"ok": True, "bibliothek": grund,
                "geraete": [_dienst(a) for a in z.get("abos", [])],
                "schluessel": bool(z.get("vapid_pem"))}
    except Exception as e:
        return {"ok": False, "bibliothek": grund, "fehler": str(e)}


def push_pruefen_api(key):
    """Schickt eine Probe-Benachrichtigung und sagt, was dabei geschah."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    bereit, grund = _push_bereit()
    try:
        geraete = len(_push_zustand().get("abos", []))
    except Exception as e:
        return {"ok": False, "fehler": f"Push-Zustand nicht lesbar: {e}"}
    if not bereit:
        return {"ok": False, "bibliothek": grund, "geraete": geraete,
                "fehler": "Versandbibliothek fehlt im Space."}
    if not geraete:
        return {"ok": False, "bibliothek": grund, "geraete": 0,
                "fehler": "Kein Gerät angemeldet."}
    try:
        zugestellt, fehler = _push_senden(
            "Probe", "Wenn du das liest, funktionieren Benachrichtigungen.")
    except Exception as e:
        traceback.print_exc()
        return {"ok": False, "bibliothek": grund, "geraete": geraete,
                "fehler": f"{type(e).__name__}: {e}"}
    return {"ok": zugestellt > 0, "bibliothek": grund, "geraete": geraete,
            "zugestellt": zugestellt, "meldungen": fehler}


def push_anmelden_api(key, abo):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    if not isinstance(abo, dict) or not abo.get("endpoint"):
        return {"ok": False, "fehler": "Ungültige Anmeldung."}
    try:
        z = _push_zustand()
        abos = [a for a in z.get("abos", [])
                if a.get("endpoint") != abo["endpoint"]]
        abos.append(abo)
        z["abos"] = abos[-10:]      # ein paar Geräte reichen
        _push_sichern(z)
    except Exception as e:
        return {"ok": False, "fehler": str(e)}
    return {"ok": True, "geraete": len(z["abos"])}


def kontingent_api(key):
    """Wie viel GPU-Zeit heute noch übrig ist.

    Selbst gezählt, weil es keine Schnittstelle dafür gibt — und deshalb
    als Schätzung gekennzeichnet, bis eine echte Kontingent-Meldung die
    Zahl bestätigt hat.
    """
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    try:
        return {"ok": True, **(kontingent_stand(_konto_laden(), time.time(),
                                                GPU_BUDGET) or {})}
    except Exception as e:
        return {"ok": False, "fehler": str(e)}


def status_api(key, request: gr.Request = None):
    """Meldet, ob die Anfrage mit HF-Token ankommt (GPU-Kontingent-Zuordnung)."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    kopf = {}
    try:
        kopf = dict(request.headers) if request is not None else {}
    except Exception:
        pass
    auth = kopf.get("authorization") or kopf.get("Authorization") or ""
    return {"ok": True, "token": auth.lower().startswith("bearer hf_")}


def vorbereiten_api(key, audio, transkript=None, kontext=None,
                    mit_abgleich=False):
    """Schritt 1: Audio entgegennehmen und in WAV wandeln.

    Wird ein Transkript mitgegeben, wird dieses später eingepasst statt
    selbst zu erkennen.
    """
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    if not audio:
        return {"ok": False, "fehler": "Kein Audio übermittelt."}
    _aufraeumen()
    # Bewusst ohne Umwandlung: ffmpeg über eine Stunde Audio dauert, und
    # solange diese Anfrage läuft, muss die App im Vordergrund bleiben.
    # Die Umwandlung ist der erste Schritt des Hintergrundauftrags.
    worte = (transkript or "").split()
    sid = secrets.token_urlsafe(16)
    SITZUNGEN[sid] = {"wav": None, "original": audio,
                      "quelle": os.path.basename(audio), "zeit": time.time(),
                      "dauer": None, "chunks": [],
                      "worte": worte, "fa_zeit": 0.0, "fa_wort": 0,
                      "mit_abgleich": bool(mit_abgleich),
                      "eigener": (kontext or "").strip(),
                      "kontext": _kontext_bauen((kontext or "").strip(),
                                                _glossar_laden())}
    return {"ok": True, "id": sid,
            "modus": "transkript" if worte else "erkennung",
            "woerter": len(worte)}


def _audio_bereitstellen(s):
    """Wandelt die hochgeladene Datei um. Erster Schritt im Hintergrund."""
    if s.get("wav"):
        return
    s["wav"] = _nach_wav(s["original"])
    s["dauer"] = loader.get_duration(s["wav"])


def diarisieren_api(key, sid, num_speakers, sprache=None):
    """Schritt 2: Sprecher über die gesamte Aufnahme erkennen."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    s = _sitzung(sid)
    if not s:
        return {"ok": False, "fehler": "Sitzung abgelaufen. Bitte erneut starten."}
    try:
        turns, stats, overlaps, abdruecke = _diarize_gpu(s["wav"], num_speakers)
    except Exception as e:
        traceback.print_exc()
        return {"ok": False,
                "fehler": f"Sprechererkennung fehlgeschlagen: {type(e).__name__}: {e}"}
    sprache = (sprache or "").strip() or None
    if not sprache and not s.get("worte"):
        try:
            sprache = _sprache_gpu(s["wav"], turns[0]["start"] if turns else 0.0)
        except Exception:
            sprache = None
    s.update(turns=turns, stats=stats, overlaps=overlaps, chunks=[],
             abdruecke=abdruecke, abgleich_chunks=[], phase="haupt",
             sprache=sprache, fa_zeit=0.0, fa_wort=0, fertig=set(),
             fenster=_fenstergrenzen(turns, s["dauer"]))
    s["qualitaet"] = aufnahme_urteil(_pegel_messen(s["wav"], turns) or {})

    # Bekannte Stimmen wiedererkennen: dieselben Personen sitzen Woche für
    # Woche im selben Meeting.
    s["vorschlag"] = stimmen_zuordnen(abdruecke, _stimmen_laden())

    fa_abschnitte = max(1, math.ceil(s["dauer"] / FA_FENSTER))
    asr_abschnitte = max(1, len(s["fenster"]) - 1)
    if s.get("worte") and s.get("mit_abgleich"):
        # Einpassen und danach zum Abgleich selbst erkennen — der einzige
        # kostenlose Hinweis darauf, wo der Text unsicher ist. Kostet
        # allerdings den doppelten Durchgang, deshalb nur auf Wunsch.
        abschnitte = fa_abschnitte + asr_abschnitte
    elif s.get("worte"):
        abschnitte = fa_abschnitte
    else:
        abschnitte = asr_abschnitte
    return {"ok": True, "sprecher": len(stats), "sprache": sprache,
            "abschnitte": abschnitte, "qualitaet": s["qualitaet"],
            "vorschlag": s["vorschlag"]}


def transkribieren_api(key, sid, index):
    """Schritt 3: nächsten Abschnitt verarbeiten (eigene Anfrage je Abschnitt)."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    s = _sitzung(sid)
    if not s or "fenster" not in s:
        return {"ok": False, "fehler": "Sitzung abgelaufen. Bitte erneut starten."}

    if s.get("worte") and s.get("phase") == "haupt":
        try:
            weiter = _align_schritt(s)
        except Exception as e:
            traceback.print_exc()
            return {"ok": False,
                    "fehler": f"Transkript einpassen fehlgeschlagen: "
                              f"{type(e).__name__}: {e}"}
        if not weiter and not s.get("mit_abgleich"):
            return {"ok": True, "weiter": False, "schritt": "einpassen",
                    "fortschritt": 1.0}
        if not weiter:
            # Eingepasst. Jetzt dieselbe Aufnahme selbst erkennen, damit
            # sich die beiden Ergebnisse gegenseitig prüfen können.
            s["phase"] = "abgleich"
            s["fertig"] = set()
            weiter = True
        return {"ok": True, "weiter": weiter, "schritt": "einpassen",
                "fortschritt": round(min(1.0, s["fa_zeit"] / max(s["dauer"], 1)), 2)}

    i = int(index)
    if not 0 <= i < len(s["fenster"]) - 1:
        return {"ok": False, "fehler": "Ungültiger Abschnitt."}
    if i in s.setdefault("fertig", set()):
        # Bereits verarbeitet — Wiederholung nach Abbruch darf nichts doppeln
        return {"ok": True, "index": i, "abschnitte": len(s["fenster"]) - 1,
                "weiter": i + 1 < len(s["fenster"]) - 1}
    ziel_liste = ("abgleich_chunks" if s.get("phase") == "abgleich"
                  else "chunks")
    try:
        s[ziel_liste] += _transkribiere_gpu(s["wav"], s["fenster"][i],
                                            s["fenster"][i + 1],
                                            s.get("sprache"),
                                            s.get("kontext", ""))
    except Exception as e:
        traceback.print_exc()
        return {"ok": False,
                "fehler": f"Transkription fehlgeschlagen: {type(e).__name__}: {e}"}
    s["fertig"].add(i)
    return {"ok": True, "index": i, "abschnitte": len(s["fenster"]) - 1,
            "weiter": i + 1 < len(s["fenster"]) - 1}


def _daten_sichern(s):
    if "daten" not in s:
        # Zwei unabhängige Erkennungen: wo sie sich widersprechen, ist der
        # Text unsicher. Der Wortlaut der Vorlage bleibt stehen — markiert
        # wird nur, was bestritten ist.
        strittig_idx = ()
        if s.get("abgleich_chunks"):
            haupt = [c["text"] for c in sorted(s["chunks"],
                                               key=lambda c: c["start"])]
            zweit = [c["text"] for c in sorted(s["abgleich_chunks"],
                                               key=lambda c: c["start"])]
            s["strittig"] = abgleich(haupt, zweit)
            strittig_idx = {x["index"] for x in s["strittig"]}
        s["chunks"] = _konfidenz_markieren(
            sorted(s["chunks"], key=lambda c: c["start"]), strittig_idx)
        daten = _zusammenfuegen(s["turns"], s["stats"], s["overlaps"],
                                s["chunks"], s["dauer"], s.get("sprache"))
        # Erst nach der Zuordnung korrigieren: die Zeitstempel stammen aus
        # dem Audio und dürfen von einer Textänderung nicht berührt werden.
        # Ein mitgeliefertes Transkript bleibt unangetastet — es ist die
        # Vorlage, nicht die Vermutung.
        if not s.get("worte"):
            g = _glossar_laden()
            ziele = _korrekturziele(g, s.get("eigener", ""))
            daten["segmente"], daten["korrekturen"] = _korrigieren(
                daten["segmente"], ziele, g.get("festen"))
        else:
            daten["korrekturen"] = []
        daten["qualitaet"] = s.get("qualitaet")
        daten["strittig"] = s.get("strittig", [])
        s["daten"] = daten
    return s["daten"]


def abschliessen_api(key, sid):
    """Schritt 5: Ergebnis ausliefern und das Glossar fortschreiben."""
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    s = _sitzung(sid)
    if not s or "turns" not in s:
        return {"ok": False, "fehler": "Sitzung abgelaufen. Bitte erneut starten."}
    daten = _daten_sichern(s)
    # Ab hier wird das Audio nicht mehr gebraucht — sofort entfernen.
    _dateien_loeschen(s)
    zeitpunkt = datetime.now(ZEITZONE)
    namen, farben = _namen_farben(daten["stats"])
    glossar, neue_begriffe = _glossar_fortschreiben(daten["segmente"], [])
    return {"ok": True, "daten": daten, "namen": namen, "farben": farben,
            "zeitpunkt": zeitpunkt.isoformat(timespec="seconds"),
            "quelle": s["quelle"], "sprache": daten.get("sprache"),
            "glossar": glossar["begriffe"][:GLOSSAR_PROMPT],
            "glossar_gesamt": len(glossar["begriffe"]),
            "neue_begriffe": neue_begriffe,
            "sprecher_bekannt": glossar["sprecher"],
            # Damit die App die Stimmen beim Speichern fortschreiben kann
            # und benannte Sprecher gleich vorausgefüllt sind.
            "abdruecke": s.get("abdruecke") or {},
            "vorschlag": s.get("vorschlag") or {},
            "qualitaet": s.get("qualitaet")}


def analysieren_api(key, audio, num_speakers, sprache=None, transkript=None,
                    kontext=None):
    """Alle Schritte in einem Aufruf — nur für kurze Aufnahmen geeignet."""
    vor = vorbereiten_api(key, audio, transkript, kontext)
    if not vor.get("ok"):
        return vor
    sid = vor["id"]
    dia = diarisieren_api(key, sid, num_speakers, sprache)
    if not dia.get("ok"):
        return dia
    i = 0
    while True:
        tr = transkribieren_api(key, sid, i)
        if not tr.get("ok"):
            return tr
        if not tr.get("weiter"):
            break
        i += 1
    return abschliessen_api(key, sid)


def speichern_api(key, analyse, namen):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    try:
        daten = analyse["daten"]
        zeitpunkt = datetime.fromisoformat(analyse["zeitpunkt"])
        quelle = analyse.get("quelle") or "aufnahme"
        daten["stats"]
    except (KeyError, TypeError, ValueError):
        return {"ok": False, "fehler": "Ungültige Analysedaten."}
    eigene = {k: v.strip() for k, v in (namen or {}).items()
              if isinstance(v, str) and v.strip()}
    md_text = _markdown(daten, quelle, zeitpunkt, eigene)
    try:
        pfad = _speichern(md_text, zeitpunkt, analyse.get("pfad"))
    except Exception as e:
        return {"ok": False, "fehler": f"Speichern fehlgeschlagen: {e}"}

    neue = []
    if eigene:                      # Sprechernamen fürs nächste Mal merken
        _, neue = _glossar_fortschreiben([], list(eigene.values()))
        # Und die Stimme dazu: beim nächsten Mal erkennt die App die
        # Person wieder, statt sie erneut nummerieren zu lassen.
        abdruecke = (analyse.get("abdruecke") or {})
        if abdruecke:
            try:
                _stimmen_sichern(stimmen_auffrischen(
                    _stimmen_laden(frisch=True), abdruecke, eigene))
            except Exception:
                traceback.print_exc()
    return {"ok": True, "pfad": pfad, "markdown": md_text,
            "neue_begriffe": neue}


def lernen_api(key, falsch, richtig):
    """Nimmt eine vom Nutzer bestätigte Korrektur auf.

    Bis hierher lernte das System ausschließlich aus der eigenen Ausgabe —
    ein Kreis ohne Wahrheit von außen, der nur die eigenen Fehler
    verstärken kann. Dies ist der einzige Eingang für echtes Wissen: was
    der Mensch beim Lesen als falsch erkannt und richtiggestellt hat.
    """
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    falsch = (falsch or "").strip()
    richtig = (richtig or "").strip()
    if not falsch or not richtig or falsch.lower() == richtig.lower():
        return {"ok": False, "fehler": "Nichts zu lernen."}
    if len(richtig) > 80 or " " in richtig:
        return {"ok": False, "fehler": "Bitte ein einzelnes Wort."}
    try:
        g = _glossar_laden(frisch=True)
        festen = dict(g.get("festen") or {})
        festen[falsch.lower()] = richtig
        zaehler = dict(g["zaehler"])
        # Das richtige Wort zählt ab jetzt als bestätigter Begriff, damit
        # auch ähnliche Verhörer darauf gezogen werden.
        zaehler[richtig] = max(zaehler.get(richtig, 0), KORREKTUR_MIN_ZAEHLER)
        _glossar_speichern(zaehler, g["sprecher"], festen)
    except Exception as e:
        traceback.print_exc()
        return {"ok": False, "fehler": f"Speichern fehlgeschlagen: {e}"}
    return {"ok": True, "falsch": falsch, "richtig": richtig}


def verlauf_api(key):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    eintraege = []
    dateien = sorted(
        (f for f in api.list_repo_files(DATEN_REPO, repo_type="dataset")
         if f.startswith("aufnahmen/") and f.endswith(".md")), reverse=True)[:20]
    for f in dateien:
        lokal = hf_hub_download(DATEN_REPO, f, repo_type="dataset", token=HF_TOKEN)
        with open(lokal, encoding="utf-8") as fh:
            fm, _ = _frontmatter(fh.read())
        eintraege.append({"pfad": f,
                          "titel": fm.get("titel", f),
                          "datum": str(fm.get("datum", "")),
                          "sprecher": fm.get("sprecher"),
                          "dauer_s": fm.get("dauer_s")})
    return {"ok": True, "eintraege": eintraege}


def eintrag_api(key, pfad):
    if not _pruefe(key):
        return {"ok": False, "fehler": "Ungültiger Zugangsschlüssel."}
    if not pfad_erlaubt(pfad):
        return {"ok": False, "fehler": "Ungültiger Pfad."}
    lokal = hf_hub_download(DATEN_REPO, pfad, repo_type="dataset", token=HF_TOKEN)
    with open(lokal, encoding="utf-8") as fh:
        text = fh.read()
    fm, body = _frontmatter(text)
    return {"ok": True, "frontmatter": fm, "body": body, "markdown": text}


with gr.Blocks(title="Sprecher-Analyse API") as demo:
    gr.Markdown(f"# 🎙️ Sprecher-Analyse — API-Backend\n"
                f"Die App selbst läuft unter **[{FRONTEND}]({FRONTEND})**. "
                f"Dieses Formular dient nur zu Testzwecken.")
    key = gr.Textbox(label="Zugangsschlüssel", type="password")
    gr.Button("Status prüfen").click(status_api, [key], gr.JSON(),
                                     api_name="status")
    with gr.Accordion("Glossar", open=False):
        g_begriffe = gr.Textbox(label="Begriffe (je Zeile einer)", lines=4)
        g_sprecher = gr.Textbox(label="Sprechernamen (je Zeile einer)", lines=3)
        gr.Button("Glossar laden").click(glossar_api, [key], gr.JSON(),
                                         api_name="glossar")
        gr.Button("Glossar speichern").click(
            glossar_speichern_api, [key, g_begriffe, g_sprecher], gr.JSON(),
            api_name="glossar_speichern")
    with gr.Tab("Analyse"):
        audio = gr.Audio(sources=["upload", "microphone"], type="filepath",
                         label="Audio")
        num_speakers = gr.Number(value=0, precision=0,
                                 label="Anzahl Sprecher (0 = automatisch)")
        sprache_feld = gr.Textbox(label="Sprache (leer = automatisch)",
                                  value="german")
        transkript_feld = gr.Textbox(
            label="Eigenes Transkript (leer = selbst erkennen)", lines=4)
        kontext_feld = gr.Textbox(label="Namen & Begriffe (optional)")
        b_analyse = gr.Button("Analysieren", variant="primary")
        out_analyse = gr.JSON()
        b_analyse.click(analysieren_api,
                        [key, audio, num_speakers, sprache_feld, transkript_feld,
                         kontext_feld],
                        out_analyse, api_name="analysieren")
    with gr.Tab("Schrittweise (lange Aufnahmen)"):
        sid = gr.Textbox(label="Sitzungs-ID")
        idx = gr.Number(value=0, precision=0, label="Abschnitt")
        abgleich_feld = gr.Checkbox(
            value=False, label="Zusätzlich selbst erkennen (Abgleich)")
        gr.Button("1 · Vorbereiten").click(
            vorbereiten_api,
            [key, audio, transkript_feld, kontext_feld, abgleich_feld],
            gr.JSON(), api_name="vorbereiten")
        gr.Button("2 · Sprecher erkennen").click(
            diarisieren_api, [key, sid, num_speakers, sprache_feld], gr.JSON(),
            api_name="diarisieren")
        gr.Button("3 · Abschnitt transkribieren").click(
            transkribieren_api, [key, sid, idx], gr.JSON(),
            api_name="transkribieren")
        gr.Button("4 · Abschließen").click(
            abschliessen_api, [key, sid], gr.JSON(), api_name="abschliessen")
        analyse_json = gr.JSON(label="Analyse-Objekt (aus /analysieren)")
        namen_json = gr.JSON(label='Sprechernamen, z. B. {"SPEAKER_00": "Nils"}')
        b_speichern = gr.Button("Als Markdown speichern")
        out_speichern = gr.JSON()
        b_speichern.click(speichern_api, [key, analyse_json, namen_json],
                          out_speichern, api_name="speichern")
        b_starten = gr.Button("Im Hintergrund verarbeiten")
        b_starten.click(starten_api,
                        [key, sid, gr.Number(value=0), gr.Textbox(value="german")],
                        gr.JSON(), api_name="starten")
        gr.Button("Auftragsstand").click(
            auftrag_api, [key, gr.Textbox(label="Auftrag")], gr.JSON(),
            api_name="auftrag")
        gr.Button("Push-Schlüssel").click(
            push_schluessel_api, [key], gr.JSON(), api_name="push_schluessel")
        gr.Button("GPU-Kontingent").click(
            kontingent_api, [key], gr.JSON(), api_name="kontingent")
        gr.Button("Push-Zustand").click(
            push_stand_api, [key], gr.JSON(), api_name="push_stand")
        gr.Button("Push prüfen").click(
            push_pruefen_api, [key], gr.JSON(), api_name="push_pruefen")
        gr.Button("Push anmelden").click(
            push_anmelden_api, [key, gr.JSON(label="Abo")], gr.JSON(),
            api_name="push_anmelden")
        falsch_feld = gr.Textbox(label="Falsch erkannt")
        richtig_feld = gr.Textbox(label="Richtig")
        gr.Button("Korrektur merken").click(
            lernen_api, [key, falsch_feld, richtig_feld], gr.JSON(),
            api_name="lernen")
    with gr.Tab("Verlauf"):
        b_verlauf = gr.Button("Verlauf laden")
        out_verlauf = gr.JSON()
        b_verlauf.click(verlauf_api, [key], out_verlauf, api_name="verlauf")
        pfad = gr.Textbox(label="Pfad (aufnahmen/....md)")
        b_eintrag = gr.Button("Eintrag laden")
        out_eintrag = gr.JSON()
        b_eintrag.click(eintrag_api, [key, pfad], out_eintrag, api_name="eintrag")

demo.launch(ssr_mode=False)