Model/scripts/hyde/main.py

"""Consolidate UPRN-matching attempts into a single domain-found result.

We have two independent attempts at resolving each property to a UPRN:

  * ``address2uprn_*``        - from the address2uprn matching service
                               (lives in output.csv, the master file)
  * ``ordnance_survey_*``     - from Ordnance Survey
                               (lives in address2uprn.csv, a 4k-row subset)

This script joins the Ordnance Survey columns onto the master by
``Organisation Reference`` and then collapses the two attempts into one
canonical result:

  * domain_found_uprn
  * domain_found_address
  * domain_scoring_confidence   (the winning source's lexiscore)

Tie-break: prefer Ordnance Survey whenever it produced a UPRN, otherwise
fall back to address2uprn.
"""

from __future__ import annotations

from copy import copy
from pathlib import Path
from typing import Any, Optional, cast

import openpyxl
import pandas as pd
from openpyxl.cell.cell import Cell
from openpyxl.worksheet.worksheet import Worksheet

HERE = Path(__file__).parent
MASTER_CSV = HERE / "output.csv"
ORDNANCE_CSV = HERE / "address2uprn.csv"
OUTPUT_CSV = HERE / "output_with_domain_uprn.csv"
BAD_POSTCODE_CSV = HERE / "bad_postcodes.csv"

# The original profiling export. We annotate a copy of it in place rather than
# rebuilding it, so every existing column, sheet and chart is preserved exactly.
ORIGINAL_XLSX = HERE / "ARA AddressProfiling_Download_28-04-2026_1050 (2).xlsx"
ANNOTATED_XLSX = HERE / "ARA AddressProfiling_with_domna.xlsx"
DATA_SHEET = "AddressProfilingResults"
DOMNA_UPRN_HEADER = "DOMNA FOUND UPRN"
DOMNA_ADDRESS_HEADER = "DOMNA FOUND ADDRESS"

JOIN_KEY = "Organisation Reference"
ORDNANCE_COLS = [
    "ordnance_survey_uprn",
    "ordnance_survey_address",
    "ordnance_survey_lexiscore",
]

# Failure markers either matching service writes into its UPRN column when the
# postcode itself could not be resolved. These mean "the postcode was wrong",
# not "no UPRN" — we surface them to the user separately.
POSTCODE_FAIL_SENTINELS = {
    "invalid postcode",
    "postcode not found in ordnance survey",
}


def _as_text(series: "pd.Series[Any]") -> "pd.Series[str]":
    """Coerce any column to a stripped string Series (NA stays NA)."""
    text: "pd.Series[str]" = series.astype("string")
    return text.str.strip()


def _norm_key(series: "pd.Series[Any]") -> "pd.Series[str]":
    """Normalise the join key so int/float/str spellings of the same
    Organisation Reference compare equal (e.g. ``13016`` vs ``13016.0``).

    Also strips leading zeros: the original spreadsheet keeps the reference as
    text (``08450115001``) while the CSV pipeline reads it as a number and drops
    the leading zero (``8450115001``). Without this they would never join."""
    text = (
        _as_text(series).str.replace(r"\.0$", "", regex=True).str.lstrip("0")
    )
    # pandas-stubs types fillna's overload with Any, tripping strict's
    # reportUnknownMemberType even though the result is a Series[str].
    return text.fillna("")  # pyright: ignore[reportUnknownMemberType]


def _is_real_uprn(series: "pd.Series[Any]") -> "pd.Series[bool]":
    """A real UPRN is numeric. The matching services write failure sentinels
    (``invalid postcode``, ``postcode not found in ordnance survey``) into the
    same column, which must NOT count as a found UPRN."""
    # pd.to_numeric's stub return contains Any; cast pins it to a known dtype.
    numeric = cast(
        "pd.Series[float]",
        pd.to_numeric(series, errors="coerce"),  # pyright: ignore[reportUnknownMemberType]
    )
    return numeric.notna()


def _is_postcode_failure(series: "pd.Series[Any]") -> "pd.Series[bool]":
    """True where the value is a known postcode-resolution failure sentinel."""
    return _as_text(series).str.lower().isin(POSTCODE_FAIL_SENTINELS)


def build_ordnance_lookup(ordnance: pd.DataFrame) -> pd.DataFrame:
    """Collapse address2uprn.csv to one row per Organisation Reference,
    keeping the first row that actually has an Ordnance Survey UPRN."""
    frame = ordnance.copy()
    frame["_key"] = _norm_key(frame[JOIN_KEY])
    frame = frame[frame["_key"] != ""]

    # Sort rows with a real OS UPRN first, then keep the first row per key, so a
    # real match always beats an empty/sentinel row for the same Organisation Ref.
    frame["_has_uprn"] = _is_real_uprn(frame["ordnance_survey_uprn"])
    frame = frame.sort_values("_has_uprn", ascending=False, kind="stable")

    lookup = frame.drop_duplicates("_key", keep="first")
    return lookup[["_key", *ORDNANCE_COLS]]


def consolidate(master: pd.DataFrame) -> pd.DataFrame:
    """Add domain_found_* columns, preferring Ordnance Survey over address2uprn.

    Only a real (numeric) UPRN counts as found; failure sentinels in either
    source are ignored. Where neither source resolved a real UPRN the
    domain_found_* columns are left empty.
    """
    os_real = _is_real_uprn(master["ordnance_survey_uprn"])
    a2_real = _is_real_uprn(master["address2uprn_uprn"])

    # Source selection per row: Ordnance Survey wins, else address2uprn, else none.
    def _pick(os_col: str, a2_col: str) -> "pd.Series[Any]":
        empty = pd.Series([pd.NA] * len(master), index=master.index, dtype="object")
        chosen = empty.where(~a2_real, master[a2_col])
        chosen = chosen.where(~os_real, master[os_col])
        return chosen

    master["domna_found_uprn"] = _pick("ordnance_survey_uprn", "address2uprn_uprn")
    master["domna_found_address"] = _pick(
        "ordnance_survey_address", "address2uprn_address"
    )
    master["domna_scoring_confidence"] = _pick(
        "ordnance_survey_lexiscore", "address2uprn_lexiscore"
    )

    # Outcome of the lookup. A bad postcode (sentinel from either source) is only
    # worth flagging when we did NOT otherwise find a real UPRN for the property.
    found = os_real | a2_real
    bad_postcode = _is_postcode_failure(master["ordnance_survey_uprn"]) | (
        _is_postcode_failure(master["address2uprn_uprn"])
    )
    status = pd.Series(["unmatched"] * len(master), index=master.index, dtype="object")
    status = status.where(~bad_postcode, "bad_postcode")
    status = status.where(~found, "matched")
    master["domna_match_status"] = status
    return master


def _norm_scalar(value: object) -> str:
    """Scalar twin of :func:`_norm_key` for iterating worksheet cells.

    Strips a trailing ``.0`` and leading zeros so the worksheet's text reference
    (``08450115001``) matches the CSV's numeric one (``8450115001``)."""
    if value is None:
        return ""
    text = str(value).strip()
    text = text[:-2] if text.endswith(".0") else text
    return text.lstrip("0")


def _build_domna_lookup(merged: pd.DataFrame) -> "dict[str, tuple[int, str]]":
    """Organisation Reference -> (found UPRN, found address) for every row where
    we resolved a real UPRN. UPRNs are returned as ints so the spreadsheet shows
    them whole rather than as ``2.02e+08``/``...0``."""
    real = merged[merged["domna_found_uprn"].notna()]
    out: "dict[str, tuple[int, str]]" = {}
    for ref, uprn, addr in zip(
        real[JOIN_KEY], real["domna_found_uprn"], real["domna_found_address"]
    ):
        key = _norm_scalar(ref)
        if not key:
            continue
        addr_text = str(addr) if addr is not None and addr == addr else ""
        out[key] = (int(float(str(uprn))), addr_text)
    return out


def _cell(ws: Worksheet, row: int, column: int) -> Cell:
    """Fetch a worksheet cell typed as ``Cell`` (openpyxl returns a wider union)."""
    return cast(Cell, ws.cell(row=row, column=column))


def _locate_columns(ws: Worksheet) -> "tuple[int, int, int]":
    """Return (header_row, organisation-ref col, UPRN col) for the data sheet,
    which sits below a metadata preamble rather than on row 1."""
    header_row = 0
    for row in range(1, 31):
        if str(_cell(ws, row, 1).value).strip() == JOIN_KEY:
            header_row = row
            break
    if header_row == 0:
        raise ValueError(f"Could not find a '{JOIN_KEY}' header in {ws.title!r}")

    uprn_col = 0
    for col in range(1, ws.max_column + 1):
        if str(_cell(ws, header_row, col).value).strip() == "UPRN":
            uprn_col = col
            break
    if uprn_col == 0:
        raise ValueError(f"Could not find a 'UPRN' header in {ws.title!r}")
    return header_row, 1, uprn_col


def _copy_header_style(src: Cell, dst: Cell) -> None:
    """Make a new header cell look native by cloning the UPRN header's style.

    openpyxl assigns these style attributes via descriptors that its stubs type
    as read-only, so strict flags the (runtime-valid) assignments."""
    dst.font = copy(src.font)  # pyright: ignore[reportAttributeAccessIssue]
    dst.fill = copy(src.fill)  # pyright: ignore[reportAttributeAccessIssue]
    dst.border = copy(src.border)  # pyright: ignore[reportAttributeAccessIssue]
    dst.alignment = copy(src.alignment)  # pyright: ignore[reportAttributeAccessIssue]
    dst.number_format = src.number_format


def annotate_original_excel(
    merged: pd.DataFrame,
    source: Optional[Path] = None,
    dest: Optional[Path] = None,
) -> int:
    """Append DOMNA FOUND UPRN / ADDRESS columns to a copy of the original
    workbook, filled only where the property had no UPRN and we found one.
    Layout, formatting and other sheets are left untouched."""
    source = source or ORIGINAL_XLSX
    dest = dest or ANNOTATED_XLSX
    lookup = _build_domna_lookup(merged)

    workbook = openpyxl.load_workbook(source)
    ws = cast(Worksheet, workbook[DATA_SHEET])
    header_row, ref_col, uprn_col = _locate_columns(ws)

    new_uprn_col = ws.max_column + 1
    new_addr_col = new_uprn_col + 1
    style_src = _cell(ws, header_row, uprn_col)
    for col, title in (
        (new_uprn_col, DOMNA_UPRN_HEADER),
        (new_addr_col, DOMNA_ADDRESS_HEADER),
    ):
        header_cell = _cell(ws, header_row, col)
        header_cell.value = title
        _copy_header_style(style_src, header_cell)

    filled = 0
    for row in range(header_row + 1, ws.max_row + 1):
        if _cell(ws, row, uprn_col).value not in (None, ""):
            continue  # property already has a UPRN — leave it alone
        match = lookup.get(_norm_scalar(_cell(ws, row, ref_col).value))
        if match is None:
            continue
        uprn, address = match
        uprn_cell = _cell(ws, row, new_uprn_col)
        uprn_cell.value = uprn
        uprn_cell.number_format = "0"
        _cell(ws, row, new_addr_col).value = address
        filled += 1

    workbook.save(dest)
    return filled


def main(
    master_path: Optional[Path] = None,
    ordnance_path: Optional[Path] = None,
    output_path: Optional[Path] = None,
) -> None:
    master_path = master_path or MASTER_CSV
    ordnance_path = ordnance_path or ORDNANCE_CSV
    output_path = output_path or OUTPUT_CSV

    master = pd.read_csv(master_path, low_memory=False)
    ordnance = pd.read_csv(ordnance_path, low_memory=False)

    master["_key"] = _norm_key(master[JOIN_KEY])
    lookup = build_ordnance_lookup(ordnance)

    merged = master.merge(lookup, on="_key", how="left").drop(columns="_key")
    merged = consolidate(merged)

    merged.to_csv(output_path, index=False)

    # Flag properties whose postcode could not be resolved by either service —
    # this means the source postcode is wrong and needs correcting at source.
    bad = merged[merged["domna_match_status"] == "bad_postcode"]
    flag_cols = [c for c in [JOIN_KEY, "Address 1", "postcode"] if c in bad.columns]
    bad[flag_cols].to_csv(BAD_POSTCODE_CSV, index=False)

    # Annotate a copy of the original workbook: fill DOMNA columns only for
    # properties that had no UPRN and where we found one.
    annotated = annotate_original_excel(merged)

    os_matched = int(_is_real_uprn(merged["ordnance_survey_uprn"]).sum())
    found = int(merged["domna_found_uprn"].notna().sum())
    print(f"rows                       : {len(merged)}")
    print(f"ordnance survey matched    : {os_matched}")
    print(f"domna_found_uprn populated : {found}")
    print(f"written to                 : {output_path}")
    print(f"excel rows filled (no UPRN): {annotated}")
    print(f"annotated workbook         : {ANNOTATED_XLSX}")
    if len(bad):
        print()
        print(f"⚠️  {len(bad)} properties have a BAD POSTCODE (could not be resolved).")
        print(f"    These postcodes are likely wrong — review: {BAD_POSTCODE_CSV.name}")


if __name__ == "__main__":
    main()