mirror of
https://github.com/Jeuners/Logpy-AgentOne.git
synced 2026-09-09 15:02:33 +02:00
1a1280e79d
Nimmt Anrufe einer Hausarztpraxis entgegen, transkribiert sie lokal, kategorisiert das Anliegen und legt es strukturiert ab. Keine Patientendaten verlassen den Rechner: Spracherkennung, Sprachmodell und Sprachsynthese laufen lokal, im Datenpfad steht kein Cloud-Dienst. Aufbau: - FreeSWITCH nimmt über einen SIP-Trunk an, spielt die Ansage und nimmt auf. - pipe.watch erkennt fertige Aufnahmen und schiebt sie durch die Pipe. - whisper.cpp transkribiert, ein lokales Sprachmodell (Ollama) ordnet in Kategorie und Dringlichkeit ein — mit erzwungenem JSON-Schema. - Ablage lokal, optional Nextcloud (WebDAV) und ein Deck-Board. - Leitstand (pipe.server): Board mit Eingang/In Bearbeitung/Rückfragen/Erledigt, Aufnahmen zum Anhören, Verlauf und Ampeln für alle beteiligten Dienste. Zugangsdaten und Anrufdaten liegen außerhalb des Repositorys (.env, ablage/, telefon/). Konfiguration siehe .env.example, Einrichtung siehe README.
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Gemeinsames Ereignisprotokoll der Pipe.
|
|
|
|
Jede Zeile ist ein JSON-Objekt (JSONL) — maschinenlesbar für das Dashboard und
|
|
trotzdem mit ``tail -f`` lesbar. Bewusst schlank: anhängen, nie sperren, nie
|
|
scheitern. Ein kaputtes Protokoll darf keinen Anruf kosten.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from . import config
|
|
|
|
DATEI = config.TELEFON / "verlauf.log"
|
|
MAX_ZEILEN = 2000 # darüber wird beim Schreiben gekürzt
|
|
|
|
|
|
def schreibe(art: str, text: str, **felder) -> None:
|
|
"""Hängt ein Ereignis an. Wirft nie."""
|
|
try:
|
|
DATEI.parent.mkdir(parents=True, exist_ok=True)
|
|
eintrag = {"zeit": datetime.now().isoformat(timespec="seconds"),
|
|
"art": art, "text": text, **felder}
|
|
with DATEI.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(eintrag, ensure_ascii=False) + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def lies(anzahl: int = 200) -> list[dict]:
|
|
"""Liest die letzten Ereignisse. Defekte Zeilen werden übersprungen."""
|
|
if not DATEI.is_file():
|
|
return []
|
|
try:
|
|
zeilen = DATEI.read_text(encoding="utf-8").splitlines()[-anzahl:]
|
|
except Exception:
|
|
return []
|
|
eintraege = []
|
|
for z in zeilen:
|
|
try:
|
|
eintraege.append(json.loads(z))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return eintraege
|
|
|
|
|
|
def kuerze() -> None:
|
|
"""Hält das Protokoll auf MAX_ZEILEN. Wirft nie."""
|
|
try:
|
|
if not DATEI.is_file():
|
|
return
|
|
zeilen = DATEI.read_text(encoding="utf-8").splitlines()
|
|
if len(zeilen) > MAX_ZEILEN:
|
|
DATEI.write_text("\n".join(zeilen[-MAX_ZEILEN:]) + "\n", encoding="utf-8")
|
|
except Exception:
|
|
pass
|