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

"""Export Intel sync — OSM Overpass, contact enrichment, tenders, fusion."""
from __future__ import annotations

import json
import re
import time
import xml.etree.ElementTree as ET
from datetime import datetime
from typing import Any, Optional
from urllib.parse import urljoin, urlparse

import httpx

from app.db import execute, execute_returning, fetch_all, fetch_one

OVERPASS_URLS = [
    "https://overpass.kumi.systems/api/interpreter",
    "https://overpass-api.de/api/interpreter",
]

TED_RSS_URLS = [
    "https://ted.europa.eu/en/simap/rss-feed/-/rss/search/15",
    "https://ted.europa.eu/en/simap/rss-feed/-/rss/search/teeq",
]

from app.export_territory_seeds import BBOX_ONLY_ISO2

from app.export_intel_sync_progress import (
    SyncCancelled,
    raise_if_cancelled,
    finish_job,
    get_active_job,
    job_step,
    set_active_job,
    start_job,
    update_job,
)

EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
SKIP_EMAIL_SUFFIX = ("example.com", "sentry.io", "wixpress.com", "png", "jpg", "jpeg")

DISTRIBUTOR_QUERIES = [
    'node["shop"="wholesale"]',
    'way["shop"="wholesale"]',
    'node["wholesale"]',
    'way["wholesale"]',
    'node["shop"="cash_and_carry"]',
    'way["shop"="cash_and_carry"]',
]

CUSTOMER_QUERIES = [
    'node["shop"="butcher"]',
    'way["shop"="butcher"]',
    'node["amenity"="restaurant"]',
    'way["amenity"="restaurant"]',
    'node["amenity"="fast_food"]',
    'way["amenity"="fast_food"]',
]

CATERER_QUERIES = [
    'node["name"~"catering",i]',
    'way["name"~"catering",i]',
    'node["name"~"cateraar",i]',
    'way["name"~"cateraar",i]',
    'node["name"~"food service",i]',
    'way["name"~"food service",i]',
]


def _fetch_overpass(query: str) -> list[dict[str, Any]]:
    last_err: Optional[str] = None
    for url in OVERPASS_URLS:
        for attempt in range(3):
            try:
                with httpx.Client(timeout=180.0) as client:
                    resp = client.post(url, data={"data": query})
                    if resp.status_code == 429:
                        time.sleep(12 * (attempt + 1))
                        continue
                    resp.raise_for_status()
                    return resp.json().get("elements", [])
            except Exception as exc:  # noqa: BLE001
                last_err = str(exc)
                time.sleep(5 * (attempt + 1))
    raise RuntimeError(f"Overpass failed: {last_err}")


def _get_territory(country_iso2: str) -> Optional[dict[str, Any]]:
    return fetch_one(
        """
        SELECT * FROM export_territories
        WHERE country_iso2 = %s AND is_active = TRUE
        ORDER BY sync_priority
        LIMIT 1
        """,
        (country_iso2.upper(),),
    )


def _bbox_str(bbox: Any) -> str:
    if isinstance(bbox, str):
        bbox = json.loads(bbox)
    if not bbox or len(bbox) != 4:
        raise ValueError("Invalid bbox")
    south, west, north, east = bbox
    return f"{south},{west},{north},{east}"


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 _osm_external_id(el: dict[str, Any]) -> str:
    return f"osm:{el.get('type')}:{el.get('id')}"


def _map_entity_type(tags: dict[str, Any], query_kind: str) -> str:
    shop = tags.get("shop") or ""
    amenity = tags.get("amenity") or ""
    name = (tags.get("name") or "").lower()
    if query_kind == "caterer" or "catering" in name or "cateraar" in name:
        return "contract_caterer"
    if shop in ("wholesale", "cash_and_carry") or tags.get("wholesale"):
        return "wholesaler"
    if shop == "butcher":
        return "butcher"
    if amenity == "fast_food":
        return "doner_shoarma" if any(x in name for x in ("doner", "döner", "shoarma", "kebab")) else "restaurant"
    if amenity == "restaurant":
        return "restaurant"
    return "foodservice"


def _upsert_osm_entity(
    el: dict[str, Any],
    country_iso2: str,
    territory_code: Optional[str],
    query_kind: str,
) -> tuple[str, Optional[int]]:
    tags = el.get("tags") or {}
    lat, lon = _coords(el)
    if lat is None or lon is None:
        return "skipped", None
    ext = _osm_external_id(el)
    name = tags.get("name") or tags.get("brand") or tags.get("operator") or "Unknown"
    entity_type = _map_entity_type(tags, query_kind)
    street = tags.get("addr:street") or ""
    hn = tags.get("addr:housenumber") or ""
    address = f"{street} {hn}".strip() or None
    city = tags.get("addr:city") or tags.get("addr:town") or tags.get("addr:village")
    phone = (
        tags.get("phone")
        or tags.get("contact:phone")
        or tags.get("contact:mobile")
        or tags.get("mobile")
        or tags.get("phone:mobile")
        or ""
    )[:64] or None
    email = (tags.get("email") or tags.get("contact:email") or "")[:255] or None
    website = (
        tags.get("website")
        or tags.get("contact:website")
        or tags.get("contact:url")
        or tags.get("url")
        or ""
    )[:512] or None

    existing = fetch_one(
        "SELECT id FROM export_market_entities WHERE metadata->>'osm_external_id' = %s",
        (ext,),
    )
    if existing:
        execute(
            """
            UPDATE export_market_entities SET
                name = %s, entity_type = %s, lat = %s, lon = %s,
                address_line = COALESCE(%s, address_line),
                city = COALESCE(%s, city),
                phone = COALESCE(%s, phone), email = COALESCE(%s, email),
                website = COALESCE(%s, website),
                sources = COALESCE(sources, '[]'::jsonb) || '["osm"]'::jsonb,
                confidence = GREATEST(confidence, 55),
                updated_at = NOW()
            WHERE id = %s
            """,
            (name[:255], entity_type, lat, lon, address, city, phone, email, website, existing["id"]),
        )
        entity_id = existing["id"]
        action = "updated"
    else:
        row = execute_returning(
            """
            INSERT INTO export_market_entities (
                name, entity_type, country_iso2, territory_code, lat, lon,
                address_line, city, phone, email, website, source, sources, confidence,
                metadata
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'osm','["osm"]'::jsonb,55,%s)
            RETURNING id
            """,
            (
                name[:255], entity_type, country_iso2.upper(), territory_code,
                lat, lon, address, city, phone, email, website,
                json.dumps({"osm_external_id": ext}),
            ),
        )
        entity_id = row["id"]
        action = "inserted"

    if email or phone:
        _upsert_contact(entity_id, email, phone, "osm", 60)
    return action, entity_id


def _upsert_contact(
    entity_id: int,
    email: Optional[str],
    phone: Optional[str],
    source: str,
    confidence: int,
    role: str = "general",
    name: Optional[str] = None,
) -> None:
    if email:
        exists = fetch_one(
            "SELECT id FROM export_entity_contacts WHERE entity_id = %s AND email = %s",
            (entity_id, email.lower()),
        )
        if not exists:
            execute(
                """
                INSERT INTO export_entity_contacts (entity_id, role, name, email, phone, source, confidence, is_primary)
                VALUES (%s, %s, %s, %s, %s, %s, %s, FALSE)
                """,
                (entity_id, role, name, email.lower(), phone, source, confidence),
            )
    elif phone:
        exists = fetch_one(
            "SELECT id FROM export_entity_contacts WHERE entity_id = %s AND phone = %s",
            (entity_id, phone),
        )
        if not exists:
            execute(
                """
                INSERT INTO export_entity_contacts (entity_id, role, name, phone, source, confidence)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (entity_id, role, name, phone, source, confidence),
            )


def _build_area_query(parts: list[str], country_iso2: str) -> str:
    body = "".join(f"{p}(area.c);" for p in parts)
    return f'[out:json][timeout:180];area["ISO3166-1"="{country_iso2}"]->.c;({body});out center tags;'


def _build_bbox_query(parts: list[str], bbox: str) -> str:
    body = "".join(f"{p}({bbox});" for p in parts)
    return f"[out:json][timeout:180];({body});out center tags;"


def _countries_for_sync(
    country_iso2: Optional[str] = None,
    region_code: Optional[str] = None,
    max_priority: int = 2,
) -> list[str]:
    if country_iso2:
        return [country_iso2.upper()]
    clauses = ["is_active = TRUE", "sync_priority <= %s"]
    params: list[Any] = [max_priority]
    if region_code:
        clauses.append("region_code = %s")
        params.append(region_code)
    rows = fetch_all(
        f"""
        SELECT country_iso2 FROM export_territories
        WHERE {' AND '.join(clauses)}
        GROUP BY country_iso2
        ORDER BY MIN(sync_priority), country_iso2
        """,
        tuple(params),
    )
    return [r["country_iso2"] for r in rows]


def _build_country_query(parts: list[str], country_iso2: str, bbox: str) -> str:
    if country_iso2.upper() in BBOX_ONLY_ISO2:
        return _build_bbox_query(parts, bbox)
    return _build_area_query(parts, country_iso2.upper())


def sync_osm_country(country_iso2: str, kinds: Optional[list[str]] = None) -> dict[str, Any]:
    territory = _get_territory(country_iso2)
    if not territory or not territory.get("bbox"):
        raise ValueError(f"No territory/bbox for {country_iso2}")
    bbox = _bbox_str(territory["bbox"])
    territory_code = territory["code"]
    kinds = kinds or ["distributors", "customers", "caterers"]
    stats = {"inserted": 0, "updated": 0, "fetched": 0, "kinds": {}}

    mapping = {
        "distributors": DISTRIBUTOR_QUERIES,
        "customers": CUSTOMER_QUERIES,
        "caterers": CATERER_QUERIES,
    }
    for kind in kinds:
        parts = mapping.get(kind, [])
        if not parts:
            continue
        active = get_active_job()
        if active:
            job_step(active, message=f"OSM {kind} · {country_iso2}", country=country_iso2, step=kind)
        query = _build_country_query(parts, country_iso2, bbox)
        elements = _fetch_overpass(query)
        ki = {"fetched": len(elements), "inserted": 0, "updated": 0}
        seen: set[str] = set()
        for el in elements:
            ext = _osm_external_id(el)
            if ext in seen:
                continue
            seen.add(ext)
            action, _ = _upsert_osm_entity(el, country_iso2, territory_code, kind.replace("s", "")[:7])
            if action == "inserted":
                ki["inserted"] += 1
                stats["inserted"] += 1
            elif action == "updated":
                ki["updated"] += 1
                stats["updated"] += 1
        stats["fetched"] += ki["fetched"]
        stats["kinds"][kind] = ki
        time.sleep(2)

    materialize_caterer_presence(country_iso2)
    return stats


def materialize_caterer_presence(country_iso2: str) -> int:
    """Create entity rows from caterer_presence + brands for country."""
    rows = fetch_all(
        """
        SELECT p.*, b.name AS brand_name, b.website
        FROM export_caterer_presence p
        JOIN export_caterer_brands b ON b.code = p.brand_code
        WHERE p.country_iso2 = %s AND p.is_active = TRUE
        """,
        (country_iso2.upper(),),
    )
    territory = _get_territory(country_iso2)
    territory_code = territory["code"] if territory else None
    lat, lon = (territory.get("lat"), territory.get("lon")) if territory else (None, None)
    created = 0
    for p in rows:
        name = p.get("local_legal_name") or p.get("brand_name")
        exists = fetch_one(
            "SELECT id FROM export_market_entities WHERE brand_code = %s AND country_iso2 = %s",
            (p["brand_code"], country_iso2.upper()),
        )
        if exists:
            if p.get("entity_id") is None:
                execute("UPDATE export_caterer_presence SET entity_id = %s WHERE id = %s", (exists["id"], p["id"]))
            continue
        row = execute_returning(
            """
            INSERT INTO export_market_entities (
                name, entity_type, country_iso2, territory_code, lat, lon, website, brand_code,
                source, sources, confidence, product_interest
            ) VALUES (%s,'contract_caterer',%s,%s,%s,%s,%s,%s,'caterer_registry','["caterer_registry"]'::jsonb,75,'["chicken","shoarma"]'::jsonb)
            RETURNING id
            """,
            (
                name, country_iso2.upper(), territory_code, lat, lon,
                (p.get("website") or "")[:512] or None, p["brand_code"],
            ),
        )
        execute("UPDATE export_caterer_presence SET entity_id = %s WHERE id = %s", (row["id"], p["id"]))
        created += 1
    return created



PHONE_RE = re.compile(
    r"(?:tel:|phone|telefoon)[^\d+]{0,12}(\+?[\d\s\(\)\-\.]{8,20})|(\+31[\s\-\.]?(?:\(?\d{2}\)?[\s\-\.]?)?[\d\s\-\.]{6,12})",
    re.I,
)


def _normalize_phone(raw: str) -> Optional[str]:
    s = re.sub(r"[^\d+]", "", (raw or "").strip())
    if len(s) < 8:
        return None
    if s.startswith("00"):
        s = "+" + s[2:]
    if s.startswith("0") and not s.startswith("+"):
        s = "+31" + s[1:]
    return s[:32]


def _scrape_phones(url: str) -> list[str]:
    if not url:
        return []
    if not url.startswith("http"):
        url = "https://" + url
    try:
        with httpx.Client(timeout=15.0, follow_redirects=True) as client:
            resp = client.get(url, headers={"User-Agent": "FoodlinkkExportIntel/1.0"})
            if resp.status_code >= 400:
                return []
            text = resp.text[:500000]
    except Exception:
        return []
    found: set[str] = set()
    for m in PHONE_RE.finditer(text):
        raw = (m.group(1) or m.group(2) or "").strip()
        norm = _normalize_phone(raw)
        if norm:
            found.add(norm)
    return list(found)[:3]


def _scrape_emails(url: str) -> list[str]:
    if not url:
        return []
    if not url.startswith("http"):
        url = "https://" + url
    try:
        with httpx.Client(timeout=15.0, follow_redirects=True) as client:
            resp = client.get(url, headers={"User-Agent": "FoodlinkkExportIntel/1.0"})
            if resp.status_code >= 400:
                return []
            text = resp.text[:500000]
    except Exception:
        return []
    found: set[str] = set()
    for m in EMAIL_RE.findall(text):
        low = m.lower()
        if any(low.endswith(s) or s in low for s in SKIP_EMAIL_SUFFIX):
            continue
        if low.startswith("noreply") or low.startswith("no-reply"):
            continue
        found.add(low)
    return list(found)[:5]


def enrich_contacts_country(country_iso2: str) -> dict[str, Any]:
    entities = fetch_all(
        """
        SELECT id, name, website, email, phone FROM export_market_entities
        WHERE country_iso2 = %s AND website IS NOT NULL AND website <> ''
        """,
        (country_iso2.upper(),),
    )
    added = 0
    scraped = 0
    phones_added = 0
    for ent in entities:
        site = ent["website"]
        emails = _scrape_emails(site)
        phones = _scrape_phones(site) if not ent.get("phone") else []
        # probe common contact pages (gratis, geen API)
        if not emails or not phones:
            base = site if site.startswith("http") else ("https://" + site)
            base = base.rstrip("/")
            for path in ("/contact", "/contact/", "/contact-ons", "/contactus", "/over-ons", "/about"):
                try:
                    emails = list(dict.fromkeys(emails + _scrape_emails(base + path)))
                    if not ent.get("phone"):
                        phones = list(dict.fromkeys(phones + _scrape_phones(base + path)))
                except Exception:
                    pass
                if emails and (ent.get("phone") or phones):
                    break
                time.sleep(0.15)
        scraped += 1
        for em in emails:
            role = "procurement" if any(x in em for x in ("inkoop", "procurement", "sales", "info", "order")) else "general"
            before = fetch_one(
                "SELECT id FROM export_entity_contacts WHERE entity_id = %s AND email = %s",
                (ent["id"], em),
            )
            if not before:
                _upsert_contact(ent["id"], em, None, "website_scrape", 65, role=role, name="Website")
                added += 1
        for ph in phones:
            before_p = fetch_one(
                "SELECT id FROM export_entity_contacts WHERE entity_id = %s AND phone = %s",
                (ent["id"], ph),
            )
            if not before_p:
                _upsert_contact(ent["id"], None, ph, "website_scrape", 60, role="general", name="Website")
                added += 1
                phones_added += 1
        if not ent.get("email") and emails:
            execute(
                "UPDATE export_market_entities SET email = %s, updated_at = NOW() WHERE id = %s",
                (emails[0], ent["id"]),
            )
        if not ent.get("phone") and phones:
            execute(
                "UPDATE export_market_entities SET phone = %s, updated_at = NOW() WHERE id = %s",
                (phones[0], ent["id"]),
            )
        time.sleep(0.3)
    return {"entities_scraped": scraped, "contacts_added": added, "phones_added": phones_added}


def sync_ted_tenders(limit: int = 50) -> dict[str, Any]:
    """TED Search API v3 — food / catering / meat tenders."""
    inserted = updated = 0
    queries = [
        "BT-21-Procedure ~ catering",
        "BT-21-Procedure ~ food",
        "BT-21-Procedure ~ meat",
    ]
    ted_url = "https://api.ted.europa.eu/v3/notices/search"
    fields = [
        "BT-21-Procedure",
        "organisation-name-buyer",
        "publication-number",
        "organisation-country-buyer",
    ]
    seen: set[str] = set()
    try:
        with httpx.Client(timeout=45.0) as client:
            for q in queries:
                resp = client.post(
                    ted_url,
                    json={"query": q, "limit": min(limit, 25), "page": 1, "fields": fields},
                )
                if resp.status_code >= 400:
                    continue
                data = resp.json()
                for notice in data.get("notices", []):
                    pub = notice.get("publication-number") or ""
                    if not pub or pub in seen:
                        continue
                    seen.add(pub)
                    title_raw = notice.get("BT-21-Procedure")
                    if isinstance(title_raw, list):
                        title = title_raw[0] if title_raw else f"TED {pub}"
                    else:
                        title = str(title_raw or f"TED notice {pub}")
                    country_raw = notice.get("organisation-country-buyer")
                    country_iso = None
                    if isinstance(country_raw, list) and country_raw:
                        c = country_raw[0]
                        country_iso = c[:2] if len(c) >= 2 else None
                    link = f"https://ted.europa.eu/en/notice/{pub}"
                    ext = f"ted:{pub}"
                    existing = fetch_one("SELECT id FROM export_tenders WHERE external_id = %s", (ext,))
                    if existing:
                        execute(
                            """
                            UPDATE export_tenders SET title=%s, url=%s, country_iso2=%s, updated_at=NOW()
                            WHERE external_id=%s
                            """,
                            (title[:512], link, country_iso, ext),
                        )
                        updated += 1
                    else:
                        execute(
                            """
                            INSERT INTO export_tenders (external_id, title, url, country_iso2, source, status)
                            VALUES (%s,%s,%s,%s,'ted','open')
                            """,
                            (ext, title[:512], link, country_iso),
                        )
                        inserted += 1
                time.sleep(1)
    except Exception as exc:
        return {"error": str(exc), "inserted": inserted, "updated": updated}
    return {"inserted": inserted, "updated": updated, "source": "ted_api_v3", "seen": len(seen)}


def sync_all_priority(max_priority: int = 2, region_code: Optional[str] = None) -> dict[str, Any]:
    """Full OSM + contacts sync for active territories (gratis)."""
    iso_list = _countries_for_sync(region_code=region_code, max_priority=max_priority)
    label = f"Sync {region_code or 'alle landen'} ({len(iso_list)} landen)"
    job_id = start_job("all", label, total=len(iso_list), meta={"region": region_code, "max_priority": max_priority})
    report: dict[str, Any] = {
        "countries": iso_list,
        "country_count": len(iso_list),
        "region": region_code,
        "max_priority": max_priority,
        "steps": {},
        "job_id": job_id,
    }
    try:
        update_job(job_id, step="ted", message="TED tenders ophalen…")
        report["steps"]["ted"] = sync_ted_tenders(40)

        dist: dict[str, Any] = {}
        cater: dict[str, Any] = {}
        cust: dict[str, Any] = {}
        contacts: dict[str, Any] = {}
        for i, c in enumerate(iso_list, start=1):
            raise_if_cancelled()
            job_step(job_id, message=f"Land {i}/{len(iso_list)}: {c}", country=c, step="osm", current=i, total=len(iso_list))
            try:
                materialize_caterer_presence(c)
                dist[c] = sync_osm_country(c, kinds=["distributors"])
                cater[c] = sync_osm_country(c, kinds=["caterers"])
                cust[c] = sync_osm_country(c, kinds=["customers"])
                contacts[c] = enrich_contacts_country(c)
            except Exception as exc:
                dist[c] = {"error": str(exc)}
            time.sleep(2)
        report["steps"]["distributors"] = dist
        report["steps"]["caterers"] = cater
        report["steps"]["customers"] = cust
        report["steps"]["contacts"] = contacts

        report["steps"]["google_places"] = {"skipped": True, "reason": "gratis-modus: geen Google Places"}
        finish_job(job_id, result={"country_count": len(iso_list), "region": region_code})
    except SyncCancelled as exc:
        finish_job(job_id, error=str(exc))
        report["cancelled"] = True
        return report
    except Exception as exc:
        finish_job(job_id, error=str(exc))
        raise
    finally:
        set_active_job(None)
    return report



def sync_missing_countries(max_priority: int = 3) -> dict[str, Any]:
    """OSM + contacts sync for active territories with zero entities."""
    rows = fetch_all(
        """
        SELECT t.country_iso2
        FROM export_territories t
        WHERE t.is_active
          AND t.sync_priority <= %s
          AND NOT EXISTS (
            SELECT 1 FROM export_market_entities e WHERE e.country_iso2 = t.country_iso2
          )
        GROUP BY t.country_iso2
        ORDER BY MIN(t.sync_priority), t.country_iso2
        """,
        (max_priority,),
    )
    iso_list = [r["country_iso2"] for r in rows]
    job_id = start_job("missing", f"Lege landen ({len(iso_list)})", total=len(iso_list))
    report: dict[str, Any] = {"countries": iso_list, "country_count": len(iso_list), "steps": {}, "job_id": job_id}
    dist: dict[str, Any] = {}
    cater: dict[str, Any] = {}
    cust: dict[str, Any] = {}
    contacts: dict[str, Any] = {}
    try:
        for i, c in enumerate(iso_list, start=1):
            raise_if_cancelled()
            job_step(job_id, message=f"Leeg land {i}/{len(iso_list)}: {c}", country=c, current=i, total=len(iso_list))
            try:
                materialize_caterer_presence(c)
                dist[c] = sync_osm_country(c, kinds=["distributors"])
                cater[c] = sync_osm_country(c, kinds=["caterers"])
                cust[c] = sync_osm_country(c, kinds=["customers"])
                contacts[c] = enrich_contacts_country(c)
            except Exception as exc:
                dist[c] = {"error": str(exc)}
            time.sleep(2)
        report["steps"] = {
            "distributors": dist,
            "caterers": cater,
            "customers": cust,
            "contacts": contacts,
        }
        finish_job(job_id, result={"country_count": len(iso_list)})
    except SyncCancelled as exc:
        finish_job(job_id, error=str(exc))
        report["cancelled"] = True
        return report
    except Exception as exc:
        finish_job(job_id, error=str(exc))
        raise
    finally:
        set_active_job(None)
    return report

def sync_world_regions(max_priority: int = 2) -> dict[str, Any]:
    """Sync all target regions sequentially (Europe, MENA, Africa, Americas)."""
    regions = ["europe", "middle_east", "gcc", "africa", "americas", "caribbean", "asia", "oceania"]
    out: dict[str, Any] = {"regions": {}, "totals": {"countries": 0}}
    for reg in regions:
        countries = _countries_for_sync(region_code=reg, max_priority=max_priority)
        if not countries:
            out["regions"][reg] = {"skipped": True, "reason": "no countries"}
            continue
        out["totals"]["countries"] += len(countries)
        try:
            out["regions"][reg] = sync_all_priority(max_priority=max_priority, region_code=reg)
        except Exception as exc:
            out["regions"][reg] = {"error": str(exc)}
        time.sleep(5)
    return out


def sync_distributors(country_iso2: Optional[str] = None, region_code: Optional[str] = None) -> dict[str, Any]:
    countries = _countries_for_sync(country_iso2, region_code)
    results: dict[str, Any] = {}
    for c in countries:
        try:
            results[c] = sync_osm_country(c, kinds=["distributors"])
        except Exception as exc:
            results[c] = {"error": str(exc)}
        time.sleep(3)
    enrich = {}
    if country_iso2:
        enrich = enrich_contacts_country(country_iso2.upper())
    return {"osm": results, "contact_enrichment": enrich}


def sync_caterers(country_iso2: Optional[str] = None, region_code: Optional[str] = None) -> dict[str, Any]:
    countries = _countries_for_sync(country_iso2, region_code)
    results: dict[str, Any] = {}
    materialized = 0
    for c in countries:
        try:
            materialized += materialize_caterer_presence(c)
            results[c] = sync_osm_country(c, kinds=["caterers"])
        except Exception as exc:
            results[c] = {"error": str(exc)}
        time.sleep(3)
    return {"osm": results, "materialized_presence": materialized}


def sync_contacts(country_iso2: Optional[str] = None, region_code: Optional[str] = None) -> dict[str, Any]:
    countries = _countries_for_sync(country_iso2, region_code)
    out: dict[str, Any] = {}
    for c in countries:
        out[c] = enrich_contacts_country(c)
    ted = sync_ted_tenders()
    return {"contacts": out, "ted_tenders": ted}


def sync_customers(country_iso2: Optional[str] = None, region_code: Optional[str] = None) -> dict[str, Any]:
    if country_iso2:
        job_id = start_job("customers", f"Horeca {country_iso2.upper()}", total=1, meta={"country": country_iso2.upper()})
        try:
            job_step(job_id, message=f"Horeca OSM · {country_iso2.upper()}", country=country_iso2.upper(), current=1, total=1)
            result = sync_osm_country(country_iso2, kinds=["customers"])
            finish_job(job_id, result=result)
            return result
        except Exception as exc:
            finish_job(job_id, error=str(exc))
            raise
        finally:
            set_active_job(None)
    countries = _countries_for_sync(region_code=region_code)
    label = f"Horeca {region_code or 'regio'} ({len(countries)} landen)"
    job_id = start_job("customers", label, total=len(countries), meta={"region": region_code})
    results: dict[str, Any] = {}
    try:
        for i, c in enumerate(countries, start=1):
            raise_if_cancelled()
            job_step(job_id, message=f"Horeca {i}/{len(countries)}: {c}", country=c, current=i, total=len(countries))
            try:
                results[c] = sync_osm_country(c, kinds=["customers"])
            except Exception as exc:
                results[c] = {"error": str(exc)}
            time.sleep(3)
        finish_job(job_id, result={"countries": len(countries), "region": region_code})
    except Exception as exc:
        finish_job(job_id, error=str(exc))
        raise
    finally:
        set_active_job(None)
    return {"osm": results, "job_id": job_id}