#!/usr/bin/env python
"""Weekly LOCAL CFR refresh: federal Clean Fuel Regulations approved CI list (ECCC).

Cloud egress unverified for the ECCC endpoint (INS precedent: GSE blocked HF Jobs + GitHub Actions
datacenter IPs); the file is tiny 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 ECCC CSV via /api/file (browser UA) -> schema assert (refuse loudly on drift)
    -> normalize (cfr.py) -> corpus_watch row-count bands
       (data/cfr_registers_counts.json -- a DEDICATED manifest, see tests/test_cfr_registers_counts.py
       for why this never touches the shared data/corpus_counts.json)
    -> snapshot-diff for change events -> data_plane.publish

Any download/schema/band failure leaves current serving data untouched (no publish call at all)
and is recorded loudly in the report. Registered as a weekly Windows Scheduled Task
("Sinetti PC weekly CFR refresh"), weekend slot (see docs/ins-local-refresh-runbook.md for the sibling
INS task this mirrors).

Usage:
  python scripts/cfr_local_refresh.py            # full run: fetch + publish
  python scripts/cfr_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 cfr  # noqa: E402
import data_plane  # 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_cfr_local_refresh.json"
LOG_NAME = "runner.log"
DEFAULT_STAGING_DIR = ROOT / "data" / "refresh" / "cfr_local"


def _manifest_path() -> Path:
    return ROOT / "data" / "cfr_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/, validate, and swap only the
    verified pair back into the candidate. 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-cfr-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"),
        }
    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_cfr(fetch_fn) -> dict:
    """Fetch + schema-assert + normalize CFR. Returns a refresh-leg report dict; never raises --
    a fetch or schema-drift failure comes back as decision='kept_previous'."""
    try:
        header, rows = fetch_fn()
        cfr.assert_schema(header)
    except Exception as exc:  # noqa: BLE001 -- any failure here must refuse, never crash the run
        return {"status": "error", "decision": "kept_previous",
                "error": f"{type(exc).__name__}: {exc}"}
    certs, excluded = cfr.normalize_pathways(rows)
    return {"status": "ok", "decision": "accepted", "rows_total": len(rows),
            "certs": len(certs), "excluded": len(excluded), "raw_rows": rows, "certs_list": certs}


def _snapshot_status_counts(certs_list: list[dict]) -> dict[str, int]:
    return {
        "cfr.rows_total": 0,  # filled by caller from rows_total
        "cfr.active_in_snapshot": sum(1 for c in certs_list if c.get("status") == "active"),
        "cfr.inactive_in_snapshot": sum(1 for c in certs_list if c.get("status") == "withdrawn"),
    }


def main(argv: list[str] | None = None, *, fetch_cfr_fn=None, probe_cfr_fn=None,
         rebuild_fn=None, environ=os.environ, env_file=None,
         _candidate_root: Path | 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:
        workspace = tempfile.TemporaryDirectory(prefix="sinetti-cfr-candidate-")
        try:
            return main(
                argv, fetch_cfr_fn=fetch_cfr_fn, rebuild_fn=rebuild_fn,
                environ=environ, env_file=env_file,
                _candidate_root=Path(workspace.name),
            )
        finally:
            workspace.cleanup()

    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": "cfr-local-refresh",
            "baseline": baseline,
            "refresh": {"cfr": {"status": "error", "decision": "kept_previous",
                                "error": "verified dataset baseline unavailable"}},
            "upload": {"status": "skipped-baseline-error"},
            "status": "error",
        }
        _write_report(staging, report)
        _log_line(staging, "status=error rc=1 reason=baseline_failed")
        return 1

    fetch_cfr_fn = fetch_cfr_fn or cfr.fetch_cfr_rows
    # Ask ECCC before downloading. Measured 2026-08-14: the CI list's Last-Modified was
    # 2025-12-04 and a live fetch was row-identical to our snapshot -- this weekly task had
    # been re-parsing an eight-month-old file every week since it was registered.
    cfr_gate = source_checks.gate(
        "CFR",
        probe_fn=probe_cfr_fn or cfr.probe_cfr_source,
        bundle_path=candidate_data / "cfr" / "pathways.json.gz",
        checked_at=run_id,
    )
    unchanged = bool(cfr_gate["skip"])
    if unchanged:
        cfr_leg = {"status": "unchanged", "decision": "kept_previous",
                   "reason": cfr_gate["reason"], "signal": cfr_gate["signal"],
                   "source_last_modified": cfr_gate["source_last_modified"]}
    else:
        cfr_leg = _fetch_and_normalize_cfr(fetch_cfr_fn)
        cfr_leg["probe"] = cfr_gate["reason"]
    leg_ok = cfr_leg["status"] == "ok"

    bands_report: dict = {"status": "skipped", "reason": "refresh not accepted"}
    if leg_ok:
        counts = _snapshot_status_counts(cfr_leg["certs_list"])
        counts["cfr.rows_total"] = cfr_leg["rows_total"]
        manifest = corpus_watch.load_manifest(_manifest_path())
        bands_report = corpus_watch.counts_report(counts, manifest)

    ok = leg_ok and bands_report.get("status") == "ok"

    change_events: list = []
    if ok:
        old_cfr = cfr.load_cfr_certs(path=candidate_data / "cfr" / "pathways.json.gz")
        change_events = refresh_quality.build_register_diff_events(
            run_id, "CFR", old_cfr, cfr_leg["certs_list"],
            cfr.diff_cfr_snapshot(old_cfr, cfr_leg["certs_list"]),
            detected_at=run_id,
        )
        cfr.write_snapshot(
            cfr_leg["certs_list"], path=candidate_data / "cfr" / "pathways.json.gz",
            fetched_at=run_id, source="cfr_local_refresh",
            source_last_modified=cfr_gate["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": "cfr-local-refresh",
        "baseline": baseline,
        "refresh": {
            "cfr": {**_leg_report(cfr_leg), "change_events": change_events} if cfr_leg["status"] == "ok" else cfr_leg,
            "bands": bands_report,
            "app_db": app_db_report,
        },
        "upload": {"status": "pending"},
    }

    rc = 0
    if unchanged:
        report["upload"] = {"status": "skipped-unchanged",
                            "reason": "ECCC reported the CI list unchanged; nothing to publish"}
        report["status"] = "unchanged"
    elif not leg_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="cfr_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)}}
            rc = 1

    finished = datetime.now(timezone.utc)
    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')}")
    return rc


if __name__ == "__main__":
    raise SystemExit(main())