#!/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())