"""Periodic quality/coverage report and garbage collection."""
from __future__ import annotations

import json
from typing import Any

from findata_core.db import connection, transaction

from findata_ingest.lake import Lake


def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
    dataset_id = job.get("dataset_id") or "market_bars.stock.1m.raw"
    with connection() as conn:
        tot = conn.execute("SELECT count(*) n, sum(bar_count) bars, count(DISTINCT symbol) syms, min(session_date) d0, max(session_date) d1 FROM coverage WHERE dataset_id=%s", (dataset_id,)).fetchone()
        comp = conn.execute("SELECT avg(LEAST(1.0, bar_count::float/NULLIF(expected_bars,0))) AS c FROM coverage WHERE dataset_id=%s AND expected_bars>0", (dataset_id,)).fetchone()
        anomalies = conn.execute("SELECT code, severity, count(*) n FROM quality_anomalies WHERE dataset_id=%s AND created_at > now() - interval '30 days' GROUP BY 1,2 ORDER BY n DESC", (dataset_id,)).fetchall()
    summary = {"dataset_id": dataset_id, "symbol_sessions": tot["n"], "bars": int(tot["bars"] or 0), "symbols": tot["syms"],
               "first_session": str(tot["d0"]), "last_session": str(tot["d1"]), "mean_regular_hours_completeness": float(comp["c"] or 0) if comp and comp["c"] is not None else None,
               "anomalies_30d": [dict(a) for a in anomalies]}
    with transaction() as conn:
        conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",
                     (dataset_id, job.get("run_id"), summary["bars"], sum(a["n"] for a in anomalies), json.dumps(summary, default=str)))
    return summary


def gc(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
    n = lake.gc_retired(7)
    orphans = lake.gc_orphans(2)
    removed = 0
    with transaction() as conn:
        rows = conn.execute("UPDATE exports SET status='expired' WHERE status='ready' AND expires_at < now() RETURNING path").fetchall()
        for r in rows:
            if r["path"]:
                p = lake.abs(r["path"])
                if p.exists():
                    p.unlink()
                    removed += 1
        conn.execute("DELETE FROM sessions WHERE expires_at < now() - interval '7 days'")
    return {"retired_files_deleted": n, "orphan_files_deleted": orphans, "exports_expired": len(rows), "export_files_removed": removed}