"""Streamed exports (json / ndjson / csv) with bounded limits and keyset iteration — derived data only, never raw page content."""
from __future__ import annotations

import csv
import io
from collections.abc import AsyncIterator, Callable
from typing import Any

import orjson
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import StreamingResponse

from companyatlas.api import queries as q
from companyatlas.api import serializers as ser
from companyatlas.api.common import _default
from companyatlas.db import connection, fetch_all

ORDER = 20
router = APIRouter(prefix="/api/v1", tags=["exports"])
BATCH = 1000
MAX_EVENTS = 10_000
MAX_COMPANIES = 20_000
MAX_JOBS = 10_000
MEDIA = {"json": "application/json", "ndjson": "application/x-ndjson", "csv": "text/csv; charset=utf-8"}
EVENT_COLUMNS = ["id", "detected_at", "company_slug", "company_name", "company_domain", "country", "event_type", "event_subtype", "importance", "confidence",
                 "confidence_label", "title", "summary", "old_value", "new_value", "source_url", "surface", "origin", "status"]
COMPANY_COLUMNS = ["id", "slug", "display_name", "canonical_domain", "website", "country", "hq_city", "industry_primary", "industries", "public_company", "ticker",
                   "tier", "importance", "activity_score", "hiring_momentum_30d", "open_jobs", "ai_adoption", "sensors", "events", "last_event_at"]
JOB_COLUMNS = ["id", "company_slug", "title", "department", "location_text", "city", "country", "remote", "employment_type", "seniority", "url", "posted_at",
               "first_seen_at", "last_seen_at", "removed_at", "status", "is_ai"]


def _dumps(obj: Any) -> bytes:
    return orjson.dumps(obj, option=orjson.OPT_UTC_Z | orjson.OPT_NON_STR_KEYS, default=_default)


def _csv_row(values: list[Any]) -> bytes:
    buf = io.StringIO()
    csv.writer(buf, lineterminator="\n").writerow(["" if v is None else (v.isoformat() if hasattr(v, "isoformat") else v) for v in values])
    return buf.getvalue().encode("utf-8")


def _event_flat(ev: dict[str, Any]) -> list[Any]:
    c = ev["company"]
    return [ev["id"], ev["detected_at"], c["slug"], c["display_name"], c["canonical_domain"], c["country"], ev["event_type"], ev["event_subtype"], ev["importance"],
            ev["confidence"], ev["confidence_label"], ev["title"], ev["summary"], ev["old_value"], ev["new_value"], ev["source_url"], ev["surface"], ev["origin"],
            ev["status"]]


def _company_flat(c: dict[str, Any]) -> list[Any]:
    m, n = c["metrics"], c["counts"]
    return [c["id"], c["slug"], c["display_name"], c["canonical_domain"], c["website"], c["country"], c["hq_city"], c["industry_primary"], "|".join(c["industries"]),
            c["public_company"], c["ticker"], c["tier"], c["importance"], m.get("activity_score"), m.get("hiring_momentum_30d"), m.get("open_jobs"), m.get("ai_adoption"),
            n["sensors"], n["events"], c["last_event_at"]]


def _job_flat(j: dict[str, Any]) -> list[Any]:
    return [j["id"], j.get("company_slug"), j["title"], j["department"], j["location_text"], j["city"], j["country"], j["remote"], j["employment_type"], j["seniority"],
            j["url"], j["posted_at"], j["first_seen_at"], j["last_seen_at"], j["removed_at"], j["status"], j["is_ai"]]


async def _stream(fmt: str, rows: AsyncIterator[dict[str, Any]], columns: list[str], flat: Callable[[dict[str, Any]], list[Any]]) -> AsyncIterator[bytes]:
    if fmt == "csv":
        yield _csv_row(columns)
        async for r in rows:
            yield _csv_row(flat(r))
        return
    if fmt == "ndjson":
        async for r in rows:
            yield _dumps(r) + b"\n"
        return
    yield b"["
    first = True
    async for r in rows:
        yield (b"" if first else b",") + _dumps(r)
        first = False
    yield b"]"


def _response(fmt: str, name: str, body: AsyncIterator[bytes]) -> StreamingResponse:
    if fmt not in MEDIA:
        raise HTTPException(status_code=404, detail="unsupported format (json, ndjson, csv)")
    return StreamingResponse(body, media_type=MEDIA[fmt], headers={"content-disposition": f'attachment; filename="company-atlas-{name}.{fmt}"',
                                                                  "cache-control": "public, max-age=300", "x-accel-buffering": "no"})


async def _iter_events(where: list[str], params: dict[str, Any], limit: int) -> AsyncIterator[dict[str, Any]]:
    sent = 0
    cursor: tuple[Any, str] | None = None
    async with connection() as conn:
        while sent < limit:
            w = list(where)
            p = dict(params)
            if cursor:
                w.append("(e.detected_at, e.id) < (:cur_at, :cur_id)")
                p.update(cur_at=cursor[0], cur_id=cursor[1])
            rows = await q.fetch_events(conn, w, p, sort="recent", limit=min(BATCH, limit - sent))
            if not rows:
                return
            for r in rows:
                yield ser.event(r)
            sent += len(rows)
            cursor = (rows[-1]["detected_at"], rows[-1]["id"])


@router.get("/export/events.{fmt}", summary="Export events (≤ 10 000 rows)")
async def export_events(fmt: str, since: str | None = None, until: str | None = None, event_type: str | None = None, country: str | None = None,
                        industry: str | None = None, company: str | None = None, min_importance: float | None = Query(None, ge=0, le=1),
                        limit: int = Query(MAX_EVENTS, ge=1, le=MAX_EVENTS)) -> StreamingResponse:
    company_id = None
    if company:
        async with connection() as conn:
            company_id = (await q.require_company(conn, company))["id"]
    where, params = q.event_filters(company_id=company_id, event_type=event_type, country=country, industry=industry, since=q.parse_iso(since),
                                    until=q.parse_iso(until, "until"), min_importance=min_importance)
    return _response(fmt, "events", _stream(fmt, _iter_events(where, params, limit), EVENT_COLUMNS, _event_flat))


async def _iter_companies(where: list[str], params: dict[str, Any], limit: int) -> AsyncIterator[dict[str, Any]]:
    sent = 0
    last_id = ""
    async with connection() as conn:
        while sent < limit:
            w = list(where) + ["c.id > :cur_id"]
            rows = await fetch_all(conn, f"select c.id from companies c where {' and '.join(w)} order by c.id limit :lim", **params, cur_id=last_id,
                                   lim=min(BATCH, limit - sent))
            if not rows:
                return
            ids = [r["id"] for r in rows]
            for card in await q.fetch_cards_by_ids(conn, ids):
                yield ser.company_card(card)
            sent += len(ids)
            last_id = ids[-1]


@router.get("/export/companies.{fmt}", summary="Export companies (≤ 20 000 rows)")
async def export_companies(fmt: str, country: str | None = None, industry: str | None = None, status: str | None = None, tier: int | None = Query(None, ge=1, le=4),
                           limit: int = Query(MAX_COMPANIES, ge=1, le=MAX_COMPANIES)) -> StreamingResponse:
    where, params = q.company_filters(country=country, industry=industry, status=status, tier=tier)
    return _response(fmt, "companies", _stream(fmt, _iter_companies(where or ["true"], params, limit), COMPANY_COLUMNS, _company_flat))


async def _iter_jobs(where: list[str], params: dict[str, Any], limit: int) -> AsyncIterator[dict[str, Any]]:
    sent = 0
    last_id = ""
    async with connection() as conn:
        while sent < limit:
            w = list(where) + ["j.id > :cur_id"]
            rows = await fetch_all(conn, f"select j.*, c.slug as company_slug from jobs j join companies c on c.id = j.company_id where {' and '.join(w)} "
                                         "order by j.id limit :lim", **params, cur_id=last_id, lim=min(BATCH, limit - sent))
            if not rows:
                return
            for r in rows:
                item = ser.job(r)
                item["company_slug"] = r["company_slug"]
                yield item
            sent += len(rows)
            last_id = rows[-1]["id"]


@router.get("/export/jobs.{fmt}", summary="Export jobs (≤ 10 000 rows)")
async def export_jobs(fmt: str, company: str | None = None, since: str | None = None, status: str = Query("open", pattern="^(open|removed|all)$"),
                      country: str | None = None, ai: str | None = None, limit: int = Query(MAX_JOBS, ge=1, le=MAX_JOBS)) -> StreamingResponse:
    where: list[str] = ["true"]
    params: dict[str, Any] = {}
    if company:
        async with connection() as conn:
            params["cid"] = (await q.require_company(conn, company))["id"]
        where.append("j.company_id = :cid")
    since_dt = q.parse_iso(since)
    if since_dt:
        where.append("j.first_seen_at >= :since")
        params["since"] = since_dt
    if status == "open":
        where.append("j.status = 'open'")
    elif status == "removed":
        where.append("j.status <> 'open'")
    if country:
        where.append("j.country = cast(:country as char(2))")
        params["country"] = country.upper()[:2]
    if q.parse_bool(ai):
        where.append("j.is_ai")
    return _response(fmt, "jobs", _stream(fmt, _iter_jobs(where, params, limit), JOB_COLUMNS, _job_flat))