From 37745939b351f39dd931941f3f699518dee238d5 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 13 Jul 2026 16:22:53 +0000 Subject: [PATCH] Resolve Calico UPRNs from the address-matched S3 CSV lookup Calico Asset References don't match our landlord_property_id, so the runner now resolves UPRNs from the address-matched CSV (UprnLookupS3) uploaded to S3, not the property-table PropertyUprnLookup. run_calico_load accepts any UprnLookup (null_uprn count applies only to the property-table lookup). Adds a dry_run option to reconcile without writing. Co-Authored-By: Claude Opus 4.8 (1M context) --- applications/condition/calico_load.py | 16 +++++-- applications/condition/calico_runner.py | 60 ++++++++++++------------- 2 files changed, 43 insertions(+), 33 deletions(-) diff --git a/applications/condition/calico_load.py b/applications/condition/calico_load.py index 73fbca1df..ea17687ef 100644 --- a/applications/condition/calico_load.py +++ b/applications/condition/calico_load.py @@ -6,6 +6,7 @@ from domain.condition.property_condition_survey import PropertyConditionSurvey from infrastructure.condition.parsing.calico_parser import CalicoParser from repositories.condition.condition_writer import ConditionWriter from repositories.condition.property_uprn_lookup import PropertyUprnLookup +from repositories.condition.uprn_lookup import UprnLookup from utils.logger import setup_logger logger = setup_logger() @@ -13,13 +14,17 @@ logger = setup_logger() def run_calico_load( file_stream: BinaryIO, - uprn_lookup: PropertyUprnLookup, + uprn_lookup: UprnLookup, export_year: int, writer: ConditionWriter, ) -> LoadReport: """Parse a Calico export, map each roof-covering observation to a condition survey keyed by UPRN, persist them, and return the reconciliation report - (ADR-0064). ``export_year`` anchors renewal years so they don't drift.""" + (ADR-0064). ``export_year`` anchors renewal years so they don't drift. + + Works with any ``UprnLookup``. A CSV/S3 lookup carries only resolved UPRNs, + so the ``null_uprn`` count is only meaningful for the property-table lookup + (which surfaces portfolio properties missing a UPRN); it is zero otherwise.""" parser = CalicoParser(uprn_lookup) properties = parser.parse(file_stream) @@ -31,11 +36,16 @@ def run_calico_load( writer.bulk_insert_surveys(surveys) + null_uprn_references = ( + uprn_lookup.null_uprn_references() + if isinstance(uprn_lookup, PropertyUprnLookup) + else [] + ) report = LoadReport.from_calico_load( properties=properties, blank_placeholder_count=parser.blank_placeholder_count, unmatched_reference_count=parser.unmatched_reference_count, - null_uprn_references=uprn_lookup.null_uprn_references(), + null_uprn_references=null_uprn_references, ) logger.info(f"[calico_load] {report}") return report diff --git a/applications/condition/calico_runner.py b/applications/condition/calico_runner.py index 4a5f7bead..26db31af7 100644 --- a/applications/condition/calico_runner.py +++ b/applications/condition/calico_runner.py @@ -2,64 +2,64 @@ Calico is a bulk one-off import, not a recurring feed, so it runs through this runner rather than the SQS-Lambda path Peabody/LBWF use (ADR-0064). The raw -export lives in S3; references resolve to UPRNs via the ``property`` table for -Calico's portfolio. +export and the Asset-Reference->UPRN lookup both live in S3. Not exercised by the unit suite -- it is S3 + Postgres wiring around the unit-tested ``run_calico_load``; run it where a database and S3 are available. -NOTE (2026-07-13): the ``PropertyUprnLookup`` wired below does **not** resolve -Calico yet -- Calico Asset References have zero overlap with portfolio 824's -stored ``landlord_property_id`` (ADR-0064 update). Swap in a CSV ``UprnLookup`` -once Calico supplies an ``Asset Reference -> UPRN`` table, or add address -matching. Running as-is would drop every row as unmatched. +Calico Asset References do not match our stored ``landlord_property_id``, so the +UPRN is resolved from an address-matched CSV lookup (built against portfolio 824 +and uploaded to S3), not the property table. Pass ``dry_run=True`` to reconcile +without writing. """ -import os from io import BytesIO +from typing import List from applications.condition.calico_load import run_calico_load from applications.condition.load_report import LoadReport -from infrastructure.postgres.config import PostgresConfig -from infrastructure.postgres.engine import make_engine, transactional_session +from domain.condition.property_condition_survey import PropertyConditionSurvey +from infrastructure.condition.lookups.uprn_lookup_s3 import UprnLookupS3 from repositories.condition.condition_postgres import ConditionPostgres -from repositories.condition.property_reference_reader_postgres import ( - PropertyReferenceReaderPostgres, -) -from repositories.condition.property_uprn_lookup import PropertyUprnLookup from utils.logger import setup_logger from utils.s3 import read_io_from_s3 logger = setup_logger() -CALICO_PORTFOLIO_ID = 824 # The export date the Calico file was generated (Roof Covering 2.6.26); anchors # renewal years so they don't drift on re-run (ADR-0064). CALICO_EXPORT_YEAR = 2026 -# Confirmed source location. CALICO_BUCKET = "condition-data-dev" CALICO_KEY = "input/calico/Roof Covering 2.6.26.xlsx" +CALICO_UPRN_LOOKUP_KEY = "input/calico/uprn-lookup/calico_reference_to_uprn.csv" + + +class _NoOpWriter: + """Discards surveys -- used for a dry run so the load can be reconciled + without touching the database.""" + + def bulk_insert_surveys( + self, surveys: List[PropertyConditionSurvey] + ) -> None: + logger.info(f"[calico_runner] dry run -- would persist {len(surveys)} surveys") def run_calico_s3_load( bucket: str = CALICO_BUCKET, key: str = CALICO_KEY, - portfolio_id: int = CALICO_PORTFOLIO_ID, + uprn_lookup_key: str = CALICO_UPRN_LOOKUP_KEY, export_year: int = CALICO_EXPORT_YEAR, + dry_run: bool = False, ) -> LoadReport: file_bytes: BytesIO = read_io_from_s3(bucket_name=bucket, file_key=key) + lookup = UprnLookupS3(bucket=bucket, key=uprn_lookup_key) + writer = _NoOpWriter() if dry_run else ConditionPostgres() - engine = make_engine(PostgresConfig.from_env(dict(os.environ))) - with transactional_session(engine) as session: - lookup = PropertyUprnLookup( - PropertyReferenceReaderPostgres(session), portfolio_id - ) - report = run_calico_load( - file_stream=file_bytes, - uprn_lookup=lookup, - export_year=export_year, - writer=ConditionPostgres(), - ) - - logger.info(f"[calico_runner] {report}") + report = run_calico_load( + file_stream=file_bytes, + uprn_lookup=lookup, + export_year=export_year, + writer=writer, + ) + logger.info(f"[calico_runner] {'(dry run) ' if dry_run else ''}{report}") return report