refactor(address2uprn): name the match/decision return types; rename helper

Address PR review (dancafc):
- introduce UprnMatch NamedTuple (datatypes/address_match.py) for the
  (uprn, address, lexiscore, certificate_number) return, replacing the bare
  4-tuple in get_uprn_from_epc_df / get_uprn_from_historic_epc /
  HistoricEpcResolver.resolve_uprn. Tuple-compatible, so unpacking is unchanged.
- rename get_uprn_with_epc_df -> get_uprn_from_epc_df (+ callers).
- type resolve_group_ambiguity via a GroupDecision NamedTuple and trim its
  docstring.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/address2UPRN/main.py

@@ -14,6 +14,7 @@ from utils.s3 import (
 )
 from datetime import datetime
 
+from datatypes.address_match import UprnMatch
 from backend.utils.addressMatch import AddressMatch
 from backend.address2UPRN.scoring import (
     all_uprns_match,
@@ -52,7 +53,7 @@ def get_epc_data_with_postcode(postcode: str) -> pd.DataFrame:
 def get_uprn_from_historic_epc(
     user_inputed_address: str,
     postcode: str,
-) -> Optional[tuple[str, str, float, Optional[str]]]:
+) -> Optional[UprnMatch]:
     """Resolve a UPRN via historic EPC S3 data.
 
     Returns (uprn, address, lexiscore, certificate_number) when the historic
@@ -66,11 +67,11 @@ def get_uprn_from_historic_epc(
     return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode)
 
 
-def get_uprn_with_epc_df(
+def get_uprn_from_epc_df(
     user_inputed_address: str,
     epc_df: pd.DataFrame,
     verbose: bool = False,
-) -> Optional[str | tuple[str, str, float, Optional[str]]]:
+) -> Optional[str | UprnMatch]:
     """
     Return uprn (str) using a pre-fetched EPC dataframe.
     This avoids calling the API multiple times for the same postcode.
@@ -110,7 +111,7 @@ def get_uprn_with_epc_df(
         return None
 
     if verbose:
-        return (found_uprn, address, score, certificate_number)
+        return UprnMatch(found_uprn, address, score, certificate_number)
     else:
         return found_uprn
 
@@ -128,11 +129,11 @@ def get_uprn(
     back to the historic EPC dataset on S3.
 
     For processing multiple addresses in the same postcode, use
-    get_uprn_with_epc_df instead.
+    get_uprn_from_epc_df instead.
     """
     df = get_epc_data_with_postcode(postcode=postcode)
 
-    result: Optional[tuple[str, str, float, Optional[str]]] = get_uprn_with_epc_df(
+    result: Optional[str | UprnMatch] = get_uprn_from_epc_df(
         user_inputed_address=user_inputed_address,
         epc_df=df,
         verbose=True,
@@ -448,9 +449,7 @@ def handler(event, context, local=False):
                             continue
 
                         # Get UPRN using the pre-fetched EPC data with all return options
-                        result: Optional[
-                            tuple[str, str, float, Optional[str]]
-                        ] = get_uprn_with_epc_df(
+                        result: Optional[UprnMatch] = get_uprn_from_epc_df(
                             user_inputed_address=address2uprn_user_input,
                             epc_df=epc_df,
                             verbose=True,

backend/address2UPRN/scoring.py

@@ -1,47 +1,42 @@
 from collections import defaultdict
-from typing import Optional
+from typing import NamedTuple, Optional
 
 import pandas as pd
 
 from backend.utils.addressMatch import AddressMatch
 
 
+class GroupDecision(NamedTuple):
+    """One row's outcome after cross-row ambiguity resolution (ADR-0057)."""
+
+    uprn: Optional[str]
+    status: str  # "matched" | "ambiguous_duplicate" | "unmatched"
+
+
 def resolve_group_ambiguity(
     matches: list[tuple[Optional[str], str]],
-) -> list[tuple[Optional[str], str]]:
+) -> list[GroupDecision]:
     """Resolve cross-row UPRN ambiguity within one postcode group (ADR-0057).
 
-    ``matches`` is ``(uprn, normalised_address)`` per row, in order. Each row is
-    matched independently, so nothing stops one UPRN being the best match for
-    two *different* addresses — almost always a coarse EPC record absorbing
-    several real addresses (e.g. flats in a block matched to a flat-less
-    record). Withhold that UPRN on every such row, so a distinct address is
-    never coerced onto a shared UPRN: the downstream ``property`` identity
-    insert (``on_conflict_do_nothing`` on ``(portfolio_id, uprn)``) would
-    otherwise silently merge them, and the ``property_overrides`` upsert would
-    then collide on ``(property_id, override_component, building_part)``.
-
-    A UPRN shared only by rows with the *same* normalised address is a genuine
-    re-listing of one property and is kept.
-
-    Returns ``(uprn, status)`` per row in input order, where status is
-    ``"matched"`` (kept), ``"ambiguous_duplicate"`` (withheld → uprn ``None``),
-    or ``"unmatched"`` (input uprn was already ``None``). Withheld rows keep
-    their lexiscore upstream for triage on the confirmation page.
+    ``matches`` is ``(uprn, normalised_address)`` per row. A UPRN that is the
+    best match for two rows with *different* normalised addresses is withheld
+    on both (a coarse EPC record absorbing several real addresses, e.g. flats in
+    a block); a UPRN shared only by identical addresses is a genuine re-listing
+    and kept. Returns a ``GroupDecision`` per row, in input order.
     """
     distinct_addresses: dict[str, set[str]] = defaultdict(set)
     for uprn, norm_address in matches:
         if uprn:
             distinct_addresses[uprn].add(norm_address)
 
-    resolved: list[tuple[Optional[str], str]] = []
+    resolved: list[GroupDecision] = []
     for uprn, _norm_address in matches:
         if not uprn:
-            resolved.append((None, "unmatched"))
+            resolved.append(GroupDecision(None, "unmatched"))
         elif len(distinct_addresses[uprn]) > 1:
-            resolved.append((None, "ambiguous_duplicate"))
+            resolved.append(GroupDecision(None, "ambiguous_duplicate"))
         else:
-            resolved.append((uprn, "matched"))
+            resolved.append(GroupDecision(uprn, "matched"))
     return resolved
 
 

datatypes/address_match.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from typing import NamedTuple, Optional
+
+
+class UprnMatch(NamedTuple):
+    """A confident address→UPRN match from either EPC source (new API or
+    historic). Tuple-compatible, so existing unpacking keeps working."""
+
+    uprn: str
+    address: str
+    lexiscore: float
+    certificate_number: Optional[str]

repositories/historic_epc/historic_epc_resolver.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import Optional
 
+from datatypes.address_match import UprnMatch
 from datatypes.epc.domain.historic_epc import HistoricEpc
 from datatypes.epc.domain.historic_epc_matching import (
     HistoricEpcMatches,
@@ -47,12 +48,10 @@ class HistoricEpcResolver:
             return None
         return max(certs, key=lambda r: r.lodgement_date)
 
-    def resolve_uprn(
-        self, user_address: str, postcode: str
-    ) -> Optional[tuple[str, str, float, Optional[str]]]:
-        """``(uprn, matched_address, lexiscore, certificate_number)`` for an
-        unambiguous rank-1 match, else None (no data / ambiguous tie / zero
-        score). ``certificate_number`` is the historic dataset's ``lmk_key``."""
+    def resolve_uprn(self, user_address: str, postcode: str) -> Optional[UprnMatch]:
+        """A ``UprnMatch`` for an unambiguous rank-1 match, else None (no data /
+        ambiguous tie / zero score). ``certificate_number`` is the historic
+        dataset's ``lmk_key``."""
         matches: HistoricEpcMatches = self.match(user_address, postcode)
         uprn: Optional[str] = matches.unambiguous_uprn()
         if not uprn or uprn == "nan":
@@ -60,4 +59,4 @@ class HistoricEpcResolver:
         top: Optional[ScoredHistoricEpc] = matches.top()
         if top is None:
             return None
-        return uprn, top.record.address, top.lexiscore, top.record.lmk_key
+        return UprnMatch(uprn, top.record.address, top.lexiscore, top.record.lmk_key)

scripts/resolve_uprns_for_finaliser.py

@@ -47,7 +47,7 @@ sys.path.insert(0, str(_REPO_ROOT))  # worktree root first — avoid the import
 from backend.address2UPRN.main import (  # noqa: E402
     get_epc_data_with_postcode,
     get_uprn_from_historic_epc,
-    get_uprn_with_epc_df,
+    get_uprn_from_epc_df,
 )
 from backend.ordnanceSurvey.helpers import (  # noqa: E402
     lookup_os_places,
@@ -108,7 +108,7 @@ def resolve_epc(
         epc_df = get_epc_data_with_postcode(postcode=postcode_clean)
         epc_cache[postcode_clean] = epc_df
 
-    result = get_uprn_with_epc_df(
+    result = get_uprn_from_epc_df(
         user_inputed_address=address, epc_df=epc_df, verbose=True
     )
     if isinstance(result, tuple):