From 01bf0dbd9c7aeb5f0b3ca3c4053d9bde0d9f6e4e Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Tue, 20 Jan 2026 09:58:20 +0000 Subject: [PATCH] =?UTF-8?q?Parse=20Houses=20sheet=20as=20well=20?= =?UTF-8?q?=F0=9F=9F=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/condition/parsing/lbwf_parser.py | 58 ++++++++++++++++++------ 1 file changed, 43 insertions(+), 15 deletions(-) diff --git a/backend/condition/parsing/lbwf_parser.py b/backend/condition/parsing/lbwf_parser.py index 3c2fbf93..44ae05f7 100644 --- a/backend/condition/parsing/lbwf_parser.py +++ b/backend/condition/parsing/lbwf_parser.py @@ -14,36 +14,66 @@ logger = setup_logger class LbwfParser(Parser): def parse(self, file_stream: BinaryIO) -> Any: - wb = load_workbook(file_stream) - address_to_uprn_map: Dict[str, int] = LbwfParser._generate_address_to_uprn_dict(wb) + wb: Workbook = load_workbook(file_stream) + address_to_uprn_map: Dict[str, int] = self._generate_address_to_uprn_dict(wb) + + assets = self._parse_assets(wb) + houses = self._parse_houses(wb, address_to_uprn_map) + + self._merge_assets_into_houses(assets, houses) + + return houses + + @staticmethod + def _parse_assets(wb: Workbook) -> List[LbwfAssetCondition]: + assets_sheet = wb["Houses Asset Data"] + asset_rows = assets_sheet.iter_rows(values_only=True) - # Parse assets - assets_sheet: Workbook = wb["Houses Asset Data"] - asset_rows: Iterator[Tuple[object | None, ...]] = assets_sheet.iter_rows(values_only=True) asset_headers = next(asset_rows) - asset_header_indexes: Dict[str, int] = LbwfParser._get_column_indexes_by_name(asset_headers) + asset_header_indexes = LbwfParser._get_column_indexes_by_name(asset_headers) assets: List[LbwfAssetCondition] = [] for row in asset_rows: try: - assets.append(LbwfParser._map_row_to_asset_record(row, asset_header_indexes)) + assets.append( + LbwfParser._map_row_to_asset_record(row, asset_header_indexes) + ) except Exception as e: logger.error(f"Error mapping LBWF row to asset record: {e}") - # Parse houses - houses_sheet: Workbook = wb["Houses"] - house_rows: Iterator[Tuple[object | None, ...]] = houses_sheet.iter_rows(values_only=True) + return assets + + @staticmethod + def _parse_houses( + wb: Workbook, + address_to_uprn_map: Dict[str, int], + ) -> List[LbwfHouse]: + houses_sheet = wb["Houses"] + house_rows = houses_sheet.iter_rows(values_only=True) + house_headers = next(house_rows) - house_header_indexes: Dict[str, int] = LbwfParser._get_column_indexes_by_name(house_headers) + house_header_indexes = LbwfParser._get_column_indexes_by_name(house_headers) houses: List[LbwfHouse] = [] for row in house_rows: try: - houses.append(LbwfParser._map_row_to_house_record(row, house_header_indexes, address_to_uprn_map)) + houses.append( + LbwfParser._map_row_to_house_record( + row, + house_header_indexes, + address_to_uprn_map, + ) + ) except Exception as e: logger.error(f"Error mapping LBWF row to house record: {e}") - # Merge assets and houses by Reference + return houses + + @staticmethod + def _merge_assets_into_houses( + assets: List[LbwfAssetCondition], + houses: List[LbwfHouse], + ) -> None: assets_by_ref: Dict[int, List[LbwfAssetCondition]] = defaultdict(list) for asset in assets: assets_by_ref[asset.prop_ref].append(asset) @@ -51,8 +81,6 @@ class LbwfParser(Parser): for house in houses: house.assets = assets_by_ref.get(house.reference, []) - return houses - @staticmethod def _map_row_to_house_record(