"""Reading the platform's own LLM usage.

One question, three cuts: **which of our features are spending our AI
budget?** By purpose (the one people actually ask), by model, and by
day. All three are token counts and call counts, never money — see
:mod:`app.domains.bailian.ledger` for why this module refuses to
multiply tokens by a price it does not have.

Two properties worth stating because they change how the numbers read:

* **NULL tokens are not zero.** A vendor that did not report usage
  leaves NULL, and ``SUM`` skips those rows. So every cut also carries
  ``calls_without_token_report`` — a purpose whose tokens look
  implausibly low next to its call count is usually that, not a cheap
  feature. Hiding it would make the ledger quietly disagree with the
  vendor bill.

* **``unattributed`` is shown, never dropped.** It is the coverage gap
  made visible: calls from a construction site that has not declared
  its purpose yet. A ledger that silently omitted them would produce a
  total nobody could reconcile.
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import case, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.bailian.models import LlmCallLedger

#: Widest window the read surface will assemble in one request. Beyond
#: this the honest answer is a rollup job, not a slower query.
MAX_WINDOW_DAYS = 180
DEFAULT_WINDOW_DAYS = 30


@dataclass(frozen=True)
class Window:
    days: int
    since: datetime
    until: datetime


def resolve_window(days: int | None) -> Window:
    requested = int(days or DEFAULT_WINDOW_DAYS)
    clamped = max(1, min(requested, MAX_WINDOW_DAYS))
    until = datetime.now(UTC)
    return Window(days=clamped, since=until - timedelta(days=clamped), until=until)


def _totals_columns() -> tuple[Any, ...]:
    """The measures every cut reports, defined once.

    ``calls_without_token_report`` exists because ``SUM`` over a column
    that is NULL for some rows silently under-reports, and the reader
    has no way to tell that from genuinely low usage.
    """
    return (
        func.count().label("calls"),
        func.sum(case((LlmCallLedger.ok.is_(True), 1), else_=0)).label("ok_calls"),
        func.sum(case((LlmCallLedger.ok.is_(False), 1), else_=0)).label("failed_calls"),
        func.coalesce(func.sum(LlmCallLedger.input_tokens), 0).label("input_tokens"),
        func.coalesce(func.sum(LlmCallLedger.output_tokens), 0).label("output_tokens"),
        func.sum(
            case(
                (
                    LlmCallLedger.input_tokens.is_(None)
                    & LlmCallLedger.output_tokens.is_(None),
                    1,
                ),
                else_=0,
            ),
        ).label("calls_without_token_report"),
    )


def _row_to_dict(row: Any, key_field: str, key_name: str) -> dict[str, Any]:
    calls = int(row.calls or 0)
    ok_calls = int(row.ok_calls or 0)
    return {
        key_name: getattr(row, key_field),
        "calls": calls,
        "ok_calls": ok_calls,
        "failed_calls": int(row.failed_calls or 0),
        # Rounded to one decimal: this is a health signal, not an SLO.
        "success_rate": round(ok_calls * 100.0 / calls, 1) if calls else None,
        "input_tokens": int(row.input_tokens or 0),
        "output_tokens": int(row.output_tokens or 0),
        "total_tokens": int(row.input_tokens or 0) + int(row.output_tokens or 0),
        "calls_without_token_report": int(row.calls_without_token_report or 0),
    }


async def overview(db: AsyncSession, *, days: int | None = None) -> dict[str, Any]:
    """Header totals plus the three cuts."""
    window = resolve_window(days)
    # Lower bound only. An upper bound of "now" cannot exclude a real
    # row — a call cannot be recorded in the future — but it CAN
    # exclude a real row by accident: MySQL rounds ``DATETIME`` to whole
    # seconds, so a call recorded at 12:00:00.6 is stored as 12:00:01
    # and falls outside an ``until`` of 12:00:00.8. That made this read
    # drop the most recent rows at random, which is the worst possible
    # place for a silent gap. ``window.until`` still labels the window
    # in the payload; it just does not filter.
    bounds = (LlmCallLedger.called_at >= window.since,)

    totals_row = (
        await db.execute(select(*_totals_columns()).where(*bounds))
    ).one()

    by_purpose = [
        _row_to_dict(row, "purpose", "purpose")
        for row in (
            await db.execute(
                select(LlmCallLedger.purpose, *_totals_columns())
                .where(*bounds)
                .group_by(LlmCallLedger.purpose)
                # Biggest spender first — the ordering the question
                # implies. Ties break on name so the list is stable
                # across refreshes.
                .order_by(
                    (
                        func.coalesce(func.sum(LlmCallLedger.input_tokens), 0)
                        + func.coalesce(func.sum(LlmCallLedger.output_tokens), 0)
                    ).desc(),
                    LlmCallLedger.purpose.asc(),
                ),
            )
        ).all()
    ]

    by_model = [
        _row_to_dict(row, "model", "model")
        for row in (
            await db.execute(
                select(LlmCallLedger.model, *_totals_columns())
                .where(*bounds)
                .group_by(LlmCallLedger.model)
                .order_by(
                    (
                        func.coalesce(func.sum(LlmCallLedger.input_tokens), 0)
                        + func.coalesce(func.sum(LlmCallLedger.output_tokens), 0)
                    ).desc(),
                    LlmCallLedger.model.asc(),
                ),
            )
        ).all()
    ]

    day = func.date(LlmCallLedger.called_at).label("day")
    by_day = [
        {
            "day": str(row.day),
            "calls": int(row.calls or 0),
            "failed_calls": int(row.failed_calls or 0),
            "total_tokens": int(row.input_tokens or 0) + int(row.output_tokens or 0),
        }
        for row in (
            await db.execute(
                select(day, *_totals_columns())
                .where(*bounds)
                .group_by(day)
                .order_by(day.asc()),
            )
        ).all()
    ]

    # The failure cut is what an admin opens this page for when
    # something is broken: an expired key shows up here as a wall of
    # one error code long before anyone reports the feature is down.
    top_errors = [
        {
            "error_code": row.error_code,
            "purpose": row.purpose,
            "calls": int(row.calls or 0),
        }
        for row in (
            await db.execute(
                select(
                    LlmCallLedger.error_code,
                    LlmCallLedger.purpose,
                    func.count().label("calls"),
                )
                .where(*bounds, LlmCallLedger.ok.is_(False))
                .group_by(LlmCallLedger.error_code, LlmCallLedger.purpose)
                .order_by(func.count().desc())
                .limit(10),
            )
        ).all()
    ]

    totals = _row_to_dict(totals_row, "calls", "_discard")
    totals.pop("_discard", None)
    return {
        "window_days": window.days,
        "since": window.since.isoformat(),
        "until": window.until.isoformat(),
        "totals": totals,
        "by_purpose": by_purpose,
        "by_model": by_model,
        "by_day": by_day,
        "top_errors": top_errors,
        # Said in the payload rather than only in the UI, because MCP
        # clients read this too and would otherwise be free to multiply
        # tokens by a price they invented.
        "money_note": (
            "Token counts only. This platform has no per-model price "
            "table; the vendor bill in ai-spending remains the money "
            "truth. Joining the two is a proportional-split assumption."
        ),
    }


__all__ = ["DEFAULT_WINDOW_DAYS", "MAX_WINDOW_DAYS", "Window", "overview", "resolve_window"]