Model/scripts/lisasrequest/review_flags.py

"""Flag step-1 matches that need a human eye, for review before finalising.

Reads durkan_domna_filled.csv (the step-1 output) and writes a review CSV of
only the rows carrying at least one flag, newest-doubt-first:

    not_found          no UPRN resolved at all.
    unit_not_in_match  the input flat/house number does NOT appear in the matched
                       address — the high-precision "wrong property" signal. Two
                       shapes: a near-miss ("9 VANBRUGH" matched "9A, VANBRUGH")
                       or a flat collapsing onto its building ("FLAT 1, 20 WARWICK"
                       matched "20, WARWICK ROAD").
    dup_uprn           the same UPRN was resolved for >1 input row — typically a
                       block of flats all collapsing onto the building UPRN; all
                       but one will be dropped at finalise.
    low_score          lexiscore < 0.70 (a weak match, just over the OS bar). NOTE:
                       on its own this is noisy — truncated EPC addresses and extra
                       locality tokens push correct matches below 0.70. Treat it as
                       informational unless paired with one of the flags above.

    python -m scripts.lisasrequest.review_flags
"""

from __future__ import annotations

import argparse
import csv
import re
import sys
from collections import Counter
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[2]

ADDRESS_COL = "address"
POSTCODE_COL = "postcode"
FOUND_ADDRESS_COL = "domna_address_found"
FOUND_UPRN_COL = "domna_address_uprn"
LEXISCORE_COL = "domna_lexiscore"
SOURCE_COL = "domna_source"
LOW_SCORE = 0.70

_DEFAULT_IN = _REPO_ROOT / "scripts" / "lisasrequest" / "durkan_domna_filled.csv"
_DEFAULT_OUT = _REPO_ROOT / "scripts" / "lisasrequest" / "durkan_review_flags.csv"

_REVIEW_COLS = [
    ADDRESS_COL,
    POSTCODE_COL,
    FOUND_ADDRESS_COL,
    FOUND_UPRN_COL,
    LEXISCORE_COL,
    SOURCE_COL,
    "flags",
]


def input_unit(address: str) -> str:
    """The salient unit number of an input address: the FLAT number if present,
    else the leading house number ("" if neither). Upper-cased."""
    upper = address.upper()
    flat = re.search(r"\bFLAT\s+(\d+[A-Z]?)", upper)
    if flat:
        return flat.group(1)
    lead = re.match(r"\s*(\d+[A-Z]?)\b", upper)
    return lead.group(1) if lead else ""


def address_numbers(address: str) -> set[str]:
    """All standalone number tokens in an address (e.g. {"3", "20"}). Upper-cased."""
    return set(re.findall(r"\b\d+[A-Z]?\b", address.upper()))


def _score(value: str) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def flag_rows(rows: list[dict[str, str]]) -> list[dict[str, str]]:
    """Return the flagged subset, each with a ';'-joined ``flags`` field."""
    uprn_counts = Counter(
        r.get(FOUND_UPRN_COL, "") for r in rows if r.get(FOUND_UPRN_COL)
    )

    flagged: list[dict[str, str]] = []
    for row in rows:
        uprn = row.get(FOUND_UPRN_COL, "")
        source = row.get(SOURCE_COL, "")
        flags: list[str] = []

        if source == "not_found" or not uprn:
            flags.append("not_found")
        else:
            unit = input_unit(row.get(ADDRESS_COL, ""))
            if unit and unit not in address_numbers(row.get(FOUND_ADDRESS_COL, "")):
                flags.append("unit_not_in_match")
            if uprn_counts[uprn] > 1:
                flags.append("dup_uprn")
            if _score(row.get(LEXISCORE_COL, "")) < LOW_SCORE:
                flags.append("low_score")

        if flags:
            flagged.append({**{c: row.get(c, "") for c in _REVIEW_COLS[:-1]},
                            "flags": ";".join(flags)})

    # not_found first, then mismatches, then dup/low.
    order = {"not_found": 0, "unit_not_in_match": 1, "dup_uprn": 2, "low_score": 3}
    flagged.sort(key=lambda r: order.get(r["flags"].split(";")[0], 9))
    return flagged


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--in", dest="inp", type=Path, default=_DEFAULT_IN)
    parser.add_argument("--out", type=Path, default=_DEFAULT_OUT)
    args = parser.parse_args()

    with args.inp.open(newline="", encoding="utf-8-sig") as fh:
        rows = [dict(r) for r in csv.DictReader(fh)]

    flagged = flag_rows(rows)
    with args.out.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=_REVIEW_COLS, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(flagged)

    counts = Counter(f for r in flagged for f in r["flags"].split(";"))
    print(f"{len(flagged)}/{len(rows)} rows flagged for review -> {args.out}")
    for name in ("not_found", "unit_not_in_match", "dup_uprn", "low_score"):
        print(f"  {name}: {counts.get(name, 0)}")
    return 0


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