#!/usr/bin/env python """Weekly LOCAL Oregon + BC registers refresh: Oregon CFP (state xlsx) + BC LCFS (fuel-codes API). Cloud egress unverified for both endpoints (INS precedent: GSE blocked HF Jobs + GitHub Actions datacenter IPs); both sources are small and weekly cadence suffices, so this refreshes HERE, local, and publishes through the same private-dataset candidate/release flow the daily/INS/CFR/US refreshes use -- never a direct serving-data write, never auto-activation. fetch Oregon xlsx (browser UA) + BC fuel-codes API (current + archived, paginated) -> schema assert (refuse loudly on drift) -> normalize (oregon.py / bc_lcfs.py) -> corpus_watch row-count bands (data/orbc_registers_counts.json -- a DEDICATED manifest, see tests/test_orbc_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. No geocode drip, no job auto-activation -- unlike scripts/us_local_refresh.py, this mirrors scripts/cfr_local_refresh.py's simpler ending (spec 2026-08-12 asks only for the fetch -> normalize -> band -> diff -> rebuild -> publish pipeline). Windows task name: "Sinetti PC weekly OR-BC refresh" (no slash -- Task Scheduler reads one as a folder separator and refuses with 0x80070057) -- registered ONLY after a verified manual run (CFR lesson: never register blind). Failure leaves serving untouched. Usage: python scripts/orbc_local_refresh.py # full run: fetch + publish python scripts/orbc_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.""" 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 bc_lcfs # noqa: E402 import data_plane # noqa: E402 import oregon # noqa: E402 import scripts.refresh_quality as refresh_quality # noqa: E402 from sinetti_services import corpus_watch # noqa: E402 from sinetti_services import source_checks # noqa: E402 REPORT_NAME = "last_orbc_local_refresh.json" LOG_NAME = "runner.log" DEFAULT_STAGING_DIR = ROOT / "data" / "refresh" / "orbc_local" def _manifest_path() -> Path: return ROOT / "data" / "orbc_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 _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 Oregon/BC snapshots beside the pulled register bundles), validate it, and swap only the verified pair back into the candidate. INS/US/CFR-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-orbc-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")) 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?)" ) 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"), "oregon_epa_address_fills": app_db.get("oregon_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_oregon(fetch_fn) -> dict: """Fetch + schema-assert + normalize Oregon. Returns a refresh-leg report dict; never raises. failure='fetch' means the source was unreachable (transient, previous snapshot still true); failure='schema' means a data-shape change we do not understand yet -- always refuses.""" 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: oregon.assert_schema(header) certs, excluded = oregon.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_bc(fetch_fn) -> dict: try: keys, rows = fetch_fn() except Exception as exc: # noqa: BLE001 return {"status": "error", "decision": "kept_previous", "failure": "fetch", "error": f"{type(exc).__name__}: {exc}"} try: bc_lcfs.assert_schema(keys) certs, excluded = bc_lcfs.normalize_bulletin_items(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} # 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 = { "oregon": ("oregon.rows_total", "oregon.active_in_snapshot"), "bc": ("bc.rows_total", "bc.active_in_snapshot"), } 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 = { "oregon": (oregon.load_oregon_certs, candidate_data / "oregon" / "pathways.json.gz"), "bc": (bc_lcfs.load_bc_certs, candidate_data / "bc" / "fuel_codes.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 _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) _log_line(staging, f"status={report['status']} rc={rc} upload={report['upload'].get('status')}") def main(argv: list[str] | None = None, *, fetch_oregon_fn=None, fetch_bc_fn=None, fetch_oregon_last_modified_fn=None, probe_oregon_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) -> 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() 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-orbc-candidate-") as workspace_name: final_rc = main( argv, fetch_oregon_fn=fetch_oregon_fn, fetch_bc_fn=fetch_bc_fn, fetch_oregon_last_modified_fn=fetch_oregon_last_modified_fn, probe_oregon_fn=probe_oregon_fn, rebuild_fn=rebuild_fn, environ=environ, env_file=env_file, _candidate_root=Path(workspace_name), _defer_finalize=True, _attempt_result=attempt_result, ) final_report = attempt_result.get("report") if final_report is None: return final_rc if not final_report.get("upload", {}).get("retryable_conflict") or attempt_index == 1: break conflicts += 1 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": "orbc-local-refresh", "baseline": baseline, "refresh": {"oregon": {"status": "error", "decision": "kept_previous", "error": "verified dataset baseline unavailable"}, "bc": {"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, "status=error rc=1 reason=baseline_failed") return 1 fetch_oregon_fn = fetch_oregon_fn or oregon.fetch_oregon_rows fetch_bc_fn = fetch_bc_fn or bc_lcfs.fetch_bc_rows fetch_oregon_last_modified_fn = ( fetch_oregon_last_modified_fn or oregon.fetch_oregon_source_last_modified ) # Probe Oregon before paying for the workbook. Measured 2026-08-14: DEQ's file had not # moved since 2026-04-03, so a weekly run had been downloading and re-parsing the same # bytes for 19 weeks. BC has no equivalent gate (its API answers no HEAD) and needs none: # both bulletin buckets are a few hundred rows, and BC is the register that actually moves. oregon_gate = source_checks.gate( "Oregon", probe_fn=probe_oregon_fn or oregon.probe_oregon_source, bundle_path=candidate_data / "oregon" / "pathways.json.gz", checked_at=run_id, ) if oregon_gate["skip"]: oregon_leg = {"status": "unchanged", "decision": "kept_previous", "reason": oregon_gate["reason"], "signal": oregon_gate["signal"], "source_last_modified": oregon_gate["source_last_modified"]} else: oregon_leg = _fetch_and_normalize_oregon(fetch_oregon_fn) oregon_leg["probe"] = oregon_gate["reason"] bc_leg = _fetch_and_normalize_bc(fetch_bc_fn) # Advisory only -- Oregon DEQ's own Last-Modified header, separate from the row fetch above # and never allowed to affect the schema/publish decision. oregon_source_last_modified = oregon_gate["source_last_modified"] or "" if oregon_leg["status"] == "ok" and not oregon_source_last_modified: # The gate normally supplies this from its own HEAD; fall back to the dedicated call # only when the probe came back empty, so a fetched leg still records DEQ's own date. try: oregon_source_last_modified = fetch_oregon_last_modified_fn() or "" except Exception: # noqa: BLE001 -- advisory only, never blocks the refresh oregon_source_last_modified = "" if oregon_leg["status"] == "ok": oregon_leg["source_last_modified"] = oregon_source_last_modified legs = {"oregon": oregon_leg, "bc": bc_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. 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": # The publisher told us nothing moved. Not accepted (there is nothing new to # write) and emphatically not blocking -- this is the healthy outcome. 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} snapshot in the " "candidate baseline to keep)" ).strip() blocking.append(name) accepted = [name for name, leg in legs.items() if leg["status"] == "ok"] # Every register said "unchanged": a complete, successful run with nothing to publish. # Distinguished from legs_ok so it never reaches the band/rebuild/publish path at all. all_unchanged = not blocking and not accepted and all( leg["status"] == "unchanged" for leg in legs.values() ) legs_ok = not blocking and bool(accepted) bands_report: dict = {"status": "skipped", "reason": "refresh not accepted"} if legs_ok: counts: dict[str, int] = {} if "oregon" in accepted: counts["oregon.rows_total"] = oregon_leg["rows_total"] counts["oregon.active_in_snapshot"] = sum( 1 for c in oregon_leg["certs_list"] if c.get("status") == "active" ) if "bc" in accepted: counts["bc.rows_total"] = bc_leg["rows_total"] counts["bc.active_in_snapshot"] = sum( 1 for c in bc_leg["certs_list"] if c.get("status") == "active" ) 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 = {"oregon": [], "bc": []} if ok: if "oregon" in accepted: old_oregon = oregon.load_oregon_certs(path=candidate_data / "oregon" / "pathways.json.gz") oregon_diff = oregon.diff_oregon_snapshot(old_oregon, oregon_leg["certs_list"]) change_events["oregon"] = refresh_quality.build_register_diff_events( run_id, "Oregon", old_oregon, oregon_leg["certs_list"], oregon_diff, detected_at=run_id, ) oregon.write_snapshot( oregon_leg["certs_list"], path=candidate_data / "oregon" / "pathways.json.gz", fetched_at=run_id, source="orbc_local_refresh", source_last_modified=oregon_source_last_modified or None, ) if "bc" in accepted: old_bc = bc_lcfs.load_bc_certs(path=candidate_data / "bc" / "fuel_codes.json.gz") bc_diff = bc_lcfs.diff_bc_snapshot(old_bc, bc_leg["certs_list"]) change_events["bc"] = refresh_quality.build_register_diff_events( run_id, "BC", old_bc, bc_leg["certs_list"], bc_diff, detected_at=run_id, ) bc_source_last_modified = bc_lcfs.derive_bc_source_last_modified( bc_leg["raw_rows"], asof=started.date() ) bc_leg["source_last_modified"] = bc_source_last_modified bc_lcfs.write_snapshot( bc_leg["certs_list"], path=candidate_data / "bc" / "fuel_codes.json.gz", fetched_at=run_id, source="orbc_local_refresh", source_last_modified=bc_source_last_modified or None, ) 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" def _leg_report(leg: dict) -> dict: return {k: v for k, v in leg.items() if k not in ("raw_rows", "certs_list")} report: dict = { "started_at": run_id, "runner": "orbc-local-refresh", "baseline": baseline, "refresh": { "oregon": {**_leg_report(oregon_leg), "change_events": change_events["oregon"]} if oregon_leg["status"] == "ok" else oregon_leg, "bc": {**_leg_report(bc_leg), "change_events": change_events["bc"]} if bc_leg["status"] == "ok" else bc_leg, "bands": bands_report, "app_db": app_db_report, }, "upload": {"status": "pending"}, } 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="orbc_local_refresh", expected_parent=baseline.get("hub_revision"), ) report["upload"] = result if result.get("status") != "ok": rc = 1 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 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())