"""Live-Dashboard: zeigt Anrufe, Systemzustand und das laufende Protokoll. python3 -m pipe.server [--port 8088] Rein lokal, stdlib. Liest die Ablage, das Ereignisprotokoll und den Zustand der beteiligten Dienste. Die Seite fragt alle paar Sekunden nach — kein WebSocket, kein Framework, nichts, was im Praxisbetrieb kaputtgehen kann. """ from __future__ import annotations import argparse import base64 import hmac import json import re import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlparse from . import config, dashboard, protokoll, stapel, 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 def anrufe() -> list[dict]: liste = [] for d in dashboard._anrufe(): e = d["auswertung"] zeit = d["empfangen"] ordner = None try: z = datetime.fromisoformat(zeit) tag = f"{z:%Y-%m-%d}" for kandidat in (config.ABLAGE_LOKAL / tag).iterdir(): if kandidat.is_dir() and kandidat.name.startswith(f"{z:%H-%M-%S}"): ordner = f"{tag}/{kandidat.name}" break except Exception: pass if ordner and stapel.ist_archiviert(config.ABLAGE_LOKAL / ordner): continue # archiviert (simuliert) - verschwindet vom Board audio = None if ordner: for endung in (".wav", ".mp3", ".opus", ".m4a"): if (config.ABLAGE_LOKAL / ordner / f"aufnahme{endung}").is_file(): audio = f"/audio/{ordner}/aufnahme{endung}" break liste.append({ "id": ordner or "", "stapel": stapel.lies(config.ABLAGE_LOKAL / ordner) if ordner else stapel.erlaubt()[0], "empfangen": zeit, "kategorie": e["kategorie"], "dringlichkeit": e["dringlichkeit"], "name": e.get("anrufer_name"), "nummer": d.get("anrufer_nummer") or e.get("rueckrufnummer"), "rueckruf": bool(e.get("rueckruf_gewuenscht")), "anliegen": e.get("anliegen_kurz") or "", "stichworte": e.get("stichworte") or [], "transkript": d.get("transkript") or "", "audio": audio, "kontakt": (d.get("bekannter_kontakt") or {}).get("name"), }) return liste def archiviere_tag() -> int: """Markiert alle 'Erledigt'-Anrufe als archiviert (simuliert - es wird nichts exportiert oder geloescht, nur vom Board geraeumt). Gibt die Anzahl zurueck.""" letzter = stapel.erlaubt()[-1] anzahl = 0 for a in anrufe(): if a["stapel"] != letzter or not a["id"]: continue ordner = stapel.ordner_zu(a["id"]) if ordner and stapel.archiviere(ordner): anzahl += 1 return anzahl FS_LOG = config.FREESWITCH_LOG _ZEIT = re.compile(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}") # Was im FreeSWITCH-Log für die Praxis interessant ist - und wie es heißen soll. _FS_MUSTER = [ ("New Channel sofia/external/", "anruf", "Anruf von {nummer}"), ("has been answered", "anruf", "abgenommen, Ansage läuft"), ("Hangup sofia/external/", "anruf", "Anrufer hat aufgelegt"), ("Failed Registration", "fehler", "Telefon: Registrierung fehlgeschlagen"), ("Ping failed", "fehler", "Telefon: Provider antwortet nicht"), ] def _telefon_ereignisse(anzahl: int = 60) -> list[dict]: """Liest Anruf-Ereignisse aus dem FreeSWITCH-Log. So sieht man den Anruf schon beim Klingeln, nicht erst wenn die fertige Aufnahme durch die Pipe gelaufen ist. """ if not FS_LOG.is_file(): return [] try: with FS_LOG.open("rb") as f: # nur das Ende lesen f.seek(0, 2) f.seek(max(0, f.tell() - 300_000)) zeilen = f.read().decode("utf-8", "replace").splitlines() except Exception: return [] ereignisse = [] for zeile in zeilen[-4000:]: for schnipsel, art, vorlage in _FS_MUSTER: if schnipsel not in zeile: continue # Bei Kanal-Ereignissen steht die Channel-UUID vor dem Zeitstempel, # die Position ist also nicht verlässlich - daher per Muster suchen. treffer = _ZEIT.search(zeile) zeit = treffer.group(0) if treffer else "" nummer = "" if "sofia/external/" in zeile: rest = zeile.split("sofia/external/", 1)[1] nummer = rest.split("@", 1)[0].strip() ereignisse.append({ "zeit": zeit.replace(" ", "T"), "art": art, "text": vorlage.format(nummer=nummer or "unbekannt"), }) break return ereignisse[-anzahl:] def verlauf(anzahl: int = 150) -> list[dict]: """Pipe-Protokoll und Telefonie-Ereignisse, zeitlich zusammengeführt.""" alle = protokoll.lies(anzahl) + _telefon_ereignisse() alle.sort(key=lambda e: e.get("zeit") or "") return alle[-anzahl:] class Handler(BaseHTTPRequestHandler): def log_message(self, *_): # kein Zugriffslog auf der Konsole pass def _angemeldet(self) -> bool: """Prüft die Zugangsdaten, sofern eines gesetzt ist. Ohne Passwort bleibt der Leitstand offen - das ist nur zulässig, wenn er an 127.0.0.1 hängt; darauf besteht main() beim Start. """ if not config.LEITSTAND_PASS: return True kopf = self.headers.get("Authorization", "") if not kopf.startswith("Basic "): return False try: nutzer, _, wort = base64.b64decode(kopf[6:]).decode("utf-8").partition(":") except Exception: return False # compare_digest: Vergleichsdauer verrät nichts über das Passwort. return (hmac.compare_digest(nutzer, config.LEITSTAND_USER) and hmac.compare_digest(wort, config.LEITSTAND_PASS)) def _anmeldung_verlangen(self) -> None: self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Anruf-Leitstand"') self.send_header("Content-Length", "0") self.end_headers() def _sende(self, inhalt: bytes, typ: str, code: int = 200) -> None: self.send_response(code) self.send_header("Content-Type", typ) self.send_header("Content-Length", str(len(inhalt))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(inhalt) def _json(self, daten) -> None: self._sende(json.dumps(daten, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def do_GET(self) -> None: if not self._angemeldet(): self._anmeldung_verlangen() return pfad = unquote(urlparse(self.path).path) if pfad == "/": self._sende(SEITE.encode("utf-8"), "text/html; charset=utf-8") elif pfad == "/api/status": self._json(status()) elif pfad == "/api/anrufe": self._json({"stapel": stapel.erlaubt(), "anrufe": anrufe()}) elif pfad == "/api/verlauf": self._json(verlauf(150)) elif pfad.startswith("/audio/"): self._audio(pfad[len("/audio/"):]) else: self._sende(b"nicht gefunden", "text/plain; charset=utf-8", 404) def do_POST(self) -> None: if not self._angemeldet(): self._anmeldung_verlangen() return pfad = unquote(urlparse(self.path).path) if pfad == "/api/archivieren": anzahl = archiviere_tag() protokoll.schreibe("archiv", f"{anzahl} Anruf(e) archiviert (simuliert)", anzahl=anzahl) self._json({"ok": True, "anzahl": anzahl, "simuliert": True}) return if pfad != "/api/stapel": self._sende(b"nicht gefunden", "text/plain; charset=utf-8", 404) return try: laenge = int(self.headers.get("Content-Length") or 0) if laenge > 10_000: raise ValueError("Anfrage zu groß") wunsch = json.loads(self.rfile.read(laenge).decode("utf-8")) kennung, ziel = wunsch["id"], wunsch["stapel"] except Exception: self._json({"ok": False, "fehler": "ungültige Anfrage"}) return ordner = stapel.ordner_zu(kennung) if ordner is None: self._json({"ok": False, "fehler": "Anruf nicht gefunden"}) return if not stapel.setze(ordner, ziel): self._json({"ok": False, "fehler": f"unbekannter Stapel: {ziel}"}) return protokoll.schreibe("stapel", f"{ordner.name} → {ziel}") self._json({"ok": True}) def _audio(self, rest: str) -> None: """Liefert eine Aufnahme aus der Ablage - nur von dort. Muss HTTP-Range unterstuetzen: der