"""One-shot backfill of the parquet climate cache into the TimescaleDB tables.

Run ONCE during the cutover, after ``alembic upgrade head`` has created the
``climate_history`` / ``climate_recent`` / ``climate_sync`` tables. It loads every
cached cell's parquet record into Postgres so a freshly-cut-over server serves
history/recent from the database immediately instead of refetching decades of
archive (which the upstream rate limit would throttle hard).

Idempotent and resumable: a cell whose stored sync timestamp already matches its
parquet mtime is skipped, and history rows upsert (``ON CONFLICT``), so re-running
after an interruption only does the missing work.

Key detail: rows are read RAW (``pl.read_parquet`` + ``_normalize_read``), NOT via
``climate.load_cached_*`` — the loaders apply ``_derive_metrics``, which replaces
raw relative humidity with absolute humidity. The DB must store the raw record the
live write path persists, so absolute humidity / wet-bulb stay read-time
derivations. The per-cell ``synced_at`` is set to the parquet FILE MTIME (not now),
so ``recent_stamp`` — hence every derived-payload validity token — is unchanged
across the cutover.

Usage:
    THERMOGRAPH_DATABASE_URL=postgresql+psycopg://thermograph:PW@127.0.0.1:5432/thermograph \\
      python migrate_cache_to_pg.py            # reads data/cache/*.parquet
"""
import os
import sys

import polars as pl

from data import climate
from data import climate_store
from data import grid


def _is_schema_complete(df: pl.DataFrame) -> bool:
    """The same NEW_COLS guard climate._read_history_cache applies: a record missing
    any post-original column predates the wind/humidity features and is skipped so
    the server refetches it lazily (with the current schema) on first touch."""
    return all(c in df.columns for c in climate.NEW_COLS)


def migrate() -> int:
    if not climate_store.is_postgres():
        sys.exit("THERMOGRAPH_DATABASE_URL must be a postgresql URL (the migration target)")
    if not os.path.isdir(climate.CACHE_DIR):
        print("No parquet cache directory yet — nothing to migrate.")
        return 0

    hist = recent = current = skipped = 0
    for fname in sorted(os.listdir(climate.CACHE_DIR)):
        if not fname.endswith(".parquet"):
            continue
        path = os.path.join(climate.CACHE_DIR, fname)
        is_recent = fname.endswith("_rf.parquet")
        cell_id = fname[:-len("_rf.parquet")] if is_recent else fname[:-len(".parquet")]
        try:
            grid.from_id(cell_id)   # validate the id is a real cell (skip stray files)
        except ValueError:
            print(f"  {fname}: unrecognized cache filename — skipped")
            skipped += 1
            continue

        mtime = os.path.getmtime(path)
        synced = (climate_store.recent_synced_at(cell_id) if is_recent
                  else climate_store.history_synced_at(cell_id))
        if synced >= mtime:
            current += 1
            continue

        try:
            df = climate._normalize_read(pl.read_parquet(path))
        except Exception as e:  # noqa: BLE001 - a truncated/corrupt file is a skip, not a crash
            print(f"  {fname}: unreadable ({e}) — skipped")
            skipped += 1
            continue

        if is_recent:
            climate_store.write_recent(cell_id, df, mtime)
            recent += 1
            print(f"  {cell_id}: recent bundle ({df.height} rows)")
        else:
            if not _is_schema_complete(df):
                print(f"  {cell_id}: pre-schema history — skipped (refetches lazily)")
                skipped += 1
                continue
            climate_store.write_history(cell_id, df, mtime)
            hist += 1
            print(f"  {cell_id}: history ({df.height} rows)")

    print(f"{hist} history + {recent} recent loaded, {current} already current, "
          f"{skipped} skipped.")
    return 0


if __name__ == "__main__":
    sys.exit(migrate())