Spaces:
Running
Running
| """Budgeted Google geocoder for physical sites from v_app_sites. | |
| Priority: annex / operating sites → place-of-audit → clean page-1. | |
| One Google call per physical_site_id. Respects geocode_budget ledger. | |
| Run: | |
| APP_BACKEND=local python scripts/geocode_sites.py --dry-run | |
| APP_BACKEND=local python scripts/geocode_sites.py --max-n 300 | |
| APP_BACKEND=local python scripts/geocode_sites.py --backfill-country --dry-run | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| import urllib.error | |
| import urllib.parse | |
| import urllib.request | |
| from contextlib import contextmanager | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Callable, Iterable, Iterator | |
| try: | |
| sys.stdout.reconfigure(encoding="utf-8", errors="replace") | |
| except (AttributeError, ValueError): | |
| pass | |
| ROOT = Path(__file__).resolve().parents[1] | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| import scripts.geocode_budget as geocode_budget # noqa: E402 | |
| import rehearsal # noqa: E402 | |
| import state_paths # noqa: E402 | |
| from sinetti_services import site_geo as sg # noqa: E402 | |
| _NEGATIVE_CACHE_MAX_ATTEMPTS = 1 | |
| # Answers that repeat for the same query: paying for them again buys nothing (SIN-0809). A Google | |
| # HTTP 400 is this one query's failure (c245311c1); network / 5xx answers stay retryable. | |
| _PERMANENT_FAILURE_STATUSES = frozenset({None, "ZERO_RESULTS", "http_error_400", "country_centroid"}) | |
| # A key-wide 400 would otherwise negative-cache a whole night's queue. | |
| _MAX_CONSECUTIVE_HTTP_400 = 3 | |
| # The only v_app_sites columns the queue reads; the wide JSON columns stay on disk. | |
| _SITE_QUEUE_COLUMNS = ( | |
| "scheme", "certificate_id", "site_index", "site_source", "site_name", "address", "postcode", | |
| "city", "site_country", "source_site_id", "holder_norm", "holder_name", "status", | |
| ) | |
| _FATAL_GOOGLE_STATUSES = frozenset( | |
| {"OVER_DAILY_LIMIT", "OVER_QUERY_LIMIT", "REQUEST_DENIED", "INVALID_REQUEST"} | |
| ) | |
| _PRIORITY = { | |
| "iscc_annex_ii": 0, | |
| "ins_rina_annex": 0, | |
| "ins_sede_operativa": 0, | |
| "ins_sgs_storage": 0, | |
| "ins_sgs_production": 0, | |
| "ins_office_branch": 0, | |
| "redcert_operating_sites": 0, | |
| "iscc_place_of_audit": 1, | |
| "page1_header": 2, | |
| "ins_sede_legale": 2, # same band as page1_header (site_geo rank parity) | |
| "epa_facility": 3, | |
| "lcfs_facility": 3, | |
| "cfr_facility": 3, | |
| "renovabio_facility": 3, | |
| "lcfs_facility_coarse": 4, | |
| } | |
| class _RequestMeter: | |
| """Reserve Google request budget before each HTTP attempt.""" | |
| def __init__( | |
| self, | |
| limit: int, | |
| *, | |
| ledger_path: Path | None = None, | |
| daily_limit: int = 0, | |
| monthly_limit: int = 0, | |
| metadata: dict | None = None, | |
| ): | |
| self.limit = max(0, int(limit)) | |
| self.attempted = 0 | |
| self._ledger_path = ledger_path | |
| self._daily_limit = max(0, int(daily_limit)) | |
| self._monthly_limit = max(0, int(monthly_limit)) | |
| self._metadata = metadata | |
| self._finalized = False | |
| def available(self) -> bool: | |
| return self.attempted < self.limit | |
| def reserve(self) -> bool: | |
| if not self.available: | |
| return False | |
| if self._ledger_path is not None and not geocode_budget.reserve_call( | |
| self._ledger_path, | |
| daily_limit=self._daily_limit, | |
| monthly_limit=self._monthly_limit, | |
| ): | |
| return False | |
| self.attempted += 1 | |
| return True | |
| def finalize(self) -> None: | |
| if self._finalized: | |
| return | |
| self._finalized = True | |
| if self._ledger_path is not None: | |
| geocode_budget.append_run_summary( | |
| self._ledger_path, | |
| self.attempted, | |
| metadata=self._metadata, | |
| ) | |
| def _utc_now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _env_limit(name: str, default: int) -> int: | |
| try: | |
| return max(0, int(os.environ.get(name, default))) | |
| except (TypeError, ValueError): | |
| return default | |
| def _key(explicit: str | None = None) -> str: | |
| if explicit is not None and str(explicit).strip(): | |
| return str(explicit).strip() | |
| 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 _normalise_country_code(value: object) -> str | None: | |
| code = str(value or "").strip().upper() | |
| return code if len(code) == 2 and code.isalpha() else None | |
| def _country_code_from_result(result: dict) -> str | None: | |
| """Return a normalized ISO-2 code from a Google geocoder result.""" | |
| for component in result.get("address_components") or []: | |
| if not isinstance(component, dict) or "country" not in (component.get("types") or []): | |
| continue | |
| country_code = _normalise_country_code(component.get("short_name")) | |
| if country_code: | |
| return country_code | |
| return None | |
| def _has_authoritative_site_country(label: dict) -> bool: | |
| """Keep any non-junk physical-country text out of reverse geocoding.""" | |
| _query, country = sg.build_geocode_query(country=str(label.get("site_country") or "")) | |
| return bool(country) | |
| def _google_geocode_request( | |
| params: dict[str, str], key: str, *, request_meter: _RequestMeter | None = None | |
| ) -> tuple[dict | None, str]: | |
| url = "https://maps.googleapis.com/maps/api/geocode/json?" + urllib.parse.urlencode( | |
| {**params, "key": key} | |
| ) | |
| # Google intermittently returns transient 5xx; without a retry one blip | |
| # aborts a whole batch run mid-flight (loses un-checkpointed coords and the | |
| # budget-ledger write). Retry briefly, then record the site as a failure. | |
| if rehearsal.is_rehearsal(): | |
| # Paid per request; "budget-exhausted" is the signal every caller already handles. | |
| return None, "budget-exhausted" | |
| data = None | |
| for attempt in range(3): | |
| if request_meter is not None and not request_meter.reserve(): | |
| return None, "budget-exhausted" | |
| try: | |
| with urllib.request.urlopen(url, timeout=25) as r: | |
| data = json.loads(r.read()) | |
| break | |
| except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as exc: | |
| code = getattr(exc, "code", None) | |
| transient = code is None or int(code) >= 500 | |
| if not transient or attempt == 2: | |
| return None, f"http_error_{code or 'network'}" | |
| time.sleep(2.0 * (attempt + 1)) | |
| if data is None: | |
| return None, "http_error_network" | |
| status = data.get("status") | |
| return data, status or "ZERO_RESULTS" | |
| def _geocode(q: str, cc: str, key: str, *, request_meter: _RequestMeter | None = None): | |
| params = {"address": q} | |
| if cc: | |
| params["components"] = f"country:{cc}" | |
| data, status = _google_geocode_request(params, key, request_meter=request_meter) | |
| if status == "OK" and data and data.get("results"): | |
| result = data["results"][0] | |
| loc = result["geometry"]["location"] | |
| return ( | |
| round(loc["lat"], 4), | |
| round(loc["lng"], 4), | |
| _country_code_from_result(result), | |
| status, | |
| ) | |
| return None, None, None, status | |
| def _reverse_geocode_country( | |
| lat: float, lng: float, key: str, *, request_meter: _RequestMeter | None = None | |
| ) -> tuple[str | None, str]: | |
| """Look up the country component for an existing coordinate pair.""" | |
| data, status = _google_geocode_request( | |
| {"latlng": f"{lat},{lng}"}, key, request_meter=request_meter | |
| ) | |
| if status == "OK" and data and data.get("results"): | |
| for result in data["results"]: | |
| country_code = _country_code_from_result(result) | |
| if country_code: | |
| return country_code, status | |
| return None, status | |
| def _load_failures(path: Path) -> dict: | |
| from sinetti_services.geocode_workspace import load_object | |
| return load_object(path) | |
| def _save_failures(path: Path, failures: dict) -> None: | |
| from sinetti_services.geocode_workspace import atomic_write | |
| atomic_write(path, failures) | |
| def _failure_hit(failures: dict, pid: str, query: str, country: str) -> bool: | |
| entry = failures.get(pid) | |
| if not isinstance(entry, dict): | |
| return False | |
| if entry.get("query") != query or (entry.get("country") or "").upper() != (country or "").upper(): | |
| return False | |
| if entry.get('last_status') not in _PERMANENT_FAILURE_STATUSES: | |
| return False | |
| return int(entry.get("attempts") or 0) >= _NEGATIVE_CACHE_MAX_ATTEMPTS | |
| def _record_failure(failures: dict, pid: str, query: str, country: str, status: str) -> None: | |
| entry = failures.get(pid) if isinstance(failures.get(pid), dict) else {} | |
| attempts = int(entry.get("attempts") or 0) + 1 | |
| failures[pid] = { | |
| "physical_site_id": pid, | |
| "country": (country or "").upper(), | |
| "query": query, | |
| "attempts": attempts, | |
| "last_status": status, | |
| "last_attempted_at": _utc_now(), | |
| } | |
| def _best_source(contributions) -> str: | |
| return min( | |
| (str(c.site_source or "") for c in contributions), | |
| key=lambda s: _PRIORITY.get(s, 99), | |
| default="page1_header", | |
| ) | |
| def _queue_from_rows( | |
| rows: list[dict], | |
| *, | |
| country_bias: str = "", | |
| site_accurate_only: bool = False, | |
| ) -> list[dict]: | |
| index = sg.build_physical_sites(rows) | |
| bias = (country_bias or "").strip().upper() | |
| if bias == "CHINA": | |
| bias = "CN" | |
| queued: list[dict] = [] | |
| for pid, meta in index.items(): | |
| label = meta["label"] | |
| source = _best_source(meta["contributions"]) | |
| accuracy = sg.accuracy_class(source) | |
| if site_accurate_only and accuracy != "site_accurate": | |
| continue | |
| site_cc = str(label.get("site_country") or "").strip() | |
| ok, _reason = sg.address_quality_ok( | |
| label.get("address", ""), | |
| label.get("postcode", ""), | |
| label.get("city", ""), | |
| site_cc or bias, | |
| ) | |
| if not ok: | |
| continue | |
| query, cc = sg.build_geocode_query( | |
| address=label.get("address", ""), | |
| postcode=label.get("postcode", ""), | |
| city=label.get("city", ""), | |
| country=site_cc or bias, | |
| ) | |
| if not query: | |
| continue | |
| if bias and not cc: | |
| cc = bias | |
| queued.append( | |
| { | |
| "physical_site_id": pid, | |
| "query": query, | |
| "country": cc, | |
| "site_source": source, | |
| "priority": _PRIORITY.get(source, 99), | |
| "accuracy_class": accuracy, | |
| } | |
| ) | |
| queued.sort(key=lambda item: (item["priority"], item["physical_site_id"])) | |
| return queued | |
| def filter_sites_by_certificate_country( | |
| rows: list[dict], certificate_country: str | |
| ) -> list[dict]: | |
| """Keep site rows whose certificate_country matches (CN also accepts CHINA).""" | |
| want = (certificate_country or "").strip().upper() | |
| if not want: | |
| return list(rows) | |
| aliases = {want} | |
| if want == "CN": | |
| aliases |= {"CHINA", "P.R. CHINA", "PRC"} | |
| return [ | |
| row | |
| for row in rows | |
| if str(row.get("certificate_country") or "").strip().upper() in aliases | |
| ] | |
| def filter_china_cp_poo_sites(rows: list[dict]) -> list[dict]: | |
| """Keep China Collecting Point / Point of Origin site rows for the footprint pilot. | |
| Uses ``scope_json`` / ``scope`` on each site row (from ``v_app_sites``). Does not | |
| re-check in-force status — pair with ``scan_site_footprints`` for that gate. | |
| """ | |
| from sinetti_services.site_footprint_select import is_pilot_scope, _scopes | |
| china = filter_sites_by_certificate_country(rows, "CN") | |
| return [row for row in china if is_pilot_scope(_scopes(row))] | |
| def run_geocode_sites( | |
| *, | |
| max_n: int = 300, | |
| key: str | None = None, | |
| dry_run: bool = False, | |
| sleep_s: float = 0.06, | |
| sites_rows: Iterable[dict] | None = None, | |
| coords_path: Path | None = None, | |
| failures_path: Path | None = None, | |
| save_coords: Callable[[dict], None] | None = None, | |
| budget_ledger_path: Path | None = None, | |
| daily_limit: int = geocode_budget.DEFAULT_DAILY_LIMIT, | |
| monthly_limit: int = geocode_budget.DEFAULT_MONTHLY_LIMIT, | |
| certificate_country: str | None = None, | |
| site_accurate_only: bool = False, | |
| china_cp_poo_only: bool = False, | |
| ) -> dict: | |
| coords_path = Path(coords_path or sg.default_site_coords_path(ROOT)) | |
| failures_path = Path(failures_path or coords_path.with_name("site_geocode_failures.json")) | |
| budget_ledger_path = Path( | |
| budget_ledger_path if budget_ledger_path is not None else state_paths.refresh_usage_path() | |
| ) | |
| # Iterated once: a streamed artifact read never becomes a list of full rows here. | |
| rows = sites_rows if sites_rows is not None else [] | |
| if china_cp_poo_only: | |
| rows = filter_china_cp_poo_sites(rows) | |
| certificate_country = certificate_country or "CN" | |
| if certificate_country: | |
| rows = filter_sites_by_certificate_country(rows, certificate_country) | |
| queue = _queue_from_rows( | |
| rows, | |
| country_bias=(certificate_country or "").strip().upper(), | |
| site_accurate_only=site_accurate_only, | |
| ) | |
| from sinetti_services.geocode_workspace import load_object | |
| coords = load_object(coords_path) | |
| persist = save_coords if save_coords is not None else lambda value: sg.save_site_coords(coords_path, value) | |
| failures = _load_failures(failures_path) | |
| todo = [] | |
| for item in queue: | |
| pid = item["physical_site_id"] | |
| existing = coords.get(pid) if isinstance(coords.get(pid), dict) else None | |
| if existing is not None and existing.get("lat") is not None and existing.get("lng") is not None: | |
| try: | |
| elat, elng = float(existing["lat"]), float(existing["lng"]) | |
| except (TypeError, ValueError): | |
| elat = elng = None | |
| # Re-queue country-centroid junk pins (placeholder queries). | |
| if elat is not None and not sg.looks_like_country_centroid( | |
| elat, elng, country=item.get("country") or "" | |
| ): | |
| continue | |
| coords.pop(pid, None) | |
| if _failure_hit(failures, pid, item["query"], item["country"]): | |
| continue | |
| todo.append(item) | |
| report = { | |
| "status": "ok", | |
| "queued": len(queue), | |
| "todo": len(todo), | |
| "geocoded": 0, | |
| "failed": 0, | |
| "attempted": 0, | |
| "skipped_existing": len(queue) - len(todo), | |
| } | |
| if dry_run: | |
| report["status"] = "dry-run" | |
| report["would_geocode"] = min(len(todo), max(0, int(max_n))) | |
| return report | |
| api_key = _key(key) | |
| if not api_key: | |
| report["status"] = "skipped-no-google-key" | |
| return report | |
| ledger = geocode_budget.load_usage(budget_ledger_path) | |
| remaining = geocode_budget.remaining_calls( | |
| ledger, daily_limit=int(daily_limit), monthly_limit=int(monthly_limit) | |
| ) | |
| if remaining <= 0: | |
| report["status"] = "budget-exhausted" | |
| report["remaining"] = 0 | |
| return report | |
| limit = min(max(0, int(max_n)), len(todo)) | |
| request_meter = _RequestMeter( | |
| remaining, | |
| ledger_path=budget_ledger_path, | |
| daily_limit=int(daily_limit), | |
| monthly_limit=int(monthly_limit), | |
| metadata={"job": "geocode_sites"}, | |
| ) | |
| consecutive_400 = 0 | |
| try: | |
| for item in todo[:limit]: | |
| if not request_meter.available: | |
| report["status"] = "budget-exhausted" | |
| break | |
| pid = item["physical_site_id"] | |
| attempted_before = request_meter.attempted | |
| try: | |
| lat, lng, country_code, status = _geocode( | |
| item["query"], item["country"], api_key, request_meter=request_meter | |
| ) | |
| country_code = _normalise_country_code(country_code) | |
| except RuntimeError as exc: | |
| if request_meter.attempted == attempted_before: | |
| request_meter.reserve() | |
| report["attempted"] = request_meter.attempted | |
| report["status"] = "google-error" | |
| report["error"] = str(exc) | |
| break | |
| if request_meter.attempted == attempted_before and status != "budget-exhausted": | |
| request_meter.reserve() | |
| report["attempted"] = request_meter.attempted | |
| if status == "budget-exhausted": | |
| report["status"] = status | |
| break | |
| if status in _FATAL_GOOGLE_STATUSES: | |
| report["status"] = "google-error" | |
| report["error"] = f"Google status {status}" | |
| break | |
| if str(status).startswith("http_error_4") and status != "http_error_400": | |
| # A refused key or a rate limit refuses every next request too: stop, keep the cap. | |
| report["status"] = "google-error" | |
| report["error"] = f"Google {status}" | |
| break | |
| if status == "http_error_400": | |
| consecutive_400 += 1 | |
| if consecutive_400 >= _MAX_CONSECUTIVE_HTTP_400: | |
| # Several in a row is the key or request shape, not these sites: stop | |
| # rather than negative-cache the night's whole queue. | |
| report["status"] = "google-error" | |
| report["error"] = f"Google HTTP 400 x{consecutive_400} in a row" | |
| break | |
| else: | |
| consecutive_400 = 0 | |
| if lat is None: | |
| _record_failure(failures, pid, item["query"], item["country"], status) | |
| report["failed"] += 1 | |
| elif sg.looks_like_country_centroid(lat, lng, country=item.get("country") or ""): | |
| _record_failure( | |
| failures, pid, item["query"], item["country"], "country_centroid" | |
| ) | |
| report["failed"] += 1 | |
| else: | |
| entry = { | |
| "lat": lat, | |
| "lng": lng, | |
| "accuracy_class": item["accuracy_class"], | |
| "query": item["query"], | |
| "status": status, | |
| "updated_at": _utc_now(), | |
| } | |
| if country_code: | |
| entry["country_code"] = country_code | |
| coords[pid] = entry | |
| failures.pop(pid, None) | |
| persist(coords) | |
| report["geocoded"] += 1 | |
| if request_meter.attempted % 50 == 0: | |
| persist(coords) | |
| _save_failures(failures_path, failures) | |
| if sleep_s: | |
| time.sleep(float(sleep_s)) | |
| finally: | |
| request_meter.finalize() | |
| persist(coords) | |
| _save_failures(failures_path, failures) | |
| report["remaining"] = geocode_budget.remaining_calls( | |
| geocode_budget.load_usage(budget_ledger_path), | |
| daily_limit=int(daily_limit), | |
| monthly_limit=int(monthly_limit), | |
| ) | |
| return report | |
| def run_country_backfill( | |
| *, | |
| max_n: int = 300, | |
| key: str | None = None, | |
| dry_run: bool = False, | |
| sleep_s: float = 0.06, | |
| sites_rows: list[dict] | None = None, | |
| coords_path: Path | None = None, | |
| failures_path: Path | None = None, | |
| budget_ledger_path: Path | None = None, | |
| daily_limit: int = geocode_budget.DEFAULT_DAILY_LIMIT, | |
| monthly_limit: int = geocode_budget.DEFAULT_MONTHLY_LIMIT, | |
| ) -> dict: | |
| """Backfill blank coordinate country codes with budgeted reverse geocoding.""" | |
| coords_path = Path(coords_path or sg.default_site_coords_path(ROOT)) | |
| failures_path = Path(failures_path or coords_path.with_name("site_geocode_failures.json")) | |
| budget_ledger_path = Path( | |
| budget_ledger_path if budget_ledger_path is not None else state_paths.refresh_usage_path() | |
| ) | |
| coords = sg.load_site_coords(coords_path) | |
| failures = _load_failures(failures_path) | |
| site_index = sg.build_physical_sites(list(sites_rows or [])) | |
| todo: list[tuple[str, dict, float, float]] = [] | |
| for pid, entry in coords.items(): | |
| if not isinstance(entry, dict) or str(entry.get("country_code") or "").strip(): | |
| continue | |
| meta = site_index.get(pid) | |
| label = meta.get("label") if isinstance(meta, dict) else None | |
| if not isinstance(label, dict): | |
| continue | |
| if _has_authoritative_site_country(label): | |
| continue | |
| if sg.resolve_site_flag_country( | |
| site_country=label.get("site_country", ""), | |
| site_name=label.get("site_name", ""), | |
| address=label.get("address", ""), | |
| city=label.get("city", ""), | |
| postcode=label.get("postcode", ""), | |
| ): | |
| continue | |
| try: | |
| lat, lng = float(entry["lat"]), float(entry["lng"]) | |
| except (KeyError, TypeError, ValueError): | |
| continue | |
| todo.append((pid, entry, lat, lng)) | |
| report = { | |
| "status": "ok", | |
| "scanned": len(coords), | |
| "queued": len(todo), | |
| "todo": len(todo), | |
| "backfilled": 0, | |
| "failed": 0, | |
| "attempted": 0, | |
| "skipped_existing": len(coords) - len(todo), | |
| } | |
| if dry_run: | |
| report["status"] = "dry-run" | |
| report["would_backfill"] = min(len(todo), max(0, int(max_n))) | |
| return report | |
| api_key = _key(key) | |
| if not api_key: | |
| report["status"] = "skipped-no-google-key" | |
| return report | |
| ledger = geocode_budget.load_usage(budget_ledger_path) | |
| remaining = geocode_budget.remaining_calls( | |
| ledger, daily_limit=int(daily_limit), monthly_limit=int(monthly_limit) | |
| ) | |
| if remaining <= 0: | |
| report["status"] = "budget-exhausted" | |
| report["remaining"] = 0 | |
| return report | |
| request_meter = _RequestMeter( | |
| remaining, | |
| ledger_path=budget_ledger_path, | |
| daily_limit=int(daily_limit), | |
| monthly_limit=int(monthly_limit), | |
| metadata={"job": "geocode_site_country_backfill"}, | |
| ) | |
| limit = min(max(0, int(max_n)), len(todo)) | |
| try: | |
| for pid, entry, lat, lng in todo[:limit]: | |
| if not request_meter.available: | |
| report["status"] = "budget-exhausted" | |
| break | |
| attempted_before = request_meter.attempted | |
| try: | |
| country_code, status = _reverse_geocode_country( | |
| lat, lng, api_key, request_meter=request_meter | |
| ) | |
| country_code = _normalise_country_code(country_code) | |
| except RuntimeError as exc: | |
| if request_meter.attempted == attempted_before: | |
| request_meter.reserve() | |
| report["attempted"] = request_meter.attempted | |
| report["status"] = "google-error" | |
| report["error"] = str(exc) | |
| break | |
| if request_meter.attempted == attempted_before and status != "budget-exhausted": | |
| request_meter.reserve() | |
| report["attempted"] = request_meter.attempted | |
| if status == "budget-exhausted": | |
| report["status"] = status | |
| break | |
| if status in _FATAL_GOOGLE_STATUSES: | |
| report["status"] = "google-error" | |
| report["error"] = f"Google status {status}" | |
| break | |
| if country_code: | |
| entry["country_code"] = country_code | |
| failures.pop(pid, None) | |
| report["backfilled"] += 1 | |
| else: | |
| _record_failure(failures, pid, f"{lat:.4f},{lng:.4f}", "", status) | |
| report["failed"] += 1 | |
| if request_meter.attempted % 50 == 0: | |
| sg.save_site_coords(coords_path, coords) | |
| _save_failures(failures_path, failures) | |
| if sleep_s: | |
| time.sleep(float(sleep_s)) | |
| finally: | |
| request_meter.finalize() | |
| sg.save_site_coords(coords_path, coords) | |
| _save_failures(failures_path, failures) | |
| report["remaining"] = geocode_budget.remaining_calls( | |
| geocode_budget.load_usage(budget_ledger_path), | |
| daily_limit=int(daily_limit), | |
| monthly_limit=int(monthly_limit), | |
| ) | |
| return report | |
| def stream_site_rows(artifact: str | Path) -> Iterator[Iterator[dict]]: | |
| """Stream the queue columns of ``v_app_sites`` from a serving artifact, one row at a time. | |
| Read through the serving seam (never the bundled fallback). The nightly job is memory-bound, | |
| so the wide JSON columns are never selected and no list of rows is built. A missing artifact | |
| yields nothing. | |
| """ | |
| import sqlite3 | |
| import sinetti_data | |
| artifact = Path(artifact) | |
| if not artifact.exists(): | |
| yield iter(()) | |
| return | |
| with sinetti_data.open_serving_sqlite(artifact) as sqlite_path: | |
| conn = sinetti_data.connect_serving_sqlite(sqlite_path) | |
| conn.row_factory = sqlite3.Row | |
| try: | |
| present = {row[1] for row in conn.execute("PRAGMA table_info(v_app_sites)")} | |
| columns = ", ".join(c for c in _SITE_QUEUE_COLUMNS if c in present) | |
| if not columns: # an artifact built before the sites view | |
| yield iter(()) | |
| return | |
| cursor = conn.execute( | |
| f"SELECT {columns} FROM v_app_sites ORDER BY scheme, certificate_id, site_index" | |
| ) | |
| yield (dict(row) for row in cursor) | |
| finally: | |
| conn.close() | |
| def _load_rows_from_artifact(artifact: Path | None) -> list[dict]: | |
| from sinetti_data import load_app_sites | |
| if artifact is None: | |
| return list(load_app_sites()) | |
| if not Path(artifact).exists(): | |
| return [] | |
| return list(load_app_sites(artifact)) | |
| def main(argv: list[str] | None = None) -> int: | |
| ap = argparse.ArgumentParser(description=__doc__) | |
| ap.add_argument("--max-n", type=int, default=300) | |
| ap.add_argument("--dry-run", action="store_true") | |
| ap.add_argument( | |
| "--backfill-country", | |
| action="store_true", | |
| help="Reverse-geocode stored coordinates missing country_code", | |
| ) | |
| ap.add_argument("--coords", type=Path, default=None) | |
| ap.add_argument("--failures", type=Path, default=None) | |
| ap.add_argument("--ledger", type=Path, default=None) | |
| ap.add_argument("--artifact", type=Path, default=None, help="sinetti_app.sqlite.gz for site rows") | |
| ap.add_argument( | |
| "--daily-limit", | |
| type=int, | |
| default=_env_limit("GEOCODE_DAILY_LIMIT", geocode_budget.DEFAULT_DAILY_LIMIT), | |
| ) | |
| ap.add_argument( | |
| "--monthly-limit", | |
| type=int, | |
| default=_env_limit("GEOCODE_MONTHLY_LIMIT", geocode_budget.DEFAULT_MONTHLY_LIMIT), | |
| ) | |
| ap.add_argument("--sleep", type=float, default=0.06) | |
| ap.add_argument( | |
| "--certificate-country", | |
| default="", | |
| help="Only geocode sites for this certificate country (e.g. CN)", | |
| ) | |
| ap.add_argument( | |
| "--site-accurate-only", | |
| action="store_true", | |
| help="Skip hq_approximate page-1 pins; annex / place-of-audit / operating sites only", | |
| ) | |
| ap.add_argument( | |
| "--china-cp-poo-only", | |
| action="store_true", | |
| help="Only China Collecting Point / Point of Origin sites (footprint pilot queue)", | |
| ) | |
| args = ap.parse_args(argv) | |
| if args.backfill_country: | |
| rows = _load_rows_from_artifact(args.artifact) | |
| report = run_country_backfill( | |
| max_n=args.max_n, | |
| dry_run=args.dry_run, | |
| sleep_s=args.sleep, | |
| sites_rows=rows, | |
| coords_path=args.coords, | |
| failures_path=args.failures, | |
| budget_ledger_path=args.ledger, | |
| daily_limit=args.daily_limit, | |
| monthly_limit=args.monthly_limit, | |
| ) | |
| else: | |
| rows = _load_rows_from_artifact(args.artifact) | |
| report = run_geocode_sites( | |
| max_n=args.max_n, | |
| dry_run=args.dry_run, | |
| sleep_s=args.sleep, | |
| sites_rows=rows, | |
| coords_path=args.coords, | |
| failures_path=args.failures, | |
| budget_ledger_path=args.ledger, | |
| daily_limit=args.daily_limit, | |
| monthly_limit=args.monthly_limit, | |
| certificate_country=args.certificate_country or None, | |
| site_accurate_only=bool(args.site_accurate_only), | |
| china_cp_poo_only=bool(args.china_cp_poo_only), | |
| ) | |
| print(json.dumps(report, indent=2, sort_keys=True), flush=True) | |
| return 0 if report.get("status") in {"ok", "dry-run", "budget-exhausted", "skipped-no-google-key"} else 1 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |