Aissa
5d60d33db1
Volledige Foodlinkk Command Center uitbreiding met social automatisering, reclamefolder filters, Proxmox monitoring en documentatie.
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)}
|