Explorer
/opt/struktur/telefon-agent/scripts/run_twilio_test_server.py
← Zurück ↓ Download
#!/usr/bin/env python3
"""Lokaler Twilio-TEST-Server (Phase 3) — startet NUR mit vollständiger Konfiguration.

Bindet ausschließlich 127.0.0.1:5090. Öffentliche Erreichbarkeit (nginx/Tunnel)
ist bewusst NICHT enthalten und braucht eine separate Freigabe.
Kein Recording, keine Audio-Speicherung, nur freigegebene Templates.
"""

import pathlib
import sys
import urllib.parse
from http.server import BaseHTTPRequestHandler, HTTPServer

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

import os  # noqa: E402

from telefon_agent import twilio_gateway as gw  # noqa: E402

BIND = ("127.0.0.1", 5090)


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        params = {k: v[0] for k, v in urllib.parse.parse_qs(
            self.rfile.read(length).decode()).items()}

        # Twilio-Signaturprüfung (Pflicht im Testbetrieb)
        url = os.environ["PUBLIC_BASE_URL"].rstrip("/") + self.path
        sig = self.headers.get("X-Twilio-Signature", "")
        if not gw.validate_signature(url, params, sig, os.environ["TWILIO_AUTH_TOKEN"]):
            self.send_response(403)
            self.end_headers()
            return

        call_sid = params.get("CallSid", "")
        if self.path == gw.ROUTE_INBOUND:
            body, _ = gw.handle_inbound(call_sid, params.get("From", ""))
        elif self.path == gw.ROUTE_STEP:
            body, _ = gw.handle_step(call_sid, params.get("Digits", ""),
                                     params.get("SpeechResult", ""))
        elif self.path == gw.ROUTE_STATUS:
            body = gw.handle_status(call_sid, params.get("CallStatus", ""))
        else:
            self.send_response(404)
            self.end_headers()
            return
        data = body.encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/xml; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, fmt, *args):
        print("[twilio-test]", fmt % args)


def main():
    fehlend = gw.missing_config()
    if fehlend:
        print("ABBRUCH: Konfiguration unvollständig. Fehlende Umgebungsvariablen:")
        for k in fehlend:
            print(" -", k)
        print("Werte in /opt/struktur/telefon-agent/.env pflegen (chmod 600, gitignored).")
        return 2
    print(f"Twilio-TEST-Server auf http://{BIND[0]}:{BIND[1]} — Routen: "
          f"{gw.ROUTE_INBOUND}, {gw.ROUTE_STEP}, {gw.ROUTE_STATUS}")
    print("Nur Testnummer. Kein Recording. Öffentliche Erreichbarkeit erfordert separate Freigabe.")
    HTTPServer(BIND, Handler).serve_forever()


if __name__ == "__main__":
    sys.exit(main())