"""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 html import json import re import sys import time from datetime import datetime from functools import lru_cache from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, unquote, urlparse from . import bewertung, config, dashboard, dienste, export, kategorien, modellwahl, protokoll, stapel, testzugang 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"), "rueckrufnummer": e.get("rueckrufnummer"), "email": e.get("email"), "email_korrigiert": bool(e.get("email_vermutete_korrektur")), "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"), "bewertung": (bewertung.lies(config.ABLAGE_LOKAL / ordner) or {}).get("bewertung") if ordner else None, }) return liste def archiviere_tag() -> int: """Markiert alle 'Erledigt'-Anrufe als archiviert (simuliert - es wird nichts geloescht, nur vom Board geraeumt). Bewertete Anrufe werden dabei zusaetzlich als Trainingsdaten exportiert, siehe pipe.export. Gibt die Anzahl archivierter Anrufe 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:] # Bei Testzugang (kein Passwort, oeffentlicher Link) werden Rufnummern nicht # nur im Frontend per CSS geblurred (das verdeckt nur die Anzeige - im # Seitenquelltext/DevTools/Netzwerk-Tab stehen sie trotzdem echt drin), # sondern schon hier serverseitig ersetzt. Die echten Werte verlassen den # Server in diesem Modus gar nicht erst. _MASKE = "•••• •••• •••" def _maskiere_anrufe(anrufe_liste: list[dict]) -> list[dict]: for a in anrufe_liste: if a.get("nummer"): a["nummer"] = _MASKE if a.get("rueckrufnummer"): a["rueckrufnummer"] = _MASKE return anrufe_liste def _maskiere_verlauf(eintraege: list[dict]) -> list[dict]: for e in eintraege: if e.get("text"): e["text"] = re.sub(r"\d{9,}", "•••••••••", e["text"]) return eintraege 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. Zusätzlich akzeptiert: ein gültiges signiertes Testzugangs-Token (siehe pipe.testzugang) - per ?zugang=... in der URL (so kommt es aus dilles-agent.php's "agent-one test"-Kommando) oder per zugang-Cookie für Folgeanfragen (fetch()-Aufrufe der Seite tragen die URL nicht mit). Ein frisches URL-Token setzt das Cookie für den Rest der Stunde. """ self._zusatz_header: list[tuple[str, str]] = [] self._via_token = False # True nur bei Zugang per Testzugangs-Token, nicht Passwort self._nutzer: str | None = None # wer angemeldet ist - fuer den Verlauf (wer hat was verschoben) konten = config.leitstand_konten() if not konten: return True token = parse_qs(urlparse(self.path).query).get("zugang", [None])[0] if token and testzugang.gueltig(token): rest_s = max(0, int(token.split(".", 1)[0]) - int(time.time())) self._zusatz_header.append( ("Set-Cookie", f"zugang={token}; Max-Age={rest_s}; Path=/; HttpOnly; Secure; SameSite=Lax")) self._via_token = True self._nutzer = "Testzugang" return True cookie_wert = SimpleCookie(self.headers.get("Cookie", "")).get("zugang") if cookie_wert and testzugang.gueltig(cookie_wert.value): self._via_token = True self._nutzer = "Testzugang" 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. erwartet = konten.get(nutzer) if erwartet is None or not hmac.compare_digest(wort, erwartet): return False self._nutzer = nutzer return True 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") for schluessel, wert in getattr(self, "_zusatz_header", []): self.send_header(schluessel, wert) 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": # Kopie, nicht das gecachte dienste.status()-Objekt direkt # mutieren - der ThreadingHTTPServer bedient mehrere Nutzer # gleichzeitig, sonst koennte der Nutzername der einen Anfrage in # der Antwort einer anderen landen. daten = dict(dienste.status()) # Testzugang (per "agent-one test"-Link) -> Sicher-Modus im # Frontend erzwingen, siehe sicherSetzen() im JS. daten["testzugang"] = getattr(self, "_via_token", False) daten["nutzer"] = self._nutzer self._json(daten) elif pfad == "/api/anrufe": liste = anrufe() if getattr(self, "_via_token", False): liste = _maskiere_anrufe(liste) self._json({"stapel": stapel.erlaubt(), "anrufe": liste}) elif pfad == "/api/verlauf": eintraege = verlauf(150) if getattr(self, "_via_token", False): eintraege = _maskiere_verlauf(eintraege) self._json(eintraege) elif pfad == "/api/modell": self._json({"aktuell": modellwahl.aktuelle_id(), "optionen": modellwahl.auswahl()}) 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) _, export_anzahl = export.exportiere() if export_anzahl: protokoll.schreibe("export", f"{export_anzahl} bewertete Anrufe exportiert") # Verlauf-Log sichern + leeren, fuer eine frische Ansicht ab jetzt. # audit.log (dauerhaft, unveraenderlich) ist davon nicht betroffen. log_archiv = protokoll.archiviere_und_leere() if log_archiv: protokoll.schreibe("system", f"Verlauf archiviert nach {log_archiv.name} und geleert") self._json({"ok": True, "anzahl": anzahl, "simuliert": True, "export_anzahl": export_anzahl, "log_archiviert": log_archiv.name if log_archiv else None}) return if pfad == "/api/export": ziel, anzahl = export.exportiere() protokoll.schreibe("export", f"{anzahl} bewertete Anrufe exportiert nach {ziel.name}") self._json({"ok": True, "anzahl": anzahl, "datei": ziel.name}) return if pfad == "/api/bewertung": try: laenge = int(self.headers.get("Content-Length") or 0) wunsch = json.loads(self.rfile.read(min(laenge, 10_000)).decode("utf-8")) kennung, wert = wunsch["id"], wunsch["bewertung"] korrektur = wunsch.get("korrektur") 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 bewertung.setze(ordner, wert, korrektur): self._json({"ok": False, "fehler": f"ungültige Bewertung: {wert}"}) return protokoll.schreibe("bewertung", f"{ordner.name} → {wert}" + (f" (Korrektur: {korrektur})" if korrektur else "")) self._json({"ok": True}) return if pfad == "/api/modell": try: laenge = int(self.headers.get("Content-Length") or 0) kennung = json.loads(self.rfile.read(min(laenge, 10_000)).decode("utf-8"))["id"] except Exception: self._json({"ok": False, "fehler": "ungültige Anfrage"}) return if not modellwahl.setze(kennung): self._json({"ok": False, "fehler": f"unbekanntes Modell: {kennung}"}) return protokoll.schreibe("system", f"Kategorisierungs-Modell → {kennung}") self._json({"ok": 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 zusatz = f" (von {self._nutzer})" if self._nutzer else "" protokoll.schreibe("stapel", f"{ordner.name} → {ziel}{zusatz}", nutzer=self._nutzer) self._json({"ok": True}) def _audio(self, rest: str) -> None: """Liefert eine Aufnahme aus der Ablage - nur von dort. Muss HTTP-Range unterstuetzen: der