229 lines
8.4 KiB
Python
229 lines
8.4 KiB
Python
"""Eiendom.no enrichment, unit vector, and similar units client."""
|
|
|
|
import base64
|
|
import logging
|
|
from typing import Any
|
|
|
|
import msgpack
|
|
|
|
from .config import (
|
|
EIENDOM_NO_BASE_URL,
|
|
EIENDOM_NO_ENABLED,
|
|
EIENDOM_NO_REQUEST_DELAY_SECONDS,
|
|
EIENDOM_NO_SIMILAR_UNITS_DEFAULT_STATUS,
|
|
)
|
|
from .http import HTTPClient
|
|
from .models import EiendomUnit, SimilarUnit, UnitVector
|
|
from .parser import extract_finnkode_from_url, normalize_finnkode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _extract_coordinates(geometry: dict) -> tuple[float | None, float | None]:
|
|
if not isinstance(geometry, dict):
|
|
return None, None
|
|
coords = geometry.get("coordinates") or []
|
|
if isinstance(coords, (list, tuple)) and len(coords) >= 2:
|
|
return coords[0], coords[1]
|
|
return None, None
|
|
|
|
|
|
def parse_eiendom_unit_json(unit_data: dict) -> EiendomUnit:
|
|
geometry = unit_data.get("geometry", {})
|
|
lon, lat = _extract_coordinates(geometry)
|
|
specification = unit_data.get("specification", {})
|
|
valuation = unit_data.get("valuation", {})
|
|
market = unit_data.get("latestMarketData", {})
|
|
unit_images = market.get("unitImages") or unit_data.get("unitImages") or []
|
|
|
|
return EiendomUnit(
|
|
unit_code=unit_data.get("unitCode", ""),
|
|
address=unit_data.get("address") or unit_data.get("streetAddress"),
|
|
lat=lat or unit_data.get("lat"),
|
|
lng=lon or unit_data.get("lon"),
|
|
property_type=specification.get("propertyType") or unit_data.get("propertyType"),
|
|
floor=specification.get("floor") or unit_data.get("floor"),
|
|
rooms=specification.get("rooms") or unit_data.get("rooms"),
|
|
construction_year=specification.get("constructionYear")
|
|
or unit_data.get("constructionYear"),
|
|
usable_area=specification.get("usableArea") or unit_data.get("usableArea"),
|
|
estimated_selling_price=valuation.get("estimatedSellingPrice")
|
|
or unit_data.get("estimatedSellingPrice"),
|
|
estimated_selling_price_lower=valuation.get("estimatedSellingPriceLower")
|
|
or unit_data.get("estimatedSellingPriceLower"),
|
|
estimated_selling_price_upper=valuation.get("estimatedSellingPriceUpper")
|
|
or unit_data.get("estimatedSellingPriceUpper"),
|
|
listing_price=market.get("listingPrice") or unit_data.get("listingPrice"),
|
|
listing_sqm_price=market.get("squareMeterPrice")
|
|
or unit_data.get("listingSquareMeterPrice"),
|
|
common_costs=market.get("monthlyCosts")
|
|
or market.get("commonCosts")
|
|
or unit_data.get("commonCosts"),
|
|
days_on_market=market.get("daysOnMarket") or unit_data.get("daysOnMarket"),
|
|
sale_status=market.get("saleStatus") or unit_data.get("saleStatus"),
|
|
market_placement_score=market.get("marketPlacementScore")
|
|
or unit_data.get("marketPlacementScore"),
|
|
unit_images=unit_images if unit_images else None,
|
|
)
|
|
|
|
|
|
def parse_similar_units_json(response_data: dict) -> list[SimilarUnit]:
|
|
units: list[SimilarUnit] = []
|
|
for item in response_data.get("units", []):
|
|
geometry = item.get("geometry", {})
|
|
lon, lat = _extract_coordinates(geometry)
|
|
specification = item.get("specification", {})
|
|
market = item.get("marketData", {})
|
|
units.append(
|
|
SimilarUnit(
|
|
unit_code=item.get("unitCode", ""),
|
|
address=item.get("address"),
|
|
lat=lat or item.get("lat"),
|
|
lng=lon or item.get("lon"),
|
|
property_type=specification.get("propertyType") or item.get("propertyType"),
|
|
floor=specification.get("floor") or item.get("floor"),
|
|
rooms=specification.get("rooms") or item.get("rooms"),
|
|
construction_year=specification.get("constructionYear")
|
|
or item.get("constructionYear"),
|
|
usable_area=specification.get("usableArea") or item.get("usableArea"),
|
|
listing_price=market.get("listingPrice") or item.get("listingPrice"),
|
|
selling_price=market.get("sellingPrice") or item.get("sellingPrice"),
|
|
shared_debt=market.get("jointDebt") or item.get("sharedDebt"),
|
|
common_costs=market.get("monthlyCosts") or item.get("commonCosts"),
|
|
sqm_price=market.get("squareMeterPrice") or item.get("squareMeterPrice"),
|
|
days_on_market=market.get("daysOnMarket") or item.get("daysOnMarket"),
|
|
sale_status=market.get("saleStatus") or item.get("saleStatus"),
|
|
finalized_at=item.get("finalizedAt") or market.get("finalizedAt"),
|
|
listing_status=item.get("listingStatus", "RECENTLY_SOLD"),
|
|
)
|
|
)
|
|
return units
|
|
|
|
|
|
def build_unit_vector(unit: EiendomUnit) -> str:
|
|
"""Build a base64url-encoded unit_vector from EiendomUnit data."""
|
|
payload = UnitVector(
|
|
lon=unit.lng or 0.0,
|
|
lat=unit.lat or 0.0,
|
|
ptype=unit.property_type or "APARTMENT",
|
|
floor=unit.floor,
|
|
rooms=unit.rooms,
|
|
built=unit.construction_year,
|
|
area=unit.usable_area,
|
|
price=unit.listing_price or unit.estimated_selling_price,
|
|
)
|
|
packed = msgpack.packb(payload.model_dump(), use_bin_type=True)
|
|
encoded = base64.urlsafe_b64encode(packed).decode("utf-8").rstrip("=")
|
|
return encoded
|
|
|
|
|
|
def decode_unit_vector(vector_str: str) -> dict:
|
|
"""Decode a base64url unit_vector for debugging."""
|
|
padding = 4 - (len(vector_str) % 4)
|
|
if padding != 4:
|
|
vector_str += "=" * padding
|
|
packed = base64.urlsafe_b64decode(vector_str.encode("utf-8"))
|
|
return msgpack.unpackb(packed, raw=False)
|
|
|
|
|
|
async def search_unit_from_finn_url(
|
|
finn_url: str,
|
|
client: HTTPClient | None = None,
|
|
) -> EiendomUnit | None:
|
|
if not EIENDOM_NO_ENABLED or not finn_url:
|
|
logger.info("Eiendom.no unit search is disabled or finn_url is empty")
|
|
return None
|
|
|
|
client = client or HTTPClient(
|
|
base_url=EIENDOM_NO_BASE_URL,
|
|
request_delay_seconds=EIENDOM_NO_REQUEST_DELAY_SECONDS,
|
|
)
|
|
response = await client.get(
|
|
"/geodata/units/search/",
|
|
params={"search": finn_url},
|
|
)
|
|
data = response.json()
|
|
units = data.get("units", [])
|
|
if not units:
|
|
return None
|
|
return parse_eiendom_unit_json(units[0])
|
|
|
|
|
|
async def get_unit(
|
|
unit_code: str,
|
|
client: HTTPClient | None = None,
|
|
) -> EiendomUnit | None:
|
|
if not EIENDOM_NO_ENABLED:
|
|
logger.info("Eiendom.no enrichment is disabled")
|
|
return None
|
|
|
|
client = client or HTTPClient(
|
|
base_url=EIENDOM_NO_BASE_URL,
|
|
request_delay_seconds=EIENDOM_NO_REQUEST_DELAY_SECONDS,
|
|
)
|
|
path = f"/geodata/units/{unit_code}/"
|
|
response = await client.get(path)
|
|
data = response.json()
|
|
units = data.get("units") or []
|
|
if not units and isinstance(data, dict) and data.get("unitCode"):
|
|
return parse_eiendom_unit_json(data)
|
|
if not units:
|
|
return None
|
|
return parse_eiendom_unit_json(units[0])
|
|
|
|
|
|
async def get_eiendom_unit(
|
|
unit_code: str,
|
|
client: HTTPClient | None = None,
|
|
) -> EiendomUnit | None:
|
|
return await get_unit(unit_code, client=client)
|
|
|
|
|
|
async def get_similar_units(
|
|
unit_vector: str,
|
|
listing_status: str = EIENDOM_NO_SIMILAR_UNITS_DEFAULT_STATUS,
|
|
client: HTTPClient | None = None,
|
|
) -> list[SimilarUnit]:
|
|
if not EIENDOM_NO_ENABLED:
|
|
logger.info("Eiendom.no similar-units disabled")
|
|
return []
|
|
|
|
client = client or HTTPClient(
|
|
base_url=EIENDOM_NO_BASE_URL,
|
|
request_delay_seconds=EIENDOM_NO_REQUEST_DELAY_SECONDS,
|
|
)
|
|
response = await client.get(
|
|
"/geodata/units/similar/",
|
|
params={"unit_vector": unit_vector},
|
|
)
|
|
data = response.json()
|
|
units = parse_similar_units_json(data)
|
|
|
|
listing_status = (listing_status or "").upper()
|
|
if listing_status == "RECENTLY_SOLD":
|
|
units = [
|
|
unit
|
|
for unit in units
|
|
if unit.sale_status and unit.sale_status.upper() == "SOLD" and unit.finalized_at
|
|
]
|
|
elif listing_status == "FOR_SALE":
|
|
units = [
|
|
unit for unit in units if unit.sale_status and unit.sale_status.upper() == "FORSALE"
|
|
]
|
|
|
|
return units
|
|
|
|
|
|
async def enrich_ad_with_eiendom_no(
|
|
ad: Any,
|
|
unit_code: str | None = None,
|
|
client: HTTPClient | None = None,
|
|
) -> EiendomUnit | None:
|
|
if not unit_code:
|
|
return None
|
|
unit = await get_eiendom_unit(unit_code, client=client)
|
|
if unit is None:
|
|
return None
|
|
unit.unit_vector = build_unit_vector(unit)
|
|
return unit
|