-
"""
diff --git a/pipe/deck.py b/pipe/deck.py
index c1b117e..394e302 100644
--- a/pipe/deck.py
+++ b/pipe/deck.py
@@ -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)
diff --git a/pipe/dienste.py b/pipe/dienste.py
deleted file mode 100644
index c17342f..0000000
--- a/pipe/dienste.py
+++ /dev/null
@@ -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
diff --git a/pipe/kategorien.py b/pipe/kategorien.py
deleted file mode 100644
index 4365a04..0000000
--- a/pipe/kategorien.py
+++ /dev/null
@@ -1,48 +0,0 @@
-"""Kategorien-Schema des aktiven Branchen-Profils (profile/Anruf-Übersicht · {html.escape(kategorien.PROFIL_NAME)}
+Anruf-Übersicht
{len(anrufe)} Anrufe · {zusammenfassung or '—'} · lokal erzeugt aus der Ablage
{karten}
-
+
…
-
-
@@ -657,8 +616,9 @@ 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
deleted file mode 100644
index 381c60b..0000000
--- a/pipe/testzugang.py
+++ /dev/null
@@ -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: "Logpy:AgentOne
Anruf-Leitstand · __PROFIL_NAME__Logpy:AgentOne
Anruf-LeitstandDienste
@@ -629,14 +596,6 @@ 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