80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
import logging
|
|
|
|
import requests
|
|
from constance import config
|
|
from django.contrib.gis.geos import Point
|
|
from django.db import models
|
|
|
|
logger = logging.getLogger("django")
|
|
|
|
|
|
class AddressManager(models.Manager):
|
|
def create(self, **kwargs):
|
|
if not kwargs.get("raw_data"):
|
|
raise ValueError("'raw_data' (address string) must be provided.")
|
|
|
|
params: dict[str, str | int] = { # type: ignore [annotation-unchecked]
|
|
"format": "json",
|
|
"addressdetails": 1,
|
|
"q": kwargs.get("raw_data"),
|
|
}
|
|
resp = requests.get(config.NOMINATIM_URL.rstrip("/") + "/search", params=params)
|
|
resp.raise_for_status()
|
|
results = resp.json()
|
|
if not results:
|
|
raise ValueError(f"No geocoding result for address: {kwargs.get('raw_data')}")
|
|
data = results[0]
|
|
|
|
addr = data.get("address", {})
|
|
street = f"{addr.get('road', '') or addr.get('pedestrian', '')}, {addr.get('house_number', '')}"
|
|
district = addr.get("city_district") or addr.get("suburb") or ""
|
|
city = addr.get("city") or addr.get("town") or addr.get("village") or ""
|
|
region = addr.get("state") or addr.get("region") or ""
|
|
postal_code = addr.get("postcode") or ""
|
|
country = addr.get("country") or ""
|
|
|
|
try:
|
|
lat = float(data.get("lat"))
|
|
lon = float(data.get("lon"))
|
|
location = Point(lon, lat, srid=4326)
|
|
except (TypeError, ValueError):
|
|
location = None
|
|
|
|
try:
|
|
address_line_1 = kwargs.pop("address_line_1")
|
|
except KeyError as e:
|
|
raise ValueError("Missing required field 'address_line_1'") from e
|
|
|
|
try:
|
|
address_line_2 = kwargs.pop("address_line_2")
|
|
except KeyError:
|
|
address_line_2 = ""
|
|
|
|
return super().get_or_create(
|
|
raw_data=kwargs.get("raw_data"),
|
|
address_line=f"{address_line_1}, {address_line_2}",
|
|
street=street,
|
|
district=district,
|
|
city=city,
|
|
region=region,
|
|
postal_code=postal_code,
|
|
country=country,
|
|
user=kwargs.pop("user"),
|
|
defaults={"api_response": data, "location": location},
|
|
)[0]
|
|
|
|
|
|
class ProductManager(models.Manager):
|
|
def available(self):
|
|
return self.filter(
|
|
is_active=True, brand__is_active=True, category__is_active=True, stocks__vendor__is_active=True
|
|
)
|
|
|
|
def available_in_stock(self):
|
|
return self.filter(
|
|
is_active=True,
|
|
brand__is_active=True,
|
|
category__is_active=True,
|
|
stocks__vendor__is_active=True,
|
|
stocks__quantity__gt=0,
|
|
)
|