"""Pydantic shapes for the AI Spending dashboard (4-bucket model). PR #253 (Phase 2) collapsed the legacy 5-source dedup ladder into four canonical buckets, each backed by a single authoritative data source: * ``seats`` — ``software_monthly_bills`` JOIN ``asset_software_catalog`` WHERE ``category='ai'``. * ``bailian`` — ``bailian_usage_daily`` (Aliyun BSS-derived; per-API-key breakdown intentionally unavailable — BSS only exposes account-level cost). * ``foundry`` — ``cmdb_cost_facts`` filtered to Azure integration rows whose ``service`` matches the Azure entry of ``platform_settings.ai_spending.service_whitelist``. * ``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. All monetary values cross the wire as ``MoneyPair`` (USD + CNY at the period's reference rate) so the frontend renders either currency without re-fetching. The ``usd_per_cny`` field on each pair makes the conversion auditable and surfaces stale-rate situations. """ from __future__ import annotations from datetime import date, datetime from decimal import Decimal from typing import Any from uuid import UUID from pydantic import BaseModel, ConfigDict, Field # ─── Currency-pair amount ──────────────────────────────────────── class MoneyPair(BaseModel): """Amount denominated in both USD and CNY at the rate used for the request. ``usd_per_cny`` makes the conversion auditable. The stored platform setting ``ai_spending.cny_to_usd_rate`` is USD-per-CNY form (≈0.14): multiply a CNY amount by the rate to get USD, divide a USD amount by the rate to get CNY. """ model_config = ConfigDict(from_attributes=True) usd: float cny: float usd_per_cny: float = Field( ..., description=( "USD-per-CNY rate applied (1 CNY = N USD). Multiply a " "CNY amount by this to get USD; divide a USD amount by " "this to get CNY." ), ) # ─── 4-bucket Overview ────────────────────────────────────────── class AiSpendingBucket(BaseModel): """One of the canonical AI spend buckets.""" model_config = ConfigDict(from_attributes=True) key: str = Field( ..., description=( "Stable bucket identifier: ``seats`` / ``bailian`` / " "``foundry`` / ``bedrock`` / ``vertex``. Mirrors the i18n key " "namespace ``aiSpending.bucket..label`` on the frontend." ), ) label_en: str label_zh: str amount: MoneyPair previous_amount: MoneyPair | None = None mom_change_pct: float | None = Field( None, description=( "Period-over-period percent change against ``previous_amount``. " "Null when previous_amount is zero (∞% would mislead — the UI " "renders 'NEW' instead)." ), ) share_pct: float = Field( ..., description="Percent of the period grand total (0-100).", ) fact_count: int = Field( ..., description=( "Number of underlying line-item rows that contributed to " "``amount`` — lets the dashboard show 'N facts' for " "data-density transparency." ), ) distinct_subjects: int = Field( ..., description=( "Number of distinct attribution subjects within the bucket — " "users for seats, Aliyun sub-accounts for Bailian, RGs for " "Foundry, IAM users for Bedrock, scopes for Vertex. Drives " "the 'X users / X RGs' counters. (This said 'models for " "Bailian' until 2026-08-15, which matched neither " "``_subject_key_for_bucket`` nor the data.)" ), ) data_caveat: str | None = Field( None, description=( "Plain-English explanation when the bucket has data-quality " "limitations. For Bailian: 'Aliyun BSS exposes Bailian spend " "at account granularity only — per-API-key breakdown is not " "available.' Null when the data is fully resolved." ), ) amount_after_discount: MoneyPair | None = Field( None, description=( "Middle price tier (优惠后): list price minus the negotiated " "discount, BEFORE coupons + round-off. Populated only for " "Bailian today (= PretaxGrossAmount − InvoiceDiscount from the " "commodity bill). Null for buckets without a distinct mid-tier " "(their net already equals after-discount)." ), ) amount_gross: MoneyPair | None = Field( None, description=( "Pre-discount (list / 原价) amount for the portion of this " "bucket whose source exposes a gross price. Currently only " "Bailian (BSS PretaxGrossAmount) and AWS (CUR " "publicOnDemandCost) populate it; GCP and Azure aren't wired " "yet, so their gross stays null. Null also for AI Seats and " "any row without a list price." ), ) discount_amount: MoneyPair | None = Field( None, description=( "``amount_gross`` minus the post-discount net of the SAME " "covered rows. Signed: negative when credits/refunds make " "net exceed gross. Null when gross is unavailable." ), ) discount_pct: float | None = Field( None, description=( "Discount as a percent of gross for the covered portion: " "(gross − net) / gross × 100. Signed, not clamped. Null when " "gross is unavailable. For 2026-05 Bailian ≈ 63.8." ), ) gross_coverage_pct: float = Field( 0.0, description=( "Percent of this bucket's net ``amount`` that has a known " "gross price (0-100). 100 = fully resolvable discount " "(Bailian); <100 = partial (Azure forward-only); 0 = none." ), ) currency_warning: bool = Field( False, description=( "Seats bucket only: true when at least one active-seat " "`category='ai'` plan carries a non-USD currency, so the USD " "figure may be FX-mis-converted (a mis-entered currency shrinks " "the bucket). The dashboard renders a warning chip; the " "/admin/reconciliation cost lens has the per-plan detail." ), ) class AiSpendingOverview(BaseModel): """Top-of-page summary — grand total + per-bucket breakdown.""" period: str = Field(..., description="YYYY-MM") period_start: date period_end: date is_partial_period: bool = Field( ..., description=( "True when the requested period is the current month — " "values reflect MTD spend and continue to grow." ), ) grand_total: MoneyPair grand_total_gross: MoneyPair | None = Field( None, description=( "Pre-discount (list / 原价) grand total — sum of gross over " "ONLY the rows whose source exposes a list price. Null when " "no bucket carries gross. Partial coverage is expected today " "(Bailian + AWS Bedrock have gross; GCP/Azure/Seats don't) — " "read alongside ``gross_coverage_pct``; this is NOT a " "platform-wide list price." ), ) grand_discount_pct: float | None = Field( None, description=( "(gross − net) / gross × 100 over the covered rows only. " "Signed, not clamped. Null when no gross is available." ), ) gross_coverage_pct: float = Field( 0.0, description=( "Percent of ``grand_total`` (net) that has a known list " "price (0-100). <100 means the pre-discount total covers " "only part of spend — the dashboard surfaces this so the " "discount isn't misread as platform-wide." ), ) previous_period: str | None = None previous_grand_total: MoneyPair | None = None yoy_period: str | None = None yoy_grand_total: MoneyPair | None = None mom_change_pct: float | None = None yoy_change_pct: float | None = None buckets: list[AiSpendingBucket] active_user_count: int = Field( ..., description=( "Distinct people who incurred any attributable AI spend in " "the period. Counts: seat holders (≥1 day of allocation), " "Foundry RG owners (with non-zero spend), Bedrock IAM-user " "owners (with non-zero spend). Bailian intentionally NOT " "counted — BSS gives no per-user breakdown." ), ) cost_per_active_user: MoneyPair | None = Field( None, description=( "grand_total ÷ active_user_count. Null when there are no " "active users. First-class so the KPI tile and its MoM " "delta don't re-derive it client-side." ), ) previous_cost_per_active_user: MoneyPair | None = None cost_per_active_user_change_pct: float | None = Field( None, description="MoM change in cost-per-active-user. Null if no baseline.", ) leaver_residual_amount: MoneyPair | None = Field( None, description=( "Subset of seats whose holder has employment_status='left' " "but the seat remained active for ≥1 day in the period — " "the 'use one day = one month' billing rule. Useful for " "spotting unreclaimed seats post-offboarding." ), ) # ─── Breakdown (one tab on the dashboard) ────────────────────── class AiSpendingBreakdownRow(BaseModel): """One row of the by-dimension breakdown table.""" model_config = ConfigDict(from_attributes=True) dimension_key: str | None label_en: str label_zh: str amount: MoneyPair share_pct: float bucket: str | None = Field( None, description=( "Which bucket contributed this row (``seats`` / ``bailian`` / " "``foundry`` / ``bedrock``). Lets the UI colour-tag rows by " "bucket when the same dimension straddles multiple buckets " "(e.g. dimension=user: same user may show seats AND Foundry " "spend)." ), ) yoy_amount: MoneyPair | None = Field( None, description=( "This dimension key's spend in the same month last year " "(period − 12). Null when the key did not exist a year ago " "(new hire / new model / new service)." ), ) yoy_change_pct: float | None = Field( None, description=( "(amount − yoy_amount) / yoy_amount × 100. Null when " "``yoy_amount`` is null or zero. Lets the table flag " "'who/what grew vs a year ago' per row." ), ) class AiSpendingBreakdown(BaseModel): period: str dimension: str = Field( ..., description=( "One of: bucket / user / department / model / service / " "resource_group / iam_user / software." ), ) rows: list[AiSpendingBreakdownRow] total: MoneyPair class AiSpendingDetailRow(BaseModel): """One unaggregated cost row that contributes to a breakdown bucket.""" date: date bucket: str = Field( ..., description=( "Which bucket this row belongs to: seats / bailian / " "foundry / bedrock / vertex." ), ) cost_usd: Decimal cost_cny: Decimal user_id: str | None = None user_label: str | None = None department: str | None = None software_id: str | None = None software_label: str | None = None plan_name: str | None = None model_name: str | None = None service_name: str | None = None resource_group: str | None = None iam_user: str | None = None integration_id: str | None = None integration_name: str | None = None scope_id: str | None = None scope_label: str | None = None # How the row's ``user_id`` was decided. Mirrors the values stamped # on ``cmdb_cost_facts.auto_owner_source`` for Foundry/Bedrock/Vertex # rows; Bailian rows now also carry one of ``api_key_owner`` / # ``ak_ram_user`` / ``scope_owner`` / ``none`` populated by the # bailian loader's owner ladder. Lets the UI distinguish "this is # the user's actual AK usage" from "this is just the subscription # owner getting the bill". auto_owner_source: str | None = None # The underlying cloud resource identifier (Azure resource ID / # AWS ARN / GCP resource link / Aliyun account ID + workspace). # Populated for cloud-billing-fact-backed rows so admins can drill # past the bucket label to the actual resource. resource_id: str | None = None # API key assignment linkage when the cost is attributable to a # specific Helix-issued AK (Foundry: matched by integration+RG; # Bedrock: matched by integration+IAM user; Bailian: matched by # api_key_pk or RAM user). Lets the UI link a row to its AK # request without a second roundtrip. ak_assignment_id: str | None = None ak_request_number: str | None = None ak_owner_label: str | None = None class AiSpendingDetailGroupRow(BaseModel): """One aggregated row inside a detail drill-down rollup. The drawer's raw view lists per-day rows; these group rows let the user collapse the same rows by owner / subscription / service so "who / what is driving this number" reads at a glance without eyeballing 88 daily lines. Computed backend-side (FE is pure presentation) so the money sums use the same FX rate as ``total``. """ # The grouping value. ``None`` is the unattributed/unknown bucket # (e.g. a cost row with no owner) and renders as "未归属" on the UI. key: str | None = None label: str | None = None # Secondary line for context: department for owner rollups, # integration name for subscription rollups. May be empty. secondary: str | None = None amount: MoneyPair row_count: int = Field(..., description="Raw detail rows folded in.") share: float = Field( ..., description="Fraction 0..1 of the drill-down total.", ) class AiSpendingDetailGroup(BaseModel): """A single rollup axis (owner / scope / service) over the same rows the drill-down emits. Only axes with at least one identifying value are emitted, so seats (no cloud scope) won't carry an empty ``scope`` group.""" dimension: str = Field( ..., description="owner | scope | service.", ) rows: list[AiSpendingDetailGroupRow] class AiSpendingDetail(BaseModel): period: str dimension: str dimension_key: str label: str | None = None rows: list[AiSpendingDetailRow] total: MoneyPair # By-owner / by-subscription / by-service rollups of ``rows``. Each # is sorted by amount desc. Empty for axes that have no identifying # value among the matched rows. groups: list[AiSpendingDetailGroup] = Field(default_factory=list) # ─── Trend (line + bar) ──────────────────────────────────────── class AiSpendingTrendPoint(BaseModel): """One data point on the trend chart (day or month).""" model_config = ConfigDict(from_attributes=True) bucket: str = Field( ..., description="ISO date for daily, YYYY-MM for monthly.", ) amount: MoneyPair by_bucket: dict[str, float] = Field( default_factory=dict, description=( "Per-bucket USD split for stacked rendering. Map keys are " "bucket keys (``seats`` / ``bailian`` / ``foundry`` / " "``bedrock``); values are USD floats. Populated ONLY when the " "trend's ``dimension`` is ``bucket`` (the default) so the KPI " "+ per-bucket sparklines keep reading it. Empty for other " "dimensions — read ``by_series`` instead." ), ) by_series: dict[str, float] = Field( default_factory=dict, description=( "Per-series USD split when ``dimension`` ≠ ``bucket`` " "(model / service / department). Keys are the dimension " "values, capped to the window's top-8 by total; the rest " "fold into the sentinel key ``__other__``. Rows with no value " "for the dimension fold into ``__none__``. Empty for the " "``bucket`` dimension." ), ) class AiSpendingTrend(BaseModel): granularity: str = Field(..., description="daily | monthly") points: list[AiSpendingTrendPoint] dimension: str = Field( "bucket", description="bucket (default) | model | service | department.", ) series_order: list[str] = Field( default_factory=list, description=( "Stable stack/legend order of series keys across the whole " "window (top series first, ``__other__`` last). For the " "``bucket`` dimension the FE may ignore this and use its own " "bucket order; for other dimensions it drives rendering." ), ) series_labels: dict[str, str] = Field( default_factory=dict, description=( "Display label per series key (identity for model/service/" "department since those keys are already human-readable). " "Sentinel keys ``__none__`` / ``__other__`` carry an empty " "label — the FE localises them." ), ) # ─── Per-AK / per-resource attribution table ────────────────── class AiBucketAttributionRow(BaseModel): """One row of the per-bucket attribution table — replaces the legacy ``cmdb_ai_resources``-backed chargeback view. A Foundry row, a Bedrock row, and a Bailian-account row each surface here with the period amount, owner, and bucket-specific identifying fields populated. """ model_config = ConfigDict(from_attributes=True) bucket: str = Field( ..., description="seats / bailian / foundry / bedrock / vertex.", ) # Assignment linkage when the row maps 1:1 to an AK request. assignment_id: UUID | None = None request_number: str | None = None integration_id: UUID | None = None integration_name: str | None = None scope_id: str | None = None scope_display_name: str | None = None # Bucket-specific identifiers (exactly one populated per row). resource_group: str | None = Field( None, description="Azure RG name (foundry rows).", ) iam_user: str | None = Field( None, description="AWS IAM user name (bedrock rows).", ) bailian_workspace: str | None = Field( None, description="Bailian workspace label (bailian rows).", ) seat_software_id: UUID | None = Field( None, description="Software id (seat rows).", ) label_en: str label_zh: str owner_person_id: UUID | None = None owner_label: str | None = None owner_department: str | None = None auto_owner_source: str | None = Field( None, description=( "How owner was resolved: ``rule`` (allocation rule matched), " "``scope_owner`` (scope-owner fallback), ``manual`` (admin " "override), or ``bailian_account`` (Aliyun account owner)." ), ) period_amount: MoneyPair class AiBucketAttribution(BaseModel): period: str rows: list[AiBucketAttributionRow] unattributed_amount: MoneyPair = Field( ..., description=( "Sum of period_amount where owner_person_id is NULL. " "Highlighted on the UI so admins assign these promptly." ), ) # ─── Summary (LLM-generated) ────────────────────────────────── class AiSummaryFinding(BaseModel): model_config = ConfigDict(extra="ignore") title_en: str title_zh: str detail_en: str | None = None detail_zh: str | None = None severity: str = Field( "info", description="info | warn | crit; renders different colour on UI.", ) class AiSummarySuggestion(BaseModel): model_config = ConfigDict(extra="ignore") action_en: str action_zh: str expected_saving_usd: float | None = None priority: str = Field( "medium", description="low | medium | high.", ) rationale_en: str | None = None rationale_zh: str | None = None class AiSpendingSummaryResponse(BaseModel): model_config = ConfigDict(from_attributes=True) period: str available: bool = Field( ..., description=( "False when no summary has been generated for this period " "yet — the frontend then prompts the admin to click " "'regenerate'." ), ) summary_md_en: str | None = None summary_md_zh: str | None = None findings: list[AiSummaryFinding] = [] suggestions: list[AiSummarySuggestion] = [] aggregates: dict[str, Any] | None = None model_used: str | None = None input_token_count: int | None = None output_token_count: int | None = None generated_at: datetime | None = None generated_by_user_label: str | None = None class AiSpendingSummaryRegenerateBody(BaseModel): period: str | None = Field( None, description=( "YYYY-MM. Defaults to the previous calendar month when " "omitted (matches the auto-job's period selection)." ), ) class AiSpendingRollupRefreshBody(BaseModel): period: str | None = Field( None, pattern=r"^\d{4}-(0[1-9]|1[0-2])$", description=( "YYYY-MM closed month to re-materialise. Must be strictly " "before the current month (open months are computed live, " "never materialised). Omit to refresh the most-recent 2 " "closed months — same window the daily cron warms." ), ) # ─── Settings (admin-editable) ──────────────────────────────── class AiServiceWhitelist(BaseModel): """Per-provider AI service substring patterns. Empty list = no rows match for that provider. Aliyun is intentionally absent — see :mod:`app.domains.ai_spending.whitelist` for why.""" model_config = ConfigDict(extra="forbid") azure: list[str] = Field(default_factory=list) aws: list[str] = Field(default_factory=list) gcp: list[str] = Field(default_factory=list) class AiSpendingSettings(BaseModel): cny_to_usd_rate: float cny_rate_last_refreshed_at: datetime | None = None llm_model: str llm_region: str = Field( "intl", description=( "Which DashScope region the Bailian client hits. " "``cn`` = mainland (dashscope.aliyuncs.com); " "``intl`` = Singapore (dashscope-intl.aliyuncs.com)." ), ) llm_api_key_present: bool service_whitelist: AiServiceWhitelist regenerate_per_user_per_day: int class AiSpendingSettingsUpdate(BaseModel): cny_to_usd_rate: float | None = None llm_model: str | None = None llm_region: str | None = Field( None, description=( "Bailian region: ``cn`` (mainland) or ``intl`` (Singapore)." ), ) llm_api_key: str | None = Field( None, description=( "sk-xxx Bailian API key. Pass empty string to clear; " "omit to keep the existing key." ), ) service_whitelist: AiServiceWhitelist | None = None regenerate_per_user_per_day: int | None = Field( None, ge=1, le=100, ) # ─── Movers (period-over-period changers) ──────────────────── class AiSpendingMoverRow(BaseModel): model_config = ConfigDict(from_attributes=True) dimension_key: str | None label_en: str label_zh: str current: MoneyPair previous: MoneyPair delta: MoneyPair = Field( ..., description="current - previous (signed). Negative = drop.", ) delta_pct: float | None = Field( None, description=( "Percentage change vs previous period. Null when previous " "is zero (∞% would mislead — frontend should render 'NEW')." ), ) bucket: str | None = Field( None, description="Bucket the row belongs to, when dimension allows.", ) class AiSpendingMovers(BaseModel): period: str previous_period: str dimension: str = Field( ..., description="user | department | model | service | bucket.", ) jumps: list[AiSpendingMoverRow] = Field( ..., description="Top N rows sorted by delta DESC (biggest increases).", ) drops: list[AiSpendingMoverRow] = Field( ..., description="Top N rows sorted by delta ASC (biggest decreases).", ) # ─── Forecast (month-end projection) ────────────────────────── class AiSpendingForecast(BaseModel): """Linear projection of the current period's grand total.""" period: str is_partial: bool days_in_period: int days_elapsed: int days_remaining: int mtd_total: MoneyPair daily_pace: MoneyPair = Field( ..., description="mtd_total / days_elapsed (avg daily spend so far).", ) projected_total: MoneyPair = Field( ..., description=( "mtd_total + daily_pace * days_remaining. Naïve linear " "projection — fine for a rough month-end indicator. " "When period is past, equals mtd_total." ), ) previous_period: str previous_total: MoneyPair | None = None projected_vs_previous_pct: float | None = Field( None, description=( "(projected_total - previous_total) / previous_total * 100. " "Negative = on track to come in below last month." ), ) # ─── BU rollup ──────────────────────────────────────────────── class AiSpendingByBuRow(BaseModel): model_config = ConfigDict(from_attributes=True) bu: str member_count: int amount: MoneyPair share_pct: float active_user_count: int class AiSpendingByBu(BaseModel): period: str rows: list[AiSpendingByBuRow] unattributed_amount: MoneyPair = Field( ..., description=( "Spend rows that could not be attributed to a BU — bucket-by-" "bucket: Bailian (account-level, no per-user breakdown); " "Foundry/Bedrock facts whose resolved owner has no primary " "org membership; seats where the seat holder has no BU." ), ) # ─── Money helper ───────────────────────────────────────────── def quantize(amount: float | Decimal | int) -> float: """Round to two decimals for display.""" return float(round(Decimal(str(amount)), 2)) __all__ = [ "AiBucketAttribution", "AiBucketAttributionRow", "AiServiceWhitelist", "AiSpendingBreakdown", "AiSpendingBreakdownRow", "AiSpendingBucket", "AiSpendingByBu", "AiSpendingByBuRow", "AiSpendingDetail", "AiSpendingDetailRow", "AiSpendingForecast", "AiSpendingMoverRow", "AiSpendingMovers", "AiSpendingOverview", "AiSpendingSettings", "AiSpendingSettingsUpdate", "AiSpendingSummaryRegenerateBody", "AiSpendingSummaryResponse", "AiSpendingTrend", "AiSpendingTrendPoint", "AiSummaryFinding", "AiSummarySuggestion", "MoneyPair", "quantize", ]