Two-way number masking

import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

FE = os.environ.get("FE", "918888800001")              # field agent real number
CUSTOMER = os.environ.get("CUSTOMER", "918888800002")  # customer real number
M1 = os.environ.get("M1", "919000000001")              # masking: FE -> Customer
M2 = os.environ.get("M2", "919000000002")              # masking: Customer -> FE


@dataclass
class Leg:
    masking_number: str   # the number that was dialled (the `To`)
    allowed_caller: str   # only this registered real number may use it
    destination: str      # the real number to bridge to
    caller_id_shown: str  # the masking number the destination should SEE


# Keyed by the number that was *dialled* (the `To`).
# Cross-mapped caller ID: dial M1 -> customer sees M2; dial M2 -> agent sees M1.
MAPPINGS: dict[str, Leg] = {
    M1: Leg(masking_number=M1, allowed_caller=FE, destination=CUSTOMER, caller_id_shown=M2),
    M2: Leg(masking_number=M2, allowed_caller=CUSTOMER, destination=FE, caller_id_shown=M1),
}


def _norm(number: str) -> str:
    """Digits only, last 10 - tolerant of '+', spaces, leading 0, country code."""
    digits = "".join(c for c in (number or "") if c.isdigit())
    return digits[-10:] if len(digits) >= 10 else digits


_NORM_MAPPINGS = {_norm(k): v for k, v in MAPPINGS.items()}


def resolve(to_number: str, from_number: str) -> Leg | None:
    """Return the Leg if `to_number` is a masking number AND `from_number` is its
    registered caller; otherwise None."""
    leg = _NORM_MAPPINGS.get(_norm(to_number))
    if leg is None or _norm(leg.allowed_caller) != _norm(from_number):
        return None
    return leg

from fastapi import FastAPI, Form, Response

import mappings
from vobiz_xml import dial, hangup, response, speak

app = FastAPI(title="Vobiz masking - two-way (client-hosted)")

XML = "application/xml"


@app.post("/answer")
async def answer(
    From: str = Form(...),
    To: str = Form(...),
    CallUUID: str = Form(default=""),
    Direction: str = Form(default=""),
    CallStatus: str = Form(default=""),
) -> Response:
    """Vobiz POSTs call details here when a masking number is dialled."""
    leg = mappings.resolve(to_number=To, from_number=From)
    if leg is None:
        xml = response(
            speak("Sorry, this number is not authorised for your account."),
            hangup(reason="rejected"),
        )
        return Response(content=xml, media_type=XML)

    xml = response(
        dial(leg.destination, caller_id=leg.caller_id_shown, timeout=30),
    )
    return Response(content=xml, media_type=XML)


@app.get("/health")
async def health() -> dict:
    return {"status": "ok", "strategy": "two-way-client-hosted"}