"""Finance ledger read surface — summary / pivot / timeseries / detail /
movers / completeness / digest.

Everything reads ONLY the materialised ``finance_spend_facts`` (+ close /
restatement / fx rows the finance domain owns). Cross-domain reconciliation
lives in the coordinator; this module never touches another domain's
tables except for the owner display-name resolve (identity), which mirrors
how every other dashboard resolves person labels.

Dimension vocabulary (the pivot/timeseries ``group_by`` axes):
``cost_type``, ``basis``, ``department``, ``cost_center``, ``vendor``,
``sub_category``, ``owner``, ``allocation_source``.

The "未分摊" bucket is first-class: a NULL ``department`` / ``cost_center``
groups under the explicit label ``__unallocated__`` so finance can see —
and drill into — exactly how much spend isn't attributed, rather than
having it silently dropped from a filter.
"""
from __future__ import annotations

from decimal import ROUND_HALF_UP, Decimal

from sqlalchemy import func, literal, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import fiscal_year as fy
from app.core.exceptions import AppException
from app.domains.finance.models import (
    BASIS_CASH,
    BASIS_CONSUMPTION,
    BASIS_PROCUREMENT,
    COST_TYPE_AI,
    COST_TYPE_CLOUD,
    COST_TYPE_HARDWARE,
    COST_TYPE_PROCUREMENT,
    COST_TYPE_SOFTWARE,
    EXPENSE_COST_TYPE,
    FinanceGLActual,
    FinanceNetSuiteAccount,
    FinanceNetSuiteAccountingPeriod,
    FinanceNetSuiteSubsidiary,
    FinancePeriodClose,
    FinanceRestatement,
    FinanceSpendFact,
)

# The four consumption axes the BU fiscal-year view reports, in display
# order. ``procurement`` / ``expense`` are different bases (cash /
# external) and are NOT part of the consumption roll-up.
BU_AXES = (COST_TYPE_CLOUD, COST_TYPE_SOFTWARE, COST_TYPE_AI, COST_TYPE_HARDWARE)

from app.core.safe_logging import get_safe_logger

logger = get_safe_logger(__name__)

_USD = Decimal("0.01")
UNALLOCATED = "__unallocated__"
# The label ``pivot`` renders for a NULL on a dimension whose NULL is not
# an "unattributed" bucket (vendor / sub_category / allocation_source /
# cost_type / basis). Both this and ``UNALLOCATED`` are SENTINELS on the
# way back in: a drill from such a cell sends the sentinel and
# ``_apply_filters`` turns it into an ``IS NULL`` predicate. Matching them
# literally returns zero rows (#2227) — no ledger row's vendor is the
# five-character string "(unknown)".
UNKNOWN_LABEL = "(unknown)"

# group_by dimension → ledger column.
_DIM_COLUMNS = {
    "cost_type": FinanceSpendFact.cost_type,
    "basis": FinanceSpendFact.basis,
    "department": FinanceSpendFact.department,
    "cost_center": FinanceSpendFact.cost_center,
    "vendor": FinanceSpendFact.vendor,
    "sub_category": FinanceSpendFact.sub_category,
    "owner": FinanceSpendFact.owner_person_id,
    "allocation_source": FinanceSpendFact.allocation_source,
}
# Dimensions whose NULL means "unattributed" → folded into the explicit
# unallocated bucket rather than a NULL group.
_UNALLOCATED_DIMS = {"department", "cost_center", "owner"}


def _q2(x) -> Decimal:
    return Decimal(str(x or 0)).quantize(_USD, rounding=ROUND_HALF_UP)


def _shift_period(period: str, months_back: int) -> str:
    year, month = (int(x) for x in period.split("-"))
    idx = year * 12 + (month - 1) - months_back
    y, m = divmod(idx, 12)
    return f"{y:04d}-{m + 1:02d}"


# Widest month span any range-capable read accepts. 36 months = 3 fiscal
# years; the facts table is already monthly-aggregated so the scan stays
# bounded, but an unbounded range would let one request walk the whole table.
RANGE_MAX_MONTHS = 36


def enumerate_periods(start: str, end: str) -> list[str]:
    """Inclusive ascending ``YYYY-MM`` list between the two endpoints.

    Endpoints may arrive in either order (``YYYY-MM`` strings sort
    chronologically). Shared by every range-capable read (pivot / detail /
    timeseries / export / vendor picker) so they can never disagree on
    which months a range means.

    Validates the format itself rather than trusting the caller: a
    malformed endpoint ("2026-1") would never equal ``hi`` and would walk
    the loop to the cap, surfacing as a confusing RANGE_TOO_WIDE instead
    of "you passed a bad month". Every HTTP/MCP caller already validates —
    this keeps the shared helper safe for the next one that forgets.
    """
    from app.domains.finance.service import validate_period

    start, end = validate_period(start), validate_period(end)
    lo, hi = sorted([start, end])
    periods: list[str] = []
    p = lo
    for _ in range(RANGE_MAX_MONTHS):
        periods.append(p)
        if p == hi:
            return periods
        p = _shift_period(p, -1)  # months_back=-1 steps FORWARD one month
    raise AppException(
        code="RANGE_TOO_WIDE",
        message=f"range capped at {RANGE_MAX_MONTHS} months ({lo} → {hi})",
        status_code=400,
    )


async def provisional_periods(db: AsyncSession, periods: list[str]) -> list[str]:
    """Which of ``periods`` are NOT closed yet.

    A range that mixes closed and open months mixes final numbers with
    numbers that can still move (a later rollup re-run restates them), so
    every range read reports the open ones instead of quietly blending
    the two — the range analogue of ``get_summary``'s ``is_provisional``.
    """
    closed = set(
        (
            await db.execute(
                select(FinancePeriodClose.period).where(
                    FinancePeriodClose.period.in_(periods),
                    FinancePeriodClose.status == "closed",
                )
            )
        ).scalars().all()
    )
    return [p for p in periods if p not in closed]


def _validate_dims(group_by: list[str]) -> list[str]:
    bad = [d for d in group_by if d not in _DIM_COLUMNS]
    if bad:
        raise AppException(
            code="BAD_GROUP_BY",
            message=f"unknown group_by dimension(s): {bad}; "
                    f"supported: {sorted(_DIM_COLUMNS)}",
            status_code=400,
        )
    return group_by


# ── Invoice-measure confidentiality (DECISIONS 2026-07-29 #1) ────────
#
# ``finance_spend_facts`` materialises cloud rows straight off
# ``cmdb_cost_facts.amount_usd`` — i.e. INVOICE (post-discount) money.
# The red line is "no reader outside the confidentiality circle may
# obtain any invoice-basis figure from any surface", because a
# post-discount total divided by the list-basis total for the same
# scope (which those readers CAN see, via /cost/breakdown and
# /cmdb/billing/monthly) reconstructs the company's cloud discount rate.
#
# ``/finance/spend/reconcile/internal`` was narrowed for exactly this
# reason on 2026-07-29 — but the same ledger has eight other direct read
# surfaces at a FINER grain (pivot / detail / export go down to
# provider × service × month), and they all kept ``can_view_finance``,
# which admits auditor + management (#2121).
#
# The fix here is measure routing, not a blanket narrowing: an
# out-of-circle reader keeps every non-cloud cost type (hardware,
# software, AI — AI is usage-billed and carries no discount, so it is
# explicitly exempt) and loses only the cloud rows. Withholding is
# always DECLARED in the payload (``withheld_cost_types``); a number
# that silently changes meaning by reader is worse than one that is
# openly incomplete.
WITHHELD_FROM_OUTSIDE_CIRCLE: tuple[str, ...] = (COST_TYPE_CLOUD,)


def _withhold(stmt, withhold_cloud: bool):
    """Drop invoice-basis cloud rows for out-of-circle readers."""
    if not withhold_cloud:
        return stmt
    return stmt.where(
        FinanceSpendFact.cost_type.not_in(WITHHELD_FROM_OUTSIDE_CIRCLE)
    )


def withheld_meta(withhold_cloud: bool) -> list[str]:
    """The declaration that rides along with every narrowed payload."""
    return list(WITHHELD_FROM_OUTSIDE_CIRCLE) if withhold_cloud else []


# A ``vendor`` IN-list is unbounded on the wire (repeatable query param
# over HTTP, a JSON array over MCP). Not a security issue — it is a
# self-inflicted slow query — but an unbounded IN against the ledger is
# worth one explicit 400 rather than a mystery timeout (#2189-1). Far
# above the picker's own 500-vendor ceiling, so no real UI hits it.
#
# #2238: the cap lives HERE, in the shared filter layer, not on the HTTP
# router helper. It used to sit in ``finance/router.py::_check_vendor_cap``
# (reached only from ``_filters_from_query``), while MCP's
# ``_finance_vendor_filter`` built the same filter dict by hand — so the
# one surface the cap's own comment named as unbounded ("a JSON array over
# MCP") was the one it did not cover.
VENDOR_FILTER_MAX = 500


def enforce_vendor_cap(vendor: object) -> None:
    """Reject an over-long vendor IN-list on ANY surface (HTTP / MCP)."""
    if isinstance(vendor, (list, tuple, set)) and len(vendor) > VENDOR_FILTER_MAX:
        raise AppException(
            code="TOO_MANY_VENDORS",
            message=(
                f"at most {VENDOR_FILTER_MAX} vendors per query "
                f"(got {len(vendor)})"
            ),
            status_code=400,
        )


def _sentinel_predicate(col, value: str, *, unallocated_dim: bool):
    """Turn a pivot label sentinel back into the predicate it came from.

    ``pivot`` renders NULLs as ``__unallocated__`` (department /
    cost_center / owner, where NULL means unattributed) or ``(unknown)``
    (everything else). Drilling such a cell has to reproduce EXACTLY that
    group — matching the label literally yields 0 rows, and dropping the
    constraint yields a superset spanning every other value of the
    dimension (#2227). Both directions break the drill invariant that the
    drawer's rows add up to the cell that was clicked.
    """
    if unallocated_dim:
        # Mirrors ``coalesce(nullif(col, ''), UNALLOCATED)``.
        return or_(col.is_(None), col == "")
    return col.is_(None)


def _apply_filters(stmt, filters: dict):
    """Apply the common filter set to a finance_spend_facts query.

    Supported filters: ``cost_type`` (list), ``basis`` (str),
    ``vendor`` (str), ``cost_center`` (str), ``department`` (subtree
    list, resolved by the caller), ``owner`` (person id).
    """
    enforce_vendor_cap(filters.get("vendor"))
    if filters.get("withhold_cloud"):
        # Reader is outside the invoice-measure confidentiality circle —
        # see WITHHELD_FROM_OUTSIDE_CIRCLE above. Applied first and
        # unconditionally: an explicit ``cost_type=cloud`` from such a
        # reader intersects to an empty set rather than overriding this.
        stmt = _withhold(stmt, True)
    cost_types = filters.get("cost_type")
    if cost_types:
        stmt = stmt.where(FinanceSpendFact.cost_type.in_(cost_types))
    basis = filters.get("basis")
    if basis:
        stmt = stmt.where(FinanceSpendFact.basis == basis)
    elif filters.get("exclude_cash"):
        # Drop the overlapping hardware purchase-cash lens so a no-basis query
        # can't double-count hardware against its own depreciation.
        stmt = stmt.where(FinanceSpendFact.basis != BASIS_CASH)
    vendor = filters.get("vendor")
    if vendor:
        # A list (multi-select: "these three vendors") or a single value.
        # Exact match either way — the picker feeds real ledger values, so
        # there is no partial-name guessing to support.
        vendor_list = (
            list(vendor) if isinstance(vendor, (list, tuple, set)) else [vendor]
        )
        # A "(unknown)" cell drills to the NULL-vendor group (#2227). Mixed
        # selections (two real vendors + the unknown bucket) stay one OR.
        real = [v for v in vendor_list if v != UNKNOWN_LABEL]
        preds = []
        if real:
            preds.append(FinanceSpendFact.vendor.in_(real))
        if len(real) != len(vendor_list):
            preds.append(FinanceSpendFact.vendor.is_(None))
        stmt = stmt.where(or_(*preds) if len(preds) > 1 else preds[0])
    cost_center = filters.get("cost_center")
    if cost_center:
        stmt = stmt.where(
            _sentinel_predicate(
                FinanceSpendFact.cost_center, cost_center, unallocated_dim=True,
            )
            if cost_center == UNALLOCATED
            else FinanceSpendFact.cost_center == cost_center
        )
    owner = filters.get("owner")
    if owner:
        stmt = stmt.where(
            _sentinel_predicate(
                FinanceSpendFact.owner_person_id, owner, unallocated_dim=True,
            )
            if owner == UNALLOCATED
            else FinanceSpendFact.owner_person_id == owner
        )
    # Both are valid ``group_by`` dimensions, so a drill from a cell
    # grouped by either one has to be able to narrow by it — otherwise
    # the drawer returns the slice WITHOUT that constraint and its total
    # cannot match the row that opened it (#2148).
    sub_category = filters.get("sub_category")
    if sub_category:
        stmt = stmt.where(
            _sentinel_predicate(
                FinanceSpendFact.sub_category, sub_category,
                unallocated_dim=False,
            )
            if sub_category == UNKNOWN_LABEL
            else FinanceSpendFact.sub_category == sub_category
        )
    allocation_source = filters.get("allocation_source")
    if allocation_source:
        stmt = stmt.where(
            _sentinel_predicate(
                FinanceSpendFact.allocation_source, allocation_source,
                unallocated_dim=False,
            )
            if allocation_source == UNKNOWN_LABEL
            else FinanceSpendFact.allocation_source == allocation_source
        )
    # ``department`` is the one dimension whose filter is pre-resolved to a
    # subtree name list by the caller; the unallocated sentinel therefore
    # rides its own key (``department_unallocated``) rather than the name
    # list — see ``_resolve_department_filter``.
    if filters.get("department_unallocated"):
        stmt = stmt.where(
            _sentinel_predicate(
                FinanceSpendFact.department, UNALLOCATED, unallocated_dim=True,
            )
        )
    dept_names = filters.get("department_names")
    if dept_names:
        stmt = stmt.where(FinanceSpendFact.department.in_(dept_names))
    return stmt


async def _is_provisional(db: AsyncSession, period: str) -> bool:
    row = (
        await db.execute(
            select(FinancePeriodClose.status).where(
                FinancePeriodClose.period == period
            )
        )
    ).scalar_one_or_none()
    # No close row, or status != closed → provisional.
    return row != "closed"


async def _resolve_department_filter(
    db: AsyncSession, department: str | None,
) -> list[str] | None:
    """Resolve a department to its subtree names (parent rolls up).

    Mirrors every other department filter on the platform via
    ``app.core.org_scope.resolve_department_subtree_names``.

    The ``__unallocated__`` sentinel (a drill from the unallocated pivot
    cell, #2227) is NOT a department name — it is handled as an IS NULL
    predicate in ``_apply_filters``, so resolve it to no name list.
    """
    if not department or department == UNALLOCATED:
        return None
    from app.core.org_scope import resolve_department_subtree_names

    names = await resolve_department_subtree_names(db, department)
    return list(names) if names else [department]


async def _resolve_department_into(db: AsyncSession, filters: dict) -> None:
    """Populate the department-derived filter keys in place.

    Two keys, because a department filter has two shapes: a real name
    (resolved to its subtree name list) and the ``__unallocated__``
    sentinel from a pivot cell, which is an IS NULL predicate rather than
    a name (#2227). Every read that filters by department goes through
    here so the two can never drift apart.
    """
    department = filters.get("department")
    filters["department_unallocated"] = department == UNALLOCATED
    filters["department_names"] = await _resolve_department_filter(
        db, department,
    )


# ─── Summary (KPI hero) ─────────────────────────────────────────


async def get_summary(
    db: AsyncSession, period: str, *, withhold_cloud: bool = False,
) -> dict:
    """Top-of-page KPIs for a period.

    Consumption (cloud+software+ai+hardware-depreciation) and procurement
    are reported as SEPARATE totals — never summed — because they are
    different bases on the same money. The hardware CASH lens (purchase
    outflow in the buy month) is a THIRD basis and is excluded from both
    headline totals: lumping it into consumption double-counts the asset
    against its own depreciation (the platform invariant — never sum across
    bases). ``unallocated_usd`` surfaces the spend with no department.
    """
    prev = _shift_period(period, 1)

    # Per-(basis, cost_type) totals in one pass.
    rows = (
        await db.execute(
            _withhold(
                select(
                    FinanceSpendFact.basis,
                    FinanceSpendFact.cost_type,
                    func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0),
                )
                .where(FinanceSpendFact.period == period),
                withhold_cloud,
            ).group_by(FinanceSpendFact.basis, FinanceSpendFact.cost_type)
        )
    ).all()

    by_cost_type: dict[str, Decimal] = {}
    consumption_total = Decimal("0")
    procurement_total = Decimal("0")
    for basis, cost_type, usd in rows:
        amt = _q2(usd)
        if basis == BASIS_PROCUREMENT:
            procurement_total += amt
        elif basis == BASIS_CONSUMPTION:
            # Consumption basis only — the cash lens (hardware purchase) is a
            # separate view (surfaced via the BU cash toggle), never added
            # into the consumption headline or its cost-type mix.
            by_cost_type[cost_type] = by_cost_type.get(cost_type, Decimal("0")) + amt
            consumption_total += amt

    # Unallocated (consumption basis, no department).
    unalloc = _q2((
        await db.execute(
            _withhold(
                select(
                    func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0)
                ).where(
                    FinanceSpendFact.period == period,
                    FinanceSpendFact.basis == BASIS_CONSUMPTION,
                    or_(
                        FinanceSpendFact.department.is_(None),
                        FinanceSpendFact.department == "",
                    ),
                ),
                withhold_cloud,
            )
        )
    ).scalar())

    prev_consumption = _q2((
        await db.execute(
            _withhold(
                select(
                    func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0)
                ).where(
                    FinanceSpendFact.period == prev,
                    FinanceSpendFact.basis == BASIS_CONSUMPTION,
                ),
                withhold_cloud,
            )
        )
    ).scalar())

    mom_pct = None
    if prev_consumption > 0:
        mom_pct = round(
            float((consumption_total - prev_consumption) / prev_consumption * 100), 2
        )

    ct_list = [
        {
            "cost_type": ct,
            "usd": str(amt),
            "pct": (
                round(float(amt / consumption_total * 100), 2)
                if consumption_total > 0 and ct != "procurement" else None
            ),
        }
        for ct, amt in sorted(by_cost_type.items(), key=lambda kv: -kv[1])
    ]

    return {
        "period": period,
        "is_provisional": await _is_provisional(db, period),
        "consumption_total_usd": str(_q2(consumption_total)),
        "procurement_total_usd": str(_q2(procurement_total)),
        "by_cost_type": ct_list,
        "unallocated_usd": str(unalloc),
        "unallocated_pct": (
            round(float(unalloc / consumption_total * 100), 2)
            if consumption_total > 0 else 0.0
        ),
        "prev_period": prev,
        "prev_consumption_usd": str(prev_consumption),
        "mom_pct": mom_pct,
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }


# ─── Pivot (the explorer) ───────────────────────────────────────


async def pivot(
    db: AsyncSession, periods: str | list[str], group_by: list[str], filters: dict,
) -> dict:
    """Group the facts of one or MORE months by 1–3 dimensions, summing USD.

    Returns rows with the dimension key columns + ``usd`` + ``row_count``,
    sorted by USD desc, plus the grand total. NULL values on
    unallocated-eligible dims fold into the ``__unallocated__`` label so
    they stay visible.

    ``periods`` is the inclusive month list (one entry = the old
    single-month behaviour). Summing months is safe here in a way it is
    NOT on the software-billing surface: every fact is already
    USD-normalised at its own month's frozen FX rate, so a range total is
    a real number rather than a cross-currency mash. What a range CAN mix
    is closed and open months — ``provisional_periods`` in the response
    names the open ones.
    """
    # A bare "2026-05" is accepted as a one-month range — several callers
    # (and every pre-range test) pass the single period string, and
    # ``period.in_("2026-05")`` would otherwise iterate the characters.
    periods = [periods] if isinstance(periods, str) else list(periods)
    group_by = _validate_dims(group_by or ["cost_type"])
    if len(group_by) > 3:
        raise AppException(
            code="TOO_MANY_DIMS",
            message="at most 3 group_by dimensions",
            status_code=400,
        )
    filters = dict(filters or {})
    await _resolve_department_into(db, filters)
    # Exclude the hardware CASH lens unless the caller explicitly asked for a
    # basis or is splitting BY basis. Cash (purchase outflow in the buy month)
    # is the one basis that OVERLAPS another — hardware also booked as
    # consumption depreciation — so a default pivot that includes it sums a
    # hardware asset against its own depreciation. Consumption and procurement
    # don't share cost_types, so they stay visible (the documented behaviour);
    # only the double-counting cash rows are dropped. Matches get_summary,
    # which also excludes cash from its headline totals.
    if not filters.get("basis") and "basis" not in group_by:
        filters["exclude_cash"] = True

    # Build the SELECT: dimension columns (NULL→unallocated label for
    # the eligible dims), USD sum, row count.
    select_cols = []
    group_cols = []
    for dim in group_by:
        col = _DIM_COLUMNS[dim]
        if dim in _UNALLOCATED_DIMS:
            expr = func.coalesce(
                func.nullif(col, literal("")), literal(UNALLOCATED)
            ).label(dim)
        else:
            expr = func.coalesce(col, literal(UNKNOWN_LABEL)).label(dim)
        select_cols.append(expr)
        group_cols.append(expr)

    stmt = (
        select(
            *select_cols,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
            func.count().label("row_count"),
        )
        .where(FinanceSpendFact.period.in_(periods))
        .group_by(*group_cols)
    )
    stmt = _apply_filters(stmt, filters)
    rows = (await db.execute(stmt)).all()

    out = []
    total = Decimal("0")
    owner_ids: set[str] = set()
    for r in rows:
        d = r._mapping
        usd = _q2(d["usd"])
        total += usd
        entry = {dim: d[dim] for dim in group_by}
        entry["usd"] = str(usd)
        entry["row_count"] = int(d["row_count"])
        if "owner" in group_by and entry["owner"] not in (UNALLOCATED, None):
            owner_ids.add(entry["owner"])
        out.append(entry)

    # Resolve owner display names if grouping by owner.
    if owner_ids:
        names = await _resolve_person_names(db, owner_ids)
        for entry in out:
            if "owner" in entry:
                entry["owner_name"] = names.get(entry["owner"])

    out.sort(key=lambda e: Decimal(e["usd"]), reverse=True)
    return {
        # ``period`` stays for back-compat with single-month callers (it is
        # the range's last month); ``periods``/``start``/``end`` are the
        # range-aware fields.
        "period": periods[-1],
        "periods": periods,
        "start": periods[0],
        "end": periods[-1],
        "provisional_periods": await provisional_periods(db, periods),
        "group_by": group_by,
        "rows": out,
        "total_usd": str(_q2(total)),
    }


# ─── Timeseries (trend) ─────────────────────────────────────────


async def timeseries(
    db: AsyncSession, start_period: str, end_period: str,
    group_by: str | None, filters: dict,
) -> dict:
    """Monthly USD series from ``start_period`` to ``end_period`` inclusive.

    Optionally split by a single dimension (``group_by``). Zero-fills
    months with no spend so the trend line is continuous.
    """
    if group_by:
        _validate_dims([group_by])
    # Inclusive month range, endpoints in either order, capped — shared with
    # pivot / detail / export / vendor picker so the four can never disagree
    # on what a range means.
    periods = enumerate_periods(start_period, end_period)
    filters = dict(filters or {})
    await _resolve_department_into(db, filters)
    # Exclude the overlapping hardware CASH lens unless explicitly requested
    # or split BY basis, so the trend line never sums a hardware asset against
    # its own depreciation (see pivot / get_summary for the same invariant).
    if not filters.get("basis") and group_by != "basis":
        filters["exclude_cash"] = True

    cols = [FinanceSpendFact.period.label("period")]
    grp = [FinanceSpendFact.period]
    if group_by:
        gcol = _DIM_COLUMNS[group_by]
        if group_by in _UNALLOCATED_DIMS:
            gexpr = func.coalesce(
                func.nullif(gcol, literal("")), literal(UNALLOCATED)
            ).label("key")
        else:
            gexpr = func.coalesce(gcol, literal(UNKNOWN_LABEL)).label("key")
        cols.append(gexpr)
        grp.append(gexpr)

    stmt = (
        select(*cols, func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"))
        .where(FinanceSpendFact.period.in_(periods))
        .group_by(*grp)
    )
    stmt = _apply_filters(stmt, filters)
    rows = (await db.execute(stmt)).all()

    if not group_by:
        by_period = {pp: Decimal("0") for pp in periods}
        for r in rows:
            by_period[r._mapping["period"]] = _q2(r._mapping["usd"])
        return {
            "periods": periods,
            "series": [{"period": pp, "usd": str(by_period[pp])} for pp in periods],
        }

    # Split series: {key: {period: usd}}
    keys: set[str] = set()
    grid: dict[str, dict[str, Decimal]] = {}
    for r in rows:
        m = r._mapping
        keys.add(m["key"])
        grid.setdefault(m["key"], {})[m["period"]] = _q2(m["usd"])
    series = [
        {
            "key": k,
            "points": [
                {"period": pp, "usd": str(grid.get(k, {}).get(pp, Decimal("0")))}
                for pp in periods
            ],
        }
        for k in sorted(keys)
    ]
    return {"periods": periods, "group_by": group_by, "series": series}


# ─── Detail (drill) ─────────────────────────────────────────────


async def detail(
    db: AsyncSession, periods: str | list[str], filters: dict,
    *, limit: int = 200, offset: int = 0,
) -> dict:
    """Raw ledger rows for a month range + filters (drill from any aggregate).

    Each row carries the full provenance (fx, source_domain, source_ref)
    so the UI can show exactly how a number was derived. ``periods`` is the
    inclusive month list — one entry = the old single-month drill; the row
    payload carries its own ``period`` so a range drill stays readable.
    """
    # Single period string accepted as a one-month range — see ``pivot``.
    periods = [periods] if isinstance(periods, str) else list(periods)
    filters = dict(filters or {})
    await _resolve_department_into(db, filters)
    # Same cash-lens exclusion as ``pivot`` / ``timeseries`` /
    # ``get_summary`` (#2155). This is a DRILL: the rows it returns must
    # add up to the aggregate cell the user clicked. Without the rule,
    # clicking a "hardware" cell showing depreciation-basis $X opened a
    # drawer whose banner count and whose "Pivot-ready" xlsx export both
    # added the purchase-CASH rows back — the hardware asset summed
    # against its own depreciation, which is the exact double-count
    # ``pivot``'s docstring exists to prevent, and the wider the range
    # the wider the gap.
    #
    # ``group_by_has_basis`` (#2237) is the third case ``pivot`` already
    # honours and this side did not: when the user SPLITS BY basis, the
    # pivot keeps the hardware-CASH rows (split apart, the overlap is
    # visible rather than double-counted) — so the export button, which
    # only ships the filter bar and has no group_by of its own, produced
    # an xlsx WITHOUT the cash rows the table on screen was showing.
    # Screen total ≠ file total, on a feature whose whole premise is
    # "the export IS that query".
    if not filters.get("basis") and not filters.get("group_by_has_basis"):
        filters["exclude_cash"] = True
    base = select(FinanceSpendFact).where(FinanceSpendFact.period.in_(periods))
    base = _apply_filters(base, filters)
    total = (await db.execute(
        select(func.count()).select_from(base.subquery())
    )).scalar() or 0
    rows = (await db.execute(
        # ``id`` tiebreaker: without it, rows with equal ``amount_usd``
        # have no defined order, so an OFFSET page can repeat or skip
        # rows. Ties were rare on a single month and became routine once
        # the drill went range-capable — same class of fix as
        # ``top_resources``'s (integration_id, resource_id) tiebreaker.
        base.order_by(
            FinanceSpendFact.amount_usd.desc(), FinanceSpendFact.id.asc(),
        ).limit(limit).offset(offset)
    )).scalars().all()

    owner_ids = {r.owner_person_id for r in rows if r.owner_person_id}
    names = await _resolve_person_names(db, owner_ids)

    data = [
        {
            "period": r.period,
            "cost_type": r.cost_type,
            "basis": r.basis,
            "sub_category": r.sub_category,
            "vendor": r.vendor,
            "department": r.department,
            "cost_center": r.cost_center,
            "owner_person_id": r.owner_person_id,
            "owner_name": names.get(r.owner_person_id) if r.owner_person_id else None,
            "allocation_source": r.allocation_source,
            "amount_native": str(r.amount_native),
            "currency_native": r.currency_native,
            "amount_usd": str(r.amount_usd),
            "fx_rate_used": str(r.fx_rate_used) if r.fx_rate_used is not None else None,
            "fx_basis": r.fx_basis,
            "is_provisional": r.is_provisional,
            "source_domain": r.source_domain,
            "source_ref": r.source_ref,
        }
        for r in rows
    ]
    return {"period": periods[-1], "periods": periods, "total": int(total),
            "limit": limit, "offset": offset, "rows": data}


# ─── Vendor picker (range-aware) ─────────────────────────────────

# The picker lists every vendor that actually spent in the range. Capped so
# one request can't return an unbounded option list; the cap is REPORTED
# (``truncated``) rather than silently trimming the tail off a filter UI.
VENDOR_PICKER_LIMIT = 500


async def vendors_in_range(
    db: AsyncSession, periods: str | list[str], filters: dict,
) -> dict:
    """Distinct vendors with USD totals over a month range, biggest first.

    Feeds the explorer's vendor multi-select. Honours the same filters as
    the pivot (minus ``vendor`` itself — the option list must not be
    narrowed by the selection it feeds), so picking ``cost_type=software``
    lists exactly the software vendors that billed in the range.
    """
    periods = [periods] if isinstance(periods, str) else list(periods)
    filters = dict(filters or {})
    filters.pop("vendor", None)
    await _resolve_department_into(db, filters)
    if not filters.get("basis"):
        filters["exclude_cash"] = True
    stmt = (
        select(
            FinanceSpendFact.vendor,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.period.in_(periods),
            FinanceSpendFact.vendor.isnot(None),
        )
        .group_by(FinanceSpendFact.vendor)
        .order_by(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).desc())
        .limit(VENDOR_PICKER_LIMIT + 1)
    )
    rows = (await db.execute(_apply_filters(stmt, filters))).all()
    truncated = len(rows) > VENDOR_PICKER_LIMIT
    if truncated:
        rows = rows[:VENDOR_PICKER_LIMIT]
        logger.warning(
            "finance.vendors_in_range: option list truncated at %d for %s→%s",
            VENDOR_PICKER_LIMIT, periods[0], periods[-1],
        )
    return {
        "start": periods[0],
        "end": periods[-1],
        "vendors": [
            {
                "vendor": r._mapping["vendor"],
                "amount_usd": str(_q2(r._mapping["usd"])),
            }
            for r in rows
        ],
        "truncated": truncated,
    }


# ─── Movers (MoM change) ────────────────────────────────────────


async def movers(
    db: AsyncSession, period: str, dim: str, *, top: int = 10,
    withhold_cloud: bool = False,
) -> dict:
    """Top USD increases / decreases vs the previous month, by dimension.

    Answers "what changed / is it good?" — the variance lens. Compares
    the period's per-key USD against the prior month's; returns the
    largest absolute movers split into up / down.
    """
    _validate_dims([dim])
    prev = _shift_period(period, 1)
    col = _DIM_COLUMNS[dim]
    if dim in _UNALLOCATED_DIMS:
        kexpr = func.coalesce(func.nullif(col, literal("")), literal(UNALLOCATED))
    else:
        kexpr = func.coalesce(col, literal(UNKNOWN_LABEL))

    async def _totals(p):
        rows = (await db.execute(
            _withhold(
                select(
                    kexpr.label("k"),
                    func.coalesce(
                        func.sum(FinanceSpendFact.amount_usd), 0,
                    ).label("usd"),
                )
                .where(FinanceSpendFact.period == p,
                       FinanceSpendFact.basis == BASIS_CONSUMPTION),
                withhold_cloud,
            ).group_by(kexpr)
        )).all()
        return {r._mapping["k"]: _q2(r._mapping["usd"]) for r in rows}

    cur = await _totals(period)
    prv = await _totals(prev)
    keys = set(cur) | set(prv)
    deltas = []
    for k in keys:
        c = cur.get(k, Decimal("0"))
        p = prv.get(k, Decimal("0"))
        d = _q2(c - p)
        if d == 0:
            continue
        deltas.append({
            "key": k,
            "cur_usd": str(c),
            "prev_usd": str(p),
            "delta_usd": str(d),
            "delta_pct": (round(float((c - p) / p * 100), 2) if p > 0 else None),
        })
    ups = sorted([d for d in deltas if Decimal(d["delta_usd"]) > 0],
                 key=lambda d: Decimal(d["delta_usd"]), reverse=True)[:top]
    downs = sorted([d for d in deltas if Decimal(d["delta_usd"]) < 0],
                   key=lambda d: Decimal(d["delta_usd"]))[:top]
    return {"period": period, "prev_period": prev, "dim": dim,
            "increases": ups, "decreases": downs}


# ─── Completeness / trust panel ─────────────────────────────────


async def completeness(
    db: AsyncSession, period: str, *, withhold_cloud: bool = False,
) -> dict:
    """Per-source freshness + the period's reconciliation + restatements.

    The trust panel: tells finance whether this period's number can be
    relied on (sources landed, ledger reconciles, no open restatement)
    before they sign off.
    """
    rows = (await db.execute(
        _withhold(
            select(
                FinanceSpendFact.source_domain,
                func.count().label("rows"),
                func.coalesce(
                    func.sum(FinanceSpendFact.amount_usd), 0,
                ).label("usd"),
                func.max(FinanceSpendFact.computed_at).label("computed_at"),
                func.max(FinanceSpendFact.is_provisional).label("prov"),
            )
            .where(FinanceSpendFact.period == period),
            withhold_cloud,
        ).group_by(FinanceSpendFact.source_domain)
    )).all()
    sources = [
        {
            "source_domain": r._mapping["source_domain"],
            "rows": int(r._mapping["rows"]),
            "usd": str(_q2(r._mapping["usd"])),
            "computed_at": (
                r._mapping["computed_at"].isoformat()
                if r._mapping["computed_at"] else None
            ),
            "provisional": bool(r._mapping["prov"]),
        }
        for r in rows
    ]
    restatements = (await db.execute(
        select(FinanceRestatement).where(FinanceRestatement.period == period)
        .order_by(FinanceRestatement.detected_at.desc())
    )).scalars().all()
    return {
        "period": period,
        "is_provisional": await _is_provisional(db, period),
        "sources": sources,
        "restatements": [
            {
                "prev_usd": str(r.prev_usd), "new_usd": str(r.new_usd),
                "delta_usd": str(r.delta_usd), "reason": r.reason,
                "detected_at": r.detected_at.isoformat() if r.detected_at else None,
            }
            for r in restatements
        ],
    }


# ─── Monthly digest (deterministic narrative) ───────────────────


async def digest(
    db: AsyncSession, period: str, *, withhold_cloud: bool = False,
) -> dict:
    """A deterministic, auditable monthly narrative for finance.

    Composed from the same aggregates the dashboard shows (no LLM, so it
    never drifts from the numbers): total + MoM, cost-type mix, top
    departments, top movers, unallocated, provisional/restatement
    warnings. Returned as structured ``lines`` the UI renders as a card
    and the export embeds.
    """
    s = await get_summary(db, period, withhold_cloud=withhold_cloud)
    top_depts = await pivot(
        db, period, ["department"],
        {"basis": BASIS_CONSUMPTION, "withhold_cloud": withhold_cloud},
    )
    movers_dept = await movers(
        db, period, "department", top=3, withhold_cloud=withhold_cloud,
    )

    consumption = Decimal(s["consumption_total_usd"])
    lines: list[str] = []
    lines.append(
        f"{period} 消费总额 ${consumption:,.2f}"
        + (f"(环比 {s['mom_pct']:+.1f}%)" if s["mom_pct"] is not None else "")
        + ("(本月数据仍在汇总,为临时值)" if s["is_provisional"] else "(已关账)")
    )
    if Decimal(s["procurement_total_usd"]) > 0:
        lines.append(f"采购实际支出 ${Decimal(s['procurement_total_usd']):,.2f}(独立口径,不计入消费总额)")
    mix = "、".join(
        f"{c['cost_type']} ${Decimal(c['usd']):,.0f}"
        + (f"({c['pct']:.0f}%)" if c.get("pct") is not None else "")
        for c in s["by_cost_type"] if c["cost_type"] != "procurement"
    )
    if mix:
        lines.append(f"构成:{mix}")
    top3 = [r for r in top_depts["rows"] if r["department"] != UNALLOCATED][:3]
    if top3:
        lines.append(
            "花费最高部门:"
            + "、".join(f"{r['department']} ${Decimal(r['usd']):,.0f}" for r in top3)
        )
    if movers_dept["increases"]:
        up = movers_dept["increases"][0]
        lines.append(
            f"环比增长最大:{up['key']} +${Decimal(up['delta_usd']):,.0f}"
            + (f"({up['delta_pct']:+.0f}%)" if up.get("delta_pct") is not None else "")
        )
    unalloc = Decimal(s["unallocated_usd"])
    if unalloc > 0:
        lines.append(
            f"⚠ 未分摊到部门:${unalloc:,.2f}({s['unallocated_pct']:.1f}%)——建议补归属规则"
        )
    return {
        "period": period, "lines": lines, "summary": s,
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }


# ─── Helpers ────────────────────────────────────────────────────


async def _resolve_person_names(
    db: AsyncSession, person_ids: set[str],
) -> dict[str, str | None]:
    ids = [p for p in person_ids if p]
    if not ids:
        return {}
    from app.identity.person_models import IdentityPerson

    out: dict[str, str | None] = {}
    for i in range(0, len(ids), 500):
        chunk = ids[i:i + 500]
        rows = (await db.execute(
            select(IdentityPerson.id, IdentityPerson.display_name)
            .where(IdentityPerson.id.in_(chunk))
        )).all()
        for pid, name in rows:
            out[str(pid)] = name
    return out


# ─── External-actuals reconciliation (Phase 2 — vs 汇联易) ───────


def _external_statement_paid_predicate():
    """SQL boolean: is a 汇联易 external statement PAID?

    ``FinanceExternalStatement.payment_status`` is stored VERBATIM from the
    汇联易 ERP (a free ``String(32)``, not normalized at ingest — see
    ``finance_huilianyi_coordinator._clean``), so the ERP emits things like
    ``已支付`` / ``paid`` / ``Paid`` / numeric codes, NOT the ASCII literal
    ``"PAID"``. A bare ``payment_status == "PAID"`` therefore matched almost
    nothing — silently zeroing the paid-count pivot and, worse, the
    ``paid_unverified`` compliance control (paid-but-not-invoice-verified).

    Key primarily off ``paid_at`` — the coordinator populates it from the
    document's payment date, so a non-null value is unambiguous proof of
    payment across every representation — with a case-insensitive
    ``payment_status`` fallback for ASCII rows that lack a parsed date.
    """
    from app.domains.finance.models import FinanceExternalStatement
    return or_(
        FinanceExternalStatement.paid_at.isnot(None),
        func.upper(func.trim(FinanceExternalStatement.payment_status)) == "PAID",
    )


async def external_reconcile(db: AsyncSession, period: str, dim: str = "cost_center") -> dict:
    """Internal computed spend vs external actual (汇联易) for a period.

    By ``dim`` ∈ {cost_center, department, vendor}: internal USD
    (``finance_spend_facts`` EXCLUDING the promoted ``expense`` cost_type
    — i.e. the cloud/software/ai/procurement Helix computes) vs external
    USD (``finance_external_statements`` — actual booked/paid). The two
    are different bases (internal = computed consumption; external =
    reimbursed/invoiced actual), so the diff is an investigative signal,
    not an error — large gaps flag a vendor/cost-centre where the two
    systems disagree. Also rolls up the external chain (docs / invoices
    verified / paid) so finance sees procure→consume→invoice→pay status.
    """
    from app.domains.finance.models import FinanceExternalStatement

    if dim not in ("cost_center", "department", "vendor"):
        raise AppException(code="BAD_DIM", message="dim must be cost_center/department/vendor", status_code=400)
    ext_col = getattr(FinanceExternalStatement, dim)
    int_col = _DIM_COLUMNS[dim]

    def _label(col):
        return func.coalesce(func.nullif(col, literal("")), literal(UNALLOCATED))

    # External side.
    ext_rows = (await db.execute(
        select(
            _label(ext_col).label("k"),
            func.coalesce(func.sum(FinanceExternalStatement.amount_usd), 0).label("usd"),
            func.count().label("docs"),
            func.coalesce(func.sum(func.if_(FinanceExternalStatement.invoice_verified.is_(True), 1, 0)), 0).label("verified"),
            func.coalesce(func.sum(func.if_(_external_statement_paid_predicate(), 1, 0)), 0).label("paid"),
        ).where(FinanceExternalStatement.period == period).group_by(_label(ext_col))
    )).all()
    ext = {r._mapping["k"]: r._mapping for r in ext_rows}

    # Internal side — exclude the promoted 'expense' cost_type (that IS
    # the external data; including it would self-compare).
    from app.domains.finance.models import EXPENSE_COST_TYPE
    int_rows = (await db.execute(
        select(
            _label(int_col).label("k"),
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        ).where(
            FinanceSpendFact.period == period,
            # Consumption basis ONLY. External statements are promoted into
            # the ledger as consumption-basis facts (see finance_spend_
            # coordinator.collect_expense_rows, basis=BASIS_CONSUMPTION), so
            # the external actuals correspond to the consumption lens. Without
            # this filter the internal side also sums procurement (PO
            # commitments) + cash (hardware purchase), producing a bogus
            # diff_usd against external actuals the systems actually agree on.
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
            FinanceSpendFact.cost_type != EXPENSE_COST_TYPE,
        ).group_by(_label(int_col))
    )).all()
    internal = {r._mapping["k"]: _q2(r._mapping["usd"]) for r in int_rows}

    keys = set(ext) | set(internal)
    out = []
    for k in keys:
        e = ext.get(k)
        ext_usd = _q2(e["usd"]) if e else Decimal("0")
        int_usd = internal.get(k, Decimal("0"))
        out.append({
            dim: k,
            "internal_usd": str(int_usd),
            "external_usd": str(ext_usd),
            "diff_usd": str(_q2(int_usd - ext_usd)),
            "ext_docs": int(e["docs"]) if e else 0,
            "ext_invoices_verified": int(e["verified"]) if e else 0,
            "ext_paid": int(e["paid"]) if e else 0,
        })
    out.sort(key=lambda r: abs(Decimal(r["diff_usd"])), reverse=True)
    return {"period": period, "dim": dim, "rows": out}


async def external_statements(
    db: AsyncSession, period: str, *, cost_center: str | None = None,
    vendor: str | None = None, promoted: bool | None = None,
    limit: int = 200, offset: int = 0,
) -> dict:
    """Paginated raw external-actual lines (汇联易) for drill / browse."""
    from app.domains.finance.models import FinanceExternalStatement

    base = select(FinanceExternalStatement).where(FinanceExternalStatement.period == period)
    if cost_center:
        base = base.where(FinanceExternalStatement.cost_center == cost_center)
    if vendor:
        base = base.where(FinanceExternalStatement.vendor == vendor)
    if promoted is not None:
        base = base.where(FinanceExternalStatement.promote_to_ledger.is_(promoted))
    total = (await db.execute(select(func.count()).select_from(base.subquery()))).scalar() or 0
    rows = (await db.execute(
        base.order_by(FinanceExternalStatement.amount_usd.desc()).limit(limit).offset(offset)
    )).scalars().all()
    data = [
        {
            "external_doc_no": r.external_doc_no, "expense_category": r.expense_category,
            "mapped_cost_type": r.mapped_cost_type, "promote_to_ledger": r.promote_to_ledger,
            "cost_center": r.cost_center, "department": r.department, "vendor": r.vendor,
            "amount_native": str(r.amount_native), "currency": r.currency,
            "amount_usd": str(r.amount_usd), "invoice_no": r.invoice_no,
            "invoice_verified": r.invoice_verified, "doc_status": r.doc_status,
            "payment_status": r.payment_status,
        }
        for r in rows
    ]
    return {"period": period, "total": int(total), "limit": limit, "offset": offset, "rows": data}


# ─── Cost-centre authority + expense governance (PR-C) ──────────


async def cost_center_coverage(
    db: AsyncSession, period: str, *, withhold_cloud: bool = False,
) -> dict:
    """Authoritative cost-centre coverage for a period.

    Compares the cost centres that actually carry spend (internal ledger
    + external statements) against the authoritative master mirrored from
    汇联易 (``finance_cost_centers``). Surfaces:
      * **orphan** — spend booked to a cost centre NOT in the master (an
        attribution gap / stale code to fix),
      * **unused** — master cost centres with no spend this period,
      * coverage % of spend that maps to a known master code.
    """
    from app.domains.finance.models import (
        FinanceCostCenter,
        FinanceExternalStatement,
    )

    master_rows = (await db.execute(
        select(
            FinanceCostCenter.code, FinanceCostCenter.name,
            FinanceCostCenter.source, FinanceCostCenter.headcount,
        )
    )).all()
    # A code may exist under several sources (hris floor + huilianyi
    # enrichment); collapse to one entry, preferring a non-null name and
    # summing nothing (headcount comes from the hris row). Track which
    # sources back the master so the UI can show provenance.
    master: dict[str, str | None] = {}
    master_headcount: dict[str, int | None] = {}
    by_source: dict[str, int] = {}
    for code, name, src, headcount in master_rows:
        by_source[src] = by_source.get(src, 0) + 1
        if code not in master or (master.get(code) is None and name):
            master[code] = name
        if headcount is not None:
            master_headcount[code] = headcount

    # Spend per cost-centre, this period. Internal ledger covers
    # cloud/software/ai/procurement AND the promoted 'expense' (mirrored
    # from external promote=True rows), so the external side adds ONLY the
    # reconciliation-only rows (promote_to_ledger=False) — otherwise
    # promoted expense would be counted twice (once in the ledger, once
    # here from the statement).
    spend: dict[str, Decimal] = {}
    internal_rows = (await db.execute(_withhold(
        select(FinanceSpendFact.cost_center, func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0))
        # Consumption basis ONLY — the platform invariant (see models.py) is
        # that consumption / procurement / cash are three lenses on the SAME
        # money and must never be summed. Without this, a hardware asset
        # (cash purchase + monthly depreciation) and a cloud PO (procurement
        # commitment + its consumption bill) each land twice on the same
        # cost_center, inflating total_spend / orphan_spend and skewing
        # coverage_pct. Mirrors get_summary / top_vendors / top_depts.
        .where(FinanceSpendFact.period == period,
               FinanceSpendFact.basis == BASIS_CONSUMPTION,
               FinanceSpendFact.cost_center.isnot(None), FinanceSpendFact.cost_center != "")
        .group_by(FinanceSpendFact.cost_center),
        withhold_cloud,
    ))).all()
    external_rows = (await db.execute(
        select(FinanceExternalStatement.cost_center,
               func.coalesce(func.sum(FinanceExternalStatement.amount_usd), 0))
        .where(FinanceExternalStatement.period == period,
               FinanceExternalStatement.cost_center.isnot(None),
               FinanceExternalStatement.cost_center != "",
               FinanceExternalStatement.promote_to_ledger.is_(False))
        .group_by(FinanceExternalStatement.cost_center)
    )).all()
    for code, usd in [*internal_rows, *external_rows]:
        spend[code] = spend.get(code, Decimal("0")) + _q2(usd)

    withheld_here = withheld_meta(withhold_cloud)
    used = set(spend)
    master_codes = set(master)
    orphan_codes = used - master_codes
    unused_codes = master_codes - used
    total_spend = sum(spend.values(), Decimal("0"))
    orphan_spend = sum((spend[c] for c in orphan_codes), Decimal("0"))
    coverage_pct = (
        round(float((total_spend - orphan_spend) / total_spend * 100), 2)
        if total_spend > 0 else 100.0
    )

    # Second attribution axis: owner coverage. A fact can sit in a valid
    # cost centre yet still have no person owner (owner_person_id NULL) —
    # the "(unallocated)" bucket finance also cares about. Surface it here
    # so the cost-centre page reports BOTH gaps (code-not-in-master AND
    # spend-with-no-owner) instead of only the first.
    owner_split = (await db.execute(_withhold(
        select(
            FinanceSpendFact.owner_person_id.is_(None),
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0),
        )
        # Same basis invariant as internal_rows above — the owner-coverage
        # ratio must be computed on a single (consumption) basis, else the
        # cross-basis double-count inflates both ledger_total and
        # unallocated_owner_usd and distorts owner_coverage_pct.
        .where(FinanceSpendFact.period == period,
               FinanceSpendFact.basis == BASIS_CONSUMPTION)
        .group_by(FinanceSpendFact.owner_person_id.is_(None)),
        withhold_cloud,
    ))).all()
    unallocated_owner_usd = Decimal("0")
    ledger_total = Decimal("0")
    for is_null, usd in owner_split:
        ledger_total += _q2(usd)
        if is_null:
            unallocated_owner_usd += _q2(usd)
    owner_coverage_pct = (
        round(float((ledger_total - unallocated_owner_usd) / ledger_total * 100), 2)
        if ledger_total > 0 else 100.0
    )

    return {
        "period": period,
        "withheld_cost_types": withheld_here,
        "master_count": len(master_codes),
        "master_by_source": by_source,
        "used_count": len(used),
        "coverage_pct": coverage_pct,
        "orphan_spend_usd": str(_q2(orphan_spend)),
        "total_spend_usd": str(_q2(total_spend)),
        # Owner-attribution axis (independent of cost-centre coverage).
        "owner_coverage_pct": owner_coverage_pct,
        "unallocated_owner_usd": str(_q2(unallocated_owner_usd)),
        "orphans": sorted(
            ({"code": c, "name": master.get(c), "spend_usd": str(spend[c])}
             for c in orphan_codes),
            key=lambda r: Decimal(r["spend_usd"]), reverse=True,
        ),
        "unused": [
            {"code": c, "name": master.get(c),
             "headcount": master_headcount.get(c)}
            for c in sorted(unused_codes)
        ][:200],
    }


async def expense_governance(db: AsyncSession, period: str) -> dict:
    """Reimbursement-hygiene findings over the external statements.

    * **duplicate_invoices** — one invoice number on >1 distinct document
      (potential double submission / double payment).
    * **paid_unverified** — line marked PAID but the invoice was never
      tax-verified.
    Both are concrete finance-control issues worth a human look.
    """
    from app.domains.finance.models import FinanceExternalStatement

    # Duplicate invoice numbers across distinct documents.
    dup_rows = (await db.execute(
        select(
            FinanceExternalStatement.invoice_no,
            func.count(func.distinct(FinanceExternalStatement.external_doc_no)).label("docs"),
            func.coalesce(func.sum(FinanceExternalStatement.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceExternalStatement.period == period,
            FinanceExternalStatement.invoice_no.isnot(None),
            FinanceExternalStatement.invoice_no != "",
        )
        .group_by(FinanceExternalStatement.invoice_no)
        .having(func.count(func.distinct(FinanceExternalStatement.external_doc_no)) > 1)
    )).all()
    duplicate_invoices = [
        {"invoice_no": r._mapping["invoice_no"], "doc_count": int(r._mapping["docs"]),
         "amount_usd": str(_q2(r._mapping["usd"]))}
        for r in dup_rows
    ]

    paid_unverified_rows = (await db.execute(
        select(FinanceExternalStatement)
        .where(
            FinanceExternalStatement.period == period,
            _external_statement_paid_predicate(),
            or_(
                FinanceExternalStatement.invoice_verified.is_(False),
                FinanceExternalStatement.invoice_verified.is_(None),
            ),
        )
        .order_by(FinanceExternalStatement.amount_usd.desc())
        .limit(200)
    )).scalars().all()
    paid_unverified = [
        {"external_doc_no": r.external_doc_no, "vendor": r.vendor,
         "amount_usd": str(r.amount_usd), "invoice_no": r.invoice_no}
        for r in paid_unverified_rows
    ]
    return {
        "period": period,
        "duplicate_invoices": duplicate_invoices,
        "paid_unverified": paid_unverified,
    }


# ─── BU fiscal-year cost view (per 一级部门, four axes, vs budget) ───


def _hardware_basis_value(hardware_basis: str) -> str:
    """Map the view toggle to the stored hardware ``basis``.

    ``depreciation`` (default) → the accrual lens that joins the
    consumption universe; ``cash`` → the purchase-outflow lens. Anything
    else falls back to depreciation.
    """
    return BASIS_CASH if hardware_basis == "cash" else BASIS_CONSUMPTION


async def _build_bu_resolver(
    db: AsyncSession,
) -> tuple[dict[str, str], dict[str, str]]:
    """One pass over primary memberships → (owner_id→BU, dept_name→BU).

    A spend fact is attributed to a BU by its owner first (most precise),
    then by its department name, then falls into the unallocated bucket.
    Both maps are derived from the same ``org_path`` rule via
    :func:`bu_from_org_path` so the BU view agrees with the AI dashboard
    and asset cost-stats.
    """
    from app.identity.bu_helpers import bu_from_org_path
    from app.identity.person_models import IdentityOrgMembership

    rows = (await db.execute(
        select(
            IdentityOrgMembership.person_id,
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_name,
        )
        .where(IdentityOrgMembership.is_primary.is_(True))
        .where(IdentityOrgMembership.deleted_at.is_(None))
    )).all()
    owner_bu: dict[str, str] = {}
    dept_bu: dict[str, str] = {}
    for pid, org_path, leaf in rows:
        bu = bu_from_org_path(org_path, fallback_name=leaf)
        if not bu:
            continue
        owner_bu[str(pid)] = bu
        if leaf and leaf not in dept_bu:
            dept_bu[leaf] = bu
    return owner_bu, dept_bu


def _bu_of_fact(owner, department, owner_bu, dept_bu) -> str:
    """Resolve a fact's BU: owner → department-name → unallocated."""
    if owner and str(owner) in owner_bu:
        return owner_bu[str(owner)]
    if department and department in dept_bu:
        return dept_bu[department]
    return UNALLOCATED


async def bu_fiscal_year_costs(
    db: AsyncSession,
    fiscal_year: int,
    *,
    as_of_period: str | None = None,
    hardware_basis: str = "depreciation",
    withhold_cloud: bool = False,
) -> dict:
    """Per-BU fiscal-year cost across the four consumption axes vs budget.

    Answers, for every 一级部门 (BU): how much it has spent this fiscal
    year (cloud / software / ai / hardware), how that maps against its
    budget, the YTD-prorated pace, and a run-rate year-end forecast.
    Hardware uses the depreciation (accrual) lens by default; pass
    ``hardware_basis='cash'`` for the purchase-cash lens.
    """
    from app.domains.finance import budget_service

    start_month = await fy.get_fiscal_year_start_month(db)
    if as_of_period is None:
        as_of_period = fy.default_as_of_period(fiscal_year, start_month)
    fy_lo, fy_hi = fy.fy_bounds(fiscal_year, start_month)
    elapsed = fy.months_elapsed(fiscal_year, as_of_period, start_month)
    hw_basis = _hardware_basis_value(hardware_basis)
    withheld_axes = set(withheld_meta(withhold_cloud))

    # YTD facts across the four axes. Cloud/software/ai always consumption;
    # hardware on the selected basis. Restrict to FY periods <= as_of.
    consumption_axes = (COST_TYPE_CLOUD, COST_TYPE_SOFTWARE, COST_TYPE_AI)
    axis_basis = or_(
        (FinanceSpendFact.cost_type.in_(consumption_axes))
        & (FinanceSpendFact.basis == BASIS_CONSUMPTION),
        (FinanceSpendFact.cost_type == COST_TYPE_HARDWARE)
        & (FinanceSpendFact.basis == hw_basis),
    )
    stmt = (
        select(
            FinanceSpendFact.cost_type,
            FinanceSpendFact.department,
            FinanceSpendFact.owner_person_id,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(FinanceSpendFact.period >= fy_lo)
        .where(FinanceSpendFact.period <= as_of_period)
        .where(axis_basis)
        .group_by(
            FinanceSpendFact.cost_type,
            FinanceSpendFact.department,
            FinanceSpendFact.owner_person_id,
        )
    )
    stmt = _withhold(stmt, withhold_cloud)
    fact_rows = (await db.execute(stmt)).all()

    owner_bu, dept_bu = await _build_bu_resolver(db)

    # bu → {axis → usd}, bu → total
    bu_axis: dict[str, dict[str, Decimal]] = {}
    for r in fact_rows:
        d = r._mapping
        bu = _bu_of_fact(d["owner_person_id"], d["department"], owner_bu, dept_bu)
        axis = d["cost_type"]
        bucket = bu_axis.setdefault(bu, {a: Decimal("0") for a in BU_AXES})
        bucket[axis] = bucket.get(axis, Decimal("0")) + _q2(d["usd"])

    # Budgets for the FY, resolved to a BU by their department scope. A
    # budget with no department (company-wide / cost-type-only) can't be
    # attributed to one BU; rather than silently drop it (which would make
    # the company total budget understate vs the company total actual), it
    # is tracked separately and added back into the company total + exposed
    # as ``company_wide_budget_usd`` so the totals reconcile honestly.
    budgets = await budget_service.list_budgets(db, fiscal_year)
    budget_scope_conflicts, conflicted_budget_ids = (
        budget_service.diagnose_scope_conflicts(budgets)
    )
    bu_budget: dict[str, Decimal] = {}
    company_wide_budget = Decimal("0")
    for b in budgets:
        if b.id in conflicted_budget_ids:
            # Same fail-closed rule as budget_vs_actual: historical overlap
            # has no safe automatic winner, so neither ceiling participates
            # in BU/company totals until Finance resolves the diagnostic.
            continue
        if not b.department:
            company_wide_budget += _q2(b.amount_usd)
            continue
        bu = dept_bu.get(b.department, b.department)  # admins may enter a BU name
        bu_budget[bu] = bu_budget.get(bu, Decimal("0")) + _q2(b.amount_usd)

    all_bus = set(bu_axis) | set(bu_budget)
    items = []
    tot_actual = Decimal("0")
    tot_budget = Decimal("0")
    tot_forecast = Decimal("0")
    for bu in all_bus:
        axes = bu_axis.get(bu, {a: Decimal("0") for a in BU_AXES})
        ytd = sum(axes.values(), Decimal("0"))
        annual_budget = _q2(bu_budget.get(bu, Decimal("0")))
        ytd_budget = _q2(annual_budget * Decimal(elapsed) / Decimal(12)) if elapsed else Decimal("0")
        forecast = _q2(ytd / Decimal(elapsed) * Decimal(12)) if elapsed else Decimal("0")
        pace = round(float(ytd / ytd_budget * 100), 1) if ytd_budget > 0 else None
        if annual_budget > 0 and forecast > annual_budget:
            status = "over"
        elif pace is not None and pace >= 100:
            status = "warning"
        else:
            status = "ok"
        tot_actual += ytd
        tot_budget += annual_budget
        tot_forecast += forecast
        items.append({
            "bu": bu,
            "by_axis": {
                a: str(_q2(axes.get(a, Decimal("0"))))
                for a in BU_AXES if a not in withheld_axes
            },
            "ytd_actual_usd": str(_q2(ytd)),
            "annual_budget_usd": str(annual_budget),
            "ytd_budget_usd": str(ytd_budget),
            "forecast_year_end_usd": str(forecast),
            "projected_variance_usd": str(_q2(annual_budget - forecast)) if annual_budget > 0 else None,
            "pace_pct": pace,
            "status": status,
        })

    items.sort(key=lambda e: Decimal(e["ytd_actual_usd"]), reverse=True)
    return {
        "fiscal_year": fiscal_year,
        "fiscal_year_start_month": start_month,
        "fy_label": fy.fy_label(fiscal_year, start_month),
        "fy_period_start": fy_lo,
        "fy_period_end": fy_hi,
        "as_of_period": as_of_period,
        "months_elapsed": elapsed,
        "hardware_basis": "cash" if hw_basis == BASIS_CASH else "depreciation",
        # A withheld axis is REMOVED from the axis list rather than reported
        # as a hard zero — a zero column reads as "this BU spent nothing on
        # cloud", which is a different (and false) statement from "you are
        # not shown cloud". ``withheld_cost_types`` says which.
        "axes": [a for a in BU_AXES if a not in withheld_axes],
        "withheld_cost_types": withheld_meta(withhold_cloud),
        "bu_count": len(items),
        "budget_scope_conflict_count": len(budget_scope_conflicts),
        "budget_scope_conflicts": budget_scope_conflicts,
        "budget_totals_exclude_scope_conflicts": bool(budget_scope_conflicts),
        "total_ytd_actual_usd": str(_q2(tot_actual)),
        # Company total budget = per-BU budgets + company-wide (unattributed)
        # budgets, so it never understates against the company total actual.
        "total_annual_budget_usd": str(_q2(tot_budget + company_wide_budget)),
        "company_wide_budget_usd": str(_q2(company_wide_budget)),
        "total_forecast_year_end_usd": str(_q2(tot_forecast)),
        "items": items,
    }


_BU_DRILL_DIMS = {"cost_type", "vendor", "sub_category", "owner", "cost_center"}


async def bu_cost_drill(
    db: AsyncSession,
    fiscal_year: int,
    bu: str,
    *,
    as_of_period: str | None = None,
    hardware_basis: str = "depreciation",
    group_by: str = "vendor",
    cost_type: str | None = None,
    withhold_cloud: bool = False,
) -> dict:
    """Drill one BU's fiscal-year spend by vendor / product / person / etc.

    The same four-axis YTD window as :func:`bu_fiscal_year_costs`, narrowed
    to a single BU (resolved via owner / department) and optionally one
    ``cost_type``, grouped by one dimension. Powers the per-BU expand row
    and the "每个部门的人 AI/云花了多少" person drill.
    """
    if group_by not in _BU_DRILL_DIMS:
        raise AppException(
            code="INVALID_DIM",
            message=f"group_by must be one of {sorted(_BU_DRILL_DIMS)}",
            status_code=400,
        )
    start_month = await fy.get_fiscal_year_start_month(db)
    if as_of_period is None:
        as_of_period = fy.default_as_of_period(fiscal_year, start_month)
    fy_lo, _ = fy.fy_bounds(fiscal_year, start_month)
    hw_basis = _hardware_basis_value(hardware_basis)

    consumption_axes = (COST_TYPE_CLOUD, COST_TYPE_SOFTWARE, COST_TYPE_AI)
    axis_basis = or_(
        (FinanceSpendFact.cost_type.in_(consumption_axes))
        & (FinanceSpendFact.basis == BASIS_CONSUMPTION),
        (FinanceSpendFact.cost_type == COST_TYPE_HARDWARE)
        & (FinanceSpendFact.basis == hw_basis),
    )
    stmt = (
        select(
            FinanceSpendFact.cost_type,
            FinanceSpendFact.vendor,
            FinanceSpendFact.sub_category,
            FinanceSpendFact.cost_center,
            FinanceSpendFact.owner_person_id,
            FinanceSpendFact.department,
            FinanceSpendFact.amount_usd,
        )
        .where(FinanceSpendFact.period >= fy_lo)
        .where(FinanceSpendFact.period <= as_of_period)
        .where(axis_basis)
    )
    if cost_type:
        stmt = stmt.where(FinanceSpendFact.cost_type == cost_type)
    # Applied AFTER the explicit cost_type narrow, so an out-of-circle
    # reader asking for ``cost_type=cloud`` intersects to the empty set
    # rather than overriding the gate (same rule as ``_apply_filters``).
    stmt = _withhold(stmt, withhold_cloud)
    rows = (await db.execute(stmt)).all()

    owner_bu, dept_bu = await _build_bu_resolver(db)
    agg: dict[str, Decimal] = {}
    owner_ids: set[str] = set()
    for r in rows:
        d = r._mapping
        if _bu_of_fact(d["owner_person_id"], d["department"], owner_bu, dept_bu) != bu:
            continue
        if group_by == "owner":
            key = str(d["owner_person_id"]) if d["owner_person_id"] else UNALLOCATED
            if key != UNALLOCATED:
                owner_ids.add(key)
        elif group_by in ("cost_center",):
            key = d["cost_center"] or UNALLOCATED
        else:
            key = d[group_by] or UNKNOWN_LABEL
        agg[key] = agg.get(key, Decimal("0")) + _q2(d["amount_usd"])

    names = await _resolve_person_names(db, owner_ids) if owner_ids else {}
    out = []
    total = Decimal("0")
    for key, usd in agg.items():
        total += usd
        entry = {group_by: key, "usd": str(_q2(usd))}
        if group_by == "owner":
            entry["owner_name"] = names.get(key)
        out.append(entry)
    out.sort(key=lambda e: Decimal(e["usd"]), reverse=True)
    return {
        "fiscal_year": fiscal_year,
        "bu": bu,
        "group_by": group_by,
        "cost_type": cost_type,
        "hardware_basis": "cash" if hw_basis == BASIS_CASH else "depreciation",
        "as_of_period": as_of_period,
        "rows": out,
        "total_usd": str(_q2(total)),
    }


# ─── NetSuite GL actuals (book of record) read surface ──────────────
#
# Reads ONLY the finance-owned NetSuite mirror tables (GL actuals +
# masters). Connector status / live pulls live in the coordinator (the
# sanctioned cross-domain glue), keeping this module domain-pure.


async def netsuite_gl_summary(db: AsyncSession, period: str) -> dict:
    """Summarized NetSuite GL for a month, sliced by account type,
    subsidiary (with the Helix legal-entity label when mapped), department,
    and the largest accounts. Numbers are the signed GL movement in USD."""
    rows = (await db.execute(
        select(FinanceGLActual).where(FinanceGLActual.period == period)
    )).scalars().all()

    # Subsidiary → human label + mapped legal entity (single pass over the
    # master so the summary can name subsidiaries rather than show raw ids).
    subs = {
        s.subsidiary_id: s for s in (
            await db.execute(select(FinanceNetSuiteSubsidiary))
        ).scalars().all()
    }

    by_type: dict[str, Decimal] = {}
    by_sub: dict[str, Decimal] = {}
    by_dept: dict[str, Decimal] = {}
    by_account: dict[tuple[str, str], Decimal] = {}
    total = Decimal("0")
    latest_pulled = None
    any_closed = False
    for r in rows:
        amt = Decimal(str(r.amount_usd or 0))
        total += amt
        by_type[r.account_type or "未分类"] = (
            by_type.get(r.account_type or "未分类", Decimal("0")) + amt
        )
        by_sub[r.subsidiary_id or "__none__"] = (
            by_sub.get(r.subsidiary_id or "__none__", Decimal("0")) + amt
        )
        by_dept[r.department_name or UNALLOCATED] = (
            by_dept.get(r.department_name or UNALLOCATED, Decimal("0")) + amt
        )
        akey = (r.account_number or r.account_id, r.account_name or "")
        by_account[akey] = by_account.get(akey, Decimal("0")) + amt
        if r.pulled_at and (latest_pulled is None or r.pulled_at > latest_pulled):
            latest_pulled = r.pulled_at
        any_closed = any_closed or bool(r.is_closed)

    def _sub_label(sid: str) -> dict:
        if sid == "__none__":
            return {"subsidiary_id": None, "name": "(无主体)", "legal_entity_id": None}
        s = subs.get(sid)
        return {
            "subsidiary_id": sid,
            "name": (s.name if s else None) or sid,
            "legal_entity_id": s.legal_entity_id if s else None,
        }

    return {
        "period": period,
        "row_count": len(rows),
        "total_usd": str(_q2(total)),
        "is_closed": any_closed,
        "as_of": latest_pulled.isoformat() if latest_pulled else None,
        "by_account_type": [
            {"account_type": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_type.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "by_subsidiary": [
            {**_sub_label(k), "amount_usd": str(_q2(v))}
            for k, v in sorted(by_sub.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "by_department": [
            {"department": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_dept.items(), key=lambda kv: kv[1], reverse=True)[:30]
        ],
        "top_accounts": [
            {"account_number": k[0], "account_name": k[1],
             "amount_usd": str(_q2(v))}
            for k, v in sorted(by_account.items(), key=lambda kv: abs(kv[1]),
                               reverse=True)[:30]
        ],
    }


async def netsuite_subsidiaries(db: AsyncSession) -> dict:
    """NetSuite subsidiary master + the human-set legal-entity mapping."""
    rows = (await db.execute(
        select(FinanceNetSuiteSubsidiary).order_by(
            FinanceNetSuiteSubsidiary.is_inactive,
            FinanceNetSuiteSubsidiary.name,
        )
    )).scalars().all()
    return {
        "subsidiaries": [
            {
                "subsidiary_id": s.subsidiary_id,
                "name": s.name,
                "full_name": s.full_name,
                "currency": s.currency,
                "parent_id": s.parent_id,
                "is_elimination": s.is_elimination,
                "is_inactive": s.is_inactive,
                "legal_entity_id": s.legal_entity_id,
            }
            for s in rows
        ],
        "mapped": sum(1 for s in rows if s.legal_entity_id),
        "total": len(rows),
    }


async def netsuite_periods(db: AsyncSession, *, limit: int = 60) -> dict:
    """NetSuite accounting-period master (closed flags), newest first."""
    rows = (await db.execute(
        select(FinanceNetSuiteAccountingPeriod)
        .order_by(FinanceNetSuiteAccountingPeriod.start_date.desc())
        .limit(max(1, min(limit, 500)))
    )).scalars().all()
    return {
        "periods": [
            {
                "period_id": p.period_id,
                "period_name": p.period_name,
                "period": p.period,
                "start_date": p.start_date,
                "end_date": p.end_date,
                "closed": p.closed,
                "is_adjustment": p.is_adjustment,
                "is_quarter": p.is_quarter,
                "is_year": p.is_year,
            }
            for p in rows
        ],
    }


async def netsuite_sync_status(db: AsyncSession) -> dict:
    """Data-presence status for the NetSuite GL mirror (finance tables
    only): row/period coverage, freshness, master counts. Connector
    configuration is reported by the probe endpoint."""
    gl_rows = (await db.execute(
        select(
            FinanceGLActual.period,
            func.count().label("n"),
            func.max(FinanceGLActual.pulled_at).label("pulled"),
            func.max(FinanceGLActual.is_closed).label("closed"),
        ).group_by(FinanceGLActual.period)
        .order_by(FinanceGLActual.period.desc())
    )).all()
    sub_n = (await db.execute(
        select(func.count()).select_from(FinanceNetSuiteSubsidiary)
    )).scalar() or 0
    acct_n = (await db.execute(
        select(func.count()).select_from(FinanceNetSuiteAccount)
    )).scalar() or 0
    period_n = (await db.execute(
        select(func.count()).select_from(FinanceNetSuiteAccountingPeriod)
    )).scalar() or 0
    return {
        "periods": [
            {
                "period": r.period,
                "rows": int(r.n),
                "as_of": r.pulled.isoformat() if r.pulled else None,
                "is_closed": bool(r.closed),
            }
            for r in gl_rows
        ],
        "masters": {
            "subsidiaries": int(sub_n),
            "accounts": int(acct_n),
            "accounting_periods": int(period_n),
        },
    }


# ─── NS GL ↔ Helix bottom-up reconciliation ─────────────────────────
#
# The trust loop: does Helix's operational spend tie to what NetSuite
# booked? Per cost_type, compare Helix bottom-up `finance_spend_facts`
# against the GL movement on accounts classified to that cost_type. The
# diff is the signal (booked-not-in-Helix / in-Helix-not-booked = accrual
# candidate / mis-classification). Reconciliation compares COST streams
# only — never consolidated statements (NetSuite intercompany
# elimination + CTA can't be externally recomputed; the flux .xls report
# stays the only faithful consolidated source — see DECISIONS 2026-06-27).

# Which spend_facts basis each cost_type reconciles on (the accrual view
# that matches booked GL expense). procurement uses its own commitment
# basis; expense (汇联易-promoted) carries no single basis so we sum all.
_RECON_BASIS = {
    COST_TYPE_CLOUD: BASIS_CONSUMPTION,
    COST_TYPE_SOFTWARE: BASIS_CONSUMPTION,
    COST_TYPE_AI: BASIS_CONSUMPTION,
    COST_TYPE_HARDWARE: BASIS_CONSUMPTION,
    COST_TYPE_PROCUREMENT: BASIS_PROCUREMENT,
}
_RECON_COST_TYPES = (
    COST_TYPE_CLOUD, COST_TYPE_SOFTWARE, COST_TYPE_AI,
    COST_TYPE_HARDWARE, COST_TYPE_PROCUREMENT, EXPENSE_COST_TYPE,
)


def _recon_match(internal: Decimal, gl: Decimal) -> bool:
    """Tolerant match: within max($1k, 5% of the larger side)."""
    tol = max(Decimal("1000"), (max(abs(internal), abs(gl)) * Decimal("0.05")))
    return abs(internal - gl) <= tol


async def gl_reconcile(db: AsyncSession, period: str) -> dict:
    """Per-cost-type reconciliation of Helix bottom-up spend vs NetSuite
    booked GL, plus the unclassified-GL exception list."""
    # Internal: spend_facts grouped by (cost_type, basis) — pick the
    # reconciliation basis per cost_type in Python.
    int_rows = (await db.execute(
        select(
            FinanceSpendFact.cost_type,
            FinanceSpendFact.basis,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        ).where(FinanceSpendFact.period == period)
        .group_by(FinanceSpendFact.cost_type, FinanceSpendFact.basis)
    )).all()
    internal: dict[str, Decimal] = {}
    for r in int_rows:
        ct = r._mapping["cost_type"]
        basis = r._mapping["basis"]
        want = _RECON_BASIS.get(ct)  # None for 'expense' → accept any basis
        if want is not None and basis != want:
            continue
        internal[ct] = internal.get(ct, Decimal("0")) + _q2(r._mapping["usd"])

    # GL: actuals joined to the account master's mapped_cost_type.
    gl_rows = (await db.execute(
        select(
            FinanceNetSuiteAccount.mapped_cost_type.label("ct"),
            func.coalesce(func.sum(FinanceGLActual.amount_usd), 0).label("usd"),
        )
        .select_from(FinanceGLActual)
        .join(
            FinanceNetSuiteAccount,
            FinanceNetSuiteAccount.account_id == FinanceGLActual.account_id,
        )
        .where(FinanceGLActual.period == period)
        .group_by(FinanceNetSuiteAccount.mapped_cost_type)
    )).all()
    gl: dict[str, Decimal] = {}
    for gr in gl_rows:
        ct = gr._mapping["ct"]
        if not ct or ct == "__none__":
            continue
        gl[ct] = gl.get(ct, Decimal("0")) + _q2(gr._mapping["usd"])

    legs = []
    all_match = True
    for ct in _RECON_COST_TYPES:
        iv = internal.get(ct, Decimal("0"))
        gv = gl.get(ct, Decimal("0"))
        if iv == 0 and gv == 0:
            continue
        match = _recon_match(iv, gv)
        all_match = all_match and match
        legs.append({
            "cost_type": ct,
            "internal_usd": str(iv),
            "gl_usd": str(gv),
            "diff_usd": str(_q2(iv - gv)),
            "match": match,
        })

    # Exceptions: material GL booked to accounts with NO cost_type (NULL)
    # — "booked but unclassified", the finance work item to map.
    unmapped = (await db.execute(
        select(
            FinanceGLActual.account_id,
            func.max(FinanceGLActual.account_number).label("num"),
            func.max(FinanceGLActual.account_name).label("name"),
            func.max(FinanceGLActual.account_type).label("type"),
            func.coalesce(func.sum(FinanceGLActual.amount_usd), 0).label("usd"),
        )
        .select_from(FinanceGLActual)
        .outerjoin(
            FinanceNetSuiteAccount,
            FinanceNetSuiteAccount.account_id == FinanceGLActual.account_id,
        )
        .where(
            FinanceGLActual.period == period,
            FinanceNetSuiteAccount.mapped_cost_type.is_(None),
        )
        .group_by(FinanceGLActual.account_id)
    )).all()
    # Only surface P&L cost-shaped unclassified accounts with material spend.
    # Balance-sheet / income accounts (Bank, AP/AR, equity, assets,
    # liabilities, deferred, revenue) are LEGITIMATELY unmapped — not cost
    # classification work items — and NetSuite stores accttype as CODES
    # (AcctPay / FixedAsset / OthCurrAsset / ...), so we gate on the canonical
    # cost-code allowlist (Expense / COGS / OthExpense), shared with the ingest
    # heuristic. Without this, once GL syncs, material balance-sheet movements
    # (e.g. a Bank account at -359,589.75, or 263 OthAsset accounts) flood the
    # "unclassified" worklist, inflate unmapped_total_usd as if it were
    # unclassified COST, and permanently hold close_cockpit.ready_to_close
    # at false.
    from app.core.coordinators.finance_netsuite_coordinator import (
        is_cost_account_type,
    )

    unmapped_rows = [
        {
            "account_id": r._mapping["account_id"],
            "account_number": r._mapping["num"],
            "account_name": r._mapping["name"],
            "account_type": r._mapping["type"],
            "amount_usd": str(_q2(r._mapping["usd"])),
        }
        for r in unmapped
        if abs(Decimal(str(r._mapping["usd"] or 0))) >= Decimal("1000")
        and is_cost_account_type(r._mapping["type"])
    ]
    unmapped_rows.sort(key=lambda x: abs(Decimal(x["amount_usd"])), reverse=True)

    return {
        "period": period,
        "all_match": all_match and bool(legs),
        "has_gl": bool(gl_rows),
        "legs": legs,
        "unmapped_accounts": unmapped_rows[:50],
        "unmapped_total_usd": str(_q2(sum(
            (Decimal(x["amount_usd"]) for x in unmapped_rows), Decimal("0"),
        ))),
    }


async def close_cockpit(db: AsyncSession, period: str) -> dict:
    """Controller close cockpit: a single readiness view combining data
    completeness, NetSuite GL sync + reconciliation, and the period close
    state — with a derived go/no-go checklist."""
    comp = await completeness(db, period)
    recon = await gl_reconcile(db, period)
    sync = await netsuite_sync_status(db)
    gl_period = next((p for p in sync["periods"] if p["period"] == period), None)

    close = (await db.execute(
        select(FinancePeriodClose).where(FinancePeriodClose.period == period)
    )).scalar_one_or_none()
    close_state = close.status if close else "open"

    # NetSuite-side period closed? (book frozen on the vendor)
    ns_period = (await db.execute(
        select(FinanceNetSuiteAccountingPeriod).where(
            FinanceNetSuiteAccountingPeriod.period == period,
            FinanceNetSuiteAccountingPeriod.is_quarter.is_(False),
            FinanceNetSuiteAccountingPeriod.is_year.is_(False),
        )
    )).scalars().all()
    ns_closed = bool(ns_period) and all(p.closed for p in ns_period)

    no_open_restatements = not comp["restatements"]
    checklist = [
        {"key": "data_complete", "ok": bool(comp["sources"]),
         "detail": f"{len(comp['sources'])} sources materialised"},
        {"key": "gl_synced", "ok": bool(gl_period),
         "detail": (f"{gl_period['rows']} GL rows, as of {gl_period['as_of']}"
                    if gl_period else "GL not pulled for this period")},
        {"key": "ns_period_closed", "ok": ns_closed,
         "detail": "NetSuite period closed" if ns_closed
                   else "NetSuite period still open"},
        {"key": "recon_clean", "ok": recon["all_match"] and not recon["unmapped_accounts"],
         "detail": (f"{len(recon['unmapped_accounts'])} unclassified GL accounts"
                    if recon["unmapped_accounts"]
                    else ("all cost types tie out" if recon["all_match"]
                          else "variance beyond tolerance"))},
        {"key": "no_open_restatements", "ok": no_open_restatements,
         "detail": (f"{len(comp['restatements'])} restatement(s)"
                    if not no_open_restatements else "none")},
    ]
    ready_to_close = all(c["ok"] for c in checklist)

    return {
        "period": period,
        "close_state": close_state,
        "closed_total_usd": (
            str(close.closed_total_usd)
            if close and close.closed_total_usd is not None else None
        ),
        "closed_at": close.closed_at.isoformat() if close and close.closed_at else None,
        "ready_to_close": ready_to_close,
        "checklist": checklist,
        "completeness": comp,
        "reconciliation": recon,
        "gl_period": gl_period,
        "ns_period_closed": ns_closed,
    }


# ─── Expense intelligence (汇联易 external statements) ───────────────
#
# Owner attribution on external statements is now populated by the
# coordinator's email resolver, so these surface per-employee + the
# employee's own reimbursements. All read the finance-owned
# finance_external_statements; reconciliation/hygiene live elsewhere.


async def my_expenses(db: AsyncSession, person_id: str, *, limit: int = 200) -> dict:
    """An employee's own reimbursement/expense documents + a status summary.
    Self-service: the caller passes their own resolved person id."""
    from app.domains.finance.models import FinanceExternalStatement

    rows = (await db.execute(
        select(FinanceExternalStatement)
        .where(FinanceExternalStatement.owner_person_id == person_id)
        .order_by(
            FinanceExternalStatement.period.desc(),
            FinanceExternalStatement.paid_at.desc(),
        )
        .limit(max(1, min(limit, 500)))
    )).scalars().all()

    by_status: dict[str, Decimal] = {}
    by_category: dict[str, Decimal] = {}
    total = Decimal("0")
    for r in rows:
        amt = Decimal(str(r.amount_usd or 0))
        total += amt
        st = (r.payment_status or "unknown").lower()
        by_status[st] = by_status.get(st, Decimal("0")) + amt
        cat = r.expense_category or "未分类"
        by_category[cat] = by_category.get(cat, Decimal("0")) + amt

    return {
        "person_id": person_id,
        "count": len(rows),
        "total_usd": str(_q2(total)),
        "by_status": [
            {"status": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_status.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "by_category": [
            {"category": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_category.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "documents": [
            {
                "doc_no": r.external_doc_no,
                "period": r.period,
                "category": r.expense_category,
                "cost_center": r.cost_center,
                "vendor": r.vendor,
                "amount_usd": str(_q2(Decimal(str(r.amount_usd or 0)))),
                "amount_native": str(r.amount_native),
                "currency": r.currency,
                "invoice_no": r.invoice_no,
                "invoice_verified": r.invoice_verified,
                "doc_status": r.doc_status,
                "payment_status": r.payment_status,
                "paid_at": r.paid_at.isoformat() if r.paid_at else None,
            }
            for r in rows
        ],
    }


async def expense_analytics(db: AsyncSession, period: str) -> dict:
    """T&E / expense intelligence for a month: by category, by department,
    by payment status, and top spenders (owner-resolved)."""
    from app.domains.finance.models import FinanceExternalStatement

    rows = (await db.execute(
        select(FinanceExternalStatement).where(
            FinanceExternalStatement.period == period,
        )
    )).scalars().all()

    by_category: dict[str, Decimal] = {}
    by_department: dict[str, Decimal] = {}
    by_status: dict[str, Decimal] = {}
    by_owner: dict[str, Decimal] = {}
    total = Decimal("0")
    for r in rows:
        amt = Decimal(str(r.amount_usd or 0))
        total += amt
        by_category[r.expense_category or "未分类"] = (
            by_category.get(r.expense_category or "未分类", Decimal("0")) + amt
        )
        by_department[r.department or UNALLOCATED] = (
            by_department.get(r.department or UNALLOCATED, Decimal("0")) + amt
        )
        by_status[(r.payment_status or "unknown").lower()] = (
            by_status.get((r.payment_status or "unknown").lower(), Decimal("0")) + amt
        )
        if r.owner_person_id:
            by_owner[r.owner_person_id] = (
                by_owner.get(r.owner_person_id, Decimal("0")) + amt
            )

    top_owner_ids = [
        oid for oid, _ in sorted(by_owner.items(), key=lambda kv: kv[1], reverse=True)[:15]
    ]
    names = await _resolve_person_names(db, set(top_owner_ids))

    def _rank(d: dict[str, Decimal], n: int = 20) -> list[dict]:
        return [
            {"key": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(d.items(), key=lambda kv: kv[1], reverse=True)[:n]
        ]

    return {
        "period": period,
        "row_count": len(rows),
        "total_usd": str(_q2(total)),
        "attributed_count": sum(1 for r in rows if r.owner_person_id),
        "by_category": _rank(by_category),
        "by_department": _rank(by_department),
        "by_status": _rank(by_status),
        "top_spenders": [
            {"owner_person_id": oid, "owner_name": names.get(oid),
             "amount_usd": str(_q2(by_owner[oid]))}
            for oid in top_owner_ids
        ],
    }


# ─── Employee "my money" + BU-owner self-service ────────────────────


def _period_minus(period: str, months: int) -> str:
    y, m = (int(x) for x in period.split("-"))
    idx = y * 12 + (m - 1) - months
    yy, mm = divmod(idx, 12)
    return f"{yy:04d}-{mm + 1:02d}"


async def my_attributed_cost(
    db: AsyncSession, person_id: str, period: str, *,
    withhold_cloud: bool = False,
) -> dict:
    """A person's own attributed cost to the company for a month: the
    unified ledger sliced to ``owner_person_id = me`` (cloud / software /
    ai / hardware / procurement, consumption basis) + their 汇联易 expenses,
    plus a trailing-12-month total trend. Self-service; everyone sees only
    their own."""
    # Ledger, owner = me, consumption basis (the accrual the person 'costs').
    # 汇联易 reimbursements promoted into the ledger ALREADY appear here under
    # ``cost_type='expense'`` — finance_spend_coordinator.collect_expense_rows
    # is the single, no-double-count source (only ``promote_to_ledger=True``
    # rows). So expenses are read from the ledger like every other cost type
    # and are NEVER re-added from finance_external_statements: re-summing the
    # promoted rows would double-count them (a prior bug), and summing the
    # non-promoted rows would double-count reconciliation-only invoices that
    # are already booked under their own cost type (software / cloud / ...).
    rows = (await db.execute(_withhold(
        select(
            FinanceSpendFact.cost_type,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.owner_person_id == person_id,
            FinanceSpendFact.period == period,
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        )
        .group_by(FinanceSpendFact.cost_type),
        # Self-scope does NOT put a reader inside the invoice-measure
        # circle (#2216): the caller can already read their own list-basis
        # cloud usage from /cmdb/analytics/my, so handing them the invoice
        # figure for the same resources hands them the discount rate.
        withhold_cloud,
    ))).all()
    by_cost_type = {r._mapping["cost_type"]: _q2(r._mapping["usd"]) for r in rows}

    # "My reimbursements" surfaced as a distinct field, but sourced from the
    # ledger's expense slice (above) — not a second query — so it cannot
    # double-count.
    expense_usd = by_cost_type.get(EXPENSE_COST_TYPE, Decimal("0"))

    total = sum(by_cost_type.values(), Decimal("0"))

    # Trailing 12-month total trend — ledger only (promoted expenses already
    # included; no second external leg, per the by_cost_type note above).
    lo = _period_minus(period, 11)
    trend_rows = (await db.execute(_withhold(
        select(
            FinanceSpendFact.period,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.owner_person_id == person_id,
            FinanceSpendFact.period >= lo,
            FinanceSpendFact.period <= period,
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        )
        .group_by(FinanceSpendFact.period),
        withhold_cloud,
    ))).all()
    trend_map: dict[str, Decimal] = {}
    for r in trend_rows:
        trend_map[r._mapping["period"]] = trend_map.get(
            r._mapping["period"], Decimal("0")) + _q2(r._mapping["usd"])

    return {
        "person_id": person_id,
        "period": period,
        "total_usd": str(_q2(total)),
        "by_cost_type": [
            {"cost_type": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_cost_type.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "expense_usd": str(expense_usd),
        "trend": [
            {"period": p, "usd": str(_q2(trend_map.get(p, Decimal("0"))))}
            for p in (_period_minus(period, n) for n in range(11, -1, -1))
        ],
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }


async def my_bu_costs(
    db: AsyncSession, person_id: str, fiscal_year: int, *,
    as_of_period: str | None = None, hardware_basis: str = "depreciation",
    withhold_cloud: bool = False,
) -> dict:
    """The BU a person leads/belongs to, scoped to just that BU — the
    self-service slice of :func:`bu_fiscal_year_costs`. Returns an empty
    ``items`` (with ``bu=None``) when the person maps to no BU, so a
    non-owner simply sees nothing rather than a forbidden error."""
    owner_bu, _dept_bu = await _build_bu_resolver(db)
    my_bu = owner_bu.get(str(person_id))
    if not my_bu or my_bu == UNALLOCATED:
        return {"bu": None, "fiscal_year": fiscal_year, "items": []}
    full = await bu_fiscal_year_costs(
        db, fiscal_year, as_of_period=as_of_period, hardware_basis=hardware_basis,
        withhold_cloud=withhold_cloud,
    )
    mine = [i for i in full["items"] if i.get("bu") == my_bu]
    return {
        "bu": my_bu,
        "fiscal_year": full["fiscal_year"],
        "fy_label": full["fy_label"],
        "as_of_period": full["as_of_period"],
        "hardware_basis": full["hardware_basis"],
        "axes": full["axes"],
        "items": mine,
        "withheld_cost_types": full["withheld_cost_types"],
    }


# ─── Vendor 360 + CFO leadership dashboard ──────────────────────────


async def top_vendors(
    db: AsyncSession, period: str, *, limit: int = 15,
    withhold_cloud: bool = False,
) -> dict:
    """Largest vendors for a month by consumption cost (across all
    consumption cost types — cloud / software / ai / hardware), for the
    Vendor 360 picker + the leadership dashboard's vendor tile.

    Scoped to ``consumption`` basis — the canonical cost lens used by
    every other finance read. Summing across bases would double-count
    (a vendor's procurement PO and its recognized consumption describe
    the same money from two angles); see the model docstring's
    basis invariant."""
    rows = (await db.execute(_withhold(
        select(
            FinanceSpendFact.vendor,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.period == period,
            FinanceSpendFact.vendor.isnot(None),
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        )
        .group_by(FinanceSpendFact.vendor)
        .order_by(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).desc())
        .limit(max(1, min(limit, 100))),
        withhold_cloud,
    ))).all()
    return {
        "period": period,
        "vendors": [
            {"vendor": r._mapping["vendor"], "amount_usd": str(_q2(r._mapping["usd"]))}
            for r in rows
        ],
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }


async def vendor_360(
    db: AsyncSession, period: str, vendor: str, *,
    withhold_cloud: bool = False,
) -> dict:
    """One vendor's consumption cost for a month, unified across consumption
    cost types (cloud / software / ai / hardware — the split a plain vendor
    pivot hides), by department, a 12-month trend, and the vendor's
    concentration as a share of company consumption. Answers 'what does
    <vendor> cost us across everything we run'.

    Scoped to ``consumption`` basis: the bases are different views of the
    same money (a procurement PO and its recognized seat/depreciation
    cost), so summing across them would double-count — the platform
    invariant the model docstring calls out. Procurement/cash commitment
    views live on the budget/commitment surfaces, not here."""
    rows = (await db.execute(_withhold(
        select(
            FinanceSpendFact.cost_type,
            FinanceSpendFact.department,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.period == period,
            FinanceSpendFact.vendor == vendor,
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        )
        .group_by(FinanceSpendFact.cost_type, FinanceSpendFact.department),
        withhold_cloud,
    ))).all()
    by_cost_type: dict[str, Decimal] = {}
    by_department: dict[str, Decimal] = {}
    total = Decimal("0")
    for r in rows:
        amt = _q2(r._mapping["usd"])
        total += amt
        ct = r._mapping["cost_type"]
        by_cost_type[ct] = by_cost_type.get(ct, Decimal("0")) + amt
        dep = r._mapping["department"] or UNALLOCATED
        by_department[dep] = by_department.get(dep, Decimal("0")) + amt

    lo = _period_minus(period, 11)
    trend_rows = (await db.execute(_withhold(
        select(
            FinanceSpendFact.period,
            func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0).label("usd"),
        )
        .where(
            FinanceSpendFact.vendor == vendor,
            FinanceSpendFact.period >= lo, FinanceSpendFact.period <= period,
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        )
        .group_by(FinanceSpendFact.period),
        withhold_cloud,
    ))).all()
    trend_map = {r._mapping["period"]: _q2(r._mapping["usd"]) for r in trend_rows}

    # Concentration: vendor's share of the month's company consumption
    # (same basis as the numerator, so the ratio is honest).
    company = _q2((await db.execute(_withhold(
        select(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0))
        .where(
            FinanceSpendFact.period == period,
            FinanceSpendFact.basis == BASIS_CONSUMPTION,
        ),
        # Numerator and denominator must be withheld the SAME way or the
        # concentration % becomes a back-door: a vendor's cloud-free total
        # over a cloud-inclusive company total is a different ratio, and
        # the difference is exactly the withheld measure.
        withhold_cloud,
    ))).scalar())
    concentration_pct = (
        round(float(total / company * 100), 2) if company > 0 else 0.0
    )

    return {
        "period": period,
        "vendor": vendor,
        "total_usd": str(_q2(total)),
        "concentration_pct": concentration_pct,
        "by_cost_type": [
            {"cost_type": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_cost_type.items(), key=lambda kv: kv[1], reverse=True)
        ],
        "by_department": [
            {"department": k, "amount_usd": str(_q2(v))}
            for k, v in sorted(by_department.items(), key=lambda kv: kv[1], reverse=True)[:20]
        ],
        "trend": [
            {"period": p, "usd": str(trend_map.get(p, Decimal("0")))}
            for p in (_period_minus(period, n) for n in range(11, -1, -1))
        ],
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }


async def _active_headcount(db: AsyncSession) -> int:
    from app.identity.person_models import IdentityPerson
    # ``deleted_at IS NULL`` is mandatory alongside employment_status:
    # a soft-deleted person keeps ``employment_status='active'`` (platform
    # gotcha), so without this guard the count is inflated and the
    # ``cost_per_headcount_usd`` on the leadership dashboard is understated.
    return int((await db.execute(
        select(func.count()).select_from(IdentityPerson)
        .where(
            IdentityPerson.employment_status == "active",
            IdentityPerson.deleted_at.is_(None),
        )
    )).scalar() or 0)


async def leadership_dashboard(
    db: AsyncSession, period: str, *, withhold_cloud: bool = False,
) -> dict:
    """Company-wide leadership snapshot for one month, composed from the
    existing reads (one consistent call for the CFO 大屏): headline totals +
    mix, top department movers, top vendors, per-BU cost vs budget, budget
    health, unallocated %, and cost-per-headcount."""
    from app.domains.finance import budget_service

    summ = await get_summary(db, period, withhold_cloud=withhold_cloud)
    dept_movers = await movers(
        db, period, "department", top=8, withhold_cloud=withhold_cloud,
    )
    tv = await top_vendors(db, period, limit=10, withhold_cloud=withhold_cloud)
    start_month = await fy.get_fiscal_year_start_month(db)
    fyear = fy.fy_of_period(period, start_month)
    bu = await bu_fiscal_year_costs(
        db, fyear, as_of_period=period, withhold_cloud=withhold_cloud,
    )
    # Budget health rides the SAME ledger. Left un-withheld it is a
    # subtraction channel: ytd_actual (all four axes) minus the three axes
    # bu_costs still shows == the cloud number this whole gate exists to
    # withhold.
    budget = await budget_service.budget_vs_actual(db, fyear, as_of_period=period,
                                                   start_month=start_month,
                                                   withhold_cloud=withhold_cloud)
    headcount = await _active_headcount(db)
    consumption = Decimal(summ["consumption_total_usd"])
    cost_per_head = (
        str(_q2(consumption / Decimal(headcount))) if headcount > 0 else None
    )
    return {
        "period": period,
        "summary": summ,
        "headcount": headcount,
        "cost_per_headcount_usd": cost_per_head,
        "dept_movers": dept_movers,
        "top_vendors": tv["vendors"],
        "bu_costs": bu["items"],
        "fy_label": bu["fy_label"],
        "budget_health": {
            "over_count": budget["over_count"],
            "warning_count": budget["warning_count"],
            "total_annual_budget_usd": budget["total_annual_budget_usd"],
            "total_ytd_actual_usd": budget["total_ytd_actual_usd"],
            "total_committed_usd": budget.get("total_committed_usd"),
            "total_forecast_year_end_usd": budget["total_forecast_year_end_usd"],
            "scope_conflict_count": budget["scope_conflict_count"],
        },
        "withheld_cost_types": withheld_meta(withhold_cloud),
    }