Model/scripts/lisasrequest/resolve_uprns_for_finaliser.py

"""Step 2 (Durkan portfolio): split step-1 matches, reshape the confident ones.

Reads durkan_domna_filled.csv (step 1) and SPLITS it in two — no re-resolution,
just column work:

  * Rows we cannot confidently insert are held back to a client-clarification CSV
    (durkan_client_clarification.csv) for Khalim to take to the client. Reasons:
      not_found_no_match     no UPRN was resolved.
      no_flat_level_uprn     a block of flats all collapsed onto one building
                             UPRN — OS/EPC carry no flat-level records, so we
                             can't tell the flats apart.
      unit_number_mismatch   the matched house number differs from the input
                             (e.g. "9 ..." matched "9A ..."), so the property is
                             ambiguous.
  * Every remaining row is reshaped into the columns the finaliser reads
    (bulk_upload_finaliser_orchestrator), written to durkan_finaliser_input.csv
    ready for step 3:
      Address 1/2/3 | postcode | Internal Reference | address2uprn_uprn
      | address2uprn_address | address2uprn_lexiscore
    Internal Reference is left blank (landlord_property_id null, by decision).

    python -m scripts.lisasrequest.resolve_uprns_for_finaliser

This stage hits no APIs. The held rows are not lost — once the client confirms
them they can be appended to the finaliser input by hand.
"""

from __future__ import annotations

import argparse
import csv
import sys
from collections import Counter
from pathlib import Path
from typing import Optional

_REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_REPO_ROOT))  # worktree root first — avoid the import trap

from scripts.lisasrequest.fill_domna_address import (  # noqa: E402
    ADDRESS_COL,
    FOUND_ADDRESS_COL,
    FOUND_UPRN_COL,
    LEXISCORE_COL,
    POSTCODE_COL,
    SOURCE_COL,
)
from scripts.lisasrequest.review_flags import address_numbers, input_unit  # noqa: E402

# Finaliser input columns — must match bulk_upload_finaliser_orchestrator
# (ADDRESS_COLS / POSTCODE_COL / INTERNAL_REF_COL / UPRN_COL /
# MATCHED_ADDRESS_COL / LEXISCORE_COL). Hard-coded to keep this a light,
# stdlib-only reshape; step 3 imports the real orchestrator and will fail loudly
# if these ever drift.
FIN_ADDRESS_1, FIN_ADDRESS_2, FIN_ADDRESS_3 = "Address 1", "Address 2", "Address 3"
FIN_POSTCODE = "postcode"
FIN_INTERNAL_REF = "Internal Reference"
FIN_UPRN = "address2uprn_uprn"
FIN_MATCHED_ADDRESS = "address2uprn_address"
FIN_LEXISCORE = "address2uprn_lexiscore"
_FINALISER_COLS = [
    FIN_ADDRESS_1,
    FIN_ADDRESS_2,
    FIN_ADDRESS_3,
    FIN_POSTCODE,
    FIN_INTERNAL_REF,
    FIN_UPRN,
    FIN_MATCHED_ADDRESS,
    FIN_LEXISCORE,
]

# Client-clarification report columns (kept human-readable for the client).
CONTEXT_COLS = ["address", "postcode", "No.", "Address Block"]
DOMNA_COLS = [FOUND_ADDRESS_COL, FOUND_UPRN_COL, LEXISCORE_COL, SOURCE_COL]
REASON_COL = "clarification_reason"
ACTION_COL = "action_needed"
_CLARIFY_COLS = CONTEXT_COLS + DOMNA_COLS + [REASON_COL, ACTION_COL]

_REASON_ORDER = {
    "not_found_no_match": 0,
    "no_flat_level_uprn": 1,
    "unit_number_mismatch": 2,
}
_REASON_ACTION = {
    "not_found_no_match": "No UPRN found for this address — please confirm the "
    "exact address or provide the UPRN.",
    "no_flat_level_uprn": "Address registers hold only the building, not the "
    "individual flats — please provide a UPRN per flat, or confirm a "
    "building-level record is acceptable.",
    "unit_number_mismatch": "Closest match has a different unit number (see "
    "domna_address_found) — please confirm the correct property / UPRN.",
}

_DEFAULT_IN = _REPO_ROOT / "scripts" / "lisasrequest" / "durkan_domna_filled.csv"
_DEFAULT_FINALISER = _REPO_ROOT / "scripts" / "lisasrequest" / "durkan_finaliser_input.csv"
_DEFAULT_CLARIFY = (
    _REPO_ROOT / "scripts" / "lisasrequest" / "durkan_client_clarification.csv"
)


def read_rows(path: Path) -> list[dict[str, str]]:
    with path.open(newline="", encoding="utf-8-sig") as fh:
        return [dict(row) for row in csv.DictReader(fh)]


def clarification_reason(
    row: dict[str, str], uprn_counts: Counter[str]
) -> Optional[str]:
    """Why this row can't be inserted yet, or None if it's safe to finalise."""
    uprn = row.get(FOUND_UPRN_COL, "")
    if row.get(SOURCE_COL) == "not_found" or not uprn:
        return "not_found_no_match"

    unit = input_unit(row.get(ADDRESS_COL, ""))
    unit_missing = bool(unit) and unit not in address_numbers(
        row.get(FOUND_ADDRESS_COL, "")
    )
    duplicate = uprn_counts[uprn] > 1
    if unit_missing:
        return "no_flat_level_uprn" if duplicate else "unit_number_mismatch"
    if duplicate:
        # A shared UPRN with the right unit number still collides at finalise.
        return "no_flat_level_uprn"
    return None


def to_finaliser_row(row: dict[str, str]) -> dict[str, str]:
    """Rename a confident step-1 row into the finaliser's input columns."""
    return {
        FIN_ADDRESS_1: row.get(ADDRESS_COL, ""),
        FIN_ADDRESS_2: "",
        FIN_ADDRESS_3: "",
        FIN_POSTCODE: row.get(POSTCODE_COL, ""),
        FIN_INTERNAL_REF: "",  # landlord_property_id null, by decision
        FIN_UPRN: row.get(FOUND_UPRN_COL, ""),
        FIN_MATCHED_ADDRESS: row.get(FOUND_ADDRESS_COL, ""),
        FIN_LEXISCORE: row.get(LEXISCORE_COL, ""),
    }


def to_clarify_row(row: dict[str, str], reason: str) -> dict[str, str]:
    out = {col: row.get(col, "") for col in CONTEXT_COLS + DOMNA_COLS}
    out[REASON_COL] = reason
    out[ACTION_COL] = _REASON_ACTION[reason]
    return out


def split(
    rows: list[dict[str, str]],
    *,
    accept_unit_mismatch: bool = False,
) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
    """Return (finaliser_rows, clarification_rows).

    ``accept_unit_mismatch`` reshapes the ``unit_number_mismatch`` rows (a
    near-miss like 9 -> 9A the client has already confirmed) into the finaliser
    input instead of holding them back.
    """
    uprn_counts: Counter[str] = Counter(
        r.get(FOUND_UPRN_COL, "") for r in rows if r.get(FOUND_UPRN_COL)
    )
    finaliser: list[dict[str, str]] = []
    clarify: list[dict[str, str]] = []
    for row in rows:
        reason = clarification_reason(row, uprn_counts)
        if reason is None or (
            accept_unit_mismatch and reason == "unit_number_mismatch"
        ):
            finaliser.append(to_finaliser_row(row))
        else:
            clarify.append(to_clarify_row(row, reason))
    clarify.sort(key=lambda r: _REASON_ORDER.get(r[REASON_COL], 9))
    return finaliser, clarify


def write_csv(rows: list[dict[str, str]], path: Path, fieldnames: list[str]) -> None:
    with path.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--in", dest="inp", type=Path, default=_DEFAULT_IN)
    parser.add_argument("--finaliser-out", type=Path, default=_DEFAULT_FINALISER)
    parser.add_argument("--clarify-out", type=Path, default=_DEFAULT_CLARIFY)
    parser.add_argument(
        "--accept-unit-mismatch",
        action="store_true",
        help="reshape unit_number_mismatch rows (e.g. 9->9A) into the finaliser "
        "input instead of holding them for the client",
    )
    args = parser.parse_args()

    rows = read_rows(args.inp)
    finaliser, clarify = split(rows, accept_unit_mismatch=args.accept_unit_mismatch)

    write_csv(finaliser, args.finaliser_out, _FINALISER_COLS)
    write_csv(clarify, args.clarify_out, _CLARIFY_COLS)

    counts = Counter(r[REASON_COL] for r in clarify)
    print(f"Read {len(rows)} step-1 rows.")
    print(f"  -> {len(finaliser)} confident rows reshaped -> {args.finaliser_out}")
    print(f"  -> {len(clarify)} held for client -> {args.clarify_out}")
    for reason in _REASON_ORDER:
        print(f"       {reason}: {counts.get(reason, 0)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())