`services/common/normalize.py`

143 lines
"""Shared normalization + masking primitives (stdlib only).

Raw source values are never mutated by these functions — callers store the raw string
alongside the normalized output. Production may swap in ICU/libpostal for parsing, but
the CONTRACT (what 'normalized' means per field) is defined here so ingestion, search,
and matching all agree.
"""
from __future__ import annotations

import re
import unicodedata
from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN

CORPORATE_SUFFIXES = {
    "INC", "INCORPORATED", "LLC", "LLC.", "L L C", "LTD", "LIMITED",
    "CORP", "CORPORATION", "CO", "COMPANY", "LP", "LLP", "PLLC",
}


def normalize_text(value: str | None) -> str:
    """Uppercase, strip accents, collapse to single-spaced alphanumerics."""
    if not value:
        return ""
    decomposed = unicodedata.normalize("NFKD", value)
    ascii_like = "".join(ch for ch in decomposed if not unicodedata.combining(ch))
    cleaned = re.sub(r"[^A-Z0-9]+", " ", ascii_like.upper())
    return re.sub(r"\s+", " ", cleaned).strip()


def normalize_search(value: str | None) -> str:
    """Fully-collapsed alphanumeric SEARCH key: normalize_text with spaces removed.

    'Catherine O\\'Neil' -> 'CATHERINEONEIL'. Substring-matching a same-collapsed query
    ('Oneil' -> 'ONEIL') then finds it regardless of internal apostrophes/hyphens/spaces —
    fixing the apostrophe/hyphen recall gap without widening normalize_text (which feeds
    masks, blocking keys, and phonetic codes). Production = an OpenSearch char-filter analyzer.
    """
    return normalize_text(value).replace(" ", "")


def normalize_business(value: str | None) -> str:
    """Normalize a business name, dropping legal suffixes but keeping brand tokens."""
    tokens = normalize_text(value).split()
    meaningful = [t for t in tokens if t not in CORPORATE_SUFFIXES]
    return " ".join(meaningful) or normalize_text(value)


_CENTS = Decimal("0.01")


def parse_decimal(value: str | None) -> Decimal | None:
    # m4: quantize to 2 dp so downstream arithmetic stays consistent (Decimal("123")
    # != Decimal("123.00") in comparisons, and string-formatting varies without a fixed
    # precision). Banker's rounding (ROUND_HALF_EVEN) is the financial-calculation default.
    if not value or not value.strip():
        return None
    try:
        raw = Decimal(value.replace("$", "").replace(",", "").strip())
        return raw.quantize(_CENTS, rounding=ROUND_HALF_EVEN)
    except InvalidOperation as exc:
        raise ValueError(f"Invalid amount: {value!r}") from exc


# Soundex digit groups (the classic mapping). We build a phonetic key by coding EVERY
# consonant (including the first letter) and dropping vowels/H/W/Y — so "Catherine" and
# "Kathryn" collapse to the same code (C and K are both group 2), which first-letter
# Soundex would miss. Production swaps this for Double Metaphone; the FEATURE contract
# (phonetic agreement) is what matters here.
_SOUNDEX_MAP = {
    **{c: "1" for c in "BFPV"},
    **{c: "2" for c in "CGJKQSXZ"},
    **{c: "3" for c in "DT"},
    "L": "4",
    **{c: "5" for c in "MN"},
    "R": "6",
}


def phonetic_key(value: str | None) -> str:
    """Space-joined per-token phonetic code (vowel-stripped, adjacent-duplicate-collapsed)."""
    out_tokens = []
    for token in normalize_text(value).split():
        digits: list[str] = []
        for ch in token:
            code = _SOUNDEX_MAP.get(ch)
            if code and (not digits or digits[-1] != code):
                digits.append(code)
            elif not code:
                # a vowel/H/W/Y breaks a run so a real repeated sound isn't over-collapsed
                if digits and digits[-1] == "":
                    continue
                digits.append("")
        joined = "".join(d for d in digits if d)
        if joined:
            out_tokens.append(joined)
    return " ".join(out_tokens)


def normalize_postal(value: str | None) -> str | None:
    if not value:
        return None
    digits = re.sub(r"\D", "", value)[:9]
    return digits or None


# Constant mask width — MUST NOT leak the true token length. Revealing first-initial +
# exact length turns anonymous search into a per-person CONFIRMATION oracle (a third party
# who already knows a name+city can confirm that person has property, and the amount band).
# A fixed-width mask lets a rightful owner recognize their own record without letting a
# stranger confirm it about someone else. (Security audit finding 1.1.)
_MASK_WIDTH = 3


def mask_name(raw_name: str | None) -> str:
    """Public-search masking: first char of each token + a CONSTANT-width mask.

    'CATHERINE ONEILL' -> 'C••• O•••'  (length is NOT revealed).
    Per-jurisdiction policy may mask even more coarsely via jurisdiction_policy.config_json.
    """
    norm = normalize_text(raw_name)
    if not norm:
        return ""
    return " ".join(tok[0] + ("•" * _MASK_WIDTH) for tok in norm.split())


# Coarse public amount bands — never expose the exact figure through anonymous search.
_BANDS = [
    (Decimal("50"), "Under $50"),
    (Decimal("100"), "$50–$100"),
    (Decimal("500"), "$100–$500"),
    (Decimal("1000"), "$500–$1,000"),
    (Decimal("5000"), "$1,000–$5,000"),
]


def amount_band(amount: Decimal | None) -> str | None:
    if amount is None:
        return None
    for ceiling, label in _BANDS:
        if amount < ceiling:
            return label
    return "$5,000+"