-
"""
diff --git a/pipe/deck.py b/pipe/deck.py
index 394e302..c1b117e 100644
--- a/pipe/deck.py
+++ b/pipe/deck.py
@@ -18,24 +18,15 @@ import urllib.error
import urllib.request
from functools import lru_cache
-from . import config
+from . import config, kategorien
_API = "/index.php/apps/deck/api/v1.1"
-# Farben (6-stelliges Hex ohne #) je Dringlichkeit.
-DRINGLICHKEIT_FARBE = {
- "niedrig": "31CC7C", # grün
- "normal": "F1DB50", # gelb
- "hoch": "E0A339", # orange
- "notfall": "E9322D", # rot
-}
+# Farben je Dringlichkeit - siehe pipe/kategorien.py.
+DRINGLICHKEIT_FARBE = kategorien.DRINGLICHKEIT_FARBE
# Lesbare Namen der Kategorien für den Kartentitel - der Stapel sagt jetzt,
# wie weit die Bearbeitung ist, nicht mehr worum es geht.
-KATEGORIE_LABEL = {
- "termin": "Termin", "rezept": "Rezept", "ueberweisung": "Überweisung",
- "befund": "Befund", "verwaltung": "Verwaltung", "beschwerden": "Beschwerden", "notfall": "Notfall",
- "rueckruf": "Rückruf", "sonstiges": "Sonstiges",
-}
+KATEGORIE_LABEL = kategorien.KATEGORIE_LABEL
@lru_cache(maxsize=1)
diff --git a/pipe/dienste.py b/pipe/dienste.py
new file mode 100644
index 0000000..c17342f
--- /dev/null
+++ b/pipe/dienste.py
@@ -0,0 +1,96 @@
+"""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
diff --git a/pipe/kategorien.py b/pipe/kategorien.py
new file mode 100644
index 0000000..4365a04
--- /dev/null
+++ b/pipe/kategorien.py
@@ -0,0 +1,48 @@
+"""Kategorien-Schema des aktiven Branchen-Profils (profile/Anruf-Übersicht
+Anruf-Übersicht · {html.escape(kategorien.PROFIL_NAME)}
{len(anrufe)} Anrufe · {zusammenfassung or '—'} · lokal erzeugt aus der Ablage
{karten}
-
+
…
+
+
@@ -616,9 +657,8 @@ color:var(--gut);border-radius:999px;padding:1px 8px;font-size:11px;font-weight:
"""
+@lru_cache(maxsize=1)
+def _seite() -> str:
+ """SEITE mit den Kategorie-Labels des aktiven Profils (pipe.kategorien).
+
+ Fest pro Prozesslauf (Profil wechselt nur per .env + Neustart, wie die
+ Kategorien selbst) - deshalb einmalig gecacht statt bei jeder Anfrage neu
+ zusammengesetzt.
+ """
+ seite = SEITE.replace("__KAT_JSON__", json.dumps(kategorien.KATEGORIE_LABEL, ensure_ascii=False))
+ farben = {k: f"#{v}" for k, v in kategorien.DRINGLICHKEIT_FARBE.items()}
+ seite = seite.replace("__FARBE_JSON__", json.dumps(farben, ensure_ascii=False))
+ return seite.replace("__PROFIL_NAME__", html.escape(kategorien.PROFIL_NAME))
+
+
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description="Live-Dashboard der Anruf-Pipe.")
p.add_argument("--port", type=int, default=config.LEITSTAND_PORT)
diff --git a/pipe/testzugang.py b/pipe/testzugang.py
new file mode 100644
index 0000000..381c60b
--- /dev/null
+++ b/pipe/testzugang.py
@@ -0,0 +1,35 @@
+"""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: "Logpy:AgentOne
Anruf-LeitstandLogpy:AgentOne
Anruf-Leitstand · __PROFIL_NAME__Dienste
@@ -596,6 +629,14 @@ color:var(--gut);border-radius:999px;padding:1px 8px;font-size:11px;font-weight:
1.36.87 2.29 0 3.28-2 4-3.9 4.22.31.27.58.79.58 1.6 0 1.15-.01 2.08-.01 2.36 0 .21.15.46.55.38A8.01 8.01 0 0 0 16 8c0-4.42-3.58-8-8-8z"/>
Projekt auf GitHub
+
+
+ LinkedIn
+
+ 📞 +49 152 23062462
+ ☁️ Nextcloud
☕ Unterstützen