foodlinkk-command-center/tools-api/app/connectors/cbs.py

"""CBS Open Data — gemeente demografie via OData."""
from __future__ import annotations

import json
import re
import urllib.parse
import urllib.request
from typing import Any, Optional

CBS_BASE = "https://opendata.cbs.nl/ODataApi/OData"
_GEMEENTE_CACHE: dict[str, dict[str, Any]] = {}


def _int_val(raw: Any) -> Optional[int]:
    if raw is None:
        return None
    s = str(raw).strip().replace(".", "")
    if not s or s == ".":
        return None
    try:
        return int(s)
    except ValueError:
        return None


def _float_val(raw: Any) -> Optional[float]:
    if raw is None:
        return None
    s = str(raw).strip()
    if not s or s == ".":
        return None
    try:
        return float(s)
    except ValueError:
        return None


def _fetch_untyped(dataset: str, filter_expr: str, top: int = 1) -> list[dict[str, Any]]:
    params = urllib.parse.urlencode(
        {"$filter": filter_expr, "$top": str(top), "$format": "json"},
        quote_via=urllib.parse.quote,
    )
    url = f"{CBS_BASE}/{dataset}/UntypedDataSet?{params}"
    with urllib.request.urlopen(url, timeout=45) as resp:
        data = json.loads(resp.read().decode())
    return data.get("value", [])


def _normalize_gm(code: str) -> str:
    code = (code or "").strip().upper()
    if code.startswith("GM"):
        return code
    digits = re.sub(r"\D", "", code)
    return f"GM{digits}" if digits else code


def fetch_gemeente_stats(gemeente_code: str) -> Optional[dict[str, Any]]:
    gm = _normalize_gm(gemeente_code)
    if not gm:
        return None
    if gm in _GEMEENTE_CACHE:
        return _GEMEENTE_CACHE[gm]

    pop_rows = _fetch_untyped(
        "03759ned",
        f"RegioS eq '{gm}' and Leeftijd eq '10000' and Geslacht eq 'T001038' "
        f"and BurgerlijkeStaat eq 'T001019' and substringof('2024',Perioden)",
    )
    income_rows = _fetch_untyped(
        "86005NED",
        f"RegioS eq '{gm}' and substringof('2023',Perioden) and Geslacht eq 'T001038'",
    )
    area_rows = _fetch_untyped(
        "84583NED",
        f"startswith(WijkenEnBuurten,'{gm}') and SoortRegio_2 eq 'Gemeente  '",
    )

    population = _int_val(pop_rows[0].get("BevolkingOp1Januari_1")) if pop_rows else None
    avg_income = None
    median_income = None
    if income_rows:
        avg_income = _float_val(income_rows[0].get("GemiddeldPersoonlijkInkomen_6"))
        median_income = _float_val(income_rows[0].get("MediaanPersoonlijkInkomen_7"))
        if avg_income:
            avg_income *= 1000
        if median_income:
            median_income *= 1000

    area = area_rows[0] if area_rows else {}
    pop_area = _int_val(area.get("AantalInwoners_5")) or population
    households = _int_val(area.get("HuishoudensTotaal_28"))
    niet_westers = _int_val(area.get("NietWestersTotaal_18"))
    marokko = _int_val(area.get("Marokko_19"))
    turkije = _int_val(area.get("Turkije_22"))
    suriname = _int_val(area.get("Suriname_21"))
    avg_hh_size = _float_val(area.get("GemiddeldeHuishoudensgrootte_32"))
    income_per_inhabitant = _float_val(area.get("GemiddeldInkomenPerInwoner_72"))
    if income_per_inhabitant and not avg_income:
        avg_income = income_per_inhabitant * 1000

    muslim_proxy_pct = None
    niet_westers_pct = None
    if pop_area and pop_area > 0:
        if marokko is not None and turkije is not None:
            muslim_proxy_pct = round((marokko + turkije) / pop_area * 100, 2)
        if niet_westers is not None:
            niet_westers_pct = round(niet_westers / pop_area * 100, 2)

    stats = {
        "gemeente_code": gm,
        "population": pop_area,
        "households": households,
        "avg_household_size": avg_hh_size,
        "avg_income": avg_income,
        "median_income": median_income,
        "unemployment_rate": None,
        "ethnic_composition": {
            "niet_westers_totaal": niet_westers,
            "niet_westers_pct": niet_westers_pct,
            "marokko": marokko,
            "turkije": turkije,
            "suriname": suriname,
        },
        "religious_composition": {
            "muslim_proxy_pct": muslim_proxy_pct,
            "note": "Indicatief: Marokko+Turkije / bevolking (CBS Kerncijfers wijken en buurten)",
        },
        "education_level": {
            "laag": _int_val(area.get("OpleidingsniveauLaag_64")),
            "middelbaar": _int_val(area.get("OpleidingsniveauMiddelbaar_65")),
            "hoog": _int_val(area.get("OpleidingsniveauHoog_66")),
        },
        "housing_type": {
            "koop_pct": _float_val(area.get("Koopwoningen_40")),
            "huur_pct": _float_val(area.get("HuurwoningenTotaal_41")),
        },
        "car_ownership": _float_val(area.get("PersonenautoSPerHuishouden_102")),
        "data_granularity": "gemeente",
        "data_source": "cbs+pdok",
    }
    _GEMEENTE_CACHE[gm] = stats
    return stats