From 371874380115febec25753cec19189efdbfa16ae Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 13:26:49 +0000 Subject: [PATCH 01/13] Carry the EPC certificate number through address2uprn to property EpcClientService.search_by_postcode already returns the matched certificate number alongside the UPRN, but it was dropped before persistence. Thread it through get_epc_data_with_postcode -> get_uprn_with_epc_df / get_uprn_from_historic_epc (using the historic dataset's lmk_key) -> the address2uprn_certificate_number result column -> PropertyIdentityInsert -> the property table's new certificate_number column (assessment-model PR #362). Co-Authored-By: Claude Sonnet 5 --- backend/address2UPRN/main.py | 40 +++++++++++++------ infrastructure/postgres/property_table.py | 1 + .../bulk_upload_finaliser_orchestrator.py | 7 ++++ .../historic_epc/historic_epc_resolver.py | 9 +++-- .../property/property_postgres_repository.py | 1 + repositories/property/property_repository.py | 7 +++- ...test_bulk_upload_finaliser_orchestrator.py | 4 ++ .../test_historic_epc_resolver.py | 3 +- 8 files changed, 52 insertions(+), 20 deletions(-) diff --git a/backend/address2UPRN/main.py b/backend/address2UPRN/main.py index 599cbaa6b..acebe03ca 100644 --- a/backend/address2UPRN/main.py +++ b/backend/address2UPRN/main.py @@ -34,22 +34,29 @@ def get_epc_data_with_postcode(postcode: str) -> pd.DataFrame: service = EpcClientService(auth_token=token) results = service.search_by_postcode(postcode) return pd.DataFrame( - [{"address": r.address_line_1, "uprn": r.uprn} for r in results] + [ + { + "address": r.address_line_1, + "uprn": r.uprn, + "certificate_number": r.certificate_number, + } + for r in results + ] ) def get_uprn_from_historic_epc( user_inputed_address: str, postcode: str, -) -> Optional[tuple[str, str, float]]: +) -> Optional[tuple[str, str, float, Optional[str]]]: """Resolve a UPRN via historic EPC S3 data. - Returns (uprn, address, lexiscore) when the historic dataset agrees on a - single rank-1 UPRN, None otherwise (no stored data, zero score, or - ambiguous top rank). The score gate is `unambiguous_uprn`'s own (score > 0); - the 0.7 heuristic used for the new-EPC source isn't applied here because - historic addresses use a more verbose format that systematically depresses - lexiscores. + Returns (uprn, address, lexiscore, certificate_number) when the historic + dataset agrees on a single rank-1 UPRN, None otherwise (no stored data, + zero score, or ambiguous top rank). The score gate is `unambiguous_uprn`'s + own (score > 0); the 0.7 heuristic used for the new-EPC source isn't + applied here because historic addresses use a more verbose format that + systematically depresses lexiscores. """ repo = HistoricEpcS3Repository.with_default_s3_client() return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode) @@ -59,7 +66,7 @@ def get_uprn_with_epc_df( user_inputed_address: str, epc_df: pd.DataFrame, verbose: bool = False, -) -> Optional[str | tuple[str, str, float]]: +) -> Optional[str | tuple[str, str, float, Optional[str]]]: """ Return uprn (str) using a pre-fetched EPC dataframe. This avoids calling the API multiple times for the same postcode. @@ -88,6 +95,7 @@ def get_uprn_with_epc_df( address = top_rank_df["address"].values[0] score = float(top_rank_df["lexiscore"].values[0]) + certificate_number = top_rank_df["certificate_number"].values[0] logger.info(f"Address found to be: {address}, with lexiscore {score}") # Safe to return the agreed UPRN @@ -98,7 +106,7 @@ def get_uprn_with_epc_df( return None if verbose: - return (found_uprn, address, score) + return (found_uprn, address, score, certificate_number) else: return found_uprn @@ -120,7 +128,7 @@ def get_uprn( """ df = get_epc_data_with_postcode(postcode=postcode) - result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df( + result: Optional[tuple[str, str, float, Optional[str]]] = get_uprn_with_epc_df( user_inputed_address=user_inputed_address, epc_df=df, verbose=True, @@ -393,6 +401,7 @@ def handler(event, context, local=False): "address2uprn_uprn": "invalid postcode", "address2uprn_address": "invalid postcode", "address2uprn_lexiscore": "invalid postcode", + "address2uprn_certificate_number": "invalid postcode", } ) continue @@ -428,7 +437,9 @@ 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]] = get_uprn_with_epc_df( + result: Optional[ + tuple[str, str, float, Optional[str]] + ] = get_uprn_with_epc_df( user_inputed_address=address2uprn_user_input, epc_df=epc_df, verbose=True, @@ -453,7 +464,7 @@ def handler(event, context, local=False): # Parse result tuple if successful if result: - uprn, found_address, score = result + uprn, found_address, score, certificate_number = result logger.info( f"Found UPRN for {address2uprn_user_input} in {postcode}: {uprn} (score: {score})" ) @@ -464,6 +475,7 @@ def handler(event, context, local=False): "address2uprn_uprn": uprn, "address2uprn_address": found_address, "address2uprn_lexiscore": score, + "address2uprn_certificate_number": certificate_number, } ) else: @@ -476,6 +488,7 @@ def handler(event, context, local=False): "address2uprn_uprn": None, "address2uprn_address": None, "address2uprn_lexiscore": None, + "address2uprn_certificate_number": None, } ) @@ -490,6 +503,7 @@ def handler(event, context, local=False): "address2uprn_uprn": None, "address2uprn_address": None, "address2uprn_lexiscore": None, + "address2uprn_certificate_number": None, "error": str(e), } ) diff --git a/infrastructure/postgres/property_table.py b/infrastructure/postgres/property_table.py index 56d0b2fa7..9664b7885 100644 --- a/infrastructure/postgres/property_table.py +++ b/infrastructure/postgres/property_table.py @@ -49,6 +49,7 @@ class PropertyRow(SQLModel, table=True): user_inputted_address: Optional[str] = Field(default=None) user_inputted_postcode: Optional[str] = Field(default=None) lexiscore: Optional[float] = Field(default=None) + certificate_number: Optional[str] = Field(default=None) # FE-owned columns the modelling pipeline now WRITES to record a run: the old # engine set `has_recommendations` (engine.py); we mirror that, and bump diff --git a/orchestration/bulk_upload_finaliser_orchestrator.py b/orchestration/bulk_upload_finaliser_orchestrator.py index 1d707a8d7..534835e20 100644 --- a/orchestration/bulk_upload_finaliser_orchestrator.py +++ b/orchestration/bulk_upload_finaliser_orchestrator.py @@ -41,6 +41,7 @@ INTERNAL_REF_COL = "Internal Reference" UPRN_COL = "address2uprn_uprn" MATCHED_ADDRESS_COL = "address2uprn_address" LEXISCORE_COL = "address2uprn_lexiscore" +CERTIFICATE_NUMBER_COL = "address2uprn_certificate_number" MISSING_SENTINEL = "invalid postcode" UK_POSTCODE_RE = re.compile(r"[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}", re.IGNORECASE) @@ -362,6 +363,11 @@ class BulkUploadFinaliserOrchestrator: internal_ref = _normalize(raw.get(INTERNAL_REF_COL)) or None lexiscore = _parse_lexiscore(raw.get(LEXISCORE_COL)) + certificate_number_raw = _normalize(raw.get(CERTIFICATE_NUMBER_COL)) + certificate_number = ( + None if _is_missing(certificate_number_raw) else certificate_number_raw + ) + return PropertyIdentityInsert( portfolio_id=portfolio_id, uprn=uprn, @@ -371,5 +377,6 @@ class BulkUploadFinaliserOrchestrator: user_inputted_address=user_inputted_address, user_inputted_postcode=user_inputted_postcode, lexiscore=lexiscore, + certificate_number=certificate_number, creation_status="READY", ) diff --git a/repositories/historic_epc/historic_epc_resolver.py b/repositories/historic_epc/historic_epc_resolver.py index 21a7457be..37413f6c7 100644 --- a/repositories/historic_epc/historic_epc_resolver.py +++ b/repositories/historic_epc/historic_epc_resolver.py @@ -49,9 +49,10 @@ class HistoricEpcResolver: def resolve_uprn( self, user_address: str, postcode: str - ) -> Optional[tuple[str, str, float]]: - """``(uprn, matched_address, lexiscore)`` for an unambiguous rank-1 - match, else None (no data / ambiguous tie / zero score).""" + ) -> 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``.""" matches: HistoricEpcMatches = self.match(user_address, postcode) uprn: Optional[str] = matches.unambiguous_uprn() if not uprn or uprn == "nan": @@ -59,4 +60,4 @@ class HistoricEpcResolver: top: Optional[ScoredHistoricEpc] = matches.top() if top is None: return None - return uprn, top.record.address, top.lexiscore + return uprn, top.record.address, top.lexiscore, top.record.lmk_key diff --git a/repositories/property/property_postgres_repository.py b/repositories/property/property_postgres_repository.py index 1072c18d8..57fb9ef54 100644 --- a/repositories/property/property_postgres_repository.py +++ b/repositories/property/property_postgres_repository.py @@ -162,6 +162,7 @@ class PropertyPostgresRepository(PropertyRepository): "user_inputted_address": r.user_inputted_address, "user_inputted_postcode": r.user_inputted_postcode, "lexiscore": r.lexiscore, + "certificate_number": r.certificate_number, } for r in rows ] diff --git a/repositories/property/property_repository.py b/repositories/property/property_repository.py index e2f8284ae..3911ab3ca 100644 --- a/repositories/property/property_repository.py +++ b/repositories/property/property_repository.py @@ -13,8 +13,10 @@ class PropertyIdentityInsert: """One row inserted into the FE-owned ``property`` table at Finalise (ADR-0013). Mirrors the exact column set today's Next.js ``/finalize`` writes: nine fields - plus ``creation_status='READY'``. ``address``/``postcode`` are the resolved - (matched ?? user-inputted) values and may be ``None``. + plus ``creation_status='READY'``, plus ``certificate_number`` (the EPC + certificate the address2uprn lookup matched, not written by the old + Next.js route). ``address``/``postcode`` are the resolved (matched ?? + user-inputted) values and may be ``None``. """ portfolio_id: int @@ -25,6 +27,7 @@ class PropertyIdentityInsert: user_inputted_address: Optional[str] user_inputted_postcode: Optional[str] lexiscore: Optional[float] + certificate_number: Optional[str] = None creation_status: str = "READY" diff --git a/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py b/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py index 335a3e919..0462e9792 100644 --- a/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py +++ b/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py @@ -100,6 +100,7 @@ def test_finalise_inserts_resolved_rows_and_marks_complete() -> None: "address2uprn_uprn": "100023", "address2uprn_address": "1 SOME STREET, TOWN, SW1A 1AA", "address2uprn_lexiscore": "0.95", + "address2uprn_certificate_number": "1234-5678-9012-3456-7890", }, ] @@ -114,6 +115,7 @@ def test_finalise_inserts_resolved_rows_and_marks_complete() -> None: assert row.postcode == "SW1A 1AA" # extracted from the matched address assert row.user_inputted_address == "1 Some Street" assert row.lexiscore == 0.95 + assert row.certificate_number == "1234-5678-9012-3456-7890" assert row.creation_status == "READY" # The upload is marked complete via the injected status writer. assert status.calls == [(task_id, "complete")] @@ -128,6 +130,7 @@ def test_finalise_falls_back_to_user_input_when_unmatched() -> None: "address2uprn_uprn": "invalid postcode", "address2uprn_address": "invalid postcode", "address2uprn_lexiscore": "", + "address2uprn_certificate_number": "invalid postcode", }, ] @@ -138,6 +141,7 @@ def test_finalise_falls_back_to_user_input_when_unmatched() -> None: assert row.address == "2 Other Road" # falls back to user input assert row.postcode == "B1 1BB" # falls back to user-inputted postcode assert row.lexiscore is None + assert row.certificate_number is None # missing sentinel -> None def test_finalise_parses_pandas_float_uprn() -> None: diff --git a/tests/repositories/historic_epc/test_historic_epc_resolver.py b/tests/repositories/historic_epc/test_historic_epc_resolver.py index c4b05f8b7..f28787c1b 100644 --- a/tests/repositories/historic_epc/test_historic_epc_resolver.py +++ b/tests/repositories/historic_epc/test_historic_epc_resolver.py @@ -73,10 +73,11 @@ def test_resolve_uprn_returns_unambiguous_match(): # Assert assert result is not None - uprn, address, score = result + uprn, address, score, certificate_number = result assert uprn == "100" assert address == "47 GORDON ROAD" assert score > 0 + assert certificate_number == "" def test_resolve_uprn_is_none_when_postcode_has_no_data(): From 7abc71ed95973285094569fa33ddc3436614c152 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 14:16:55 +0000 Subject: [PATCH 02/13] Drop property-table wiring for certificate_number MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On reflection the certificate number only needs to travel through the address2uprn pipeline (result CSV, S3 output) as an internal value — not persisted to property. Reverts the PropertyIdentityInsert / property_table.py / property_postgres_repository.py changes; keeps certificate_number flowing through get_epc_data_with_postcode, get_uprn_with_epc_df, get_uprn_from_historic_epc, and the address2uprn_certificate_number result column. Co-Authored-By: Claude Sonnet 5 --- infrastructure/postgres/property_table.py | 1 - orchestration/bulk_upload_finaliser_orchestrator.py | 7 ------- repositories/property/property_postgres_repository.py | 1 - repositories/property/property_repository.py | 7 ++----- .../test_bulk_upload_finaliser_orchestrator.py | 4 ---- 5 files changed, 2 insertions(+), 18 deletions(-) diff --git a/infrastructure/postgres/property_table.py b/infrastructure/postgres/property_table.py index 9664b7885..56d0b2fa7 100644 --- a/infrastructure/postgres/property_table.py +++ b/infrastructure/postgres/property_table.py @@ -49,7 +49,6 @@ class PropertyRow(SQLModel, table=True): user_inputted_address: Optional[str] = Field(default=None) user_inputted_postcode: Optional[str] = Field(default=None) lexiscore: Optional[float] = Field(default=None) - certificate_number: Optional[str] = Field(default=None) # FE-owned columns the modelling pipeline now WRITES to record a run: the old # engine set `has_recommendations` (engine.py); we mirror that, and bump diff --git a/orchestration/bulk_upload_finaliser_orchestrator.py b/orchestration/bulk_upload_finaliser_orchestrator.py index 534835e20..1d707a8d7 100644 --- a/orchestration/bulk_upload_finaliser_orchestrator.py +++ b/orchestration/bulk_upload_finaliser_orchestrator.py @@ -41,7 +41,6 @@ INTERNAL_REF_COL = "Internal Reference" UPRN_COL = "address2uprn_uprn" MATCHED_ADDRESS_COL = "address2uprn_address" LEXISCORE_COL = "address2uprn_lexiscore" -CERTIFICATE_NUMBER_COL = "address2uprn_certificate_number" MISSING_SENTINEL = "invalid postcode" UK_POSTCODE_RE = re.compile(r"[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}", re.IGNORECASE) @@ -363,11 +362,6 @@ class BulkUploadFinaliserOrchestrator: internal_ref = _normalize(raw.get(INTERNAL_REF_COL)) or None lexiscore = _parse_lexiscore(raw.get(LEXISCORE_COL)) - certificate_number_raw = _normalize(raw.get(CERTIFICATE_NUMBER_COL)) - certificate_number = ( - None if _is_missing(certificate_number_raw) else certificate_number_raw - ) - return PropertyIdentityInsert( portfolio_id=portfolio_id, uprn=uprn, @@ -377,6 +371,5 @@ class BulkUploadFinaliserOrchestrator: user_inputted_address=user_inputted_address, user_inputted_postcode=user_inputted_postcode, lexiscore=lexiscore, - certificate_number=certificate_number, creation_status="READY", ) diff --git a/repositories/property/property_postgres_repository.py b/repositories/property/property_postgres_repository.py index 57fb9ef54..1072c18d8 100644 --- a/repositories/property/property_postgres_repository.py +++ b/repositories/property/property_postgres_repository.py @@ -162,7 +162,6 @@ class PropertyPostgresRepository(PropertyRepository): "user_inputted_address": r.user_inputted_address, "user_inputted_postcode": r.user_inputted_postcode, "lexiscore": r.lexiscore, - "certificate_number": r.certificate_number, } for r in rows ] diff --git a/repositories/property/property_repository.py b/repositories/property/property_repository.py index 3911ab3ca..e2f8284ae 100644 --- a/repositories/property/property_repository.py +++ b/repositories/property/property_repository.py @@ -13,10 +13,8 @@ class PropertyIdentityInsert: """One row inserted into the FE-owned ``property`` table at Finalise (ADR-0013). Mirrors the exact column set today's Next.js ``/finalize`` writes: nine fields - plus ``creation_status='READY'``, plus ``certificate_number`` (the EPC - certificate the address2uprn lookup matched, not written by the old - Next.js route). ``address``/``postcode`` are the resolved (matched ?? - user-inputted) values and may be ``None``. + plus ``creation_status='READY'``. ``address``/``postcode`` are the resolved + (matched ?? user-inputted) values and may be ``None``. """ portfolio_id: int @@ -27,7 +25,6 @@ class PropertyIdentityInsert: user_inputted_address: Optional[str] user_inputted_postcode: Optional[str] lexiscore: Optional[float] - certificate_number: Optional[str] = None creation_status: str = "READY" diff --git a/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py b/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py index 0462e9792..335a3e919 100644 --- a/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py +++ b/tests/orchestration/test_bulk_upload_finaliser_orchestrator.py @@ -100,7 +100,6 @@ def test_finalise_inserts_resolved_rows_and_marks_complete() -> None: "address2uprn_uprn": "100023", "address2uprn_address": "1 SOME STREET, TOWN, SW1A 1AA", "address2uprn_lexiscore": "0.95", - "address2uprn_certificate_number": "1234-5678-9012-3456-7890", }, ] @@ -115,7 +114,6 @@ def test_finalise_inserts_resolved_rows_and_marks_complete() -> None: assert row.postcode == "SW1A 1AA" # extracted from the matched address assert row.user_inputted_address == "1 Some Street" assert row.lexiscore == 0.95 - assert row.certificate_number == "1234-5678-9012-3456-7890" assert row.creation_status == "READY" # The upload is marked complete via the injected status writer. assert status.calls == [(task_id, "complete")] @@ -130,7 +128,6 @@ def test_finalise_falls_back_to_user_input_when_unmatched() -> None: "address2uprn_uprn": "invalid postcode", "address2uprn_address": "invalid postcode", "address2uprn_lexiscore": "", - "address2uprn_certificate_number": "invalid postcode", }, ] @@ -141,7 +138,6 @@ def test_finalise_falls_back_to_user_input_when_unmatched() -> None: assert row.address == "2 Other Road" # falls back to user input assert row.postcode == "B1 1BB" # falls back to user-inputted postcode assert row.lexiscore is None - assert row.certificate_number is None # missing sentinel -> None def test_finalise_parses_pandas_float_uprn() -> None: From daa1cd79679ec99109f4029b0591bc79315d04b4 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 15:15:59 +0000 Subject: [PATCH 03/13] feat(address2uprn): withhold ambiguous cross-row UPRN matches (ADR-0057) Phase 1 of confirming UPRNs before finalise. address2uprn matched each row independently, so one UPRN could be the best match for two distinct addresses (a coarse EPC record absorbing several real addresses, e.g. flats in a block). Those distinct addresses were then silently merged by the property identity insert, and collided in property_overrides. resolve_group_ambiguity() withholds a UPRN claimed by >=2 distinct normalised addresses within a postcode group (keeps genuine same-address re-listings), and the handler now emits an address2uprn_status column (matched | ambiguous_duplicate | unmatched | invalid_postcode | error). Withheld rows drop to a null UPRN but keep their lexiscore for triage on the (upcoming) confirmation page. Also adds the ADR-0057 backstop dedup in property_overrides upsert_all so the ON CONFLICT statement can never double-touch a row. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/address2UPRN/main.py | 77 ++++++++++++++----- backend/address2UPRN/scoring.py | 42 ++++++++++ .../property_override_postgres_repository.py | 13 +++- .../property/property_override_repository.py | 6 +- 4 files changed, 117 insertions(+), 21 deletions(-) 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).""" ... From 269aade481178cc6f304624ba983d7c55787e852 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 15:21:04 +0000 Subject: [PATCH 04/13] test(address2uprn): cover ambiguity withholding + override dedup; add ADR-0057 - unit tests for resolve_group_ambiguity (distinct addresses withheld, same-address re-listing kept, order preserved) - Postgres integration tests for upsert_all's backstop dedup - ADR-0057 recording the "confirm UPRNs before finalise" decision Co-Authored-By: Claude Opus 4.8 (1M context) --- ...057-uprn-confirmation-precedes-finalise.md | 80 ++++++++++++++ tests/backend/address2UPRN/__init__.py | 0 .../test_resolve_group_ambiguity.py | 70 ++++++++++++ ...t_property_override_postgres_repository.py | 102 ++++++++++++++++++ 4 files changed, 252 insertions(+) create mode 100644 docs/adr/0057-uprn-confirmation-precedes-finalise.md create mode 100644 tests/backend/address2UPRN/__init__.py create mode 100644 tests/backend/address2UPRN/test_resolve_group_ambiguity.py create mode 100644 tests/repositories/property/test_property_override_postgres_repository.py diff --git a/docs/adr/0057-uprn-confirmation-precedes-finalise.md b/docs/adr/0057-uprn-confirmation-precedes-finalise.md new file mode 100644 index 000000000..81eb1185e --- /dev/null +++ b/docs/adr/0057-uprn-confirmation-precedes-finalise.md @@ -0,0 +1,80 @@ +# UPRN confirmation precedes finalise; the bulk-upload review is one two-tab page and the portfolio "unmatched" tab is retired + +## Status + +proposed + +## Context + +The bulk-upload pipeline is a chain of SQS-driven Lambda stages: +`address2uprn → landlord-overrides → combiner → (review) → finaliser`. +`address2uprn` matches each input row independently against per-postcode EPC +candidates, keeping the best candidate above a `lexiscore >= 0.7` floor. There +is **no cross-row uniqueness guard**: nothing stops one UPRN being the best +match for two *different* addresses. The textbook case is a coarse EPC record +`42 Moreton Road` (no flat token) that `Flat 1, 42 Moreton Road` and +`Flat 2, 42 Moreton Road` both clear — the building-number guard passes and the +flat guard never fires, so both distinct flats collapse onto one UPRN. + +Downstream this is corrosive, not just noisy: + +- The `property` identity insert is `on_conflict_do_nothing` on the partial + unique index `(portfolio_id, uprn) WHERE uprn IS NOT NULL`. Two distinct + addresses carrying the same UPRN therefore **merge into a single property** — + one address silently loses its identity. +- `_build_overrides` (the finaliser) keys `property_overrides` on + `(property_id, override_component, building_part)`. Both merged rows emit the + same tuple, so the single `INSERT ... ON CONFLICT DO UPDATE` raises + `CardinalityViolation` ("cannot affect row a second time"). + +The classifier and finaliser were built assuming a UPRN is always intact and +unique; it is not. Separately, a *post-onboarding* portfolio-level "Unmatched +properties" tab let users fix no-UPRN properties after the fact — too late (the +identity merge has already happened) and in a different place from the rest of +review. + +## Decision + +1. **`address2uprn` emits a UPRN only when confident *and* unambiguous.** Within + a postcode group, a UPRN that is the best match for two or more rows whose + *normalised* input addresses differ is **withheld** (dropped to null) on + every such row; a UPRN shared only by rows with the *same* normalised address + is a genuine re-listing and is kept. `address2uprn` emits an + `address2uprn_status` column (`matched | ambiguous_duplicate | unmatched | + invalid_postcode | error`); withheld rows keep their `lexiscore` for triage. + A `lexiscore` review band `[0.7, X)` that also withholds shaky matches is a + follow-on once `X` is calibrated (see Consequences). +2. **One pre-finalise review page, two tabs**, styled like the retired unmatched + tab: + - **Addresses** — the flagged rows (`unmatched` / `ambiguous_duplicate` / + low-score); the user resolves each via OS Places (reusing the existing + `MatchAddress` / `searchAddresses` / `useAssignUprn` flow + `postcode_search` + cache), producing a valid residential UPRN, or explicitly marks *no-UPRN*. + - **Classification** — the existing description→value verify gate, semantics + unchanged (every `Unknown` must be mapped; the finaliser still fails loud on + an unresolved value). + Finalise is enabled only when **both** tabs are clear. +3. The finaliser runs only after confirmation, so every row carries a confirmed + UPRN **or** an explicit no-UPRN (a first-class terminal state: `property` + identity with `uprn NULL`, no `property_overrides`, building-passport shows + the "not matched yet" card). +4. **Retire the portfolio-level "Unmatched properties" tab** — properties now + arrive pre-matched, so post-onboarding UPRN cleanup is unnecessary. +5. `PropertyOverridePostgresRepository.upsert_all` keeps a **defensive + last-write-wins dedup** as a backstop for the single-statement invariant, but + it is no longer the primary fix. + +## Consequences + +- Distinct addresses can no longer silently merge into one property; the + `CardinalityViolation` can no longer arise from real data. +- Onboarding reaching "complete" now implies confirmed identities, so portfolio + pages need no unmatched tab. +- The ambiguous/low-score subset adds bounded user work at review time; confident + rows pass untouched. +- The `lexiscore` review-band threshold `X` needs calibrating against a real + portfolio's score distribution before it is switched on; until then only the + ambiguous-duplicate rule (which needs no threshold) is active. +- Migration is safe stage-by-stage: withheld UPRNs are caught by the *existing* + unmatched tab until the two-tab page ships, and the tab is removed only after + it does — so an ambiguous property always has somewhere to be resolved. diff --git a/tests/backend/address2UPRN/__init__.py b/tests/backend/address2UPRN/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/address2UPRN/test_resolve_group_ambiguity.py b/tests/backend/address2UPRN/test_resolve_group_ambiguity.py new file mode 100644 index 000000000..1279cd1d6 --- /dev/null +++ b/tests/backend/address2UPRN/test_resolve_group_ambiguity.py @@ -0,0 +1,70 @@ +"""Unit tests for ``resolve_group_ambiguity`` (ADR-0057). + +Each address is matched independently, so one UPRN can be the best match for +two *different* addresses (a coarse EPC record absorbing several real +addresses). Withholding that UPRN is a pure function of a postcode group's +``(uprn, normalised_address)`` pairs — no S3, EPC API, or scoring needed — so +it is unit-tested directly here. +""" + +from __future__ import annotations + +from backend.address2UPRN.scoring import resolve_group_ambiguity + + +def test_distinct_addresses_sharing_one_uprn_are_withheld() -> None: + # Flat 1 and Flat 2 both matched the same coarse EPC UPRN — distinct + # addresses, so both are withheld rather than silently merged. + result = resolve_group_ambiguity( + [ + ("100", "flat 1 42 moreton road"), + ("100", "flat 2 42 moreton road"), + ] + ) + assert result == [(None, "ambiguous_duplicate"), (None, "ambiguous_duplicate")] + + +def test_same_address_listed_twice_keeps_the_uprn() -> None: + # A genuine re-listing of one property (identical normalised address) is a + # real duplicate, not an ambiguous match — the UPRN is kept on both. + result = resolve_group_ambiguity( + [ + ("100", "42 moreton road"), + ("100", "42 moreton road"), + ] + ) + assert result == [("100", "matched"), ("100", "matched")] + + +def test_distinct_uprns_are_each_matched() -> None: + result = resolve_group_ambiguity( + [ + ("100", "42 moreton road"), + ("101", "44 moreton road"), + ] + ) + assert result == [("100", "matched"), ("101", "matched")] + + +def test_none_uprn_is_unmatched() -> None: + result = resolve_group_ambiguity([(None, "99 nowhere lane")]) + assert result == [(None, "unmatched")] + + +def test_order_is_preserved_across_mixed_group() -> None: + # A withheld pair, an unmatched row, and a clean match — all in one group; + # the output aligns positionally with the input. + result = resolve_group_ambiguity( + [ + ("100", "flat 1 42 moreton road"), # ambiguous (with row 3) + (None, "no epc candidate"), # unmatched + ("100", "flat 2 42 moreton road"), # ambiguous (with row 0) + ("200", "sole match road"), # matched + ] + ) + assert result == [ + (None, "ambiguous_duplicate"), + (None, "unmatched"), + (None, "ambiguous_duplicate"), + ("200", "matched"), + ] diff --git a/tests/repositories/property/test_property_override_postgres_repository.py b/tests/repositories/property/test_property_override_postgres_repository.py new file mode 100644 index 000000000..635b32ee5 --- /dev/null +++ b/tests/repositories/property/test_property_override_postgres_repository.py @@ -0,0 +1,102 @@ +"""Integration tests for ``PropertyOverridePostgresRepository.upsert_all``. + +The conflict handling lives entirely in SQL (``INSERT ... ON CONFLICT +(property_id, override_component, building_part) DO UPDATE``), so it can only be +verified against a real Postgres -- the ``db_engine`` fixture in +``tests/conftest.py`` spins one up per test. The batch-dedup below is exactly +the case an in-memory fake cannot catch: only a real ``ON CONFLICT`` statement +raises ``CardinalityViolation`` when a batch would touch one target row twice +(the ADR-0057 backstop). +""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from sqlalchemy import Engine +from sqlmodel import Session, select + +from infrastructure.postgres.property_override_table import PropertyOverrideRow +from repositories.property.property_override_postgres_repository import ( + PropertyOverridePostgresRepository, +) +from repositories.property.property_override_repository import PropertyOverrideInsert + + +@pytest.fixture +def session(db_engine: Engine) -> Iterator[Session]: + with Session(db_engine) as s: + yield s + + +def _insert( + property_id: int, + override_value: str, + description: str, + *, + building_part: int = 0, + component: str = "property_type", +) -> PropertyOverrideInsert: + return PropertyOverrideInsert( + property_id=property_id, + portfolio_id=820, + building_part=building_part, + override_component=component, + override_value=override_value, + original_spreadsheet_description=description, + ) + + +def _all_rows(session: Session) -> list[PropertyOverrideRow]: + return list(session.exec(select(PropertyOverrideRow)).all()) + + +def test_duplicate_conflict_key_in_one_batch_collapses_to_last( + session: Session, +) -> None: + # arrange: two batch rows sharing the conflict key + # (property_id, override_component, building_part) — before the backstop this + # raised CardinalityViolation ("cannot affect row a second time"). + repo = PropertyOverridePostgresRepository(session) + + # act + repo.upsert_all( + [ + _insert(742961, "House", "House: Semi-Detached"), + _insert(742961, "House", "House: Mid-Terrace"), + ] + ) + session.commit() + + # assert: one row, last occurrence wins (matches ON CONFLICT DO UPDATE). + rows = _all_rows(session) + assert len(rows) == 1 + assert rows[0].override_value == "House" + assert rows[0].original_spreadsheet_description == "House: Mid-Terrace" + + +def test_distinct_conflict_keys_are_all_written(session: Session) -> None: + # arrange: dedup must only collapse exact conflict-key duplicates — a + # different property_id, building_part, or component is a distinct row. + repo = PropertyOverridePostgresRepository(session) + + # act + repo.upsert_all( + [ + _insert(742961, "House", "House: Semi-Detached"), + _insert(742962, "House", "House: Mid-Terrace"), # different property + _insert(742961, "House", "ext", building_part=1), # different part + _insert(742961, "SolidBrick", "solid", component="wall_type"), # component + ] + ) + session.commit() + + # assert + assert len(_all_rows(session)) == 4 + + +def test_empty_batch_is_a_noop(session: Session) -> None: + repo = PropertyOverridePostgresRepository(session) + assert repo.upsert_all([]) == 0 + assert _all_rows(session) == [] From 2e0cc7432cca729adfb8d3163db3c7c6e3e1b7bf Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Tue, 7 Jul 2026 15:23:44 +0000 Subject: [PATCH 05/13] terraform correction --- .github/workflows/deploy_terraform.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/deploy_terraform.yml b/.github/workflows/deploy_terraform.yml index 510ac147a..228b1fcea 100644 --- a/.github/workflows/deploy_terraform.yml +++ b/.github/workflows/deploy_terraform.yml @@ -820,7 +820,7 @@ jobs: # Deploy Hubspot ETL Lambda # ============================================================ hubspot_etl_lambda: - needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda, abri_lambda] + needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda, abri_api_lambda] uses: ./.github/workflows/_deploy_lambda.yml with: lambda_name: hubspot-etl-to-ara From dbcdf29bd900248628072189f23ee4d24a61b3e9 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 16:00:00 +0000 Subject: [PATCH 06/13] 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) --- backend/address2UPRN/main.py | 17 ++++---- backend/address2UPRN/scoring.py | 41 ++++++++----------- datatypes/address_match.py | 13 ++++++ .../historic_epc/historic_epc_resolver.py | 13 +++--- scripts/resolve_uprns_for_finaliser.py | 4 +- 5 files changed, 47 insertions(+), 41 deletions(-) create mode 100644 datatypes/address_match.py diff --git a/backend/address2UPRN/main.py b/backend/address2UPRN/main.py index ccab46b1e..a3f0b086a 100644 --- a/backend/address2UPRN/main.py +++ b/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, diff --git a/backend/address2UPRN/scoring.py b/backend/address2UPRN/scoring.py index cc3082fbc..d158a9680 100644 --- a/backend/address2UPRN/scoring.py +++ b/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 diff --git a/datatypes/address_match.py b/datatypes/address_match.py new file mode 100644 index 000000000..74ed9ce6e --- /dev/null +++ b/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] diff --git a/repositories/historic_epc/historic_epc_resolver.py b/repositories/historic_epc/historic_epc_resolver.py index 37413f6c7..eb7e37af3 100644 --- a/repositories/historic_epc/historic_epc_resolver.py +++ b/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) diff --git a/scripts/resolve_uprns_for_finaliser.py b/scripts/resolve_uprns_for_finaliser.py index c01a55ed4..b9a44a2ce 100644 --- a/scripts/resolve_uprns_for_finaliser.py +++ b/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): From 015db4275b0adf8432388d025c10168814a6d6ef Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:08:30 +0000 Subject: [PATCH 07/13] =?UTF-8?q?The=20plan=20repository=20reports=20which?= =?UTF-8?q?=20properties=20already=20have=20a=20default=20plan=20?= =?UTF-8?q?=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- repositories/plan/plan_postgres_repository.py | 3 +++ repositories/plan/plan_repository.py | 8 +++++++ tests/orchestration/fakes.py | 5 +++++ .../plan/test_plan_postgres_repository.py | 21 +++++++++++++++++++ 4 files changed, 37 insertions(+) diff --git a/repositories/plan/plan_postgres_repository.py b/repositories/plan/plan_postgres_repository.py index e81e90844..c2685ace4 100644 --- a/repositories/plan/plan_postgres_repository.py +++ b/repositories/plan/plan_postgres_repository.py @@ -45,6 +45,9 @@ class PlanPostgresRepository(PlanRepository): [PlanSaveRequest(plan, property_id=property_id, scenario_id=scenario_id, portfolio_id=portfolio_id, is_default=is_default)] )[0] + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + raise NotImplementedError + def save_batch(self, requests: list[PlanSaveRequest]) -> list[int]: """Persist all Plans in three statements regardless of batch size. diff --git a/repositories/plan/plan_repository.py b/repositories/plan/plan_repository.py index 6fdc16d63..149a88618 100644 --- a/repositories/plan/plan_repository.py +++ b/repositories/plan/plan_repository.py @@ -54,3 +54,11 @@ class PlanRepository(ABC): UPDATE only when at least one request has ``is_default=True``. Keeps prior Plans as history (ADR-0017).""" ... + + @abstractmethod + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + """The subset of ``property_ids`` that already have a default Plan. + A property outside this set is receiving its first Plan, which becomes + its default whatever the Scenario says (the legacy ``p.is_new`` rule — + readers need one default Plan per property).""" + ... diff --git a/tests/orchestration/fakes.py b/tests/orchestration/fakes.py index d5fe016a8..52f409693 100644 --- a/tests/orchestration/fakes.py +++ b/tests/orchestration/fakes.py @@ -240,6 +240,11 @@ class FakePlanRepository(PlanRepository): for r in requests ] + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + # The fake store does not track defaults; every property reads as + # first-time (no default yet). + return set() + class _UnsetProductRepo(ProductRepository): """Default for a `FakeUnitOfWork` built without a catalogue — raises if a diff --git a/tests/repositories/plan/test_plan_postgres_repository.py b/tests/repositories/plan/test_plan_postgres_repository.py index 200c38d6b..2d3fb7473 100644 --- a/tests/repositories/plan/test_plan_postgres_repository.py +++ b/tests/repositories/plan/test_plan_postgres_repository.py @@ -206,3 +206,24 @@ def test_rerun_as_non_default_does_not_demote_the_prior_default( assert len(plan_rows) == 2 assert by_id[first_id].is_default is True + + +def test_reports_which_properties_already_have_a_default_plan( + db_engine: Engine, +) -> None: + # Arrange — property 20 has a default Plan, 21 only a non-default one, + # 22 has no Plans at all + with Session(db_engine) as session: + repo = PlanPostgresRepository(session) + repo.save(_plan(), property_id=20, scenario_id=7, portfolio_id=1, is_default=True) + repo.save(_plan(), property_id=21, scenario_id=7, portfolio_id=1, is_default=False) + session.commit() + + # Act + with Session(db_engine) as session: + have_default = PlanPostgresRepository(session).property_ids_with_default_plans( + [20, 21, 22] + ) + + # Assert — only the property with a default Plan is reported + assert have_default == {20} From 371240c8856daf9d31ded655b2f1179839636b34 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:11:32 +0000 Subject: [PATCH 08/13] =?UTF-8?q?The=20plan=20repository=20reports=20which?= =?UTF-8?q?=20properties=20already=20have=20a=20default=20plan=20?= =?UTF-8?q?=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- repositories/plan/plan_postgres_repository.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/repositories/plan/plan_postgres_repository.py b/repositories/plan/plan_postgres_repository.py index c2685ace4..aa1e30157 100644 --- a/repositories/plan/plan_postgres_repository.py +++ b/repositories/plan/plan_postgres_repository.py @@ -3,7 +3,7 @@ from __future__ import annotations from typing import Any from sqlalchemy import insert as _sa_insert -from sqlmodel import Session, col, update +from sqlmodel import Session, col, select, update from domain.modelling.plan import Plan from infrastructure.postgres.modelling import PlanModel, RecommendationModel @@ -46,7 +46,15 @@ class PlanPostgresRepository(PlanRepository): )[0] def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: - raise NotImplementedError + rows = self._session.exec( + select(col(PlanModel.property_id)) + .distinct() + .where( + col(PlanModel.property_id).in_(property_ids), + col(PlanModel.is_default).is_(True), + ) + ).all() + return {int(row) for row in rows} def save_batch(self, requests: list[PlanSaveRequest]) -> list[int]: """Persist all Plans in three statements regardless of batch size. From 101c1f1f2149bb970e0095e007bc45d38b0ebe62 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:12:34 +0000 Subject: [PATCH 09/13] =?UTF-8?q?A=20property's=20first=20plan=20becomes?= =?UTF-8?q?=20its=20default=20whatever=20the=20scenario=20says=20?= =?UTF-8?q?=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- .../modelling_e2e/test_handler.py | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index f2e33187b..32319bcd5 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -511,6 +511,70 @@ def test_attach_mode_partial_failure_persists_successes_then_records_failure() - ] +def test_first_plan_for_a_property_becomes_its_default() -> None: + """A property's first Plan is its default whatever the Scenario says (the + legacy p.is_new rule — readers select one default Plan per property); a + property that already has a default Plan keeps the Scenario's flag.""" + # Arrange — a non-default scenario; 111 has no default plan yet, 222 does + pid_first, pid_has_default = 111, 222 + mock_engine = _engine_mock( + [pid_first, pid_has_default], [1001, 1002], [POSTCODE, POSTCODE] + ) + scenario = MagicMock() + scenario.is_default = False + + with ExitStack() as stack: + stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV)) + stack.enter_context( + patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() + stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository")) + stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient")) + stack.enter_context( + patch("applications.modelling_e2e.handler._spatial_for", return_value=None) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.overlays_from", return_value=[]) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ).return_value.overrides_for_many.return_value = {} + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [scenario] + stack.enter_context(patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides")) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch("applications.modelling_e2e.handler.run_modelling", return_value=_plan_mock()) + ) + MockUoW = stack.enter_context(patch("applications.modelling_e2e.handler.PostgresUnitOfWork")) + mock_uow = MagicMock() + mock_uow.plan.property_ids_with_default_plans.return_value = {pid_has_default} + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + from applications.modelling_e2e.handler import handler + handler.__wrapped__( # type: ignore[attr-defined] + {"property_ids": [pid_first, pid_has_default], + "portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID, + "refetch_solar": False, "dry_run": False}, + None, _mock_orchestrator(), uuid4(), + ) + + # Assert — the first-timer's plan is promoted; the other keeps the flag + plan_requests = mock_uow.plan.save_batch.call_args.args[0] + by_pid = {r.property_id: r for r in plan_requests} + assert by_pid[pid_first].is_default is True + assert by_pid[pid_has_default].is_default is False + + # --------------------------------------------------------------------------- # Lodged EPC path # --------------------------------------------------------------------------- From 0eaf6fa99cb570329dd2b43a3f34d3d276764755 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:13:47 +0000 Subject: [PATCH 10/13] =?UTF-8?q?A=20property's=20first=20plan=20becomes?= =?UTF-8?q?=20its=20default=20whatever=20the=20scenario=20says=20?= =?UTF-8?q?=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- applications/modelling_e2e/handler.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index b580a37d4..55a646676 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -205,13 +205,21 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: uow.epc.save_batch(lodged_requests) if predicted_requests: uow.epc.save_batch(predicted_requests) + # First plan wins: a property with no default Plan yet gets this Plan + # as its default whatever the Scenario's flag says (the legacy + # p.is_new rule — readers select one default Plan per property, so a + # first run under a non-default Scenario must still produce one). + # Also heals properties left default-less by runs before this rule. + have_default: set[int] = uow.plan.property_ids_with_default_plans( + [w.property_id for w in writes] + ) plan_requests = [ PlanSaveRequest( w.plan, property_id=w.property_id, scenario_id=w.scenario_id, portfolio_id=w.portfolio_id, - is_default=w.is_default, + is_default=w.is_default or w.property_id not in have_default, ) for w in writes ] From aa94e76d3d27d4ea62e0ef42af72828b7efe90f0 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:41:48 +0000 Subject: [PATCH 11/13] =?UTF-8?q?A=20new=20default=20plan=20demotes=20the?= =?UTF-8?q?=20prior=20default=20across=20scenarios=20=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- .../plan/test_plan_postgres_repository.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/repositories/plan/test_plan_postgres_repository.py b/tests/repositories/plan/test_plan_postgres_repository.py index 2d3fb7473..7198a6926 100644 --- a/tests/repositories/plan/test_plan_postgres_repository.py +++ b/tests/repositories/plan/test_plan_postgres_repository.py @@ -227,3 +227,35 @@ def test_reports_which_properties_already_have_a_default_plan( # Assert — only the property with a default Plan is reported assert have_default == {20} + + +def test_a_new_default_demotes_the_prior_default_across_scenarios( + db_engine: Engine, +) -> None: + """One default Plan per property: a promoted first Plan (non-default + scenario) must lose the default when the default scenario's Plan lands — + readers filter portfolio + is_default only, never scenario (#1490).""" + # Arrange — the property's first Plan, promoted under non-default scenario 7 + with Session(db_engine) as session: + first_id = PlanPostgresRepository(session).save( + _plan(), property_id=30, scenario_id=7, portfolio_id=1, is_default=True + ) + session.commit() + + # Act — the default scenario (8) models the same property + with Session(db_engine) as session: + second_id = PlanPostgresRepository(session).save( + _plan(), property_id=30, scenario_id=8, portfolio_id=1, is_default=True + ) + session.commit() + + # Assert — exactly one default for the property, the newest plan + with Session(db_engine) as session: + plan_rows = session.exec( + select(PlanModel).where(col(PlanModel.property_id) == 30) + ).all() + by_id = {p.id: p for p in plan_rows} + + assert by_id[first_id].is_default is False # demoted across scenarios + assert by_id[second_id].is_default is True + assert sum(1 for p in plan_rows if p.is_default) == 1 From 5c982beba403fab9ce430c0742a0095b060fbf82 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:42:47 +0000 Subject: [PATCH 12/13] =?UTF-8?q?A=20new=20default=20plan=20demotes=20the?= =?UTF-8?q?=20prior=20default=20across=20scenarios=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- repositories/plan/plan_postgres_repository.py | 19 ++++++++++--------- repositories/plan/plan_repository.py | 3 ++- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/repositories/plan/plan_postgres_repository.py b/repositories/plan/plan_postgres_repository.py index aa1e30157..21540523a 100644 --- a/repositories/plan/plan_postgres_repository.py +++ b/repositories/plan/plan_postgres_repository.py @@ -26,8 +26,9 @@ class PlanPostgresRepository(PlanRepository): A re-run INSERTs a fresh Plan rather than deleting the prior one (the cascade delete was slow); when the new Plan is the default it demotes any prior - default Plan for the same (property_id, scenario_id) to ``is_default=False``, - so readers can select the current Plan via ``is_default=True``.""" + default Plan for the same property — across all Scenarios — to + ``is_default=False``, so readers can select the property's one current Plan + via ``is_default=True``.""" def __init__(self, session: Session) -> None: self._session = session @@ -67,17 +68,17 @@ class PlanPostgresRepository(PlanRepository): return [] # Demote prior default Plans for every property in the batch that is - # receiving a new default Plan — one UPDATE for the whole batch. + # receiving a new default Plan — one UPDATE for the whole batch, + # across ALL scenarios: readers select a property's default by + # portfolio + is_default alone, so the invariant is one default per + # property. Scenario-scoped demotion left a promoted first Plan + # (non-default scenario) defaulted forever alongside the default + # scenario's later Plan (#1490). default_pids = [r.property_id for r in requests if r.is_default] if default_pids: - # scenario_id is uniform per batch (one scenario per SQS message). - scenario_id = requests[0].scenario_id self._session.exec( # type: ignore[call-overload] update(PlanModel) - .where( - col(PlanModel.property_id).in_(default_pids), - col(PlanModel.scenario_id) == scenario_id, - ) + .where(col(PlanModel.property_id).in_(default_pids)) .values(is_default=False) ) diff --git a/repositories/plan/plan_repository.py b/repositories/plan/plan_repository.py index 149a88618..15cafced1 100644 --- a/repositories/plan/plan_repository.py +++ b/repositories/plan/plan_repository.py @@ -43,7 +43,8 @@ class PlanRepository(ABC): ) -> int: """Persist ``plan`` and return its Plan id. Keeps prior Plans for ``(property_id, scenario_id)`` as history; when ``is_default`` is True, - demotes those prior Plans to ``is_default=False``.""" + demotes the property's prior default Plan — whatever its Scenario — to + ``is_default=False`` (one default Plan per property).""" ... @abstractmethod From 8dd125f199e363530c1ac99e3e96ab3a06b78d70 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 7 Jul 2026 16:44:23 +0000 Subject: [PATCH 13/13] =?UTF-8?q?Review=20nitpicks:=20pids=5Fwith=5Fdefaul?= =?UTF-8?q?t=20naming,=20empty-batch=20guard,=20truthful=20fake=20?= =?UTF-8?q?=F0=9F=9F=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- applications/modelling_e2e/handler.py | 4 ++-- repositories/plan/plan_postgres_repository.py | 2 ++ tests/orchestration/fakes.py | 7 ++++--- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 55a646676..51c3e8149 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -210,7 +210,7 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: # p.is_new rule — readers select one default Plan per property, so a # first run under a non-default Scenario must still produce one). # Also heals properties left default-less by runs before this rule. - have_default: set[int] = uow.plan.property_ids_with_default_plans( + pids_with_default: set[int] = uow.plan.property_ids_with_default_plans( [w.property_id for w in writes] ) plan_requests = [ @@ -219,7 +219,7 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: property_id=w.property_id, scenario_id=w.scenario_id, portfolio_id=w.portfolio_id, - is_default=w.is_default or w.property_id not in have_default, + is_default=w.is_default or w.property_id not in pids_with_default, ) for w in writes ] diff --git a/repositories/plan/plan_postgres_repository.py b/repositories/plan/plan_postgres_repository.py index 21540523a..dc3b135a7 100644 --- a/repositories/plan/plan_postgres_repository.py +++ b/repositories/plan/plan_postgres_repository.py @@ -47,6 +47,8 @@ class PlanPostgresRepository(PlanRepository): )[0] def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + if not property_ids: + return set() rows = self._session.exec( select(col(PlanModel.property_id)) .distinct() diff --git a/tests/orchestration/fakes.py b/tests/orchestration/fakes.py index 52f409693..4c6d323b9 100644 --- a/tests/orchestration/fakes.py +++ b/tests/orchestration/fakes.py @@ -212,6 +212,7 @@ class FakePlanRepository(PlanRepository): def __init__(self) -> None: self.saved: dict[tuple[int, int], Plan] = {} + self.default_property_ids: set[int] = set() self._next_id = 1 def save( @@ -224,6 +225,8 @@ class FakePlanRepository(PlanRepository): is_default: bool, ) -> int: self.saved[(property_id, scenario_id)] = plan + if is_default: + self.default_property_ids.add(property_id) plan_id = self._next_id self._next_id += 1 return plan_id @@ -241,9 +244,7 @@ class FakePlanRepository(PlanRepository): ] def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: - # The fake store does not track defaults; every property reads as - # first-time (no default yet). - return set() + return self.default_property_ids.intersection(property_ids) class _UnsetProductRepo(ProductRepository):