diarization / app.py
Monstermango's picture
Count the GPU call from outside, where the result survives
2e783b7 verified
Raw
History Blame Contribute Delete
64.9 kB
"""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)