"""Resolve a CSV of addresses to UPRNs, ready to feed the bulk-upload finaliser. Takes a CSV with `Address 1/2/3` + `postcode` columns and, per row, resolves a UPRN by trying — in order — the new EPC API (address2uprn), the historic EPC S3 dataset, then the Ordnance Survey Places API as a fallback. Whichever source wins, the result is written into the SAME three columns the finaliser reads (`bulk_upload_finaliser_orchestrator`): address2uprn_uprn UPRN integer (empty when unresolved) address2uprn_address the matched address address2uprn_lexiscore the match score in [0, 1] A `resolution_source` diagnostic column (epc / epc_historic / ordnance_survey / none) is appended too — the finaliser ignores unknown columns. All original columns are preserved in their original order, so the output CSV drops straight into the finaliser. python -m scripts.resolve_uprns_for_finaliser input.csv -o resolved.csv # OS-only / EPC-only, custom postcode column, custom OS score threshold python -m scripts.resolve_uprns_for_finaliser in.csv -o out.csv --no-epc python -m scripts.resolve_uprns_for_finaliser in.csv -o out.csv --postcode-col Postcode --os-threshold 0.6 Keys are read from backend/.env: OPEN_EPC_API_TOKEN (EPC) and ORDNANCE_SURVEY_API_KEY (OS Places). Run from the worktree root (import trap). The module-level functions (`load_keys`, `read_rows`, `resolve_row`, `process`, `write_rows`) are written to be driven line-by-line from a REPL as well as via the CLI. """ from __future__ import annotations import argparse import csv import os import sys from pathlib import Path from typing import Optional import pandas as pd from dotenv import load_dotenv _REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import trap from backend.address2UPRN.main import ( # noqa: E402 get_epc_data_with_postcode, get_uprn_from_historic_epc, get_uprn_with_epc_df, ) from backend.ordnanceSurvey.helpers import ( # noqa: E402 lookup_os_places, os_places_results_to_dataframe, ) from backend.utils.addressMatch import AddressMatch # noqa: E402 # Columns the finaliser reads (bulk_upload_finaliser_orchestrator). UPRN_COL = "address2uprn_uprn" MATCHED_ADDRESS_COL = "address2uprn_address" LEXISCORE_COL = "address2uprn_lexiscore" SOURCE_COL = "resolution_source" _RESULT_COLS = (UPRN_COL, MATCHED_ADDRESS_COL, LEXISCORE_COL, SOURCE_COL) # A resolved hit: (uprn, matched_address, lexiscore, source). Resolution = tuple[str, str, float, str] def load_keys() -> tuple[Optional[str], Optional[str]]: """Load (epc_token, os_api_key) from backend/.env (and the process env).""" load_dotenv(_REPO_ROOT / "backend" / ".env") epc_token = os.environ.get("OPEN_EPC_API_TOKEN") os_api_key = os.environ.get("ORDNANCE_SURVEY_API_KEY") return epc_token, os_api_key def read_rows(path: Path) -> tuple[list[dict[str, str]], list[str]]: """Read a CSV into (rows, fieldnames). Preserves column order.""" with path.open(newline="", encoding="utf-8-sig") as fh: reader = csv.DictReader(fh) fieldnames = list(reader.fieldnames or []) rows = [dict(row) for row in reader] return rows, fieldnames def clean_postcode(postcode: str) -> str: """Sanitise to the no-space upper form the EPC/OS lookups expect (e.g. E84SQ).""" return postcode.upper().replace(" ", "").strip() def build_address(row: dict[str, str]) -> str: """Concatenate Address 1/2/3 the same way the address2uprn lambda does.""" return " ".join( str(row.get(col, "") or "").strip() for col in ("Address 1", "Address 2", "Address 3") ).strip() def resolve_epc( address: str, postcode_clean: str, epc_cache: dict[str, pd.DataFrame] ) -> Optional[Resolution]: """Resolve via the new EPC API (cached per postcode), then historic EPC S3. `epc_cache` is mutated to memoise one EPC API call per postcode — pass the same dict across rows so a postcode is only fetched once. """ epc_df = epc_cache.get(postcode_clean) if epc_df is None: epc_df = get_epc_data_with_postcode(postcode=postcode_clean) epc_cache[postcode_clean] = epc_df result = get_uprn_with_epc_df( user_inputed_address=address, epc_df=epc_df, verbose=True ) if isinstance(result, tuple): uprn, matched, score = result return str(uprn), str(matched), float(score), "epc" historic = get_uprn_from_historic_epc( user_inputed_address=address, postcode=postcode_clean ) if historic is not None: uprn, matched, score = historic return str(uprn), str(matched), float(score), "epc_historic" return None def resolve_os( address: str, postcode_clean: str, os_api_key: str, os_cache: dict[str, pd.DataFrame], threshold: float, ) -> Optional[Resolution]: """Resolve via the OS Places API: best-scoring address above `threshold`. `os_cache` memoises one OS Places call per postcode. """ places_df = os_cache.get(postcode_clean) if places_df is None: response = lookup_os_places(postcode_clean, os_api_key) if response.get("status") != 200 or "data" not in response: places_df = pd.DataFrame() else: places_df = os_places_results_to_dataframe(response["data"]) os_cache[postcode_clean] = places_df if places_df.empty or "ADDRESS" not in places_df.columns: return None # Iterate plain records — avoids pandas' partially-unknown indexing types. records: list[dict[str, object]] = places_df.to_dict(orient="records") best: Optional[Resolution] = None for rec in records: candidate = str(rec.get("ADDRESS", "")) score = AddressMatch.score(address, candidate) if score >= threshold and (best is None or score > best[2]): best = (str(rec.get("UPRN", "")), candidate, score, "ordnance_survey") return best def resolve_row( row: dict[str, str], *, epc_token: Optional[str], os_api_key: Optional[str], epc_cache: dict[str, pd.DataFrame], os_cache: dict[str, pd.DataFrame], postcode_col: str, use_epc: bool, use_os: bool, os_threshold: float, validate_postcode: bool, ) -> dict[str, str]: """Resolve one row in place and return it with the finaliser columns filled. Tries EPC (new + historic) first, then OS Places. On no match the three result columns are written empty and `resolution_source` is "none". """ address = build_address(row) postcode_clean = clean_postcode(str(row.get(postcode_col, "") or "")) def write(res: Optional[Resolution]) -> dict[str, str]: if res is None: row[UPRN_COL] = "" row[MATCHED_ADDRESS_COL] = "" row[LEXISCORE_COL] = "" row[SOURCE_COL] = "none" else: uprn, matched, score, source = res row[UPRN_COL] = uprn row[MATCHED_ADDRESS_COL] = matched row[LEXISCORE_COL] = str(score) row[SOURCE_COL] = source return row if not address or not postcode_clean: return write(None) if validate_postcode and not AddressMatch.is_valid_postcode(postcode_clean): return write(None) if use_epc and epc_token: try: res = resolve_epc(address, postcode_clean, epc_cache) if res is not None: return write(res) except Exception as exc: # keep going on a per-row API/lookup failure print(f" EPC lookup failed for {address!r} / {postcode_clean}: {exc}") if use_os and os_api_key: try: res = resolve_os(address, postcode_clean, os_api_key, os_cache, os_threshold) if res is not None: return write(res) except Exception as exc: print(f" OS lookup failed for {address!r} / {postcode_clean}: {exc}") return write(None) def process( rows: list[dict[str, str]], *, epc_token: Optional[str], os_api_key: Optional[str], postcode_col: str = "postcode", use_epc: bool = True, use_os: bool = True, os_threshold: float = 0.5, validate_postcode: bool = True, ) -> list[dict[str, str]]: """Resolve every row, printing a per-row line so REPL/CLI progress is visible.""" epc_cache: dict[str, pd.DataFrame] = {} os_cache: dict[str, pd.DataFrame] = {} for i, row in enumerate(rows, start=1): resolve_row( row, epc_token=epc_token, os_api_key=os_api_key, epc_cache=epc_cache, os_cache=os_cache, postcode_col=postcode_col, use_epc=use_epc, use_os=use_os, os_threshold=os_threshold, validate_postcode=validate_postcode, ) print( f"[{i}/{len(rows)}] {build_address(row)!r} -> " f"{row[UPRN_COL] or '(no match)'} ({row[SOURCE_COL]})" ) return rows def write_rows(rows: list[dict[str, str]], path: Path, fieldnames: list[str]) -> None: """Write rows to CSV, preserving input columns and appending the result columns.""" out_fields = list(fieldnames) for col in _RESULT_COLS: if col not in out_fields: out_fields.append(col) with path.open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=out_fields, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", type=Path, help="input CSV (Address 1/2/3 + postcode)") parser.add_argument( "-o", "--out", type=Path, required=True, help="output CSV for the finaliser" ) parser.add_argument("--postcode-col", default="postcode", help="postcode column name") parser.add_argument("--no-epc", action="store_true", help="skip EPC resolution") parser.add_argument("--no-os", action="store_true", help="skip Ordnance Survey fallback") parser.add_argument( "--os-threshold", type=float, default=0.5, help="min OS match score (default 0.5)" ) parser.add_argument( "--no-validate-postcode", action="store_true", help="skip the postcodes.io validity check (one HTTP call per postcode)", ) parser.add_argument("--limit", type=int, default=None, help="process only the first N rows") return parser.parse_args() def main() -> int: args = _parse_args() epc_token, os_api_key = load_keys() use_epc = not args.no_epc use_os = not args.no_os if use_epc and not epc_token: print("OPEN_EPC_API_TOKEN not set (backend/.env) — EPC resolution disabled") use_epc = False if use_os and not os_api_key: print("ORDNANCE_SURVEY_API_KEY not set (backend/.env) — OS fallback disabled") use_os = False if not use_epc and not use_os: print("No resolver enabled (missing keys or both --no-* flags). Nothing to do.") return 2 rows, fieldnames = read_rows(args.input) if args.limit is not None: rows = rows[: args.limit] print(f"Loaded {len(rows)} rows from {args.input}") process( rows, epc_token=epc_token, os_api_key=os_api_key, postcode_col=args.postcode_col, use_epc=use_epc, use_os=use_os, os_threshold=args.os_threshold, validate_postcode=not args.no_validate_postcode, ) write_rows(rows, args.out, fieldnames) matched = sum(1 for r in rows if r.get(UPRN_COL)) print(f"\nResolved {matched}/{len(rows)} rows. Wrote {args.out}") return 0 if __name__ == "__main__": sys.exit(main())