mirror of
https://github.com/Jeuners/Logpy-AgentOne.git
synced 2026-09-09 15:02:33 +02:00
af67ee89e3
Gegenstueck zum "agent-one test"-Kommando in Alice (dillenberg.net,
dilles-agent.php). Kein Live-Aufruf zwischen den Systemen noetig - beide
teilen sich TESTZUGANG_SECRET, das Token ist ein HMAC-signierter
Ablauf-Zeitstempel ("ablauf.signatur"), stateless geprueft.
pipe/testzugang.py: gueltig(token) prueft Signatur (hmac.compare_digest)
und Ablauf. _angemeldet() in server.py akzeptiert das Token zusaetzlich zum
Passwort - per ?zugang=... in der URL oder per zugang-Cookie fuer
Folgeanfragen (fetch()-Aufrufe der Seite tragen die URL nicht mit). Ein
frisches URL-Token setzt das Cookie (HttpOnly, Secure, SameSite=Lax) fuer
den Rest der Stunde.
End-to-End getestet: PHP-generiertes Token wird von Python akzeptiert
(Cross-Language-HMAC-Kompatibilitaet bestaetigt), ungueltige Signatur und
abgelaufenes Token werden korrekt abgelehnt (401), Cookie-Folgeanfrage ohne
erneutes Token/Passwort funktioniert.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vtvcyNkapuDh583643hNs
177 lines
8.1 KiB
Python
177 lines
8.1 KiB
Python
"""Zentrale Konfiguration der Anruf-Pipeline.
|
|
|
|
Alle Werte kommen aus Umgebungsvariablen (optional aus einer .env-Datei im
|
|
Projektwurzelverzeichnis). Keine Geheimnisse im Code — die Nextcloud-Zugangs-
|
|
daten werden erst gesetzt, wenn sie vorliegen; bis dahin bleibt die Ablage rein
|
|
lokal.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
WURZEL = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _lade_dotenv(pfad: Path) -> None:
|
|
"""Minimaler .env-Loader ohne externe Abhängigkeit."""
|
|
if not pfad.is_file():
|
|
return
|
|
for zeile in pfad.read_text(encoding="utf-8").splitlines():
|
|
zeile = zeile.strip()
|
|
if not zeile or zeile.startswith("#") or "=" not in zeile:
|
|
continue
|
|
schluessel, _, wert = zeile.partition("=")
|
|
schluessel = schluessel.strip()
|
|
wert = wert.strip().strip('"').strip("'")
|
|
os.environ.setdefault(schluessel, wert)
|
|
|
|
|
|
_lade_dotenv(WURZEL / ".env")
|
|
|
|
|
|
def _bool(name: str, standard: bool) -> bool:
|
|
wert = os.environ.get(name)
|
|
if wert is None:
|
|
return standard
|
|
return wert.strip().lower() in {"1", "true", "yes", "ja", "on"}
|
|
|
|
|
|
# --- Spracherkennung (lokal) -------------------------------------------------
|
|
# Backend: "whispercpp" (Apple-GPU, nutzt vorhandenes ggml-Modell, kein Download)
|
|
# "faster" (faster-whisper / CTranslate2, lädt Modell bei Bedarf)
|
|
STT_BACKEND = os.environ.get("STT_BACKEND", "whispercpp").strip().lower()
|
|
|
|
# Feste Sprache erzwingen (z. B. "de") oder "auto" für Auto-Erkennung.
|
|
WHISPER_SPRACHE = os.environ.get("WHISPER_SPRACHE", "de").strip() or "auto"
|
|
|
|
# whisper.cpp
|
|
WHISPERCPP_BIN = os.environ.get("WHISPERCPP_BIN", "whisper-cli")
|
|
WHISPERCPP_MODELL = os.environ.get(
|
|
"WHISPERCPP_MODELL", str(Path.home() / "whisper-models" / "ggml-large-v3-turbo.bin")
|
|
)
|
|
WHISPER_THREADS = os.environ.get("WHISPER_THREADS", "8")
|
|
|
|
# Kontext für die Spracherkennung. Whisper erkennt Namen, Rufnummern und
|
|
# Praxis-Vokabular deutlich zuverlässiger, wenn es weiß, worum es geht - bei
|
|
# der schlechten Tonqualität einer Telefonleitung macht das den Unterschied.
|
|
WHISPER_PROMPT = os.environ.get(
|
|
"WHISPER_PROMPT",
|
|
"Anruf auf dem Anrufbeantworter einer Hausarztpraxis. Der Anrufer nennt "
|
|
"seinen Namen, sein Anliegen und eine Rückrufnummer. Häufig geht es um "
|
|
"Termin, Rezept, Überweisung, Befund, Krankschreibung oder Schmerzen. "
|
|
"Nennt der Anrufer eine E-Mail-Adresse, wird sie mit @-Zeichen "
|
|
"geschrieben, zum Beispiel max.mueller@example.de.",
|
|
).strip()
|
|
|
|
# faster-whisper (nur bei STT_BACKEND=faster)
|
|
WHISPER_MODELL = os.environ.get("WHISPER_MODELL", "medium")
|
|
WHISPER_COMPUTE = os.environ.get("WHISPER_COMPUTE", "int8")
|
|
|
|
# --- Kategorisierung -----------------------------------------------------
|
|
# Welches Modell aktiv ist, waehlt der Leitstand per Dropdown (pipe.modellwahl,
|
|
# geteilte Zustandsdatei telefon/modell.json) - kein Neustart noetig. Hier nur
|
|
# die Bausteine dafuer: Ollama-Zugang (lokal, Default-Auswahl) und der
|
|
# OpenRouter-Zugang fuers Testen staerkerer Cloud-Modelle. Patientendaten
|
|
# (Transkript) verlassen bei OpenRouter den Rechner - nur zu Testzwecken,
|
|
# nicht im Praxisbetrieb.
|
|
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
|
|
OLLAMA_MODELL = os.environ.get("OLLAMA_MODELL", "qwen3:8b")
|
|
|
|
OPENROUTER_KEY = os.environ.get("OPENROUTER_KEY", "")
|
|
# Optional: eigenes Testmodell zusaetzlich zu den zwei fest im Dropdown
|
|
# hinterlegten (siehe pipe/modellwahl.py) - leer lassen wenn nicht gebraucht.
|
|
OPENROUTER_MODELL = os.environ.get("OPENROUTER_MODELL", "")
|
|
|
|
PROMPT_DATEI = WURZEL / "prompts" / os.environ.get(
|
|
"PROMPT_DATEI", "categorize_de.txt"
|
|
)
|
|
|
|
# --- Ablage ------------------------------------------------------------------
|
|
ABLAGE_LOKAL = Path(os.environ.get("ABLAGE_LOKAL", str(WURZEL / "ablage")))
|
|
|
|
# --- Leitstand (Weboberfläche) ----------------------------------------------
|
|
# Log der Telefonanlage - der Leitstand liest daraus die Anruf-Ereignisse.
|
|
FREESWITCH_LOG = Path(os.environ.get(
|
|
"FREESWITCH_LOG", "/opt/homebrew/var/log/freeswitch/freeswitch.log"))
|
|
|
|
# Event-Socket-Port - muss mit autoload_configs/event_socket.conf.xml
|
|
# uebereinstimmen. Der Homebrew-Default variiert je nach Installation.
|
|
FS_PORT = os.environ.get("FS_PORT", "8021")
|
|
|
|
LEITSTAND_HOST = os.environ.get("LEITSTAND_HOST", "127.0.0.1")
|
|
LEITSTAND_PORT = int(os.environ.get("LEITSTAND_PORT", "8088"))
|
|
# Zugangsschutz. Pflicht, sobald der Leitstand nicht nur lokal erreichbar ist -
|
|
# auf den Seiten stehen Transkripte und Aufnahmen von Patienten.
|
|
LEITSTAND_USER = os.environ.get("LEITSTAND_USER", "praxis")
|
|
LEITSTAND_PASS = os.environ.get("LEITSTAND_PASS", "")
|
|
|
|
# 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:
|
|
return LEITSTAND_HOST not in {"127.0.0.1", "localhost", "::1"}
|
|
|
|
|
|
# --- Telefonie ---------------------------------------------------------------
|
|
# Die Anlage legt Aufnahmen im Eingang ab; der Watcher (pipe.watch) räumt sie
|
|
# nach der Verarbeitung weg.
|
|
TELEFON = Path(os.environ.get("TELEFON_ORDNER", str(WURZEL / "telefon")))
|
|
TELEFON_EINGANG = TELEFON / "eingang"
|
|
TELEFON_VERARBEITET = TELEFON / "verarbeitet"
|
|
TELEFON_FEHLER = TELEFON / "fehler"
|
|
|
|
# SIP-Zugang der Telefonanlage (Werte in .env, nie im Code).
|
|
SIP_USER = os.environ.get("SIP_USER", "")
|
|
SIP_DOMAIN = os.environ.get("SIP_DOMAIN", "")
|
|
SIP_PASS = os.environ.get("SIP_PASS", "")
|
|
|
|
# Nextcloud (steckbar): nur aktiv, wenn URL + Nutzer + Passwort gesetzt sind.
|
|
NEXTCLOUD_URL = os.environ.get("NEXTCLOUD_URL", "").rstrip("/")
|
|
NEXTCLOUD_USER = os.environ.get("NEXTCLOUD_USER", "") # Login (kann E-Mail sein)
|
|
NEXTCLOUD_PASS = os.environ.get("NEXTCLOUD_PASS", "") # App-Passwort
|
|
# WebDAV-Pfad braucht die interne User-ID (weicht bei E-Mail-Login ab).
|
|
NEXTCLOUD_USERID = os.environ.get("NEXTCLOUD_USERID", "") or NEXTCLOUD_USER
|
|
NEXTCLOUD_ORDNER = os.environ.get("NEXTCLOUD_ORDNER", "Anrufe").strip("/")
|
|
# Adressbuch (CardDAV) fuer den Kontakt-Abgleich bei eingehenden Anrufen.
|
|
# Nextclouds Standard-Adressbuch heisst "contacts"; eigene Adressbuecher
|
|
# tragen ihre URI (z. B. "hgod").
|
|
NEXTCLOUD_ADRESSBUCH = os.environ.get("NEXTCLOUD_ADRESSBUCH", "contacts")
|
|
# Getrennter Schalter fuer den Datei-Upload: Zugangsdaten allein reichen fuer
|
|
# den Kontakt-Abgleich (kontakte.py) - wer nur das will, ohne dass Aufnahmen
|
|
# und Transkripte zusaetzlich nach Nextcloud hochgeladen werden, setzt hier 0.
|
|
# Default an, damit sich am bisherigen Verhalten (Zugangsdaten -> Upload) fuer
|
|
# bestehende Installationen nichts aendert.
|
|
NEXTCLOUD_UPLOAD = os.environ.get("NEXTCLOUD_UPLOAD", "1").strip().lower() in {
|
|
"1", "true", "yes", "ja", "on"
|
|
}
|
|
|
|
# Deck-Triage-Board (optional): pro Anruf eine Karte, Stapel = Kategorie.
|
|
NEXTCLOUD_DECK = os.environ.get("NEXTCLOUD_DECK", "").strip().lower() in {
|
|
"1", "true", "yes", "ja", "on"
|
|
}
|
|
DECK_BOARD = os.environ.get("DECK_BOARD", "Anrufe")
|
|
# Stapel des Boards = Arbeitsablauf, nicht Kategorie. Neue Anrufe landen immer
|
|
# im ersten Stapel; die Kategorie steht auf der Karte.
|
|
DECK_STAPEL = [s.strip() for s in os.environ.get(
|
|
"DECK_STAPEL", "Eingang,In Bearbeitung,Rückfragen,Erledigt").split(",") if s.strip()]
|
|
# Nutzer, die das Board sehen sollen (kommagetrennt). Ohne Freigabe sieht nur
|
|
# das Konto der Pipe die Karten - die Praxis schaut in ein leeres Deck.
|
|
DECK_TEILEN = [n.strip() for n in os.environ.get("DECK_TEILEN", "").split(",") if n.strip()]
|
|
|
|
|
|
def nextcloud_aktiv() -> bool:
|
|
return bool(NEXTCLOUD_URL and NEXTCLOUD_USER and NEXTCLOUD_PASS)
|
|
|
|
|
|
# --- 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
|
|
# wird automatisch geloescht). Greift nur fuer Anrufe im letzten Stapel
|
|
# (DECK_STAPEL[-1], per Default "Erledigt") - unbearbeitete Anrufe bleiben
|
|
# unangetastet, egal wie alt.
|
|
LOESCHFRIST_TAGE = int(os.environ.get("LOESCHFRIST_TAGE", "0") or "0")
|