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) <noreply@anthropic.com>

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

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