Do you really need AI for that?

PROMPT = (
    "Find every piece of personal contact information in the message below.\n"
    "Answer with JSON only: a list of objects with keys `text` and `kind`,\n"
    "where `kind` is one of email, phone, iban, address.\n"
    "If there is none, answer with an empty list.\n\n"
    "Message:\n{message}"
)

MAX_CHARACTERS = 8000


class MaskingUnavailable(Exception):
    """The provider could not be reached, or answered something unusable."""


def mask(message: str, client=None, *, attempts: int = 3) -> str:
    """
    Replace contact details with a label naming what was removed.

    `client` is injected so this function can be tested without a network call.
    In production it defaults to a real provider client.
    """
    if client is None:  # pragma: no cover - needs a key and a network
        from openai import OpenAI

        client = OpenAI()

    # A model charges by the token. Refusing oversized input is not an
    # optimisation, it is a cost control.
    if len(message) > MAX_CHARACTERS:
        raise ValueError(f"message longer than {MAX_CHARACTERS} characters")

    found = _ask(client, message, attempts)

    # Replace the longest matches first, so a substring never eats its parent.
    for item in sorted(found, key=lambda i: len(i.get("text", "")), reverse=True):
        text, kind = item.get("text"), item.get("kind")
        if text and kind:
            message = message.replace(text, f"[{kind}]")
    return message


def _ask(client, message: str, attempts: int) -> list[dict]:
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            answer = client.complete(prompt=PROMPT.format(message=message), temperature=0)
            parsed = json.loads(answer)
            if isinstance(parsed, list):
                return parsed
            last_error = ValueError("the model answered something that is not a list")
        except Exception as error:  # noqa: BLE001 - any provider failure is retried
            last_error = error
    raise MaskingUnavailable(str(last_error))

UNUSUAL_SPACES = re.compile(r"[     ]")

EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")

# French numbers: 0X XX XX XX XX, or +33 X XX XX XX XX. The separator between
# digits may be a space, a dot or a dash, or absent.
SEP = r"[ .-]?"
PHONE = re.compile(rf"(?<![\d+]){SEP}(?:\+{SEP}33{SEP}|0)[1-9](?:{SEP}\d){{8}}(?!\d)")

# IBAN: two letters, two check digits, then up to thirty alphanumerics,
# conventionally grouped in fours.
IBAN = re.compile(r"(?<![A-Z0-9])[A-Z]{2} ?\d{2}(?: ?[A-Z0-9]){10,28}(?![A-Z0-9])")

# Order matters: an email may contain digits that would otherwise be read as
# the start of a phone number.
PATTERNS = ((EMAIL, "[email]"), (IBAN, "[iban]"), (PHONE, "[phone]"))


def normalise(text: str) -> str:
    """Reduce the many spellings of a space to a plain one."""
    return UNUSUAL_SPACES.sub(" ", unicodedata.normalize("NFKC", text))


def mask(text: str) -> str:
    """
    Replace contact details with a label naming what was removed.

    A label beats a row of asterisks: whoever reads the thread later can see
    that a phone number was removed, not merely that something was.
    """
    text = normalise(text)
    for pattern, label in PATTERNS:
        text = pattern.sub(label, text)
    return text