Spaces:

Couscouskabyle
/

iscc-app

Running

iscc-app / scripts /geocode_holders_google.py

Deploy Sinetti V2 (FastAPI + React) [publication:ce9378e5d2984478a403032d3eeafd86]

e459b62 verified 21 days ago

20.7 kB

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

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