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_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 ---
# Backend: "ollama" (Default, lokal) oder "openrouter" (Cloud, nur zum Testen
# ob ein staerkeres Modell besser extrahiert - Transkript verlaesst dabei den
@ -51,14 +45,6 @@ NEXTCLOUD_ADRESSBUCH=contacts
# Anrufe bleiben unangetastet, egal wie alt.
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) ---
SIP_USER=
SIP_DOMAIN=sip.plusnet.de
@ -99,13 +85,6 @@ LEITSTAND_HOST=127.0.0.1
LEITSTAND_PORT=8088
LEITSTAND_USER=praxis
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) ---
# Event-Socket-Port von FreeSWITCH. Muss zu listen-port in

View file

@ -1,44 +1,5 @@
# 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
Erste stabile Version. Läuft produktiv (mit Testdaten) auf macOS/Apple

126
README.md
View file

@ -1,31 +1,30 @@
# 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
ab (lokal und optional in Nextcloud). Für welchen Einsatzbereich (Hausarztpraxis,
24h-Notdienst, …) ist reine Konfiguration — siehe
[Branchen-Profil](#kategorien--branchen-profil). Zwei Beispiele liegen bei:
`praxis` (Hausarztpraxis) und `aufzug-notdienst`.
ab (lokal und optional in Nextcloud).
**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
> erfunden — auch im Screenshot und in der Beispielkonfiguration. „Praxis
> Musterhausen" ist ein Platzhalter. Echte Zugangsdaten und Anrufdaten liegen
> 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
1. **Die Pipe** (fertig): Audio → Transkript → Kategorie → Ablage.
2. **Telefonie** (steht): FreeSWITCH nimmt über den Plusnet-Trunk an, spielt die
Ansage, nimmt auf und legt die Datei im Eingang der Pipe ab.
3. **Feinschliff** (teilweise): Löschfristen ✅, Notfall-Alarm ✅,
Verschlüsselung at rest ✅ (Betreiber-Pflicht: FileVault, siehe
[Datenschutz](#datenschutz-dsgvo)), Monitoring/Alerting bei Diensteausfall ✅
(`pipe/monitor.py`, siehe unten).
3. **Feinschliff** (teilweise): Löschfristen ✅, Notfall-Alarm ✅ — offen:
Monitoring/Alerting bei Diensteausfall, Verschlüsselung at rest.
## Bausteine
@ -38,62 +37,19 @@ ab (lokal und optional in Nextcloud). Für welchen Einsatzbereich (Hausarztpraxi
| `pipe/config.py` | Konfiguration aus `.env` |
| `pipe/watch.py` | Beobachtet den Eingang und schiebt Aufnahmen durch die Pipe |
| `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/audit.py` | Unveränderliches Audit-Log (append-only, siehe unten) |
| `pipe/loeschen.py` | Löscht erledigte Anrufe nach Ablauf der Aufbewahrungsfrist |
| `pipe/bewertung.py` | 👍/👎-Bewertung pro Anruf (Leitstand) |
| `pipe/export.py` | Exportiert bewertete Anrufe als JSONL-Trainingsdaten |
| `pipe/kategorien.py` | Lädt Kategorien-Schema des aktiven Branchen-Profils (siehe unten) |
| `profile/<name>/` | Branchen-Profil: Prompt, Kategorien, Sicherheitsnetz, Ansage |
| `telefon/` | FreeSWITCH-Vorlagen, Start- und Einrichtungsskripte |
| `prompts/categorize_de.txt` | Kategorisierungs-Prompt (deutsch, mehrsprachig-tauglich) |
| `telefon/` | Ansage, FreeSWITCH-Vorlagen, Start- und Einrichtungsskripte |
## Kategorien & Branchen-Profil
## Kategorien
Alles, was fachlich vom Einsatzzweck abhängt, steckt in einem **Branchen-Profil**
unter `profile/<name>/` — vier Dateien, die zueinander passen müssen:
| 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.
`termin · rezept · ueberweisung · befund · verwaltung · beschwerden · notfall ·
rueckruf · sonstiges`
Dringlichkeit: `niedrig · normal · hoch · notfall`
## 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.
```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/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
setzt deren Rechte auf 600 — im Git liegen nur Vorlagen mit Platzhaltern.
Text ändern: `profile/$PROFIL/ansage.txt` bearbeiten (siehe
[Branchen-Profil](#kategorien--branchen-profil)), `ansage_bauen.sh` erneut
Text ändern: `telefon/ansage.txt` bearbeiten, `ansage_bauen.sh` erneut
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
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
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
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`).
- Verarbeitung lokal (Whisper + Qwen) — kein Cloud-Dienst im Datenpfad.
> **Offener Punkt:** Die aktuelle `profile/praxis/ansage.txt` erfüllt die
> ersten beiden Punkte noch nicht (kein Notfall-Hinweis, keine
> Aufnahme-Einwilligung) — Text anpassen und `ansage_bauen.sh` neu laufen
> lassen, bevor die Anlage im 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.
> **Offener Punkt:** Die aktuelle `telefon/ansage.txt` erfüllt die ersten
> beiden Punkte noch nicht (kein Notfall-Hinweis, keine Aufnahme-Einwilligung)
> — Text anpassen und `ansage_bauen.sh` neu laufen lassen, bevor die Anlage im
> echten Praxisbetrieb ans Netz geht.

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
from functools import lru_cache
from . import config, kategorien, modellwahl
from . import config, modellwahl
KATEGORIEN = kategorien.KATEGORIEN
DRINGLICHKEITEN = kategorien.DRINGLICHKEITEN
KATEGORIEN = [
"termin", "rezept", "ueberweisung", "befund",
"verwaltung", "beschwerden", "notfall", "rueckruf", "sonstiges",
]
DRINGLICHKEITEN = ["niedrig", "normal", "hoch", "notfall"]
@lru_cache(maxsize=1)
def _sicherheitsnetz_muster() -> list[re.Pattern]:
"""Lädt die Eskalations-Regex-Muster aus profile/<PROFIL>/sicherheitsnetz.txt.
Ein Muster pro Zeile, Kommentare ("#") und Leerzeilen werden übersprungen.
Deterministisch, unabhängig vom LLM - siehe Rationale in der Profil-Datei
selbst. Eskaliert nur nach oben, nie nach unten.
"""
if not config.SICHERHEITSNETZ_DATEI.is_file():
return []
return [
re.compile(zeile, re.IGNORECASE)
for zeile in (z.strip() for z in config.SICHERHEITSNETZ_DATEI.read_text(encoding="utf-8").splitlines())
if zeile and not zeile.startswith("#")
]
# Deterministisches Sicherheitsnetz, unabhaengig vom LLM: dieselben Symptome,
# die der Prompt selbst als Notfall-Beispiele nennt (Atemnot, Brustschmerz,
# Bewusstlosigkeit, starke Blutung). Grund: 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. 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).
_NOTFALL_MUSTER = [re.compile(p, re.IGNORECASE) for p in (
r"atemnot",
r"(kriege|bekomme|krieg)\w*\s+(kaum|keine|schwer)\s+luft",
r"kann\s+(kaum|nicht)\s+(mehr\s+)?atmen",
r"brust\w*schmerz",
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:
ziel = kategorien.SICHERHEITSNETZ_KATEGORIE
if d.get("kategorie") != ziel and any(m.search(transkript) for m in _sicherheitsnetz_muster()):
d["kategorie"] = ziel
if d.get("dringlichkeit") != "notfall" and any(m.search(transkript) for m in _NOTFALL_MUSTER):
d["kategorie"] = "notfall"
d["dringlichkeit"] = "notfall"
return d
@ -237,11 +244,11 @@ def _bereinige(d: dict) -> dict:
d["kategorie"] = "sonstiges"
if d.get("dringlichkeit") not in DRINGLICHKEITEN:
d["dringlichkeit"] = "normal"
# Das Modell vergibt gelegentlich die Sicherheitsnetz-Kategorie (z. B.
# "notfall") und stuft die Dringlichkeit trotzdem nur auf "hoch" ein. Das
# ist widersprüchlich und führt zu einer harmloseren Farbe auf dem Board,
# als der Anruf verdient. Im Zweifel eskaliert man, statt abzuschwächen.
if d["kategorie"] == kategorien.SICHERHEITSNETZ_KATEGORIE:
# Das Modell vergibt gelegentlich die Kategorie "notfall" und stuft die
# Dringlichkeit trotzdem nur auf "hoch" ein. Das ist widersprüchlich und
# führt zu einer harmloseren Farbe auf dem Board, als der Anruf verdient.
# In der Praxis eskaliert man im Zweifel, statt abzuschwächen.
if d["kategorie"] == "notfall":
d["dringlichkeit"] = "notfall"
d.setdefault("anrufer_name", 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.
OPENROUTER_MODELL = os.environ.get("OPENROUTER_MODELL", "")
# --- Branchen-Profil -----------------------------------------------------
# Buendelt alles, was fachlich vom Einsatzzweck abhaengt (Kategorien, Prompt,
# 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"
PROMPT_DATEI = WURZEL / "prompts" / os.environ.get(
"PROMPT_DATEI", "categorize_de.txt"
)
# --- 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.
LEITSTAND_USER = os.environ.get("LEITSTAND_USER", "praxis")
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:
@ -196,19 +162,6 @@ def nextcloud_aktiv() -> bool:
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) --------------------------------------
# Nach so vielen Tagen wird ein Anruf-Ordner (Aufnahme + Transkript + Meta)
# geloescht - lokal und, falls hochgeladen, auch bei Nextcloud. 0 = aus (nichts

View file

@ -11,11 +11,15 @@ import json
import sys
from pathlib import Path
from . import config, kategorien
from . import config
RANG = kategorien.DRINGLICHKEIT_RANG
FARBE = {k: f"#{v}" for k, v in kategorien.DRINGLICHKEIT_FARBE.items()}
KAT_LABEL = kategorien.KATEGORIE_LABEL
RANG = {"notfall": 0, "hoch": 1, "normal": 2, "niedrig": 3}
FARBE = {"notfall": "#E9322D", "hoch": "#E0A339", "normal": "#C9A227", "niedrig": "#31CC7C"}
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]:
@ -61,7 +65,7 @@ def baue(ausgabe: Path) -> Path:
karten = "\n".join(_karte(d) for d in anrufe)
doc = f"""<!doctype html><html lang="de"><head><meta charset="utf-8">
<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>
: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}}}}
@ -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}}
.tr{{color:var(--muted);font-size:13px;margin:6px 0 0}}
</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>
{karten}
</div></body></html>"""

View file

@ -18,15 +18,24 @@ import urllib.error
import urllib.request
from functools import lru_cache
from . import config, kategorien
from . import config
_API = "/index.php/apps/deck/api/v1.1"
# Farben je Dringlichkeit - siehe pipe/kategorien.py.
DRINGLICHKEIT_FARBE = kategorien.DRINGLICHKEIT_FARBE
# Farben (6-stelliges Hex ohne #) je Dringlichkeit.
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,
# 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)

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
JSON-Datei geteilt, die categorize.py bei jedem Anruf frisch liest (kein
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
import json
import time
import urllib.error
import urllib.request
from . import config
DATEI = config.TELEFON / "modell.json"
# Ollama-Abfrage ist ein Netzwerkaufruf - kurz zwischenspeichern, damit ein
# offener Leitstand (Polling alle paar Sekunden) Ollama nicht dauerbelastet.
_CACHE: dict[str, tuple[float, list[dict]]] = {}
CACHE_S = 30.0
def _ollama_modelle() -> list[str]:
"""Namen der lokal installierten, textfähigen Ollama-Modelle (leer bei Fehler).
Reine Embedding-Modelle (z. B. nomic-embed-text) werden ausgefiltert - sie
können nicht kategorisieren, ein Aufruf würde nur mit einem Fehler enden.
Ältere Ollama-Versionen liefern kein "capabilities"-Feld - dann bleibt das
Modell sicherheitshalber drin, statt es zu Unrecht zu verstecken.
"""
req = urllib.request.Request(f"{config.OLLAMA_URL}/api/tags",
headers={"User-Agent": "praxis-telefon-agent/1.0"})
try:
with urllib.request.urlopen(req, timeout=5) as antwort:
daten = json.loads(antwort.read().decode("utf-8"))
except Exception:
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"]
# Feste, kuratierte Auswahl statt freier Eingabe - ein Tippfehler in einer
# Modell-ID im Leitstand darf nicht den naechsten Anruf scheitern lassen.
# "openrouter"-Eintraege sind zum Testen gedacht (Transkript verlaesst dabei
# den Rechner) - nicht fuer den Praxisbetrieb.
AUSWAHL = [
{"id": "ollama", "label": f"{config.OLLAMA_MODELL} (lokal)",
"backend": "ollama", "modell": config.OLLAMA_MODELL},
{"id": "or-ultra", "label": "Nemotron Ultra 550B (OpenRouter, Test)",
"backend": "openrouter", "modell": "nvidia/nemotron-3-ultra-550b-a55b:free"},
{"id": "or-lightning", "label": "Nemotron 3.5 Lightning (OpenRouter, Test)",
"backend": "openrouter", "modell": "nvidia/nemotron-3.5-lightning:free"},
]
# Optionaler vierter Eintrag zum schnellen Ausprobieren eines beliebigen
# OpenRouter-Modells, ohne Code zu aendern - einfach OPENROUTER_MODELL in
# .env setzen.
if config.OPENROUTER_MODELL:
AUSWAHL.append({"id": "or-custom", "label": f"{config.OPENROUTER_MODELL} (OpenRouter, Test)",
"backend": "openrouter", "modell": config.OPENROUTER_MODELL})
_STANDARD = AUSWAHL[0]["id"]
_NACH_ID = {e["id"]: e for e in AUSWAHL}
def aktuelle_id() -> str:
"""Liest die aktive Auswahl. Unbekanntes/Fehlendes gilt als Standard."""
optionen = auswahl()
standard = _standard_id(optionen)
try:
wert = json.loads(DATEI.read_text(encoding="utf-8")).get("id")
except Exception:
return standard
return wert if wert in {e["id"] for e in optionen} else standard
return _STANDARD
return wert if wert in _NACH_ID else _STANDARD
def setze(kennung: str) -> bool:
"""Schreibt die Auswahl. False, wenn die ID nicht (mehr) vorgesehen ist."""
if kennung not in {e["id"] for e in auswahl()}:
"""Schreibt die Auswahl. False, wenn die ID nicht vorgesehen ist."""
if kennung not in _NACH_ID:
return False
try:
DATEI.parent.mkdir(parents=True, exist_ok=True)
@ -114,6 +59,5 @@ def setze(kennung: str) -> bool:
def backend_und_modell() -> tuple[str, str]:
"""Backend + Modell-ID der aktiven Auswahl, fuer categorize.py."""
kennung = aktuelle_id()
eintrag = next(e for e in auswahl() if e["id"] == kennung)
eintrag = _NACH_ID[aktuelle_id()]
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 shutil
import time
from datetime import datetime
from pathlib import Path
@ -80,26 +79,3 @@ def archiviere_und_leere() -> Path | None:
return ziel
except Exception:
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:
while True:
durchlauf(eingang)
if time.monotonic() - letzter_loeschlauf >= LOESCH_TAKT_S:
if config.LOESCHFRIST_TAGE:
anzahl = loeschen.laeuft()
if anzahl:
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).")
if config.LOESCHFRIST_TAGE and time.monotonic() - letzter_loeschlauf >= LOESCH_TAKT_S:
anzahl = loeschen.laeuft()
if anzahl:
print(f"🗑 {anzahl} Anruf(e) nach Aufbewahrungsfrist geloescht.")
letzter_loeschlauf = time.monotonic()
time.sleep(TAKT_S)
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
# Erzeugt die Telefonansage aus profile/$PROFIL/ansage.txt (Branchen-Profil,
# siehe README).
# Erzeugt die Telefonansage aus telefon/ansage.txt.
# Ergebnis: telefon/ansage.wav in Telefonqualitaet (8 kHz mono, PCM16).
#
# Vier Quellen:
# Drei Sprachsynthesen:
# piper (Standard, natuerlicher, lokal) - Modelle unter ~/piper-voices
# say (macOS-Bordmittel, lokal, Rueckfall wenn Piper fehlt)
# mistral (Voxtral-API, Cloud - nur fuer diesen einmaligen Erzeugungslauf;
# die fertige ansage.wav wird danach lokal fuer jeden Anruf
# 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:
# PROFIL=<Ordnername unter profile/, Default "praxis">
# ANSAGE_TTS=piper|say|mistral|datei
# ANSAGE_TTS=piper|say|mistral
# ANSAGE_STIMME=<Piper-Modellname, say-Stimme, oder Voxtral-Stimmen-Slug>
# ANSAGE_EMOTION=<Sprecher-ID, nur bei thorsten_emotional: 0=amused,
# 1=angry, 2=disgusted, 3=drunk, 4=neutral, 5=sleepy,
@ -36,10 +32,6 @@ if [ -f .env ]; then
done < .env
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}"
STIMME="${ANSAGE_STIMME:-de_DE-thorsten_emotional-medium}"
EMOTION="${ANSAGE_EMOTION:-0}"
@ -58,28 +50,10 @@ fi
TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT
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)
MODELL="$STIMMEN_ORDNER/$STIMME.onnx"
[ -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" \
-f "$TMP/sprache_roh.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_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"
BESCHREIBUNG="say/$SAY_STIMME"
;;
@ -95,7 +69,7 @@ case "$TTS" in
MISTRAL_STIMME="${ANSAGE_STIMME:-en_paul_neutral}"
MISTRAL_MODELL="${MISTRAL_TTS_MODEL:-voxtral-mini-tts-latest}"
[ -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 '
import json, os, sys
print(json.dumps({

View file

@ -18,20 +18,6 @@ if [ -f .env ]; then
done < .env
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
melde() { printf ' %-38s %s\n' "$1" "$2"; }
@ -113,21 +99,15 @@ if [ "${status:-}" = "UP" ]; then
else
# nice 10: Transkription und Kategorisierung dürfen warten, ein laufendes
# 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)"
fi
if pgrep -f "[p]ipe.server" >/dev/null; then
echo "Leitstand laeuft bereits."
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
tail -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)"
head -1 telefon/leitstand.log
fi
sleep 2
echo