Spaces:
Running
Running
| """ | |
| Command line for the intel store (run from backend/ with INTEL_DB_PATH set): | |
| py -m intel.cli init create tables, seed sources/streams | |
| py -m intel.cli locations build the state/LGA gazetteer | |
| py -m intel.cli calendar holidays + seasons, 2000 → +10 years | |
| py -m intel.cli places load the Overture extract (run overture_extract first) | |
| py -m intel.cli weather-history [--budget N] [--start YYYY-MM-DD] resumable backfill | |
| py -m intel.cli weather-forecast 16-day forecast at state centres | |
| py -m intel.cli status Layer 1 report (exit 1 if unhealthy) | |
| py -m intel.cli export [--out DIR] CSVs + manifest for the Colab notebooks | |
| py -m intel.cli resolve "TEXT" which location does this text mention? | |
| py -m intel.cli run-daily forecast + history top-up + status, one run id | |
| `run-daily` is what a scheduler will call once one exists (none yet — see the | |
| design doc's "Conflicts"). | |
| """ | |
| import argparse | |
| import json | |
| import sys | |
| import uuid | |
| from datetime import date | |
| from pathlib import Path | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| from intel import db # noqa: E402 (env must be loaded first) | |
| EXPORT_DIR = Path(__file__).resolve().parent / "data" / "export" | |
| def _print(obj) -> None: | |
| print(json.dumps(obj, indent=2, default=str)) | |
| def main(argv: list[str] | None = None) -> int: | |
| p = argparse.ArgumentParser(prog="intel") | |
| sub = p.add_subparsers(dest="cmd", required=True) | |
| sub.add_parser("init") | |
| sub.add_parser("locations") | |
| sub.add_parser("calendar") | |
| sub.add_parser("places") | |
| wh = sub.add_parser("weather-history") | |
| wh.add_argument("--budget", type=float, default=None) | |
| wh.add_argument("--start", type=date.fromisoformat, default=None) | |
| wh.add_argument("--estimate", action="store_true", help="only print the weighted calls still needed") | |
| sub.add_parser("weather-forecast") | |
| sub.add_parser("status") | |
| ex = sub.add_parser("export") | |
| ex.add_argument("--out", type=Path, default=EXPORT_DIR) | |
| rs = sub.add_parser("resolve") | |
| rs.add_argument("text") | |
| sub.add_parser("run-daily") | |
| args = p.parse_args(argv) | |
| con = db.connect() | |
| run_id = f"{args.cmd}-{uuid.uuid4().hex[:8]}" | |
| try: | |
| if args.cmd == "init": | |
| db.init(con) | |
| _print({"initialised": db.db_path()}) | |
| elif args.cmd == "locations": | |
| from intel.locations import build | |
| _print(build(con)) | |
| elif args.cmd == "calendar": | |
| from intel.ingest.calendar import ingest | |
| _print(ingest(con, run_id=run_id)) | |
| elif args.cmd == "places": | |
| from intel.ingest.places import ingest | |
| _print(ingest(con, run_id=run_id)) | |
| elif args.cmd == "weather-history": | |
| from intel.ingest import weather | |
| start = args.start or weather.DEFAULT_HISTORY_START | |
| if args.estimate: | |
| _print({"weighted_calls_needed": round(weather.estimate_backfill_calls(con, start=start), 1)}) | |
| else: | |
| _print(weather.backfill_history(con, start=start, budget=args.budget or weather.DEFAULT_RUN_BUDGET, run_id=run_id)) | |
| elif args.cmd == "weather-forecast": | |
| from intel.ingest.weather import fetch_forecast | |
| _print(fetch_forecast(con, run_id=run_id)) | |
| elif args.cmd == "status": | |
| from intel.freshness import report | |
| r = report(con) | |
| _print(r) | |
| return 0 if r["healthy"] else 1 | |
| elif args.cmd == "export": | |
| from intel.export import export | |
| _print(export(con, args.out)["rows"]) | |
| elif args.cmd == "resolve": | |
| from intel.locations import resolve | |
| _print({"text": args.text, "location_id": resolve(con, args.text)}) | |
| elif args.cmd == "run-daily": | |
| from intel.freshness import report | |
| from intel.ingest import weather | |
| out = {"run_id": run_id, "forecast": weather.fetch_forecast(con, run_id=run_id), | |
| "history": weather.backfill_history(con, run_id=run_id)} | |
| out["status"] = report(con) | |
| _print(out) | |
| return 0 if out["status"]["healthy"] else 1 | |
| finally: | |
| con.close() | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |