mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
37745939b3
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>
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
from typing import BinaryIO, List
|
|
|
|
from applications.condition.load_report import LoadReport
|
|
from domain.condition.mapping.calico.calico_mapper import CalicoMapper
|
|
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()
|
|
|
|
|
|
def run_calico_load(
|
|
file_stream: BinaryIO,
|
|
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.
|
|
|
|
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)
|
|
|
|
mapper = CalicoMapper()
|
|
surveys: List[PropertyConditionSurvey] = [
|
|
mapper.map_asset_conditions_for_property(prop, export_year)
|
|
for prop in properties
|
|
]
|
|
|
|
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=null_uprn_references,
|
|
)
|
|
logger.info(f"[calico_load] {report}")
|
|
return report
|