foodlinkk-command-center/tools-api/app/retail_enrichment.py

"""Enrich supermarket postcodes with PDOK geocoding + CBS demografie."""
from __future__ import annotations

import time
from typing import Any, Optional

from app.connectors import cbs, pdok
from app.db import execute, execute_returning, fetch_all, fetch_one, json_param


def _postcodes_to_enrich(limit: int = 100, offset: int = 0) -> list[str]:
    rows = fetch_all(
        """
        SELECT DISTINCT s.postcode
        FROM supermarkets s
        LEFT JOIN area_analysis a ON a.postcode = s.postcode
        WHERE s.postcode IS NOT NULL
          AND s.postcode <> '0000AA'
          AND a.id IS NULL
        ORDER BY s.postcode
        LIMIT %s OFFSET %s
        """,
        (limit, offset),
    )
    return [r["postcode"] for r in rows]


def enrich_postcode(postcode: str) -> dict[str, Any]:
    pc = pdok.normalize_postcode(postcode)
    pdok_data = pdok.lookup_postcode(pc)
    if not pdok_data:
        return {"postcode": pc, "status": "pdok_not_found"}

    gm_code = pdok_data.get("municipality_code")
    cbs_data = cbs.fetch_gemeente_stats(gm_code) if gm_code else None
    if not cbs_data:
        return {"postcode": pc, "status": "cbs_not_found", "pdok": pdok_data}

    religious = cbs_data.get("religious_composition") or {}
    row = execute_returning(
        """
        INSERT INTO area_analysis (
            postcode, city, population, households, avg_household_size,
            avg_income, median_income, education_level, ethnic_composition,
            religious_composition, unemployment_rate, housing_type, car_ownership,
            data_source, last_updated
        ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW())
        ON CONFLICT (postcode) DO UPDATE SET
            city = EXCLUDED.city,
            population = EXCLUDED.population,
            households = EXCLUDED.households,
            avg_household_size = EXCLUDED.avg_household_size,
            avg_income = EXCLUDED.avg_income,
            median_income = EXCLUDED.median_income,
            education_level = EXCLUDED.education_level,
            ethnic_composition = EXCLUDED.ethnic_composition,
            religious_composition = EXCLUDED.religious_composition,
            housing_type = EXCLUDED.housing_type,
            car_ownership = EXCLUDED.car_ownership,
            data_source = EXCLUDED.data_source,
            last_updated = NOW()
        RETURNING id, postcode
        """,
        (
            pc,
            pdok_data.get("city") or cbs_data.get("city"),
            cbs_data.get("population"),
            cbs_data.get("households"),
            cbs_data.get("avg_household_size"),
            cbs_data.get("avg_income"),
            cbs_data.get("median_income"),
            json_param(cbs_data.get("education_level")),
            json_param(cbs_data.get("ethnic_composition")),
            json_param(religious),
            cbs_data.get("unemployment_rate"),
            json_param(cbs_data.get("housing_type")),
            cbs_data.get("car_ownership"),
            f"cbs+pdok ({cbs_data.get('data_granularity', 'gemeente')})",
        ),
    )

    execute(
        """
        UPDATE supermarkets SET
            city = COALESCE(NULLIF(city, 'Onbekend'), %s),
            province = COALESCE(province, %s),
            latitude = COALESCE(latitude, %s),
            longitude = COALESCE(longitude, %s),
            last_updated = NOW()
        WHERE postcode = %s AND (
            city = 'Onbekend' OR province IS NULL OR latitude IS NULL OR longitude IS NULL
        )
        """,
        (
            pdok_data.get("city"),
            pdok_data.get("province"),
            pdok_data.get("latitude"),
            pdok_data.get("longitude"),
            pc,
        ),
    )
    return {
        "postcode": pc,
        "status": "ok",
        "area_id": row["id"] if row else None,
        "municipality": pdok_data.get("municipality"),
        "population": cbs_data.get("population"),
        "muslim_proxy_pct": religious.get("muslim_proxy_pct"),
    }


def enrich_batch(limit: int = 50, offset: int = 0, delay_sec: float = 0.15) -> dict[str, Any]:
    postcodes = _postcodes_to_enrich(limit, offset)
    results: list[dict[str, Any]] = []
    ok = failed = 0
    for pc in postcodes:
        try:
            result = enrich_postcode(pc)
            results.append(result)
            if result.get("status") == "ok":
                ok += 1
            else:
                failed += 1
        except Exception as exc:  # noqa: BLE001
            results.append({"postcode": pc, "status": "error", "error": str(exc)})
            failed += 1
        time.sleep(delay_sec)

    remaining = fetch_one(
        """
        SELECT COUNT(DISTINCT s.postcode) AS n
        FROM supermarkets s
        LEFT JOIN area_analysis a ON a.postcode = s.postcode
        WHERE s.postcode <> '0000AA' AND a.id IS NULL
        """
    )
    return {
        "processed": len(postcodes),
        "ok": ok,
        "failed": failed,
        "remaining": int((remaining or {}).get("n") or 0),
        "results": results[:20],
    }


def enrichment_status() -> dict[str, Any]:
    total_pc = fetch_one(
        "SELECT COUNT(DISTINCT postcode) AS n FROM supermarkets WHERE postcode <> '0000AA'"
    )
    enriched = fetch_one("SELECT COUNT(*) AS n FROM area_analysis")
    providers = fetch_all(
        "SELECT name, provider_type, last_fetch_at, last_status, is_active FROM data_providers ORDER BY name"
    )
    return {
        "unique_postcodes": int((total_pc or {}).get("n") or 0),
        "enriched_postcodes": int((enriched or {}).get("n") or 0),
        "data_providers": [dict(p) for p in providers],
    }