Spaces:
Running
Running
| #!/usr/bin/env python | |
| """Weekly LOCAL US registers refresh: LCFS (CARB) + EPA (Part 80 Renewable Fuel Producers). | |
| Cloud egress unverified for both endpoints (INS precedent: GSE blocked HF Jobs + GitHub Actions | |
| datacenter IPs); the files are tiny (1.7MB + 5.8MB) and weekly cadence suffices, so this refreshes | |
| HERE, local, and publishes through the same private-dataset candidate/release flow the daily and | |
| INS refreshes use -- never a direct serving-data write, never auto-activation. | |
| fetch CARB xlsx + EPA xls (browser UA) -> schema assert (refuse loudly on drift) | |
| -> normalize (lcfs.py / epa.py) -> corpus_watch row-count bands | |
| (data/us_registers_counts.json -- a DEDICATED manifest, see tests/test_us_registers_counts.py | |
| for why this never touches the shared data/corpus_counts.json) | |
| -> snapshot-diff for change events -> data_plane.publish | |
| Any schema/band/rebuild failure leaves current serving data untouched (no publish call at all) | |
| and is recorded loudly in the report. A DOWNLOAD failure is treated differently, because an | |
| unreachable source says nothing about the data: that register keeps its previous snapshot and | |
| the other register still publishes, provided a usable previous snapshot exists and at least one | |
| register actually refreshed. Registered as a weekly Windows Scheduled Task | |
| ("Sinetti PC weekly US refresh"), weekend slot (see docs/ins-local-refresh-runbook.md for the sibling | |
| INS task this mirrors). | |
| Usage: | |
| python scripts/us_local_refresh.py # full run: fetch + publish | |
| python scripts/us_local_refresh.py --dry-run # config check only, no network | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import shutil | |
| import sys | |
| import tempfile | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| ROOT = Path(__file__).parent.parent | |
| sys.path.insert(0, str(ROOT)) | |
| os.environ.setdefault("APP_BACKEND", "local") | |
| def _devnull_headless_streams() -> None: | |
| """Under pythonw.exe (the Scheduled Task launcher) there is no console: sys.stdout/stderr | |
| are None and ANY print/progress-bar write dies with "'NoneType' object has no attribute | |
| 'write'". Route them to devnull so the run only speaks via runner.log + the report file. | |
| Confirmed live 2026-08-07: the task's first real run failed inside data_plane.boot_sync's | |
| huggingface_hub progress-bar write for exactly this reason -- must run before that import.""" | |
| if sys.stdout is None: | |
| sys.stdout = open(os.devnull, "w", encoding="utf-8") # noqa: SIM115 — lives for the process | |
| if sys.stderr is None: | |
| sys.stderr = open(os.devnull, "w", encoding="utf-8") # noqa: SIM115 | |
| _devnull_headless_streams() | |
| import data_plane # noqa: E402 | |
| import epa # noqa: E402 | |
| import lcfs # noqa: E402 | |
| import scripts.refresh_quality as refresh_quality # noqa: E402 | |
| import state_paths # noqa: E402 | |
| from sinetti_services import corpus_watch, job_activation # noqa: E402 | |
| from sinetti_services import source_checks # noqa: E402 | |
| from sinetti_services.us_local_email_job import run_us_local_email_step # noqa: E402 | |
| REPORT_NAME = "last_us_local_refresh.json" | |
| LOG_NAME = "runner.log" | |
| DEFAULT_STAGING_DIR = ROOT / "data" / "refresh" / "us_local" | |
| def _manifest_path() -> Path: | |
| return ROOT / "data" / "us_registers_counts.json" | |
| def load_env_file(path, environ=os.environ) -> int: | |
| """Load KEY=VALUE lines from a .env file into `environ`. Existing keys are NEVER overwritten | |
| and values are NEVER printed. Returns the number of keys loaded. Missing file -> 0.""" | |
| p = Path(path) | |
| if not p.exists(): | |
| return 0 | |
| loaded = 0 | |
| for line in p.read_text(encoding="utf-8").splitlines(): | |
| line = line.strip() | |
| if not line or line.startswith("#") or "=" not in line: | |
| continue | |
| key, _, value = line.partition("=") | |
| key = key.strip() | |
| value = value.strip().strip('"').strip("'") | |
| if key and key not in environ: | |
| environ[key] = value | |
| loaded += 1 | |
| return loaded | |
| def _cached_login_token() -> str: | |
| try: | |
| from huggingface_hub import get_token | |
| return get_token() or "" | |
| except Exception: | |
| return "" | |
| def _log_line(staging: Path, text: str) -> None: | |
| staging.mkdir(parents=True, exist_ok=True) | |
| stamp = datetime.now(timezone.utc).isoformat(timespec="seconds") | |
| with open(staging / LOG_NAME, "a", encoding="utf-8") as fh: | |
| fh.write(f"{stamp} {text}\n") | |
| def _write_report(staging: Path, report: dict) -> None: | |
| staging.mkdir(parents=True, exist_ok=True) | |
| (staging / REPORT_NAME).write_text(json.dumps(report, indent=2, default=str), encoding="utf-8") | |
| _SERVING_PAIR = ("sinetti_app.sqlite.gz", "sinetti_data_release.json") | |
| def _artifact_pathway_count(artifact_gz: Path) -> int: | |
| return _artifact_row_count(artifact_gz, "v_app_pathways") | |
| def _artifact_row_count(artifact_gz: Path, table: str) -> int: | |
| import gzip as _gzip | |
| import sqlite3 as _sqlite3 | |
| with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as tmp: | |
| tmp_path = Path(tmp.name) | |
| try: | |
| with _gzip.open(artifact_gz, "rb") as fi, open(tmp_path, "wb") as fo: | |
| shutil.copyfileobj(fi, fo) | |
| conn = _sqlite3.connect(tmp_path) | |
| try: | |
| return int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]) | |
| finally: | |
| conn.close() | |
| finally: | |
| tmp_path.unlink(missing_ok=True) | |
| def _drip_register_site_geocode(artifact_gz: Path | None) -> dict: | |
| """Budgeted Google drip for newly projected EPA/LCFS facility sites. Fail-open always.""" | |
| if artifact_gz is None or not Path(artifact_gz).exists(): | |
| return {"status": "skipped", "reason": "no-artifact"} | |
| try: | |
| import scripts.daily_refresh as daily_refresh | |
| import scripts.geocode_budget as geocode_budget | |
| import scripts.geocode_holders_google as google_geocoder | |
| import scripts.geocode_sites as site_geocoder | |
| import state_paths | |
| from sinetti_data import load_app_sites | |
| rows = [ | |
| row | |
| for row in load_app_sites(artifact_gz) | |
| if str(row.get("site_source") or "") | |
| in {"epa_facility", "lcfs_facility", "lcfs_facility_coarse"} | |
| ] | |
| if not rows: | |
| return {"status": "skipped", "reason": "no-register-facility-sites", "attempted": 0} | |
| # Straight to the site geocoder on the shared holder ledger: the nightly's | |
| # geocode_sites_with_budget now runs a site-accurate-only drip on its own ledger | |
| # (SIN-0809), which would drop coarse LCFS facilities and switch this drip off. | |
| ledger_path = state_paths.refresh_usage_path() | |
| daily_limit = daily_refresh._env_int("GEOCODE_DAILY_LIMIT", geocode_budget.DEFAULT_DAILY_LIMIT) | |
| monthly_limit = daily_refresh._env_int("GEOCODE_MONTHLY_LIMIT", geocode_budget.DEFAULT_MONTHLY_LIMIT) | |
| key = google_geocoder._key() | |
| if not key: | |
| return {"status": "skipped-no-google-key", "attempted": 0, "geocoded": 0} | |
| remaining = geocode_budget.remaining_calls( | |
| geocode_budget.load_usage(ledger_path), daily_limit=daily_limit, monthly_limit=monthly_limit, | |
| ) | |
| # Cap per US refresh so one weekend run cannot burn the whole daily budget. | |
| limit = min(int(os.environ.get("US_SITE_GEOCODE_MAX_N", "400") or 400), remaining) | |
| if limit <= 0: | |
| return {"status": "skipped-budget-exhausted", "remaining_before": remaining, | |
| "attempted": 0, "geocoded": 0} | |
| result = site_geocoder.run_geocode_sites( | |
| max_n=limit, | |
| key=key, | |
| dry_run=False, | |
| sites_rows=rows, | |
| budget_ledger_path=ledger_path, | |
| daily_limit=daily_limit, | |
| monthly_limit=monthly_limit, | |
| ) | |
| return { | |
| "status": result.get("status") or "ok", | |
| "remaining_before": remaining, | |
| "attempted": int(result.get("attempted") or 0), | |
| "geocoded": int(result.get("geocoded") or 0), | |
| "failed": int(result.get("failed") or 0), | |
| "todo": int(result.get("todo") or 0), | |
| "detail": result, | |
| } | |
| except Exception as exc: # noqa: BLE001 — geocode must never block publish | |
| return { | |
| "status": "error", | |
| "error": f"{type(exc).__name__}: {str(exc)[:300]}", | |
| "attempted": 0, | |
| "geocoded": 0, | |
| } | |
| def _rebuild_candidate_serving_pair(candidate_root: Path, rebuild_fn=None) -> dict: | |
| """Rebuild the candidate's serving pair from its own data/ (which now holds the fresh | |
| LCFS/EPA snapshots beside the pulled register bundles), validate it, and swap only the | |
| verified pair back into the candidate. INS-refresh pattern; any failure leaves the | |
| candidate's previous pair untouched and refuses the publish.""" | |
| if rebuild_fn is None: | |
| import scripts.daily_refresh as daily_refresh | |
| rebuild_fn = daily_refresh.rebuild_bundled_app_database | |
| import sinetti_data_release | |
| candidate_data = candidate_root / "data" | |
| try: | |
| with tempfile.TemporaryDirectory(prefix="sinetti-us-rebuild-") as temp_name: | |
| staged_root = Path(temp_name) / "candidate" | |
| staged_data = staged_root / "data" | |
| shutil.copytree(candidate_data, staged_data) | |
| app_db = rebuild_fn(root=staged_root, replace=True) | |
| if app_db.get("status") != "ok": | |
| raise RuntimeError(str(app_db.get("error") or "candidate rebuild did not complete")) | |
| validation = sinetti_data_release.validate_data_release( | |
| staged_data / "sinetti_app.sqlite.gz", | |
| manifest_path=staged_data / "sinetti_data_release.json", | |
| inspect_database=True, | |
| ) | |
| if not validation.get("ok"): | |
| raise RuntimeError(str(validation.get("reason") or "rebuilt serving pair is invalid")) | |
| # A rebuild in a workspace missing cert_pdf_structured.sqlite.gz "succeeds" with ZERO | |
| # v_app_pathways rows and would silently gut every deep-dive downstream. Refuse it. | |
| pathway_count = _artifact_pathway_count(staged_data / "sinetti_app.sqlite.gz") | |
| if pathway_count <= 0: | |
| raise RuntimeError( | |
| "rebuilt serving pair has zero v_app_pathways rows " | |
| "(structured PDF layer missing from the candidate workspace?)" | |
| ) | |
| # Same trap, second shape: a structured DB from before the sites layer imports with | |
| # a silent sites=[] fallback, gutting Certified Sites / Browse map annex sites | |
| # (release 79fc6880 shipped exactly this way on 2026-08-08). Refuse it too. | |
| sites_count = _artifact_row_count(staged_data / "sinetti_app.sqlite.gz", "sites") | |
| if sites_count <= 0: | |
| raise RuntimeError( | |
| "rebuilt serving pair has zero sites rows " | |
| "(structured DB predates the sites layer, or structured_sites is empty?)" | |
| ) | |
| for name in _SERVING_PAIR: | |
| shutil.copy2(staged_data / name, candidate_data / name) | |
| return { | |
| "status": "ok", | |
| "certificates": app_db.get("certificates"), | |
| "lcfs_epa_address_fills": app_db.get("lcfs_epa_address_fills"), | |
| } | |
| except Exception as exc: # noqa: BLE001 -- refuse the publish, never crash the run | |
| return {"status": "error", "error": f"{type(exc).__name__}: {str(exc)[:300]}"} | |
| def _fetch_and_normalize_lcfs(fetch_fn) -> dict: | |
| """Fetch + schema-assert + normalize LCFS. Returns a refresh-leg report dict; never raises. | |
| A failure records WHICH stage failed, because the two mean opposite things: failure='fetch' | |
| is the source being unreachable (transient, the previous snapshot is still true), while | |
| failure='schema' is a data-shape change we do not understand yet and must always refuse.""" | |
| try: | |
| header, rows = fetch_fn() | |
| except Exception as exc: # noqa: BLE001 -- any failure here must refuse, never crash the run | |
| return {"status": "error", "decision": "kept_previous", "failure": "fetch", | |
| "error": f"{type(exc).__name__}: {exc}"} | |
| try: | |
| lcfs.assert_schema(header) | |
| certs, excluded = lcfs.normalize_pathways(rows) | |
| except Exception as exc: # noqa: BLE001 | |
| return {"status": "error", "decision": "kept_previous", "failure": "schema", | |
| "error": f"{type(exc).__name__}: {exc}"} | |
| return {"status": "ok", "decision": "accepted", "rows_total": len(rows), | |
| "certs": len(certs), "excluded": len(excluded), "raw_rows": rows, "certs_list": certs} | |
| def _fetch_and_normalize_epa(fetch_fn) -> dict: | |
| try: | |
| header, rows = fetch_fn() | |
| except Exception as exc: # noqa: BLE001 | |
| return {"status": "error", "decision": "kept_previous", "failure": "fetch", | |
| "error": f"{type(exc).__name__}: {exc}"} | |
| try: | |
| epa.assert_schema(header) | |
| certs = epa.normalize_rows(rows) | |
| except Exception as exc: # noqa: BLE001 | |
| return {"status": "error", "decision": "kept_previous", "failure": "schema", | |
| "error": f"{type(exc).__name__}: {exc}"} | |
| return {"status": "ok", "decision": "accepted", "rows_total": len(rows), | |
| "certs": len(certs), "certs_list": certs} | |
| # Which corpus-band probes each register contributes. A leg that did not refresh contributes | |
| # none of its keys -- corpus_watch.check_counts requires the counts and the manifest to carry | |
| # EXACTLY the same keys, so a partial refresh has to narrow the manifest, never the counts. | |
| _LEG_BAND_KEYS = { | |
| "lcfs": ("lcfs.pathways_total", "lcfs.biofuel_subset"), | |
| "epa": ("epa.producer_rows",), | |
| } | |
| def _previous_snapshot_certs(candidate_data: Path, leg_name: str) -> list | None: | |
| """The previous snapshot for a register in the pulled candidate baseline, or None when there | |
| is nothing usable to keep. Never raises -- an unreadable snapshot counts as absent.""" | |
| loaders = { | |
| "lcfs": (lcfs.load_lcfs_certs, candidate_data / "lcfs" / "pathways.json.gz"), | |
| "epa": (epa.load_epa_certs, candidate_data / "epa" / "producers.json.gz"), | |
| } | |
| loader, path = loaders[leg_name] | |
| if not path.is_file(): | |
| return None | |
| try: | |
| certs = loader(path=path) | |
| except Exception: # noqa: BLE001 -- an unreadable previous snapshot is not something to keep | |
| return None | |
| return certs or None | |
| def _narrow_manifest(manifest: dict, keys) -> dict: | |
| """The band manifest restricted to `keys`, so a partial refresh band-checks exactly the | |
| counts it actually produced.""" | |
| wanted = set(keys) | |
| narrowed = dict(manifest) | |
| narrowed["counts"] = {k: v for k, v in (manifest.get("counts") or {}).items() if k in wanted} | |
| return narrowed | |
| def _register_publish_gate_review() -> dict: | |
| """LCFS/EPA have no PDF corpus and are not covered by the certificate-PDF Trust Review | |
| (sinetti_services/trust_review_service.py) -- its checks are scoped to ISCC/REDcert/INS | |
| certificate-format evidence (data_trust_control_service, cert_format_catalog.json), none of | |
| which exists for these registers. By the time upload succeeds below, this run's own hard | |
| gates have ALREADY passed -- schema assert, corpus_watch bands (this register family's | |
| coverage-check equivalent), and rebuild + validate_data_release -- because publish never | |
| happens otherwise (see `ok` above). That IS this register family's equivalent of a clean | |
| review. Shaped to satisfy job_activation.trust_review_clean() with a zero-finding | |
| "completed" review, honestly labeled as this run's own publish gate, not a PDF review.""" | |
| return {"status": "completed", "checks": [], "source": "us_local_refresh_publish_gate"} | |
| def _exception_http_status(exc: Exception) -> int | None: | |
| response = getattr(exc, "response", None) | |
| status = getattr(response, "status_code", None) | |
| return status if isinstance(status, int) else None | |
| def _is_publish_conflict(exc: Exception) -> bool: | |
| return _exception_http_status(exc) == 412 | |
| def _finalize_report(staging: Path, report: dict, rc: int) -> None: | |
| finished = datetime.now(timezone.utc) | |
| started = datetime.fromisoformat(report["started_at"]) | |
| if not (rc == 0 and report.get("status") == "unchanged"): | |
| report["status"] = "ok" if rc == 0 else "error" | |
| report["finished_at"] = finished.isoformat() | |
| report["duration_s"] = round((finished - started).total_seconds(), 2) | |
| _write_report(staging, report) | |
| refresh_quality.append_history_files(staging, report) | |
| # Operator brief last. Member flags already sit on the shared change log via | |
| # append_history_files, so this step does not send a second subscriber digest. | |
| try: | |
| report["email"] = run_us_local_email_step(report, history_dir=staging) | |
| except Exception as exc: # noqa: BLE001 | |
| report["email"] = { | |
| "status": f"error: {type(exc).__name__}: {str(exc)[:200]}", | |
| "admin": "not-attempted", "digest": "not-attempted", "red_flag_count": 0, | |
| } | |
| _write_report(staging, report) | |
| _log_line(staging, f"status={report['status']} rc={rc} upload={report['upload'].get('status')} " | |
| f"activation={report.get('job_activation', {}).get('status', '?')}") | |
| def main(argv: list[str] | None = None, *, fetch_lcfs_fn=None, fetch_epa_fn=None, | |
| fetch_lcfs_last_modified_fn=None, probe_lcfs_fn=None, probe_epa_fn=None, | |
| rebuild_fn=None, environ=os.environ, env_file=None, | |
| _candidate_root: Path | None = None, _defer_finalize: bool = False, | |
| _attempt_result: dict | None = None, _site_geocode_report: dict | None = None) -> int: | |
| from sinetti_operator.registration import register_live_providers | |
| register_live_providers() # jobs never reach full_registration (SIN-0999) | |
| env_path = env_file if env_file is not None else ROOT / ".env" | |
| load_env_file(env_path, environ=environ) | |
| ap = argparse.ArgumentParser(description=__doc__) | |
| ap.add_argument("--dry-run", action="store_true", help="print config, no network calls") | |
| ap.add_argument("--staging-dir", default=str(DEFAULT_STAGING_DIR)) | |
| args = ap.parse_args(argv) | |
| staging = Path(args.staging_dir) | |
| token = (environ.get("HF_TOKEN", "") or environ.get("HF_JOB_TOKEN", "") | |
| or environ.get("HF_CREATE_TOKEN", "") or _cached_login_token()) | |
| data_repo_id = environ.get("SINETTI_DATA_REPO_ID", "") | |
| if args.dry_run: | |
| print(json.dumps({"dry_run": True, "data_repo_id": data_repo_id, | |
| "staging_dir": str(staging), "hf_token_present": bool(token)}, indent=2)) | |
| return 0 | |
| if not data_repo_id: | |
| print("REFUSING: SINETTI_DATA_REPO_ID not set -- this pipeline publishes only through " | |
| "the private dataset candidate/release flow, no legacy fallback.") | |
| _log_line(staging, "refused: SINETTI_DATA_REPO_ID missing") | |
| return 2 | |
| if not token: | |
| print("REFUSING: no HF token (HF_TOKEN / HF_JOB_TOKEN / HF_CREATE_TOKEN) in environment, " | |
| ".env, or a cached `huggingface-cli login`.") | |
| _log_line(staging, "refused: HF_TOKEN missing") | |
| return 2 | |
| if _candidate_root is None: | |
| orchestration_started = datetime.now(timezone.utc).isoformat() | |
| previous_geocode_report = None | |
| final_report = None | |
| final_rc = 1 | |
| attempts = 0 | |
| conflicts = 0 | |
| for attempt_index in range(2): | |
| attempts += 1 | |
| attempt_result: dict = {} | |
| with tempfile.TemporaryDirectory(prefix="sinetti-us-candidate-") as workspace_name: | |
| final_rc = main( | |
| argv, fetch_lcfs_fn=fetch_lcfs_fn, fetch_epa_fn=fetch_epa_fn, | |
| fetch_lcfs_last_modified_fn=fetch_lcfs_last_modified_fn, | |
| probe_lcfs_fn=probe_lcfs_fn, probe_epa_fn=probe_epa_fn, | |
| rebuild_fn=rebuild_fn, environ=environ, env_file=env_file, | |
| _candidate_root=Path(workspace_name), _defer_finalize=True, | |
| _attempt_result=attempt_result, _site_geocode_report=previous_geocode_report, | |
| ) | |
| final_report = attempt_result.get("report") | |
| if final_report is None: | |
| # Defensive fallback for an internal attempt that returned before making a report. | |
| return final_rc | |
| if not final_report.get("upload", {}).get("retryable_conflict") or attempt_index == 1: | |
| break | |
| conflicts += 1 | |
| previous_geocode_report = final_report["refresh"].get("site_geocode") | |
| assert final_report is not None | |
| final_report["started_at"] = orchestration_started | |
| final_report["publish_attempts"] = attempts | |
| final_report["publish_conflicts_recovered"] = conflicts if final_rc == 0 else 0 | |
| _finalize_report(staging, final_report, final_rc) | |
| return final_rc | |
| candidate_root = Path(_candidate_root) | |
| candidate_data = candidate_root / "data" | |
| baseline = data_plane.boot_sync( | |
| repo_id=data_repo_id, token=token, cache_dir=candidate_root / "data-cache", | |
| data_dir=candidate_data, materialize=True, | |
| ) | |
| baseline_failed = baseline.get("status") != "ok" or not baseline.get("hub_revision") | |
| started = datetime.now(timezone.utc) | |
| run_id = started.isoformat() | |
| if baseline_failed: | |
| report = { | |
| "started_at": run_id, "runner": "us-local-refresh", | |
| "baseline": baseline, | |
| "refresh": {"lcfs": {"status": "error", "decision": "kept_previous", | |
| "error": "verified dataset baseline unavailable"}, | |
| "epa": {"status": "error", "decision": "kept_previous", | |
| "error": "verified dataset baseline unavailable"}}, | |
| "upload": {"status": "skipped-baseline-error"}, | |
| "status": "error", | |
| } | |
| if _defer_finalize: | |
| if _attempt_result is not None: | |
| _attempt_result.update({"report": report, "rc": 1}) | |
| return 1 | |
| _write_report(staging, report) | |
| _log_line(staging, f"status=error rc=1 reason=baseline_failed") | |
| return 1 | |
| fetch_lcfs_fn = fetch_lcfs_fn or lcfs.fetch_lcfs_rows | |
| fetch_epa_fn = fetch_epa_fn or epa.fetch_epa_rows | |
| fetch_lcfs_last_modified_fn = fetch_lcfs_last_modified_fn or lcfs.fetch_lcfs_source_last_modified | |
| # Probe both publishers before downloading. CARB serves Last-Modified (2026-08-08 when | |
| # measured) so its gate is real; EPA's CDX endpoint serves neither Last-Modified nor ETag, | |
| # so decide_skip falls back to Content-Length -- a weak signal, and EPA is the register | |
| # that moves most (8 added / 8 deregistered in the six days after 2026-08-08), so that | |
| # gate rarely fires. Both are advisory: a failed probe fetches as before. | |
| lcfs_gate = source_checks.gate( | |
| "LCFS", probe_fn=probe_lcfs_fn or lcfs.probe_lcfs_source, | |
| bundle_path=candidate_data / "lcfs" / "pathways.json.gz", checked_at=run_id, | |
| ) | |
| epa_gate = source_checks.gate( | |
| "EPA", probe_fn=probe_epa_fn or epa.probe_epa_source, | |
| bundle_path=candidate_data / "epa" / "producers.json.gz", checked_at=run_id, | |
| ) | |
| if lcfs_gate["skip"]: | |
| lcfs_leg = {"status": "unchanged", "decision": "kept_previous", | |
| "reason": lcfs_gate["reason"], "signal": lcfs_gate["signal"], | |
| "source_last_modified": lcfs_gate["source_last_modified"]} | |
| else: | |
| lcfs_leg = _fetch_and_normalize_lcfs(fetch_lcfs_fn) | |
| lcfs_leg["probe"] = lcfs_gate["reason"] | |
| if epa_gate["skip"]: | |
| epa_leg = {"status": "unchanged", "decision": "kept_previous", | |
| "reason": epa_gate["reason"], "signal": epa_gate["signal"]} | |
| else: | |
| epa_leg = _fetch_and_normalize_epa(fetch_epa_fn) | |
| epa_leg["probe"] = epa_gate["reason"] | |
| # Advisory only -- CARB's own Last-Modified header, separate from the row fetch above and | |
| # never allowed to affect the schema/publish decision. A HEAD-request failure here must | |
| # not touch lcfs_leg["status"]; it only leaves source_last_modified blank. | |
| lcfs_source_last_modified = lcfs_gate["source_last_modified"] or "" | |
| if lcfs_leg["status"] == "ok" and not lcfs_source_last_modified: | |
| try: | |
| lcfs_source_last_modified = fetch_lcfs_last_modified_fn() or "" | |
| except Exception: # noqa: BLE001 -- advisory only, never blocks the refresh | |
| lcfs_source_last_modified = "" | |
| if lcfs_leg["status"] == "ok": | |
| lcfs_leg["source_last_modified"] = lcfs_source_last_modified | |
| legs = {"lcfs": lcfs_leg, "epa": epa_leg} | |
| # A source being unreachable must not strand the OTHER source's fresh data, and must not | |
| # strand build-time normalisation fixes either -- the candidate is rebuilt from the | |
| # published snapshots, so a rebuild is worth doing even when only one register moved. | |
| # (CARB served HTTP 503 from its own origin on 2026-08-08 while EPA fetched fine.) The leg | |
| # that failed keeps its previous snapshot, and only if there IS one: republishing a | |
| # register as absent would look exactly like the register losing every row. | |
| # Schema drift is never tolerated here -- an unexplained data-shape change still refuses. | |
| blocking: list[str] = [] | |
| for name, leg in legs.items(): | |
| if leg["status"] == "ok": | |
| continue | |
| if leg["status"] == "unchanged": | |
| # Publisher reported nothing moved: nothing to accept, nothing to block. | |
| continue | |
| if leg.get("failure") != "fetch": | |
| blocking.append(name) | |
| continue | |
| if _previous_snapshot_certs(candidate_data, name) is None: | |
| leg["error"] = ( | |
| f"{leg.get('error', '')} (no usable previous {name.upper()} snapshot in the " | |
| "candidate baseline to keep)" | |
| ).strip() | |
| blocking.append(name) | |
| accepted = [name for name, leg in legs.items() if leg["status"] == "ok"] | |
| all_unchanged = not blocking and not accepted and all( | |
| leg["status"] == "unchanged" for leg in legs.values() | |
| ) | |
| # Nothing fresh anywhere means nothing to publish: a republish of unchanged snapshots would | |
| # burn a Hub revision and a release for no data change. | |
| legs_ok = not blocking and bool(accepted) | |
| bands_report: dict = {"status": "skipped", "reason": "refresh not accepted"} | |
| if legs_ok: | |
| counts: dict[str, int] = {} | |
| if "lcfs" in accepted: | |
| counts["lcfs.pathways_total"] = lcfs_leg["rows_total"] | |
| counts["lcfs.biofuel_subset"] = lcfs_leg["certs"] | |
| if "epa" in accepted: | |
| counts["epa.producer_rows"] = epa_leg["certs"] | |
| manifest = corpus_watch.load_manifest(_manifest_path()) | |
| bands_report = corpus_watch.counts_report(counts, _narrow_manifest(manifest, counts)) | |
| skipped_keys = sorted( | |
| key | |
| for name, leg_keys in _LEG_BAND_KEYS.items() | |
| if name not in accepted | |
| for key in leg_keys | |
| ) | |
| bands_report["skipped_keys"] = skipped_keys | |
| if skipped_keys: | |
| bands_report["skipped_reason"] = "source unavailable; previous snapshot kept" | |
| ok = legs_ok and bands_report.get("status") == "ok" | |
| change_events = {"lcfs": [], "epa": []} | |
| if ok: | |
| if "lcfs" in accepted: | |
| old_lcfs = lcfs.load_lcfs_certs(path=candidate_data / "lcfs" / "pathways.json.gz") | |
| lcfs_diff = lcfs.diff_lcfs_snapshot(old_lcfs, lcfs_leg["certs_list"]) | |
| change_events["lcfs"] = refresh_quality.build_register_diff_events( | |
| run_id, "LCFS", old_lcfs, lcfs_leg["certs_list"], lcfs_diff, detected_at=run_id, | |
| ) | |
| lcfs.write_snapshot( | |
| lcfs_leg["certs_list"], path=candidate_data / "lcfs" / "pathways.json.gz", | |
| fetched_at=run_id, source="us_local_refresh", | |
| source_last_modified=lcfs_source_last_modified or None, | |
| ) | |
| if "epa" in accepted: | |
| old_epa = epa.load_epa_certs(path=candidate_data / "epa" / "producers.json.gz") | |
| epa_diff = epa.diff_epa_snapshot(old_epa, epa_leg["certs_list"]) | |
| change_events["epa"] = refresh_quality.build_register_diff_events( | |
| run_id, "EPA", old_epa, epa_leg["certs_list"], epa_diff, detected_at=run_id, | |
| ) | |
| epa.write_snapshot( | |
| epa_leg["certs_list"], path=candidate_data / "epa" / "producers.json.gz", | |
| fetched_at=run_id, source="us_local_refresh", | |
| ) | |
| # Rebuild the candidate serving pair so the published candidate actually CONTAINS the | |
| # US registers (INS-refresh pattern): stage a copy, rebuild the app DB there, validate, | |
| # then swap only the verified pair back. Snapshot-only publishes left LCFS/EPA invisible | |
| # to serving until some other refresh happened to rebuild. | |
| app_db_report: dict = {"status": "skipped", "reason": "refresh not accepted"} | |
| if ok: | |
| app_db_report = _rebuild_candidate_serving_pair(candidate_root, rebuild_fn) | |
| ok = app_db_report.get("status") == "ok" | |
| # Fail-open drip: new EPA/LCFS facility sites need lat/lng in data/site_coords.json. | |
| # Never refuse the publish for geocode budget / missing key. | |
| site_geocode_report = _site_geocode_report | |
| if site_geocode_report is None: | |
| site_geocode_report = _drip_register_site_geocode( | |
| candidate_root / "data" / "sinetti_app.sqlite.gz" if ok else None | |
| ) | |
| def _leg_report(leg: dict) -> dict: | |
| out = {k: v for k, v in leg.items() if k not in ("raw_rows", "certs_list")} | |
| return out | |
| report: dict = { | |
| "started_at": run_id, "runner": "us-local-refresh", | |
| "baseline": baseline, | |
| "refresh": { | |
| "lcfs": {**_leg_report(lcfs_leg), "change_events": change_events["lcfs"]} if lcfs_leg["status"] == "ok" else lcfs_leg, | |
| "epa": {**_leg_report(epa_leg), "change_events": change_events["epa"]} if epa_leg["status"] == "ok" else epa_leg, | |
| "bands": bands_report, | |
| "app_db": app_db_report, | |
| "site_geocode": site_geocode_report, | |
| }, | |
| "upload": {"status": "pending"}, | |
| "job_activation": {"status": "not-attempted", "reason": "dataset release not published"}, | |
| } | |
| rc = 0 | |
| if all_unchanged: | |
| report["upload"] = {"status": "skipped-unchanged", | |
| "reason": "every source reported unchanged; nothing to publish"} | |
| report["status"] = "unchanged" | |
| elif not legs_ok: | |
| report["upload"] = {"status": "skipped-refresh-error"} | |
| rc = 1 | |
| elif bands_report.get("status") != "ok": | |
| report["upload"] = {"status": "skipped-band-violation"} | |
| rc = 1 | |
| elif app_db_report.get("status") != "ok": | |
| report["upload"] = {"status": "skipped-rebuild-error"} | |
| rc = 1 | |
| else: | |
| try: | |
| result = data_plane.publish( | |
| candidate_root, repo_id=data_repo_id, token=token, | |
| producer="us_local_refresh", expected_parent=baseline.get("hub_revision"), | |
| ) | |
| except Exception as exc: # noqa: BLE001 | |
| report["upload"] = {"status": "error", | |
| "exception": {"class": type(exc).__name__, "message": str(exc)}} | |
| http_status = _exception_http_status(exc) | |
| if http_status is not None: | |
| report["upload"]["http_status"] = http_status | |
| if _is_publish_conflict(exc): | |
| report["upload"]["retryable_conflict"] = True | |
| rc = 1 | |
| report["job_activation"] = { | |
| "status": "skipped", "reason": "dataset publish raised an exception", | |
| } | |
| else: | |
| report["upload"] = result | |
| if result.get("status") != "ok": | |
| rc = 1 | |
| report["job_activation"] = { | |
| "status": "skipped", "reason": "dataset publish did not succeed", | |
| } | |
| else: | |
| try: | |
| report["job_activation"] = job_activation.decide_and_activate( | |
| enabled=job_activation.job_auto_activate_enabled(environ), | |
| ok=True, | |
| revision=result.get("commit_sha"), | |
| candidate_data_dir=candidate_data, | |
| review=_register_publish_gate_review(), | |
| state_root=state_paths.state_root(), | |
| by="us-local-refresh", | |
| ) | |
| except Exception as exc: # noqa: BLE001 -- publish remains authoritative | |
| report["job_activation"] = { | |
| "status": "hold", | |
| "reason": "dataset published; automatic job activation raised an exception", | |
| "exception": {"class": type(exc).__name__, "message": str(exc)}, | |
| } | |
| if _defer_finalize: | |
| if _attempt_result is not None: | |
| _attempt_result.update({"report": report, "rc": rc}) | |
| return rc | |
| _finalize_report(staging, report, rc) | |
| return rc | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |