"""Per-subscription (scope) billing detail — month-anchored rich payload.

Powers the shared ``SubscriptionBillingDrawer`` rendered from both
the employee-facing ``/my/cloud/grants`` table AND the admin-side
``/admin/cloud/overview`` Subscriptions tab. One round-trip returns
everything the drawer needs:

* scope identity (display_name, external_id, provider, owner, budget)
* MTD summary for the requested month: total spend, % of budget,
  daily average, simple month-end projection, fact-data freshness
* daily trend across the entire requested month
* 6-month trailing monthly history (always anchored at the current
  month — independent of the requested ``month`` so the chart stays
  a stable long-trend context view as the user clicks around past
  months in the picker)
* cost-by-service ranking (top N + "other")
* top-N resources for the month (with 14-day sparkline)
* ``coverage`` — 这个月的数字完不完整（"$0 是真的 0" vs "我们没取到
  账单"）。判定规则与它无法区分的那一类见
  :mod:`app.domains.cmdb.cost_coverage`。

Data source: ``cmdb_cost_facts`` — the same ground-truth table the
admin FinOps dashboard, MCP cost tools, and chargeback reports all
read from. Per-scope filtering walks ``scope_id`` for tenant-level
scopes and ``(integration_id, resource_group)`` for sub-tenant scopes
(Azure RG), mirroring the existing ``/cmdb/billing/daily`` branch
split so the drawer's daily-trend agrees with the existing chart
to the cent.

Authorisation: enforced by the router handler — this module assumes
the caller has already been gated on scope membership / ownership /
admin role. The service never calls ``can_*`` predicates itself; it
trusts its callers.

Money handling: USD accumulated in ``Decimal`` end-to-end (cost_facts
column is ``Numeric(14, 6)``). Float boundary conversion at JSON
serialisation only, rounded to 4 dp to match the existing
``cmdb_cloud_billing_daily.total_cost_usd`` precision.

Pairs with the ``_RG_SCOPE_TYPES`` split that ``lifecycle_router``
already maintains for the daily-billing branch. Adding a new RG-typed
scope_type there automatically picks up correct routing here, since
both sites read the same ``_RG_SCOPE_TYPES`` frozenset.
"""
from __future__ import annotations

import calendar
import logging
import re
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import and_, func, select, tuple_
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import ValidationError
from app.domains.cmdb.cost_coverage import assess_scope_coverage
from app.domains.cmdb.cost_view import daily_amount_expr, fact_amount_expr
from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    GRANT_STATUS_PENDING,
    CmdbCloudAccessGrant,
    CmdbCloudBillingDaily,
    CmdbCloudScope,
    CmdbCostFact,
    CmdbResourceCreator,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

logger = logging.getLogger(__name__)


# Precision target for serialised USD: matches
# ``CmdbCloudBillingDaily.total_cost_usd`` (Numeric(14, 4)). A float
# accumulator would drift across many service-rows per day; we keep
# the math in Decimal and only step down to float at the response
# boundary.
_DP_4 = Decimal("0.0001")
_DP_2 = Decimal("0.01")

# Trailing monthly-history window. Six months balances "enough trend
# to spot quarterly drift" against payload size and the page UI which
# shows them as horizontal bars in the drawer. Always ends at the
# current calendar month — see ``_monthly_history`` for why we don't
# slide it with the month picker.
_MONTHLY_HISTORY_WINDOW = 6

# Top-N caps. Generous enough that the drawer never says "12 of 50
# shown" for a typical mid-size scope, but bounded so a tenant with
# thousands of resources doesn't ship a megabyte of JSON per drawer
# open.
_SERVICE_TOPN = 12
_RESOURCES_TOPN = 20
_SPARKLINE_DAYS = 14

# Top-N cap for the per-resource monthly-history breakdown (the
# "each Foundry × each month" stacked view). Smaller than
# ``_RESOURCES_TOPN`` — a stacked bar / matrix with >8 series becomes
# unreadable, and the long tail collapses into an ``__other__`` band so
# the column totals still reconcile with ``_monthly_history``.
_RESOURCE_MONTHLY_TOPN = 8


_MONTH_RE = re.compile(r"^(\d{4})-(0[1-9]|1[0-2])$")


def parse_month_param(month: str | None) -> tuple[int, int]:
    """Validate ``YYYY-MM`` and return ``(year, month)``.

    Strict — single-digit month (``2026-5``) is rejected so a curl /
    MCP caller doesn't get a silently-accepted query with display
    drift on the response. Mirrors the parse in
    :meth:`ApiKeyService.get_assignment_spend` so all spend surfaces
    enforce the same contract.

    Defaults to the current UTC-anchored month when ``None`` —
    matches the ``CmdbCostFact.date`` column (UTC date) so a server
    in any timezone returns the same month.

    Future months are rejected: ``cost_facts`` has no rows past today,
    so a ``2099-12`` query would produce an empty payload with
    ``month_end < month_start`` and a confusing UX. The HTTP route's
    ``Query(pattern=...)`` regex accepts any well-formed YYYY-MM; this
    is the semantic gate that says "no, you can't ask about the
    future." The frontend DatePicker independently caps at the current
    month, so this guard only fires for curl / MCP callers.
    """
    if month is None:
        today = datetime.now(UTC).date()
        return today.year, today.month
    m = _MONTH_RE.match(month)
    if not m:
        raise ValidationError(f"month must be YYYY-MM, got {month!r}")
    year, month_num = int(m.group(1)), int(m.group(2))
    today = datetime.now(UTC).date()
    if (year, month_num) > (today.year, today.month):
        raise ValidationError(
            f"month {month!r} is in the future; cost_facts has no "
            f"data past {today.year:04d}-{today.month:02d}",
        )
    return year, month_num


def _month_bounds(year: int, month_num: int) -> tuple[date, date, bool]:
    """Return (month_start, month_end, is_partial).

    ``month_end`` is clamped to today (UTC) when the requested month
    is the current month — same clamp the AK spend panel uses so the
    drawer's daily-average and projection don't include future days
    that haven't actually accrued any cost yet. ``is_partial`` flags
    that clamp so the drawer can render a "month in progress" hint.

    ``is_partial`` is keyed on *whether the requested month is the current
    calendar month*, NOT on ``full_month_end > today``. On the last day of
    the month those two diverge: ``full_month_end == today`` made the
    in-progress month read as a fully-settled past month, so the drawer
    dropped the "month in progress" hint and the projection switched to the
    full-month-days denominator (instead of anchoring on ``last_fact_date``)
    a day early. The current month is partial on every one of its days — the
    day's spend hasn't fully accrued and the cost-facts cron may lag.
    """
    month_start = date(year, month_num, 1)
    last_day = calendar.monthrange(year, month_num)[1]
    full_month_end = date(year, month_num, last_day)
    today = datetime.now(UTC).date()
    if (year, month_num) == (today.year, today.month):
        # Current (in-progress) month: clamp the window to today and flag it
        # partial regardless of which day of the month it is.
        return month_start, today, True
    return month_start, full_month_end, False


def scope_fact_filters(scope: CmdbCloudScope) -> tuple[list[Any], dict[str, str]]:
    """Build the WHERE clauses + label fields that select cost_facts
    for a given scope.

    Branch matches the ``/billing/daily`` handler:
      * tenant-level scope → ``scope_id`` direct match
      * RG-typed scope     → ``(integration_id, resource_group)`` pair

    Returns the WHERE list AND a small dict describing the chosen
    routing (echoed in the response under ``debug`` so MCP callers can
    sanity-check which branch fired without re-deriving from
    scope_type).
    """
    from app.domains.cmdb.lifecycle_router import _RG_SCOPE_TYPES

    if scope.scope_type in _RG_SCOPE_TYPES:
        # RG-typed scope: cost lives in cost_facts keyed by
        # (integration, resource_group). The scope's external_id IS
        # the RG name (set by the autogrant during creation).
        if not scope.external_id:
            return [], {"routing": "rg", "note": "rg_name_missing"}
        return [
            CmdbCostFact.integration_id == str(scope.integration_id),
            CmdbCostFact.resource_group == scope.external_id,
        ], {"routing": "rg", "rg_name": scope.external_id}
    # Tenant-level scope (AWS account / Azure sub / GCP project /
    # Aliyun account / Tencent account / standalone IAM).
    return [CmdbCostFact.scope_id == str(scope.id)], {"routing": "scope_id"}


async def _scope_summary(
    db: AsyncSession,
    *,
    scope: CmdbCloudScope,
) -> dict[str, Any]:
    """Identity block: scope header info + owner + parent integration.

    No cost data here — this is the chrome the drawer renders in the
    header (provider chip, name, owner avatar, monthly-budget pill).
    """
    integ = (await db.execute(
        select(Integration).where(Integration.id == str(scope.integration_id)),
    )).scalar_one_or_none()

    owner_block: dict[str, Any] | None = None
    if scope.owner_person_id:
        person = (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id == scope.owner_person_id,
            ),
        )).scalar_one_or_none()
        if person:
            owner_block = {
                "person_id": str(person.id),
                "display_name": person.display_name or "",
                "email": person.work_email or "",
            }

    return {
        "id": str(scope.id),
        "external_id": scope.external_id,
        "display_name": scope.display_name,
        "scope_type": scope.scope_type,
        "status": scope.status,
        "environment": scope.environment,
        "integration_id": str(scope.integration_id),
        "integration_name": integ.name if integ else None,
        "provider": integ.provider if integ else None,
        "owner": owner_block,
        "budget_usd_monthly": (
            float(scope.budget_usd_monthly)
            if scope.budget_usd_monthly is not None else None
        ),
        "currency": "USD",
        "last_synced_at": (
            scope.last_synced_at.isoformat() if scope.last_synced_at else None
        ),
    }


async def _members_summary(
    db: AsyncSession,
    *,
    amount_col: Any,
    scope: CmdbCloudScope,
    redact_roles: bool,
) -> list[dict[str, Any]]:
    """Members on this scope — light-weight roster for the drawer.

    Same disclosure semantics as ``/my-subscriptions``: when the
    caller does NOT carry ``can_read_cloud``, the response omits the
    individual ``roles`` list (so a plain employee can see colleagues
    share the scope without learning their privilege level). The
    ``redact_roles`` flag is set by the caller based on the user's
    role set; this module does not re-derive it.

    Excludes status ∉ {active, pending} — expired/revoked rows belong
    on a different tab. Sorted by display name for stable rendering.
    """
    grants = list((await db.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.scope_id == str(scope.id),
            CmdbCloudAccessGrant.status.in_(
                (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING),
            ),
        ),
    )).scalars().all())
    if not grants:
        return []
    person_ids = {g.person_id for g in grants if g.person_id}
    person_map: dict[str, IdentityPerson] = {}
    if person_ids:
        person_rows = (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id.in_(
                    [uuid.UUID(p) for p in person_ids],
                ),
            ),
        )).scalars().all()
        person_map = {str(p.id): p for p in person_rows}

    buckets: dict[str, dict[str, Any]] = {}
    for g in grants:
        if not g.person_id:
            continue
        person = person_map.get(g.person_id)
        if not person:
            continue
        b = buckets.setdefault(g.person_id, {
            "person_id": g.person_id,
            "display_name": person.display_name or "",
            "email": person.work_email or "",
            "roles": [],
            "active_count": 0,
            "pending_count": 0,
        })
        if not redact_roles:
            b["roles"].append(g.role_display_name or g.role_key)
        if g.status == GRANT_STATUS_ACTIVE:
            b["active_count"] += 1
        elif g.status == GRANT_STATUS_PENDING:
            b["pending_count"] += 1
    out = list(buckets.values())
    out.sort(key=lambda m: (m["display_name"] or "").lower())
    return out


async def _daily_trend(
    db: AsyncSession,
    *,
    amount_col: Any,
    fact_where: list[Any],
    month_start: date,
    month_end: date,
) -> tuple[list[dict[str, Any]], Decimal, date | None]:
    """Per-day USD sum across the month, truncated at ``last_fact_date``.

    Returns ``(daily_series, month_total, last_fact_date)``.

    The series ends at ``last_fact_date`` — trailing days for which
    ``cmdb_cost_facts`` has no rows are NOT zero-filled and are NOT
    emitted. Rationale: when the cron has not yet ingested the latest
    day(s), zero-filling rendered as ``$0`` bars on the daily-trend
    chart, which read as "I spent nothing yesterday" — misleading.
    The drawer surfaces the truncation via the ``as_of_date`` field on
    the summary block + the data-lag warning banner; the chart's
    x-axis simply stops at the last day we have data for.

    Mid-month gaps inside ``[month_start, last_fact_date]`` (i.e. a
    day for which the cron failed but later days succeeded) still
    zero-fill — those are rare and the alternative (chart "holes")
    would mislead worse than a single $0 dip.

    ``month_total`` is the Decimal sum of the actually-present days
    (matches what the user sees on the page). ``last_fact_date`` is
    the latest ``date`` we have a row for (None when the month has no
    facts at all) — drives the "freshness lag" hint the drawer shows
    when the ingest cron is behind.
    """
    if not fact_where:
        return [], Decimal("0"), None

    rows = (await db.execute(
        select(
            CmdbCostFact.date,
            func.coalesce(func.sum(amount_col), 0).label("usd"),
        )
        .where(and_(*fact_where, CmdbCostFact.date >= month_start,
                    CmdbCostFact.date <= month_end))
        .group_by(CmdbCostFact.date)
        .order_by(CmdbCostFact.date.asc()),
    )).all()
    per_day: dict[date, Decimal] = {
        r.date: Decimal(str(r.usd or 0)) for r in rows
    }

    if not per_day:
        return [], Decimal("0"), None

    # Truncate to last_fact_date — trailing missing days would zero-fill
    # to misleading "$0 spend" bars otherwise. See docstring.
    last_fact_date = max(per_day.keys())
    series: list[dict[str, Any]] = []
    total = Decimal("0")
    cur = month_start
    while cur <= last_fact_date:
        usd_d = per_day.get(cur, Decimal("0"))
        total += usd_d
        series.append({
            "date": cur.isoformat(),
            "usd": float(usd_d.quantize(_DP_4)),
        })
        cur += timedelta(days=1)

    return series, total, last_fact_date


async def _monthly_history(
    db: AsyncSession,
    *,
    amount_col: Any,
    fact_where: list[Any],
) -> list[dict[str, Any]]:
    """Trailing N-month bar series ending at the current calendar month.

    Deliberately anchored at ``today`` rather than the page's selected
    billing month: the cards / daily-trend chart above already react
    to the month picker, so this chart serves as a STABLE long-trend
    context view. Sliding it with the picker made the rightmost bar
    move in lockstep with the month selector, which read as "the chart
    just shows the selected month" and looked silly when historical
    months had no data ingested yet.

    Single GROUP BY query that buckets facts by (year, month). Faster
    than N separate queries; ``_MONTHLY_HISTORY_WINDOW`` of them
    typically means 1 query vs 6.
    """
    if not fact_where:
        return []

    # Walk back ``_MONTHLY_HISTORY_WINDOW`` months from today.
    today = datetime.now(UTC).date()
    months: list[tuple[int, int]] = []
    y, m = today.year, today.month
    for _ in range(_MONTHLY_HISTORY_WINDOW):
        months.append((y, m))
        m -= 1
        if m == 0:
            m = 12
            y -= 1
    months.reverse()
    earliest_start = date(months[0][0], months[0][1], 1)
    # ``anchor`` is the FULL last-day-of-month, not the clamped
    # month_end — historical months should show their final totals
    # even when the current month is partial.
    last_y, last_m = months[-1]
    anchor_last_day = calendar.monthrange(last_y, last_m)[1]
    latest_end = date(last_y, last_m, anchor_last_day)

    rows = (await db.execute(
        select(
            func.extract("year", CmdbCostFact.date).label("y"),
            func.extract("month", CmdbCostFact.date).label("m"),
            func.coalesce(func.sum(amount_col), 0).label("usd"),
        )
        .where(and_(*fact_where,
                    CmdbCostFact.date >= earliest_start,
                    CmdbCostFact.date <= latest_end))
        .group_by("y", "m"),
    )).all()
    bucket: dict[tuple[int, int], Decimal] = {
        (int(r.y), int(r.m)): Decimal(str(r.usd or 0)) for r in rows
    }
    out: list[dict[str, Any]] = []
    for (y, m) in months:
        usd_d = bucket.get((y, m), Decimal("0"))
        out.append({
            "month": f"{y:04d}-{m:02d}",
            "usd": float(usd_d.quantize(_DP_4)),
        })
    return out


async def _resource_monthly_history(
    db: AsyncSession,
    *,
    amount_col: Any,
    fact_where: list[Any],
) -> dict[str, Any]:
    """Trailing N-month spend broken down per resource (per-Foundry).

    This is the "how much did each Foundry spend each month" view. A
    Foundry (Azure ``Microsoft.CognitiveServices/accounts``) has no
    dedicated inventory table — its cost lives ONLY in ``cmdb_cost_facts``
    keyed by ``resource_id`` / ``resource_name`` (see the removed
    ``CmdbAiResource`` note in models.py), so per-resource grouping over
    the fact table IS the canonical per-Foundry monthly breakdown.

    Returns ``{"months": [...], "resources": [...]}`` where every
    resource entry's ``monthly`` array is aligned index-for-index with
    ``months``. Window matches :func:`_monthly_history` — trailing
    ``_MONTHLY_HISTORY_WINDOW`` months anchored at the current calendar
    month — so this stacked view shares an x-axis with the subscription
    total bars and the columns reconcile.

    Top-``_RESOURCE_MONTHLY_TOPN`` resources by window total are emitted
    individually; the remainder collapse into a single
    ``resource_id=None`` / ``resource_name="__other__"`` row (carrying
    ``rest_count``) so a subscription with a long resource tail still
    sums back to the ``_monthly_history`` total.

    NULL ``resource_id`` rows (tenant-aggregate SKUs — marketplace /
    support / cross-resource discounts) are dropped for the same reason
    as :func:`_top_resources`: they would otherwise form a giant
    "(unknown)" band that dwarfs every real resource.
    """
    if not fact_where:
        return {"months": [], "resources": []}

    # Same window derivation as ``_monthly_history`` — anchored at the
    # current calendar month, ``_MONTHLY_HISTORY_WINDOW`` months back —
    # so the two charts share an x-axis and stack consistently.
    today = datetime.now(UTC).date()
    months: list[tuple[int, int]] = []
    y, m = today.year, today.month
    for _ in range(_MONTHLY_HISTORY_WINDOW):
        months.append((y, m))
        m -= 1
        if m == 0:
            m = 12
            y -= 1
    months.reverse()
    earliest_start = date(months[0][0], months[0][1], 1)
    last_y, last_m = months[-1]
    latest_end = date(last_y, last_m, calendar.monthrange(last_y, last_m)[1])
    month_keys = [f"{yy:04d}-{mm:02d}" for (yy, mm) in months]

    rows = (await db.execute(
        select(
            CmdbCostFact.resource_id,
            CmdbCostFact.integration_id,
            func.max(CmdbCostFact.resource_name).label("name"),
            func.max(CmdbCostFact.service).label("service"),
            func.max(CmdbCostFact.resource_type).label("rtype"),
            func.extract("year", CmdbCostFact.date).label("y"),
            func.extract("month", CmdbCostFact.date).label("m"),
            func.coalesce(func.sum(amount_col), 0).label("usd"),
        )
        .where(and_(
            *fact_where,
            CmdbCostFact.date >= earliest_start,
            CmdbCostFact.date <= latest_end,
            CmdbCostFact.resource_id.is_not(None),
        ))
        .group_by(
            CmdbCostFact.resource_id,
            CmdbCostFact.integration_id,
            "y",
            "m",
        ),
    )).all()
    if not rows:
        return {"months": month_keys, "resources": []}

    # Accumulate per (integration_id, resource_id): window total + a
    # per-month map + descriptive labels. Keyed on the composite pair
    # because resource ids aren't guaranteed unique across integrations
    # (same convention as ``_top_resources``).
    agg: dict[tuple[str, str], dict[str, Any]] = {}
    for r in rows:
        key = (str(r.integration_id), r.resource_id)
        entry = agg.setdefault(key, {
            "resource_id": r.resource_id,
            "integration_id": str(r.integration_id),
            "resource_name": r.name,
            "service": r.service,
            "resource_type": r.rtype,
            "total": Decimal("0"),
            "by_month": {},
        })
        usd_d = Decimal(str(r.usd or 0))
        mk = f"{int(r.y):04d}-{int(r.m):02d}"
        entry["by_month"][mk] = entry["by_month"].get(mk, Decimal("0")) + usd_d
        entry["total"] += usd_d
        # ``resource_name`` / ``service`` are MAX()'d per (resource,
        # month) group, so a NULL can slip through for a month where the
        # billing row lacked the label; keep the first non-null we see.
        if not entry["resource_name"] and r.name:
            entry["resource_name"] = r.name
        if not entry["service"] and r.service:
            entry["service"] = r.service

    ranked = sorted(agg.values(), key=lambda e: e["total"], reverse=True)
    top = ranked[:_RESOURCE_MONTHLY_TOPN]
    rest = ranked[_RESOURCE_MONTHLY_TOPN:]

    resources: list[dict[str, Any]] = []
    for e in top:
        resources.append({
            "resource_id": e["resource_id"],
            "integration_id": e["integration_id"],
            "resource_name": e["resource_name"],
            "service": e["service"],
            "resource_type": e["resource_type"],
            "total_usd": float(e["total"].quantize(_DP_4)),
            "monthly": [
                float(e["by_month"].get(mk, Decimal("0")).quantize(_DP_4))
                for mk in month_keys
            ],
        })
    if rest:
        rest_by_month: dict[str, Decimal] = {}
        rest_total = Decimal("0")
        for e in rest:
            rest_total += e["total"]
            for mk, v in e["by_month"].items():
                rest_by_month[mk] = rest_by_month.get(mk, Decimal("0")) + v
        resources.append({
            "resource_id": None,
            "integration_id": None,
            "resource_name": "__other__",
            "service": None,
            "resource_type": None,
            "total_usd": float(rest_total.quantize(_DP_4)),
            "rest_count": len(rest),
            "monthly": [
                float(rest_by_month.get(mk, Decimal("0")).quantize(_DP_4))
                for mk in month_keys
            ],
        })
    return {"months": month_keys, "resources": resources}


async def _cost_by_service(
    db: AsyncSession,
    *,
    amount_col: Any,
    fact_where: list[Any],
    month_start: date,
    month_end: date,
    total: Decimal,
) -> list[dict[str, Any]]:
    """Top-N services + "other" bucket for the donut/list chart.

    ``share`` is the fraction of month total — pre-computed server-
    side so every consumer (web + MCP) renders the same percentage
    without re-doing the division (and risking a 99.7%-not-100% sum
    from independent rounding).
    """
    if not fact_where:
        return []

    rows = (await db.execute(
        select(
            CmdbCostFact.service,
            func.coalesce(func.sum(amount_col), 0).label("usd"),
        )
        .where(and_(*fact_where, CmdbCostFact.date >= month_start,
                    CmdbCostFact.date <= month_end))
        .group_by(CmdbCostFact.service)
        .order_by(func.sum(amount_col).desc()),
    )).all()
    if not rows:
        return []

    # PR #339 R2: split the 'Savings Plans' row out before slicing
    # top-N. The plain ORDER BY desc sort puts SP-discount rows (the
    # largest negative on linked accounts) at the BOTTOM of the list;
    # without this split, on any scope with > _SERVICE_TOPN services
    # the SP row falls into ``__other__`` and the frontend's
    # "Discounts applied" badge stays hidden — user sees a fully-
    # positive donut alongside a netted total card with no visible
    # reconciliation. Always emit the SP row as its own entry so the
    # FE always has signal to render the badge.
    SP_LABEL = "Savings Plans"
    sp_row = next(
        (r for r in rows if (r.service or "") == SP_LABEL), None,
    )
    non_sp_rows = [r for r in rows if (r.service or "") != SP_LABEL]
    top_rows = non_sp_rows[:_SERVICE_TOPN]
    rest_rows = non_sp_rows[_SERVICE_TOPN:]

    out: list[dict[str, Any]] = []
    for r in top_rows:
        usd_d = Decimal(str(r.usd or 0))
        share = (
            float((usd_d / total).quantize(Decimal("0.0001")))
            if total > 0 else 0.0
        )
        out.append({
            "service": (r.service or "unknown") or "unknown",
            "usd": float(usd_d.quantize(_DP_4)),
            "share": share,
        })
    if rest_rows:
        rest_total = sum(
            (Decimal(str(r.usd or 0)) for r in rest_rows), Decimal("0"),
        )
        rest_share = (
            float((rest_total / total).quantize(Decimal("0.0001")))
            if total > 0 else 0.0
        )
        out.append({
            "service": "__other__",
            "usd": float(rest_total.quantize(_DP_4)),
            "share": rest_share,
            "rest_count": len(rest_rows),
        })
    if sp_row is not None:
        sp_usd_d = Decimal(str(sp_row.usd or 0))
        sp_share = (
            float((sp_usd_d / total).quantize(Decimal("0.0001")))
            if total > 0 else 0.0
        )
        out.append({
            "service": SP_LABEL,
            "usd": float(sp_usd_d.quantize(_DP_4)),
            "share": sp_share,
            "is_discount": float(sp_usd_d) < 0,
        })
    return out


async def _top_resources(
    db: AsyncSession,
    *,
    amount_col: Any,
    fact_where: list[Any],
    month_start: date,
    month_end: date,
) -> list[dict[str, Any]]:
    """Top-N resources within the requested month, with sparkline.

    Two-pass: aggregate first, then fetch 14-day sparkline data for
    just the top rows. Mirrors :func:`cost_service.top_resources` but
    scoped to a single scope's filter — we can't call top_resources
    directly because it doesn't expose the RG-routing branch.

    Sparkline window is the trailing 14 days of the requested month,
    so a past-month query sees daily values for the last 14 days OF
    THAT MONTH (not the calendar's "today minus 14"). Frontend renders
    the strip as a tiny line behind the resource name.

    Filters out NULL ``resource_id`` rows — these are tenant-aggregate
    facts (some providers report account-level rollups when a SKU has
    no resource attribution; they would otherwise collapse into a
    single "(unknown)" mega-row that dwarfs every real resource).
    """
    if not fact_where:
        return []

    first_pass = (await db.execute(
        select(
            CmdbCostFact.resource_id,
            CmdbCostFact.integration_id,
            func.max(CmdbCostFact.resource_name).label("name"),
            func.max(CmdbCostFact.resource_type).label("rtype"),
            func.max(CmdbCostFact.service).label("service"),
            func.max(CmdbCostFact.region).label("region"),
            func.coalesce(func.sum(amount_col), 0).label("total"),
        )
        .where(and_(
            *fact_where,
            CmdbCostFact.date >= month_start,
            CmdbCostFact.date <= month_end,
            CmdbCostFact.resource_id.is_not(None),
        ))
        .group_by(CmdbCostFact.resource_id, CmdbCostFact.integration_id)
        .order_by(func.sum(amount_col).desc())
        .limit(_RESOURCES_TOPN),
    )).all()
    if not first_pass:
        return []

    spark_end = month_end
    spark_start = max(month_start, month_end - timedelta(days=_SPARKLINE_DAYS - 1))
    rid_set = {r.resource_id for r in first_pass}
    int_set = {str(r.integration_id) for r in first_pass}
    second_pass = (await db.execute(
        select(
            CmdbCostFact.resource_id,
            CmdbCostFact.integration_id,
            CmdbCostFact.date,
            func.coalesce(func.sum(amount_col), 0).label("daily"),
        )
        .where(and_(
            *fact_where,
            CmdbCostFact.date >= spark_start,
            CmdbCostFact.date <= spark_end,
            CmdbCostFact.resource_id.in_(rid_set),
            CmdbCostFact.integration_id.in_(int_set),
        ))
        .group_by(
            CmdbCostFact.resource_id,
            CmdbCostFact.integration_id,
            CmdbCostFact.date,
        ),
    )).all()
    sparks: dict[tuple[str, str], dict[date, Decimal]] = {}
    for sr in second_pass:
        # Key by (integration_id, resource_id) — resource ids aren't
        # guaranteed unique across integrations (see top_resources
        # in cost_service.py for the same pattern).
        key = (str(sr.integration_id), sr.resource_id)
        sparks.setdefault(key, {})[sr.date] = Decimal(str(sr.daily or 0))
    spark_dates = [
        spark_start + timedelta(days=i)
        for i in range((spark_end - spark_start).days + 1)
    ]

    # Third pass: latest non-empty tags dict per (integration_id,
    # resource_id). Mirrors the cost_service.top_resources pattern —
    # JSON columns can't be MAX()'d cleanly across MySQL + TiDB, so
    # we order by date DESC and let Python keep the first non-empty
    # entry per key. Hard SQL ``LIMIT`` bounds the worst-case row
    # payload regardless of topN cardinality / window length; the
    # ``tags IS NOT NULL`` filter skips never-tagged rows at the DB
    # layer. Early-termination once every topN resource has a tags
    # dict captured — typically lands on the first few hundred rows.
    _TAG_SCAN_HARD_CAP = 50_000
    tag_keys_by_rid: dict[tuple[str, str], dict[str, Any]] = {}
    # ``(integration_id, resource_id)`` set drawn from the first pass —
    # ``rid_set`` × ``int_set`` is a Cartesian permutation that may
    # include pairs that weren't actually a topN result; constrain
    # the early-termination break and the per-row keep/skip decision
    # to this exact set.
    expected_unique_keys = {(str(r.integration_id), r.resource_id) for r in first_pass}
    tags_pass = (await db.execute(
        select(
            CmdbCostFact.integration_id,
            CmdbCostFact.resource_id,
            CmdbCostFact.tags,
        )
        .where(and_(
            *fact_where,
            CmdbCostFact.date >= month_start,
            CmdbCostFact.date <= month_end,
            CmdbCostFact.resource_id.in_(rid_set),
            CmdbCostFact.integration_id.in_(int_set),
            CmdbCostFact.tags.is_not(None),
            # ``tags IS NOT NULL`` alone is insufficient — ingest
            # writes JSON ``{}`` for untagged facts; skip them at
            # the DB layer so the LIMIT cap doesn't fill with
            # discardable empty-dict rows.
            func.json_length(CmdbCostFact.tags) > 0,
        ))
        .order_by(CmdbCostFact.date.desc())
        .limit(_TAG_SCAN_HARD_CAP),
    )).all()
    for tr in tags_pass:
        key = (str(tr.integration_id), tr.resource_id)
        if key not in expected_unique_keys:
            # ``rid_set`` × ``int_set`` is a permutation bound, so the
            # query may stream rows that satisfy both .in_() filters
            # without actually being a topN result. Skip them.
            continue
        if key in tag_keys_by_rid:
            continue
        td = tr.tags or {}
        if not td or not isinstance(td, dict):
            continue
        tag_keys_by_rid[key] = td
        if len(tag_keys_by_rid) >= len(expected_unique_keys):
            break

    # Fourth pass: per-resource "who created this" attribution from the
    # ``cmdb_resource_creators`` cache. One bulk SELECT keyed on the
    # composite ``(integration_id, resource_id)`` unique index — the
    # top-N is capped at :data:`_RESOURCES_TOPN` so we send a single
    # query with ≤20 tuple members regardless of fleet size. Joined
    # against ``identity_persons`` so we can prefer the Person's
    # canonical display_name over the audit-log-derived
    # ``creator_display_name`` when the email matched a Helix user.
    # Resources with no creator row yet (cron sweep hasn't processed
    # them) get ``creator=None``; the FE renders "—" + "未归属" tooltip
    # for that state.
    expected_keys = {(str(r.integration_id), r.resource_id) for r in first_pass}
    creators_by_key: dict[tuple[str, str], dict[str, Any]] = {}
    if expected_keys:
        creator_rows = (await db.execute(
            select(
                CmdbResourceCreator.integration_id,
                CmdbResourceCreator.resource_id,
                CmdbResourceCreator.status,
                CmdbResourceCreator.creator_email,
                CmdbResourceCreator.creator_display_name,
                CmdbResourceCreator.creator_person_id,
                CmdbResourceCreator.source,
                IdentityPerson.display_name.label("person_display_name"),
            )
            .select_from(CmdbResourceCreator)
            .outerjoin(
                IdentityPerson,
                IdentityPerson.id == CmdbResourceCreator.creator_person_id,
            )
            .where(
                # tuple_().in_() compiles to the row-constructor IN
                # form which TiDB + MySQL 8.0 both accept and is the
                # natural fit for the composite unique index
                # ``uq_resource_creators_integ_rid``.
                tuple_(
                    CmdbResourceCreator.integration_id,
                    CmdbResourceCreator.resource_id,
                ).in_([
                    (i, r) for i, r in expected_keys
                ]),
            ),
        )).all()
        for cr in creator_rows:
            creators_by_key[(str(cr.integration_id), cr.resource_id)] = {
                "status": cr.status,
                "email": cr.creator_email,
                # Prefer the Person's canonical display name when the
                # email matched a Helix user — keeps the column
                # consistent with /admin/users/<id> headers and the
                # ResourceCreatorSection drawer. Falls back to the
                # audit-log-derived name when no Person match.
                "display_name": (
                    cr.person_display_name or cr.creator_display_name
                ),
                "person_id": (
                    str(cr.creator_person_id)
                    if cr.creator_person_id else None
                ),
                "source": cr.source,
            }

    out: list[dict[str, Any]] = []
    for r in first_pass:
        key = (str(r.integration_id), r.resource_id)
        out.append({
            "resource_id": r.resource_id,
            "integration_id": str(r.integration_id),
            "resource_name": r.name,
            "resource_type": r.rtype,
            "service": r.service,
            "region": r.region,
            "total_usd": float(
                Decimal(str(r.total or 0)).quantize(_DP_4),
            ),
            "spark": [
                float(sparks.get(key, {}).get(d, Decimal("0")).quantize(_DP_4))
                for d in spark_dates
            ],
            "tags": tag_keys_by_rid.get(key, {}),
            # ``None`` when no creator row exists yet (cron sweep
            # pending); the FE renders a distinct "未归属" state for
            # this vs. a row with ``status='not_found'`` (we tried
            # but the audit log didn't have a matching create event).
            "creator": creators_by_key.get(key),
        })
    return out


def _resources_count(top_rows: list[dict[str, Any]]) -> int:
    """Cheap caveat — we know AT LEAST N from the top-N pass. Avoids
    a separate COUNT(DISTINCT resource_id) query when the drawer only
    needs an order-of-magnitude hint (`shown: 20 / scope: 35+`)."""
    return len(top_rows)


def _projection(
    month_total: Decimal,
    month_start: date,
    month_end: date,
    year: int,
    month_num: int,
    is_partial: bool,
    last_fact_date: date | None,
) -> dict[str, Any]:
    """Simple linear month-end projection for the current month.

    For past months (``is_partial=False``) the projection equals the
    month total — the month is settled, no projection needed. For the
    current month we extrapolate from the days elapsed.

    ``days_elapsed`` semantics:

      * **Current month** (``is_partial=True``): anchor on
        ``last_fact_date`` when present (or ``month_end`` clamped to
        today if no data yet). Without this, dividing ``month_total``
        by days-up-to-today understates the daily average when the
        cost-facts cron lags behind today (the lagging days contribute
        $0 to the sum but extend the denominator) and that understated
        daily then drives a falsely-low month-end projection.
      * **Past months** (``is_partial=False``): always use the full
        calendar month (``month_end``). Conventional finance reporting
        treats daily-average for March as ``total / 31`` regardless of
        which days have line items — using ``last_fact_date`` here
        would inflate the daily average on partially-backfilled past
        months (e.g. a single $20 fact on 3/5 would render as $4/day
        instead of $0.65/day).

    Trailing-14d smoother (used by ``cost_service.forecast_month_end``)
    would be more accurate but adds another query; for the drawer
    a single linear estimate is enough to convey "are we on track to
    blow the budget."
    """
    last_day = calendar.monthrange(year, month_num)[1]
    days_in_month = last_day
    # Anchor selection — see docstring. Current month uses
    # last_fact_date (when data exists) to keep daily_avg consistent
    # with the truncated daily-trend chart; past months use month_end
    # so the daily_avg matches the conventional total/days_in_month
    # finance figure.
    if is_partial and last_fact_date is not None:
        effective_end = last_fact_date
    else:
        effective_end = month_end
    days_elapsed = max(0, (effective_end - month_start).days + 1)
    daily_average = (
        month_total / Decimal(days_elapsed) if days_elapsed > 0 else Decimal("0")
    )
    if not is_partial:
        # Past month — final number is known.
        projection = month_total
    else:
        # Linear projection: MTD × (days_in_month / days_elapsed).
        if days_elapsed > 0:
            projection = (
                month_total * Decimal(days_in_month) / Decimal(days_elapsed)
            )
        else:
            projection = Decimal("0")
    return {
        "days_in_month": days_in_month,
        "days_elapsed": days_elapsed,
        "daily_average_usd": float(daily_average.quantize(_DP_4)),
        "month_end_projection_usd": float(projection.quantize(_DP_4)),
    }


async def _billing_daily_fallback(
    db: AsyncSession,
    *,
    scope: CmdbCloudScope,
    month_start: date,
    month_end: date,
    measure: str = "list",
) -> tuple[list[dict[str, Any]], Decimal, date | None, str, str | None]:
    """Fallback daily series sourced from ``cmdb_cloud_billing_daily``.

    Fires only when ``cmdb_cost_facts`` has zero rows for the requested
    scope+month — bridges the case where the line-item ingest cron is
    lagging behind the per-day aggregate cron (the two crons run on
    independent schedules and can disagree for hours). Without this
    fallback the page shows ``$0.00`` while the alert (which reads
    billing_daily) fires for the same scope — see bug #3.

    Currency labelling: ``total_cost_usd`` is misnamed for CN clouds —
    rows from Aliyun / Tencent carry CNY in that column. When ANY row's
    ``currency`` field is not "USD", we surface that mixed/native state
    on the response (``currency`` = the row's currency; a note marks
    the unconverted-state caveat) rather than silently presenting CNY
    as USD on the page.

    Returns ``(daily_series, month_total, last_fact_date, currency,
    note)``. ``note`` is non-None only when the rendered currency is
    not USD.
    """
    rows = (await db.execute(
        select(
            CmdbCloudBillingDaily.date,
            # 口径路由:研发面 = COALESCE(list, invoice)(回填过渡语义
            # 见 cost_view 模块 docstring);invoice 面 = 净额原列。
            daily_amount_expr(measure).label("total_cost_usd"),
            CmdbCloudBillingDaily.currency,
        )
        .where(and_(
            CmdbCloudBillingDaily.integration_id == str(scope.integration_id),
            CmdbCloudBillingDaily.scope_id == str(scope.id),
            CmdbCloudBillingDaily.date >= month_start,
            CmdbCloudBillingDaily.date <= month_end,
        ))
        .order_by(CmdbCloudBillingDaily.date.asc()),
    )).all()

    per_day: dict[date, Decimal] = {}
    currencies: set[str] = set()
    # Pre-launch hardening 2026-05-16 follow-up (LS1 P2): track currency
    # per-day so we can detect mixed-currency days and refuse to sum
    # them blindly. Pre-fix, a day with `CNY 100 + USD 50` rows would
    # render as a single `150` number — wrong by an order of magnitude
    # when the currencies actually differ.
    per_day_currencies: dict[date, set[str]] = {}
    for r in rows:
        cur_label = r.currency or "USD"
        currencies.add(cur_label)
        day_set = per_day_currencies.setdefault(r.date, set())
        day_set.add(cur_label)
        if len(day_set) > 1:
            # Mixed-currency day — refuse to sum; renders as 0 with
            # a clear note attached to the response so finance sees
            # "this day is unsummable" instead of a wrong number.
            per_day[r.date] = Decimal("0")
            continue
        per_day[r.date] = per_day.get(r.date, Decimal("0")) + Decimal(
            str(r.total_cost_usd or 0),
        )

    # Decide the rendered currency. If every row is USD (or no rows)
    # the page renders USD. If any row carries a non-USD label, surface
    # the most common one and add a note — for CN clouds the column is
    # actually native CNY despite the column name.
    note: str | None = None
    if not currencies or currencies == {"USD"}:
        rendered_currency = "USD"
    else:
        # Pick the most common non-USD label (typical mixed case is
        # all-CNY or all-USD; mixed is rare but possible if a scope
        # spans two providers via aliasing).
        # Pre-launch hardening 2026-05-16 (LS1 P2): tie-break is
        # deterministic — ``Counter`` + set iteration order had
        # implementation-defined behaviour on ties; sort by
        # ``(-count, code)`` so two providers with equal day counts
        # always render in alphabetical order rather than flickering
        # day-to-day.
        from collections import Counter
        counter = Counter(r.currency or "USD" for r in rows)
        non_usd = sorted(
            (c for c in counter if c != "USD"),
            key=lambda c: (-counter[c], c),
        )
        if non_usd:
            rendered_currency = non_usd[0]
        else:
            rendered_currency = counter.most_common(1)[0][0]
        mixed_days = [
            d.isoformat() for d, codes in per_day_currencies.items()
            if len(codes) > 1
        ]
        if mixed_days:
            note = (
                f"rendered in {rendered_currency} (cost_facts not yet "
                f"synced for USD conversion); {len(mixed_days)} day(s) "
                "had multiple currencies on the same date and render "
                "as 0 to avoid a wrong sum: "
                + ", ".join(sorted(mixed_days)[:5])
                + ("…" if len(mixed_days) > 5 else "")
            )
        else:
            note = (
                f"rendered in {rendered_currency} (cost_facts not yet "
                f"synced for USD conversion)"
            )

    if not per_day:
        return [], Decimal("0"), None, rendered_currency, note

    # Truncate to last_fact_date — see ``_daily_trend`` rationale; same
    # zero-fill-is-misleading concern applies to the fallback path.
    last_fact_date = max(per_day.keys())
    series: list[dict[str, Any]] = []
    total = Decimal("0")
    cur = month_start
    while cur <= last_fact_date:
        usd_d = per_day.get(cur, Decimal("0"))
        total += usd_d
        series.append({
            "date": cur.isoformat(),
            "usd": float(usd_d.quantize(_DP_4)),
        })
        cur += timedelta(days=1)

    return series, total, last_fact_date, rendered_currency, note


async def get_scope_billing_detail(
    db: AsyncSession,
    *,
    scope: CmdbCloudScope,
    month: str | None,
    redact_member_roles: bool,
    measure: str = "list",
) -> dict[str, Any]:
    """Build the full billing-detail payload for ``scope``.

    Pre-condition: caller has already authorised the read (membership,
    ownership, or admin role).

    ``month`` defaults to the current UTC month when ``None``; raises
    :class:`ValidationError` on a malformed value.

    ``redact_member_roles=True`` strips per-member cloud-role lists
    from the member roster (employees viewing other members on a
    shared scope must not learn their privilege level). Admins pass
    ``False``.

    Returns a dict that the router wraps in the standard envelope.
    """
    year, month_num = parse_month_param(month)
    month_start, month_end, is_partial = _month_bounds(year, month_num)
    fact_where, routing = scope_fact_filters(scope)
    # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 =
    # 业务折前成本,COALESCE 过渡语义见 cost_view 模块 docstring。
    amount_col = fact_amount_expr(measure)

    scope_block = await _scope_summary(db, scope=scope)
    daily_series, month_total, last_fact_date = await _daily_trend(
        db,
        amount_col=amount_col,
        fact_where=fact_where,
        month_start=month_start,
        month_end=month_end,
    )

    # ── billing_daily fallback ─────────────────────────────────────
    # cost_facts is the canonical line-item source, but its ingest cron
    # runs on a different schedule than the per-day-aggregate cron that
    # feeds ``cmdb_cloud_billing_daily``. When the line-item cron lags,
    # the page would otherwise show $0 while the daily-spike alert
    # (reading billing_daily) fires the same scope — see bug #3. Fall
    # back to billing_daily for the daily series + month total so the
    # two surfaces agree. RG-typed scopes skip fallback: billing_daily
    # is keyed by (integration, scope_id) and cannot answer
    # (integration, resource_group) queries.
    data_source = "cost_facts"
    data_source_note: str | None = None
    rendered_currency = "USD"
    if (
        month_total == 0
        and last_fact_date is None
        and routing.get("routing") == "scope_id"
    ):
        (
            fb_series, fb_total, fb_last_date, fb_currency, fb_note,
        ) = await _billing_daily_fallback(
            db,
            scope=scope,
            month_start=month_start,
            month_end=month_end,
            measure=measure,
        )
        if fb_last_date is not None:
            # Replace the (empty) cost_facts series with the fallback.
            daily_series = fb_series
            month_total = fb_total
            last_fact_date = fb_last_date
            data_source = "billing_daily_fallback"
            data_source_note = (
                fb_note
                if fb_note is not None
                else "成本明细暂未同步,数字来自每日账单汇总"
            )
            rendered_currency = fb_currency

    # ── 覆盖判定:这个数字(尤其 $0)完不完整 ─────────────────────────
    # 为什么和 ``data_available`` 并存而不是替换它:``data_available``
    # 只回答「这个月有没有行」,对「真的没花钱」和「没取到账单」都是
    # false,读者无从分辨 —— 2026-08-14 三个 GCP 项目被挪到新账单账号后
    # 页面显示 $0.00 整 25 天,没有任何一处提示这不是真的 0(详见
    # ``cost_coverage`` 模块 docstring)。判定跑在**同一批行、同一口径**
    # 上(传入同一个 fact_where / amount_col),否则会出现「页面 $0、
    # 判定说完整」这种自相矛盾。
    coverage = await assess_scope_coverage(
        db,
        integration_id=str(scope.integration_id),
        amount_col=amount_col,
        fact_where=fact_where,
        month_start=month_start,
        month_end=month_end,
    )

    monthly_hist = await _monthly_history(
        db, amount_col=amount_col, fact_where=fact_where,
    )
    resource_monthly = await _resource_monthly_history(
        db, amount_col=amount_col, fact_where=fact_where,
    )
    services = await _cost_by_service(
        db,
        amount_col=amount_col,
        fact_where=fact_where,
        month_start=month_start,
        month_end=month_end,
        total=month_total,
    )
    resources = await _top_resources(
        db,
        amount_col=amount_col,
        fact_where=fact_where,
        month_start=month_start,
        month_end=month_end,
    )
    members = await _members_summary(
        db, amount_col=amount_col, scope=scope,
        redact_roles=redact_member_roles,
    )

    # Recompute projection + budget % off the (possibly-fallback)
    # ``month_total`` so the page's budget bar agrees with the
    # rendered total. If we left these computed off the pre-fallback
    # zero, the drawer would show "$5,505 / 0% of budget" — confusing.
    #
    # Pre-launch hardening 2026-05-16 (CR3 / LS1 P1): when fallback
    # rendered CNY, ``month_total`` is CNY but ``budget_usd_monthly``
    # is USD — naive ratio inflates budget% by ~7×, lighting up every
    # CN-cloud scope's budget bar red. Convert via the global
    # CNY→USD rate before computing the ratio. Projection is rendered
    # in ``rendered_currency`` (CNY for these scopes) so it stays
    # consistent with the drawer's "currency" field.
    month_total_for_budget = month_total
    budget_currency_note: str | None = None
    if rendered_currency != "USD":
        try:
            from app.core.fx_rate import resolve_cny_per_usd
            cny_per_usd = await resolve_cny_per_usd(db)
            if rendered_currency == "CNY" and cny_per_usd > 0:
                month_total_for_budget = month_total / Decimal(
                    str(cny_per_usd),
                )
                budget_currency_note = (
                    f"budget_progress computed after CNY→USD conversion "
                    f"(rate={float(cny_per_usd):.4f})"
                )
            else:
                # Non-CNY, non-USD path — refuse to compute a misleading
                # ratio and surface the gap to the FE.
                month_total_for_budget = None  # type: ignore[assignment]
                budget_currency_note = (
                    f"budget_progress unavailable: {rendered_currency} not"
                    " supported by global FX layer"
                )
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "scope_billing_detail: FX conversion failed for scope %s: %s",
                scope.id, exc,
            )
            month_total_for_budget = None  # type: ignore[assignment]
            budget_currency_note = (
                "budget_progress unavailable: FX rate lookup failed"
            )

    proj = _projection(
        month_total, month_start, month_end,
        year=year, month_num=month_num, is_partial=is_partial,
        last_fact_date=last_fact_date,
    )

    budget_d = (
        Decimal(str(scope.budget_usd_monthly))
        if scope.budget_usd_monthly is not None else Decimal("0")
    )
    if budget_d > 0 and month_total_for_budget is not None:
        budget_progress = float(
            (month_total_for_budget / budget_d * Decimal(100)).quantize(_DP_2),
        )
    else:
        budget_progress = None

    # Pre-launch hardening 2026-05-16 (LS1 P0): when the fallback path
    # synthesises ``total_usd`` from billing_daily, the breakdown
    # queries (services / resources / monthly_hist) still hit cost_facts
    # and return empty. Without a marker the FE renders "$5,505" alongside
    # blank charts — admin ranking sees the same scope as $0 in
    # ``_cost_by_scope_ranking`` (it only reads cost_facts). The
    # ``breakdowns_available`` flag lets the FE render an explicit
    # "明细暂未同步" banner instead of mysterious empties.
    breakdowns_available = data_source == "cost_facts"

    # Round-22 audit: the response previously emitted ``total_usd``
    # carrying a CNY value when ``rendered_currency != "USD"`` (i.e.
    # the billing_daily_fallback path for AliCloud/Tencent CN scopes).
    # The field name is `_usd` but the value is the CNY amount,
    # which inflates the dashboard ~7×. Convert when possible (FX
    # rate already resolved above for the budget pill — reuse it),
    # ALWAYS emit the canonical USD value in ``total_usd``, and add
    # an explicit ``total_native`` + ``native_currency`` block so the
    # FE / CSV / MCP consumers can render the native figure without
    # mis-summing field names.
    total_native_amount = float(month_total.quantize(_DP_4))
    if rendered_currency == "USD":
        total_usd_value: float = total_native_amount
    elif (
        month_total_for_budget is not None
        and rendered_currency == "CNY"
    ):
        # ``month_total_for_budget`` was already converted to USD just
        # above for the budget ratio; reuse so we don't run FX twice.
        total_usd_value = float(month_total_for_budget.quantize(_DP_4))
    else:
        # FX unavailable or unsupported currency — emit ``None`` so the
        # FE can render "USD total unavailable" instead of silently
        # showing the native number under a USD-named field.
        total_usd_value = None  # type: ignore[assignment]

    summary: dict[str, Any] = {
        "total_usd": total_usd_value,
        "total_native": total_native_amount,
        "native_currency": rendered_currency,
        "budget_usd": (
            float(budget_d.quantize(_DP_4)) if budget_d > 0 else None
        ),
        "budget_progress_pct": budget_progress,
        "budget_currency_note": budget_currency_note,
        "currency": rendered_currency,
        "last_fact_date": (
            last_fact_date.isoformat() if last_fact_date else None
        ),
        "data_available": last_fact_date is not None,
        "data_source": data_source,
        "data_source_note": data_source_note,
        "breakdowns_available": breakdowns_available,
        **proj,
    }

    return {
        "scope": scope_block,
        "month": f"{year:04d}-{month_num:02d}",
        "month_window": {
            "start": month_start.isoformat(),
            "end": month_end.isoformat(),
            "is_partial": is_partial,
        },
        # 口径标识:FE 据此渲染「业务折前成本」/「实付」徽标。研发面
        # 恒 'list'(含 COALESCE 过渡语义);invoice 仅保密圈可请求。
        "measure": measure,
        # 覆盖判定。``measured_on`` 说明这个判定针对哪条链路:抽屉在
        # cost_facts 空时会回落 billing_daily 渲染数字,而判定始终跑在
        # cost_facts 上 —— 回落生效时读侧应当优先渲染
        # ``summary.data_source_note``(它已经说明了数字换了来源),别把
        # 一个针对明细链路的 gap 结论贴在一个来自日汇总的数字旁边。
        "coverage": {**coverage.as_dict(), "measured_on": "cost_facts"},
        "summary": summary,
        "daily_trend": daily_series,
        "monthly_history": monthly_hist,
        "resource_monthly_history": resource_monthly,
        "cost_by_service": services,
        "top_resources": resources,
        "members": members,
        "routing": routing,
    }