"""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 sys import time from datetime import datetime 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, 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 if not config.LEITSTAND_PASS: 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 return True cookie_wert = SimpleCookie(self.headers.get("Cookie", "")).get("zugang") if cookie_wert and testzugang.gueltig(cookie_wert.value): self._via_token = True 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") 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": daten = dienste.status() # Testzugang (per "agent-one test"-Link) -> Sicher-Modus im # Frontend erzwingen, siehe sicherSetzen() im JS. daten["testzugang"] = getattr(self, "_via_token", False) 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 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