Compare commits

..

No commits in common. "main" and "v1.0.0" have entirely different histories.
main ... v1.0.0

30 changed files with 351 additions and 1006 deletions

View file

@ -12,12 +12,6 @@ WHISPER_THREADS=8
WHISPER_MODELL=medium WHISPER_MODELL=medium
WHISPER_COMPUTE=int8 WHISPER_COMPUTE=int8
# --- Branchen-Profil (Kategorien, Prompt, Sicherheitsnetz, Ansage) ---
# Waehlt den Ordner unter profile/ - buendelt alles, was fachlich vom
# Einsatzzweck abhaengt. Mitgeliefert: "praxis" (Hausarztpraxis, Default),
# "aufzug-notdienst" (Beispiel/Vorlage). Siehe README "Kategorien & Branchen-Profil".
PROFIL=praxis
# --- Kategorisierung --- # --- Kategorisierung ---
# Backend: "ollama" (Default, lokal) oder "openrouter" (Cloud, nur zum Testen # Backend: "ollama" (Default, lokal) oder "openrouter" (Cloud, nur zum Testen
# ob ein staerkeres Modell besser extrahiert - Transkript verlaesst dabei den # ob ein staerkeres Modell besser extrahiert - Transkript verlaesst dabei den
@ -51,14 +45,6 @@ NEXTCLOUD_ADRESSBUCH=contacts
# Anrufe bleiben unangetastet, egal wie alt. # Anrufe bleiben unangetastet, egal wie alt.
LOESCHFRIST_TAGE=0 LOESCHFRIST_TAGE=0
# --- Störungswache (pipe.monitor, eigener Prozess, siehe telefon/starten.sh) ---
# Prüft die Dienste-Ampel (Telefon, Verarbeitung, Spracherkennung, Nextcloud)
# und alarmiert lokal per macOS-Benachrichtigung + Ton, wenn niemand auf den
# Leitstand schaut.
MONITOR_TAKT_S=60
MONITOR_WIEDERHOLUNG_MIN=15
MONITOR_TON=Sosumi
# --- Telefonie (SIP-Zugang der Praxis, leer lassen = keine Telefonannahme) --- # --- Telefonie (SIP-Zugang der Praxis, leer lassen = keine Telefonannahme) ---
SIP_USER= SIP_USER=
SIP_DOMAIN=sip.plusnet.de SIP_DOMAIN=sip.plusnet.de
@ -99,13 +85,6 @@ LEITSTAND_HOST=127.0.0.1
LEITSTAND_PORT=8088 LEITSTAND_PORT=8088
LEITSTAND_USER=praxis LEITSTAND_USER=praxis
LEITSTAND_PASS= LEITSTAND_PASS=
# Weitere Konten zusaetzlich zum obigen - kommagetrennt "nutzer:passwort",
# z.B. fuer ein zweites Team/einen zweiten Einsatzzweck neben "praxis".
LEITSTAND_ZUGAENGE=
# Optional: geteiltes Geheimnis fuer signierte, 1h gueltige Testzugangslinks
# (z.B. per Chatbot-Kommando ausgegeben). Leer = Feature aus.
TESTZUGANG_SECRET=
# --- Telefonanlage (nur Stufe 2) --- # --- Telefonanlage (nur Stufe 2) ---
# Event-Socket-Port von FreeSWITCH. Muss zu listen-port in # Event-Socket-Port von FreeSWITCH. Muss zu listen-port in

View file

@ -1,44 +1,5 @@
# Changelog # Changelog
## v1.1.0 — 2026-08-30
Branchen-Profile machen das Projekt für andere Einsatzzwecke als die
Hausarztpraxis konfigurierbar (z. B. 24h-Notdienst), dazu mehrere
Leitstand-Härtungen und Aufräumarbeiten.
### Branchen-Profile (neu)
- Projekt ist jetzt per `PROFIL`-Wahl für andere Einsatzzwecke konfigurierbar
(Praxis, Aufzug-Notdienst, …) statt hart auf die Arztpraxis zugeschnitten
- aktives Branchen-Profil im Leitstand/Dashboard sichtbar
- README-Einleitung an den Profil-Umbau angepasst
### Leitstand & Zugriff
- mehrere gleichzeitige Leitstand-Zugänge (`LEITSTAND_ZUGAENGE`), z. B. ein
eigenes Notdienst-Konto neben dem Praxis-Konto
- Abmelden-Button im Leitstand, zusätzlich per `?logoff` in der URL auslösbar
- eingeloggter Nutzer im Header sichtbar; Verlauf zeigt jetzt, wer eine Karte
verschoben hat
- Modell-Dropdown listet lokale Ollama-Modelle live statt fest im Code
- Dringlichkeitsfarben getauscht und zentralisiert; `/api/anrufe` und
`/api/verlauf` werden jetzt gecacht (Performance)
### Testzugang & Sicherheit
- signierte, 1h gültige Testzugangslinks für den Leitstand ("agent-one
test"-Kommando)
- Sicher-Modus bei Testzugang erzwungen, nicht abschaltbar
- Rufnummern bei Testzugang serverseitig maskiert statt nur per CSS geblurrt
(vorher im Rohdaten-Response sichtbar)
### Zuverlässigkeit
- Störungswache (`pipe.monitor`) + Verschlüsselung-at-rest dokumentiert
- Log-Rotation für Rohlogs + Aufräumen alter Verlauf-Archivkopien
- Linux-Setup-Hinweis in der README ergänzt
### Sonstiges
- echte Aufnahme statt Sprachsynthese als Ansage möglich (`ANSAGE_TTS=datei`)
- Nextcloud-Profil, LinkedIn und Telefonnummer (internationales Format) im
Footer
## v1.0.0 — 2026-08-25 ## v1.0.0 — 2026-08-25
Erste stabile Version. Läuft produktiv (mit Testdaten) auf macOS/Apple Erste stabile Version. Läuft produktiv (mit Testdaten) auf macOS/Apple

126
README.md
View file

@ -1,31 +1,30 @@
# Logpy:AgentOne # Logpy:AgentOne
**v1.1.0** — siehe [CHANGELOG.md](CHANGELOG.md) **v1.0.0** — siehe [CHANGELOG.md](CHANGELOG.md)
Anruf-Intake-Agent — nimmt Anrufe (Sprachnachrichten) entgegen, **transkribiert** Praxis-Telefon-Agent — nimmt Anrufe (Sprachnachrichten) einer Hausarztpraxis entgegen, **transkribiert**
sie lokal, **kategorisiert** das Anliegen und **legt** das Ergebnis strukturiert sie lokal, **kategorisiert** das Anliegen und **legt** das Ergebnis strukturiert
ab (lokal und optional in Nextcloud). Für welchen Einsatzbereich (Hausarztpraxis, ab (lokal und optional in Nextcloud).
24h-Notdienst, …) ist reine Konfiguration — siehe
[Branchen-Profil](#kategorien--branchen-profil). Zwei Beispiele liegen bei:
`praxis` (Hausarztpraxis) und `aufzug-notdienst`.
**Vollständig on-prem** — keine Anrufdaten verlassen den Rechner. **Vollständig on-prem** — keine Patientendaten verlassen den Rechner.
![Der Leitstand im Dunkelmodus: Dienste-Ampeln und Verlauf oben, darunter das Board mit Eingang, In Bearbeitung, Rückfragen und Erledigt](docs/agentone-screen-1.jpg)
> **Hinweis:** Alle Namen, Rufnummern und Anliegen in diesem Repository sind > **Hinweis:** Alle Namen, Rufnummern und Anliegen in diesem Repository sind
> erfunden — auch im Screenshot und in der Beispielkonfiguration. „Praxis > erfunden — auch im Screenshot und in der Beispielkonfiguration. „Praxis
> Musterhausen" ist ein Platzhalter. Echte Zugangsdaten und Anrufdaten liegen > Musterhausen" ist ein Platzhalter. Echte Zugangsdaten und Anrufdaten liegen
> außerhalb des Repositorys (`.env`, `ablage/`, `telefon/`). > außerhalb des Repositorys (`.env`, `ablage/`, `telefon/`).
Wer das Projekt auf einem weiteren Rechner einrichtet: [docs/MITGABE.md](docs/MITGABE.md)
sammelt die Erfahrungswerte, die in dieser Anleitung nicht stehen.
## Stufen ## Stufen
1. **Die Pipe** (fertig): Audio → Transkript → Kategorie → Ablage. 1. **Die Pipe** (fertig): Audio → Transkript → Kategorie → Ablage.
2. **Telefonie** (steht): FreeSWITCH nimmt über den Plusnet-Trunk an, spielt die 2. **Telefonie** (steht): FreeSWITCH nimmt über den Plusnet-Trunk an, spielt die
Ansage, nimmt auf und legt die Datei im Eingang der Pipe ab. Ansage, nimmt auf und legt die Datei im Eingang der Pipe ab.
3. **Feinschliff** (teilweise): Löschfristen ✅, Notfall-Alarm ✅, 3. **Feinschliff** (teilweise): Löschfristen ✅, Notfall-Alarm ✅ — offen:
Verschlüsselung at rest ✅ (Betreiber-Pflicht: FileVault, siehe Monitoring/Alerting bei Diensteausfall, Verschlüsselung at rest.
[Datenschutz](#datenschutz-dsgvo)), Monitoring/Alerting bei Diensteausfall ✅
(`pipe/monitor.py`, siehe unten).
## Bausteine ## Bausteine
@ -38,62 +37,19 @@ ab (lokal und optional in Nextcloud). Für welchen Einsatzbereich (Hausarztpraxi
| `pipe/config.py` | Konfiguration aus `.env` | | `pipe/config.py` | Konfiguration aus `.env` |
| `pipe/watch.py` | Beobachtet den Eingang und schiebt Aufnahmen durch die Pipe | | `pipe/watch.py` | Beobachtet den Eingang und schiebt Aufnahmen durch die Pipe |
| `pipe/server.py` | Leitstand: Live-Ansicht mit Status, Anrufen und Verlauf | | `pipe/server.py` | Leitstand: Live-Ansicht mit Status, Anrufen und Verlauf |
| `pipe/dienste.py` | Zustand der beteiligten Dienste (Telefon, Verarbeitung, Spracherkennung, Nextcloud) — geteilt von Leitstand und Störungswache |
| `pipe/monitor.py` | Störungswache: alarmiert lokal (macOS-Benachrichtigung + Ton), wenn ein Dienst ausfällt |
| `pipe/alarm.py` | Verschickt den lokalen macOS-Alarm (osascript) |
| `pipe/protokoll.py` | Ereignisprotokoll (JSONL) für den Leitstand | | `pipe/protokoll.py` | Ereignisprotokoll (JSONL) für den Leitstand |
| `pipe/audit.py` | Unveränderliches Audit-Log (append-only, siehe unten) | | `pipe/audit.py` | Unveränderliches Audit-Log (append-only, siehe unten) |
| `pipe/loeschen.py` | Löscht erledigte Anrufe nach Ablauf der Aufbewahrungsfrist | | `pipe/loeschen.py` | Löscht erledigte Anrufe nach Ablauf der Aufbewahrungsfrist |
| `pipe/bewertung.py` | 👍/👎-Bewertung pro Anruf (Leitstand) | | `pipe/bewertung.py` | 👍/👎-Bewertung pro Anruf (Leitstand) |
| `pipe/export.py` | Exportiert bewertete Anrufe als JSONL-Trainingsdaten | | `pipe/export.py` | Exportiert bewertete Anrufe als JSONL-Trainingsdaten |
| `pipe/kategorien.py` | Lädt Kategorien-Schema des aktiven Branchen-Profils (siehe unten) | | `prompts/categorize_de.txt` | Kategorisierungs-Prompt (deutsch, mehrsprachig-tauglich) |
| `profile/<name>/` | Branchen-Profil: Prompt, Kategorien, Sicherheitsnetz, Ansage | | `telefon/` | Ansage, FreeSWITCH-Vorlagen, Start- und Einrichtungsskripte |
| `telefon/` | FreeSWITCH-Vorlagen, Start- und Einrichtungsskripte |
## Kategorien & Branchen-Profil ## Kategorien
Alles, was fachlich vom Einsatzzweck abhängt, steckt in einem **Branchen-Profil** `termin · rezept · ueberweisung · befund · verwaltung · beschwerden · notfall ·
unter `profile/<name>/` — vier Dateien, die zueinander passen müssen: rueckruf · sonstiges`
Dringlichkeit: `niedrig · normal · hoch · notfall`
| Datei | Inhalt |
|---|---|
| `prompt.txt` | System-Prompt für `categorize.py`, inkl. Kategorie-Beschreibungen |
| `kategorien.json` | `kategorien` (Enum fürs JSON-Schema), `kategorie_labels` (Anzeige in Leitstand/Deck/Dashboard), `sicherheitsnetz_kategorie` (Eskalationsziel) |
| `sicherheitsnetz.txt` | Ein Regex-Muster pro Zeile — deterministische Eskalation, unabhängig vom LLM |
| `ansage.txt` | Text für `telefon/ansage_bauen.sh` |
Welches Profil aktiv ist, bestimmt `PROFIL` in der `.env` (Default `praxis`) —
ein Wechsel braucht einen Neustart von `pipe.watch`/`pipe.server`, ist aber sonst
nichts weiter als diese eine Variable. Mitgeliefert:
- **`profile/praxis/`** (Default) — Hausarztpraxis:
`termin · rezept · ueberweisung · befund · verwaltung · beschwerden · notfall ·
rueckruf · sonstiges`
- **`profile/aufzug-notdienst/`** (Beispiel/Vorlage) — 24-Stunden-Aufzug-Notdienst:
`eingeschlossen · stoerung · wartung · rueckruf · sonstiges`
> **Wichtig bei `aufzug-notdienst`:** nur eine Vorlage, um den Mechanismus zu
> zeigen — nicht fachlich/rechtlich geprüft für echten Betrieb. Und generell
> gilt für jedes Profil mit akuter Lebensgefahr am Telefon (Person im Aufzug
> eingeschlossen, medizinischer Notfall, …): Diese Pipe nimmt nur auf und
> kategorisiert **nachträglich** — es gibt keine Live-Weiterleitung während
> des Anrufs. Das Sicherheitsnetz stuft eine Aufnahme nach der Verarbeitung
> als Notfall ein, greift aber nicht ein, während der Anrufer noch in der
> Leitung ist. Für Szenarien, in denen eine sofortige Reaktion während des
> Anrufs überlebenswichtig ist, reicht eine reine Anrufbeantworter-Architektur
> allein nicht — das ist nicht Teil dieses Projekts.
Ein drittes Profil anlegen: `profile/praxis/` kopieren, alle vier Dateien
inhaltlich anpassen, `PROFIL=<name>` setzen. Die Dringlichkeitsstufen
(`niedrig · normal · hoch · notfall`) sind **nicht** Teil des Profils — die
Eskalationsskala ist einsatzzweck-unabhängig und in Leitstand/Dashboard/Deck
fest verdrahtet (Farben, Sortierung, Notfall-Alarm).
> **Hinweis:** `prompt.txt` beschreibt die Kategorien in Prosa fürs Modell,
> `kategorien.json` erzwingt sie als Enum im JSON-Schema — bewusst zwei
> Dateien, aber sie müssen inhaltlich zusammenpassen. Ändert man die
> Kategorien-Liste, muss der Prompt-Text mitgezogen werden, sonst kann das
> Modell eine im Prompt beschriebene Kategorie wählen, die das Schema ablehnt.
## Einrichtung ## Einrichtung
@ -180,7 +136,7 @@ eingehende Anrufe an, spielt die Ansage und nimmt auf. Die Aufnahme landet unter
sie durch die Pipe und räumt sie nach `telefon/verarbeitet/` weg. sie durch die Pipe und räumt sie nach `telefon/verarbeitet/` weg.
```bash ```bash
./telefon/ansage_bauen.sh # Ansage aus profile/$PROFIL/ansage.txt erzeugen ./telefon/ansage_bauen.sh # Ansage aus telefon/ansage.txt erzeugen
./telefon/freeswitch/einrichten.sh # Trunk + Dialplan in FreeSWITCH eintragen ./telefon/freeswitch/einrichten.sh # Trunk + Dialplan in FreeSWITCH eintragen
./telefon/starten.sh # startet FreeSWITCH, Verarbeitung, Leitstand ./telefon/starten.sh # startet FreeSWITCH, Verarbeitung, Leitstand
``` ```
@ -236,8 +192,7 @@ Der SIP-Zugang steht in der `.env` (`SIP_USER`, `SIP_DOMAIN`, `SIP_PASS`). Das
Einrichtungsskript trägt das Passwort in die FreeSWITCH-Konfiguration ein und Einrichtungsskript trägt das Passwort in die FreeSWITCH-Konfiguration ein und
setzt deren Rechte auf 600 — im Git liegen nur Vorlagen mit Platzhaltern. setzt deren Rechte auf 600 — im Git liegen nur Vorlagen mit Platzhaltern.
Text ändern: `profile/$PROFIL/ansage.txt` bearbeiten (siehe Text ändern: `telefon/ansage.txt` bearbeiten, `ansage_bauen.sh` erneut
[Branchen-Profil](#kategorien--branchen-profil)), `ansage_bauen.sh` erneut
ausführen. Für den Praxisbetrieb muss die Ansage den Notruf 112 nennen und auf ausführen. Für den Praxisbetrieb muss die Ansage den Notruf 112 nennen und auf
die Aufzeichnung hinweisen, bevor der Signalton kommt — beides ist bei einer die Aufzeichnung hinweisen, bevor der Signalton kommt — beides ist bei einer
Arztpraxis Pflicht, siehe „Offener Punkt" unter Datenschutz weiter unten. Arztpraxis Pflicht, siehe „Offener Punkt" unter Datenschutz weiter unten.
@ -292,29 +247,6 @@ in `telefon/audit.log` — anders als `verlauf.log` (fürs Leitstand-UI, wird au
zusätzlich per macOS `chflags uappnd` gegen nachträgliches Ändern/Kürzen zusätzlich per macOS `chflags uappnd` gegen nachträgliches Ändern/Kürzen
abgesichert (auch der eigene Prozess kommt danach nur noch per Anhängen rein). abgesichert (auch der eigene Prozess kommt danach nur noch per Anhängen rein).
## Störungswache (Monitoring/Alerting)
Die Dienste-Ampel im Leitstand (Telefon, Verarbeitung, Spracherkennung,
Nextcloud) zeigt den Zustand nur an, wenn gerade jemand die Seite offen hat.
`pipe/monitor.py` läuft deshalb als **eigener Prozess** (von
`telefon/starten.sh` mitgestartet, Log `telefon/monitor.log`), fragt dieselbe
Ampel (`pipe/dienste.py`) alle `MONITOR_TAKT_S` Sekunden (Default 60) ab und
alarmiert bei einer echten Störung (`schlecht`/`aus` — nicht bei `unklar`,
z. B. Nextcloud unkonfiguriert) lokal per macOS-Systembenachrichtigung + Ton
(`pipe/alarm.py`, `osascript`). Solange die Störung anhält, wird alle
`MONITOR_WIEDERHOLUNG_MIN` Minuten (Default 15) erneut erinnert; bei
Wiederherstellung einmalig „wieder ok".
Bewusst ein eigener, dritter Prozess statt in `pipe.watch` oder `pipe.server`
eingebaut: fällt einer der beiden aus, kann die Wache das trotzdem noch
melden. Wirkt nur lokal auf diesem Mac — für Alarme aufs Handy (z. B. per
Push/E-Mail) `pipe/alarm.py` erweitern.
```bash
python3 -m pipe.monitor # dauerhaft beobachten (normalerweise per starten.sh)
python3 -m pipe.monitor --einmal # einmal prüfen, für Tests
```
## Bewertung & Trainingsdaten-Export ## Bewertung & Trainingsdaten-Export
Jede Karte im Leitstand hat 👍/👎 — war die Kategorisierung richtig? Bei 👎 Jede Karte im Leitstand hat 👍/👎 — war die Kategorisierung richtig? Bei 👎
@ -339,21 +271,7 @@ Gesundheitsdaten — Pflichten für den Produktivbetrieb (Stufe 2):
- Löschfristen — siehe oben (`LOESCHFRIST_TAGE`). - Löschfristen — siehe oben (`LOESCHFRIST_TAGE`).
- Verarbeitung lokal (Whisper + Qwen) — kein Cloud-Dienst im Datenpfad. - Verarbeitung lokal (Whisper + Qwen) — kein Cloud-Dienst im Datenpfad.
> **Offener Punkt:** Die aktuelle `profile/praxis/ansage.txt` erfüllt die > **Offener Punkt:** Die aktuelle `telefon/ansage.txt` erfüllt die ersten
> ersten beiden Punkte noch nicht (kein Notfall-Hinweis, keine > beiden Punkte noch nicht (kein Notfall-Hinweis, keine Aufnahme-Einwilligung)
> Aufnahme-Einwilligung) — Text anpassen und `ansage_bauen.sh` neu laufen > — Text anpassen und `ansage_bauen.sh` neu laufen lassen, bevor die Anlage im
> lassen, bevor die Anlage im echten Praxisbetrieb ans Netz geht. > echten Praxisbetrieb ans Netz geht.
> **Verschlüsselung at rest — Pflicht des Betreibers:** Die Pipe legt
> Aufnahmen, Transkripte und Metadaten (`ablage/`, `telefon/`, `training/`)
> unverschlüsselt auf der Platte ab und verlässt sich auf
> Festplattenverschlüsselung durch das Betriebssystem. **FileVault muss auf
> jedem Rechner aktiviert sein, der diese Pipe betreibt**
> (Systemeinstellungen → Datenschutz & Sicherheit → FileVault, oder
> `sudo fdesetup enable`) — den Recovery-Key sicher verwahren (Passwort-
> Tresor), da er das einzige Mittel bei vergessenem Login-Passwort ist.
> Ist FileVault auf einem Zielsystem nicht aktivierbar oder nicht zulässig
> (z. B. Mehrbenutzer-Rechner, andere Auflagen), reicht das nicht aus — dann
> braucht die App eine eigene Verschlüsselung der Ablage, die es aktuell
> **nicht** gibt. Das Aktivieren/Prüfen von FileVault ist Aufgabe des
> Betreibers vor Ort, nicht Teil dieses Repos.

129
docs/MITGABE.md Normal file
View file

@ -0,0 +1,129 @@
# Mitgabe für die Einrichtung auf einem zweiten Rechner
Diese Datei ergänzt README (Einrichtung) und HANDOFF (Stand). Sie enthält das,
was auf der anderen Seite nicht im Repository ankommt: die Erfahrungswerte aus
dem ersten Volltest, die Stellen, an denen die Vorlagen unvollständig sind, und
die Reihenfolge, in der man prüft, ob es wirklich läuft.
## 1. Was im Repo *nicht* liegt
| Fehlt | Woher |
|---|---|
| `.env` | aus `.env.example` kopieren, Werte eintragen (Abschnitt 3) |
| Whisper-Modell (~1,6 GB) | `~/whisper-models/ggml-large-v3-turbo.bin`, Download siehe README |
| Ollama-Modell (~5,2 GB) | `ollama pull qwen3:8b` |
| Piper-Stimme | `~/piper-voices/…`, nur für die Ansage (Stufe 2) nötig |
| `ablage/`, `telefon/eingang/`, `dashboard.html` | Anruf- und Patientendaten, bleiben absichtlich lokal |
| FreeSWITCH-Konfiguration mit echtem SIP-Passwort | `telefon/freeswitch/einrichten.sh` schreibt sie aus der `.env` |
Alle Namen, Rufnummern und Beispieldaten im Repo sind erfunden. „Praxis
Musterhausen" ist Platzhalter — auch im Screenshot.
## 2. Der kurze Weg: Stufe 1 ohne Telefonie
Telefonie (FreeSWITCH, SIP-Trunk) ist für die Erprobung nicht nötig und macht
die meiste Arbeit. Für einen ersten Durchlauf reicht:
```bash
brew install ffmpeg whisper-cpp
brew install ollama && ollama serve &
ollama pull qwen3:8b
mkdir -p ~/whisper-models && cd ~/whisper-models
curl -LO https://huggingface.co/ggerganov/whisper.cpp/resolve/main/ggml-large-v3-turbo.bin
cd ~/praxis-telefon-agent && cp .env.example .env
python3 -m pipe.process_call test/rezept.wav --nummer 021031234567
```
Läuft das durch (~25 s), liegt das Ergebnis unter
`ablage/JJJJ-MM-TT/HH-MM-SS_021031234567/`. Nextcloud, Deck, Leitstand und
Telefonie sind alle abschaltbar und kommen erst danach dran.
## 3. Was die `.env.example` verschweigt
Die Vorlage deckt nicht alle Schalter ab, die `pipe/config.py` liest. Die
Voreinstellungen greifen still — mit diesen Folgen:
| Variable | Voreinstellung | Warum das wichtig ist |
|---|---|---|
| `STT_BACKEND` | `whispercpp` | `WHISPER_MODELL`/`WHISPER_COMPUTE` aus der Vorlage gelten **nur** für `faster`. In der Voreinstellung werden sie ignoriert. |
| `WHISPERCPP_MODELL` | `~/whisper-models/ggml-large-v3-turbo.bin` | Liegt das Modell woanders, bricht die Erkennung ab. |
| `WHISPERCPP_BIN` | `whisper-cli` | Muss im `PATH` liegen (Homebrew `whisper-cpp`). |
| `WHISPER_THREADS` | `8` | Auf kleineren Maschinen herunterdrehen. |
| `NEXTCLOUD_DECK` | *aus* | **Häufigste Verwirrung:** `DECK_TEILEN` und `DECK_STAPEL` stehen in der Vorlage, der Hauptschalter nicht. Ohne `NEXTCLOUD_DECK=1` entsteht kein Board. |
| `NEXTCLOUD_USERID` | = `NEXTCLOUD_USER` | Bei E-Mail-Login weicht die interne User-ID ab, dann zeigt der WebDAV-Pfad ins Leere. |
| `FREESWITCH_LOG` | `/opt/homebrew/var/log/freeswitch/…` | Apple-Silicon-Homebrew. Auf Intel-Macs `/usr/local/...`. Sonst bleiben die Telefon-Ereignisse im Leitstand leer. |
| `ABLAGE_LOKAL`, `TELEFON_ORDNER` | im Projektordner | Für eine Ablage auf einem anderen Volume hier umbiegen. |
| `WHISPER_PROMPT` | Praxis-Kontext | Steuert, wie zuverlässig Namen und Rufnummern erkannt werden. Bei anderem Einsatzzweck anpassen. |
`starten.sh` nutzt außerdem `FS_ETC` (Voreinstellung `/opt/homebrew/etc/freeswitch`)
und `FS_PORT` (`8099`, weil 8021 auf dem Ursprungsrechner belegt war). Auf einer
frischen Maschine ist 8021 meist frei — dann entweder `FS_PORT=8021` setzen oder
den Port in `autoload_configs/event_socket.conf.xml` auf 8099 lassen. Beides
muss zusammenpassen, sonst antwortet `fs_cli` nicht.
## 4. Prüfleiter — in dieser Reihenfolge
1. **Erkennung:** `python3 -m pipe.process_call test/termin.wav --nummer 0…`
`meta.json` enthält ein plausibles `transkript`.
2. **Kategorisierung:** `test/rezept.wav``kategorie: rezept`,
`test/notfall.wav``dringlichkeit: notfall`. Bleibt `kategorie` leer,
läuft Ollama nicht oder das Modell fehlt.
3. **Name und Rückrufnummer** stehen als eigene Felder in `meta.json` — nicht
nur im Fließtext von `anliegen_kurz` (siehe Falle 1 unten).
4. **Nextcloud:** Zugangsdaten eintragen, erneut einen Anruf verarbeiten, dann
`python3 -m pipe.resync` — meldet 0 Fehler.
5. **Deck:** `NEXTCLOUD_DECK=1` **und** `DECK_TEILEN` setzen. Ohne Freigabe legt
die Pipe Karten an, die Praxis sieht ein leeres Board.
6. **Leitstand:** `python3 -m pipe.server`, dann <http://127.0.0.1:8088>.
7. **Telefonie** zuletzt: `./telefon/starten.sh` prüft selbst vor und bricht mit
Klartext ab.
## 5. Fallen, die uns beim ersten Volltest getroffen haben
1. **Name und Rückrufnummer gingen still verloren.** Beide Felder fehlten in der
`required`-Liste des JSON-Schemas; das Modell ließ sie einfach weg, obwohl es
sie kannte. Behoben — aber beim Anpassen des Schemas in `pipe/categorize.py`
sofort wieder einbaubar. Immer gegen `test/notfall.wav` gegenprüfen.
2. **WebDAV ohne Wiederholung.** Ein einzelner träger Antwortlauf schickt den
Anruf in den manuellen Nachsync. Jetzt 3 Versuche mit Backoff, Timeout 60 s.
3. **Ordnername aus der geratenen statt der CLIP-Nummer.** Die Nummer kommt aus
der Telefonanlage, nicht aus dem Transkript.
4. **Stumme ffmpeg-/whisper-Fehler.** Jetzt mit Klartext — bei eigenen
Änderungen nicht wieder auf nackte `CalledProcessError` zurückfallen.
5. **Telephone.app streitet um dieselbe SIP-Registrierung.** Vor dem Start
beenden; `starten.sh` prüft das.
6. **Rotierende Provider-Adressen.** Löst der SIP-Hostname auf mehrere IPs auf,
bricht die Registrierung dauernd ab. `SIP_PROXY` auf einen Host setzen, der
auf genau eine IP zeigt.
7. **Leerzeichen im Projektpfad brechen die Aufnahme.** FreeSWITCHs `record`
zerlegt seine Argumente an Leerzeichen (`record <pfad> <zeitlimit>
<stille-schwelle> <stille-treffer>`). Liegt das Projekt z. B. unter
`~/Desktop/Server AI/…`, landet die Aufnahme als `…/Server.PCMA` neben dem
Projekt, der Eingang bleibt leer und es sieht aus wie ein Watcher-Fehler —
die Ansage läuft ja. `einrichten.sh` bricht deshalb bei Leerzeichen ab und
nennt den Symlink-Ausweg. Getroffen auf der zweiten Maschine, 23.08.2026.
8. **FreeSWITCH liest aus dem Cellar**, wenn `-conf` fehlt — dann überschreibt
jedes Homebrew-Upgrade Trunk und Dialplan.
## 6. Was bewusst noch fehlt (vor Produktivbetrieb klären)
- **Notfall löst keinen Alarm aus.** `dringlichkeit: notfall` bekommt ein rotes
Label und steht im Dashboard oben — kein Ton, keine Push, keine Mail. Eine
Pipe, die Notfälle stumm ablegt, ist für eine Praxis gefährlicher als gar
keine. Das ist der nächste Schritt, vor der Telefonie.
- **Keine Löschfristen**, keine Verschlüsselung at rest.
- **Leitstand im Netz** (`LEITSTAND_HOST=0.0.0.0`) verlangt `LEITSTAND_PASS`
die Seiten zeigen Transkripte und Aufnahmen von Patienten.
- **Ansage** nennt 112 und weist auf die Aufzeichnung hin. Beides ist bei einer
Arztpraxis Pflicht und darf beim Umtexten nicht wegfallen.
- **Mehrsprachigkeit:** `WHISPER_SPRACHE=de` ist fest. Für EN/AR auf `auto`.
## 7. Nützliche Rückmeldung
Wenn auf der anderen Seite etwas abweicht, sind vor allem diese Punkte für uns
hier interessant:
- andere Hardware/OS-Version und wie lange ein Anruf dann braucht (hier ~24 s),
- ob `qwen3:8b` bei 16 GB RAM neben laufender Telefonie noch passt,
- jede Stelle, an der README oder `.env.example` nicht gereicht haben — die
gehört dann dort hinein, nicht in diese Datei.

BIN
docs/agentone-screen-1.jpg Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 413 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 187 KiB

View file

@ -1,27 +0,0 @@
"""Lokaler macOS-Alarm (Systembenachrichtigung + Ton) für pipe.monitor.
Wirft nie - eine fehlgeschlagene Benachrichtigung darf die Wache nicht zum
Absturz bringen. Nutzt osascript (stdlib-Unterprozess, kein Framework, keine
Abhängigkeit von einem laufenden Nextcloud/Cloud-Dienst).
"""
from __future__ import annotations
import subprocess
from . import config
def _quote(text: str) -> str:
"""AppleScript-String-Literal - Anführungszeichen/Backslashes escapen."""
return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"'
def sende(titel: str, text: str) -> None:
skript = (
f"display notification {_quote(text)} with title {_quote(titel)} "
f"sound name {_quote(config.MONITOR_TON)}"
)
try:
subprocess.run(["osascript", "-e", skript], capture_output=True, timeout=10)
except Exception as fehler:
print(f" Alarm konnte nicht zugestellt werden: {fehler}")

View file

@ -15,33 +15,40 @@ import urllib.error
import urllib.request import urllib.request
from functools import lru_cache from functools import lru_cache
from . import config, kategorien, modellwahl from . import config, modellwahl
KATEGORIEN = kategorien.KATEGORIEN KATEGORIEN = [
DRINGLICHKEITEN = kategorien.DRINGLICHKEITEN "termin", "rezept", "ueberweisung", "befund",
"verwaltung", "beschwerden", "notfall", "rueckruf", "sonstiges",
]
DRINGLICHKEITEN = ["niedrig", "normal", "hoch", "notfall"]
# Deterministisches Sicherheitsnetz, unabhaengig vom LLM: dieselben Symptome,
@lru_cache(maxsize=1) # die der Prompt selbst als Notfall-Beispiele nennt (Atemnot, Brustschmerz,
def _sicherheitsnetz_muster() -> list[re.Pattern]: # Bewusstlosigkeit, starke Blutung). Grund: getestet mit qwen3:8b, 3/3
"""Lädt die Eskalations-Regex-Muster aus profile/<PROFIL>/sicherheitsnetz.txt. # identische Laeufe stuften "starke Brustschmerzen ... kriege kaum Luft" nur
# als "hoch" statt "notfall" ein - obwohl das Modell selbst "Brustschmerzen"
Ein Muster pro Zeile, Kommentare ("#") und Leerzeilen werden übersprungen. # und "Atemnot" als stichworte erkannte. Bei Notfall-Erkennung darf man sich
Deterministisch, unabhängig vom LLM - siehe Rationale in der Profil-Datei # nicht allein auf ein LLM verlassen. Eskaliert nur nach oben, nie nach unten
selbst. Eskaliert nur nach oben, nie nach unten. # - ein Fehlalarm kostet einen Blick, ein uebersehener Notfall kann einen
""" # Patienten kosten (siehe auch die "im Zweifel hoeher"-Regel im Prompt).
if not config.SICHERHEITSNETZ_DATEI.is_file(): _NOTFALL_MUSTER = [re.compile(p, re.IGNORECASE) for p in (
return [] r"atemnot",
return [ r"(kriege|bekomme|krieg)\w*\s+(kaum|keine|schwer)\s+luft",
re.compile(zeile, re.IGNORECASE) r"kann\s+(kaum|nicht)\s+(mehr\s+)?atmen",
for zeile in (z.strip() for z in config.SICHERHEITSNETZ_DATEI.read_text(encoding="utf-8").splitlines()) r"brust\w*schmerz",
if zeile and not zeile.startswith("#") r"schmerzen?\s+in\s+der\s+brust",
] r"bewusstlos",
r"ohnm[aä]chtig",
r"nicht\s+ansprechbar",
r"blutet\s+(stark|heftig|sehr)",
r"starke?\s+blutung",
)]
def _notfall_sicherheitsnetz(transkript: str, d: dict) -> dict: def _notfall_sicherheitsnetz(transkript: str, d: dict) -> dict:
ziel = kategorien.SICHERHEITSNETZ_KATEGORIE if d.get("dringlichkeit") != "notfall" and any(m.search(transkript) for m in _NOTFALL_MUSTER):
if d.get("kategorie") != ziel and any(m.search(transkript) for m in _sicherheitsnetz_muster()): d["kategorie"] = "notfall"
d["kategorie"] = ziel
d["dringlichkeit"] = "notfall" d["dringlichkeit"] = "notfall"
return d return d
@ -237,11 +244,11 @@ def _bereinige(d: dict) -> dict:
d["kategorie"] = "sonstiges" d["kategorie"] = "sonstiges"
if d.get("dringlichkeit") not in DRINGLICHKEITEN: if d.get("dringlichkeit") not in DRINGLICHKEITEN:
d["dringlichkeit"] = "normal" d["dringlichkeit"] = "normal"
# Das Modell vergibt gelegentlich die Sicherheitsnetz-Kategorie (z. B. # Das Modell vergibt gelegentlich die Kategorie "notfall" und stuft die
# "notfall") und stuft die Dringlichkeit trotzdem nur auf "hoch" ein. Das # Dringlichkeit trotzdem nur auf "hoch" ein. Das ist widersprüchlich und
# ist widersprüchlich und führt zu einer harmloseren Farbe auf dem Board, # führt zu einer harmloseren Farbe auf dem Board, als der Anruf verdient.
# als der Anruf verdient. Im Zweifel eskaliert man, statt abzuschwächen. # In der Praxis eskaliert man im Zweifel, statt abzuschwächen.
if d["kategorie"] == kategorien.SICHERHEITSNETZ_KATEGORIE: if d["kategorie"] == "notfall":
d["dringlichkeit"] = "notfall" d["dringlichkeit"] = "notfall"
d.setdefault("anrufer_name", None) d.setdefault("anrufer_name", None)
d.setdefault("rueckrufnummer", None) d.setdefault("rueckrufnummer", None)

View file

@ -83,21 +83,9 @@ OPENROUTER_KEY = os.environ.get("OPENROUTER_KEY", "")
# hinterlegten (siehe pipe/modellwahl.py) - leer lassen wenn nicht gebraucht. # hinterlegten (siehe pipe/modellwahl.py) - leer lassen wenn nicht gebraucht.
OPENROUTER_MODELL = os.environ.get("OPENROUTER_MODELL", "") OPENROUTER_MODELL = os.environ.get("OPENROUTER_MODELL", "")
# --- Branchen-Profil ----------------------------------------------------- PROMPT_DATEI = WURZEL / "prompts" / os.environ.get(
# Buendelt alles, was fachlich vom Einsatzzweck abhaengt (Kategorien, Prompt, "PROMPT_DATEI", "categorize_de.txt"
# Notfall-Sicherheitsnetz, Ansage) an einer Stelle - siehe profile/<name>/. )
# So bleiben diese vier Teile beim Wechsel des Einsatzzwecks (z.B. Praxis vs.
# 24h-Aufzug-Notdienst) automatisch zueinander konsistent, statt dass man eine
# Datei beim Umstellen vergisst. Neues Profil anlegen: profile/<name>/ Ordner
# mit prompt.txt, kategorien.json, sicherheitsnetz.txt, ansage.txt anlegen
# (profile/praxis/ als Vorlage nehmen) und hier PROFIL=<name> setzen.
PROFIL = os.environ.get("PROFIL", "praxis").strip() or "praxis"
PROFIL_ORDNER = WURZEL / "profile" / PROFIL
PROMPT_DATEI = PROFIL_ORDNER / "prompt.txt"
KATEGORIEN_DATEI = PROFIL_ORDNER / "kategorien.json"
SICHERHEITSNETZ_DATEI = PROFIL_ORDNER / "sicherheitsnetz.txt"
ANSAGE_DATEI = PROFIL_ORDNER / "ansage.txt"
# --- Ablage ------------------------------------------------------------------ # --- Ablage ------------------------------------------------------------------
ABLAGE_LOKAL = Path(os.environ.get("ABLAGE_LOKAL", str(WURZEL / "ablage"))) ABLAGE_LOKAL = Path(os.environ.get("ABLAGE_LOKAL", str(WURZEL / "ablage")))
@ -117,28 +105,6 @@ LEITSTAND_PORT = int(os.environ.get("LEITSTAND_PORT", "8088"))
# auf den Seiten stehen Transkripte und Aufnahmen von Patienten. # auf den Seiten stehen Transkripte und Aufnahmen von Patienten.
LEITSTAND_USER = os.environ.get("LEITSTAND_USER", "praxis") LEITSTAND_USER = os.environ.get("LEITSTAND_USER", "praxis")
LEITSTAND_PASS = os.environ.get("LEITSTAND_PASS", "") LEITSTAND_PASS = os.environ.get("LEITSTAND_PASS", "")
# Weitere Konten zusaetzlich zum obigen (z.B. ein eigenes Konto fuer den
# Notdienst neben dem Praxis-Konto) - kommagetrennt "nutzer:passwort",
# z.B. "notdienst:Berg,spaeter:nochwas".
LEITSTAND_ZUGAENGE = os.environ.get("LEITSTAND_ZUGAENGE", "")
def leitstand_konten() -> dict[str, str]:
"""Alle gueltigen Leitstand-Logins als {nutzername: passwort}."""
konten = {}
if LEITSTAND_PASS:
konten[LEITSTAND_USER] = LEITSTAND_PASS
for eintrag in LEITSTAND_ZUGAENGE.split(","):
nutzer, _, passwort = eintrag.strip().partition(":")
if nutzer and passwort:
konten[nutzer] = passwort
return konten
# Geteiltes Geheimnis mit dilles-agent.php ("agent-one test"-Kommando auf
# dillenberg.net) fuer signierte, 1h gueltige Testzugangslinks - siehe
# pipe.testzugang. Leer lassen deaktiviert das Feature (kein Fallback-Wert,
# ein leeres Secret darf niemals gueltige Signaturen erzeugen).
TESTZUGANG_SECRET = os.environ.get("TESTZUGANG_SECRET", "")
def leitstand_oeffentlich() -> bool: def leitstand_oeffentlich() -> bool:
@ -196,19 +162,6 @@ def nextcloud_aktiv() -> bool:
return bool(NEXTCLOUD_URL and NEXTCLOUD_USER and NEXTCLOUD_PASS) return bool(NEXTCLOUD_URL and NEXTCLOUD_USER and NEXTCLOUD_PASS)
# --- Störungswache (pipe.monitor) --------------------------------------------
# Prueft die Dienste-Ampel (pipe.dienste: Telefon, Verarbeitung, Spracherkennung,
# Nextcloud) periodisch und alarmiert lokal per macOS-Benachrichtigung + Ton,
# wenn niemand auf den Leitstand schaut. Eigener Prozess (siehe
# telefon/starten.sh) - faellt pipe.watch oder pipe.server aus, meldet die
# Wache das trotzdem noch.
MONITOR_TAKT_S = int(os.environ.get("MONITOR_TAKT_S", "60") or "60")
# Erinnerungsabstand, waehrend eine Stoerung anhaelt - sonst nur ein einziger
# Alarm beim Ausfall, der leicht untergeht.
MONITOR_WIEDERHOLUNG_MIN = int(os.environ.get("MONITOR_WIEDERHOLUNG_MIN", "15") or "15")
# macOS-Systemton (siehe /System/Library/Sounds), der bei jedem Alarm abgespielt wird.
MONITOR_TON = os.environ.get("MONITOR_TON", "Sosumi").strip() or "Sosumi"
# --- Aufbewahrungsfrist (pipe.loeschen) -------------------------------------- # --- Aufbewahrungsfrist (pipe.loeschen) --------------------------------------
# Nach so vielen Tagen wird ein Anruf-Ordner (Aufnahme + Transkript + Meta) # Nach so vielen Tagen wird ein Anruf-Ordner (Aufnahme + Transkript + Meta)
# geloescht - lokal und, falls hochgeladen, auch bei Nextcloud. 0 = aus (nichts # geloescht - lokal und, falls hochgeladen, auch bei Nextcloud. 0 = aus (nichts

View file

@ -11,11 +11,15 @@ import json
import sys import sys
from pathlib import Path from pathlib import Path
from . import config, kategorien from . import config
RANG = kategorien.DRINGLICHKEIT_RANG RANG = {"notfall": 0, "hoch": 1, "normal": 2, "niedrig": 3}
FARBE = {k: f"#{v}" for k, v in kategorien.DRINGLICHKEIT_FARBE.items()} FARBE = {"notfall": "#E9322D", "hoch": "#E0A339", "normal": "#C9A227", "niedrig": "#31CC7C"}
KAT_LABEL = kategorien.KATEGORIE_LABEL KAT_LABEL = {
"termin": "Termin", "rezept": "Rezept", "ueberweisung": "Überweisung",
"befund": "Befund", "verwaltung": "Verwaltung", "beschwerden": "Beschwerden", "notfall": "Notfall",
"rueckruf": "Rückruf", "sonstiges": "Sonstiges",
}
def _anrufe() -> list[dict]: def _anrufe() -> list[dict]:
@ -61,7 +65,7 @@ def baue(ausgabe: Path) -> Path:
karten = "\n".join(_karte(d) for d in anrufe) karten = "\n".join(_karte(d) for d in anrufe)
doc = f"""<!doctype html><html lang="de"><head><meta charset="utf-8"> doc = f"""<!doctype html><html lang="de"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"> <meta name="viewport" content="width=device-width,initial-scale=1">
<title>Anruf-Übersicht · {html.escape(kategorien.PROFIL_NAME)}</title> <title>Anruf-Übersicht Praxis Musterhausen</title>
<style> <style>
:root{{--bg:#f6f7f9;--fg:#1a2230;--karte:#fff;--rand:#e4e8ee;--muted:#6b7688}} :root{{--bg:#f6f7f9;--fg:#1a2230;--karte:#fff;--rand:#e4e8ee;--muted:#6b7688}}
@media(prefers-color-scheme:dark){{:root{{--bg:#12151b;--fg:#e7ecf3;--karte:#1b2029;--rand:#2a313d;--muted:#93a0b4}}}} @media(prefers-color-scheme:dark){{:root{{--bg:#12151b;--fg:#e7ecf3;--karte:#1b2029;--rand:#2a313d;--muted:#93a0b4}}}}
@ -82,7 +86,7 @@ padding:1px 7px;font-size:12px;color:var(--muted);margin:2px 4px 2px 0}}
details{{margin-top:6px}}summary{{cursor:pointer;color:var(--muted);font-size:13px}} details{{margin-top:6px}}summary{{cursor:pointer;color:var(--muted);font-size:13px}}
.tr{{color:var(--muted);font-size:13px;margin:6px 0 0}} .tr{{color:var(--muted);font-size:13px;margin:6px 0 0}}
</style></head><body><div class="wrap"> </style></head><body><div class="wrap">
<h1>Anruf-Übersicht · {html.escape(kategorien.PROFIL_NAME)}</h1> <h1>Anruf-Übersicht</h1>
<p class="sub">{len(anrufe)} Anrufe · {zusammenfassung or ''} · lokal erzeugt aus der Ablage</p> <p class="sub">{len(anrufe)} Anrufe · {zusammenfassung or ''} · lokal erzeugt aus der Ablage</p>
{karten} {karten}
</div></body></html>""" </div></body></html>"""

View file

@ -18,15 +18,24 @@ import urllib.error
import urllib.request import urllib.request
from functools import lru_cache from functools import lru_cache
from . import config, kategorien from . import config
_API = "/index.php/apps/deck/api/v1.1" _API = "/index.php/apps/deck/api/v1.1"
# Farben je Dringlichkeit - siehe pipe/kategorien.py. # Farben (6-stelliges Hex ohne #) je Dringlichkeit.
DRINGLICHKEIT_FARBE = kategorien.DRINGLICHKEIT_FARBE DRINGLICHKEIT_FARBE = {
"niedrig": "31CC7C", # grün
"normal": "F1DB50", # gelb
"hoch": "E0A339", # orange
"notfall": "E9322D", # rot
}
# Lesbare Namen der Kategorien für den Kartentitel - der Stapel sagt jetzt, # Lesbare Namen der Kategorien für den Kartentitel - der Stapel sagt jetzt,
# wie weit die Bearbeitung ist, nicht mehr worum es geht. # wie weit die Bearbeitung ist, nicht mehr worum es geht.
KATEGORIE_LABEL = kategorien.KATEGORIE_LABEL KATEGORIE_LABEL = {
"termin": "Termin", "rezept": "Rezept", "ueberweisung": "Überweisung",
"befund": "Befund", "verwaltung": "Verwaltung", "beschwerden": "Beschwerden", "notfall": "Notfall",
"rueckruf": "Rückruf", "sonstiges": "Sonstiges",
}
@lru_cache(maxsize=1) @lru_cache(maxsize=1)

View file

@ -1,96 +0,0 @@
"""Zustand der beteiligten Dienste (Telefon, Verarbeitung, Spracherkennung,
Nextcloud) geteilt zwischen Leitstand (pipe.server) und Störungswache
(pipe.monitor), damit beide dieselbe Sicht auf "läuft/steht" haben.
"""
from __future__ import annotations
import subprocess
import time
import urllib.error
import urllib.request
from datetime import datetime
from . import config, store
# Statusabfragen sind teuer (Netz, Unterprozesse) - kurz zwischenspeichern,
# damit haeufiges Nachfragen den Rechner nicht belastet.
_CACHE: dict[str, tuple[float, dict]] = {}
CACHE_S = 4.0
def _laeuft(muster: str) -> bool:
try:
return subprocess.run(["pgrep", "-f", muster], capture_output=True).returncode == 0
except Exception:
return False
def _http_ok(url: str, timeout: float = 5.0) -> bool:
"""Erreichbarkeitstest. Mit eigenem User-Agent - ein WAF vor Nextcloud
blockt den Standardnamen von urllib und meldete fälschlich 'nicht
erreichbar'."""
req = urllib.request.Request(url, headers={"User-Agent": "praxis-telefon-agent/1.0"})
try:
with urllib.request.urlopen(req, timeout=timeout, context=store._ssl_kontext()):
return True
except urllib.error.HTTPError:
return True # antwortet - reicht als Lebenszeichen
except Exception:
return False
def _trunk() -> tuple[str, str]:
"""(zustand, text) des SIP-Trunks."""
if not _laeuft("[f]reeswitch"):
return "aus", "FreeSWITCH läuft nicht"
try:
roh = subprocess.run(
["fs_cli", "-P", config.FS_PORT, "-x", "sofia status gateway plusnet"],
capture_output=True, text=True, timeout=8).stdout
for zeile in roh.splitlines():
if zeile.startswith("Status"):
wert = zeile.split()[-1]
return ("gut", "registriert") if wert == "UP" else ("schlecht", f"Trunk {wert}")
return "unklar", "Gateway unbekannt"
except Exception:
return "unklar", "fs_cli nicht erreichbar"
def status() -> dict:
jetzt = time.time()
gepuffert = _CACHE.get("status")
if gepuffert and jetzt - gepuffert[0] < CACHE_S:
return gepuffert[1]
trunk_zustand, trunk_text = _trunk()
watcher = _laeuft("[p]ipe.watch")
ollama = _http_ok(f"{config.OLLAMA_URL}/api/version")
nextcloud = (_http_ok(f"{config.NEXTCLOUD_URL}/status.php")
if config.nextcloud_aktiv() else None)
offen = 0
if config.TELEFON_EINGANG.is_dir():
offen = sum(1 for d in config.TELEFON_EINGANG.iterdir()
if d.is_file() and not d.name.startswith("."))
fehler = 0
if config.TELEFON_FEHLER.is_dir():
fehler = sum(1 for d in config.TELEFON_FEHLER.iterdir() if d.is_file())
daten = {
"dienste": [
{"name": "Telefon", "zustand": trunk_zustand, "text": trunk_text},
{"name": "Verarbeitung", "zustand": "gut" if watcher else "aus",
"text": "beobachtet Eingang" if watcher else "Watcher läuft nicht"},
{"name": "Spracherkennung", "zustand": "gut" if ollama else "schlecht",
"text": "Ollama bereit" if ollama else "Ollama nicht erreichbar"},
{"name": "Nextcloud", "zustand": "unklar" if nextcloud is None
else ("gut" if nextcloud else "schlecht"),
"text": "nicht konfiguriert" if nextcloud is None
else ("erreichbar" if nextcloud else "nicht erreichbar")},
],
"eingang_offen": offen,
"fehler": fehler,
"stand": datetime.now().strftime("%H:%M:%S"),
}
_CACHE["status"] = (jetzt, daten)
return daten

View file

@ -1,48 +0,0 @@
"""Kategorien-Schema des aktiven Branchen-Profils (profile/<PROFIL>/kategorien.json).
Gebündelt an einer Stelle, damit categorize.py (JSON-Schema, Eskalationsziel
des Sicherheitsnetzes) und die Anzeige (dashboard.py, deck.py: Kategorie-
Label) dieselbe Quelle nutzen - ein neues Profil ändert nur die JSON-Datei,
kein Python-Code.
Dringlichkeitsstufen sind bewusst NICHT Teil des Profils: die vierstufige
Eskalationsskala (niedrig/normal/hoch/notfall) ist einsatzzweck-unabhängig.
Farbe und Sortierrang je Stufe stehen deshalb ebenfalls hier - eine Quelle
für dashboard.py, deck.py und den Leitstand (server.py), statt an drei
Stellen von Hand synchron zu halten.
"""
from __future__ import annotations
import json
from functools import lru_cache
from . import config
DRINGLICHKEITEN = ["niedrig", "normal", "hoch", "notfall"]
# Farben (6-stelliges Hex ohne "#" - so will es Nextcloud Deck; dashboard.py
# und server.py stellen sich das "#" selbst voran).
DRINGLICHKEIT_FARBE = {
"niedrig": "8A93A3", # grau - unwichtig
"normal": "31CC7C", # grün
"hoch": "E0A339", # orange
"notfall": "E9322D", # rot
}
# Sortierrang (0 = dringlichst zuerst) - aus DRINGLICHKEITEN abgeleitet statt
# separat gepflegt.
DRINGLICHKEIT_RANG = {d: i for i, d in enumerate(reversed(DRINGLICHKEITEN))}
@lru_cache(maxsize=1)
def _daten() -> dict:
return json.loads(config.KATEGORIEN_DATEI.read_text(encoding="utf-8"))
KATEGORIEN = _daten()["kategorien"]
KATEGORIE_LABEL = _daten().get("kategorie_labels", {})
# Anzeigename des aktiven Profils (Leitstand-Header, Dashboard) - Fallback auf
# den rohen Ordnernamen, falls ein Profil das Feld nicht setzt.
PROFIL_NAME = _daten().get("profil_name", config.PROFIL)
# Kategorie, auf die das deterministische Sicherheitsnetz eskaliert (siehe
# categorize._notfall_sicherheitsnetz) - bei "praxis" ist das "notfall", bei
# anderen Profilen ggf. ein anderer Name (z. B. "eingeschlossen").
SICHERHEITSNETZ_KATEGORIE = _daten().get("sicherheitsnetz_kategorie", "notfall")

View file

@ -5,104 +5,49 @@ Der Watcher (pipe.watch, Langzeit-Prozess) und der Leitstand (pipe.server,
eigener Prozess) laufen getrennt - die Auswahl wird deshalb in einer kleinen eigener Prozess) laufen getrennt - die Auswahl wird deshalb in einer kleinen
JSON-Datei geteilt, die categorize.py bei jedem Anruf frisch liest (kein JSON-Datei geteilt, die categorize.py bei jedem Anruf frisch liest (kein
Caching), damit eine Umschaltung sofort beim naechsten Anruf greift. Caching), damit eine Umschaltung sofort beim naechsten Anruf greift.
Die lokalen Ollama-Modelle werden live von Ollama abgefragt (GET /api/tags)
statt fest im Code zu stehen - jedes per "ollama pull" installierte Modell
taucht automatisch im Dropdown auf, ohne Code-Aenderung. Eine feste
OpenRouter-Testauswahl bleibt daneben bestehen: die Modell-IDs dort lassen
sich nicht lokal verifizieren, ein Tippfehler wuerde sonst erst beim naechsten
Anruf sichtbar - fuer freie Eingabe siehe OPENROUTER_MODELL in der .env.
""" """
from __future__ import annotations from __future__ import annotations
import json import json
import time
import urllib.error
import urllib.request
from . import config from . import config
DATEI = config.TELEFON / "modell.json" DATEI = config.TELEFON / "modell.json"
# Ollama-Abfrage ist ein Netzwerkaufruf - kurz zwischenspeichern, damit ein # Feste, kuratierte Auswahl statt freier Eingabe - ein Tippfehler in einer
# offener Leitstand (Polling alle paar Sekunden) Ollama nicht dauerbelastet. # Modell-ID im Leitstand darf nicht den naechsten Anruf scheitern lassen.
_CACHE: dict[str, tuple[float, list[dict]]] = {} # "openrouter"-Eintraege sind zum Testen gedacht (Transkript verlaesst dabei
CACHE_S = 30.0 # den Rechner) - nicht fuer den Praxisbetrieb.
AUSWAHL = [
{"id": "ollama", "label": f"{config.OLLAMA_MODELL} (lokal)",
def _ollama_modelle() -> list[str]: "backend": "ollama", "modell": config.OLLAMA_MODELL},
"""Namen der lokal installierten, textfähigen Ollama-Modelle (leer bei Fehler). {"id": "or-ultra", "label": "Nemotron Ultra 550B (OpenRouter, Test)",
"backend": "openrouter", "modell": "nvidia/nemotron-3-ultra-550b-a55b:free"},
Reine Embedding-Modelle (z. B. nomic-embed-text) werden ausgefiltert - sie {"id": "or-lightning", "label": "Nemotron 3.5 Lightning (OpenRouter, Test)",
können nicht kategorisieren, ein Aufruf würde nur mit einem Fehler enden. "backend": "openrouter", "modell": "nvidia/nemotron-3.5-lightning:free"},
Ältere Ollama-Versionen liefern kein "capabilities"-Feld - dann bleibt das ]
Modell sicherheitshalber drin, statt es zu Unrecht zu verstecken. # Optionaler vierter Eintrag zum schnellen Ausprobieren eines beliebigen
""" # OpenRouter-Modells, ohne Code zu aendern - einfach OPENROUTER_MODELL in
req = urllib.request.Request(f"{config.OLLAMA_URL}/api/tags", # .env setzen.
headers={"User-Agent": "praxis-telefon-agent/1.0"}) if config.OPENROUTER_MODELL:
try: AUSWAHL.append({"id": "or-custom", "label": f"{config.OPENROUTER_MODELL} (OpenRouter, Test)",
with urllib.request.urlopen(req, timeout=5) as antwort: "backend": "openrouter", "modell": config.OPENROUTER_MODELL})
daten = json.loads(antwort.read().decode("utf-8")) _STANDARD = AUSWAHL[0]["id"]
except Exception: _NACH_ID = {e["id"]: e for e in AUSWAHL}
return []
namen = []
for m in daten.get("models", []):
name = m.get("name")
faehigkeiten = m.get("capabilities")
if name and (faehigkeiten is None or "completion" in faehigkeiten):
namen.append(name)
return sorted(namen)
def auswahl() -> list[dict]:
"""Aktuelle Dropdown-Optionen: lokale Ollama-Modelle live + feste OpenRouter-Testeintraege."""
jetzt = time.time()
gepuffert = _CACHE.get("auswahl")
if gepuffert and jetzt - gepuffert[0] < CACHE_S:
return gepuffert[1]
# Ist Ollama gerade nicht erreichbar, bleibt wenigstens die konfigurierte
# Standardauswahl im Dropdown - ein leeres Dropdown waere schlechter.
lokale = _ollama_modelle() or [config.OLLAMA_MODELL]
liste = [{"id": f"ollama:{name}", "label": f"{name} (lokal)",
"backend": "ollama", "modell": name} for name in lokale]
liste.append({"id": "or-ultra", "label": "Nemotron Ultra 550B (OpenRouter, Test)",
"backend": "openrouter", "modell": "nvidia/nemotron-3-ultra-550b-a55b:free"})
liste.append({"id": "or-lightning", "label": "Nemotron 3.5 Lightning (OpenRouter, Test)",
"backend": "openrouter", "modell": "nvidia/nemotron-3.5-lightning:free"})
# Optionaler Eintrag zum schnellen Ausprobieren eines beliebigen
# OpenRouter-Modells, ohne Code zu aendern - einfach OPENROUTER_MODELL in
# .env setzen.
if config.OPENROUTER_MODELL:
liste.append({"id": "or-custom", "label": f"{config.OPENROUTER_MODELL} (OpenRouter, Test)",
"backend": "openrouter", "modell": config.OPENROUTER_MODELL})
_CACHE["auswahl"] = (jetzt, liste)
return liste
def _standard_id(optionen: list[dict]) -> str:
"""Bevorzugt das in der .env konfigurierte Modell, sonst die erste Option."""
for eintrag in optionen:
if eintrag["backend"] == "ollama" and eintrag["modell"] == config.OLLAMA_MODELL:
return eintrag["id"]
return optionen[0]["id"]
def aktuelle_id() -> str: def aktuelle_id() -> str:
"""Liest die aktive Auswahl. Unbekanntes/Fehlendes gilt als Standard.""" """Liest die aktive Auswahl. Unbekanntes/Fehlendes gilt als Standard."""
optionen = auswahl()
standard = _standard_id(optionen)
try: try:
wert = json.loads(DATEI.read_text(encoding="utf-8")).get("id") wert = json.loads(DATEI.read_text(encoding="utf-8")).get("id")
except Exception: except Exception:
return standard return _STANDARD
return wert if wert in {e["id"] for e in optionen} else standard return wert if wert in _NACH_ID else _STANDARD
def setze(kennung: str) -> bool: def setze(kennung: str) -> bool:
"""Schreibt die Auswahl. False, wenn die ID nicht (mehr) vorgesehen ist.""" """Schreibt die Auswahl. False, wenn die ID nicht vorgesehen ist."""
if kennung not in {e["id"] for e in auswahl()}: if kennung not in _NACH_ID:
return False return False
try: try:
DATEI.parent.mkdir(parents=True, exist_ok=True) DATEI.parent.mkdir(parents=True, exist_ok=True)
@ -114,6 +59,5 @@ def setze(kennung: str) -> bool:
def backend_und_modell() -> tuple[str, str]: def backend_und_modell() -> tuple[str, str]:
"""Backend + Modell-ID der aktiven Auswahl, fuer categorize.py.""" """Backend + Modell-ID der aktiven Auswahl, fuer categorize.py."""
kennung = aktuelle_id() eintrag = _NACH_ID[aktuelle_id()]
eintrag = next(e for e in auswahl() if e["id"] == kennung)
return eintrag["backend"], eintrag["modell"] return eintrag["backend"], eintrag["modell"]

View file

@ -1,70 +0,0 @@
"""Störungswache: prüft die Dienste-Ampel (pipe.dienste) periodisch und
alarmiert lokal (macOS-Benachrichtigung + Ton), wenn ein Dienst ausfällt -
auch wenn gerade niemand auf den Leitstand schaut.
python3 -m pipe.monitor [--einmal]
Läuft als eigener Prozess (siehe telefon/starten.sh), unabhängig von
pipe.watch und pipe.server - fällt einer der beiden aus, kann die Wache das
trotzdem noch melden. Alarmiert nur bei echten Störungen ("schlecht"/"aus"),
nicht bei "unklar" (z. B. Nextcloud nicht konfiguriert - das ist gewollt,
kein Fehler).
"""
from __future__ import annotations
import argparse
import time
from . import alarm, config, dienste
GESTOERT = {"schlecht", "aus"}
def pruefe(letzte_alarme: dict[str, float]) -> None:
"""Ein Durchlauf: Ampel abfragen, bei Bedarf alarmieren.
`letzte_alarme` hält je Dienstname den Zeitpunkt (time.monotonic) des
letzten Alarms - fehlt ein Name, gilt der Dienst aktuell als nicht
gestört. Der Aufrufer reicht dasselbe dict über alle Durchläufe weiter.
"""
jetzt = time.monotonic()
for dienst in dienste.status()["dienste"]:
name, zustand, text = dienst["name"], dienst["zustand"], dienst["text"]
letzter_alarm = letzte_alarme.get(name)
if zustand in GESTOERT:
if letzter_alarm is None:
alarm.sende(f"{name} gestört", text)
letzte_alarme[name] = jetzt
elif jetzt - letzter_alarm >= config.MONITOR_WIEDERHOLUNG_MIN * 60:
alarm.sende(f"{name} weiterhin gestört", text)
letzte_alarme[name] = jetzt
elif zustand == "gut" and letzter_alarm is not None:
alarm.sende(f"{name} wieder ok", text)
del letzte_alarme[name]
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(
description="Dienste-Ampel beobachten und bei Störung lokal alarmieren.")
p.add_argument("--einmal", action="store_true",
help="nur einmal prüfen statt dauerhaft zu beobachten")
args = p.parse_args(argv)
letzte_alarme: dict[str, float] = {}
if args.einmal:
pruefe(letzte_alarme)
return 0
print(f"Störungswache aktiv - Takt {config.MONITOR_TAKT_S}s, "
f"Erinnerung alle {config.MONITOR_WIEDERHOLUNG_MIN} min - Abbruch mit Strg-C")
try:
while True:
pruefe(letzte_alarme)
time.sleep(config.MONITOR_TAKT_S)
except KeyboardInterrupt:
print("\nBeobachtung beendet.")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -12,7 +12,6 @@ from __future__ import annotations
import json import json
import shutil import shutil
import time
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
@ -80,26 +79,3 @@ def archiviere_und_leere() -> Path | None:
return ziel return ziel
except Exception: except Exception:
return None return None
ARCHIV_FRIST_TAGE = 90
def raeume_archiv_auf(tage: int = ARCHIV_FRIST_TAGE) -> int:
"""Löscht Verlauf-Archivkopien (nicht audit.log!) älter als `tage` Tage.
Ein neues verlauf_*.log entsteht bei jedem "Tag archivieren" - ohne
Aufräumen sammeln sich diese sonst unbegrenzt an. Wirft nie, gibt die
Anzahl gelöschter Dateien zurück."""
if not ARCHIV_ORDNER.is_dir():
return 0
grenze = time.time() - tage * 86400
geloescht = 0
for datei in ARCHIV_ORDNER.glob("verlauf_*.log"):
try:
if datei.stat().st_mtime < grenze:
datei.unlink()
geloescht += 1
except Exception:
continue
return geloescht

File diff suppressed because one or more lines are too long

View file

@ -1,35 +0,0 @@
"""Signierte, zeitlich begrenzte Testzugangs-Links zum Leitstand.
Gegenstueck zu dilles_agent_one_test_command() in dillenberg.net's
dilles-agent.php ("agent-one test"-Kommando im Alice-Chat). Beide Seiten
teilen sich TESTZUGANG_SECRET (config.py) - der Link wird komplett auf der
PHP-Seite erzeugt, ohne Live-Aufruf hierher; hier wird nur die Signatur
geprueft. Stateless: kein Server-seitiger Speicher fuer ausgegebene Tokens,
die Gueltigkeit steckt komplett im signierten Ablauf-Zeitstempel.
Token-Format: "<ablauf-unixzeit>.<hmac-sha256-hexdigest>"
"""
from __future__ import annotations
import hmac
import time
from hashlib import sha256
from . import config
def gueltig(token: str) -> bool:
"""Prueft ein Token aus ?zugang=... oder dem zugang-Cookie. Wirft nie."""
if not config.TESTZUGANG_SECRET or not token:
return False
try:
ablauf_str, signatur = token.split(".", 1)
ablauf = int(ablauf_str)
except (ValueError, AttributeError):
return False
erwartet = hmac.new(
config.TESTZUGANG_SECRET.encode("utf-8"), ablauf_str.encode("utf-8"), sha256
).hexdigest()
if not hmac.compare_digest(erwartet, signatur):
return False
return time.time() < ablauf

View file

@ -152,18 +152,10 @@ def main(argv: list[str] | None = None) -> int:
try: try:
while True: while True:
durchlauf(eingang) durchlauf(eingang)
if time.monotonic() - letzter_loeschlauf >= LOESCH_TAKT_S: if config.LOESCHFRIST_TAGE and time.monotonic() - letzter_loeschlauf >= LOESCH_TAKT_S:
if config.LOESCHFRIST_TAGE: anzahl = loeschen.laeuft()
anzahl = loeschen.laeuft() if anzahl:
if anzahl: print(f"🗑 {anzahl} Anruf(e) nach Aufbewahrungsfrist geloescht.")
print(f"🗑 {anzahl} Anruf(e) nach Aufbewahrungsfrist geloescht.")
# Verlauf-Archivkopien (pipe.protokoll, "Tag archivieren")
# sammeln sich sonst unbegrenzt an - unabhaengig von
# LOESCHFRIST_TAGE, das nur Anrufe betrifft, nicht Log-Kopien.
archiv_geloescht = protokoll.raeume_archiv_auf()
if archiv_geloescht:
print(f"🗑 {archiv_geloescht} alte Verlauf-Archivkopie(n) geloescht "
f"(> {protokoll.ARCHIV_FRIST_TAGE} Tage).")
letzter_loeschlauf = time.monotonic() letzter_loeschlauf = time.monotonic()
time.sleep(TAKT_S) time.sleep(TAKT_S)
except KeyboardInterrupt: except KeyboardInterrupt:

Binary file not shown.

View file

@ -1 +0,0 @@
Aufzug-Notdienst, Testansage. Bitte hinterlassen Sie eine Nachricht nach dem Signal.

View file

@ -1,12 +0,0 @@
{
"profil_name": "Aufzug-Notdienst",
"kategorien": ["eingeschlossen", "stoerung", "wartung", "rueckruf", "sonstiges"],
"kategorie_labels": {
"eingeschlossen": "Person eingeschlossen",
"stoerung": "Störung",
"wartung": "Wartung",
"rueckruf": "Rückruf",
"sonstiges": "Sonstiges"
},
"sicherheitsnetz_kategorie": "eingeschlossen"
}

View file

@ -1,44 +0,0 @@
Du bist der Anruf-Assistent eines 24-Stunden-Notdienstes für Aufzugsanlagen.
Du bekommst das Transkript einer Sprachnachricht, die ein Anrufer auf den
Anrufbeantworter gesprochen hat. Deine Aufgabe: das Anliegen sachlich
erfassen und einordnen.
WICHTIG:
- Du gibst KEINE technischen Anweisungen und versuchst nicht, das Problem
selbst zu lösen (z. B. "versuchen Sie die Tür aufzudrücken").
- Du erfindest nichts. Fehlt eine Angabe im Transkript, setze null.
- Fasse neutral und knapp zusammen, in der dritten Person.
- Das Transkript kann fehlerhaft sein (automatische Spracherkennung). Interpretiere wohlwollend.
- Antworte AUSSCHLIESSLICH mit dem geforderten JSON-Objekt, ohne weiteren Text.
Kategorien (Feld "kategorie", genau einer dieser Werte):
- "eingeschlossen" : akut jemand im Aufzug eingeschlossen/eingeklemmt, kann
nicht selbst raus
- "stoerung" : Aufzug ausser Betrieb, meldet Fehler, ungewöhnliche
Geräusche o. Ä. - aber NIEMAND eingeschlossen
- "wartung" : Wartungs-/Inspektionstermin vereinbaren, Rückfrage zu
Vertrag/Rechnung
- "rueckruf" : bittet um Rückruf ohne klar anderes Anliegen
- "sonstiges" : passt in keine der obigen
Dringlichkeit (Feld "dringlichkeit"): "niedrig" | "normal" | "hoch" | "notfall"
- "notfall" nur bei akut eingeschlossener/eingeklemmter Person - jede Minute zählt.
- Ein stehender/gestörter Aufzug ohne eingeschlossene Person ist mindestens
"hoch", wenn er das einzige Fahrzeug im Gebäude ist oder Personen mit
Mobilitätseinschränkung betroffen sind - sonst "normal".
- Im Zweifel die höhere Stufe wählen: eine zu hoch eingestufte Nachricht
kostet einen Blick, eine zu niedrig eingestufte kann jemanden im Aufzug
gefangen lassen.
Gib ein JSON-Objekt mit genau diesen Feldern zurück:
{
"kategorie": "<eine der Kategorien>",
"dringlichkeit": "<niedrig|normal|hoch|notfall>",
"anrufer_name": "<Name oder null>",
"rueckrufnummer": "<Telefonnummer aus dem Text oder null>",
"rueckruf_gewuenscht": <true|false>,
"email": "<E-Mail-Adresse aus dem Text oder null, im Format name@domain.tld>",
"sprache": "<ISO-Code der Anrufersprache, z. B. de, en, ar>",
"anliegen_kurz": "<1-2 Sätze, neutrale Zusammenfassung auf Deutsch>",
"stichworte": ["<kurze Schlagworte, z. B. Adresse/Gebäude, Aufzug-Nr.>"]
}

View file

@ -1,16 +0,0 @@
# Deterministisches Sicherheitsnetz, unabhaengig vom LLM: ein Regex-Muster
# pro Zeile (Python re, IGNORECASE), leere Zeilen und "#"-Kommentare werden
# uebersprungen. Findet ein Muster im Transkript, eskaliert categorize.py
# Kategorie + Dringlichkeit hart auf "sicherheitsnetz_kategorie" aus
# kategorien.json ("eingeschlossen") - unabhaengig davon, was das Modell
# selbst einordnet. Nur nach oben, nie nach unten.
eingeschlossen
eingeklemmt
(sitzt|steckt|stecken)\s+fest
komm(e|en)\s+nicht\s+(mehr\s+)?raus
kommt?\s+nicht\s+(mehr\s+)?(raus|heraus)
t[üu]r\s+(geht|l[äa]sst\s+sich)\s+nicht\s+(auf|[öo]ffnen)
notruftaste
kabine\s+(h[äa]ngt|steht)\s+fest
seit\s+\w+\s+(stunden?|minuten?)\s+im\s+aufzug
person\s+im\s+aufzug

View file

@ -1,19 +0,0 @@
{
"profil_name": "Hausarztpraxis",
"kategorien": [
"termin", "rezept", "ueberweisung", "befund",
"verwaltung", "beschwerden", "notfall", "rueckruf", "sonstiges"
],
"kategorie_labels": {
"termin": "Termin",
"rezept": "Rezept",
"ueberweisung": "Überweisung",
"befund": "Befund",
"verwaltung": "Verwaltung",
"beschwerden": "Beschwerden",
"notfall": "Notfall",
"rueckruf": "Rückruf",
"sonstiges": "Sonstiges"
},
"sicherheitsnetz_kategorie": "notfall"
}

View file

@ -1,23 +0,0 @@
# Deterministisches Sicherheitsnetz, unabhaengig vom LLM: ein Regex-Muster
# pro Zeile (Python re, IGNORECASE), leere Zeilen und "#"-Kommentare werden
# uebersprungen. Findet ein Muster im Transkript, eskaliert categorize.py
# Kategorie + Dringlichkeit hart auf "sicherheitsnetz_kategorie" aus
# kategorien.json ("notfall") - unabhaengig davon, was das Modell selbst
# einordnet. Nur nach oben, nie nach unten: ein Fehlalarm kostet einen Blick,
# ein uebersehener Notfall kann einen Patienten kosten.
#
# Grund fuer dieses Netz: getestet mit qwen3:8b, 3/3 identische Laeufe
# stuften "starke Brustschmerzen ... kriege kaum Luft" nur als "hoch" statt
# "notfall" ein - obwohl das Modell selbst "Brustschmerzen" und "Atemnot"
# als Stichworte erkannte. Bei Notfall-Erkennung darf man sich nicht allein
# auf ein LLM verlassen.
atemnot
(kriege|bekomme|krieg)\w*\s+(kaum|keine|schwer)\s+luft
kann\s+(kaum|nicht)\s+(mehr\s+)?atmen
brust\w*schmerz
schmerzen?\s+in\s+der\s+brust
bewusstlos
ohnm[aä]chtig
nicht\s+ansprechbar
blutet\s+(stark|heftig|sehr)
starke?\s+blutung

View file

@ -1,20 +1,16 @@
#!/bin/bash #!/bin/bash
# Erzeugt die Telefonansage aus profile/$PROFIL/ansage.txt (Branchen-Profil, # Erzeugt die Telefonansage aus telefon/ansage.txt.
# siehe README).
# Ergebnis: telefon/ansage.wav in Telefonqualitaet (8 kHz mono, PCM16). # Ergebnis: telefon/ansage.wav in Telefonqualitaet (8 kHz mono, PCM16).
# #
# Vier Quellen: # Drei Sprachsynthesen:
# piper (Standard, natuerlicher, lokal) - Modelle unter ~/piper-voices # piper (Standard, natuerlicher, lokal) - Modelle unter ~/piper-voices
# say (macOS-Bordmittel, lokal, Rueckfall wenn Piper fehlt) # say (macOS-Bordmittel, lokal, Rueckfall wenn Piper fehlt)
# mistral (Voxtral-API, Cloud - nur fuer diesen einmaligen Erzeugungslauf; # mistral (Voxtral-API, Cloud - nur fuer diesen einmaligen Erzeugungslauf;
# die fertige ansage.wav wird danach lokal fuer jeden Anruf # die fertige ansage.wav wird danach lokal fuer jeden Anruf
# wiederverwendet, es geht dabei keine Patientendaten raus) # wiederverwendet, es geht dabei keine Patientendaten raus)
# datei (echte Aufnahme statt Synthese - profile/$PROFIL/ansage.<ext>,
# ansage.txt wird dabei nicht gelesen)
# #
# Steuerung ueber Umgebungsvariablen bzw. .env: # Steuerung ueber Umgebungsvariablen bzw. .env:
# PROFIL=<Ordnername unter profile/, Default "praxis"> # ANSAGE_TTS=piper|say|mistral
# ANSAGE_TTS=piper|say|mistral|datei
# ANSAGE_STIMME=<Piper-Modellname, say-Stimme, oder Voxtral-Stimmen-Slug> # ANSAGE_STIMME=<Piper-Modellname, say-Stimme, oder Voxtral-Stimmen-Slug>
# ANSAGE_EMOTION=<Sprecher-ID, nur bei thorsten_emotional: 0=amused, # ANSAGE_EMOTION=<Sprecher-ID, nur bei thorsten_emotional: 0=amused,
# 1=angry, 2=disgusted, 3=drunk, 4=neutral, 5=sleepy, # 1=angry, 2=disgusted, 3=drunk, 4=neutral, 5=sleepy,
@ -36,10 +32,6 @@ if [ -f .env ]; then
done < .env done < .env
fi fi
PROFIL="${PROFIL:-praxis}"
ANSAGE_TXT="profile/$PROFIL/ansage.txt"
[ -f "$ANSAGE_TXT" ] || { echo "Ansage-Text fehlt: $ANSAGE_TXT (PROFIL=$PROFIL)" >&2; exit 1; }
STIMMEN_ORDNER="${PIPER_STIMMEN:-$HOME/piper-voices}" STIMMEN_ORDNER="${PIPER_STIMMEN:-$HOME/piper-voices}"
STIMME="${ANSAGE_STIMME:-de_DE-thorsten_emotional-medium}" STIMME="${ANSAGE_STIMME:-de_DE-thorsten_emotional-medium}"
EMOTION="${ANSAGE_EMOTION:-0}" EMOTION="${ANSAGE_EMOTION:-0}"
@ -58,28 +50,10 @@ fi
TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT
case "$TTS" in case "$TTS" in
datei)
# Echte Aufnahme statt Sprachsynthese - profile/$PROFIL/ansage.<ext>
# (erste gefundene Endung gewinnt). ansage.txt wird in diesem Modus nicht
# gelesen; sie bleibt trotzdem als Beleg/Backup stehen, was gesagt wird.
QUELLE=""
for ext in ogg mp3 wav m4a aiff; do
if [ -f "profile/$PROFIL/ansage.$ext" ]; then
QUELLE="profile/$PROFIL/ansage.$ext"
break
fi
done
[ -n "$QUELLE" ] || {
echo "Keine Ansage-Audiodatei gefunden (profile/$PROFIL/ansage.{ogg,mp3,wav,m4a,aiff})" >&2
exit 1
}
ffmpeg -v error -y -i "$QUELLE" -ar 8000 -ac 1 "$TMP/sprache.wav"
BESCHREIBUNG="datei/$QUELLE"
;;
piper) piper)
MODELL="$STIMMEN_ORDNER/$STIMME.onnx" MODELL="$STIMMEN_ORDNER/$STIMME.onnx"
[ -f "$MODELL" ] || { echo "Piper-Stimme fehlt: $MODELL" >&2; exit 1; } [ -f "$MODELL" ] || { echo "Piper-Stimme fehlt: $MODELL" >&2; exit 1; }
tr '\n' ' ' < "$ANSAGE_TXT" \ tr '\n' ' ' < telefon/ansage.txt \
| python3 -m piper -m "$MODELL" -s "$EMOTION" --length-scale "$TEMPO" \ | python3 -m piper -m "$MODELL" -s "$EMOTION" --length-scale "$TEMPO" \
-f "$TMP/sprache_roh.wav" -f "$TMP/sprache_roh.wav"
ffmpeg -v error -y -i "$TMP/sprache_roh.wav" -ar 8000 -ac 1 "$TMP/sprache.wav" ffmpeg -v error -y -i "$TMP/sprache_roh.wav" -ar 8000 -ac 1 "$TMP/sprache.wav"
@ -87,7 +61,7 @@ case "$TTS" in
;; ;;
say) say)
SAY_STIMME="${ANSAGE_STIMME:-Anna}" SAY_STIMME="${ANSAGE_STIMME:-Anna}"
say -v "$SAY_STIMME" -r 180 -f "$ANSAGE_TXT" -o "$TMP/sprache.aiff" say -v "$SAY_STIMME" -r 180 -f telefon/ansage.txt -o "$TMP/sprache.aiff"
ffmpeg -v error -y -i "$TMP/sprache.aiff" -ar 8000 -ac 1 "$TMP/sprache.wav" ffmpeg -v error -y -i "$TMP/sprache.aiff" -ar 8000 -ac 1 "$TMP/sprache.wav"
BESCHREIBUNG="say/$SAY_STIMME" BESCHREIBUNG="say/$SAY_STIMME"
;; ;;
@ -95,7 +69,7 @@ case "$TTS" in
MISTRAL_STIMME="${ANSAGE_STIMME:-en_paul_neutral}" MISTRAL_STIMME="${ANSAGE_STIMME:-en_paul_neutral}"
MISTRAL_MODELL="${MISTRAL_TTS_MODEL:-voxtral-mini-tts-latest}" MISTRAL_MODELL="${MISTRAL_TTS_MODEL:-voxtral-mini-tts-latest}"
[ -n "${MISTRAL_KEY:-}" ] || { echo "MISTRAL_KEY fehlt (ANSAGE_TTS=mistral)" >&2; exit 1; } [ -n "${MISTRAL_KEY:-}" ] || { echo "MISTRAL_KEY fehlt (ANSAGE_TTS=mistral)" >&2; exit 1; }
TEXT="$(tr '\n' ' ' < "$ANSAGE_TXT")" TEXT="$(tr '\n' ' ' < telefon/ansage.txt)"
PAYLOAD="$(TTS_MODEL="$MISTRAL_MODELL" TTS_VOICE="$MISTRAL_STIMME" python3 -c ' PAYLOAD="$(TTS_MODEL="$MISTRAL_MODELL" TTS_VOICE="$MISTRAL_STIMME" python3 -c '
import json, os, sys import json, os, sys
print(json.dumps({ print(json.dumps({

View file

@ -18,20 +18,6 @@ if [ -f .env ]; then
done < .env done < .env
fi fi
# Rohe Prozess-Logs (stdout/stderr von watcher/leitstand/monitor + die
# eigene Ausgabe hier, wenn per launchd-Watchdog alle 5 Min aufgerufen)
# wachsen sonst unbegrenzt - anders als verlauf.log (protokoll.kuerze(), nach
# jedem Anruf) und audit.log (bewusst permanent, siehe pipe/audit.py). Ab
# 2 MB auf die letzten 2000 Zeilen kuerzen, in-place (gleiches Inode) - sicher
# fuer >>-Redirects (O_APPEND springt bei jedem Schreiben ans Dateiende,
# unabhaengig vom vorherigen Offset).
LOG_GRENZE_BYTES=2000000
for log in telefon/autostart.log telefon/watcher.log telefon/leitstand.log telefon/monitor.log; do
if [ -f "$log" ] && [ "$(wc -c < "$log" 2>/dev/null || echo 0)" -gt "$LOG_GRENZE_BYTES" ]; then
tail -n 2000 "$log" > "$log.tmp" 2>/dev/null && cat "$log.tmp" > "$log" && rm -f "$log.tmp"
fi
done
fehler=0 fehler=0
melde() { printf ' %-38s %s\n' "$1" "$2"; } melde() { printf ' %-38s %s\n' "$1" "$2"; }
@ -113,21 +99,15 @@ if [ "${status:-}" = "UP" ]; then
else else
# nice 10: Transkription und Kategorisierung dürfen warten, ein laufendes # nice 10: Transkription und Kategorisierung dürfen warten, ein laufendes
# Telefonat nicht. Auf 16 GB RAM konkurrieren beide sonst um die Maschine. # Telefonat nicht. Auf 16 GB RAM konkurrieren beide sonst um die Maschine.
nohup nice -n 10 python3 -u -m pipe.watch >> telefon/watcher.log 2>&1 & nohup nice -n 10 python3 -u -m pipe.watch > telefon/watcher.log 2>&1 &
echo "Verarbeitung gestartet (Log: telefon/watcher.log)" echo "Verarbeitung gestartet (Log: telefon/watcher.log)"
fi fi
if pgrep -f "[p]ipe.server" >/dev/null; then if pgrep -f "[p]ipe.server" >/dev/null; then
echo "Leitstand laeuft bereits." echo "Leitstand laeuft bereits."
else else
nohup python3 -u -m pipe.server >> telefon/leitstand.log 2>&1 & nohup python3 -u -m pipe.server > telefon/leitstand.log 2>&1 &
sleep 2 sleep 2
tail -1 telefon/leitstand.log head -1 telefon/leitstand.log
fi
if pgrep -f "[p]ipe.monitor" >/dev/null; then
echo "Stoerungswache laeuft bereits."
else
nohup python3 -u -m pipe.monitor >> telefon/monitor.log 2>&1 &
echo "Stoerungswache gestartet (Log: telefon/monitor.log)"
fi fi
sleep 2 sleep 2
echo echo