"""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()