mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy import func
|
|
from backend.app.db.models.addresses import PostcodeSearch
|
|
from utils.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def _get_associated_records(results, uprn, uprn_key="UPRN"):
|
|
matched_record = []
|
|
for x in results:
|
|
if "DPA" in x:
|
|
if x["DPA"].get(uprn_key) == str(uprn):
|
|
matched_record.append(x["DPA"])
|
|
else:
|
|
if x["LPI"].get(uprn_key) == str(uprn):
|
|
matched_record.append(x["LPI"])
|
|
|
|
return matched_record
|
|
|
|
|
|
def get_associated_uprns(postcode_search: PostcodeSearch, uprn: str | int):
|
|
"""
|
|
Given a postcode and UPRN, for a remote assessment, fetch all associated UPRNs, based
|
|
on parent UPRN. This will be properties in the same building
|
|
|
|
Parent UPRN is referenced in the following docs:
|
|
https://static.geoplace.co.uk/downloads/GeoPlace-Data-Entry-Conventions-Best-Practice-for-Addresses.pdf
|
|
|
|
:param PostcodeSearch postcode_search: The postcode search record
|
|
:param uprn: The UPRN string to match
|
|
:return: The matching PostcodeSearch record, or None if not found
|
|
"""
|
|
|
|
if not postcode_search:
|
|
return []
|
|
|
|
if isinstance(uprn, int):
|
|
# For this, coerce to string
|
|
uprn = str(uprn)
|
|
|
|
matched_record = _get_associated_records(results=postcode_search.result_data["results"], uprn=uprn)
|
|
|
|
if len(matched_record) != 1:
|
|
return []
|
|
|
|
if not matched_record[0].get("PARENT_UPRN"):
|
|
logger.info("No parent UPRN found, cannot get associated records")
|
|
return []
|
|
|
|
associated_records = _get_associated_records(
|
|
results=postcode_search.result_data["results"], uprn=matched_record[0]["PARENT_UPRN"], uprn_key="PARENT_UPRN"
|
|
)
|
|
# We now fetch all UPRNS with the same parent UPRN
|
|
associated_uprns = [int(x["UPRN"]) for x in associated_records if x["UPRN"] != str(uprn)]
|
|
|
|
return associated_uprns
|
|
|
|
|
|
def get_by_postcodes(session: Session, postcodes: list[str]) -> dict[str, PostcodeSearch]:
|
|
"""
|
|
Given a list of postcodes, retrieves postcode data from the database form the PostcodeSearch table
|
|
:param session:
|
|
:param postcodes:
|
|
:return:
|
|
"""
|
|
if not postcodes:
|
|
return {}
|
|
|
|
normalised = {p.upper() for p in postcodes if p}
|
|
|
|
records = (
|
|
session.query(PostcodeSearch)
|
|
.filter(func.upper(PostcodeSearch.postcode).in_(normalised))
|
|
.all()
|
|
)
|
|
|
|
return {r.postcode.upper(): r for r in records}
|
|
|
|
|
|
def get_associated_uprns_from_record(record: PostcodeSearch, uprn: str) -> list[int]:
|
|
"""
|
|
Given the postcode sra
|
|
:param record:
|
|
:param uprn:
|
|
:return:
|
|
"""
|
|
if not record:
|
|
return []
|
|
|
|
matched_record = _get_associated_records(
|
|
results=record.result_data["results"],
|
|
uprn=uprn
|
|
)
|
|
|
|
if len(matched_record) != 1:
|
|
return []
|
|
|
|
parent_uprn = matched_record[0].get("PARENT_UPRN")
|
|
if not parent_uprn:
|
|
return []
|
|
|
|
associated_records = _get_associated_records(
|
|
results=record.result_data["results"],
|
|
uprn=parent_uprn,
|
|
uprn_key="PARENT_UPRN"
|
|
)
|
|
|
|
return [
|
|
int(x["UPRN"])
|
|
for x in associated_records
|
|
if x["UPRN"] != str(uprn)
|
|
]
|