from __future__ import annotations

import time

from fastapi import APIRouter, Depends, Request
from findata_core.config import get_settings
from findata_ingest.reader import dataset_views, run_sql
from pydantic import BaseModel, Field

from findata_api import auth as A
from findata_api import metrics as M
from findata_api.errors import ApiError
from findata_api.ratelimit import record_usage

router = APIRouter(prefix="/query", tags=["query"])


def _used(sql: str, views: dict) -> list[str]:
    from findata_ingest.reader import _referenced_views
    return _referenced_views(sql, views)


class QueryIn(BaseModel):
    sql: str = Field(..., max_length=20000)
    max_rows: int = Field(1000, ge=1, le=100_000)


@router.post("")
def query(body: QueryIn, request: Request, p: A.Principal = Depends(A.require("query"))):
    s = get_settings()
    views = dataset_views()
    t0 = time.monotonic()
    try:
        tbl, meta = run_sql(body.sql, min(body.max_rows, s.query_max_rows), s.query_timeout_seconds, views)
    except ValueError as e:
        raise ApiError(400, "query_rejected", str(e)) from None
    except TimeoutError as e:
        raise ApiError(408, "query_timeout", str(e)) from None
    except Exception as e:
        raise ApiError(400, "query_error", str(e)[:500]) from None
    rows = tbl.to_pylist()
    for r in rows:
        for k, v in r.items():
            if hasattr(v, "isoformat"):
                r[k] = v.isoformat()
    M.ROWS.labels("query").inc(len(rows))
    record_usage(p.label, rows=len(rows))
    A.audit(request, "query.sql", None, rows=len(rows), ms=int((time.monotonic() - t0) * 1000), sql=body.sql[:500])
    return {"columns": [{"name": f.name, "type": str(f.type)} for f in tbl.schema], "rows": rows,
            "meta": {**meta, "rows": len(rows), "elapsed_ms": int((time.monotonic() - t0) * 1000),
                     "views_used": _used(body.sql, views), "engine": "duckdb (read-only, sandboxed, 1 statement, hive-partition pruning)"}}


@router.get("/views")
def views(p: A.Principal = Depends(A.require("query"))):
    v = dataset_views()
    return {"views": [{"name": k, "kind": spec.get("kind"), "layouts": spec.get("layouts"), "root": spec.get("root")} for k, spec in sorted(v.items())],
            "note": "Views are Hive-partitioned directory scans: filter on the partition columns (symbol, year, session_date; underlying, period for options) "
                    "to prune files — e.g. WHERE symbol='AAPL' AND year=2024. Whole-dataset scans over millions of files time out (30 s).",
            "examples": ["SELECT session_date, min(low) lo, max(high) hi, sum(volume) vol FROM market_bars_stock_1m_raw WHERE symbol='AAPL' AND year=2024 GROUP BY 1 ORDER BY 1",
                         "SELECT trade_date, strike, call_put, bid, ask, open_interest, delta FROM options_eod WHERE underlying='SPY' AND period='2024q1' AND expiry=DATE '2024-03-28' ORDER BY trade_date, strike LIMIT 100",
                         "SELECT symbol, count(*) bars FROM market_bars_futures_1m_continuous_ratio WHERE symbol IN ('ES','NQ') GROUP BY 1"]}