254 lines
8.8 KiB
Python
254 lines
8.8 KiB
Python
|
"""Import supermarket locations from OpenStreetMap via Overpass API."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
import time
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from app.db import execute, fetch_one, json_param
|
||
|
|
|
||
|
|
OVERPASS_URLS = [
|
||
|
|
"https://overpass-api.de/api/interpreter",
|
||
|
|
"https://overpass.kumi.systems/api/interpreter",
|
||
|
|
]
|
||
|
|
|
||
|
|
CHAIN_CONFIG: dict[str, dict[str, Any]] = {
|
||
|
|
"ah": {
|
||
|
|
"label": "Albert Heijn",
|
||
|
|
"brands": ["Albert Heijn", "Albert Heijn XL", "AH"],
|
||
|
|
"db_chain": "Albert Heijn",
|
||
|
|
},
|
||
|
|
"jumbo": {
|
||
|
|
"label": "Jumbo",
|
||
|
|
"brands": ["Jumbo"],
|
||
|
|
"db_chain": "Jumbo",
|
||
|
|
},
|
||
|
|
"plus": {
|
||
|
|
"label": "Plus",
|
||
|
|
"brands": ["Plus", "PLUS"],
|
||
|
|
"brand_regex": "Plus",
|
||
|
|
"operators": ["Plus", "Plus Retail", "Plus Supermarkt"],
|
||
|
|
"db_chain": "Plus",
|
||
|
|
},
|
||
|
|
"lidl": {
|
||
|
|
"label": "Lidl",
|
||
|
|
"brands": ["Lidl"],
|
||
|
|
"db_chain": "Lidl",
|
||
|
|
},
|
||
|
|
"aldi": {
|
||
|
|
"label": "ALDI",
|
||
|
|
"brands": ["ALDI", "Aldi"],
|
||
|
|
"db_chain": "ALDI",
|
||
|
|
},
|
||
|
|
"dirk": {
|
||
|
|
"label": "Dirk",
|
||
|
|
"brands": ["Dirk", "Dirk van den Broek"],
|
||
|
|
"db_chain": "Dirk",
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
POSTCODE_RE = re.compile(r"^\d{4}\s?[A-Za-z]{2}$")
|
||
|
|
|
||
|
|
|
||
|
|
def list_chains() -> list[dict[str, str]]:
|
||
|
|
return [{"key": k, "label": v["label"], "db_chain": v["db_chain"]} for k, v in CHAIN_CONFIG.items()]
|
||
|
|
|
||
|
|
|
||
|
|
def _build_overpass_query(cfg: dict[str, Any]) -> str:
|
||
|
|
brands: list[str] = cfg.get("brands", [])
|
||
|
|
brand_regex: Optional[str] = cfg.get("brand_regex")
|
||
|
|
operators: list[str] = cfg.get("operators", [])
|
||
|
|
parts: list[str] = []
|
||
|
|
for b in brands:
|
||
|
|
parts.append(f'node["shop"="supermarket"]["brand"="{b}"](area.nl);')
|
||
|
|
parts.append(f'way["shop"="supermarket"]["brand"="{b}"](area.nl);')
|
||
|
|
if brand_regex:
|
||
|
|
parts.append(f'node["shop"="supermarket"]["brand"~"{brand_regex}",i](area.nl);')
|
||
|
|
parts.append(f'way["shop"="supermarket"]["brand"~"{brand_regex}",i](area.nl);')
|
||
|
|
for op in operators:
|
||
|
|
parts.append(f'node["shop"="supermarket"]["operator"="{op}"](area.nl);')
|
||
|
|
parts.append(f'way["shop"="supermarket"]["operator"="{op}"](area.nl);')
|
||
|
|
return f'[out:json][timeout:180];area["ISO3166-1"="NL"]->.nl;({" ".join(parts)});out center tags;'
|
||
|
|
|
||
|
|
|
||
|
|
def _build_overpass_query_legacy(brands: list[str]) -> str:
|
||
|
|
return _build_overpass_query({"brands": brands})
|
||
|
|
|
||
|
|
|
||
|
|
def _fetch_overpass(query: str) -> list[dict[str, Any]]:
|
||
|
|
last_error: Optional[str] = None
|
||
|
|
for url in OVERPASS_URLS:
|
||
|
|
for attempt in range(3):
|
||
|
|
try:
|
||
|
|
with httpx.Client(timeout=200.0) as client:
|
||
|
|
resp = client.post(url, data={"data": query})
|
||
|
|
if resp.status_code == 429:
|
||
|
|
time.sleep(15 * (attempt + 1))
|
||
|
|
continue
|
||
|
|
resp.raise_for_status()
|
||
|
|
data = resp.json()
|
||
|
|
return data.get("elements", [])
|
||
|
|
except Exception as exc: # noqa: BLE001
|
||
|
|
last_error = str(exc)
|
||
|
|
time.sleep(5 * (attempt + 1))
|
||
|
|
raise RuntimeError(f"Overpass query failed: {last_error}")
|
||
|
|
|
||
|
|
|
||
|
|
def _coords(el: dict[str, Any]) -> tuple[Optional[float], Optional[float]]:
|
||
|
|
if el.get("type") == "node":
|
||
|
|
return el.get("lat"), el.get("lon")
|
||
|
|
center = el.get("center") or {}
|
||
|
|
return center.get("lat"), center.get("lon")
|
||
|
|
|
||
|
|
|
||
|
|
def _normalize_postcode(raw: Optional[str]) -> str:
|
||
|
|
if not raw:
|
||
|
|
return "0000AA"
|
||
|
|
cleaned = raw.strip().upper().replace(" ", "")
|
||
|
|
if len(cleaned) == 6 and cleaned[:4].isdigit() and cleaned[4:].isalpha():
|
||
|
|
return cleaned
|
||
|
|
return "0000AA"
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_store(el: dict[str, Any], db_chain: str) -> Optional[dict[str, Any]]:
|
||
|
|
tags = el.get("tags") or {}
|
||
|
|
lat, lon = _coords(el)
|
||
|
|
if lat is None or lon is None:
|
||
|
|
return None
|
||
|
|
|
||
|
|
street = tags.get("addr:street") or tags.get("addr:place") or ""
|
||
|
|
housenumber = tags.get("addr:housenumber") or ""
|
||
|
|
address = " ".join(p for p in [street, housenumber] if p).strip()
|
||
|
|
if not address:
|
||
|
|
address = tags.get("name") or f"{db_chain} ({lat:.4f}, {lon:.4f})"
|
||
|
|
|
||
|
|
city = tags.get("addr:city") or tags.get("addr:town") or tags.get("addr:village") or "Onbekend"
|
||
|
|
province = tags.get("addr:province") or tags.get("is_in:state")
|
||
|
|
name = tags.get("name") or tags.get("brand") or db_chain
|
||
|
|
|
||
|
|
brand = tags.get("brand") or db_chain
|
||
|
|
store_type = None
|
||
|
|
if "XL" in brand or tags.get("shop") == "supermarket" and "xl" in name.lower():
|
||
|
|
store_type = "XL"
|
||
|
|
elif brand == "AH" or "to go" in name.lower():
|
||
|
|
store_type = "To Go"
|
||
|
|
|
||
|
|
external_id = f"osm:{el.get('type')}:{el.get('id')}"
|
||
|
|
opening_hours = tags.get("opening_hours")
|
||
|
|
|
||
|
|
return {
|
||
|
|
"external_id": external_id,
|
||
|
|
"name": name[:255],
|
||
|
|
"chain": db_chain,
|
||
|
|
"address": address,
|
||
|
|
"postcode": _normalize_postcode(tags.get("addr:postcode")),
|
||
|
|
"city": city[:100],
|
||
|
|
"province": (province or "")[:50] or None,
|
||
|
|
"latitude": lat,
|
||
|
|
"longitude": lon,
|
||
|
|
"store_type": store_type,
|
||
|
|
"phone": (tags.get("phone") or tags.get("contact:phone") or "")[:20] or None,
|
||
|
|
"website": (tags.get("website") or tags.get("contact:website") or "")[:255] or None,
|
||
|
|
"opening_hours": {"raw": opening_hours} if opening_hours else None,
|
||
|
|
"data_source": "openstreetmap",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _ensure_schema() -> None:
|
||
|
|
col = fetch_one(
|
||
|
|
"""
|
||
|
|
SELECT column_name FROM information_schema.columns
|
||
|
|
WHERE table_name = 'supermarkets' AND column_name = 'external_id'
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
if not col:
|
||
|
|
execute("ALTER TABLE supermarkets ADD COLUMN IF NOT EXISTS external_id VARCHAR(64)")
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_supermarkets_external_id
|
||
|
|
ON supermarkets (external_id) WHERE external_id IS NOT NULL
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def import_chain(chain_key: str) -> dict[str, Any]:
|
||
|
|
cfg = CHAIN_CONFIG.get(chain_key)
|
||
|
|
if not cfg:
|
||
|
|
raise ValueError(f"Unknown chain: {chain_key}")
|
||
|
|
|
||
|
|
_ensure_schema()
|
||
|
|
query = _build_overpass_query(cfg)
|
||
|
|
elements = _fetch_overpass(query)
|
||
|
|
|
||
|
|
parsed: list[dict[str, Any]] = []
|
||
|
|
for el in elements:
|
||
|
|
store = _parse_store(el, cfg["db_chain"])
|
||
|
|
if store:
|
||
|
|
parsed.append(store)
|
||
|
|
|
||
|
|
inserted = updated = skipped = 0
|
||
|
|
for store in parsed:
|
||
|
|
existing = fetch_one(
|
||
|
|
"SELECT id FROM supermarkets WHERE external_id = %s",
|
||
|
|
(store["external_id"],),
|
||
|
|
)
|
||
|
|
if existing:
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
UPDATE supermarkets SET
|
||
|
|
name = %s, chain = %s, address = %s, postcode = %s, city = %s,
|
||
|
|
province = %s, latitude = %s, longitude = %s, store_type = %s,
|
||
|
|
phone = %s, website = %s, opening_hours = %s,
|
||
|
|
last_updated = NOW(), data_source = %s
|
||
|
|
WHERE external_id = %s
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
store["name"], store["chain"], store["address"], store["postcode"],
|
||
|
|
store["city"], store["province"], store["latitude"], store["longitude"],
|
||
|
|
store["store_type"], store["phone"], store["website"],
|
||
|
|
json_param(store["opening_hours"]), store["data_source"], store["external_id"],
|
||
|
|
),
|
||
|
|
)
|
||
|
|
updated += 1
|
||
|
|
else:
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO supermarkets (
|
||
|
|
external_id, name, chain, address, postcode, city, province,
|
||
|
|
latitude, longitude, store_type, phone, website, opening_hours,
|
||
|
|
data_source, partnership_status
|
||
|
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'none')
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
store["external_id"], store["name"], store["chain"], store["address"],
|
||
|
|
store["postcode"], store["city"], store["province"], store["latitude"],
|
||
|
|
store["longitude"], store["store_type"], store["phone"], store["website"],
|
||
|
|
json_param(store["opening_hours"]), store["data_source"],
|
||
|
|
),
|
||
|
|
)
|
||
|
|
inserted += 1
|
||
|
|
|
||
|
|
return {
|
||
|
|
"chain": chain_key,
|
||
|
|
"label": cfg["label"],
|
||
|
|
"fetched": len(elements),
|
||
|
|
"parsed": len(parsed),
|
||
|
|
"inserted": inserted,
|
||
|
|
"updated": updated,
|
||
|
|
"skipped": skipped,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def import_all_chains() -> dict[str, Any]:
|
||
|
|
results = []
|
||
|
|
for key in CHAIN_CONFIG:
|
||
|
|
try:
|
||
|
|
results.append(import_chain(key))
|
||
|
|
time.sleep(8)
|
||
|
|
except Exception as exc: # noqa: BLE001
|
||
|
|
results.append({"chain": key, "error": str(exc)})
|
||
|
|
total = fetch_one("SELECT COUNT(*) AS n FROM supermarkets WHERE data_source = 'openstreetmap'")
|
||
|
|
return {"chains": results, "total_osm_stores": int((total or {}).get("n") or 0)}
|