Blame for scripts/geocode_holders_google.py

"""DEV-ONLY gap-filling geocoder via the Google Maps Geocoding API.



Fills the holders MISSING from data/iscc/holder_coords.json and merges them in. Resumable (skips

already-geocoded), saves incrementally, biases each lookup to the holder's country for accuracy.

BUILD-TIME ONLY — the app loads the stored JSON at runtime (no live calls), so the map stays

Streamlit-in-Snowflake safe. Google is fast/no daily cap, so one run finishes the backlog.



TERMS NOTE: Google Maps Platform restricts permanently storing geocoded lat/lng (≤30-day caching).

Storing coords here is a TOS gray area accepted for this internal tool — your account, your call.

A ~13k one-time batch typically sits inside Google's $200/month free credit (≈ $0).



Key: env GOOGLE_MAPS_KEY, or scripts/.google_maps_key (both gitignored).



Run:  APP_BACKEND=local GOOGLE_MAPS_KEY=xxxx python scripts/geocode_holders_google.py [scheme] [max]

      scheme = iscc (default) | redcert | ins - max = cap on lookups this run (default 20000)

"""
from __future__ import annotations

import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable

# Windows consoles default to cp1252, which can't encode the · / → in our status lines.
try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
    pass

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import iscc_fetch  # noqa: E402
import rehearsal  # noqa: E402
import schemes as _schemes  # noqa: E402
import state_paths  # noqa: E402
from scripts import geocode_budget  # noqa: E402

_DATA = Path(__file__).parent.parent / "data"
_SUPPORTED_SCHEMES = set(_schemes.CERTIFICATION_SLUGS)
_NEGATIVE_CACHE_MAX_ATTEMPTS = 1


class _BudgetExhausted(Exception):
    pass


def _normalise_scheme(value: str) -> str:
    scheme = str(value or "iscc").strip().lower()
    return scheme if scheme in _SUPPORTED_SCHEMES else "iscc"


def _ins_country_bias(cert: dict, addr: str) -> str:
    import ins  # noqa: E402
    text = " ".join(str(cert.get(k, "")) for k in (
        "country", "codice_fiscale", "partita_iva", "holder_full", "holder_name"
    ))
    return (cert.get("country") or ins.infer_foreign_country(f"{text} {addr}") or "IT").upper()


def _redcert_query(cert: dict) -> str:
    from scripts.geocode_redcert import _normalise_postcode  # noqa: E402
    cc = (cert.get("country") or "").upper()
    pc = _normalise_postcode(cc, cert.get("postcode", ""))
    city = str(cert.get("city") or "").strip()
    return " ".join(p for p in (pc, city) if p).strip()


def _twobsvs_query(cert: dict) -> str:
    """2BSvs carries a structured address: street, postcode, city (country goes in as the region

    bias). Street first so a shared postcode still lands on the right plant."""
    from scripts.geocode_redcert import _normalise_postcode  # noqa: E402
    cc = (cert.get("country") or "").upper()
    pc = _normalise_postcode(cc, cert.get("postcode", ""))
    city = str(cert.get("city") or "").strip()
    street = str(cert.get("street") or cert.get("street_address") or "").strip()
    place = " ".join(p for p in (pc, city) if p).strip()
    return ", ".join(p for p in (street, place) if p)


def _rsb_query(cert: dict) -> str:
    """RSB has no city/postcode fields: the address is whatever follows the holder name in

    holder_full (filled from the certificate PDF by rsb.apply_address_cache). A holder_full that

    is only "Name, Country" holds no address -- geocoding the bare country name would drop a pin

    on the country centroid, so it returns '' and the holder keeps the location fallback."""
    import regions  # noqa: E402
    name = str(cert.get("holder_name") or "").strip()
    full = str(cert.get("holder_full") or cert.get("address") or "").strip()
    address = full[len(name):].strip(", ").strip() if name and full.startswith(name) else full
    if not address:
        return ""
    cc = (cert.get("country") or "").upper()
    countryish = {str(cert.get("country_text") or "").strip().casefold(),
                  regions.country_name(cc).casefold() if cc else ""}
    countryish.discard("")
    if address.casefold() in countryish:
        return ""
    return address


def _holder_items(holders) -> list[tuple[str, str, str]]:
    items = holders.items() if hasattr(holders, "items") else holders
    out = []
    for key, value in items:
        q, cc = value
        hn = key[0] if isinstance(key, tuple) else key
        out.append((hn, q, (cc or "").upper()))
    return out


def _has_coord(coords: dict, scheme: str, holder_norm: str, country: str) -> bool:
    v = coords.get(holder_norm)
    if v is None:
        return False
    cc = (country or "").upper()
    # scheme-registry-ok: deliberate subset - only these two key their coordinates by country.
    if scheme in {"redcert", "ins", "2bsvs"} and isinstance(v, dict):
        return cc in v
    if scheme == "ins" and cc and cc != "IT":
        return False
    return True


def _store_coord(coords: dict, scheme: str, holder_norm: str, country: str, latlng) -> None:
    ll = [latlng[0], latlng[1]]
    cc = (country or "").upper()
    if scheme in {"redcert", "2bsvs"}:  # scheme-registry-ok: the country-keyed coordinate shape
        coords.setdefault(holder_norm, {})[cc] = ll
    elif scheme == "ins" and cc and cc != "IT":
        existing = coords.get(holder_norm)
        if isinstance(existing, dict):
            bucket = existing
        else:
            bucket = {}
            if existing is not None:
                bucket["_default"] = existing
        bucket[cc] = ll
        coords[holder_norm] = bucket
    else:
        coords[holder_norm] = ll


_HONG_KONG = re.compile(r"hong\s*kong", re.IGNORECASE)
_MACAU = re.compile(r"\bmaca[uo]\b", re.IGNORECASE)
# The Dutch Caribbean, filed under NL ("Kralendijk Bonaire, Netherlands Antilles, Netherlands").
_NL_CARIBBEAN = (
    (re.compile(r"\b(bonaire|sint eustatius|statia|saba)\b", re.IGNORECASE), "BQ"),
    (re.compile(r"\bcura[cç]ao\b", re.IGNORECASE), "CW"),
    (re.compile(r"\baruba\b", re.IGNORECASE), "AW"),
    (re.compile(r"\bsint maarten\b", re.IGNORECASE), "SX"),
)


def search_region(query: str, country: str) -> str:
    """Region for Google's `components=country:` filter. The registers file Hong Kong and Macau

    holders under CN ("..., Hong Kong, China") and the Dutch Caribbean under NL, but Google

    treats each as a region of its own, so the country-restricted search returns ZERO_RESULTS

    for every such address. The coordinate is still stored under the register's country; only

    the search filter changes."""
    cc = (country or "").upper()
    text = query or ""
    if cc in ("CN", ""):
        if _HONG_KONG.search(text):
            return "HK"
        if _MACAU.search(text):
            return "MO"
    if cc in ("NL", ""):
        for pattern, region in _NL_CARIBBEAN:
            if pattern.search(text):
                return region
    return cc


def _negative_cache_path(coords_path: Path) -> Path:
    return coords_path.with_name("holder_geocode_failures.json")


def _load_negative_cache(coords_path: Path, failures_path: Path | None = None) -> dict:
    from sinetti_services.geocode_workspace import load_object
    return load_object(failures_path if failures_path is not None else _negative_cache_path(coords_path))


def _save_negative_cache(coords_path: Path, failures: dict, failures_path: Path | None = None) -> None:
    from sinetti_services.geocode_workspace import atomic_write
    atomic_write(failures_path if failures_path is not None else _negative_cache_path(coords_path), failures)


def _negative_cache_key(holder_norm: str, country: str) -> str:
    return f"{holder_norm}|{(country or '').upper()}"


def _negative_cache_hit(failures: dict, holder_norm: str, query: str, country: str) -> bool:
    entry = failures.get(_negative_cache_key(holder_norm, country))
    if not isinstance(entry, dict):
        return False
    if entry.get("query") != query or (entry.get("country") or "").upper() != (country or "").upper():
        return False
    # A failure recorded under another search region (a Hong Kong address searched as CN before
    # search_region existed) says nothing about the current search, so it is retried once.
    if (entry.get("region") or entry.get("country") or "").upper() != search_region(query, country):
        return False
    return int(entry.get("attempts") or 0) >= _NEGATIVE_CACHE_MAX_ATTEMPTS


def _record_negative_cache(failures: dict, holder_norm: str, query: str, country: str) -> None:
    key = _negative_cache_key(holder_norm, country)
    entry = failures.get(key) if isinstance(failures.get(key), dict) else {}
    attempts = int(entry.get("attempts") or 0) + 1
    failures[key] = {
        "holder_norm": holder_norm,
        "country": (country or "").upper(),
        "region": search_region(query, country),
        "query": query,
        "attempts": attempts,
        "last_status": "ZERO_RESULTS",
        "last_attempted_at": datetime.now(timezone.utc).isoformat(),
    }


def _clear_negative_cache(failures: dict, holder_norm: str, country: str) -> None:
    failures.pop(_negative_cache_key(holder_norm, country), None)


def _holders_for(scheme: str):
    """(coords_path, {holder_norm: (address_query, country_bias)}) for the given scheme. ISCC's

    address is the text after the company name (first comma); INS folds the address into holder_full."""
    scheme = _normalise_scheme(scheme)
    if scheme == "ins":
        import ins  # noqa: E402
        certs = ins.load_ins_certs()
        out = {}
        for c in certs:
            hn = c.get("holder_norm")
            if not hn or hn in out:
                continue
            name = c.get("holder_name", "") or ""
            addr = (c.get("holder_full", "") or "")[len(name):].strip(", ").strip()
            out[hn] = (addr, _ins_country_bias(c, addr))
        return _DATA / "ins" / "holder_coords.json", out
    if scheme == "redcert":
        import redcert  # noqa: E402
        out = {}
        for c in redcert.load_redcert_certs():
            hn = c.get("holder_norm")
            cc = (c.get("country") or "").upper()
            if not hn or not cc:
                continue
            q = _redcert_query(c)
            if q:
                out.setdefault((hn, cc), (q, cc))
        return _DATA / "redcert" / "holder_coords.json", out
    if scheme == "2bsvs":
        import twobsvs  # noqa: E402
        out = {}
        for c in twobsvs.load_2bsvs_certs():
            hn = c.get("holder_norm")
            cc = (c.get("country") or "").upper()
            if not hn or not cc:
                continue
            q = _twobsvs_query(c)
            if q:
                out.setdefault((hn, cc), (q, cc))
        return _DATA / "2bsvs" / "holder_coords.json", out
    if scheme == "rsb":
        import rsb  # noqa: E402
        out = {}
        for c in rsb.load_rsb_certs():
            hn = c.get("holder_norm")
            if not hn or hn in out:
                continue
            q = _rsb_query(c)
            if q:
                out[hn] = (q, (c.get("country") or "").upper())
        return _DATA / "rsb" / "holder_coords.json", out
    certs = iscc_fetch.load_bundle(_DATA / "iscc" / "snapshot.csv.gz")
    out = {}
    for c in certs:
        hn = c.get("holder_norm")
        if hn and hn not in out:
            out[hn] = (_addr_query(c.get("holder_name", "")), c.get("country", ""))
    return _DATA / "iscc" / "holder_coords.json", out


def _key() -> str:
    k = os.environ.get("GOOGLE_MAPS_KEY", "").strip()
    if not k:
        f = Path(__file__).with_name(".google_maps_key")
        if f.exists():
            k = f.read_text(encoding="utf-8").strip()
    return k


def _addr_query(holder_name: str) -> str:
    """The postal address (everything after the company name); '' if the holder has no address."""
    parts = [p.strip() for p in (holder_name or "").split(",")]
    return ", ".join(parts[1:]) if len(parts) > 1 else ""


_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
_WHITESPACE_RUN = re.compile(r"\s+")


def _clean_query(q: str) -> str:
    """Control characters (a raw \\x02, newlines or tabs in an INS address) make Google answer

    HTTP 400. Send them as spaces. Callers keep the original string as the negative-cache key."""
    return _WHITESPACE_RUN.sub(" ", _CONTROL_CHARS.sub(" ", q or "")).strip()


def _geocode(

    q: str,

    cc: str,

    key: str,

    *,

    reserve_request: Callable[[], bool] | None = None,

):
    q = _clean_query(q)
    params = {"address": q, "key": key}
    region = search_region(q, cc)
    if region:
        params["components"] = f"country:{region}"
    url = "https://maps.googleapis.com/maps/api/geocode/json?" + urllib.parse.urlencode(params)
    if rehearsal.is_rehearsal():
        # Paid per request. The callers already stop cleanly on an exhausted budget, so reusing
        # that signal keeps the rest of the run honest instead of faking a coordinate.
        raise _BudgetExhausted
    if reserve_request is not None and not reserve_request():
        raise _BudgetExhausted
    try:
        with urllib.request.urlopen(url, timeout=25) as r:
            data = json.loads(r.read())
    except urllib.error.HTTPError as exc:
        # HTTPError and URLError are OSErrors, which run_geocode re-raises on purpose (disk), so
        # one rejected request used to kill the whole nightly data refresh. A 400 is this one
        # query: negative-cache it and go on. Anything else stops geocoding cleanly for tonight.
        if exc.code == 400:
            return None
        raise RuntimeError(f"Google HTTP {exc.code}") from exc
    except (urllib.error.URLError, TimeoutError) as exc:
        raise RuntimeError(f"Google unreachable: {exc}") from exc
    status = data.get("status")
    if status == "OK" and data.get("results"):
        loc = data["results"][0]["geometry"]["location"]
        return round(loc["lat"], 4), round(loc["lng"], 4)
    if status in ("OVER_DAILY_LIMIT", "OVER_QUERY_LIMIT", "REQUEST_DENIED", "INVALID_REQUEST"):
        raise RuntimeError(f"Google status {status}: {data.get('error_message', '')}")
    return None   # ZERO_RESULTS etc.


def run_geocode(

    scheme: str = "iscc",

    max_n: int = 20000,

    *,

    key: str | None = None,

    dry_run: bool = False,

    sleep_s: float = 0.06,

    reserve_request: Callable[[], bool] | None = None,

    coords_path: Path | None = None,

    failures_path: Path | None = None,

    save_coords: Callable[[dict], None] | None = None,

) -> dict:
    """Geocode up to max_n missing holders and return a usage report.



    attempted is the paid Google request count. added is the count of coordinates actually stored.

    """
    scheme = _normalise_scheme(scheme)
    key = _key() if key is None else key
    default_coords, holders = _holders_for(scheme)
    coords_path = Path(coords_path) if coords_path is not None else default_coords
    from sinetti_services.geocode_workspace import load_object, atomic_write
    persist = save_coords if save_coords is not None else lambda value: atomic_write(coords_path, value)
    coords = load_object(coords_path)
    failures = _load_negative_cache(coords_path, failures_path)
    holder_items = _holder_items(holders)
    missing = [(hn, q, cc) for hn, q, cc in holder_items
               if q and not _has_coord(coords, scheme, hn, cc)]
    skipped_negative_cache = sum(
        1 for hn, q, cc in missing if _negative_cache_hit(failures, hn, q, cc)
    )
    todo = [(hn, q, cc) for hn, q, cc in missing
            if not _negative_cache_hit(failures, hn, q, cc)]
    report = {
        "scheme": scheme,
        "status": "dry-run" if dry_run else "ok",
        "coords_path": str(coords_path),
        "negative_cache_path": str(failures_path if failures_path is not None else _negative_cache_path(coords_path)),
        "existing": len(coords),
        "holders": len(holder_items),
        "missing": len(missing),
        "actionable_missing": len(todo),
        "skipped_negative_cache": skipped_negative_cache,
        "limit": max(0, int(max_n)),
        "attempted": 0,
        "added": 0,
        "negative_cached": 0,
        "errors": 0,
        "stop_error": "",
    }
    if dry_run:
        return report
    if not key:
        report["status"] = "skipped-no-google-key"
        return report

    done = 0
    attempted = 0
    for hn, q, cc in todo[:max(0, int(max_n))]:
        try:
            if reserve_request is None:
                attempted += 1
                ll = _geocode(q, cc, key)
            else:
                def reserve_and_count() -> bool:
                    nonlocal attempted
                    if not reserve_request():
                        return False
                    attempted += 1
                    return True

                ll = _geocode(q, cc, key, reserve_request=reserve_and_count)
            if ll:
                _store_coord(coords, scheme, hn, cc, ll)
                _clear_negative_cache(failures, hn, cc)
                persist(coords)
                done += 1
            else:
                _record_negative_cache(failures, hn, q, cc)
                report["negative_cached"] += 1
        except _BudgetExhausted:
            report["status"] = "budget-exhausted"
            break
        except RuntimeError as e:        # auth / quota - stop cleanly, keep what we have
            report["status"] = "stopped"
            report["stop_error"] = str(e)
            break
        except OSError:
            raise
        except Exception:
            report["errors"] += 1
        if sleep_s:
            time.sleep(sleep_s)
        if attempted and attempted % 200 == 0:
            persist(coords)
            _save_negative_cache(coords_path, failures, failures_path)
    persist(coords)
    _save_negative_cache(coords_path, failures, failures_path)
    report["attempted"] = attempted
    report["added"] = done
    report["existing_after"] = len(coords)
    return report


def _env_limit(name: str, default: int) -> int:
    try:
        return max(0, int(os.environ.get(name, default)))
    except (TypeError, ValueError):
        return default


def main(argv: list[str] | None = None) -> None:
    key = _key()
    if not key:
        print("No Google key — set GOOGLE_MAPS_KEY env var or write scripts/.google_maps_key")
        return
    args = list(sys.argv[1:] if argv is None else argv)
    scheme = _normalise_scheme(args[0]) if args else "iscc"
    if args and args[0].lower() in _SUPPORTED_SCHEMES:
        args = args[1:]
    max_n = int(args[0]) if args else 20000
    ledger_path = state_paths.refresh_usage_path()
    daily_limit = _env_limit("GEOCODE_DAILY_LIMIT", geocode_budget.DEFAULT_DAILY_LIMIT)
    monthly_limit = _env_limit("GEOCODE_MONTHLY_LIMIT", geocode_budget.DEFAULT_MONTHLY_LIMIT)
    report = run_geocode(
        scheme,
        max_n,
        key=key,
        sleep_s=0.06,
        reserve_request=lambda: geocode_budget.reserve_call(
            ledger_path,
            daily_limit=daily_limit,
            monthly_limit=monthly_limit,
        ),
    )
    attempted = int(report.get("attempted") or 0)
    added = int(report.get("added") or 0)
    if attempted:
        geocode_budget.append_run_summary(
            ledger_path,
            attempted,
            metadata={
                "scheme": scheme,
                "added": added,
                "status": report.get("status", ""),
            },
        )
    print(
        f"[{scheme}] {report.get('existing', 0)} already geocoded · "
        f"{report.get('missing', 0)} missing · up to {max_n} this run"
    )
    if report.get("stop_error"):
        print("STOP:", report["stop_error"])
    print(
        f"[{scheme}] {report.get('status', 'ok')}: +{added} new · "
        f"{attempted} paid requests · {report.get('existing_after', report.get('existing', 0))} total"
    )


if __name__ == "__main__":
    main()