import asyncio
import csv
import io
import json
import re
import uuid
from datetime import datetime

import httpx
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi.responses import StreamingResponse
from openpyxl import Workbook
from reportlab.lib import colors
from reportlab.lib.pagesizes import landscape, letter
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.ai_client import chat_completion, strip_code_fence
from app.core.database import get_db
from app.core.timezones import localize_rows, tenant_zone
from app.core.permissions import check_portal_action, require_permission, resolve_action_matrix, resolve_portal_grants
from app.core.security import AuthContext, get_auth_context
from app.crud import record_sql
from app.crud.audit import write_audit_log
from app.crud.base import CRUDBase
from app.crud.dataset_query import resolve_grouped_series, resolve_rows
from app.models.master import Master
from app.models.reporting import Dataset
from app.models.workflow import Form
from app.schemas.reporting import (
    AiQueryRequest,
    AiQueryResult,
    DatasetCreate,
    DatasetFilter,
    DatasetQueryRequest,
    DatasetQueryResult,
    DatasetRead,
    DatasetSeriesQueryRequest,
    DatasetSeriesQueryResult,
    DatasetUpdate,
)

router = APIRouter(prefix="/datasets", tags=["datasets"])
crud = CRUDBase[Dataset, DatasetCreate, DatasetUpdate](Dataset)


def _cell_value(value: object) -> object:
    if isinstance(value, (dict, list)):
        return str(value)
    return value


async def _export_headers(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID, columns: list[str]) -> list[str]:
    """The header row an export prints for `columns` — a Master column's alias
    (display name) where it has one, else the raw name. Row values are still
    looked up by the raw name; only the printed header changes."""
    if source_type != "master":
        return list(columns)
    master = (await db.execute(select(Master).where(Master.id == source_id, Master.tenant_id == tenant_id))).scalar_one_or_none()
    aliases = {f.get("name"): (f.get("alias") or "").strip() for f in (master.field_schema if master else []) if isinstance(f, dict)}
    return [aliases.get(c) or c for c in columns]


def _csv_text(value: object) -> str:
    """A cell as CSV text — and a leading = + - @ neutralised, so a value
    can't run as a formula when the file is opened in a spreadsheet."""
    if value is None:
        return ""
    text = str(value) if not isinstance(value, (dict, list)) else json.dumps(value, ensure_ascii=False, default=str)
    return "'" + text if text[:1] in ("=", "+", "-", "@") and not _NUMBER.fullmatch(text) else text


_NUMBER = re.compile(r"-?\d+(\.\d+)?")


def rows_to_csv(columns: list[str], rows: list[dict], headers: list[str] | None = None) -> io.BytesIO:
    """UTF-8 with a byte-order mark, so Excel reads Malayalam and other
    scripts correctly."""
    out = io.StringIO()
    writer = csv.writer(out)
    writer.writerow(headers or columns)
    for row in rows:
        writer.writerow([_csv_text(row.get(c)) for c in columns])
    return io.BytesIO(("\ufeff" + out.getvalue()).encode("utf-8"))


def _rows_to_xlsx(sheet_title: str, columns: list[str], rows: list[dict], headers: list[str] | None = None) -> io.BytesIO:
    wb = Workbook()
    ws = wb.active
    ws.title = sheet_title[:31] or "Sheet1"  # Excel's own 31-char sheet-name limit
    ws.append(headers or columns)
    for row in rows:
        ws.append([_cell_value(row.get(c)) for c in columns])
    buffer = io.BytesIO()
    wb.save(buffer)
    buffer.seek(0)
    return buffer


def _rows_to_pdf(title: str, columns: list[str], rows: list[dict], headers: list[str] | None = None) -> io.BytesIO:
    """Builds a landscape, paginated table PDF via reportlab's platypus
    layout engine (Table/SimpleDocTemplate handle page breaks and repeating
    the header row automatically — unlike app/crud/tasks.py's PDF-generation
    node, which hand-rolls plain-text-line pagination with pdfgen.canvas;
    that approach isn't reusable for tabular data). Runs sync — callers must
    offload via asyncio.to_thread, same as _exec_pdf_generate does."""
    buffer = io.BytesIO()
    doc = SimpleDocTemplate(buffer, pagesize=landscape(letter), title=title)
    styles = getSampleStyleSheet()
    cell_style = styles["BodyText"]
    cell_style.fontSize = 8
    header_style = styles["BodyText"].clone("header")
    header_style.fontSize = 8
    header_style.textColor = colors.white

    header_row = [Paragraph(f"<b>{c}</b>", header_style) for c in (headers or columns)]
    body_rows = [
        [Paragraph(str(_cell_value(row.get(c))) if row.get(c) is not None else "", cell_style) for c in columns]
        for row in rows
    ]
    col_width = doc.width / len(columns) if columns else doc.width
    table = Table([header_row, *body_rows], repeatRows=1, colWidths=[col_width] * len(columns))
    table.setStyle(
        TableStyle(
            [
                ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1e293b")),
                ("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#cbd5e1")),
                ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#f8fafc")]),
                ("VALIGN", (0, 0), (-1, -1), "TOP"),
            ]
        )
    )
    doc.build([Paragraph(title, styles["Title"]), Spacer(1, 12), table])
    buffer.seek(0)
    return buffer

async def _require_query_access(db: AsyncSession, ctx: AuthContext, source_type: str, source_id: uuid.UUID) -> None:
    """Shared access check for every endpoint that resolves rows off a raw
    source_type/source_id pair (query/query-series/export.xlsx/export.pdf) —
    not a plain require_permission("analytics", "view") dependency, since a
    vendor caller's access has to be checked per source record (the same
    masters/forms item grants a Portal role holds, see
    app/core/permissions.py::require_portal_item), not by an "analytics"
    grant that no portal role can ever carry. Raises 403 on failure.
    A Data Query (source_type "query") is governed by its own roles instead."""
    if source_type == "query":
        from app.crud import data_queries

        query = await data_queries.load(db, ctx.tenant_id, source_id)
        if query is None:
            raise HTTPException(status_code=404, detail="Data query not found.")
        if not await data_queries.can_run(db, ctx.tenant_id, ctx.user_id, query, vendor=ctx.vendor_id is not None):
            raise HTTPException(status_code=403, detail="Your role can't run this data query.")
        return
    if ctx.vendor_id is not None:
        grants = await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)
        module = "masters" if source_type == "master" else "forms"
        if str(source_id).lower() not in grants.get(module, set()):
            raise HTTPException(status_code=403, detail="Your portal role doesn't grant access to this item.")
    else:
        matrix, is_system_admin = await resolve_action_matrix(db, ctx.tenant_id, ctx.user_id)
        if not is_system_admin and "view" not in matrix.get("analytics", []):
            raise HTTPException(status_code=403, detail="Your role doesn't have 'view' access to analytics.")


async def _require_export(db: AsyncSession, ctx: AuthContext, source_type: str, source_id: uuid.UUID) -> None:
    """Exporting rows: a portal user needs the source's "export" permission
    (Portal Roles; a role with no detailed setting has it only with the
    Report Builder — permissions.py::portal_item_actions). A Data Query
    source, which no portal role grants item by item, still needs the
    Report Builder. A no-op for Studio users (their /query access applies)."""
    if ctx.vendor_id is None:
        return
    if source_type in ("master", "form"):
        await check_portal_action(db, ctx, "masters" if source_type == "master" else "forms", source_id, "export")
    else:
        await _require_reports_feature(db, ctx)


async def _require_reports_feature(db: AsyncSession, ctx: AuthContext) -> None:
    """Extra gate for the Report Builder's AI query (and the export of a
    Data Query source, see _require_export): a vendor also needs the
    whole-feature "reports" grant a Portal role can hold (see
    PortalRoles.tsx), on top of _require_query_access's per-source-record
    check. Deliberately separate from _require_query_access, which stays
    open to /query and /query-series regardless of this flag — those back
    Page/Dashboard widgets, which aren't the Report Builder and shouldn't
    require it. A no-op for non-vendor (Studio) callers."""
    if ctx.vendor_id is None:
        return
    grants = await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)
    if not grants.get("reports"):
        raise HTTPException(status_code=403, detail="Your portal role doesn't grant access to Reports.")


_VALID_OPERATORS = {
    "equals", "not_equals", "gt", "gte", "lt", "lte", "in", "contains", "is_empty", "is_not_empty",
}
_VALID_AGGREGATE_FNS = {"count", "sum", "avg", "min", "max"}

_AI_QUERY_SYSTEM_PROMPT = """You are the query assistant inside FlowFabric's Data Explorer. The user describes, in \
plain language, what data they want to see. Translate that into a single structured query against exactly ONE of \
the tables listed below under AVAILABLE TABLES. Never invent a table or field name that isn't listed there.

A field marked "(reference -> Other Table)" is a link to another table's records, already resolved to that \
record's display name wherever you use it as a column, group_by, or filter field — this is how you pull in data \
that lives in a different table (e.g. a Country name shown on a Packages row) without naming a second table.

Respond with ONLY a JSON object of this exact shape, no prose outside it:
{{"message": "<one short sentence explaining what you built>",
 "source_type": "master"|"form",
 "source_name": "<the table's exact name, copied verbatim from the list below>",
 "columns": ["<field name>", "..."],
 "filters": [{{"field": "<field name>", "operator": "<see allowed operators>", "value": <a JSON value, or null for is_empty/is_not_empty>}}],
 "sort_field": "<field name, or the aggregate's own output field below, or null for unsorted>",
 "sort_dir": "asc"|"desc",
 "group_by": "<field name to group by, or null>",
 "aggregate_field": "<field name to sum/average/min/max, or null — omit/null when aggregate_fn is \\"count\\">",
 "aggregate_fn": "count"|"sum"|"avg"|"min"|"max"|null,
 "suggested_name": "<a short, human-friendly name for this query, e.g. \\"Active vendors in California\\">"}}

Allowed operators: equals, not_equals, gt, gte, lt, lte, in, contains, is_empty, is_not_empty. "in" takes a JSON \
array as its value. Leave "columns" as [] to include every field on the table — only list specific ones when the \
user clearly asked to see just a subset (and always leave it [] when group_by is set, since the output columns \
become the group_by field plus the aggregate result automatically). Only ever use field names that exist on the \
table you chose, plus "id" (every table has one — its record's UUID primary key — even though it isn't listed \
per-field below); only include "id" in "columns" when the user actually asked for an identifier.

Use group_by/aggregate_fn whenever the user asks for a COUNT, TOTAL, SUM, AVERAGE, or "how many ... per/by ..." — \
e.g. "count of currencies by status" -> group_by "status", aggregate_fn "count". "total price of packages by \
country" -> group_by the country reference field, aggregate_fn "sum", aggregate_field "price". Leave group_by/\
aggregate_fn null for a plain row listing.

If nothing in the request plausibly maps to any table below, respond instead with exactly:
{{"message": "<a short, helpful explanation of why nothing matched>", "error": true}}

AVAILABLE TABLES:
{catalog}
"""


def _describe_master_field(f: dict, masters_by_id: dict[str, Master]) -> str:
    field_type = f.get("type", "text")
    if field_type == "reference":
        linked = masters_by_id.get(str(f.get("reference_master_id")))
        if f.get("reference_form_id"):
            return f"{f['name']} (reference -> a form's submissions)"
        return f"{f['name']} (reference -> {linked.name if linked else 'another table'})"
    if field_type == "single_select" and f.get("options"):
        return f"{f['name']} (single_select: {', '.join(f['options'])})"
    return f"{f['name']} ({field_type})"


def _describe_form_field(f: dict) -> str:
    field_type = f.get("type", "text")
    if field_type in ("dropdown", "radio", "multiselect", "checkbox_group") and f.get("options"):
        return f"{f['name']} ({field_type}: {', '.join(f['options'])})"
    return f"{f['name']} ({field_type})"


def _build_catalog(masters: list[Master], forms: list[Form]) -> str:
    masters_by_id = {str(m.id): m for m in masters}
    lines = [
        f"[master] {m.name}: "
        + (", ".join(_describe_master_field(f, masters_by_id) for f in m.field_schema) or "(no fields yet)")
        for m in masters
    ] + [
        f"[form] {f.name}: " + (", ".join(_describe_form_field(fd) for fd in f.field_definitions) or "(no fields yet)")
        for f in forms
    ]
    return "\n".join(lines) if lines else "(no datasources or forms exist yet)"


@router.get("", response_model=list[DatasetRead])
async def list_datasets(
    skip: int = 0,
    limit: int = 100,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "view")),
) -> list[Dataset]:
    return await crud.list(db, ctx.tenant_id, skip=skip, limit=limit)


@router.get("/{dataset_id}", response_model=DatasetRead)
async def get_dataset(
    dataset_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "view")),
) -> Dataset:
    return await crud.get(db, ctx.tenant_id, dataset_id)


@router.post("", response_model=DatasetRead, status_code=status.HTTP_201_CREATED)
async def create_dataset(
    payload: DatasetCreate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "create")),
) -> Dataset:
    return await crud.create(db, ctx.tenant_id, payload, created_by=ctx.user_id)


@router.patch("/{dataset_id}", response_model=DatasetRead)
async def update_dataset(
    dataset_id: uuid.UUID,
    payload: DatasetUpdate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "edit")),
) -> Dataset:
    return await crud.update(db, ctx.tenant_id, dataset_id, payload)


@router.delete("/{dataset_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_dataset(
    dataset_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "delete")),
) -> None:
    await crud.delete(db, ctx.tenant_id, dataset_id)


@router.get("/{dataset_id}/data", response_model=DatasetQueryResult)
async def get_dataset_data(
    dataset_id: uuid.UUID,
    limit: int = Query(500, ge=0, le=record_sql.MAX_PYTHON_ROWS),
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "view")),
) -> DatasetQueryResult:
    """Resolves a saved dataset's rows fresh (no materialization) — used by
    the Data Explorer to preview a saved dataset and by dashboard widgets
    bound to data_source "dataset:<id>"."""
    dataset = await crud.get(db, ctx.tenant_id, dataset_id)
    rows, total, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        dataset.source_type,
        dataset.source_id,
        [f if isinstance(f, dict) else f.model_dump() for f in dataset.filters],
        dataset.sort_field,
        dataset.sort_dir,
        dataset.columns,
        limit=limit,
        group_by=dataset.group_by,
        aggregate_field=dataset.aggregate_field,
        aggregate_fn=dataset.aggregate_fn,
        user_id=ctx.user_id,
        computed=list(dataset.computed or []),
        joins=list(dataset.joins or []),
    )
    return DatasetQueryResult(columns=columns, rows=rows, total=total)


@router.get("/{dataset_id}/export.xlsx")
async def export_dataset_xlsx(
    dataset_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "view")),
) -> StreamingResponse:
    dataset = await crud.get(db, ctx.tenant_id, dataset_id)
    if ctx.vendor_id is not None:
        await _require_query_access(db, ctx, dataset.source_type, dataset.source_id)
        await _require_export(db, ctx, dataset.source_type, dataset.source_id)
        for join in dataset.joins or []:
            await _require_query_access(db, ctx, join["source_type"], uuid.UUID(str(join["source_id"])))
    rows, _, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        dataset.source_type,
        dataset.source_id,
        [f if isinstance(f, dict) else f.model_dump() for f in dataset.filters],
        dataset.sort_field,
        dataset.sort_dir,
        dataset.columns,
        limit=10_000,
        group_by=dataset.group_by,
        aggregate_field=dataset.aggregate_field,
        aggregate_fn=dataset.aggregate_fn,
        user_id=ctx.user_id,
        computed=list(dataset.computed or []),
        joins=list(dataset.joins or []),
    )
    rows = localize_rows(rows, await tenant_zone(db, ctx.tenant_id))
    buffer = _rows_to_xlsx(dataset.name, columns, rows, await _export_headers(db, ctx.tenant_id, dataset.source_type, dataset.source_id, columns))
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "dataset.exported", "dataset", dataset_id, {"format": "xlsx", "rows": len(rows)})
    await db.flush()
    filename = f"{dataset.name.replace(' ', '_')}.xlsx"
    return StreamingResponse(
        buffer,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


@router.get("/{dataset_id}/export.csv")
async def export_dataset_csv(
    dataset_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("analytics", "view")),
) -> StreamingResponse:
    """The same rows as export.xlsx, as CSV."""
    dataset = await crud.get(db, ctx.tenant_id, dataset_id)
    if ctx.vendor_id is not None:
        await _require_query_access(db, ctx, dataset.source_type, dataset.source_id)
        await _require_export(db, ctx, dataset.source_type, dataset.source_id)
        for join in dataset.joins or []:
            await _require_query_access(db, ctx, join["source_type"], uuid.UUID(str(join["source_id"])))
    rows, _, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        dataset.source_type,
        dataset.source_id,
        [f if isinstance(f, dict) else f.model_dump() for f in dataset.filters],
        dataset.sort_field,
        dataset.sort_dir,
        dataset.columns,
        limit=10_000,
        group_by=dataset.group_by,
        aggregate_field=dataset.aggregate_field,
        aggregate_fn=dataset.aggregate_fn,
        user_id=ctx.user_id,
        computed=list(dataset.computed or []),
        joins=list(dataset.joins or []),
    )
    rows = localize_rows(rows, await tenant_zone(db, ctx.tenant_id))
    buffer = rows_to_csv(columns, rows, await _export_headers(db, ctx.tenant_id, dataset.source_type, dataset.source_id, columns))
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "dataset.exported", "dataset", dataset_id, {"format": "csv", "rows": len(rows)})
    await db.flush()
    filename = f"{dataset.name.replace(' ', '_')}.csv"
    return StreamingResponse(buffer, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{filename}"'})


@router.post("/ai-query", response_model=AiQueryResult)
async def ai_query_datasets(
    payload: AiQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> AiQueryResult:
    """The Data Explorer's AI Query assistant (FR-AIA-006/007's provider
    plumbing, reused for a new purpose): the user describes what they want
    in plain language, the model picks a real table from the tenant's own
    Masters/Forms and drafts columns/filters/sort against it. The AI never
    sees or supplies a UUID — it names the table, and this endpoint resolves
    that name against the tenant's actual records and drops any field name
    or operator the model invented that doesn't really exist, so a
    hallucinated response can't silently misquery or fail with a 500.

    Not a plain require_permission("analytics", "view") dependency (same
    reasoning as query_dataset below): a vendor caller's catalog is
    pre-filtered to only the Masters/Forms their Portal role grants (see
    resolve_portal_grants), so the AI can never pick — and the final
    resolve_rows call can never touch — a source they aren't granted. A
    Studio caller still needs a real "analytics" view permission. A vendor
    also needs the "reports" feature grant — see _require_reports_feature."""
    await _require_reports_feature(db, ctx)
    masters = list((await db.execute(select(Master).where(Master.tenant_id == ctx.tenant_id))).scalars().all())
    forms = list(
        (
            await db.execute(
                select(Form).where(Form.tenant_id == ctx.tenant_id, Form.is_current.is_(True)).order_by(Form.name)
            )
        )
        .scalars()
        .all()
    )

    if ctx.vendor_id is not None:
        grants = await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)
        masters = [m for m in masters if str(m.id).lower() in grants.get("masters", set())]
        forms = [f for f in forms if str(f.form_id).lower() in grants.get("forms", set())]
    else:
        matrix, is_system_admin = await resolve_action_matrix(db, ctx.tenant_id, ctx.user_id)
        if not is_system_admin and "view" not in matrix.get("analytics", []):
            raise HTTPException(status_code=403, detail="Your role doesn't have 'view' access to analytics.")
    if not masters and not forms:
        return AiQueryResult(message="", error="No datasources or forms exist yet to query.")

    messages = [
        {"role": "system", "content": _AI_QUERY_SYSTEM_PROMPT.format(catalog=_build_catalog(masters, forms))},
        {"role": "user", "content": payload.prompt},
    ]
    try:
        content = await chat_completion(messages)
        parsed = json.loads(strip_code_fence(content))
    except (httpx.HTTPError, KeyError, json.JSONDecodeError) as exc:
        return AiQueryResult(
            message="",
            error=f"AI query assistant is unavailable ({exc}); build the query manually instead.",
            provider_ok=False,
        )

    if parsed.get("error"):
        msg = parsed.get("message") or "Could not interpret that request."
        return AiQueryResult(message=msg, error=msg)

    source_type = parsed.get("source_type")
    source_name = (parsed.get("source_name") or "").strip()
    if source_type not in ("master", "form") or not source_name:
        return AiQueryResult(message=parsed.get("message", ""), error="The AI's response didn't name a valid table.")

    match_master = next((m for m in masters if m.name.lower() == source_name.lower()), None) if source_type == "master" else None
    match_form = next((f for f in forms if f.name.lower() == source_name.lower()), None) if source_type == "form" else None
    if not match_master and not match_form:
        return AiQueryResult(
            message=parsed.get("message", ""),
            error=f'Couldn\'t match a {source_type} named "{source_name}" — pick a table manually on the left instead.',
        )

    resolved_id = match_master.id if match_master else match_form.form_id
    resolved_name = match_master.name if match_master else match_form.name
    valid_fields = (
        {f["name"] for f in match_master.field_schema} if match_master else {f["name"] for f in match_form.field_definitions}
    )
    valid_fields.add("id")  # every record's own UUID primary key — see resolve_rows

    columns = [c for c in (parsed.get("columns") or []) if isinstance(c, str) and c in valid_fields]

    filters: list[DatasetFilter] = []
    for f in parsed.get("filters") or []:
        if not isinstance(f, dict):
            continue
        field, operator = f.get("field"), f.get("operator")
        if field in valid_fields and operator in _VALID_OPERATORS:
            filters.append(DatasetFilter(field=field, operator=operator, value=f.get("value")))

    group_by = parsed.get("group_by")
    group_by = group_by if group_by in valid_fields else None

    aggregate_fn = parsed.get("aggregate_fn")
    aggregate_fn = aggregate_fn if aggregate_fn in _VALID_AGGREGATE_FNS else None
    if not group_by:
        aggregate_fn = None  # an aggregate_fn without a group_by is meaningless — drop it rather than guess

    aggregate_field = parsed.get("aggregate_field")
    aggregate_field = aggregate_field if aggregate_fn and aggregate_fn != "count" and aggregate_field in valid_fields else None
    if aggregate_fn and aggregate_fn != "count" and not aggregate_field:
        aggregate_fn = None  # sum/avg/min/max need a real numeric field to operate on — fall back to a plain query

    sort_field = parsed.get("sort_field")
    valid_sort_fields = valid_fields | ({"count", f"{aggregate_fn}_{aggregate_field}"} if aggregate_fn else set())
    sort_field = sort_field if sort_field in valid_sort_fields else None
    sort_dir = parsed.get("sort_dir") if parsed.get("sort_dir") in ("asc", "desc") else "asc"

    rows, total, result_columns = await resolve_rows(
        db, ctx.tenant_id, source_type, resolved_id,
        [f.model_dump() for f in filters], sort_field, sort_dir, columns, limit=200,
        group_by=group_by, aggregate_field=aggregate_field, aggregate_fn=aggregate_fn,
        user_id=ctx.user_id,
    )

    return AiQueryResult(
        message=parsed.get("message", ""),
        source_type=source_type,
        source_id=resolved_id,
        source_name=resolved_name,
        columns=columns,
        filters=filters,
        sort_field=sort_field,
        sort_dir=sort_dir,
        group_by=group_by,
        aggregate_field=aggregate_field,
        aggregate_fn=aggregate_fn,
        suggested_name=parsed.get("suggested_name") or None,
        rows=rows,
        result_columns=result_columns,
        total=total,
    )


@router.post("/query", response_model=DatasetQueryResult)
async def query_dataset(
    payload: DatasetQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> DatasetQueryResult:
    """Ad-hoc, unsaved query — the Data Explorer's "Run" button (internal),
    evaluated against live data before the user decides whether to save it —
    and also what a Page's data_list block or a Dashboard widget calls to
    resolve its rows (see PageBlockRenderer.tsx/DashboardWidgetView.tsx's
    shared parseWidgetDataSource), including from the Public Portal.

    Not a plain require_permission("analytics", "view") dependency (unlike
    every other route in this router) because a vendor caller's access here
    has to be checked per source record, not per module+action: it's
    governed by the same masters/forms item grants a Portal role holds for
    the record this query actually reads (see
    app/core/permissions.py::require_portal_item), not by an "analytics"
    grant that no portal role can ever carry."""
    await _require_query_access(db, ctx, payload.source_type, payload.source_id)
    for join in payload.joins:
        await _require_query_access(db, ctx, join.source_type, join.source_id)
    rows, total, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        payload.source_type,
        payload.source_id,
        [f.model_dump() for f in payload.filters],
        payload.sort_field,
        payload.sort_dir,
        payload.columns,
        limit=payload.limit,
        group_by=payload.group_by,
        aggregate_field=payload.aggregate_field,
        aggregate_fn=payload.aggregate_fn,
        user_id=ctx.user_id,
        computed=[c.model_dump() for c in payload.computed],
        joins=[j.model_dump(mode="json") for j in payload.joins],
        expression=payload.expression,
    )
    return DatasetQueryResult(columns=columns, rows=rows, total=total)


@router.post("/query-series", response_model=DatasetSeriesQueryResult)
async def query_dataset_series(
    payload: DatasetSeriesQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> DatasetSeriesQueryResult:
    """Axis-configured chart query (Dashboard/Page bar_chart/line_chart
    widgets with explicit x_field/series, see
    app/crud/dataset_query.py::resolve_grouped_series) — same access-control
    shape as /query above (vendor callers checked per source-record grant,
    everyone else per the "analytics" module's view permission), since this
    resolves the exact same underlying rows, just grouped differently."""
    await _require_query_access(db, ctx, payload.source_type, payload.source_id)
    for join in payload.joins:
        await _require_query_access(db, ctx, join.source_type, join.source_id)
    rows, series = await resolve_grouped_series(
        db,
        ctx.tenant_id,
        payload.source_type,
        payload.source_id,
        [f.model_dump() for f in payload.filters],
        payload.x_field,
        payload.x_bucket,
        [s.model_dump() for s in payload.series],
        limit=payload.limit,
        user_id=ctx.user_id,
        computed=[c.model_dump() for c in payload.computed],
        joins=[j.model_dump(mode="json") for j in payload.joins],
    )
    return DatasetSeriesQueryResult(rows=rows, series=series)


@router.post("/export.xlsx")
async def export_query_xlsx(
    payload: DatasetQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> StreamingResponse:
    """Exports an ad-hoc, unsaved query — the Data Explorer's "Export to
    Excel" button (and the Public Portal's Report Builder's own Excel
    export), same query shape /query already takes, so exporting exactly
    what's on screen needs no separate "save this first" step. Same
    per-record vendor grant check as /query — see _require_query_access — plus
    the source's "export" permission (_require_export) that /query itself
    doesn't require."""
    await _require_query_access(db, ctx, payload.source_type, payload.source_id)
    for join in payload.joins:
        await _require_query_access(db, ctx, join.source_type, join.source_id)
    await _require_export(db, ctx, payload.source_type, payload.source_id)
    rows, _, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        payload.source_type,
        payload.source_id,
        [f.model_dump() for f in payload.filters],
        payload.sort_field,
        payload.sort_dir,
        payload.columns,
        limit=10_000,
        group_by=payload.group_by,
        aggregate_field=payload.aggregate_field,
        aggregate_fn=payload.aggregate_fn,
        user_id=ctx.user_id,
        computed=[c.model_dump() for c in payload.computed],
        joins=[j.model_dump(mode="json") for j in payload.joins],
    )
    rows = localize_rows(rows, await tenant_zone(db, ctx.tenant_id))
    buffer = _rows_to_xlsx("Export", columns, rows, await _export_headers(db, ctx.tenant_id, payload.source_type, payload.source_id, columns))
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, "dataset.exported", "dataset", None, {"format": "xlsx", "rows": len(rows), "ad_hoc": True}
    )
    await db.flush()
    return StreamingResponse(
        buffer,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": 'attachment; filename="export.xlsx"'},
    )


@router.post("/export.csv")
async def export_query_csv(
    payload: DatasetQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> StreamingResponse:
    """Same ad-hoc query and permissions as /export.xlsx, as CSV."""
    await _require_query_access(db, ctx, payload.source_type, payload.source_id)
    for join in payload.joins:
        await _require_query_access(db, ctx, join.source_type, join.source_id)
    await _require_export(db, ctx, payload.source_type, payload.source_id)
    rows, _, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        payload.source_type,
        payload.source_id,
        [f.model_dump() for f in payload.filters],
        payload.sort_field,
        payload.sort_dir,
        payload.columns,
        limit=10_000,
        group_by=payload.group_by,
        aggregate_field=payload.aggregate_field,
        aggregate_fn=payload.aggregate_fn,
        user_id=ctx.user_id,
        computed=[c.model_dump() for c in payload.computed],
        joins=[j.model_dump(mode="json") for j in payload.joins],
    )
    rows = localize_rows(rows, await tenant_zone(db, ctx.tenant_id))
    buffer = rows_to_csv(columns, rows, await _export_headers(db, ctx.tenant_id, payload.source_type, payload.source_id, columns))
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "dataset.exported", "dataset", None, {"format": "csv", "rows": len(rows), "ad_hoc": True})
    await db.flush()
    return StreamingResponse(buffer, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="export.csv"'})


@router.post("/export.pdf")
async def export_query_pdf(
    payload: DatasetQueryRequest,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
) -> StreamingResponse:
    """Same ad-hoc query as /export.xlsx, rendered as a landscape table PDF
    instead (see _rows_to_pdf) — the Public Portal Report Builder's PDF
    export button. Capped at a lower row limit than the xlsx export
    (2,000 vs 10,000): a PDF table doesn't scale to large row counts the
    way a spreadsheet does. Same export permission as /export.xlsx."""
    await _require_query_access(db, ctx, payload.source_type, payload.source_id)
    for join in payload.joins:
        await _require_query_access(db, ctx, join.source_type, join.source_id)
    await _require_export(db, ctx, payload.source_type, payload.source_id)
    rows, _, columns = await resolve_rows(
        db,
        ctx.tenant_id,
        payload.source_type,
        payload.source_id,
        [f.model_dump() for f in payload.filters],
        payload.sort_field,
        payload.sort_dir,
        payload.columns,
        limit=2_000,
        group_by=payload.group_by,
        aggregate_field=payload.aggregate_field,
        aggregate_fn=payload.aggregate_fn,
        user_id=ctx.user_id,
        computed=[c.model_dump() for c in payload.computed],
        joins=[j.model_dump(mode="json") for j in payload.joins],
    )
    rows = localize_rows(rows, await tenant_zone(db, ctx.tenant_id))
    buffer = await asyncio.to_thread(_rows_to_pdf, "Report", columns, rows, await _export_headers(db, ctx.tenant_id, payload.source_type, payload.source_id, columns))
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, "dataset.exported", "dataset", None, {"format": "pdf", "rows": len(rows), "ad_hoc": True}
    )
    await db.flush()
    return StreamingResponse(
        buffer,
        media_type="application/pdf",
        headers={"Content-Disposition": 'attachment; filename="report.pdf"'},
    )