"""Derive 5-minute (or other) bars from validated 1-minute bars, per session_date (delta layout, supersede)."""
from __future__ import annotations

from datetime import date
from typing import Any

import pyarrow as pa
from findata_core.db import connection, transaction
from findata_quality import aggregate_bars
from findata_schema import Interval

from findata_ingest.lake import Lake, live_files
from findata_ingest.reader import read_bars
from findata_ingest.tasks.common import live_delta_dates, live_delta_paths_for_dates, upsert_coverage


def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
    p = job.get("params") or {}
    src = p["source_dataset"]
    dst = job.get("dataset_id") or src.replace(".1m.", ".5m.")
    target = Interval(p.get("target_interval", "5m"))
    dates = [date.fromisoformat(d) for d in p.get("dates", [])]
    symbols = p.get("symbols")
    run_id = job.get("run_id")
    with connection() as conn:
        if not dates:
            # all live source dates for requested symbols
            fl = live_files(conn, src, symbols)
            ds = set()
            for f in fl:
                d = f["date_min"]
                while d <= f["date_max"]:
                    ds.add(d)
                    d = d.fromordinal(d.toordinal() + 1)
            dates = sorted(ds)
        all_syms = symbols or [r["symbol"] for r in conn.execute("SELECT DISTINCT symbol FROM coverage WHERE dataset_id=%s AND session_date = ANY(%s)", (src, dates)).fetchall()]
    written, rows = [], 0
    tables = {}
    if symbols:
        # backfill path: aggregate per symbol over all its sessions and write base files per (symbol, year)
        import pyarrow.compute as pc
        retire: list[str] = []
        for s in sorted(set(x.upper() for x in symbols)):
            t, _ = read_bars(src, [s], None, None, lake=lake)
            if t.num_rows == 0:
                continue
            agg = aggregate_bars(t, target, Interval.m1)
            years = pc.year(agg.column("session_date"))
            with connection() as conn:
                old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s",
                                   (dst, s, s)).fetchall()
                delta_dates = live_delta_dates(conn, dst)
            if delta_dates:
                agg = agg.filter(pc.invert(pc.is_in(agg.column("session_date"), value_set=pa.array(sorted(delta_dates), pa.date32()))))
                years = pc.year(agg.column("session_date"))
            for y in sorted(set(years.to_pylist())):
                ty = agg.filter(pc.equal(years, y))
                written.append(lake.write_table(dst, ty, "base", {"symbol": s, "year": str(y)}))
            retire.extend(o["path"] for o in old)
            rows += agg.num_rows
            with transaction() as conn:
                upsert_coverage(conn, dst, agg, target)
            del t, agg
            import gc
            gc.collect()
            pa.default_memory_pool().release_unused()
        if not written:
            return {"rows": 0, "symbols": len(symbols)}
        pub = lake.publish(dst, written, sorted(set(retire)), run_id, note=f"derive {target.value} (base) from {src} for {len(symbols)} symbols")
        log.info("derive.done", rows=rows, files=len(written), **pub)
        return {"rows": rows, "files": len(written), "symbols": len(symbols), **pub}
    for d in dates:
        t, _ = read_bars(src, all_syms, d, d, lake=lake)
        if t.num_rows == 0:
            continue
        agg = aggregate_bars(t, target, Interval.m1)
        tables[d] = agg
        rows += agg.num_rows
        written.append(lake.write_table(dst, agg, "delta", {"session_date": d.isoformat()}))
    if not written:
        return {"rows": 0, "dates": len(dates)}
    with connection() as conn:
        retire = live_delta_paths_for_dates(conn, dst, list(tables)) if not symbols else []
    pub = lake.publish(dst, written, retire, run_id, note=f"derive {target.value} from {src}")
    with transaction() as conn:
        for t in tables.values():
            upsert_coverage(conn, dst, t, target)
    log.info("derive.done", rows=rows, files=len(written), **pub)
    return {"rows": rows, "files": len(written), "dates": len(tables), **pub}