diff --git a/backend/address2UPRN/main.py b/backend/address2UPRN/main.py index 599cbaa6b..fcc6a11c3 100644 --- a/backend/address2UPRN/main.py +++ b/backend/address2UPRN/main.py @@ -15,7 +15,11 @@ from utils.s3 import ( from datetime import datetime from backend.utils.addressMatch import AddressMatch -from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity +from backend.address2UPRN.scoring import ( + all_uprns_match, + rank_address_similarity, + resolve_group_ambiguity, +) from infrastructure.epc_client.epc_client_service import EpcClientService from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver from repositories.historic_epc.historic_epc_s3_repository import ( @@ -393,6 +397,7 @@ def handler(event, context, local=False): "address2uprn_uprn": "invalid postcode", "address2uprn_address": "invalid postcode", "address2uprn_lexiscore": "invalid postcode", + "address2uprn_status": "invalid_postcode", } ) continue @@ -409,7 +414,13 @@ def handler(event, context, local=False): ) continue - # Process each address in this postcode with the same EPC data + # Match each address in this postcode against the same EPC data, + # collecting per-row results first. Cross-row ambiguity is resolved + # for the whole group afterwards (ADR-0057) — a UPRN that is the + # best match for two *different* addresses is withheld, so distinct + # addresses are never coerced onto one UPRN (which the identity + # insert would then silently merge). + group_records: list[dict] = [] for row in postcode_rows: try: # Concatenate Address columns directly @@ -451,31 +462,38 @@ def handler(event, context, local=False): f"Historic EPC matched {address2uprn_user_input} in {postcode}" ) + norm_address = AddressMatch.normalise_address( + address2uprn_user_input + ) + # Parse result tuple if successful if result: uprn, found_address, score = result logger.info( f"Found UPRN for {address2uprn_user_input} in {postcode}: {uprn} (score: {score})" ) - - results_data.append( + group_records.append( { - **row, # Include all original data - "address2uprn_uprn": uprn, - "address2uprn_address": found_address, - "address2uprn_lexiscore": score, + "row": row, + "uprn": uprn, + "address": found_address, + "lexiscore": score, + "norm_address": norm_address, + "error": None, } ) else: logger.warning( f"No UPRN found for {address2uprn_user_input} in {postcode}" ) - results_data.append( + group_records.append( { - **row, # Include all original data - "address2uprn_uprn": None, - "address2uprn_address": None, - "address2uprn_lexiscore": None, + "row": row, + "uprn": None, + "address": None, + "lexiscore": None, + "norm_address": norm_address, + "error": None, } ) @@ -483,18 +501,39 @@ def handler(event, context, local=False): logger.error( f"Error processing address {row.get('address2uprn_user_input', 'unknown')}: {e}" ) - # Still add the row with error markers - results_data.append( + # Still record the row with error markers + group_records.append( { - **row, - "address2uprn_uprn": None, - "address2uprn_address": None, - "address2uprn_lexiscore": None, + "row": row, + "uprn": None, + "address": None, + "lexiscore": None, + "norm_address": "", "error": str(e), } ) continue + # Resolve cross-row ambiguity for the whole postcode group, then + # emit. A withheld (ambiguous) UPRN drops to None but keeps its + # lexiscore so the confirmation page can triage it (ADR-0057). + decisions = resolve_group_ambiguity( + [(rec["uprn"], rec["norm_address"]) for rec in group_records] + ) + for rec, (final_uprn, status) in zip(group_records, decisions): + emitted = { + **rec["row"], # Include all original data + "address2uprn_uprn": final_uprn, + "address2uprn_address": rec["address"] if final_uprn else None, + "address2uprn_lexiscore": rec["lexiscore"], + "address2uprn_status": ( + "error" if rec["error"] is not None else status + ), + } + if rec["error"] is not None: + emitted["error"] = rec["error"] + results_data.append(emitted) + # Create results DataFrame result_df = pd.DataFrame(results_data) diff --git a/backend/address2UPRN/scoring.py b/backend/address2UPRN/scoring.py index dcb86d498..cc3082fbc 100644 --- a/backend/address2UPRN/scoring.py +++ b/backend/address2UPRN/scoring.py @@ -1,8 +1,50 @@ +from collections import defaultdict +from typing import Optional + import pandas as pd from backend.utils.addressMatch import AddressMatch +def resolve_group_ambiguity( + matches: list[tuple[Optional[str], str]], +) -> list[tuple[Optional[str], str]]: + """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. + """ + 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]] = [] + for uprn, _norm_address in matches: + if not uprn: + resolved.append((None, "unmatched")) + elif len(distinct_addresses[uprn]) > 1: + resolved.append((None, "ambiguous_duplicate")) + else: + resolved.append((uprn, "matched")) + return resolved + + def all_uprns_match( df: pd.DataFrame, target_uprn: str, diff --git a/repositories/property/property_override_postgres_repository.py b/repositories/property/property_override_postgres_repository.py index be69292aa..d44a2ad65 100644 --- a/repositories/property/property_override_postgres_repository.py +++ b/repositories/property/property_override_postgres_repository.py @@ -31,6 +31,17 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository): return 0 now = datetime.now(timezone.utc) + # Defensive backstop (ADR-0057): a single INSERT ... ON CONFLICT DO + # UPDATE may not touch the same target row twice — two batch rows sharing + # the conflict key (property_id, override_component, building_part) make + # Postgres raise CardinalityViolation ("cannot affect row a second + # time"). ADR-0057 stops distinct addresses being coerced onto one UPRN + # upstream, so this should no longer arise from real data; collapse to + # the last occurrence (last-write-wins, the same outcome DO UPDATE gives + # across separate statements) so the statement can never be invalid. + deduped: dict[tuple[int, str, int], PropertyOverrideInsert] = { + (r.property_id, r.override_component, r.building_part): r for r in rows + } values = [ { "property_id": r.property_id, @@ -42,7 +53,7 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository): "created_at": now, "updated_at": now, } - for r in rows + for r in deduped.values() ] stmt = pg_insert(self._table).values(values) diff --git a/repositories/property/property_override_repository.py b/repositories/property/property_override_repository.py index 0a1a0a2e8..9a57d3baf 100644 --- a/repositories/property/property_override_repository.py +++ b/repositories/property/property_override_repository.py @@ -34,5 +34,9 @@ class PropertyOverrideRepository(ABC): """Upsert each row on ``(property_id, override_component, building_part)``, refreshing ``override_value`` + ``original_spreadsheet_description`` + ``updated_at`` on conflict (recalculate-on-rerun). Returns the number of - rows affected; an empty list is a no-op returning 0.""" + rows affected; an empty list is a no-op returning 0. + + Rows sharing a conflict key within one call are collapsed to the last + occurrence (last-write-wins) before writing, so a single batch never + upserts the same target row twice (ADR-0057 backstop).""" ...