"""Async exports: bounded request validated synchronously, executed in a background thread, Parquet/CSV file served
from lake/exports with expiration. Dataset snapshot (version) is pinned at submission time."""
from __future__ import annotations

import json
import traceback
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, date, datetime, timedelta
from typing import Literal

import pyarrow.parquet as pq
from fastapi import APIRouter, Depends, Request
from fastapi.responses import FileResponse
from findata_core.config import get_settings
from findata_core.db import connection, transaction
from findata_ingest.lake import Lake
from findata_ingest.reader import apply_adjustment, read_bars
from pydantic import BaseModel, Field

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

router = APIRouter(prefix="/exports", tags=["exports"])
_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="export")


class ExportIn(BaseModel):
    dataset_id: str = "market_bars.stock.1m.raw"
    symbols: list[str] = Field(..., min_length=1, max_length=500)
    start: date
    end: date
    adjustment: Literal["raw", "split", "split_div"] = "raw"
    format: Literal["parquet", "csv"] = "parquet"


def _run(export_id: str, req: dict, principal: str) -> None:
    lake = Lake()
    try:
        with transaction() as conn:
            conn.execute("UPDATE exports SET status='running' WHERE id=%s", (export_id,))
        syms = [s.upper() for s in req["symbols"]]
        table, meta = read_bars(req["dataset_id"], syms, date.fromisoformat(req["start"]), date.fromisoformat(req["end"]), version=req.get("version"))
        if req["adjustment"] != "raw":
            table, _ = apply_adjustment(table, req["adjustment"])
        ext = "parquet" if req["format"] == "parquet" else "csv"
        rel = f"exports/{export_id}.{ext}"
        out = lake.abs(rel)
        if ext == "parquet":
            pq.write_table(table, out, compression="zstd")
        else:
            import pyarrow.csv as pcsv
            pcsv.write_csv(table.drop_columns(["quality_flags"]), out)
        ttl = get_settings().export_ttl_hours
        with transaction() as conn:
            conn.execute("UPDATE exports SET status='ready', path=%s, bytes=%s, row_count=%s, finished_at=now(), expires_at=%s, snapshot=%s WHERE id=%s",
                         (rel, out.stat().st_size, table.num_rows, datetime.now(UTC) + timedelta(hours=ttl),
                          json.dumps({"dataset_id": req["dataset_id"], "version": meta.get("version"), "files_scanned": meta.get("files_scanned")}), export_id))
        record_usage(principal, rows=table.num_rows, bytes_exported=out.stat().st_size)
    except Exception as e:
        with transaction() as conn:
            conn.execute("UPDATE exports SET status='failed', error=%s, finished_at=now() WHERE id=%s", (f"{e}\n{traceback.format_exc()[-800:]}"[:2000], export_id))


@router.post("", status_code=202)
def create(body: ExportIn, request: Request, p: A.Principal = Depends(A.require("export"))):
    if body.end < body.start:
        raise ApiError(400, "bad_range", "end must be >= start")
    span = (body.end - body.start).days
    if span > 366 * 12:
        raise ApiError(413, "range_too_large", "exports are limited to 12 years per request")
    with connection() as conn:
        d = conn.execute("SELECT id, status, current_version FROM datasets WHERE id=%s", (body.dataset_id,)).fetchone()
        if not d:
            raise ApiError(404, "not_found", "unknown dataset")
        if d["status"] != "active":
            raise ApiError(409, "dataset_not_active", f"dataset status is {d['status']}")
        n = conn.execute("SELECT count(*) FROM exports WHERE principal=%s AND status IN ('queued','running')", (p.label,)).fetchone()["count"]
        if n >= 3:
            raise ApiError(429, "too_many_exports", "at most 3 concurrent exports per principal")
    req = body.model_dump(mode="json")
    req["version"] = d["current_version"]
    with transaction() as conn:
        r = conn.execute("INSERT INTO exports(principal, request, format) VALUES (%s,%s,%s) RETURNING id", (p.label, json.dumps(req), body.format)).fetchone()
    eid = str(r["id"])
    A.audit(request, "export.create", eid, dataset=body.dataset_id, symbols=len(body.symbols), start=str(body.start), end=str(body.end))
    _pool.submit(_run, eid, req, p.label)
    return {"id": eid, "status": "queued", "poll": f"/api/v1/exports/{eid}"}


@router.get("")
def list_exports(p: A.Principal = Depends(A.require("export"))):
    with connection() as conn:
        rows = conn.execute("SELECT id, status, request, format, bytes, row_count, created_at, finished_at, expires_at, error, snapshot FROM exports WHERE principal=%s OR %s ORDER BY created_at DESC LIMIT 100",
                            (p.label, p.has("admin"))).fetchall()
    return {"data": rows}


@router.get("/{export_id}")
def get(export_id: str, p: A.Principal = Depends(A.require("export"))):
    with connection() as conn:
        r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()
    if not r or (r["principal"] != p.label and not p.has("admin")):
        raise ApiError(404, "not_found", "unknown export")
    r = dict(r)
    if r["status"] == "ready":
        r["download"] = f"/api/v1/exports/{export_id}/download"
    return r


@router.get("/{export_id}/download")
def download(export_id: str, request: Request, p: A.Principal = Depends(A.require("export"))):
    with connection() as conn:
        r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()
    if not r or (r["principal"] != p.label and not p.has("admin")):
        raise ApiError(404, "not_found", "unknown export")
    if r["status"] != "ready":
        raise ApiError(409, "not_ready", f"export status is {r['status']}")
    if r["expires_at"] and r["expires_at"] < datetime.now(UTC):
        raise ApiError(410, "expired", "export expired")
    path = Lake().abs(r["path"])
    A.audit(request, "export.download", export_id, bytes=r["bytes"])
    return FileResponse(path, filename=f"findata-{export_id}.{r['format']}", media_type="application/octet-stream")