from __future__ import annotations

import json
from datetime import UTC, date, datetime
from typing import Any

import pyarrow as pa
from findata_quality.validators import session_coverage
from findata_schema import Interval

DATASET_ASSET = {"stock": "market_bars.stock.1m.raw", "etf": "market_bars.etf.1m.raw", "index": "market_bars.index.1m.raw",
                 "fx": "market_bars.fx.1m.raw", "crypto": "market_bars.crypto.1m.raw", "futures": "market_bars.futures.1m.continuous"}


def dataset_for_params(params: dict[str, Any]) -> str:
    from findata_market_firstrate.catalog import dataset_id_for
    return dataset_id_for(params.get("type", "stock"), params.get("timeframe", "1min"), params.get("adjustment"),
                          contracts=bool(params.get("contracts")))


def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str, Any], rel_path: str, sha256: str,
                       nbytes: int, file_count: int | None, vendor_last_update: date | None, run_id: str | None) -> tuple[str, bool]:
    """Insert raw archive row; returns (id, is_new). Same sha256 → already known (idempotent re-fetch)."""
    r = conn.execute("SELECT id, status FROM raw_archives WHERE source_id=%s AND sha256=%s", (source_id, sha256)).fetchone()
    if r:
        return str(r["id"]), False
    r = conn.execute(
        "INSERT INTO raw_archives(source_id, dataset_id, request, path, sha256, bytes, file_count, vendor_last_update, ingest_run_id, status) "
        "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'stored') RETURNING id",
        (source_id, dataset_id, json.dumps(request, default=str), rel_path, sha256, nbytes, file_count, vendor_last_update, run_id)).fetchone()
    return str(r["id"]), True


COVERAGE_DATASETS = {"market_bars.stock.1m.raw", "market_bars.etf.1m.raw", "market_bars.index.1m.raw",
                     "market_bars.stock.5m.raw", "market_bars.etf.5m.raw", "market_bars.stock.1d.raw", "market_bars.etf.1d.raw"}


def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:
    if dataset_id not in COVERAGE_DATASETS:
        return 0  # adjusted/coarse variants share the raw datasets' sessions; keeping per-session rows for all 64 datasets cost 130 GB
    if table.num_rows and table.column("asset_class")[0].as_py() in ("fx", "crypto", "futures"):
        return 0  # no exchange-session expectation for 24h markets
    rows = session_coverage(table, interval)
    for r in rows:
        conn.execute(
            "INSERT INTO coverage(dataset_id, symbol, session_date, bar_count, first_ts, last_ts, expected_bars, flags) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (dataset_id, symbol, session_date) DO UPDATE SET bar_count=EXCLUDED.bar_count, "
            "first_ts=EXCLUDED.first_ts, last_ts=EXCLUDED.last_ts, expected_bars=EXCLUDED.expected_bars, flags=EXCLUDED.flags, updated_at=now()",
            (dataset_id, r["symbol"], r["session_date"], r["bar_count"], r["first_ts"], r["last_ts"], r["expected_bars"], r["flags"]))
    return len(rows)


def record_anomalies(conn, dataset_id: str, anomalies: list[dict[str, Any]], run_id: str | None, limit: int = 5000) -> int:
    n = 0
    for a in anomalies[:limit]:
        conn.execute("INSERT INTO quality_anomalies(dataset_id, symbol, session_date, code, severity, detail, ingest_run_id) VALUES (%s,%s,%s,%s,%s,%s,%s)",
                     (dataset_id, a.get("symbol"), a.get("session_date"), a["code"], a["severity"], json.dumps(a.get("detail", {}), default=str), run_id))
        n += 1
    return n


def record_quality_report(conn, dataset_id: str, run_id: str | None, rows: int, anomalies: list[dict[str, Any]]) -> None:
    summary: dict[str, int] = {}
    for a in anomalies:
        summary[a["code"]] = summary.get(a["code"], 0) + int(a.get("detail", {}).get("rows", 1))
    conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",
                 (dataset_id, run_id, rows, len(anomalies), json.dumps(summary)))


def live_delta_paths_for_dates(conn, dataset_id: str, dates: list[date]) -> list[str]:
    if not dates:
        return []
    rows = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL AND date_min = ANY(%s)",
                        (dataset_id, dates)).fetchall()
    return [r["path"] for r in rows]


def live_delta_dates(conn, dataset_id: str) -> set[date]:
    rows = conn.execute("SELECT DISTINCT date_min FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL", (dataset_id,)).fetchall()
    return {r["date_min"] for r in rows}


def now() -> datetime:
    return datetime.now(UTC)