"""Warm the per-parcel value-history cache in bulk.

Normally the cache fills itself: askapi fetches a parcel's history the first
time someone opens it. This is for the case where history is wanted for *every*
parcel at once — e.g. to colour the map by 6-year growth, which needs the whole
county in the table, not just the parcels that have been clicked.

That means one request per parcel against the county Auditor's site (~99k pages,
~24 KB each). Deliberately slow: a small worker pool with a pause between
requests, resumable, and safe to stop with Ctrl-C and restart — anything already
fetched is skipped.

  python backfill_value_history.py --limit 500          # a taste
  python backfill_value_history.py --workers 4          # the lot, politely
  python backfill_value_history.py --where "sub='WHITETAIL MEADOWS'"
"""
from __future__ import annotations

import argparse
import queue
import sys
import threading
import time
from pathlib import Path

import psycopg2

sys.path.insert(0, "/home/m3ac/genoa-entwuerfe.com/askapi")
import value_history as vh   # noqa: E402

DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com")


def _dsn() -> str:
    for line in (DATA_DIR / ".env.gis").read_text().splitlines():
        if line.startswith("GISPASS="):
            return f"postgresql://gisapp:{line.split('=', 1)[1].strip()}@127.0.0.1:5432/westerville"
    raise SystemExit("GISPASS missing")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--workers", type=int, default=3, help="concurrent requests (default 3)")
    ap.add_argument("--delay", type=float, default=0.35, help="seconds each worker waits between parcels")
    ap.add_argument("--limit", type=int, default=0, help="stop after N parcels (0 = all)")
    ap.add_argument("--where", default="", help="extra SQL filter on parcels, e.g. \"sub='...'\"")
    ap.add_argument("--refetch", action="store_true", help="also re-fetch parcels already cached")
    args = ap.parse_args()

    dsn = _dsn()
    sql = ["SELECT p.id FROM parcels p"]
    if not args.refetch:
        sql.append("LEFT JOIN parcel_value_history_status s ON s.parcel_no = p.id")
    sql.append("WHERE length(regexp_replace(p.id, '\\D', '', 'g')) = 14")
    if not args.refetch:
        sql.append("AND s.parcel_no IS NULL")
    if args.where:
        sql.append(f"AND ({args.where})")
    sql.append("ORDER BY p.id")
    if args.limit:
        sql.append(f"LIMIT {args.limit}")

    conn = psycopg2.connect(dsn)
    with conn, conn.cursor() as cur:
        cur.execute(vh.DDL)
        cur.execute(" ".join(sql))
        todo = [r[0] for r in cur.fetchall()]
    print(f"[plan] {len(todo)} parcels to fetch · {args.workers} workers · {args.delay}s apart")
    if not todo:
        conn.close()
        return

    q: queue.Queue = queue.Queue()
    for pid in todo:
        q.put(pid)
    lock = threading.Lock()
    done = {"n": 0, "rows": 0, "err": 0}
    start = time.time()

    def worker():
        wconn = psycopg2.connect(dsn)
        while True:
            try:
                pid = q.get_nowait()
            except queue.Empty:
                break
            err, rows = None, []
            try:
                rows = vh.fetch_history(pid)
            except Exception as e:                    # noqa: BLE001 - recorded per parcel
                err = str(e)[:200]
            try:
                with wconn, wconn.cursor() as cur:
                    vh.store(cur, pid, rows, error=err)
            except Exception as e:                    # noqa: BLE001
                err = f"store: {e}"[:200]
            with lock:
                done["n"] += 1
                done["rows"] += len(rows)
                done["err"] += 1 if err else 0
                n = done["n"]
            if n % 100 == 0 or n == len(todo):
                rate = n / max(time.time() - start, 1e-9)
                left = (len(todo) - n) / rate if rate else 0
                print(f"  {n}/{len(todo)} · {done['rows']} rows · {done['err']} errors "
                      f"· {rate:.1f}/s · ~{left/60:.0f} min left", flush=True)
            time.sleep(args.delay)
        wconn.close()

    threads = [threading.Thread(target=worker, daemon=True) for _ in range(args.workers)]
    for t in threads:
        t.start()
    try:
        for t in threads:
            t.join()
    except KeyboardInterrupt:
        print("\n[stop] interrupted — rerun to resume where it left off")
    conn.close()
    print(f"[done] {done['n']} parcels · {done['rows']} history rows · {done['err']} errors "
          f"· {(time.time() - start)/60:.1f} min")


if __name__ == "__main__":
    main()