Reconciling Cost Code Aliases with a Canonical Registry

from __future__ import annotations

import logging
import re
from decimal import Decimal
from difflib import SequenceMatcher
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("cost_code_registry")

# Canonical general-ledger account key, e.g. "GL-1001".
CANONICAL_RE = re.compile(r"^GL-\d{4}$")
# Any non-alphanumeric run is a delimiter to be stripped during normalization.
NON_ALNUM_RE = re.compile(r"[^A-Z0-9]+")

MatchType = Literal["registered_alias", "fuzzy_alias", "unmatched"]
RouteState = Literal["auto_route", "human_review", "quarantine"]

AUTO_ROUTE = Decimal("0.92")    # site-canonical: >= 0.92 adopts the canonical code
HUMAN_REVIEW = Decimal("0.75")  # 0.75–0.92 holds for an estimator to confirm


def normalize_alias(raw: str) -> str:
    """Reduce any source spelling to one comparison key. Pure and idempotent.

    'GL1001', 'GL-1001', and 'gl 1001' all collapse to 'GL1001', so a second
    pass over the output changes nothing and a broker retry is a no-op.
    """
    return NON_ALNUM_RE.sub("", raw.strip().upper())


class AliasResolution(BaseModel):
    """A validated alias->canonical resolution carrying its own routing decision."""

    model_config = ConfigDict(frozen=True)

    source_alias: str = Field(..., description="Raw inbound spelling, preserved for audit")
    source_system: str = Field(..., description="Originating platform, for the audit trail")
    canonical_code: str | None = Field(default=None)
    match_type: MatchType
    confidence: Decimal = Field(..., ge=0, le=1)

    @field_validator("canonical_code")
    @classmethod
    def _valid_canonical(cls, v: str | None) -> str | None:
        # None is legal only for the quarantine path; a present code must be canonical.
        if v is not None and not CANONICAL_RE.match(v):
            raise ValueError(f"Non-canonical account key: {v!r}")
        return v

    @property
    def route_state(self) -> RouteState:
        if self.canonical_code is not None and self.confidence >= AUTO_ROUTE:
            return "auto_route"
        if self.confidence >= HUMAN_REVIEW:
            return "human_review"
        return "quarantine"


class CanonicalRegistry:
    """Reconciles platform-specific aliases to one canonical account key."""

    def __init__(self, entries: dict[str, list[str]]) -> None:
        # entries maps a canonical code -> the raw alias spellings observed for it.
        self._by_key: dict[str, str] = {}
        for canonical, aliases in entries.items():
            if not CANONICAL_RE.match(canonical):
                raise ValueError(f"Registry seeded with bad canonical code: {canonical!r}")
            # A canonical code is its own alias, so a clean source resolves exactly.
            for spelling in (canonical, *aliases):
                self._by_key[normalize_alias(spelling)] = canonical

    def learn(self, canonical: str, alias: str) -> None:
        """Promote a confirmed human-review spelling to an exact future hit."""
        self._by_key[normalize_alias(alias)] = canonical

    def resolve(self, raw_alias: str, source_system: str) -> AliasResolution:
        key = normalize_alias(raw_alias)

        # 1. Exact registered alias — highest confidence, auto-routes.
        if key in self._by_key:
            return AliasResolution(
                source_alias=raw_alias, source_system=source_system,
                canonical_code=self._by_key[key],
                match_type="registered_alias", confidence=Decimal("0.99"),
            )

        # 2. Fuzzy match against every known key; the band decides the route.
        best_key, best_score = None, 0.0
        for candidate in self._by_key:
            score = SequenceMatcher(None, key, candidate).ratio()
            if score > best_score:
                best_key, best_score = candidate, score
        confidence = Decimal(str(round(best_score, 2)))

        if best_key is not None and confidence >= HUMAN_REVIEW:
            return AliasResolution(
                source_alias=raw_alias, source_system=source_system,
                canonical_code=self._by_key[best_key],
                match_type="fuzzy_alias", confidence=confidence,
            )

        # 3. Nothing credible matched — quarantine, never guess an account.
        logger.warning("No canonical match for %r from %s; quarantining", raw_alias, source_system)
        return AliasResolution(
            source_alias=raw_alias, source_system=source_system,
            canonical_code=None, match_type="unmatched", confidence=confidence,
        )

def reconcile_batch(
    registry: CanonicalRegistry,
    rows: list[tuple[str, str, Decimal]],
) -> dict[str, Decimal]:
    """Resolve (alias, source_system, amount) rows and total the auto-routed ones."""
    ledger: dict[str, Decimal] = {}
    for raw_alias, source_system, amount in rows:
        resolution = registry.resolve(raw_alias, source_system)
        if resolution.route_state == "auto_route" and resolution.canonical_code:
            running = ledger.get(resolution.canonical_code, Decimal("0.00"))
            ledger[resolution.canonical_code] = running + amount
        else:
            # Held or quarantined: audit it, never fold a guessed account into a total.
            logger.info("AUDIT | %s | %s", resolution.route_state, resolution.model_dump_json())
    return ledger


if __name__ == "__main__":
    registry = CanonicalRegistry(
        {
            "GL-1001": ["GL1001", "gl 1001", "1001-CONC"],
            "GL-2200": ["GL2200", "GL 2200", "2200-STEEL"],
        }
    )
    rows = [
        ("GL1001", "erp", Decimal("150000.00")),      # exact alias -> auto
        ("gl-1001", "procore", Decimal("42500.50")),  # normalizes to a known key
        ("GL 2200", "field_csv", Decimal("87250.00")),  # exact after normalize
        ("GL-1O01", "ocr", Decimal("9900.00")),        # OCR slip: fuzzy -> review
        ("MISC-XYZ", "email", Decimal("500.00")),      # no match -> quarantine
    ]
    for code, total in reconcile_batch(registry, rows).items():
        print(f"{code}: ${total:,.2f}")