mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
import urllib.parse
|
|
from typing import Any
|
|
from pydantic import ValidationError
|
|
import requests
|
|
import pandas as pd
|
|
from utils.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def os_places_results_to_dataframe(data: dict[str, Any]) -> pd.DataFrame:
|
|
"""
|
|
Flatten the OS Places API response results into a DataFrame.
|
|
Each result contains either a DPA or LPI record.
|
|
"""
|
|
results: list[dict[str, Any]] = data.get("results", [])
|
|
rows: list[dict[str, Any]] = []
|
|
for r in results:
|
|
if "DPA" in r:
|
|
rows.append(r["DPA"])
|
|
elif "LPI" in r:
|
|
rows.append(r["LPI"])
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def lookup_os_places(postcode: str, api_key: str) -> dict[str, Any]:
|
|
"""
|
|
Lookup a postcode using the OS Places API.
|
|
Returns the full API response data or an error dict.
|
|
"""
|
|
if not api_key:
|
|
return {"error": "Ordnance Survey API key not specified", "status": 400}
|
|
|
|
encoded_postcode = urllib.parse.quote(postcode)
|
|
url = (
|
|
f"https://api.os.uk/search/places/v1/postcode?postcode={encoded_postcode}"
|
|
f"&dataset=DPA,LPI&key={api_key}"
|
|
)
|
|
|
|
response = requests.get(url)
|
|
if response.status_code != 200:
|
|
logger.error(
|
|
f"OS Places API error for postcode {postcode}: {response.status_code}"
|
|
)
|
|
return {"error": "Failed to fetch address data", "status": response.status_code}
|
|
|
|
data = response.json()
|
|
return {"data": data, "status": 200}
|