Composing base zoning and overlay districts without losing either

def audit_codes(parcels, crosswalk):
    """Every distinct code, with whether the crosswalk knows it and whether it looks
    like it has something appended."""
    import re
    rows = []
    for code, n in parcels["zoning"].value_counts().items():
        known = code in crosswalk
        # A base code is usually a letter class plus a digit. Anything after that is
        # a candidate overlay token — but "candidate" is the operative word.
        m = re.match(r"^([A-Z]{1,3}-?\d{0,2}[A-Z]?)([-/(].+)?$", code.strip().upper())
        rows.append({"code": code, "parcels": n, "in_crosswalk": known,
                     "base_guess": m.group(1) if m else None,
                     "tail": (m.group(2) or "").strip("-/()") if m else None})
    return sorted(rows, key=lambda r: (-r["parcels"]))

def classify_tail(base, tail, district_list, overlay_registry) -> str:
    if f"{base}-{tail}" in district_list or f"{base}{tail}" in district_list:
        return "distinct_district"      # R-1A is its own district: do NOT split
    if tail in overlay_registry:
        return "known_overlay"          # safe to split
    return "unknown"                    # to review, never guessed

from dataclasses import dataclass, field


@dataclass(frozen=True)
class SplitRule:
    jurisdiction: str
    separators: tuple[str, ...] = ("-", "/", "(", ")")
    # Tokens that are overlays IN THIS JURISDICTION. "HO" is a historic overlay in
    # one city and a homeowner-association overlay in the next, so this is scoped.
    overlay_tokens: frozenset[str] = frozenset()
    # Codes that must never be split, however fused they look.
    atomic_codes: frozenset[str] = frozenset()
    max_overlays: int = 4


@dataclass(frozen=True)
class Split:
    base: str
    overlays: tuple[str, ...]
    unresolved: tuple[str, ...] = ()
    rule: str = ""


def split_code(raw: str, rule: SplitRule, district_list) -> Split:
    code = " ".join(raw.upper().split())

    if code in rule.atomic_codes or code in district_list:
        # The whole string is a district in its own right. This check comes FIRST,
        # which is what stops R-1A being torn into R-1 plus a phantom overlay.
        return Split(code, (), rule=f"{rule.jurisdiction}:atomic")

    tokens = [code]
    for sep in rule.separators:
        tokens = [t for part in tokens for t in part.split(sep) if t]

    if not tokens:
        return Split(code, (), rule="empty")

    base, tail = tokens[0], tokens[1:]
    if base not in district_list:
        # The leading token is not a district either. Do not guess a base.
        return Split(code, (), unresolved=tuple(tokens),
                     rule=f"{rule.jurisdiction}:base_unknown")

    overlays = tuple(t for t in tail if t in rule.overlay_tokens)
    unresolved = tuple(t for t in tail if t not in rule.overlay_tokens)

    if len(overlays) > rule.max_overlays:
        raise ValueError(f"{code}: {len(overlays)} overlays exceeds the declared "
                         f"maximum for {rule.jurisdiction}; the split rule is wrong")

    return Split(base, overlays, unresolved, rule=f"{rule.jurisdiction}:split")

def recompose(split: Split, rule: SplitRule) -> str:
    sep = rule.separators[0]
    return sep.join((split.base, *split.overlays, *split.unresolved))


def round_trips(raw: str, rule: SplitRule, district_list) -> bool:
    s = split_code(raw, rule, district_list)
    return recompose(s, rule).replace(" ", "") == " ".join(raw.upper().split()).replace(" ", "")

def ingest_zoning(raw, rule, district_list, quarantine):
    s = split_code(raw, rule, district_list)

    if s.unresolved:
        # An unrecognised token might be an overlay nobody has registered. Recording
        # the parcel with a partial split and a flag is right; silently discarding
        # the token is how a wellhead overlay disappears.
        quarantine.record(raw, reason="unresolved_token", tokens=s.unresolved,
                          partial_base=s.base, partial_overlays=s.overlays)

    return s

ALTER TABLE parcel_version
    ADD COLUMN base_code     text  NOT NULL,
    ADD COLUMN overlay_codes text[] NOT NULL DEFAULT '{}',
    ADD COLUMN raw_zoning    text  NOT NULL,      -- exactly what the portal said
    ADD COLUMN split_rule    text  NOT NULL;      -- which rule produced the split

-- Overlay codes are sorted on write so the array is comparable and hashable; an
-- unsorted array makes every re-publication look like a change.
ALTER TABLE parcel_version
    ADD CONSTRAINT overlay_codes_sorted
    CHECK (overlay_codes = (SELECT array_agg(x ORDER BY x)
                              FROM unnest(overlay_codes) AS x));

LARIMER = SplitRule("larimer", overlay_tokens=frozenset({"HO", "WP", "AH", "TOD"}),
                    atomic_codes=frozenset({"R-1A", "R-2B"}))
DISTRICTS = {"R-1", "R-2", "MU-2", "R-1A", "R-2B"}


@pytest.mark.parametrize("raw,base,overlays", [
    ("R-1",            "R-1",  ()),
    ("R-1-HO",         "R-1",  ("HO",)),
    ("MU-2-TOD-HP",    "MU-2", ("TOD",)),        # HP unregistered → unresolved
    ("R-1/WP",         "R-1",  ("WP",)),
])
def test_splits(raw, base, overlays):
    s = split_code(raw, LARIMER, DISTRICTS)
    assert s.base == base and s.overlays == overlays


def test_atomic_district_is_never_split():
    s = split_code("R-1A", LARIMER, DISTRICTS)
    assert s.base == "R-1A" and s.overlays == ()   # not R-1 + phantom "A"


def test_unresolved_token_is_kept_not_dropped():
    s = split_code("MU-2-TOD-HP", LARIMER, DISTRICTS)
    assert "HP" in s.unresolved                   # surfaced for registration


def test_round_trip():
    for raw in ("R-1", "R-1-HO", "R-1A"):
        assert round_trips(raw, LARIMER, DISTRICTS)


def test_every_distinct_code_is_covered(parcels):
    """The gate that matters in production: no distinct code may be unhandled."""
    bad = [c for c in parcels["zoning"].unique()
           if split_code(c, LARIMER, DISTRICTS).unresolved]
    assert not bad, f"{len(bad)} distinct codes have unresolved tokens: {bad[:5]}"