#!/usr/bin/env python3 """Snapshot the tvviewers reference tables to local parquet. Why this exists --------------- `ReferenceData.from_mysql()` pulls ~6M rows across 12 frames from a remote server. That is fine once, but `run_pipeline()` calls it on EVERY run, so each upload paid several minutes of network before a single row was estimated. The reference data changes daily at most, so it is snapshotted here and read back from parquet in seconds. `worker/config.json` decides which path is used: "reference_mode": "fixtures" -> read the snapshot (fast, what the worker wants) "reference_mode": "mysql" -> go straight to the server (what this script does) Run nightly from cron: 0 4 * * * cd /var/www/html/crystal/crystal2 && python3 worker/build_cache.py The snapshot is written to a temporary directory and swapped into place only after every frame has been fetched, so a failure part-way through leaves the previous good snapshot untouched and the worker keeps running on it. """ from __future__ import annotations import json import shutil import sys import time from pathlib import Path BASE = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BASE)) # vendor/ carries pyarrow and sqlalchemy, both of which this script needs and # neither of which is installed system-wide. Without this the nightly cron job # dies on an ImportError at 4am and the snapshot silently stops refreshing -- # worker.py and fit_model.py already do this; this one had been missed. _vendor = BASE / "vendor" if _vendor.is_dir(): sys.path.insert(0, str(_vendor)) from engine.reference import ReferenceData # noqa: E402 def main() -> int: cfg_path = BASE / "worker" / "config.json" cfg = json.loads(cfg_path.read_text()) # Default to a directory of its own. `fixtures/` holds the small synthetic # set the offline tests run against; overwriting it with 6M live rows would # destroy the sandbox path. target = Path(cfg.get("fixtures_dir") or (BASE / "reference_cache")) staging = target.with_name(target.name + ".building") t0 = time.time() print(f"[cache] pulling reference from {cfg.get('ref_db_host')} …", flush=True) ref = ReferenceData.from_mysql(cfg) print(f"[cache] fetched {len(ref.frames)} frames in {time.time()-t0:.0f}s", flush=True) REPORT_FRAMES = ("ott_multipliers", "pan_channels", "pan_mapping") for name, df in sorted(ref.frames.items()): tag = " (report)" if name in REPORT_FRAMES else "" print(f"[cache] {name:38s} {len(df):>9,} rows{tag}", flush=True) missing = [n for n in REPORT_FRAMES if n not in ref.frames or ref.frames[n].empty] if missing: # Not fatal, but say so loudly: the OTT, OOH and pan sheets go to zero # without these and nothing else would signal it. print(f"[cache] WARNING: report frames empty or absent: {missing}", flush=True) print("[cache] the OTT, OOH and PAN sheets will read zero.", flush=True) if staging.exists(): shutil.rmtree(staging) ref.save_fixtures(staging) # The EPG adapter's own lookups. The event vocabulary is a GROUP BY over # global_sports, so fetching it per run cost ~85s of the wall clock before # any listing had been read. from engine.epg_adapter import EPG_FRAMES, fetch_epg_reference # noqa: E402 for name, df in zip(EPG_FRAMES, fetch_epg_reference(cfg)): df.to_parquet(staging / f"{name}.parquet", index=False) print(f"[cache] {name:38s} {len(df):>9,} rows", flush=True) # calibration.json lives in the snapshot directory, so the swap below would # delete it and the engine would silently fall back to assumed constants -- # no error, just quietly worse numbers. Carry it across, and stamp it stale: # it was fitted against the PREVIOUS snapshot, so worker/fit_model.py should # be re-run (the runbook pairs them for exactly this reason). old_cal = target / "calibration.json" if old_cal.exists(): cal = json.loads(old_cal.read_text()) cal["stale"] = True cal["stale_reason"] = ("fitted against the previous snapshot; " "re-run worker/fit_model.py") (staging / "calibration.json").write_text(json.dumps(cal, indent=1)) print("[cache] carried calibration.json across, marked stale", flush=True) else: print("[cache] no calibration.json to carry — run worker/fit_model.py", flush=True) # Swap only once the new snapshot is complete on disk. previous = target.with_name(target.name + ".previous") if previous.exists(): shutil.rmtree(previous) if target.exists(): target.rename(previous) staging.rename(target) if previous.exists(): shutil.rmtree(previous) # The worker runs as www-data and writes evidence_index.pkl INTO this # directory. The staging directory was created by whoever ran this script, # so after the swap it can easily be owned by a human with no group write # -- and then the worker silently fails to cache the index (the write is # wrapped) and pays a 14s rebuild on every single run. Force the group and # the setgid bit so it does not matter who runs the nightly job. try: shutil.chown(target, group="www-data") target.chmod(0o2775) for p in target.iterdir(): p.chmod(0o664) except (PermissionError, LookupError) as e: print(f"[cache] WARNING: could not set group www-data on {target}: {e}", flush=True) print("[cache] the worker may not be able to cache its evidence index.", flush=True) # The evidence index is derived from this snapshot, so it is stale the # moment the snapshot changes. Drop it here rather than leaving a rebuild # step for someone to remember: a stale index silently serves the previous # night's medians, which is the worst kind of wrong. stale = target / "evidence_index.pkl" if stale.exists(): stale.unlink() print("[cache] dropped evidence_index.pkl — it rebuilds on the next run", flush=True) size = sum(p.stat().st_size for p in target.glob("*.parquet")) / 1048576 print(f"[cache] wrote {target} — {size:.1f} MB in {time.time()-t0:.0f}s total", flush=True) print("[cache] set worker/config.json \"reference_mode\": \"fixtures\" to use it") return 0 if __name__ == "__main__": raise SystemExit(main())