diff --git a/backend/app/modelling/property_filters.py b/backend/app/modelling/property_filters.py index 4f45c6325..414f16889 100644 --- a/backend/app/modelling/property_filters.py +++ b/backend/app/modelling/property_filters.py @@ -12,6 +12,8 @@ from typing import Optional from pydantic import BaseModel, ConfigDict from sqlmodel import Session, col, select +from infrastructure.postgres.epc_property_table import EpcPropertyModel +from infrastructure.postgres.property_override_table import PropertyOverrideRow from infrastructure.postgres.property_table import PropertyRow @@ -54,8 +56,53 @@ def resolve_filtered_property_ids( if filters.postcodes is not None: statement = statement.where(col(PropertyRow.postcode).in_(filters.postcodes)) rows = session.exec(statement).all() + candidates = [row for row in rows if row.id is not None] + + if filters.property_types is not None: + wanted = set(filters.property_types) + types = _resolved_property_types( + session, [row.id for row in candidates if row.id is not None] + ) + candidates = [ + row + for row in candidates + if row.id is not None and types.get(row.id) in wanted + ] + return [ FilteredProperty(property_id=row.id, postcode=row.postcode) - for row in rows + for row in candidates if row.id is not None ] + + +def _resolved_property_types( + session: Session, property_ids: list[int] +) -> dict[int, str]: + """Each property's effective property_type per the ADR-0056 precedence: + override (building_part 0) → lodged EPC.""" + overrides = session.exec( + select(PropertyOverrideRow) + .where(col(PropertyOverrideRow.property_id).in_(property_ids)) + .where(PropertyOverrideRow.building_part == 0) + .where(PropertyOverrideRow.override_component == "property_type") + ).all() + by_override = {o.property_id: o.override_value for o in overrides} + + epcs = session.exec( + select(EpcPropertyModel) + .where(col(EpcPropertyModel.property_id).in_(property_ids)) + .where(EpcPropertyModel.source == "lodged") + ).all() + by_lodged_epc = { + epc.property_id: epc.property_type + for epc in epcs + if epc.property_id is not None and epc.property_type is not None + } + + resolved: dict[int, str] = {} + for pid in property_ids: + value = by_override.get(pid) or by_lodged_epc.get(pid) + if value is not None: + resolved[pid] = value + return resolved