refresh.py

"""Refresh the report end to end: pull, publish, screen, execute, verify.

    uv run python refresh.py                 pull everything, rebuild, verify
    uv run python refresh.py --skip-pulls    rebuild from the snapshots on disk
    uv run python refresh.py --months 24     pull a 24 month sales window instead of 36

Pulls land in data/raw/ with real mailing addresses and are never committed.
publish_snapshot.py redacts and compresses them into data/, which is what the repository
ships and what the notebook reads. The notebook selects the newest snapshot at or before
its ASOF setting, so a bare run always reports current data and a point-in-time run
reports whatever predates the date it was given.
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from datetime import date
from pathlib import Path

STAMP = date.today().strftime("%Y%m%d")
KERNEL_NAME = "elko-refresh"


def run(label, cmd, env=None):
    print(f"\n=== {label} ===", flush=True)
    result = subprocess.run([sys.executable, *cmd],
                            env={**os.environ, **env} if env else None)
    if result.returncode != 0:
        raise SystemExit(f"{label} failed with exit code {result.returncode}")


def write_kernelspec(root):
    """Write a kernelspec under `root` that launches this exact interpreter.

    nbconvert resolves --ExecutePreprocessor.kernel_name through the installed kernelspecs,
    and the stock python3 spec launches a bare `python` looked up on PATH at kernel start.
    That is not necessarily the interpreter running this script: on a machine with a system
    Python ahead of the project venv it binds the system one, and the report is then
    computed with whatever libraries that interpreter happens to carry rather than the
    locked ones. Synthesizing the spec here names sys.executable outright, so a refresh
    always computes the report with the locked dependencies and still needs no user-level
    kernel registration. `root` is passed to the child as JUPYTER_PATH.
    """
    spec = Path(root) / "kernels" / KERNEL_NAME
    spec.mkdir(parents=True, exist_ok=True)
    (spec / "kernel.json").write_text(json.dumps({
        "argv": [sys.executable, "-m", "ipykernel_launcher", "-f", "{connection_file}"],
        "display_name": f"elko-data refresh ({Path(sys.executable).parent.parent.name})",
        "language": "python",
    }), encoding="utf-8")
    return spec


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--skip-pulls", action="store_true",
                    help="rebuild from snapshots already on disk, no network")
    ap.add_argument("--months", type=int, default=36,
                    help="length of the rolling sales window to pull")
    ap.add_argument("--timeout", type=int, default=1800,
                    help="per-cell timeout for notebook execution, seconds")
    args = ap.parse_args()

    if not args.skip_pulls:
        run("sales window", ["pull_sales.py", "--months", str(args.months)])
        run("parcel universe", ["pull_universe.py",
                                f"data/raw/universe_sales_40_79_{STAMP}.json"])
        run("delinquent tax accounts",
            ["pull_tax.py", f"data/raw/tax_delinquent_{STAMP}.json"])
        run("FRED series", ["pull_fred.py"])

    # Publishing before executing keeps the committed snapshot and the committed notebook
    # in step: the notebook only ever reads what a clone will also have.
    if any(Path("data/raw").glob("*.json")):
        run("publish snapshot", ["publish_snapshot.py"])
        run("distress screen", ["screen.py"])
    else:
        print("\n=== publish snapshot, distress screen ===")
        print("skipped: no raw pulls in data/raw/. The committed snapshots in data/ are "
              "what the notebook reads, so this is the expected state in a fresh clone. "
              "Run without --skip-pulls to refresh them.", flush=True)
    # The kernel is synthesized rather than named from the installed set, so the notebook
    # runs on this interpreter and not on whichever python PATH resolves first.
    with tempfile.TemporaryDirectory() as kernel_root:
        write_kernelspec(kernel_root)
        run("notebook", ["-m", "jupyter", "nbconvert", "--to", "notebook", "--execute",
                         "--inplace", f"--ExecutePreprocessor.timeout={args.timeout}",
                         f"--ExecutePreprocessor.kernel_name={KERNEL_NAME}",
                         "elko_40ac_analysis.ipynb"],
            env={"JUPYTER_PATH": str(kernel_root)})
    run("gate", ["verify_outputs.py"])
    print("\nrefresh complete")


if __name__ == "__main__":
    main()