#!/usr/bin/env python3
# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Phase 6 — prospective evaluation set: sales present in the *live* Vrai Prix database but absent from the
2026-10-21 snapshot (registered after the snapshot, hence after the protocol freeze), joined to their roll unit.

Run only AFTER `av freeze`. Reads the live SQLite over SSH (read-only), never modifies it.
Output: data/raw/prospective_<date>.parquet (raw columns, same vocabulary as the snapshot) + a manifest.
"""
from __future__ import annotations

import json
import subprocess
import sys
from datetime import date
from pathlib import Path

import polars as pl

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from agentic_valuation.config import RAW, RESULTS  # noqa: E402

NODE = "M3U96a"
DB = "~/apps/vrai-prix/data/vraiprix.db"
SINCE = "2026-07-01"  # registration lag is 2-8 weeks; anything older is already in the snapshot


def ssh_query(sql: str) -> str:
    cmd = ["ssh", "-o", "ConnectTimeout=20", NODE, f"sqlite3 -json -readonly {DB} \"{sql}\""]
    return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout


def main() -> None:
    freeze = RESULTS / "freeze.json"
    if not freeze.exists():
        sys.exit("results/freeze.json not found: run `av freeze` before fetching the prospective set")
    snap = pl.read_parquet(RAW / "wp22_clean_transactions.parquet", columns=["tx_id"])
    live = pl.DataFrame(json.loads(ssh_query(f"select * from transactions where date >= '{SINCE}'") or "[]"))
    new = live.filter(~pl.col("id").is_in(snap["tx_id"].implode()))
    ids = [i for i in new["id_provinc"].drop_nulls().unique().to_list() if i]
    units = pl.DataFrame()
    if ids:
        chunks = [ids[i : i + 500] for i in range(0, len(ids), 500)]
        parts = []
        for ch in chunks:
            inl = ",".join(f"'{i}'" for i in ch)
            parts.append(pl.DataFrame(json.loads(ssh_query(f"select * from units where id_provinc in ({inl})") or "[]")))
        units = pl.concat([p for p in parts if p.height], how="diagonal_relaxed") if any(p.height for p in parts) else pl.DataFrame()
    stamp = date.today().isoformat()
    out = RAW / f"prospective_{stamp}.parquet"
    new.write_parquet(out)
    if units.height:
        units.write_parquet(RAW / f"prospective_units_{stamp}.parquet")
    man = {"fetched": stamp, "node": NODE, "since": SINCE, "live_rows_since": live.height, "new_rows": new.height, "new_date_min": str(new["date"].min()) if new.height else None, "new_date_max": str(new["date"].max()) if new.height else None, "units_rows": units.height, "freeze_hash": json.loads(freeze.read_text()).get("freeze_digest")}
    (RAW / f"prospective_{stamp}.manifest.json").write_text(json.dumps(man, indent=1))
    print(man)


if __name__ == "__main__":
    main()