"""AiSpendingService — 4-bucket cost aggregation + LLM summary. PR #253 (Phase 2 of the AI Spending refactor) rewrites this service to the four-bucket model. Each bucket is backed by exactly one authoritative source: * ``seats`` → ``software_monthly_bills`` JOIN ``asset_software_catalog`` WHERE ``category='ai'`` (Claude.ai, OPEN AI seats today; admin-editable via the seats catalog). * ``bailian`` → ``bailian_usage_daily`` (Aliyun BSS-derived; per-API- key breakdown unavailable — BSS only exposes account-level cost). * ``foundry`` → ``cmdb_cost_facts`` filtered to Azure integration rows whose ``service`` matches ``platform_settings.ai_spending.service_whitelist['azure']``. * ``bedrock`` → ``cmdb_cost_facts`` filtered to AWS integration rows whose ``service`` matches the AWS whitelist entry. A ``vertex`` bucket lives in the data model and renders only when GCP AI facts are present. Replaces the 5-source dedup ladder from PR #079 that mixed scope- level totals with per-service keyword matching. The four buckets are mutually exclusive by source table, so no dedup ladder is required — adding the buckets gives the period total with zero double-count risk. Threading --------- Read-mostly. The two writes are: * ``regenerate_summary(period)`` — calls the Bailian LLM and persists one ``ai_spending_summaries`` row per generation. Rate-limited per-user-per-day via a side-session counter on platform_settings. * ``update_settings(body, ...)`` — admin-editable CNY rate, service whitelist, LLM region/model/key, regen quota. Both writes are audited via :class:`AuditService`. """ from __future__ import annotations import asyncio import hashlib import json import re import time from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any, cast from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import and_, delete, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger from app.domains.ai_spending.models import ( AiSpendingPeriodRollup, AiSpendingSummary, ) from app.domains.ai_spending.schemas import ( AiBucketAttribution, AiBucketAttributionRow, AiServiceWhitelist, AiSpendingBreakdown, AiSpendingBreakdownRow, AiSpendingBucket, AiSpendingByBu, AiSpendingByBuRow, AiSpendingDetail, AiSpendingDetailGroup, AiSpendingDetailGroupRow, AiSpendingDetailRow, AiSpendingForecast, AiSpendingMoverRow, AiSpendingMovers, AiSpendingOverview, AiSpendingSettings, AiSpendingSettingsUpdate, AiSpendingSummaryResponse, AiSpendingTrend, AiSpendingTrendPoint, AiSummaryFinding, AiSummarySuggestion, MoneyPair, quantize, ) from app.domains.ai_spending.whitelist import ( SETTING_KEY as SETTING_SERVICE_WHITELIST, ) from app.domains.ai_spending.whitelist import ( build_service_filter_sql, load_whitelist, validate_whitelist_payload, ) from app.domains.asset.billing_service import ( # The ai_spending domain is a reporting/coordinator-class service # per PLATFORM.md's domain-rule carve-out; reaching across to # asset's billing helpers is sanctioned. _compute_amount, _parse_period, ) from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.domains.bailian.llm_client import ( DEFAULT_MODEL, DEFAULT_REGION, REGION_CN, REGION_INTL, BailianLLMClient, ) from app.domains.bailian.models import ( BAILIAN_AGGREGATE_MODEL, BailianApiKey, BailianUsageDaily, BailianWorkspace, ) from app.domains.cmdb.models import ( CmdbCloudScope, CmdbCostFact, ) from app.identity.bu_helpers import bu_for_user_ids, list_bus from app.identity.person_models import IdentityOrgMembership, IdentityPerson from app.integrations.models import Integration, PlatformSetting logger = get_safe_logger(__name__) # ─── Platform-settings keys ───────────────────────────────────── SETTING_CNY_RATE = "ai_spending.cny_to_usd_rate" SETTING_CNY_RATE_REFRESH_AT = "ai_spending.cny_rate_last_refreshed_at" SETTING_LLM_MODEL = "ai_spending.llm.model" SETTING_LLM_API_KEY = "ai_spending.llm.api_key" SETTING_LLM_REGION = "ai_spending.llm.region" SETTING_REGEN_LIMIT = "ai_spending.regenerate_per_user_per_day" SETTING_REGEN_COUNTER_PREFIX = "ai_spending.regenerate_counter." # Bucket keys (mirrors apikey/spend.py BUCKET_* constants so the two # surfaces use the same vocabulary). BUCKET_SEATS = "seats" BUCKET_BAILIAN = "bailian" BUCKET_FOUNDRY = "foundry" BUCKET_BEDROCK = "bedrock" BUCKET_VERTEX = "vertex" _BUCKET_ORDER: tuple[str, ...] = ( BUCKET_SEATS, BUCKET_BAILIAN, BUCKET_FOUNDRY, BUCKET_BEDROCK, BUCKET_VERTEX, ) # Dimensions the trend chart can stack by. ``bucket`` keeps the legacy # ``by_bucket`` payload the KPI/per-bucket sparklines read; the others # populate ``by_series`` (top-N + folded ``__other__``). ``user`` is # deliberately excluded — hundreds of series would be unreadable; the # leaderboard + drill cover per-person. _TREND_DIMENSIONS: frozenset[str] = frozenset( {"bucket", "model", "service", "department"}, ) _TREND_TOP_N = 8 _TREND_NONE_KEY = "__none__" _TREND_OTHER_KEY = "__other__" _BUCKET_LABELS_EN: dict[str, str] = { BUCKET_SEATS: "SaaS seats", BUCKET_BAILIAN: "Aliyun Bailian", BUCKET_FOUNDRY: "Azure AI Foundry", BUCKET_BEDROCK: "AWS Bedrock", BUCKET_VERTEX: "Google Vertex AI", } _BUCKET_LABELS_ZH: dict[str, str] = { BUCKET_SEATS: "SaaS 坐席", BUCKET_BAILIAN: "阿里云百炼", BUCKET_FOUNDRY: "Azure AI Foundry", BUCKET_BEDROCK: "AWS Bedrock", BUCKET_VERTEX: "Google Vertex AI", } _BAILIAN_BSS_CAVEAT = ( "This is the Aliyun BSS 'Model Studio' product-line cost " "(DescribeInstanceBill, pay-as-you-go), summed per sub-account " "across the Resource Directory. It does NOT include Bailian " "console-metered token usage paid via prepaid 资源包 (resource " "packs) — that consumption is billed separately and is not in the " "BSS product bill, so this figure can be far below the 百炼 " "console / finance total. Reconcile total usage in the Bailian " "console. Owner is resolved via the AK → RAM-user → sub-account " "ladder; per-API-key rows exist for keys Helix can enumerate " "(keys living in Resource Directory MEMBER accounts cannot be " "listed via ListApiKeys, so their cost stays account-attributed)." ) # Provider slug (``integrations.provider``) → bucket key for the # cost_facts buckets. Aliyun + Tencent + UCloud are intentionally # absent — Bailian flows via its own bucket; the other CN clouds # don't host first-class AI products today and treating their cost # rows as AI would over-count. _PROVIDER_TO_BUCKET: dict[str, str] = { "azure": BUCKET_FOUNDRY, "aws": BUCKET_BEDROCK, "gcp": BUCKET_VERTEX, } _BUCKET_TO_PROVIDER_KEY: dict[str, str] = { BUCKET_FOUNDRY: "azure", BUCKET_BEDROCK: "aws", BUCKET_VERTEX: "gcp", } # ─── Seat currency conversion ────────────────────────────────── # # Pre-launch hardening (Round-27, 2026-05-25): seat plans store a # free-form ISO-4217 ``plan.currency`` string. Helix only maintains an # FX rate for CNY (``platform_settings.ai_spending.cny_to_usd_rate``); # any other non-USD plan (HKD / EUR / GBP / JPY / SGD / etc.) used to # fall through to the ``else: usd = native`` branch and be silently # laundered as USD — a ``HKD 1000`` seat showed as ``$1000`` instead of # ~``$128`` (~7-8× inflation). Centralise the conversion so both # call-sites (``_bucket_seats`` for breakdowns + ``_sum_seats_native`` # for movers / by-bu / leaver-leakage) share one rejection path: # unsupported currencies are dropped from the USD aggregate with a # one-shot warning per ``(currency, plan_id)``. Better a visible 0 in # the bucket than a silent 7× inflated total. _SEAT_CURRENCY_USD = "USD" _SEAT_CURRENCY_CNY = "CNY" _SEAT_SUPPORTED_CURRENCIES: frozenset[str] = frozenset( {_SEAT_CURRENCY_USD, _SEAT_CURRENCY_CNY}, ) _SEAT_FX_WARNED: set[tuple[str, str]] = set() def _seat_native_to_usd( *, native: Decimal, currency: str, rate_dec: Decimal, plan_id: str, ) -> Decimal: """Convert one seat row's native price into USD. Returns ``Decimal(0)`` for unsupported currencies (anything other than USD or CNY) and for CNY rows when no FX rate is configured, emitting a one-shot warning per ``(currency, plan_id)`` so noisy plans don't flood logs while still surfacing each new offender. """ if currency == _SEAT_CURRENCY_USD: return native if currency == _SEAT_CURRENCY_CNY: if rate_dec and rate_dec > 0: return native * rate_dec key = (currency, plan_id) if key not in _SEAT_FX_WARNED: _SEAT_FX_WARNED.add(key) logger.warning( "ai_spending: CNY seat without fx rate " "(plan_id=%s); dropping cost from USD aggregate. " "Restore ai_spending.cny_to_usd_rate via " "/admin/ai-spending/settings.", plan_id, ) return Decimal("0") # Unknown currency (HKD/EUR/GBP/JPY/SGD/...) — refuse rather than # silently treat as USD. Pre-launch P0: silent USD launder produced # 7-8× inflated totals for any non-CNY non-USD plan. key = (currency, plan_id) if key not in _SEAT_FX_WARNED: _SEAT_FX_WARNED.add(key) logger.warning( "ai_spending: unsupported plan.currency=%r (plan_id=%s); " "dropping cost from USD aggregate. Either fix the plan " "currency to USD/CNY or extend ai_spending FX coverage.", currency, plan_id, ) return Decimal("0") # ─── Bucket row dataclass ────────────────────────────────────── @dataclass(slots=True) class BucketRow: """One canonical attribution row from a bucket loader. Each bucket loader produces a list of these; the aggregation core groups them by the requested dimension. The shape is deliberately flat — the caller picks the dimensions it cares about and the rest stay None. """ bucket: str date: date cost_usd: Decimal # 折扣前(原价,USD)。填充的桶:Bailian(BSS PretaxGrossAmount)、 # AWS(CUR publicOnDemandCost)、Azure(Retail-Prices 零售价 meter-join)、 # GCP(billing-export ``cost`` = credits 前的 list price,credits 即折扣)。 # None = 无原价口径(AI Seats 固定订阅价、各桶缺原价的个别行),大盘按 # 覆盖率聚合,不伪造折扣。 cost_usd_gross: Decimal | None = None # 优惠后(中间档,= 原价 − 优惠;代金券之前)。目前仅 Bailian 填充。 cost_usd_after_discount: Decimal | None = None # Owner attribution (best-effort per bucket). user_id: str | None = None department: str | None = None # How ``user_id`` was decided. Mirrors ``cmdb_cost_facts.auto_owner_ # source`` for Foundry/Bedrock/Vertex and the Bailian loader's own # ladder for Bailian: ``rule`` / ``scope_owner`` / ``api_key_owner`` # / ``ak_ram_user`` / ``manual`` / ``none``. Surfaced on the wire # so admins can tell "this person actually used it via their AK" # from "they just happen to own the subscription getting the bill". auto_owner_source: str | None = None # Bucket-specific dimensions. Exactly one set per row is the # canonical "subject" for the bucket; the others may also be # populated for richer drill-down. software_id: str | None = None plan_name: str | None = None model_name: str | None = None integration_id: str | None = None scope_id: str | None = None resource_group: str | None = None iam_user: str | None = None service_name: str | None = None aliyun_account_id: str | None = None # Underlying cloud resource identifier (Azure resource ID / AWS # ARN / GCP resource link). Carried through to the detail-row DTO # so the drill drawer can show the exact resource a cost row maps # to instead of just the bucket label. resource_id: str | None = None # Free-form raw payload — used for the LLM prompt + per-row drill # context. Not serialised on the wire. raw: dict[str, Any] = field(default_factory=dict) # ─── Period-row cache ───────────────────────────────────────── # # One dashboard page load fans out to ~10 endpoints in parallel, and # several of those endpoints internally call ``_collect_period_rows`` # 2-3 times for current / previous / yoy windows. The net effect is # ~13 redundant invocations of the same heavy ``cmdb_cost_facts`` # scan per page load. The bucket loaders are read-only and the data # they return is the same regardless of which user / endpoint asked, # so caching the unfiltered row list by ``(period_start, period_end_excl)`` # eliminates the dup-work. # # The cache is process-local (one per worker) with a short TTL — long # enough that a single dashboard load reuses the warm result across # all sibling endpoints, short enough that admin edits to the # whitelist / FX rate become visible within a minute without a manual # bust. ``invalidate_period_rows_cache()`` is exposed for # ``update_settings`` to wipe the cache immediately when an admin # changes inputs that would otherwise stale a hot entry. # # Per-key ``asyncio.Lock`` prevents thundering-herd: when the cache # is cold, only ONE coroutine runs the bucket loaders; the rest wait # on the lock and then read the warm cache. Without this, 13 parallel # sibling endpoints all miss simultaneously and each spawns its own # scan — defeating the cache. _PERIOD_ROWS_TTL_SECONDS: float = 60.0 # Bump when the meaning of a persisted rollup row changes in a way the # stored bytes can't self-describe (e.g. gross-cost semantics, a new # BucketRow dimension that affects aggregation). Part of the rollup # fingerprint so a deploy that changes computation auto-invalidates # every materialised closed-month window without a manual purge. # History: v2 switched Bailian off the QueryAccountBill trusteeship rollup; # v3 (2026-06-07) is the final correct source — BSS QueryBillOverview # (monthly commodity bill, ProductCode=sfm), currency-aware, with the # 优惠后 (after-discount) middle tier added to BucketRow/rollup. Any # rollup materialised under v1/v2 holds wrong Bailian numbers / lacks the # new tier and MUST recompute; the version bump (folded into the rollup # fingerprint) forces it. # v5 (2026-08-15): 百炼加载器不再按「实付为 0」逐行丢弃(优惠券把实付 # 抵到 0 是常态,导致整桶连折前一起消失),并补上 ``model_name != # 'aggregate'`` 过滤(此前被上面那个丢弃意外遮住,否则双计)。任何在 # v4 下物化的已结月份,百炼的行数/折前/折扣后都是错的,必须重算。 _ROLLUP_SEMANTICS_VERSION = "5" # v5: 见上；v4: Bailian → DescribeInstanceBill per-(day×API key×model) # Sentinel bucket value for the single marker row written when a closed # month materialises to ZERO BucketRows (e.g. a month before AI adoption, # within the 13-month lookback). Without it, an empty window is # indistinguishable from "not yet materialised" → it would be re-scanned # live on every read and re-computed by the cron every run, forever. The # marker makes "materialised-but-empty" a cache HIT; ``_read_rollup`` # filters it out so callers still see an empty list. _ROLLUP_EMPTY_MARKER = "__empty__" @dataclass(slots=True) class _CacheEntry: expires_at: float rows: list[BucketRow] _period_rows_cache: dict[tuple[str, str], _CacheEntry] = {} _period_rows_locks: dict[tuple[str, str], asyncio.Lock] = {} _period_rows_locks_guard = asyncio.Lock() def _cache_get(key: tuple[str, str]) -> list[BucketRow] | None: entry = _period_rows_cache.get(key) if entry is None: return None if time.monotonic() > entry.expires_at: _period_rows_cache.pop(key, None) return None return entry.rows def _cache_set(key: tuple[str, str], rows: list[BucketRow]) -> None: _period_rows_cache[key] = _CacheEntry( expires_at=time.monotonic() + _PERIOD_ROWS_TTL_SECONDS, rows=rows, ) async def _period_rows_lock_for( key: tuple[str, str], ) -> asyncio.Lock: """Return the asyncio.Lock for one cache key, creating it on demand inside a global guard so two concurrent callers see the same lock.""" async with _period_rows_locks_guard: lock = _period_rows_locks.get(key) if lock is None: lock = asyncio.Lock() _period_rows_locks[key] = lock return lock def invalidate_period_rows_cache() -> None: """Wipe the entire period-rows cache. Called by ``update_settings`` (whitelist / FX rate edits) so admin changes don't wait out the TTL before the dashboard reflects them. """ _period_rows_cache.clear() # ───────────────────────────────────────────────────────────── class AiSpendingService: """The AI Spending dashboard service — 4-bucket model. Public methods correspond 1:1 to ``ai_spending/router.py`` endpoints; bucket loaders are private and shared across them. """ def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ── Public read API ────────────────────────────────────── async def get_overview( self, *, period: str, bu: str | None = None, ) -> AiSpendingOverview: """Top-of-page summary — grand total + per-bucket breakdown + period-over-period delta + active-user count + leaver- residual on seats. Bailian is INCLUDED in the grand total at the account level but NOT in the active_user_count (BSS limitation). Same posture as the pre-rewrite service so the dashboard headline stays comparable across the refactor. """ period_start, period_end_excl = _parse_period(period) prev_period = _shift_period(period, -1) yoy_period = _shift_period(period, -12) rate = await self._get_cny_rate() cur_rows = await self._collect_period_rows( period_start, period_end_excl, bu=bu, ) prev_start, prev_end_excl = _parse_period(prev_period) prev_rows = await self._collect_period_rows( prev_start, prev_end_excl, bu=bu, ) yoy_start, yoy_end_excl = _parse_period(yoy_period) yoy_rows = await self._collect_period_rows( yoy_start, yoy_end_excl, bu=bu, ) cur_total = _sum_usd(cur_rows) prev_total = _sum_usd(prev_rows) yoy_total = _sum_usd(yoy_rows) buckets = self._buckets_from_rows(cur_rows, prev_rows, rate) # Bailian needs the BSS caveat surfaced; the seats bucket flags a # suspicious (FX-mis-convertible) non-USD plan currency. Imported # lazily — cost_drift imports this service, so a top-level import # would be circular. from app.domains.ai_spending.cost_drift import ( has_suspicious_ai_seat_currency, ) seat_currency_warning = await has_suspicious_ai_seat_currency(self.db) for b in buckets: if b.key == BUCKET_BAILIAN: b.data_caveat = _BAILIAN_BSS_CAVEAT elif b.key == BUCKET_SEATS: b.currency_warning = seat_currency_warning leaver_residual = await self._leaver_residual_usd( period_start, period_end_excl, bu=bu, ) # 折扣前(原价)总额 — 只累加能溯到原价的行;覆盖率/折扣率口径 # 与单桶一致(见 _buckets_from_rows)。覆盖率 <100% 时大盘明示, # 不把部分覆盖的原价误读成全平台原价。 gross_sum, covered_net, has_gross = _sum_gross_usd(cur_rows) grand_total_gross = _money(gross_sum, rate) if has_gross else None grand_discount_pct = ( round(float((gross_sum - covered_net) / gross_sum * 100), 2) if has_gross and gross_sum > 0 else None ) # 与单桶同一条退化规则(见 ``_buckets_from_rows``):实付合计为 0 # 时金额加权是 0/0,按**行数**算覆盖率。2026-08-15 在浏览器上 # 实测抓到 —— 桶级修好之后,KPI 页脚仍显示「折扣前 US$8,404.20 · # 省 100.0%(覆盖 0%)」:同一个公式在总览这一层还留着,而这里 # 恰恰是告诉读者「折扣前这个数有多可信」的地方。 if has_gross and cur_total > 0: gross_coverage_pct = round(float(covered_net / cur_total * 100), 2) elif has_gross and cur_rows: covered_rows = sum( 1 for r in cur_rows if r.cost_usd_gross is not None ) gross_coverage_pct = round(covered_rows / len(cur_rows) * 100, 2) else: gross_coverage_pct = 0.0 # Cost per active user — first-class so the KPI tile + its MoM # pill don't re-derive it (and so MCP/export read the same value). cur_active = _distinct_users(cur_rows) prev_active = _distinct_users(prev_rows) # CPAU numerator must match the denominator's population. The # denominator (``_distinct_users``) counts every row that carries # an owner, so the numerator does the same: spend on rows WITHOUT # an owner has no one to be divided among. # # 2026-08-15: both halves used to exclude the Bailian bucket by # name, on the (since-invalidated) premise that BSS has no # per-user grain. Bailian rows carry real owners now, so the # by-name exclusion just hid attributed people and their dollars # from the KPI. Filtering on ``user_id`` instead keeps numerator # and denominator over one population by construction — no bucket # list to keep in sync the next time a bucket gains attribution. cur_attributable = _sum_usd([r for r in cur_rows if r.user_id]) prev_attributable = _sum_usd([r for r in prev_rows if r.user_id]) cpau = ( cur_attributable / Decimal(cur_active) if cur_active > 0 else None ) prev_cpau = ( prev_attributable / Decimal(prev_active) if prev_active > 0 else None ) cpau_change = ( _pct_change(prev_cpau, cpau) if cpau is not None and prev_cpau is not None else None ) return AiSpendingOverview( period=period, period_start=period_start, period_end=period_end_excl - timedelta(days=1), is_partial_period=(period == _current_period()), grand_total=_money(cur_total, rate), grand_total_gross=grand_total_gross, grand_discount_pct=grand_discount_pct, gross_coverage_pct=gross_coverage_pct, previous_period=prev_period, previous_grand_total=( _money(prev_total, rate) if prev_total > 0 or prev_rows else None ), yoy_period=yoy_period, yoy_grand_total=( _money(yoy_total, rate) if yoy_total > 0 or yoy_rows else None ), mom_change_pct=_pct_change(prev_total, cur_total), yoy_change_pct=_pct_change(yoy_total, cur_total), buckets=buckets, active_user_count=cur_active, cost_per_active_user=( _money(cpau, rate) if cpau is not None else None ), previous_cost_per_active_user=( _money(prev_cpau, rate) if prev_cpau is not None else None ), cost_per_active_user_change_pct=cpau_change, leaver_residual_amount=( _money(leaver_residual, rate) if leaver_residual > 0 else None ), ) async def get_breakdown( self, *, period: str, dimension: str, bu: str | None = None, ) -> AiSpendingBreakdown: period_start, period_end_excl = _parse_period(period) rate = await self._get_cny_rate() rows = await self._collect_period_rows( period_start, period_end_excl, bu=bu, ) # Same-month-last-year rows, grouped by the same dimension, so # each row can carry a YoY delta ("who/what grew vs a year ago"). yoy_period = _shift_period(period, -12) yoy_start, yoy_end_excl = _parse_period(yoy_period) yoy_rows = await self._collect_period_rows( yoy_start, yoy_end_excl, bu=bu, ) yoy_grouped = self._group_by_dimension(yoy_rows, dimension) return await self._breakdown_from_rows( rows, period=period, dimension=dimension, rate=rate, yoy_grouped=yoy_grouped, ) async def get_detail( self, *, period: str, dimension: str, dimension_key: str, bu: str | None = None, ) -> AiSpendingDetail: """Drill into one breakdown bucket — returns the unaggregated rows that compose the chosen dimension_key.""" period_start, period_end_excl = _parse_period(period) rate = await self._get_cny_rate() rows = await self._collect_period_rows( period_start, period_end_excl, bu=bu, ) matches = [ r for r in rows if self._dimension_key_for(r, dimension) == (dimension_key or None) ] # Label maps for human-readable rendering. user_ids = [r.user_id for r in matches if r.user_id] software_ids = [r.software_id for r in matches if r.software_id] scope_ids = [r.scope_id for r in matches if r.scope_id] integ_ids = {r.integration_id for r in matches if r.integration_id} u_map = await self._person_label_map(user_ids) d_map = await self._person_department_map(user_ids) s_map = await self._software_label_map(software_ids) sc_map = await self._scope_label_map(scope_ids) i_map = await self._integration_label_map(integ_ids) # AK linkage maps — let admins jump from a detail row to the # AK request that "owns" it. Foundry rows map by # (integration_id, resource_group); Bedrock by # (integration_id, iam_user); Bailian rows carry the linkage # inline via ``raw["ak_assignment_id"]`` / ``raw["ak_request_ # number"]`` written by the Bailian loader's owner ladder. ak_link_map = await self._build_ak_link_map(matches) detail_rows: list[AiSpendingDetailRow] = [] for r in matches: cny = (r.cost_usd / Decimal(str(rate))) ak_assignment_id: str | None = None ak_request_number: str | None = None ak_owner_label: str | None = None if r.bucket == BUCKET_BAILIAN: ak_assignment_id = r.raw.get("ak_assignment_id") ak_request_number = r.raw.get("ak_request_number") elif r.bucket == BUCKET_FOUNDRY and r.resource_group: hit = ak_link_map.get( ("foundry", r.integration_id, r.resource_group) ) if hit: ak_assignment_id, ak_request_number, ak_owner_label = hit elif r.bucket == BUCKET_BEDROCK and r.iam_user: hit = ak_link_map.get( ("bedrock", r.integration_id, r.iam_user) ) if hit: ak_assignment_id, ak_request_number, ak_owner_label = hit # Bailian: expand the monthly account-level aggregate into the # per-(sub-account × commodity) lines that BSS QueryBillOverview # actually returned (the loader keeps them in ``raw``). The bucket # total is unchanged — this only gives the drill-down real rows # ("which sub-account spent how much on which Model Studio # commodity") instead of one opaque "aggregate". BSS still has no # per-API-key / per-call grain, so that stays unavailable. bailian_items = ( r.raw.get("bss_overview_items") if r.bucket == BUCKET_BAILIAN else None ) if bailian_items: integ_label = ( i_map.get(r.integration_id) if r.integration_id else None ) for it in bailian_items: detail_rows.append(AiSpendingDetailRow( date=r.date, bucket=r.bucket, cost_usd=Decimal(str(it.get("net_usd") or 0)), cost_cny=Decimal(str(it.get("net_cny") or 0)), service_name=( it.get("product_detail") or it.get("commodity_code") ), model_name=it.get("biz_type"), resource_group=( it.get("bill_account_name") or it.get("bill_account_id") ), resource_id=it.get("commodity_code"), integration_id=r.integration_id, integration_name=integ_label, auto_owner_source=r.auto_owner_source, ak_assignment_id=ak_assignment_id, ak_request_number=ak_request_number, ak_owner_label=ak_owner_label, )) continue detail_rows.append(AiSpendingDetailRow( date=r.date, bucket=r.bucket, cost_usd=r.cost_usd, cost_cny=cny, user_id=r.user_id, user_label=u_map.get(r.user_id) if r.user_id else None, department=r.department or d_map.get(r.user_id or ""), software_id=r.software_id, software_label=s_map.get(r.software_id) if r.software_id else None, plan_name=r.plan_name, model_name=r.model_name, service_name=r.service_name, resource_group=r.resource_group, iam_user=r.iam_user, integration_id=r.integration_id, integration_name=( i_map.get(r.integration_id) if r.integration_id else None ), scope_id=r.scope_id, scope_label=sc_map.get(r.scope_id) if r.scope_id else None, auto_owner_source=r.auto_owner_source, resource_id=r.resource_id or r.raw.get("resource_id"), ak_assignment_id=ak_assignment_id, ak_request_number=ak_request_number, ak_owner_label=ak_owner_label, )) total = _sum_usd(matches) # First-row label is good enough — all rows for one # dimension_key share the same upstream label (person id → # display name is stable within a period). Splitting hairs on # rare relabels is not worth the cost. label = self._first_label(matches, dimension, u_map, s_map, i_map, sc_map) groups = self._build_detail_groups( matches, rate, total, u_map, d_map, s_map, sc_map, i_map, ) return AiSpendingDetail( period=period, dimension=dimension, dimension_key=dimension_key, label=label, rows=sorted(detail_rows, key=lambda r: r.date), total=_money(total, rate), groups=groups, ) @staticmethod def _build_detail_groups( matches: list[BucketRow], rate: float, total_usd: Decimal, u_map: dict[str, str], d_map: dict[str, str], s_map: dict[str, str], sc_map: dict[str, str], i_map: dict[str, str], ) -> list[AiSpendingDetailGroup]: """Roll the matched rows up along owner / subscription / service so the drawer can show "按人 / 按子账号 / 按服务" summaries without a second roundtrip. Money sums reuse ``rate`` so they tie out to ``total``. An axis is dropped entirely if NO row carries an identifying value for it (e.g. seats have no cloud scope).""" denom = total_usd if total_usd and total_usd != 0 else None def build( dimension: str, key_of, label_of, secondary_of, ) -> AiSpendingDetailGroup | None: # acc[key] = [usd, count, label, secondary] acc: dict[str | None, list] = {} saw_real_key = False for r in matches: k = key_of(r) if k is not None: saw_real_key = True slot = acc.get(k) if slot is None: acc[k] = [r.cost_usd, 1, label_of(r, k), secondary_of(r, k)] else: slot[0] += r.cost_usd slot[1] += 1 if not saw_real_key: return None rows = [ AiSpendingDetailGroupRow( key=k, label=v[2], secondary=v[3], amount=_money(v[0], rate), row_count=v[1], share=(float(v[0] / denom) if denom else 0.0), ) for k, v in acc.items() ] rows.sort(key=lambda gr: gr.amount.usd, reverse=True) return AiSpendingDetailGroup(dimension=dimension, rows=rows) groups: list[AiSpendingDetailGroup] = [] # 按人 — owner attribution. owner = build( "owner", lambda r: r.user_id, lambda r, k: (u_map.get(k) if k else None), lambda r, k: (r.department or d_map.get(k or "") or None), ) if owner: groups.append(owner) # 按子账号 — cloud scope / subscription, with bucket-specific # fallbacks so Foundry RGs / Bedrock IAM users still group. scope = build( "scope", lambda r: ( r.scope_id or r.resource_group or r.iam_user or r.aliyun_account_id ), lambda r, k: ( sc_map.get(r.scope_id) if r.scope_id else None ) or r.resource_group or r.iam_user or r.aliyun_account_id or k, lambda r, k: ( i_map.get(r.integration_id) if r.integration_id else None ), ) if scope: groups.append(scope) # 按服务 — service / model / software. service = build( "service", lambda r: ( r.service_name or r.model_name or (s_map.get(r.software_id) if r.software_id else None) ), lambda r, k: k, lambda r, k: None, ) if service: groups.append(service) return groups async def get_trend( self, *, granularity: str, lookback_days: int, bu: str | None = None, dimension: str = "bucket", ) -> AiSpendingTrend: if granularity not in ("daily", "monthly"): raise AppException( code="INVALID_GRANULARITY", message=( f"granularity must be 'daily' or 'monthly', got " f"{granularity!r}" ), status_code=400, ) if dimension not in _TREND_DIMENSIONS: raise AppException( code="INVALID_DIMENSION", message=( f"trend dimension must be one of " f"{sorted(_TREND_DIMENSIONS)}, got {dimension!r}" ), status_code=400, ) rate = await self._get_cny_rate() end = datetime.now(ZoneInfo("Asia/Shanghai")).date() start = end - timedelta(days=lookback_days) # +1 because period_end_excl is exclusive rows = await self._collect_period_rows( start, end + timedelta(days=1), bu=bu, ) # ── series key per row (bucket = always present; the other # dimensions fold rows that lack a value into __none__). ── def series_key(r: BucketRow) -> str: if dimension == "bucket": return r.bucket raw = AiSpendingService._dimension_key_for(r, dimension) return raw if raw else _TREND_NONE_KEY # Group cost by (time-bucket → series → USD) and accumulate the # whole-window total per series so we can pick the top-N. by_time: dict[str, dict[str, Decimal]] = {} series_total: dict[str, Decimal] = {} def time_key(r: BucketRow) -> str: if granularity == "daily": return r.date.isoformat() return f"{r.date.year:04d}-{r.date.month:02d}" for r in rows: sk = series_key(r) tk = time_key(r) slot = by_time.setdefault(tk, {}) slot[sk] = slot.get(sk, Decimal("0")) + r.cost_usd series_total[sk] = series_total.get(sk, Decimal("0")) + r.cost_usd # ── choose displayed series + stable order. ── ranked = sorted( series_total, key=lambda k: series_total[k], reverse=True, ) fold = False if dimension == "bucket": chosen = ranked # small fixed set — keep all series_labels: dict[str, str] = {} else: chosen = ranked[:_TREND_TOP_N] fold = len(ranked) > _TREND_TOP_N # Keys are already human-readable; identity label except the # sentinels, which the FE localises. series_labels = { k: k for k in chosen if k != _TREND_NONE_KEY } if _TREND_NONE_KEY in chosen: series_labels[_TREND_NONE_KEY] = "" chosen_set = set(chosen) series_order = list(chosen) if fold: series_order.append(_TREND_OTHER_KEY) series_labels[_TREND_OTHER_KEY] = "" # ── time axis. Daily truncates trailing no-data days (cost-facts # cron lags 1–2 days; zero-filling them reads as a misleading # "$0 AI spend" dip). Monthly just uses populated months. ── if granularity == "daily": last_with_data = ( max(date.fromisoformat(t) for t in by_time) if by_time else None ) effective_end = ( min(end, last_with_data) if last_with_data is not None else None ) time_axis: list[str] = [] if effective_end is not None: cur = start while cur <= effective_end: time_axis.append(cur.isoformat()) cur += timedelta(days=1) else: time_axis = sorted(by_time) points: list[AiSpendingTrendPoint] = [] for tk in time_axis: per = by_time.get(tk, {}) total = sum(per.values(), Decimal("0")) if dimension == "bucket": points.append(AiSpendingTrendPoint( bucket=tk, amount=_money(total, rate), by_bucket={k: float(v) for k, v in per.items()}, )) else: by_series: dict[str, float] = {} top_sum = Decimal("0") for k, v in per.items(): if k in chosen_set: by_series[k] = float(v) top_sum += v if fold: other = total - top_sum if other > 0: by_series[_TREND_OTHER_KEY] = float(other) points.append(AiSpendingTrendPoint( bucket=tk, amount=_money(total, rate), by_series=by_series, )) return AiSpendingTrend( granularity=granularity, points=points, dimension=dimension, series_order=series_order, series_labels=series_labels, ) async def get_movers( self, *, period: str, dimension: str, limit: int, bu: str | None = None, ) -> AiSpendingMovers: period_start, period_end_excl = _parse_period(period) prev_period = _shift_period(period, -1) prev_start, prev_end_excl = _parse_period(prev_period) rate = await self._get_cny_rate() cur_rows = await self._collect_period_rows( period_start, period_end_excl, bu=bu, ) prev_rows = await self._collect_period_rows( prev_start, prev_end_excl, bu=bu, ) cur_by_key = self._group_by_dimension(cur_rows, dimension) prev_by_key = self._group_by_dimension(prev_rows, dimension) all_keys = set(cur_by_key.keys()) | set(prev_by_key.keys()) # Build labels in one pass for efficiency. user_ids = [ k for k in all_keys if dimension in ("user", "owner") and k is not None ] u_map = await self._person_label_map(user_ids) software_ids = [ k for k in all_keys if dimension == "software" and k is not None ] s_map = await self._software_label_map(software_ids) mover_rows: list[AiSpendingMoverRow] = [] for key in all_keys: cur_sum = cur_by_key.get(key, Decimal("0")) prev_sum = prev_by_key.get(key, Decimal("0")) delta = cur_sum - prev_sum label = self._label_for_key( key, dimension, u_map, s_map, ) bucket = self._bucket_for_key(cur_by_key, key, dimension, cur_rows) mover_rows.append(AiSpendingMoverRow( dimension_key=str(key) if key is not None else None, label_en=label or "", label_zh=label or "", current=_money(cur_sum, rate), previous=_money(prev_sum, rate), delta=_money(delta, rate), delta_pct=_pct_change(prev_sum, cur_sum), bucket=bucket, )) jumps = sorted( mover_rows, key=lambda r: r.delta.usd, reverse=True, )[:limit] drops = sorted( mover_rows, key=lambda r: r.delta.usd, )[:limit] return AiSpendingMovers( period=period, previous_period=prev_period, dimension=dimension, jumps=jumps, drops=drops, ) async def get_forecast( self, *, period: str, bu: str | None = None, ) -> AiSpendingForecast: period_start, period_end_excl = _parse_period(period) prev_period = _shift_period(period, -1) prev_start, prev_end_excl = _parse_period(prev_period) rate = await self._get_cny_rate() days_in = (period_end_excl - period_start).days today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date() is_partial = period == _current_period() # Anchor days_elapsed on "yesterday" — today's cost-facts cron # hasn't run yet, so counting today as a full elapsed day would # under-project the daily pace. if is_partial: elapsed_anchor = max(today_sh - timedelta(days=1), period_start) sum_window_end = elapsed_anchor + timedelta(days=1) else: sum_window_end = period_end_excl cur_rows = await self._collect_period_rows( period_start, sum_window_end, bu=bu, ) mtd = _sum_usd(cur_rows) # Round-41 follow-up (2026-05-19): clamp ``days_elapsed`` to the # last calendar day that actually has data, mirroring the trend # series' ``last_with_data`` truncation. Pre-fix, if the # cost-facts cron lagged 2 days behind, ``days_elapsed`` still # counted those calendar days (e.g. mtd ÷ 17 instead of mtd ÷ 15 # on a 17-day-elapsed period) and ``daily_pace`` undercounted # by ~12%, dragging the projected_total below the real burn. # Aligning with the trend keeps the dashboard's three MTD # surfaces (overview tile, forecast tile, trend chart) read # the same elapsed-days window. if is_partial: last_with_data = ( max((r.date for r in cur_rows), default=None) ) anchor = ( min(elapsed_anchor, last_with_data) if last_with_data is not None else elapsed_anchor ) days_elapsed = max((anchor - period_start).days + 1, 1) else: days_elapsed = days_in days_remaining = max(days_in - days_elapsed, 0) # Only usage-metered spend scales with elapsed days. Fixed monthly # SaaS seat subscriptions land as the FULL-month cost at period_start # (``_compute_amount`` does no day-proration for monthly plans), so # extrapolating them as daily burn over-projected the month (≈2× the # seat portion when seats dominate MTD — e.g. $3k/mo seats viewed on # day 15 projected to $6k). Pace ONLY the metered remainder; the # fixed seat cost is already the whole month and must be added once. seat_mtd = _sum_usd([r for r in cur_rows if r.bucket == BUCKET_SEATS]) metered_mtd = mtd - seat_mtd daily_pace_dec = ( metered_mtd / Decimal(days_elapsed) if days_elapsed > 0 else Decimal("0") ) projected = mtd + daily_pace_dec * Decimal(days_remaining) prev_rows = await self._collect_period_rows( prev_start, prev_end_excl, bu=bu, ) prev_total = _sum_usd(prev_rows) return AiSpendingForecast( period=period, is_partial=is_partial, days_in_period=days_in, days_elapsed=days_elapsed, days_remaining=days_remaining, mtd_total=_money(mtd, rate), daily_pace=_money(daily_pace_dec, rate), projected_total=_money(projected, rate), previous_period=prev_period, previous_total=( _money(prev_total, rate) if prev_total > 0 else None ), projected_vs_previous_pct=_pct_change(prev_total, projected), ) async def get_by_bu(self, *, period: str) -> AiSpendingByBu: """Per-BU rollup. Deliberately ignores the page-wide ``bu`` filter — the point of this card is the cross-BU comparison.""" period_start, period_end_excl = _parse_period(period) rate = await self._get_cny_rate() rows = await self._collect_period_rows( period_start, period_end_excl, bu=None, ) bus_total: dict[str, Decimal] = {} bus_users: dict[str, set[str]] = {} unattributed = Decimal("0") user_ids = {r.user_id for r in rows if r.user_id} bu_map = await bu_for_user_ids(self.db, user_ids) for r in rows: if not r.user_id: unattributed += r.cost_usd continue bu_name = bu_map.get(str(r.user_id)) if not bu_name: unattributed += r.cost_usd continue bus_total[bu_name] = bus_total.get(bu_name, Decimal("0")) + r.cost_usd bus_users.setdefault(bu_name, set()).add(r.user_id) all_bus = await list_bus(self.db) member_count_map = {bu: count for bu, count in all_bus} grand_total = sum(bus_total.values(), Decimal("0")) + unattributed result_rows = [] for bu_name, amount in sorted( bus_total.items(), key=lambda kv: kv[1], reverse=True, ): share_pct = ( float(amount / grand_total * 100) if grand_total > 0 else 0.0 ) result_rows.append(AiSpendingByBuRow( bu=bu_name, member_count=member_count_map.get(bu_name, 0), amount=_money(amount, rate), share_pct=round(share_pct, 2), active_user_count=len(bus_users.get(bu_name, set())), )) return AiSpendingByBu( period=period, rows=result_rows, unattributed_amount=_money(unattributed, rate), ) async def get_bucket_attribution( self, *, period: str, bu: str | None = None, ) -> AiBucketAttribution: """Per-AK / per-seat / per-resource chargeback table. Replaces the legacy ``get_resource_chargeback`` which read from ``cmdb_ai_resources`` (deleted in migration 151). The new view aggregates each bucket's natural attribution unit: * Foundry: per ``(integration_id, resource_group)``. * Bedrock: per ``(integration_id, iam_user)``. * Vertex: per ``(integration_id, scope_id)``. * Bailian: per ``(integration_id, aliyun_account_id)`` — accounts not API keys (BSS limitation). * Seats: per ``(software_id, user_id)``. Each row carries the bucket key + bucket-specific identifier + owner (when resolvable) + period amount. """ period_start, period_end_excl = _parse_period(period) rate = await self._get_cny_rate() rows = await self._collect_period_rows( period_start, period_end_excl, bu=bu, ) grouped: dict[tuple, dict[str, Any]] = {} for r in rows: key = self._attribution_key(r) slot = grouped.setdefault(key, { "bucket": r.bucket, "integration_id": r.integration_id, "scope_id": r.scope_id, "resource_group": r.resource_group, "iam_user": r.iam_user, "bailian_workspace": r.raw.get("workspace_label"), "aliyun_account_id": r.aliyun_account_id, "software_id": r.software_id, "user_id": r.user_id, "auto_owner_source": r.auto_owner_source, "amount": Decimal("0"), "plan_name": r.plan_name, }) slot["amount"] += r.cost_usd # Within one attribution key, every row resolves via the # same ladder step (same scope_id / same api_key_pk), so # the slot's owner source is stable. Keep the first non- # ``none`` value to avoid letting an empty fallback row # mask a meaningful source from a later row. if ( slot["auto_owner_source"] in (None, "none") and r.auto_owner_source not in (None, "none") ): slot["auto_owner_source"] = r.auto_owner_source slot["user_id"] = slot["user_id"] or r.user_id # Label maps. user_ids = [s["user_id"] for s in grouped.values() if s["user_id"]] software_ids = [s["software_id"] for s in grouped.values() if s["software_id"]] integ_ids = {s["integration_id"] for s in grouped.values() if s["integration_id"]} scope_ids = [s["scope_id"] for s in grouped.values() if s["scope_id"]] u_map = await self._person_label_map(user_ids) d_map = await self._person_department_map(user_ids) s_map = await self._software_label_map(software_ids) i_map = await self._integration_label_map(integ_ids) sc_map = await self._scope_label_map(scope_ids) # Lookup AK assignment IDs for Foundry / Bedrock so the UI can # link the chargeback row to its AK. from app.domains.apikey.models import ApiKeyAssignment ak_by_rg: dict[tuple[str, str], tuple[str, str]] = {} ak_by_iam: dict[tuple[str, str], tuple[str, str]] = {} rg_keys = [ (s["integration_id"], s["resource_group"]) for s in grouped.values() if s["bucket"] == BUCKET_FOUNDRY and s["integration_id"] and s["resource_group"] ] if rg_keys: # Single query: AKs whose cloud_scope is an Azure RG # whose external_id matches. Filter out revoked AKs — # the holder of a revoked AK has explicitly disclaimed # responsibility for spend after the revoke date; rolling # current Foundry spend onto their head would mis-charge. from app.domains.cmdb.models import ( CmdbCloudScope as _Scope, # local alias ) stmt = ( select( ApiKeyAssignment.id, ApiKeyAssignment.request_number, _Scope.integration_id, _Scope.external_id, ) .join( _Scope, _Scope.id == ApiKeyAssignment.cloud_scope_id, ) .where( _Scope.scope_type == "azure_resource_group", # AK status strings are namespaced — ``pending_approval`` # NOT ``pending``, and ``approved`` is the intermediate # state between approval and key-fill. All three are # legitimate AK holders that should attribute spend # to their owner; only revoked / expired AKs disclaim # responsibility. (Pre-launch round-2 audit P1: the # original filter used literal ``"pending"`` which # matches zero rows.) ApiKeyAssignment.status.in_( ("active", "pending_approval", "approved"), ), ) ) for ak_id, rn, integ, rg in (await self.db.execute(stmt)).all(): ak_by_rg[(str(integ), rg)] = (str(ak_id), rn) iam_keys = [ (s["integration_id"], s["iam_user"]) for s in grouped.values() if s["bucket"] == BUCKET_BEDROCK and s["integration_id"] and s["iam_user"] ] if iam_keys: # Same revoked-AK filter as the RG path above — disclaimed # AKs must not attribute current Bedrock spend. stmt = ( select( ApiKeyAssignment.id, ApiKeyAssignment.request_number, ApiKeyAssignment.iam_integration_id, ApiKeyAssignment.iam_user_arn, ) .where( ApiKeyAssignment.iam_user_arn.isnot(None), # AK status strings are namespaced — ``pending_approval`` # NOT ``pending``, and ``approved`` is the intermediate # state between approval and key-fill. All three are # legitimate AK holders that should attribute spend # to their owner; only revoked / expired AKs disclaim # responsibility. (Pre-launch round-2 audit P1: the # original filter used literal ``"pending"`` which # matches zero rows.) ApiKeyAssignment.status.in_( ("active", "pending_approval", "approved"), ), ) ) from app.domains.apikey.spend import parse_iam_user_name for ak_id, rn, integ, arn in (await self.db.execute(stmt)).all(): if not integ: continue user = parse_iam_user_name(arn) if user: ak_by_iam[(str(integ), user)] = (str(ak_id), rn) out_rows: list[AiBucketAttributionRow] = [] unattributed = Decimal("0") for slot in grouped.values(): owner_pid = slot["user_id"] if not owner_pid: unattributed += slot["amount"] label_en, label_zh = self._attribution_labels(slot, s_map, u_map) ak_id_str = None req_no = None if slot["bucket"] == BUCKET_FOUNDRY and slot["resource_group"]: hit = ak_by_rg.get( (slot["integration_id"], slot["resource_group"]), ) if hit: ak_id_str, req_no = hit elif slot["bucket"] == BUCKET_BEDROCK and slot["iam_user"]: hit = ak_by_iam.get( (slot["integration_id"], slot["iam_user"]), ) if hit: ak_id_str, req_no = hit out_rows.append(AiBucketAttributionRow( bucket=slot["bucket"], assignment_id=(UUID(ak_id_str) if ak_id_str else None), request_number=req_no, integration_id=( UUID(slot["integration_id"]) if slot["integration_id"] else None ), integration_name=i_map.get(slot["integration_id"]) if slot["integration_id"] else None, scope_id=slot["scope_id"], scope_display_name=( sc_map.get(slot["scope_id"]) if slot["scope_id"] else None ), resource_group=slot["resource_group"], iam_user=slot["iam_user"], bailian_workspace=slot["bailian_workspace"], seat_software_id=( UUID(slot["software_id"]) if slot["software_id"] else None ), label_en=label_en, label_zh=label_zh, owner_person_id=UUID(owner_pid) if owner_pid else None, owner_label=u_map.get(owner_pid) if owner_pid else None, owner_department=d_map.get(owner_pid) if owner_pid else None, auto_owner_source=slot.get("auto_owner_source"), period_amount=_money(slot["amount"], rate), )) out_rows.sort(key=lambda r: r.period_amount.usd, reverse=True) return AiBucketAttribution( period=period, rows=out_rows, unattributed_amount=_money(unattributed, rate), ) # ── Settings ────────────────────────────────────────────── async def get_settings(self) -> AiSpendingSettings: cny = await self._get_cny_rate() refreshed = await self._get_setting_value(SETTING_CNY_RATE_REFRESH_AT) wl_raw = await self._get_setting_value(SETTING_SERVICE_WHITELIST) or {} regen_limit = int( await self._get_setting_value(SETTING_REGEN_LIMIT) or 5, ) model = await self._get_setting_str(SETTING_LLM_MODEL) or DEFAULT_MODEL api_key = await self._get_setting_str(SETTING_LLM_API_KEY) or "" region = await self._get_setting_str(SETTING_LLM_REGION) or DEFAULT_REGION refreshed_dt: datetime | None = None if refreshed: try: refreshed_dt = datetime.fromisoformat(str(refreshed)) except (TypeError, ValueError): refreshed_dt = None wl = AiServiceWhitelist( azure=wl_raw.get("azure", []) if isinstance(wl_raw, dict) else [], aws=wl_raw.get("aws", []) if isinstance(wl_raw, dict) else [], gcp=wl_raw.get("gcp", []) if isinstance(wl_raw, dict) else [], ) return AiSpendingSettings( cny_to_usd_rate=cny, cny_rate_last_refreshed_at=refreshed_dt, llm_model=model, llm_region=region, llm_api_key_present=bool(api_key.strip()), service_whitelist=wl, regenerate_per_user_per_day=regen_limit, ) async def update_settings( self, body: AiSpendingSettingsUpdate, *, actor_id: UUID | str, trace_id: str, ) -> AiSpendingSettings: changed: dict[str, Any] = {} if body.cny_to_usd_rate is not None: # Align write-side validation with fx_rate.py read-side band # [0.05, 1.0] so a value accepted here is never silently rejected # by get_usd_per_cny(), which would break every CNY cloud-cost # ingestion with no obvious error message. from app.core.fx_rate import _USD_PER_CNY_MAX, _USD_PER_CNY_MIN if not (_USD_PER_CNY_MIN <= body.cny_to_usd_rate <= _USD_PER_CNY_MAX): raise AppException( code="INVALID_RATE", message=( f"cny_to_usd_rate must be in " f"[{_USD_PER_CNY_MIN}, {_USD_PER_CNY_MAX}] " f"(USD per CNY, e.g. 0.14 for 1 USD = 7.14 CNY)" ), status_code=400, ) await self._set_setting( SETTING_CNY_RATE, float(body.cny_to_usd_rate), actor_id=actor_id, ) changed["cny_to_usd_rate"] = body.cny_to_usd_rate if body.llm_model is not None: await self._set_setting( SETTING_LLM_MODEL, body.llm_model, actor_id=actor_id, ) changed["llm_model"] = body.llm_model if body.llm_region is not None: normalized = body.llm_region.strip().lower() if normalized not in (REGION_CN, REGION_INTL): raise AppException( code="INVALID_LLM_REGION", message=( f"llm_region must be one of {REGION_CN!r} or " f"{REGION_INTL!r}; got {body.llm_region!r}." ), status_code=400, ) await self._set_setting( SETTING_LLM_REGION, normalized, actor_id=actor_id, ) changed["llm_region"] = normalized if body.llm_api_key is not None: await self._set_setting( SETTING_LLM_API_KEY, body.llm_api_key, actor_id=actor_id, ) changed["llm_api_key"] = "(redacted)" if body.service_whitelist is not None: try: validated = validate_whitelist_payload( body.service_whitelist.model_dump(), ) except ValueError as exc: raise AppException( code="INVALID_SERVICE_WHITELIST", message=str(exc), status_code=400, ) from exc await self._set_setting( SETTING_SERVICE_WHITELIST, validated, actor_id=actor_id, ) changed["service_whitelist"] = validated if body.regenerate_per_user_per_day is not None: await self._set_setting( SETTING_REGEN_LIMIT, int(body.regenerate_per_user_per_day), actor_id=actor_id, ) changed["regenerate_per_user_per_day"] = ( body.regenerate_per_user_per_day ) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="ai_spending.settings.update", resource_type="platform_setting", resource_id="ai_spending", status="success", detail={"changed_keys": list(changed.keys())}, ) # Bust the period-row cache when admin edits inputs that would # otherwise leave the dashboard showing pre-edit aggregates for # up to ``_PERIOD_ROWS_TTL_SECONDS``. FX rate and whitelist are # the two inputs that materially change the bucket loaders' # output; LLM key/model/regen-limit do not affect aggregation. if ( "cny_to_usd_rate" in changed or "service_whitelist" in changed ): invalidate_period_rows_cache() return await self.get_settings() async def test_llm_connection(self) -> dict[str, Any]: """Ping the configured Bailian LLM with a minimal prompt.""" api_key = await self._get_setting_str(SETTING_LLM_API_KEY) or "" if not api_key.strip(): raise AppException( code="BAILIAN_LLM_NO_KEY", message=( "No Bailian API key configured. Save a sk-xxx key " "first, then test." ), status_code=400, ) model = await self._get_setting_str(SETTING_LLM_MODEL) or DEFAULT_MODEL region = await self._get_setting_str(SETTING_LLM_REGION) or DEFAULT_REGION client = BailianLLMClient( api_key=api_key, model=model, region=region, timeout=20.0, purpose="ai_spending.analysis") start = datetime.now(UTC) try: text = await client.chat_text( system="You are a connectivity probe.", user="Reply with the literal word: pong", max_tokens=8, temperature=0.0, ) except AppException as exc: return { "ok": False, "region": region, "endpoint": client.endpoint, "model": model, "latency_ms": int( (datetime.now(UTC) - start).total_seconds() * 1000 ), "code": exc.code, "message": exc.message, } elapsed = int((datetime.now(UTC) - start).total_seconds() * 1000) return { "ok": True, "region": region, "endpoint": client.endpoint, "model": model, "latency_ms": elapsed, "sample_reply": (text or "")[:80], "message": "ok", } # ── LLM summary ────────────────────────────────────────── async def get_summary( self, *, period: str, ) -> AiSpendingSummaryResponse: row = (await self.db.execute( select(AiSpendingSummary).where( AiSpendingSummary.period == period, ).order_by(AiSpendingSummary.generated_at.desc()).limit(1) )).scalar_one_or_none() if row is None: return AiSpendingSummaryResponse(period=period, available=False) return _summary_to_response(row) async def regenerate_summary( self, *, period: str | None, actor: IdentityPerson, trace_id: str, ) -> AiSpendingSummaryResponse: """Manual regenerate trigger (admin-only). Per-user-per-day rate-limited via the side-session counter.""" if period is None: first_of_month = ( datetime.now(ZoneInfo("Asia/Shanghai")).date().replace(day=1) ) last_month_end = first_of_month - timedelta(days=1) period = f"{last_month_end.year:04d}-{last_month_end.month:02d}" if not _PERIOD_RE.fullmatch(period): raise AppException( code="INVALID_PERIOD", message=f"period must be YYYY-MM, got {period!r}", status_code=400, ) await self._write_attempt_audit_independent( trace_id=trace_id, actor_user_id=actor.id, period=period, ) await self._consume_regen_quota(str(actor.id)) # Build the prompt from the new 4-bucket data. overview = await self.get_overview(period=period) prev_overview: AiSpendingOverview | None = None try: prev_overview = await self.get_overview( period=overview.previous_period or period, ) except Exception: # noqa: BLE001 logger.warning( "ai_spending: previous-period overview lookup failed; " "continuing without it", exc_info=True, ) breakdown_user = await self.get_breakdown( period=period, dimension="user", ) breakdown_dept = await self.get_breakdown( period=period, dimension="department", ) breakdown_model = await self.get_breakdown( period=period, dimension="model", ) breakdown_bucket = await self.get_breakdown( period=period, dimension="bucket", ) attribution = await self.get_bucket_attribution(period=period) prompt = _build_summary_prompt( overview=overview, prev_overview=prev_overview, breakdown_user=breakdown_user, breakdown_dept=breakdown_dept, breakdown_model=breakdown_model, breakdown_bucket=breakdown_bucket, attribution=attribution, ) api_key = await self._get_setting_str(SETTING_LLM_API_KEY) or "" if not api_key.strip(): raise AppException( code="BAILIAN_LLM_NO_KEY", message="No Bailian API key configured.", status_code=400, ) model = await self._get_setting_str(SETTING_LLM_MODEL) or DEFAULT_MODEL region = await self._get_setting_str(SETTING_LLM_REGION) or DEFAULT_REGION client = BailianLLMClient( api_key=api_key, model=model, region=region, timeout=120.0, purpose="ai_spending.analysis") try: llm_json = await client.chat_json( system=_SUMMARY_SYSTEM_PROMPT, user=prompt, max_tokens=4096, temperature=0.2, ) except AppException: raise except Exception as exc: # noqa: BLE001 raise AppException( code="BAILIAN_LLM_FAILED", message=f"Bailian LLM call failed: {exc}", status_code=502, ) from exc summary_row = await self._persist_summary( period=period, actor_id=actor.id, llm_json=llm_json, model=model, aggregates={ "grand_total_usd": overview.grand_total.usd, "buckets": [ {"key": b.key, "usd": b.amount.usd, "share_pct": b.share_pct} for b in overview.buckets ], }, ) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="ai_spending.summary.generate", resource_type="ai_spending_summary", resource_id=period, status="success", detail={"period": period, "model": model}, ) await self.db.flush() return _summary_to_response(summary_row) async def auto_generate_summary( self, *, period: str, trace_id: str, ) -> AiSpendingSummaryResponse: """Scheduled-job trigger — no actor, no rate-limit.""" if not _PERIOD_RE.fullmatch(period): raise AppException( code="INVALID_PERIOD", message=f"period must be YYYY-MM, got {period!r}", status_code=400, ) overview = await self.get_overview(period=period) prev_overview: AiSpendingOverview | None = None try: prev_overview = await self.get_overview( period=overview.previous_period or period, ) except Exception: # noqa: BLE001 logger.warning( "ai_spending auto: previous-period overview lookup failed; " "continuing without it", exc_info=True, ) breakdown_user = await self.get_breakdown(period=period, dimension="user") breakdown_dept = await self.get_breakdown(period=period, dimension="department") breakdown_model = await self.get_breakdown(period=period, dimension="model") breakdown_bucket = await self.get_breakdown(period=period, dimension="bucket") attribution = await self.get_bucket_attribution(period=period) prompt = _build_summary_prompt( overview=overview, prev_overview=prev_overview, breakdown_user=breakdown_user, breakdown_dept=breakdown_dept, breakdown_model=breakdown_model, breakdown_bucket=breakdown_bucket, attribution=attribution, ) api_key = await self._get_setting_str(SETTING_LLM_API_KEY) or "" if not api_key.strip(): raise AppException( code="BAILIAN_LLM_NO_KEY", message="No Bailian API key configured.", status_code=400, ) model = await self._get_setting_str(SETTING_LLM_MODEL) or DEFAULT_MODEL region = await self._get_setting_str(SETTING_LLM_REGION) or DEFAULT_REGION client = BailianLLMClient( api_key=api_key, model=model, region=region, timeout=120.0, purpose="ai_spending.analysis") try: llm_json = await client.chat_json( system=_SUMMARY_SYSTEM_PROMPT, user=prompt, max_tokens=4096, temperature=0.2, ) except AppException: raise except Exception as exc: # noqa: BLE001 raise AppException( code="BAILIAN_LLM_FAILED", message=f"Bailian LLM call failed: {exc}", status_code=502, ) from exc summary_row = await self._persist_summary( period=period, actor_id=None, llm_json=llm_json, model=model, aggregates={ "grand_total_usd": overview.grand_total.usd, "buckets": [ {"key": b.key, "usd": b.amount.usd, "share_pct": b.share_pct} for b in overview.buckets ], }, ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="ai_spending.summary.generate", resource_type="ai_spending_summary", resource_id=period, status="success", detail={"period": period, "model": model, "automatic": True}, ) await self.db.flush() return _summary_to_response(summary_row) # ── Bucket loaders (private) ───────────────────────────── async def _collect_period_rows( self, period_start: date, period_end_excl: date, *, bu: str | None = None, ) -> list[BucketRow]: """Run all bucket loaders and merge into a single canonical row list. Each row carries its bucket tag so downstream aggregation can group / filter. Cached in the module-level period-row cache (60s TTL) keyed by ``(period_start, period_end_excl)`` — see ``_PERIOD_ROWS_TTL_ SECONDS`` block above for the rationale. The cache key is the UNFILTERED window; the optional ``bu`` filter is applied after the cache lookup so endpoints with different ``bu`` arguments all share the same underlying fetch. """ if period_end_excl <= period_start: return [] key = (period_start.isoformat(), period_end_excl.isoformat()) rows = _cache_get(key) if rows is None: lock = await _period_rows_lock_for(key) async with lock: # Double-check after acquiring the lock: a concurrent # coroutine may have just populated the cache while we # were waiting. rows = _cache_get(key) if rows is None: rows = await self._load_period_rows( period_start, period_end_excl, ) _cache_set(key, rows) if bu: rows = await self._filter_rows_by_bu(rows, bu) return rows async def _load_period_rows( self, period_start: date, period_end_excl: date, ) -> list[BucketRow]: """L2 of the read path: rollup-first (closed AND open month), else live compute. Both the closed and the **current/open** month are served from ``ai_spending_period_rollup`` when a current-fingerprint materialisation exists. This is the perf fix (2026-06): the underlying cost data only changes once a day (cost_facts / bailian sync run daily, not intraday), so serving the open month from the cron-maintained rollup is at most ~1 day stale — a fine trade for not re-scanning ~2.7M ``cmdb_cost_facts`` rows on every dashboard open (the default view IS the current month, which the old "open month always live" rule left slow). A rollup miss (not yet materialised, or whitelist/FX changed → fingerprint mismatch) falls back to live compute — correct, just not yet accelerated; the cron refreshes current + recent months each run. The refresh cron is the ONLY writer, so reads never commit / never race. """ fp = await self._rollup_fingerprint() cached = await self._read_rollup(period_start, period_end_excl, fp) if cached is not None: return cached return await self._compute_period_rows(period_start, period_end_excl) async def _compute_period_rows( self, period_start: date, period_end_excl: date, ) -> list[BucketRow]: """Run the three bucket loaders live and merge. The expensive path (scans ``cmdb_cost_facts``) — the rollup exists to avoid re-running this for immutable closed months.""" seats = await self._bucket_seats(period_start, period_end_excl) bailian = await self._bucket_bailian(period_start, period_end_excl) cost_facts = await self._bucket_cost_facts( period_start, period_end_excl, ) return seats + bailian + cost_facts async def _seat_plan_fingerprint(self) -> str: """Per-plan ``(price, currency, billing_cycle)`` digest for the AI-category seat plans, folded into ``_rollup_fingerprint``. The seats bucket is **NOT frozen**: ``_bucket_seats`` reads each plan's ``price`` / ``currency`` / ``billing_cycle`` live on every recompute (then converts to USD via FX), so an admin edit to any of those fields changes what a live recompute returns for *every* month the plan's seats span — closed months included. Before this was in the fingerprint, such an edit left the rollup untouched and the dashboard kept serving the pre-edit seat USD until the refresh cron happened to rebuild that window for an unrelated reason (e.g. a currency mis-entry surfaced months late as understated AI-seat spend). Folding the plan fields in makes the edit flip the fingerprint → old-fingerprint rows become a read miss → live fallback now + cron re-materialises every window on its next run. """ ai_software_ids = (await self.db.execute( select(AssetSoftwareCatalog.id).where( AssetSoftwareCatalog.category == "ai", ) )).scalars().all() if not ai_software_ids: return "" rows = (await self.db.execute( select( SoftwareLicensePlan.id, SoftwareLicensePlan.price, SoftwareLicensePlan.currency, SoftwareLicensePlan.billing_cycle, ) .where(SoftwareLicensePlan.software_id.in_(ai_software_ids)) .order_by(SoftwareLicensePlan.id) )).all() parts = [ f"{pid}:{price}:{(currency or '').upper()}:{cycle or ''}" for pid, price, currency, cycle in rows ] return ";".join(parts) async def _rollup_fingerprint(self) -> str: """Hash the inputs that change what a live recompute would return: the AI service whitelist, the CNY→USD FX rate, the AI seat-plan price/currency/cycle digest, and the rollup-semantics version. The rollup is a faithful cache of the live compute, so the fingerprint must cover everything the live compute reads that isn't frozen in the source rows. FX is included **because the seats bucket is NOT frozen**: ``_bucket_seats`` converts CNY plan prices to USD with the live ``_get_cny_rate()`` on every compute, so an FX change shifts a closed month's seat USD. (Bailian / cost_facts USD *are* frozen at ingest, so for those buckets FX is a no-op — but the fingerprint is per-window across all buckets, so we include FX to keep the rollup from ever diverging from a live recompute. An FX change is weekly and the cron rebuilds, so the extra invalidation is cheap.) The seat-plan digest is included for the same reason — see ``_seat_plan_fingerprint``: the seats bucket reads plan price/currency/cycle live, so an edit there must invalidate the rollup just as an FX change does.""" wl = await load_whitelist(self.db) rate = await self._get_cny_rate() seat_plans = await self._seat_plan_fingerprint() payload = ( json.dumps(wl, sort_keys=True, default=str) + "|fx=" + format(float(rate or 0), ".10g") + "|seatplans=" + seat_plans + "|v=" + _ROLLUP_SEMANTICS_VERSION ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() @staticmethod def _rollup_to_bucketrow(m: AiSpendingPeriodRollup) -> BucketRow: return BucketRow( bucket=m.bucket, date=m.fact_date, cost_usd=Decimal(str(m.cost_usd)), cost_usd_gross=( Decimal(str(m.cost_usd_gross)) if m.cost_usd_gross is not None else None ), cost_usd_after_discount=( Decimal(str(m.cost_usd_after_discount)) if m.cost_usd_after_discount is not None else None ), user_id=m.user_id, department=m.department, auto_owner_source=m.auto_owner_source, software_id=m.software_id, plan_name=m.plan_name, model_name=m.model_name, integration_id=m.integration_id, scope_id=m.scope_id, resource_group=m.resource_group, iam_user=m.iam_user, service_name=m.service_name, aliyun_account_id=m.aliyun_account_id, resource_id=m.resource_id, raw=dict(m.raw_json or {}), ) async def _read_rollup( self, period_start: date, period_end_excl: date, fingerprint: str, ) -> list[BucketRow] | None: """Return materialised rows for a window at the current fingerprint, or None when not materialised (a miss → live fallback). A window that genuinely computed to zero rows still carries a single ``_ROLLUP_EMPTY_MARKER`` row, so it reads back as a HIT with an empty list (filtered out below) rather than being re-scanned live forever.""" mats = (await self.db.execute( select(AiSpendingPeriodRollup).where( AiSpendingPeriodRollup.period_start == period_start, AiSpendingPeriodRollup.period_end_excl == period_end_excl, AiSpendingPeriodRollup.whitelist_fingerprint == fingerprint, ) )).scalars().all() if not mats: return None return [ self._rollup_to_bucketrow(m) for m in mats if m.bucket != _ROLLUP_EMPTY_MARKER ] async def _rollup_window_exists( self, period_start: date, period_end_excl: date, fingerprint: str, ) -> bool: hit = (await self.db.execute( select(AiSpendingPeriodRollup.id).where( AiSpendingPeriodRollup.period_start == period_start, AiSpendingPeriodRollup.period_end_excl == period_end_excl, AiSpendingPeriodRollup.whitelist_fingerprint == fingerprint, ).limit(1) )).first() return hit is not None async def refresh_period_rollup( self, period_start: date, period_end_excl: date, ) -> int: """Recompute a window live and replace its materialisation. Window = unit of replacement: DELETE every row for ``(period_start, period_end_excl)`` (all fingerprints — this also evicts stale-fingerprint rows from a prior whitelist) then bulk INSERT the freshly computed rows tagged with the current fingerprint. Flushes but does not commit — the caller (refresh cron) owns the transaction. Returns the number of rows written. **The caller MUST hold** ``ROLLUP_WRITE_LOCK`` (see ``app.domains.ai_spending.locks``). This used to say "single-writer by construction: only the period_rollup_refresh cron calls this, and the scheduled-jobs runner single-flights it" — that stopped being true on 2026-07-28 when the drift-check cron gained an auto-heal path that also re-materialises windows (#2083). The runner's single-flight lock is keyed ``sched_job:{job_id}``, i.e. PER JOB, so two different crons are not serialised against each other; the refresh (06:30) and the drift check (07:00) only look 30 minutes apart, and the concurrency gate can delay either one. Two overlapping DELETE+INSERTs on the same window do not error — ``ai_spending_period_rollup`` has no unique constraint, only ``ix_ai_rollup_window_fp`` — they leave DUPLICATE rows, and every read SUMs them, so the dashboard silently double-counts. Hence a real lock instead of a comment. """ fp = await self._rollup_fingerprint() rows = await self._compute_period_rows(period_start, period_end_excl) await self.db.execute( delete(AiSpendingPeriodRollup).where( AiSpendingPeriodRollup.period_start == period_start, AiSpendingPeriodRollup.period_end_excl == period_end_excl, ) ) for r in rows: self.db.add(AiSpendingPeriodRollup( period_start=period_start, period_end_excl=period_end_excl, whitelist_fingerprint=fp, bucket=r.bucket, fact_date=r.date, cost_usd=r.cost_usd, cost_usd_gross=r.cost_usd_gross, cost_usd_after_discount=r.cost_usd_after_discount, user_id=r.user_id, department=r.department, auto_owner_source=r.auto_owner_source, software_id=r.software_id, plan_name=r.plan_name, model_name=r.model_name, integration_id=r.integration_id, scope_id=r.scope_id, resource_group=r.resource_group, iam_user=r.iam_user, service_name=r.service_name, aliyun_account_id=r.aliyun_account_id, resource_id=r.resource_id, raw_json=dict(r.raw or {}), )) if not rows: # Materialised-but-empty: a single marker row so this window # reads back as a HIT (empty list) instead of a miss → no # endless live re-scan / daily cron recompute for a genuinely # zero-spend month. ``_read_rollup`` filters the marker out. self.db.add(AiSpendingPeriodRollup( period_start=period_start, period_end_excl=period_end_excl, whitelist_fingerprint=fp, bucket=_ROLLUP_EMPTY_MARKER, fact_date=period_start, cost_usd=Decimal("0"), )) await self.db.flush() return len(rows) async def manual_refresh_rollup( self, *, period: str | None, actor_id: UUID, trace_id: str, ) -> dict[str, Any]: """Admin-triggered re-materialisation of one (or the recent two) closed month(s) — the reconciliation cost-lens "重算物化" fix for ``stale_fingerprint`` / ``not_materialized`` drift findings. Validates that an explicit ``period`` is a CLOSED month (strictly before the current Asia/Shanghai month — open months are served live and never materialised, so refreshing one is a no-op error, 400). With no period, refreshes the most-recent 2 closed months (same window the daily cron warms). Returns ``{periods: [...], rows_written}``. Holds ``ROLLUP_WRITE_LOCK`` around the rebuild — this path is the THIRD writer of rollup windows (after the refresh cron and the drift-check heal, #2085), reachable from both the HTTP router and the ``refresh_ai_spending_rollup`` MCP tool. Overlapping an admin click with either cron would leave duplicate rows every read SUMs. When another writer holds the lock, raises 409 ``AI_ROLLUP_REFRESH_BUSY`` — the running writer rebuilds at the current fingerprint anyway, so "retry in a few minutes" is the honest answer. **Commits its own work INSIDE the lock** (2026-07-29 review): a writer that only flushes and lets the caller commit after the ``async with`` exits does not actually get mutual exclusion — the lock is released while the rows are still uncommitted, so the next writer acquires it and starts its own DELETE+INSERT on the same window before the first writer's rows land. That is exactly the duplicate-row / double-count scenario the lock exists to prevent. The audit row is written inside the same transaction so the record and the rows commit atomically. Callers may still call ``commit()`` afterwards — it is a harmless no-op (this method leaves nothing pending) and error semantics are unchanged: a failure anywhere in the loop raises before the commit and rolls the whole rebuild back. """ # Anchor "current month" to Asia/Shanghai so the open-month guard # matches what _default_period / the dashboard consider "this # month" (the worker runs UTC; finance reads CN time). current_period = _current_period() if period is not None: if period >= current_period: raise AppException( code="ROLLUP_OPEN_MONTH", message=( f"{period} is the current/future month — open " "months are computed live and never materialised. " "Only closed months can be re-materialised." ), status_code=400, ) periods = [period] else: # Most-recent two closed months (i=1,2 relative to current). periods = [_shift_period(current_period, -1), _shift_period(current_period, -2)] from app.domains.ai_spending.locks import ( RollupLockBusy, rollup_write_lock, ) total_rows = 0 try: async with rollup_write_lock(): for p in periods: start, end_excl = _parse_period(p) total_rows += await self.refresh_period_rollup( start, end_excl, ) await AuditService(self.db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="ai_spending.rollup.manual_refresh", status="success", resource_type="ai_spending_period_rollup", resource_id=None, detail={"periods": periods, "rows_written": total_rows}, ) # Mutual exclusion must cover the COMMIT, not just the # writes — see the docstring. await self.db.commit() except RollupLockBusy as exc: raise AppException( code="AI_ROLLUP_REFRESH_BUSY", message=( "The rollup is being rebuilt right now (daily refresh " "cron or drift-check heal in progress) — it rebuilds " "at the current fingerprint, so this manual refresh " "is redundant while it runs. Try again in a few " "minutes." ), status_code=409, ) from exc return {"periods": periods, "rows_written": total_rows} async def prune_rollup_before(self, cutoff_start: date) -> int: """Delete materialised windows whose exclusive end is on/before ``cutoff_start`` — i.e. windows entirely older than the retained lookback. Bounds the table so stale-fingerprint generations left by past whitelist/FX edits on aged-out months don't accumulate forever. Flushes; caller owns the commit. Returns rows deleted.""" res = await self.db.execute( delete(AiSpendingPeriodRollup).where( AiSpendingPeriodRollup.period_end_excl <= cutoff_start, ) ) await self.db.flush() # ``rowcount`` lives on CursorResult; AsyncSession.execute is typed # as Result[Any], so cast to read it without a mypy attr-defined. return int(cast("Any", res).rowcount or 0) async def _bucket_seats( self, period_start: date, period_end_excl: date, ) -> list[BucketRow]: """Read AI seat allocations from ``software_monthly_bills`` JOIN ``asset_software_catalog``.""" ai_software_ids = (await self.db.execute( select(AssetSoftwareCatalog.id).where( AssetSoftwareCatalog.category == "ai", ) )).scalars().all() if not ai_software_ids: return [] start_dt = datetime.combine( period_start, datetime.min.time(), tzinfo=UTC, ) end_dt = datetime.combine( period_end_excl, datetime.min.time(), tzinfo=UTC, ) result = (await self.db.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan, IdentityPerson) .join( SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id, ) .outerjoin( IdentityPerson, SoftwareSeatAllocation.user_id == IdentityPerson.id, ) .where( SoftwareLicensePlan.software_id.in_(ai_software_ids), SoftwareSeatAllocation.allocated_at < end_dt, SoftwareSeatAllocation.status.in_(("active", "revoked")), or_( SoftwareSeatAllocation.revoked_at.is_(None), and_( SoftwareSeatAllocation.revoked_at >= start_dt, SoftwareSeatAllocation.revoked_at > SoftwareSeatAllocation.allocated_at, ), ), ) )).all() rate = await self._get_cny_rate() rate_dec = Decimal(str(rate)) if rate else Decimal("0") person_dept_ids = [ p.id for _alloc, _plan, p in result if p is not None ] dept_map = await self._person_department_map(person_dept_ids) # Dedup overlapping allocations: when a user is revoked and # re-added on the same plan within the period, both rows survive # the WHERE and both compute the full seat-month price = double # cost. The plan price covers ONE seat-month regardless of how # many allocation rows existed in that month. Keep the most # recent allocation (highest allocated_at) per (plan, user) # tuple. Anonymous rows (person is None) keep their own # allocation_id as the key so they don't accidentally collapse. dedup_map: dict[tuple[str, str], tuple[Any, Any, Any]] = {} for alloc, plan, person in result: user_key = str(person.id) if person is not None else f"anon:{alloc.id}" dedup_key = (str(plan.id), user_key) existing = dedup_map.get(dedup_key) if existing is None or alloc.allocated_at > existing[0].allocated_at: dedup_map[dedup_key] = (alloc, plan, person) out: list[BucketRow] = [] for alloc, plan, person in dedup_map.values(): native = _compute_amount( price=plan.price, billing_cycle=plan.billing_cycle, period_start=period_start, period_end_excl=period_end_excl, allocated_at=alloc.allocated_at, ) if native == 0: continue currency = (plan.currency or _SEAT_CURRENCY_USD).upper() usd = _seat_native_to_usd( native=native, currency=currency, rate_dec=rate_dec, plan_id=str(plan.id), ) if usd == 0: # Unsupported currency or missing FX (already logged). # Skip to avoid landing a 0-USD row that confuses # per-plan attribution; the warning surfaces the gap. continue # Anchor the day to the first day-of-month so trend bucketing # has something coherent for seat-row daily aggregation. out.append(BucketRow( bucket=BUCKET_SEATS, date=period_start, cost_usd=usd, user_id=str(person.id) if person else None, department=dept_map.get(str(person.id)) if person else None, software_id=str(plan.software_id), plan_name=plan.name, raw={"allocation_id": str(alloc.id)}, )) return out async def _bucket_bailian( self, period_start: date, period_end_excl: date, ) -> list[BucketRow]: """Read Bailian usage from ``bailian_usage_daily`` and attach per-row owner attribution via a three-level ladder: 1. **AK owner** (``source='api_key_owner'``) — when the usage row has ``api_key_pk`` populated and the referenced ``BailianApiKey`` has ``owner_person_id`` set. This is the strongest signal: someone's specific AK consumed the cost. 2. **Sub-account owner** (``source='scope_owner'``) — when the usage row carries an ``aliyun_account_id`` that matches a ``cmdb_cloud_scopes`` row of type ``aliyun_account`` with an owner. The cost is attributed to the sub-account owner as a BSS-only-granularity fallback (the canonical case today). 3. **None** (``source='none'``) — no AK match, no scope match. Row stays unattributed and lands in the dashboard's "未分摊" bucket so admins can chase it. The historical version of this loader set ``user_id=None`` for every row, which forced the entire Bailian bucket into the "aggregate / 未分摊" slot in by-user breakdowns even though the sub-account owner was always discoverable. PR #263 introduces the ladder so the dashboard shows real names on Bailian rows and the drill-down can distinguish AK-attributed from scope-owner fallback attribution. """ result = (await self.db.execute( select(BailianUsageDaily).where( BailianUsageDaily.usage_date >= period_start, BailianUsageDaily.usage_date < period_end_excl, # ``model_name='aggregate'`` 是同一天的**汇总行**,与逐 # model 明细行并存。不排掉就是双计。此前被下面的「实付 # 为 0 就跳过」意外遮住了(生产上百炼实付恒 0,汇总行和 # 明细行一起被丢),口径一改就会立刻显形。 # ``apikey/spend.py::_spend_bailian`` 一直是这么过滤的。 BailianUsageDaily.model_name != BAILIAN_AGGREGATE_MODEL, ) )).scalars().all() # Bulk-resolve workspace pk → display label so the per-row # ``bailian_workspace`` (used by the attribution table) is # populated when BSS gives us workspace-level rows. The # account-level fallback path leaves the label None and the # UI falls back to aliyun_account_id. workspace_pks = { row.workspace_pk for row in result if row.workspace_pk } ws_label_map: dict[str, str] = {} if workspace_pks: ws_rows = (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.id.in_(workspace_pks), ) )).scalars().all() ws_label_map = { str(w.id): w.name or w.workspace_id for w in ws_rows } # ── Owner ladder, step 1: AK → owner_person_id ─────────── ak_pks = {row.api_key_pk for row in result if row.api_key_pk} # Map api_key_pk (str) → # (owner_person_id | None, assignment_id | None, request_number | None) ak_owner_map: dict[str, tuple[str | None, str | None, str | None]] = {} if ak_pks: from app.domains.apikey.models import ApiKeyAssignment as ApiKeyAssn ak_stmt = ( select( BailianApiKey.id, BailianApiKey.owner_person_id, BailianApiKey.assignment_id, ApiKeyAssn.request_number, ) .outerjoin(ApiKeyAssn, ApiKeyAssn.id == BailianApiKey.assignment_id) .where(BailianApiKey.id.in_(ak_pks)) ) for ak_id, owner_pid, ass_id, req_no in ( await self.db.execute(ak_stmt) ).all(): ak_owner_map[str(ak_id)] = ( str(owner_pid) if owner_pid else None, str(ass_id) if ass_id else None, req_no, ) # ── Owner ladder, step 2: aliyun_account_id → sub-account # owner via ``cmdb_cloud_scopes`` (scope_type='aliyun_account' # OR an Alibaba-side sub-account scope, both intl + domestic). # Single batched query — group by integration_id so we can # filter ``external_id`` per integration without hitting an N+1. # Map (integration_id, aliyun_account_id) → # (scope_id, owner_person_id | None) scope_owner_map: dict[tuple[str, str], tuple[str, str | None]] = {} integ_to_accounts: dict[str, set[str]] = {} for row in result: if not (row.integration_id and row.aliyun_account_id): continue integ_to_accounts.setdefault( str(row.integration_id), set(), ).add(row.aliyun_account_id) if integ_to_accounts: conds = [ and_( CmdbCloudScope.integration_id == integ, CmdbCloudScope.external_id.in_(accounts), ) for integ, accounts in integ_to_accounts.items() ] scope_stmt = select( CmdbCloudScope.id, CmdbCloudScope.integration_id, CmdbCloudScope.external_id, CmdbCloudScope.owner_person_id, ).where( CmdbCloudScope.scope_type == "aliyun_account", CmdbCloudScope.deleted_at.is_(None), or_(*conds), ) for scope_id, integ, ext, owner_pid in ( await self.db.execute(scope_stmt) ).all(): scope_owner_map[(str(integ), ext)] = ( str(scope_id), str(owner_pid) if owner_pid else None, ) # Department lookup batched once across both owner sources. person_ids: set[str] = set() for owner_pid, _ass, _req in ak_owner_map.values(): if owner_pid: person_ids.add(owner_pid) for _sid, owner_pid in scope_owner_map.values(): if owner_pid: person_ids.add(owner_pid) dept_map = await self._person_department_map(list(person_ids)) out: list[BucketRow] = [] for row in result: cost = Decimal(str(row.cost_usd or 0)) gross = ( Decimal(str(row.cost_usd_gross)) if row.cost_usd_gross is not None else None ) after_disc = ( Decimal(str(row.cost_usd_after_discount)) if row.cost_usd_after_discount is not None else None ) # 三口径**全为空/零**才是真的「这行没有钱」。 # # 2026-08-15 修复:这里原本是 ``if cost == 0: continue`` —— # 只看实付。百炼的实付被优惠券/合同折扣抵到 0 是**常态** # (生产 2026-06/07/08 实付分别 $20.15 / $2.56 / $0.00,而 # 折前是 $3,226 / $4,345 / $8,404),于是整个桶连同它的折前 # 与折扣后数据被逐行丢光:AI 大盘上百炼不是显示「折后 $0」, # 而是 fact_count=0、distinct_subjects=0、amount_gross=None, # 整张卡片形同没有数据,也不进 grand_total_gross。 if cost == 0 and not gross and not after_disc: continue ws_label = ( ws_label_map.get(str(row.workspace_pk)) if row.workspace_pk else None ) # Walk the ladder. Stop at the first level that yields a # non-None owner so the source field unambiguously reflects # which level matched. user_id: str | None = None auto_owner_source: str | None = None matched_scope_id: str | None = None raw_extra: dict[str, Any] = {} ak_pk_str = str(row.api_key_pk) if row.api_key_pk else None if ak_pk_str and ak_pk_str in ak_owner_map: owner_pid, ass_id, req_no = ak_owner_map[ak_pk_str] if owner_pid: user_id = owner_pid auto_owner_source = "api_key_owner" if ass_id: raw_extra["ak_assignment_id"] = ass_id if req_no: raw_extra["ak_request_number"] = req_no if user_id is None and row.integration_id and row.aliyun_account_id: hit = scope_owner_map.get( (str(row.integration_id), row.aliyun_account_id), ) if hit is not None: sid, owner_pid = hit matched_scope_id = sid if owner_pid: user_id = owner_pid auto_owner_source = "scope_owner" if user_id is None: auto_owner_source = "none" # Carry the per-(sub-account × commodity) breakdown the loader # stored in ``raw_data`` through to BucketRow.raw so ``get_detail`` # can expand the monthly aggregate into real rows — AND so the # closed-month rollup (which persists BucketRow.raw → raw_json) # keeps it too. Without this propagation the drill-down only ever # sees the hand-built subset below and falls back to one opaque # "aggregate" row. src_raw = row.raw_data if isinstance(row.raw_data, dict) else {} bss_items = src_raw.get("bss_overview_items") out.append(BucketRow( bucket=BUCKET_BAILIAN, date=row.usage_date, cost_usd=cost, cost_usd_gross=gross, cost_usd_after_discount=after_disc, user_id=user_id, department=dept_map.get(user_id) if user_id else None, auto_owner_source=auto_owner_source, model_name=row.model_name, integration_id=str(row.integration_id), scope_id=matched_scope_id, aliyun_account_id=row.aliyun_account_id, raw={ "input_tokens": row.input_tokens, "output_tokens": row.output_tokens, "workspace_label": ws_label, **({"bss_overview_items": bss_items} if bss_items else {}), **raw_extra, }, )) return out async def _bucket_cost_facts( self, period_start: date, period_end_excl: date, ) -> list[BucketRow]: """Read AI-classified cost_facts (Foundry / Bedrock / Vertex). Applies the per-provider whitelist filter. Rows are tagged with the bucket key derived from the integration's provider slug. Owner attribution uses the fact's denormalised ``cost_owner_person_id`` (populated at ingest via :class:`CmdbCostAllocationRule` + scope-owner fallback). """ whitelist = await load_whitelist(self.db) # Build per-bucket subqueries — TiDB does fine with a single # union, but doing per-bucket lets us tag rows cheaply on the # Python side and avoids OR-of-(integ AND service) costs. cur_integrations = (await self.db.execute( select(Integration).where(Integration.kind == "cloud") )).scalars().all() # Map integration_id → (bucket, provider) integ_bucket: dict[str, str] = {} for integ in cur_integrations: b = _PROVIDER_TO_BUCKET.get(integ.provider) if b is not None: integ_bucket[str(integ.id)] = b out: list[BucketRow] = [] for bucket_key, provider_key in ( (BUCKET_FOUNDRY, "azure"), (BUCKET_BEDROCK, "aws"), (BUCKET_VERTEX, "gcp"), ): patterns = whitelist.get(provider_key) or [] if not patterns: continue target_integ_ids = [ iid for iid, b in integ_bucket.items() if b == bucket_key ] if not target_integ_ids: continue service_filter = build_service_filter_sql( CmdbCostFact.service, patterns, ) # Column projection, not whole ORM rows (#2189-9). A month # window is tens of thousands of facts and ``raw`` alone can # be ~16 KB each — loading it to read eleven scalar columns # costs tens of MB per call for nothing. The columns below # are exactly the ones the loop touches; adding a new ``f.x`` # read means adding it here too, which is the point (a # missing column raises AttributeError immediately rather # than quietly returning None). stmt = select( CmdbCostFact.date, CmdbCostFact.integration_id, CmdbCostFact.scope_id, CmdbCostFact.service, CmdbCostFact.resource_id, CmdbCostFact.resource_group, CmdbCostFact.amount_usd, CmdbCostFact.amount_usd_gross, CmdbCostFact.cost_owner_person_id, CmdbCostFact.cost_owner_department, CmdbCostFact.auto_owner_source, ).where( CmdbCostFact.integration_id.in_(target_integ_ids), CmdbCostFact.date >= period_start, CmdbCostFact.date < period_end_excl, service_filter, ) facts = (await self.db.execute(stmt)).all() if not facts: continue # Person department lookup in one batch. person_ids = list({f.cost_owner_person_id for f in facts if f.cost_owner_person_id}) dept_map = await self._person_department_map(person_ids) for f in facts: usd = Decimal(str(f.amount_usd or 0)) if usd == 0: continue iam_user = None if bucket_key == BUCKET_BEDROCK: iam_user = _parse_owner_from_resource_id(f.resource_id) out.append(BucketRow( bucket=bucket_key, date=f.date, cost_usd=usd, cost_usd_gross=( Decimal(str(f.amount_usd_gross)) if f.amount_usd_gross is not None else None ), user_id=f.cost_owner_person_id, department=( f.cost_owner_department or dept_map.get(f.cost_owner_person_id or "") ), auto_owner_source=f.auto_owner_source, integration_id=str(f.integration_id), scope_id=f.scope_id, resource_group=f.resource_group, iam_user=iam_user, service_name=f.service, resource_id=f.resource_id, raw={"resource_id": f.resource_id}, )) return out # ── BU filter ──────────────────────────────────────────── async def _filter_rows_by_bu( self, rows: list[BucketRow], bu: str | None, ) -> list[BucketRow]: if not bu or not bu.strip(): return rows target = bu.strip().lower() user_ids = {r.user_id for r in rows if r.user_id} bu_map = await bu_for_user_ids(self.db, user_ids) out: list[BucketRow] = [] for r in rows: if not r.user_id: continue if (bu_map.get(str(r.user_id)) or "").lower() == target: out.append(r) return out # ── Aggregation helpers ────────────────────────────────── def _buckets_from_rows( self, cur: list[BucketRow], prev: list[BucketRow], rate: float, ) -> list[AiSpendingBucket]: cur_totals: dict[str, Decimal] = {} cur_counts: dict[str, int] = {} cur_subjects: dict[str, set[str]] = {} # 折扣前(原价)聚合。``cur_gross`` 只累加有原价的行; # ``cur_gross_net`` 是这些**同一批行**的折后额 —— 折扣率必须 # 用同口径的 net 比 gross,否则覆盖率<100% 的桶(Azure 历史) # 会把无原价行的 net 也算进去,虚高折扣。 cur_gross: dict[str, Decimal] = {} cur_gross_net: dict[str, Decimal] = {} cur_after_disc: dict[str, Decimal] = {} # 优惠后(中间档) # 有原价的行数 —— 实付恒 0 的桶(百炼:优惠券把实付抵到 0) # 用它按行数算覆盖率,金额加权在 0/0 上没有意义。见下。 cur_gross_counts: dict[str, int] = {} for r in cur: cur_totals[r.bucket] = cur_totals.get(r.bucket, Decimal("0")) + r.cost_usd cur_counts[r.bucket] = cur_counts.get(r.bucket, 0) + 1 subj = self._subject_key_for_bucket(r) if subj is not None: cur_subjects.setdefault(r.bucket, set()).add(subj) if r.cost_usd_gross is not None: cur_gross[r.bucket] = ( cur_gross.get(r.bucket, Decimal("0")) + r.cost_usd_gross ) cur_gross_net[r.bucket] = ( cur_gross_net.get(r.bucket, Decimal("0")) + r.cost_usd ) cur_gross_counts[r.bucket] = ( cur_gross_counts.get(r.bucket, 0) + 1 ) if r.cost_usd_after_discount is not None: cur_after_disc[r.bucket] = ( cur_after_disc.get(r.bucket, Decimal("0")) + r.cost_usd_after_discount ) prev_totals: dict[str, Decimal] = {} for r in prev: prev_totals[r.bucket] = prev_totals.get(r.bucket, Decimal("0")) + r.cost_usd grand = sum(cur_totals.values(), Decimal("0")) result: list[AiSpendingBucket] = [] for key in _BUCKET_ORDER: amount = cur_totals.get(key, Decimal("0")) prev_amount = prev_totals.get(key, Decimal("0")) if amount == 0 and prev_amount == 0 and key == BUCKET_VERTEX: # Skip the Vertex card entirely when the tenant has no # GCP AI usage in either period. continue share = float(amount / grand * 100) if grand > 0 else 0.0 # ── 折扣前 / 折扣额 / 折扣率 ────────────────────────── amount_gross: MoneyPair | None = None amount_after_discount: MoneyPair | None = None discount_amount: MoneyPair | None = None discount_pct: float | None = None gross_coverage_pct = 0.0 ad = cur_after_disc.get(key) if ad is not None: amount_after_discount = _money(ad, rate) gross = cur_gross.get(key) if gross is not None and gross > 0: covered_net = cur_gross_net.get(key, Decimal("0")) amount_gross = _money(gross, rate) discount_amount = _money(gross - covered_net, rate) # 带符号,不 clamp:credit/退款致 net>gross 时为负折扣,如实显示。 discount_pct = float((gross - covered_net) / gross * 100) # 该桶折后额里有多少比例能溯到原价(Bailian=100%, # Azure 历史可能 <100%)。 # # 实付为 0 的桶按**行数**退化计算(2026-08-15):百炼的 # 优惠券把实付整段抵到 0,金额加权就成了 0/0 —— 原式的 # ``else 0.0`` 会把「每一行都有原价」谎报成「覆盖率 0%」, # 而 KPI 页脚正是拿这个数告诉读者折扣前的数有多可信。 # 行数口径在这种情形下是唯一有信息量的答案。 if amount > 0: gross_coverage_pct = float(covered_net / amount * 100) elif cur_counts.get(key): gross_coverage_pct = float( cur_gross_counts.get(key, 0) / cur_counts[key] * 100, ) result.append(AiSpendingBucket( key=key, label_en=_BUCKET_LABELS_EN[key], label_zh=_BUCKET_LABELS_ZH[key], amount=_money(amount, rate), previous_amount=_money(prev_amount, rate), mom_change_pct=_pct_change(prev_amount, amount), share_pct=round(share, 2), fact_count=cur_counts.get(key, 0), distinct_subjects=len(cur_subjects.get(key, set())), amount_after_discount=amount_after_discount, data_caveat=None, amount_gross=amount_gross, discount_amount=discount_amount, discount_pct=( round(discount_pct, 2) if discount_pct is not None else None ), gross_coverage_pct=round(gross_coverage_pct, 2), )) return result @staticmethod def _subject_key_for_bucket(r: BucketRow) -> str | None: if r.bucket == BUCKET_SEATS: return r.user_id if r.bucket == BUCKET_BAILIAN: return r.aliyun_account_id or r.integration_id if r.bucket == BUCKET_FOUNDRY: return f"{r.integration_id}|{r.resource_group}" if r.resource_group else None if r.bucket == BUCKET_BEDROCK: return f"{r.integration_id}|{r.iam_user}" if r.iam_user else None if r.bucket == BUCKET_VERTEX: return f"{r.integration_id}|{r.scope_id}" if r.scope_id else None return None @staticmethod def _attribution_key(r: BucketRow) -> tuple: if r.bucket == BUCKET_SEATS: return (r.bucket, r.software_id, r.user_id) if r.bucket == BUCKET_BAILIAN: return (r.bucket, r.integration_id, r.aliyun_account_id) if r.bucket == BUCKET_FOUNDRY: return (r.bucket, r.integration_id, r.resource_group) if r.bucket == BUCKET_BEDROCK: return (r.bucket, r.integration_id, r.iam_user) if r.bucket == BUCKET_VERTEX: return (r.bucket, r.integration_id, r.scope_id) return (r.bucket, "_unknown_") def _attribution_labels( self, slot: dict[str, Any], s_map: dict[str, str], u_map: dict[str, str], ) -> tuple[str, str]: bucket = slot["bucket"] if bucket == BUCKET_SEATS: sw_label = s_map.get(slot["software_id"], "?") user_label = u_map.get(slot["user_id"], "?") return ( f"{sw_label} · {user_label}", f"{sw_label} · {user_label}", ) if bucket == BUCKET_BAILIAN: return ( slot["aliyun_account_id"] or slot["bailian_workspace"] or "?", slot["aliyun_account_id"] or slot["bailian_workspace"] or "?", ) if bucket == BUCKET_FOUNDRY: return (slot["resource_group"] or "?",) * 2 if bucket == BUCKET_BEDROCK: return (slot["iam_user"] or "?",) * 2 if bucket == BUCKET_VERTEX: return (slot["scope_id"] or "?",) * 2 return ("?", "?") @staticmethod def _dimension_key_for(r: BucketRow, dimension: str) -> str | None: if dimension == "bucket": return r.bucket if dimension == "date": return r.date.isoformat() if dimension in ("user", "owner"): return r.user_id if dimension == "department": return r.department if dimension == "model": return r.model_name if dimension == "service": return r.service_name if dimension in ("software", "software_id"): return r.software_id if dimension == "resource_group": return r.resource_group if dimension == "iam_user": return r.iam_user if dimension == "scope": return r.scope_id return None def _group_by_dimension( self, rows: list[BucketRow], dimension: str, ) -> dict[str | None, Decimal]: out: dict[str | None, Decimal] = {} for r in rows: k = self._dimension_key_for(r, dimension) out[k] = out.get(k, Decimal("0")) + r.cost_usd return out @staticmethod def _bucket_for_key( cur_by_key: dict[Any, Decimal], key: Any, dimension: str, all_rows: list[BucketRow], ) -> str | None: if dimension == "bucket": return str(key) if key is not None else None # For other dimensions: return the bucket of the FIRST row # whose dimension matches — the typical case is each key has # exactly one bucket; for keys that span buckets (rare; e.g. # a user with both seat + Foundry spend), the first-bucket # tag is informational. for r in all_rows: if ( key is None and AiSpendingService._dimension_key_for(r, dimension) is None ): return r.bucket if AiSpendingService._dimension_key_for(r, dimension) == key: return r.bucket return None async def _breakdown_from_rows( self, rows: list[BucketRow], *, period: str, dimension: str, rate: float, yoy_grouped: dict[str | None, Decimal] | None = None, ) -> AiSpendingBreakdown: grouped = self._group_by_dimension(rows, dimension) total = sum(grouped.values(), Decimal("0")) user_keys = [k for k in grouped if dimension in ("user", "owner") and k] software_keys = [k for k in grouped if dimension == "software" and k] u_map = await self._person_label_map(user_keys) s_map = await self._software_label_map(software_keys) yoy_grouped = yoy_grouped or {} breakdown_rows: list[AiSpendingBreakdownRow] = [] for k, amount in sorted( grouped.items(), key=lambda kv: kv[1], reverse=True, ): label = self._label_for_key(k, dimension, u_map, s_map) bucket = self._bucket_for_key(grouped, k, dimension, rows) share = float(amount / total * 100) if total > 0 else 0.0 yoy_amt = yoy_grouped.get(k) breakdown_rows.append(AiSpendingBreakdownRow( dimension_key=str(k) if k is not None else None, label_en=label or "(unattributed)", label_zh=label or "(未分摊)", amount=_money(amount, rate), share_pct=round(share, 2), bucket=bucket, yoy_amount=( _money(yoy_amt, rate) if yoy_amt is not None and yoy_amt > 0 else None ), yoy_change_pct=( _pct_change(yoy_amt, amount) if yoy_amt is not None and yoy_amt > 0 else None ), )) return AiSpendingBreakdown( period=period, dimension=dimension, rows=breakdown_rows, total=_money(total, rate), ) @staticmethod def _first_label( rows: list[BucketRow], dimension: str, u_map: dict[str, str], s_map: dict[str, str], i_map: dict[str, str], sc_map: dict[str, str], ) -> str | None: if not rows: return None r = rows[0] key = AiSpendingService._dimension_key_for(r, dimension) if key is None: return None if dimension in ("user", "owner"): return u_map.get(str(key)) if dimension == "software": return s_map.get(str(key)) if dimension == "scope": return sc_map.get(str(key)) return str(key) @staticmethod def _label_for_key( key: Any, dimension: str, u_map: dict[str, str], s_map: dict[str, str], ) -> str: if key is None: return "" if dimension in ("user", "owner"): return u_map.get(str(key)) or str(key) if dimension == "software": return s_map.get(str(key)) or str(key) if dimension == "bucket": return _BUCKET_LABELS_EN.get(str(key), str(key)) return str(key) async def _leaver_residual_usd( self, period_start: date, period_end_excl: date, *, bu: str | None = None, ) -> Decimal: """SaaS seat residual: spend from seats whose holder has ``employment_status='left'`` but the allocation was active for ≥1 day in the period.""" ai_software_ids = (await self.db.execute( select(AssetSoftwareCatalog.id).where( AssetSoftwareCatalog.category == "ai", ) )).scalars().all() if not ai_software_ids: return Decimal("0") start_dt = datetime.combine( period_start, datetime.min.time(), tzinfo=UTC, ) end_dt = datetime.combine( period_end_excl, datetime.min.time(), tzinfo=UTC, ) result = (await self.db.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan, IdentityPerson) .join( SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id, ) .outerjoin( IdentityPerson, SoftwareSeatAllocation.user_id == IdentityPerson.id, ) .where( SoftwareLicensePlan.software_id.in_(ai_software_ids), SoftwareSeatAllocation.allocated_at < end_dt, SoftwareSeatAllocation.status.in_(("active", "revoked")), or_( SoftwareSeatAllocation.revoked_at.is_(None), and_( SoftwareSeatAllocation.revoked_at >= start_dt, SoftwareSeatAllocation.revoked_at > SoftwareSeatAllocation.allocated_at, ), ), ) )).all() leaver_user_ids = [ p.id for _alloc, _plan, p in result if p is not None and p.employment_status in ("left", "leaver", "leaver_pending") ] bu_map: dict[str, str] = {} target_lower = "" if bu and leaver_user_ids: bu_map = await bu_for_user_ids(self.db, leaver_user_ids) target_lower = bu.strip().lower() rate = await self._get_cny_rate() rate_dec = Decimal(str(rate)) if rate else Decimal("0") # Dedup overlapping allocations per (plan, leaver) so a # revoked-then-re-added seat doesn't count twice. Same # rationale as ``_bucket_seats`` (Pre-launch audit P1-3). dedup_map: dict[tuple[str, str], tuple[Any, Any, Any]] = {} for alloc, plan, person in result: if person is None: continue if person.employment_status not in ("left", "leaver", "leaver_pending"): continue if bu and (bu_map.get(str(person.id)) or "").lower() != target_lower: continue dedup_key = (str(plan.id), str(person.id)) existing = dedup_map.get(dedup_key) if existing is None or alloc.allocated_at > existing[0].allocated_at: dedup_map[dedup_key] = (alloc, plan, person) total = Decimal("0") for alloc, plan, _person in dedup_map.values(): native = _compute_amount( price=plan.price, billing_cycle=plan.billing_cycle, period_start=period_start, period_end_excl=period_end_excl, allocated_at=alloc.allocated_at, ) currency = (plan.currency or _SEAT_CURRENCY_USD).upper() total += _seat_native_to_usd( native=native, currency=currency, rate_dec=rate_dec, plan_id=str(plan.id), ) return total # ── Settings + LLM-summary helpers ──────────────────────── async def _get_setting_value(self, key: str) -> Any: row = await self.db.get(PlatformSetting, key) return row.value if row else None async def _get_setting_str(self, key: str) -> str | None: v = await self._get_setting_value(key) if v is None: return None if not isinstance(v, str): logger.warning( "ai_spending: platform setting %r expected str, got %s", key, type(v).__name__, ) return None return v async def _set_setting( self, key: str, value: Any, *, actor_id: UUID | str, ) -> None: row = await self.db.get(PlatformSetting, key) if row is None: row = PlatformSetting( key=key, value=value, updated_by=str(actor_id) if actor_id else None, ) self.db.add(row) else: row.value = value row.updated_by = str(actor_id) if actor_id else None await self.db.flush() async def _get_cny_rate(self) -> float: from app.core import fx_rate as _fx_rate rate = await _fx_rate.get_usd_per_cny(self.db) if rate is None: raise AppException( code="AI_SPENDING_RATE_MISSING", message=( "ai_spending.cny_to_usd_rate platform setting is " "missing or out of plausible range. Restore via " "/admin/ai-spending/settings." ), status_code=500, ) return rate async def _write_attempt_audit_independent( self, *, trace_id: str, actor_user_id: UUID, period: str, ) -> None: from app.core.database import async_session_factory async with async_session_factory() as side_db: side_audit = AuditService(side_db) await side_audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="ai_spending.summary.generate.attempt", resource_type="ai_spending_summary", resource_id=period, status="success", detail={"period": period, "automatic": False}, ) await side_db.commit() async def _consume_regen_quota(self, user_id: str) -> None: """Atomically debit one regenerate quota for ``user_id`` — committed in its own session so the debit survives a subsequent LLM failure.""" from app.core.database import async_session_factory limit_v = await self._get_setting_value(SETTING_REGEN_LIMIT) if limit_v is None: raise AppException( code="AI_SPENDING_REGEN_LIMIT_MISSING", message=( "ai_spending.regenerate_per_user_per_day setting is " "missing." ), status_code=500, ) try: limit = int(limit_v) except (TypeError, ValueError) as exc: raise AppException( code="AI_SPENDING_REGEN_LIMIT_CORRUPT", message=( f"ai_spending.regenerate_per_user_per_day is not " f"an integer (value={limit_v!r})." ), status_code=500, ) from exc if limit < 1: raise AppException( code="AI_SPENDING_REGEN_LIMIT_INVALID", message=( f"ai_spending.regenerate_per_user_per_day must be " f">= 1 (got {limit})." ), status_code=500, ) today = datetime.now(UTC).date().isoformat() key = f"{SETTING_REGEN_COUNTER_PREFIX}{user_id}.{today}" async with async_session_factory() as side_db: try: locked = (await side_db.execute( select(PlatformSetting) .where(PlatformSetting.key == key) .with_for_update() )).scalar_one_or_none() if locked is None: try: side_db.add(PlatformSetting( key=key, value=1, description=( "AI spending regenerate quota counter — " "auto-managed; rotates daily." ), )) await side_db.commit() return except Exception: # noqa: BLE001 await side_db.rollback() locked = (await side_db.execute( select(PlatformSetting) .where(PlatformSetting.key == key) .with_for_update() )).scalar_one_or_none() if locked is None: raise used = _coerce_quota_int(locked.value) if used >= limit: await side_db.rollback() raise AppException( code="AI_SUMMARY_RATE_LIMITED", message=( f"You have reached the daily limit of " f"{limit} regenerations. Try again after " "00:00 UTC." ), status_code=429, ) locked.value = used + 1 await side_db.commit() except AppException: raise except Exception: # noqa: BLE001 logger.exception( "ai_spending: regen quota debit failed for " "user_id=%s key=%s", user_id, key, ) raise AppException( code="AI_SPENDING_QUOTA_INTERNAL_ERROR", message=( "Internal error debiting regenerate quota; your " "action was not performed." ), status_code=500, ) async def _persist_summary( self, *, period: str, actor_id: UUID | None, llm_json: dict[str, Any], model: str, aggregates: dict[str, Any], ) -> AiSpendingSummary: summary_md_en = str(llm_json.get("summary_en") or "") summary_md_zh = str(llm_json.get("summary_zh") or "") # Delete prior bilingual row (single canonical summary per period). await self.db.execute( AiSpendingSummary.__table__.delete().where( AiSpendingSummary.period == period, ) ) row = AiSpendingSummary( period=period, language="bilingual", summary_md_en=summary_md_en or None, summary_md_zh=summary_md_zh or None, findings=llm_json.get("findings") or [], suggestions=llm_json.get("suggestions") or [], aggregates=aggregates, model_used=model, generated_by_user_id=str(actor_id) if actor_id else None, generated_at=datetime.now(UTC), ) self.db.add(row) await self.db.flush() return row # ── Label maps ──────────────────────────────────────────── async def _person_label_map( self, person_ids: list[str | None], ) -> dict[str, str]: clean = list({p for p in person_ids if p}) if not clean: return {} rows = (await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(clean)) )).scalars().all() out: dict[str, str] = {} for r in rows: email = r.work_email or "" name = r.display_name or email or str(r.id) out[str(r.id)] = ( f"{name} · {email}".strip(" ·") if email else name ) return out async def _person_department_map( self, person_ids: list[str | None], ) -> dict[str, str]: clean = list({p for p in person_ids if p}) if not clean: return {} rows = (await self.db.execute( select( IdentityOrgMembership.person_id, IdentityOrgMembership.org_node_name, ).where( IdentityOrgMembership.person_id.in_(clean), IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) )).all() return {str(r[0]): str(r[1] or "") for r in rows if r[1]} async def _software_label_map( self, software_ids: list[str | None], ) -> dict[str, str]: clean = list({s for s in software_ids if s}) if not clean: return {} rows = (await self.db.execute( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.id.in_(clean), ) )).scalars().all() return {str(r.id): r.name for r in rows} async def _scope_label_map( self, scope_ids: list[str | None], ) -> dict[str, str]: clean = list({s for s in scope_ids if s}) if not clean: return {} rows = (await self.db.execute( select(CmdbCloudScope).where(CmdbCloudScope.id.in_(clean)) )).scalars().all() return { str(r.id): r.display_name or r.external_id for r in rows } async def _integration_label_map( self, integration_ids: set[str], ) -> dict[str, str]: if not integration_ids: return {} rows = (await self.db.execute( select(Integration).where(Integration.id.in_(integration_ids)) )).scalars().all() return {str(r.id): r.name for r in rows} async def _build_ak_link_map( self, rows: list[BucketRow], ) -> dict[tuple[str, str, str], tuple[str, str, str | None]]: """Resolve (bucket, integration_id, resource_key) → ( assignment_id, request_number, applicant_label) for the detail-row drill drawer. Two parallel JOINs are issued, one per bucket family: * ``foundry`` rows use ``(integration_id, resource_group)`` via ``api_key_assignments.cloud_scope_id`` → ``cmdb_cloud_scopes`` (scope_type='azure_resource_group', ``external_id`` = the RG name). * ``bedrock`` rows use ``(integration_id, iam_user)`` via ``api_key_assignments.iam_user_arn`` → parsed IAM user name. Bailian rows are NOT covered here — the Bailian loader writes the AK linkage inline on ``BucketRow.raw`` because the join keys (``api_key_pk``) live on the usage row, not the cost row. """ out: dict[tuple[str, str, str], tuple[str, str, str | None]] = {} if not rows: return out from app.domains.apikey.models import ApiKeyAssignment as ApiKeyAssn # ── Foundry: by (integration_id, resource_group) ────────── rg_pairs = { (r.integration_id, r.resource_group) for r in rows if r.bucket == BUCKET_FOUNDRY and r.integration_id and r.resource_group } if rg_pairs: rg_conds = [ and_( CmdbCloudScope.integration_id == integ, CmdbCloudScope.external_id == rg, ) for integ, rg in rg_pairs ] stmt = ( select( ApiKeyAssn.id, ApiKeyAssn.request_number, ApiKeyAssn.assigned_to, CmdbCloudScope.integration_id, CmdbCloudScope.external_id, ) .join(CmdbCloudScope, CmdbCloudScope.id == ApiKeyAssn.cloud_scope_id) .where( CmdbCloudScope.scope_type == "azure_resource_group", or_(*rg_conds), ) ) applicant_ids: set[str] = set() tmp: list[tuple[str, str, str, str, str]] = [] for ak_id, req_no, assigned_to, integ, rg in ( await self.db.execute(stmt) ).all(): applicant_ids.add(str(assigned_to)) tmp.append( (str(ak_id), req_no, str(assigned_to), str(integ), rg), ) applicant_label_map = await self._person_label_map( list(applicant_ids) ) for ak_id, req_no, applicant_id, integ, rg in tmp: out[("foundry", integ, rg)] = ( ak_id, req_no, applicant_label_map.get(applicant_id), ) # ── Bedrock: by (integration_id, iam_user) ──────────────── iam_pairs = { (r.integration_id, r.iam_user) for r in rows if r.bucket == BUCKET_BEDROCK and r.integration_id and r.iam_user } if iam_pairs: from app.domains.apikey.spend import parse_iam_user_name stmt = select( ApiKeyAssn.id, ApiKeyAssn.request_number, ApiKeyAssn.assigned_to, ApiKeyAssn.iam_integration_id, ApiKeyAssn.iam_user_arn, ).where(ApiKeyAssn.iam_user_arn.isnot(None)) cands: list[tuple[str, str, str, str, str]] = [] applicant_ids = set() for ak_id, req_no, assigned_to, integ, arn in ( await self.db.execute(stmt) ).all(): if not integ: continue user_name = parse_iam_user_name(arn) if not user_name: continue if (str(integ), user_name) in iam_pairs: applicant_ids.add(str(assigned_to)) cands.append( (str(ak_id), req_no, str(assigned_to), str(integ), user_name), ) applicant_label_map = await self._person_label_map( list(applicant_ids) ) for ak_id, req_no, applicant_id, integ, user_name in cands: out[("bedrock", integ, user_name)] = ( ak_id, req_no, applicant_label_map.get(applicant_id), ) return out # ───────────────────────────────────────────────────────────── # module-level helpers _PERIOD_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$") def _coerce_quota_int(value: Any) -> int: if isinstance(value, bool): return 0 if isinstance(value, (int, float, Decimal)): return int(value) return 0 def _parse_owner_from_resource_id(resource_id: str | None) -> str | None: """Extract ``Owner=`` from the AWS driver's synthetic ``aws-principal:Owner=|service=`` resource_id. Returns None when: * resource_id is empty * the prefix doesn't match * the first key is not ``Owner`` (e.g. ``aws-principal:service=…`` with no Owner tag — must NOT false-match to the service value) * the Owner value is the ``(unallocated)`` sentinel (no tag was set on the IAM principal) """ if not resource_id or not resource_id.startswith("aws-principal:"): return None # aws-principal:Owner=|service= body = resource_id[len("aws-principal:"):] if "=" not in body: return None kv = body.split("|", 1)[0] key, _, value = kv.partition("=") if key.strip() != "Owner": return None value = value.strip() if not value or value == "(unallocated)": return None return value def _money(usd: Decimal | float | int, rate: float) -> MoneyPair: usd_dec = usd if isinstance(usd, Decimal) else Decimal(str(usd)) rate_dec = Decimal(str(rate)) cny_dec = usd_dec / rate_dec return MoneyPair( usd=quantize(usd_dec), cny=quantize(cny_dec), usd_per_cny=rate, ) def _pct_change(prev: Decimal | None, curr: Decimal) -> float | None: if not prev or prev == 0: return None return float((curr - prev) / prev * 100) def _shift_period(period: str, months: int) -> str: p_start, _ = _parse_period(period) new_year = p_start.year new_month = p_start.month + months while new_month > 12: new_year += 1 new_month -= 12 while new_month < 1: new_year -= 1 new_month += 12 return f"{new_year:04d}-{new_month:02d}" def _current_period() -> str: return datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m") def _sum_usd(rows: list[BucketRow]) -> Decimal: return sum((r.cost_usd for r in rows), Decimal("0")) def _sum_gross_usd( rows: list[BucketRow], ) -> tuple[Decimal, Decimal, bool]: """Sum pre-discount (gross) USD over rows that HAVE a gross price, plus the post-discount net of those SAME rows. Returns ``(gross_sum, covered_net, has_gross)``. ``covered_net`` (not the whole total) is the right denominator for the discount %, and the gap between ``covered_net`` and the full total is what ``gross_coverage_pct`` exposes — so a partially-covered total (GCP/Azure/Seats have no gross) never inflates the discount.""" gross_sum = Decimal("0") covered_net = Decimal("0") has_gross = False for r in rows: if r.cost_usd_gross is not None: has_gross = True gross_sum += r.cost_usd_gross covered_net += r.cost_usd return gross_sum, covered_net, has_gross def _distinct_users(rows: list[BucketRow]) -> int: """Distinct attributable users across **all** buckets. Bailian used to be excluded here on the premise that "BSS gives no per-user breakdown, so the rows have ``user_id=None``". That premise died with PR #263 / migration 299: ``_bucket_bailian`` resolves an owner per row through the AK → sub-account ladder, so Bailian rows routinely carry a real ``user_id``. Keeping the exclusion silently dropped those people from 活跃 AI 用户 — and, because ``cost_per_active_user`` excluded the same bucket from its numerator, the two halves of that KPI were computed over two different populations. The ``if r.user_id`` guard is what actually keeps unattributed rows out; it needs no per-bucket help. """ return len({r.user_id for r in rows if r.user_id}) def _summary_to_response( row: AiSpendingSummary, ) -> AiSpendingSummaryResponse: return AiSpendingSummaryResponse( period=row.period, available=True, summary_md_en=row.summary_md_en, summary_md_zh=row.summary_md_zh, findings=_coerce_findings(row.findings or []), suggestions=_coerce_suggestions(row.suggestions or []), aggregates=row.aggregates, model_used=row.model_used, input_token_count=row.input_token_count, output_token_count=row.output_token_count, generated_at=row.generated_at, generated_by_user_label=None, ) def _coerce_findings(value: Any) -> list[AiSummaryFinding]: if not isinstance(value, list): return [] out: list[AiSummaryFinding] = [] for v in value: if not isinstance(v, dict): continue try: out.append(AiSummaryFinding(**v)) except Exception as exc: # noqa: BLE001 logger.warning( "ai_spending: dropping malformed finding %r: %s", v, exc, ) return out def _coerce_suggestions(value: Any) -> list[AiSummarySuggestion]: if not isinstance(value, list): return [] out: list[AiSummarySuggestion] = [] for v in value: if not isinstance(v, dict): continue try: out.append(AiSummarySuggestion(**v)) except Exception as exc: # noqa: BLE001 logger.warning( "ai_spending: dropping malformed suggestion %r: %s", v, exc, ) return out _SUMMARY_SYSTEM_PROMPT = """You are a senior FinOps + AI-cost analyst at PingCAP. \ You receive a JSON snapshot of the company's monthly AI spending across four \ canonical buckets: * SaaS seats (Claude.ai, OPEN AI Team, future Cursor / Copilot) * Aliyun Bailian (per-(day × API key × model); owner resolved via the \ AK → sub-account ladder. Keys in Resource-Directory MEMBER accounts cannot be \ enumerated, so a residual share stays account-attributed) * Azure AI Foundry (per-RG attribution) * AWS Bedrock (per-IAM-user attribution via cost-allocation tag) * (Optional fifth: Google Vertex AI when present) Produce a JSON object with the following shape — and ONLY this shape: { "summary_en": "<3-6 paragraph executive summary in English markdown>", "summary_zh": "<3-6 paragraph executive summary in Chinese markdown>", "findings": [ { "title_en": "...", "title_zh": "...", "detail_en": "...", "detail_zh": "...", "severity": "info" | "warn" | "crit" }, ... ], "suggestions": [ { "action_en": "...", "action_zh": "...", "rationale_en": "...", "rationale_zh": "...", "expected_saving_usd": , "priority": "low" | "medium" | "high" }, ... ] } Findings are concrete observations about THIS period's data. Suggestions are \ concrete next-month actions with quantified expected_saving_usd where possible. \ Be specific: cite numbers, names, departments. Tone is professional, concise, \ no emoji. The English summary uses American spelling; the Chinese summary uses \ simplified characters and the same factual content as the English one. NEVER fabricate numbers. If a section has zero data, say so explicitly. \ Attribute Bailian dollars to people only where the snapshot itself names an \ owner for those rows; the residual that stays account-attributed must be \ described as unattributed, never spread across users. MIND THE COST BASIS. Buckets carry up to three tiers: gross (折扣前 / list \ price), after-discount (折扣后, Bailian only) and net paid (实付). The \ headline totals are NET PAID. Aliyun Bailian's net paid is routinely 0 \ because vouchers and contract discounts absorb it entirely — that means "we \ consumed real capacity at no invoiced cost", NOT "nobody used Bailian". When \ you discuss Bailian volume or compare it with other buckets, use its gross \ figure and say which basis you are quoting. Never compare one bucket's gross \ against another bucket's net. """ def _build_summary_prompt( *, overview: AiSpendingOverview, prev_overview: AiSpendingOverview | None, breakdown_user: AiSpendingBreakdown, breakdown_dept: AiSpendingBreakdown, breakdown_model: AiSpendingBreakdown, breakdown_bucket: AiSpendingBreakdown, attribution: AiBucketAttribution, ) -> str: payload = { "period": overview.period, "totals": { "this_period_usd": overview.grand_total.usd, "previous_period": overview.previous_period, "previous_period_usd": ( overview.previous_grand_total.usd if overview.previous_grand_total else None ), "yoy_period": overview.yoy_period, "yoy_usd": ( overview.yoy_grand_total.usd if overview.yoy_grand_total else None ), "mom_change_pct": overview.mom_change_pct, "yoy_change_pct": overview.yoy_change_pct, "active_user_count": overview.active_user_count, "leaver_residual_usd": ( overview.leaver_residual_amount.usd if overview.leaver_residual_amount else None ), "is_partial_period": overview.is_partial_period, }, "buckets": [ { "key": b.key, # ``usd`` 是**实付**。百炼的实付常年为 0(优惠券抵完), # 只给这一个数会让模型得出「没人用百炼」的结论,所以把 # 折前/折扣后一并给出 —— 提示词里那条「讨论百炼用量要 # 用折前、并注明口径」才有数据可依。 "usd": b.amount.usd, "usd_basis": "net_paid", "gross_usd": b.amount_gross.usd if b.amount_gross else None, "after_discount_usd": ( b.amount_after_discount.usd if b.amount_after_discount else None ), "gross_coverage_pct": b.gross_coverage_pct, "share_pct": b.share_pct, "mom_change_pct": b.mom_change_pct, } for b in overview.buckets ], "top_users": [ {"hash": _hash_user_id(r.dimension_key), "usd": r.amount.usd, "share_pct": r.share_pct} for r in breakdown_user.rows[:15] ], "top_departments": [ {"name": r.label_en, "usd": r.amount.usd, "share_pct": r.share_pct} for r in breakdown_dept.rows[:15] ], "top_models": [ {"name": r.label_en, "usd": r.amount.usd, "share_pct": r.share_pct} for r in breakdown_model.rows[:15] ], "by_bucket": [ {"bucket": r.label_en, "usd": r.amount.usd, "share_pct": r.share_pct} for r in breakdown_bucket.rows ], "top_attribution_rows": [ { "bucket": r.bucket, # PII-safe label: SEATS embeds `display_name · email` # in label_en for the UI, and BEDROCK's iam_user is # commonly `firstname.lastname` shape — both identify # a specific employee. Bailian is account-id only, # Foundry is RG name, Vertex is scope_id: all # organisational and safe to ship as-is. "label": _sanitise_attribution_label_for_llm(r), "owner_hash": _hash_user_id( str(r.owner_person_id) if r.owner_person_id else None ), "department": r.owner_department, "usd": r.period_amount.usd, } for r in attribution.rows[:30] ], "unattributed_usd": attribution.unattributed_amount.usd, } return ( "Here is the JSON snapshot. Respond with the JSON object " "specified in the system prompt. The data:\n\n" + json.dumps(payload, ensure_ascii=False, indent=2) ) def _hash_user_id(uid: str | None) -> str | None: """Pseudonymise a user id for the LLM prompt. The Bailian LLM is a third-party service; PR review pass 2's security finding #1 was that earlier prompt payloads sent ``top_users`` with display_name + email. Names/emails leaving the org over the wire to a 3rd-party LLM is a PII exposure we close by hashing the id → short hex token. The dashboard server-side de-pseudonymises at render time when needed. """ if not uid: return None import hashlib return "u" + hashlib.sha256(uid.encode()).hexdigest()[:8] def _sanitise_attribution_label_for_llm(r: AiBucketAttributionRow) -> str: """Return a PII-safe label for an attribution row before it enters the third-party LLM prompt payload. The UI-side ``label_en`` embeds person-identifying strings for two buckets: * ``seats`` : ``"{software} · {display_name} · {work_email}"`` built by :meth:`_attribution_labels`. Email + name to the third-party LLM is the exact PII exposure :func:`_hash_user_id` was added to prevent — but the row-level label snuck the same data past the guard. * ``bedrock``: ``label_en = iam_user``. ``iam_user`` is commonly derived from ``firstname.lastname`` (see :func:`app.integrations.drivers.aws_cloud._iam_user_name_for`), so it identifies a specific employee even though it's not a raw email. For these two buckets we substitute a pseudonymised label that keeps the **structural** signal (software / hashed-subject) the LLM uses to spot anomalies while dropping the PII. The other three buckets (bailian / foundry / vertex) carry only organisational identifiers (Aliyun account id, RG name, GCP project scope id), which are safe to ship as-is. """ if r.bucket == BUCKET_SEATS: # label_en pattern: "{sw_label} · {user_label}" — keep the # software half, hash the user half via owner_person_id. head = r.label_en.split(" · ", 1)[0] if r.label_en else "?" hashed = _hash_user_id( str(r.owner_person_id) if r.owner_person_id else None ) or "anon" return f"{head} · {hashed}" if r.bucket == BUCKET_BEDROCK: # iam_user is usually firstname.lastname — pseudonymise by # hashing the iam_user string directly (no person_id link on # bedrock rows when Owner tag is unset). hashed = _hash_user_id(r.iam_user) or "anon" return f"bedrock:{hashed}" # Bailian (account id), Foundry (RG name), Vertex (scope id) — # organisational, no PII. return r.label_en or "?" __all__ = [ "AiSpendingService", "BUCKET_BAILIAN", "BUCKET_BEDROCK", "BUCKET_FOUNDRY", "BUCKET_SEATS", "BUCKET_VERTEX", "BucketRow", ]