"""Runs a report template — a saved, multi-source report (Reports in the
Studio). Each section reads its own datasource, form or data query through
the dataset engine (app/crud/dataset_query.py), so record access policies,
hidden fields, filter tokens like {{current_user.id}} and SQL pushdown all
apply exactly as they do everywhere else.

* Report parameters fill section filters and text: {{NAME}} or {{param.NAME}}.
  A filter whose parameter is left blank is skipped; a "between" filter with
  only one end filled becomes "on or after" / "on or before".
* A table section can join columns from other sources by a shared value
  (an employee code, an order number): a left join on the first match,
  shown as "<alias>.<column>". For a datasource's own reference fields, the
  lookup columns ("customer.city") are simpler.
* A section that can't be shown (no access, a source that's gone, a data
  query that failed) carries its error; the rest of the report still runs.
"""

import html
import re
import uuid
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone
from typing import Any

from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.crud.dataset_query import _aggregate_value, resolve_rows
from app.models.master import Master

_TOKEN = re.compile(r"\{\{\s*(?:param\.)?([A-Za-z_][A-Za-z0-9_]*)\s*\}\}")
MAX_JOIN_ROWS = 20_000

# Raises HTTPException when the viewer may not read the source.
AccessCheck = Callable[[str, uuid.UUID], Awaitable[None]]


class ReportError(Exception):
    pass


# ---- Parameters -------------------------------------------------------------------------------------


def param_values(defs: list[dict], given: dict | None) -> dict[str, Any]:
    """Each declared parameter's value (given, else its default), typed;
    None when blank. Undeclared names are ignored."""
    given = given or {}
    out: dict[str, Any] = {}
    for p in defs:
        name = p["name"]
        raw = given.get(name)
        if raw is None or (isinstance(raw, str) and not raw.strip()):
            raw = p.get("default")
        if raw is None or (isinstance(raw, str) and not raw.strip()):
            if p.get("required"):
                raise ReportError(f"Enter {p.get('label') or name}.")
            out[name] = None
            continue
        if p.get("type") == "number" and not isinstance(raw, (int, float)):
            try:
                number = float(str(raw).strip())
            except ValueError as exc:
                raise ReportError(f"{p.get('label') or name} must be a number.") from exc
            raw = int(number) if number.is_integer() else number
        out[name] = raw if not isinstance(raw, str) else raw.strip()
    return out


_BLANK = object()  # a parameter token that was left blank


def _subst(value: Any, params: dict[str, Any]) -> Any:
    if not isinstance(value, str) or "{{" not in value:
        return value
    whole = _TOKEN.fullmatch(value.strip())
    if whole and whole.group(1) in params:
        v = params[whole.group(1)]
        return _BLANK if v is None else v
    return _TOKEN.sub(lambda m: "" if params.get(m.group(1)) is None else str(params[m.group(1)]) if m.group(1) in params else m.group(0), value)


def apply_params(filters: list[dict], params: dict[str, Any]) -> list[dict]:
    from app.crud.filter_tree import is_group

    out: list[dict] = []
    for f in filters:
        if is_group(f):
            # A group keeps the conditions whose parameters were given.
            out.append({**f, "filters": apply_params(f.get("filters") or [], params)})
            continue
        raw = f.get("value")
        if isinstance(raw, list):
            values = [_subst(v, params) for v in raw]
            if f.get("operator") == "between" and len(values) == 2:
                lo, hi = values
                if lo is _BLANK and hi is _BLANK:
                    continue
                if lo is _BLANK:
                    out.append({**f, "operator": "lte", "value": hi})
                    continue
                if hi is _BLANK:
                    out.append({**f, "operator": "gte", "value": lo})
                    continue
            values = [v for v in values if v is not _BLANK]
            if not values and raw:
                continue
            out.append({**f, "value": values})
        else:
            v = _subst(raw, params)
            if v is _BLANK:
                continue
            out.append({**f, "value": v})
    return out


def fill_text(text: str | None, params: dict[str, Any]) -> str | None:
    if not text:
        return text
    return _TOKEN.sub(lambda m: "" if params.get(m.group(1)) is None else str(params[m.group(1)]) if m.group(1) in params else m.group(0), text)


# ---- Sections ---------------------------------------------------------------------------------------


async def _aliases(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID) -> dict[str, str]:
    """A datasource's column display names (MasterFieldDef.alias)."""
    if source_type != "master":
        return {}
    master = (await db.execute(select(Master).where(Master.id == source_id, Master.tenant_id == tenant_id))).scalar_one_or_none()
    return {f["name"]: f["alias"].strip() for f in (master.field_schema if master else []) if isinstance(f, dict) and (f.get("alias") or "").strip()}


_TAG = re.compile(r"<[^>]+>")
_BLOCK_END = re.compile(r"</(p|div|li|h[1-6])>|<br\s*/?>", re.I)


def _plain(value: Any) -> Any:
    """Rich text (a form's rich_text/html field) as plain text — reports,
    spreadsheets and PDFs show what it says, not its markup."""
    if not isinstance(value, str) or "<" not in value or not _TAG.search(value):
        return value
    text = _TAG.sub("", _BLOCK_END.sub("\n", value))
    return re.sub(r"\n{2,}", "\n", html.unescape(text)).strip()


def _totals(columns: list[str], rows: list[dict]) -> dict[str, float] | None:
    totals: dict[str, float] = {}
    for c in columns:
        if c == "id" or c.endswith("_at"):
            continue
        values = [r.get(c) for r in rows if r.get(c) not in (None, "")]
        if values and all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values):
            total = sum(values)
            totals[c] = int(total) if float(total).is_integer() else round(total, 6)
    return totals or None


async def _join(db, tenant_id, user_id, rows: list[dict], join: dict, params: dict) -> list[dict]:
    """The rows with the joined columns added — see ReportJoin. Matches are
    taken in the order they were created, so "first" and "latest" are stable."""
    foreign = join["foreign_field"]
    wanted = list(dict.fromkeys([*join["columns"], foreign, "created_at"]))
    other, _, _ = await resolve_rows(
        db, tenant_id, join["source_type"], uuid.UUID(str(join["source_id"])), apply_params(join.get("filters") or [], params),
        "created_at", "asc", wanted, limit=MAX_JOIN_ROWS, user_id=user_id,
    )
    index: dict[str, list[dict]] = {}
    for r in other:
        key = r.get(foreign)
        if key not in (None, ""):
            index.setdefault(str(key), []).append(r)
    alias, mode = join["alias"], join.get("match") or "first"
    out: list[dict] = []
    for row in rows:
        local = row.get(join["local_field"])
        matches = index.get(str(local), []) if local not in (None, "") else []
        if mode == "first":
            matches = matches[:1]
        elif mode == "latest":
            matches = matches[-1:]
        for match in matches or [None]:
            out.append({**row, **{f"{alias}.{c}": (match.get(c) if match else None) for c in join["columns"]}})
    return out


_LAYOUT_TYPES = {"section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button"}
_SYSTEM_COLUMNS = ["created_at", "updated_at", "created_by", "updated_by"]


async def source_fields(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID) -> list[str]:
    """The field names a section's source offers to a run-time filter."""
    if source_type == "query":
        from app.crud import data_queries

        q = await data_queries.load(db, tenant_id, source_id)
        if q is None:
            return []
        return [*q.columns, *(p["name"] for p in q.parameters or [] if p["name"] not in q.columns)]
    from app.crud import record_sql

    try:
        src = await record_sql.load_source(db, tenant_id, source_type, source_id)
    except HTTPException:
        return []
    own = [f["name"] for f in src.field_defs if f.get("name") and f.get("type") not in _LAYOUT_TYPES]
    return [*own, *(c for c in _SYSTEM_COLUMNS if c not in own), "id"]


async def run_section(
    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, section: dict, params: dict[str, Any], check: AccessCheck, override: dict | None = None
) -> dict:
    override = override or {}
    out: dict[str, Any] = {
        "id": section["id"], "kind": section["kind"], "title": fill_text(section.get("title"), params), "width": section.get("width", "full"),
        "chart_type": section.get("chart_type"), "column_labels": dict(section.get("column_labels") or {}),
    }
    if section["kind"] == "text":
        out["text"] = fill_text(section.get("text"), params)
        return out
    source_type, source_id = section["source_type"], uuid.UUID(str(section["source_id"]))
    # The section's own filters, then whatever the person running it added.
    filters = apply_params([*(section.get("filters") or []), *(override.get("filters") or [])], params)
    fn, field = section.get("aggregate_fn"), section.get("aggregate_field")
    try:
        await check(source_type, source_id)
        for j in section.get("joins") or []:
            await check(j["source_type"], uuid.UUID(str(j["source_id"])))
        out["fields"] = await source_fields(db, tenant_id, source_type, source_id)
        out["column_labels"] = {
            **await _aliases(db, tenant_id, source_type, source_id),
            **out["column_labels"],
            **{k: v for k, v in (override.get("column_labels") or {}).items() if v and v.strip()},
        }

        if section["kind"] == "kpi":
            if section.get("expression"):
                rows, total, _ = await resolve_rows(db, tenant_id, source_type, source_id, filters, None, "asc", [], user_id=user_id, expression=section["expression"])
                out["value"] = rows[0]["value"] if rows else None
            elif fn == "count":
                _, total, _ = await resolve_rows(db, tenant_id, source_type, source_id, filters, None, "asc", [], limit=1, user_id=user_id)
                out["value"] = total
            else:
                rows, total, _ = await resolve_rows(db, tenant_id, source_type, source_id, filters, None, "asc", [field], user_id=user_id)
                out["value"] = _aggregate_value(fn, [r.get(field) for r in rows]) if rows else None
            out["total"] = total
            return out

        if section["kind"] in ("summary", "chart"):
            rows, total, columns = await resolve_rows(
                db, tenant_id, source_type, source_id, filters, section.get("sort_field"), section.get("sort_dir", "asc"), [],
                limit=section.get("limit", 500), group_by=section["group_by"], aggregate_field=field, aggregate_fn=fn or "count", user_id=user_id,
            )
        else:
            joins = section.get("joins") or []
            asked = list(section.get("columns") or [])
            needed = list(dict.fromkeys([*asked, *(j["local_field"] for j in joins)])) if asked else []
            rows, total, columns = await resolve_rows(
                db, tenant_id, source_type, source_id, filters, section.get("sort_field"), section.get("sort_dir", "asc"), needed,
                limit=section.get("limit", 500), user_id=user_id,
            )
            if asked:
                columns = asked
            else:
                columns = [c for c in columns if c != "id"]
            for j in joins:
                rows = await _join(db, tenant_id, user_id, rows, j, params)
                columns = [*columns, *(f"{j['alias']}.{c}" for c in j["columns"])]
            out["all_columns"] = list(columns)
            # The run's own choice of columns — only ones the section can show.
            chosen = [c for c in override.get("columns") or [] if c in columns]
            if chosen:
                columns = chosen
            rows = [{c: _plain(r.get(c)) for c in columns} for r in rows]
        if section["kind"] != "table":
            out["all_columns"] = list(columns)
        out.update(columns=columns, rows=rows, total=total)
        if section.get("show_totals"):
            out["totals"] = _totals(columns, rows)
    except HTTPException as exc:
        detail = exc.detail
        out["error"] = detail if isinstance(detail, str) else "; ".join(d.get("msg", "") for d in detail) if isinstance(detail, list) else str(detail)
    return out


async def run_report(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None,
    template: dict,
    given: dict | None,
    check: AccessCheck,
    overrides: dict[str, dict] | None = None,
) -> dict:
    params = param_values(template.get("parameters") or [], given)
    settings = template.get("settings") or {}
    overrides = overrides or {}
    sections = [await run_section(db, tenant_id, user_id, s, params, check, overrides.get(s["id"])) for s in template.get("sections") or []]
    return {
        "name": template.get("name") or "Report",
        "subtitle": fill_text(settings.get("subtitle"), params),
        "generated_at": datetime.now(timezone.utc) if settings.get("show_generated_at", True) else None,
        "params": params,
        "sections": sections,
    }