"""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()