mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
773869ada4
run_calico_load ties parse -> map -> persist together: it resolves references to UPRNs, drops blank/unmatched rows, persists the roof-covering surveys with export-year-anchored renewal years, and returns the four-count LoadReport. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
41 lines
1.5 KiB
Python
41 lines
1.5 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 utils.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def run_calico_load(
|
|
file_stream: BinaryIO,
|
|
uprn_lookup: PropertyUprnLookup,
|
|
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."""
|
|
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)
|
|
|
|
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(),
|
|
)
|
|
logger.info(f"[calico_load] {report}")
|
|
return report
|