"""BailianService — read + sync orchestration. Read endpoints serve the admin / employee console. Sync orchestration upserts workspaces / applications / usage rows from Bailian via :class:`BailianClient`. Auto-grant is implemented in ``app.domains.apikey.bailian_autogrant`` to keep the apikey-callback seam in one place — this service only deals with discovered / synced state. """ from __future__ import annotations from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from uuid import UUID from sqlalchemy import and_, delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.audit.service import AuditService from app.core.exceptions import AppException, ConflictError, NotFoundError, format_exception_for_log from app.core.safe_logging import get_safe_logger from app.domains.bailian.client import ( _MODELSTUDIO_REGIONS, _SUPPORTED_PROVIDERS, BailianClient, resolve_credentials, ) from app.domains.bailian.models import ( BAILIAN_AGGREGATE_MODEL, BailianApiKey, BailianUsageDaily, BailianWorkspace, ) from app.domains.bailian.schemas import ( BailianApiKeyResponse, BailianUsageDailyResponse, BailianUsageSummary, BailianWorkspaceResponse, ) from app.domains.cmdb.models import CmdbCloudScope from app.identity.person_models import IdentityPerson from app.integrations.models import Integration logger = get_safe_logger(__name__) # Hard backstop for the date-optional usage listing so a cleared-date-picker # call can't materialize unbounded rows (accounts × keys × days) into one JSON # body. Well above any legitimate date-scoped query; a hit is logged, never # silently truncated. _MAX_USAGE_ROWS = 10000 class BailianService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ── Integration resolution ────────────────────────────────── async def _get_integration(self, integration_id: str) -> Integration: row = await self.db.get(Integration, integration_id) if row is None: raise NotFoundError("Integration", integration_id) if row.kind != "cloud" or row.provider not in _SUPPORTED_PROVIDERS: raise AppException( code="BAILIAN_WRONG_INTEGRATION_KIND", message=( f"Bailian operations require an Aliyun-family cloud " f"integration (provider in " f"{sorted(_SUPPORTED_PROVIDERS)}; got kind={row.kind!r}, " f"provider={row.provider!r})." ), status_code=400, ) return row async def _get_client(self, integration_id: str) -> BailianClient: integration = await self._get_integration(integration_id) return BailianClient(resolve_credentials(integration)) # ── List endpoints ────────────────────────────────────────── async def list_workspaces( self, *, integration_id: str | None = None, owner_person_id: str | None = None, ) -> list[BailianWorkspaceResponse]: stmt = select(BailianWorkspace).where( BailianWorkspace.deleted_at.is_(None), ) if integration_id: stmt = stmt.where( BailianWorkspace.integration_id == integration_id, ) if owner_person_id: stmt = stmt.where( BailianWorkspace.owner_person_id == owner_person_id, ) stmt = stmt.order_by(BailianWorkspace.name) rows = (await self.db.execute(stmt)).scalars().all() # Eager-load denormalised counts + names. Two queries vs N+1. # Application counts intentionally absent — international # ModelStudio has no application catalog API; the dashboard # shows API keys / usage instead. ws_ids = [r.id for r in rows] key_count_map: dict[UUID, int] = {} if ws_ids: key_counts = (await self.db.execute( select( BailianApiKey.workspace_pk, func.count(BailianApiKey.id), ) .where( BailianApiKey.workspace_pk.in_(ws_ids), BailianApiKey.deleted_at.is_(None), ) .group_by(BailianApiKey.workspace_pk) )).all() key_count_map = {r[0]: int(r[1]) for r in key_counts} integ_ids = {r.integration_id for r in rows} integ_map = await self._load_integrations(integ_ids) owner_map = await self._load_owner_labels( [r.owner_person_id for r in rows if r.owner_person_id], ) out: list[BailianWorkspaceResponse] = [] for r in rows: integ = integ_map.get(r.integration_id) out.append(BailianWorkspaceResponse.model_validate( r, ).model_copy(update={ "integration_name": integ.name if integ else None, "owner_label": ( owner_map.get(r.owner_person_id) if r.owner_person_id else None ), "api_key_count": key_count_map.get(r.id, 0), })) return out async def list_api_keys( self, *, integration_id: str | None = None, owner_person_id: str | None = None, ) -> list[BailianApiKeyResponse]: stmt = ( select(BailianApiKey, BailianWorkspace.name) .outerjoin( BailianWorkspace, BailianApiKey.workspace_pk == BailianWorkspace.id, ) .where(BailianApiKey.deleted_at.is_(None)) ) if integration_id: stmt = stmt.where(BailianApiKey.integration_id == integration_id) if owner_person_id: stmt = stmt.where(BailianApiKey.owner_person_id == owner_person_id) stmt = stmt.order_by(BailianApiKey.created_at.desc()) rows = (await self.db.execute(stmt)).all() integ_map = await self._load_integrations( {row[0].integration_id for row in rows}, ) owner_map = await self._load_owner_labels( [row[0].owner_person_id for row in rows if row[0].owner_person_id], ) out: list[BailianApiKeyResponse] = [] for key, ws_name in rows: integ = integ_map.get(key.integration_id) out.append(BailianApiKeyResponse.model_validate(key).model_copy( update={ "integration_name": integ.name if integ else None, "workspace_name": ws_name, "owner_label": ( owner_map.get(key.owner_person_id) if key.owner_person_id else None ), }, )) return out async def list_usage_daily( self, *, integration_id: str | None = None, workspace_pk: UUID | None = None, api_key_pk: UUID | None = None, bailian_api_key_id: str | None = None, aliyun_account_id: str | None = None, owner_person_id: str | None = None, start_date: date | None = None, end_date: date | None = None, ) -> list[BailianUsageDailyResponse]: # OUTER joins all the way down — BSS billing rows are # account-scoped (workspace_pk=None, api_key_pk=None) and a # fresh Aliyun member account that BSS reports may not yet # have a matching cmdb_cloud_scopes row. INNER JOINs would # silently drop those rows even though the totals are real. # Migration 085 promoted ``aliyun_account_id`` to a real # column; we JOIN cmdb_cloud_scopes on # ``external_id = aliyun_account_id`` to attach scope name + # owner attribution. # # Migration 298 made spend per-API-key, so owner attribution now # PREFERS the API-key owner (the person the key belongs to) and # falls back to the account/scope owner only when the key isn't # one of ours (member-account keys we can't enumerate). That # matches what chargeback actually wants ("Alice's key spent # $X"), with the account owner as a backstop. KeyOwner = aliased(IdentityPerson) ScopeOwner = aliased(IdentityPerson) stmt = ( select( BailianUsageDaily, BailianWorkspace.name, BailianApiKey.display_name, CmdbCloudScope.display_name, BailianApiKey.owner_person_id, KeyOwner.display_name, KeyOwner.employee_no, KeyOwner.employment_status, CmdbCloudScope.owner_person_id, ScopeOwner.display_name, ScopeOwner.employee_no, ScopeOwner.employment_status, ) .outerjoin( BailianWorkspace, BailianUsageDaily.workspace_pk == BailianWorkspace.id, ) .outerjoin( BailianApiKey, BailianUsageDaily.api_key_pk == BailianApiKey.id, ) .outerjoin( KeyOwner, KeyOwner.id == BailianApiKey.owner_person_id, ) .outerjoin( # Restrict the JOIN to the SAME integration so a # multi-tenant deployment (two Aliyun integrations # whose RD member accounts overlap on OwnerID, or # stale duplicate scope rows) doesn't multiply usage # rows. ``cmdb_cloud_scopes`` uniqueness is # ``(integration_id, external_id)`` — see # cmdb/models.py — not ``external_id`` alone. CmdbCloudScope, (CmdbCloudScope.external_id == BailianUsageDaily.aliyun_account_id) & (CmdbCloudScope.integration_id == BailianUsageDaily.integration_id), ) .outerjoin( ScopeOwner, ScopeOwner.id == CmdbCloudScope.owner_person_id, ) ) if integration_id: stmt = stmt.where( BailianUsageDaily.integration_id == integration_id, ) if workspace_pk: stmt = stmt.where(BailianUsageDaily.workspace_pk == workspace_pk) if api_key_pk: stmt = stmt.where(BailianUsageDaily.api_key_pk == api_key_pk) if bailian_api_key_id: stmt = stmt.where( BailianUsageDaily.bailian_api_key_id == bailian_api_key_id, ) if aliyun_account_id: stmt = stmt.where( BailianUsageDaily.aliyun_account_id == aliyun_account_id, ) if owner_person_id: # Match either the API-key owner (primary) or the account # owner (fallback), so "Bailian spend for Alice" includes # both her keys and any account she owns. stmt = stmt.where( or_( BailianApiKey.owner_person_id == owner_person_id, CmdbCloudScope.owner_person_id == owner_person_id, ) ) if start_date: stmt = stmt.where(BailianUsageDaily.usage_date >= start_date) if end_date: stmt = stmt.where(BailianUsageDaily.usage_date <= end_date) # Order by date desc, then descending cost so the highest # spender of a given day surfaces first. stmt = stmt.order_by( BailianUsageDaily.usage_date.desc(), BailianUsageDaily.cost_cny.desc(), ) # Bounded backstop: usage rows grow without bound (accounts × keys × # days), and this endpoint's start_date/end_date are optional — a # caller that clears the date pickers would otherwise materialize the # entire history into memory and a single JSON body. Cap the fetch and # log (never silently truncate) so an operator hitting the ceiling # knows to narrow the date range. A legitimate date-scoped query is far # under this cap. Fetch cap+1 to detect the overflow. stmt = stmt.limit(_MAX_USAGE_ROWS + 1) rows = (await self.db.execute(stmt)).all() if len(rows) > _MAX_USAGE_ROWS: logger.warning( "bailian.list_usage_daily hit the %d-row cap (integration=%s " "start=%s end=%s) — result truncated; narrow the date range.", _MAX_USAGE_ROWS, integration_id, start_date, end_date, ) rows = rows[:_MAX_USAGE_ROWS] out: list[BailianUsageDailyResponse] = [] for (usage, ws_name, ak_name, scope_name, key_owner_id, key_owner_name, key_owner_emp_no, key_owner_emp_status, scope_owner_id, scope_owner_name, scope_owner_emp_no, scope_owner_emp_status) in rows: # Prefer the API-key owner; fall back to the account owner. # Keep the leaver indicators (employee_no / employment_status) # in lock-step with whichever owner won, so can # render the 已离职 treatment for departed owners. if key_owner_id: owner_id = key_owner_id owner_name = key_owner_name owner_emp_no = key_owner_emp_no owner_emp_status = key_owner_emp_status else: owner_id = scope_owner_id owner_name = scope_owner_name owner_emp_no = scope_owner_emp_no owner_emp_status = scope_owner_emp_status out.append( BailianUsageDailyResponse.model_validate(usage).model_copy( update={ "workspace_name": ws_name, "api_key_display_name": ak_name, "aliyun_account_name": scope_name, "owner_person_id": ( str(owner_id) if owner_id else None ), "owner_display_name": owner_name, "owner_employee_no": owner_emp_no, "owner_employment_status": owner_emp_status, }, ), ) return out async def usage_summary( self, *, integration_id: str, period_start: date | None = None, period_end: date | None = None, ) -> BailianUsageSummary: """Aggregate over a date range for the cost dashboard.""" if period_end is None: # Aliyun BSS data is stamped in Asia/Shanghai (UTC+8). Using # UTC date.today() under-counts up to 8 hours per day. Use the # Shanghai date so the summary aligns with the data source. from zoneinfo import ZoneInfo period_end = datetime.now(ZoneInfo("Asia/Shanghai")).date() if period_start is None: period_start = period_end - timedelta(days=29) if period_start > period_end: raise AppException( code="INVALID_PERIOD", message="period_start must be on or before period_end", status_code=400, ) integration = await self._get_integration(integration_id) base_filters = and_( BailianUsageDaily.integration_id == integration_id, BailianUsageDaily.usage_date >= period_start, BailianUsageDaily.usage_date <= period_end, ) totals = (await self.db.execute( select( func.coalesce(func.sum(BailianUsageDaily.input_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.output_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.request_count), 0), func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0), func.coalesce(func.sum(BailianUsageDaily.cost_cny), 0), ).where(base_filters) )).one() by_model_rows = (await self.db.execute( select( BailianUsageDaily.model_name, func.coalesce(func.sum(BailianUsageDaily.input_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.output_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.request_count), 0), func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0), ) .where(base_filters) .group_by(BailianUsageDaily.model_name) .order_by( func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).desc(), ) )).all() by_workspace_rows = (await self.db.execute( select( BailianUsageDaily.workspace_pk, BailianWorkspace.name, func.coalesce(func.sum(BailianUsageDaily.input_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.output_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.request_count), 0), func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0), ) # OUTER join — intl BSS rows have workspace_pk=None and # would be silently dropped by an INNER join. We surface # them under the synthetic name "(account aggregate)". .outerjoin( BailianWorkspace, BailianUsageDaily.workspace_pk == BailianWorkspace.id, ) .where(base_filters) .group_by(BailianUsageDaily.workspace_pk, BailianWorkspace.name) .order_by( func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).desc(), ) )).all() # by_owner — OUTER join so rows on keys we don't own (member # accounts, api_key_pk=NULL) still reconcile to the total under a # NULL "未归属" bucket instead of being silently dropped. by_owner_rows = (await self.db.execute( select( BailianApiKey.owner_person_id, func.coalesce(func.sum(BailianUsageDaily.input_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.output_tokens), 0), func.coalesce(func.sum(BailianUsageDaily.request_count), 0), func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0), ) .outerjoin( BailianApiKey, BailianUsageDaily.api_key_pk == BailianApiKey.id, ) .where(base_filters) .group_by(BailianApiKey.owner_person_id) .order_by( func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).desc(), ) )).all() owner_map = await self._load_owner_labels( [r[0] for r in by_owner_rows if r[0]], ) # by_api_key — the headline chargeback view (migration 298). Group # on the raw ModelStudio key id from the bill so member-account # keys (no local row) still appear; OUTER-join the synced key for # its display name + owner. by_api_key_rows = (await self.db.execute( select( BailianUsageDaily.bailian_api_key_id, BailianApiKey.display_name, BailianApiKey.owner_person_id, func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0), ) .outerjoin( BailianApiKey, BailianUsageDaily.api_key_pk == BailianApiKey.id, ) .where(base_filters) .group_by( BailianUsageDaily.bailian_api_key_id, BailianApiKey.display_name, BailianApiKey.owner_person_id, ) .order_by( func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).desc(), ) )).all() api_key_owner_map = await self._load_owner_labels( [r[2] for r in by_api_key_rows if r[2]], ) return BailianUsageSummary( integration_id=integration_id, integration_name=integration.name, period_start=period_start, period_end=period_end, total_input_tokens=int(totals[0]), total_output_tokens=int(totals[1]), total_request_count=int(totals[2]), total_cost_usd=Decimal(str(totals[3])), total_cost_cny=Decimal(str(totals[4])), by_model=[ { "model_name": r[0], "input_tokens": int(r[1]), "output_tokens": int(r[2]), "request_count": int(r[3]), "cost_usd": float(r[4]), } for r in by_model_rows ], by_workspace=[ { "workspace_pk": str(r[0]) if r[0] else None, # Account-aggregate rows have workspace_pk=None # (BSS bills at the account level on intl). Surface # a stable display label so the dashboard legend # isn't blank. "workspace_name": r[1] or "(account aggregate)", "input_tokens": int(r[2]), "output_tokens": int(r[3]), "request_count": int(r[4]), "cost_usd": float(r[5]), } for r in by_workspace_rows ], by_owner=[ { "owner_person_id": r[0], "owner_label": owner_map.get(r[0]) if r[0] else None, "input_tokens": int(r[1]), "output_tokens": int(r[2]), "request_count": int(r[3]), "cost_usd": float(r[4]), } for r in by_owner_rows ], by_api_key=[ { "bailian_api_key_id": r[0], "api_key_display_name": r[1], "owner_person_id": r[2], "owner_label": ( api_key_owner_map.get(r[2]) if r[2] else None ), "cost_usd": float(r[3]), } for r in by_api_key_rows ], ) # ── Sync orchestration ────────────────────────────────────── async def workspace_region( self, *, integration_id: str, workspace_id: str | None, ) -> str | None: """``workspace_id`` → 它所属的 ModelStudio 区域。 ModelStudio 是分区域的:``CreateApiKey`` / ``DeleteApiKey`` / ``ListApiKeys`` 打错区域会返回 HTTP 200 但什么也不做(删不掉、 看不见),**不报错**。凡是对某个 workspace 下的 Key 做写操作, 都要先拿这个区域。 未知 workspace / 未回填 region 的老行 → ``None``,调用方回落到 客户端默认区域(新加坡),与本次改动前的行为一致。 """ if not workspace_id: return None row = (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == integration_id, BailianWorkspace.workspace_id == workspace_id, BailianWorkspace.deleted_at.is_(None), ) )).scalar_one_or_none() return row.region if row else None async def sync_workspaces(self, *, integration_id: str) -> dict[str, Any]: """Pull all workspaces for one integration, upsert into local DB. Returns a summary dict ``{discovered, upserted, soft_deleted}`` suitable for surfacing on the manual-trigger UI and in scheduled-job logs. """ client = await self._get_client(integration_id) # **逐区域**枚举 —— ModelStudio 的 ``ListWorkspaces`` 只返回调用 # 端点所在区域的那个「默认业务空间」。只打新加坡时,北京/弗吉尼亚 # 的空间在 Helix 里根本不存在,它们的 Key 也就永远同步不到,费用 # 全部解析不到人、兜底记到账号 Owner 名下。2026-08-15 生产实测见 # ``client._MODELSTUDIO_REGIONS`` 上方注释。 # # 单个区域失败不拖垮其余区域,但**必须记下来**:静默少一个区域 # 等于静默丢一批 Key,而下面的漂移清扫会把该区域已有的行软删。 discovered: list[dict[str, Any]] = [] region_errors: list[str] = [] for reg in _MODELSTUDIO_REGIONS: try: discovered.extend(await client.list_workspaces(region=reg)) except Exception as exc: # noqa: BLE001 — 逐区域降级 region_errors.append(f"{reg}: {type(exc).__name__}") logger.warning( "bailian.sync_workspaces region failed integration=%s " "region=%s err=%s", integration_id, reg, format_exception_for_log(exc), ) if region_errors and not discovered: raise AppException( code="BAILIAN_WORKSPACE_SYNC_ALL_REGIONS_FAILED", message=( "ModelStudio ListWorkspaces failed in every region: " + "; ".join(region_errors) ), status_code=502, ) existing_map = { r.workspace_id: r for r in (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == integration_id, ) )).scalars().all() } now = datetime.now(UTC) upserted = 0 seen_ids: set[str] = set() for item in discovered: ws_id = item["workspace_id"] if not ws_id: continue seen_ids.add(ws_id) row = existing_map.get(ws_id) if row is None: row = BailianWorkspace( integration_id=integration_id, workspace_id=ws_id, name=item["name"], description=item.get("description"), region=item.get("region"), is_default=bool(item.get("is_default")), raw_data=item.get("raw"), last_synced_at=now, ) self.db.add(row) else: row.name = item["name"] or row.name row.description = item.get("description") or row.description row.region = item.get("region") or row.region # ``is_default`` is deliberately NOT overwritten here. # Intl ModelStudio's ListWorkspaces carries no isDefault # field, so ``item["is_default"]`` is always False — this # line used to wipe the admin's choice on every nightly # run, which made "mark one workspace as default" a # remediation that silently undid itself within a day. # The flag is Helix-owned; only the admin endpoint and # the insert branch below write it. row.raw_data = item.get("raw") or row.raw_data row.deleted_at = None # rediscovery un-soft-deletes row.status = "active" row.last_synced_at = now upserted += 1 # Soft-delete rows that were not seen this run — but apply the # same percentage-floor safety guard the seat-sync handlers # use (P1 from CR2). A buggy ModelStudio paginator that # returns a partial subset (HTTP 200 with truncated data plus # ``has_more=true``) would otherwise mark every existing # workspace as deleted in one tick, cascading to every # BailianApiKey + usage row in the cost dashboard. from app.scheduled_jobs._seat_sync_helpers import ( safe_to_drift_revoke, ) active_existing = [ r for r in existing_map.values() if r.deleted_at is None ] safe, refuse_reason = safe_to_drift_revoke( matched_count=len(seen_ids), existing_count=len(active_existing), label=f"Bailian workspaces (integration={integration_id})", ) soft_deleted = 0 if not safe: logger.warning( "bailian.sync_workspaces: drift cleanup refused " "(integration=%s): %s", integration_id, refuse_reason, ) else: for ws_id, row in existing_map.items(): if ws_id in seen_ids: continue if row.deleted_at is None: row.deleted_at = now row.status = "deleted" soft_deleted += 1 await self.db.flush() return { "integration_id": integration_id, "discovered": len(discovered), "upserted": upserted, "soft_deleted": soft_deleted, "soft_delete_skipped_reason": ( refuse_reason if not safe else None ), } # International ModelStudio has no application catalog API; the # applications surface (sync, list, scheduled job, table) was # removed in the same change as the model. CN Bailian had it under # a different product code — restore here behind provider routing # if Helix ever supports CN. async def sync_api_keys( self, *, integration_id: str, ) -> dict[str, Any]: """Pull Bailian API keys for every workspace in this integration. Workspace-fanout shape: one ``ListApiKeys`` call per workspace, upserted into :class:`BailianApiKey` keyed on ``(integration_id, api_key_id)``. Idempotent — re-running this method against unchanged upstream state is a no-op. Soft-deletes upstream-sourced rows that disappeared from the ModelStudio response. Autogrant-minted rows (``assignment_id IS NOT NULL``) are NEVER soft-deleted by this sync: their ``api_key_id`` is the RAM AccessKey ID (``LTAI…``), not the numeric ModelStudio ``apiKeyId``, so they will never appear in the upstream listing — wiping them every nightly run would orphan every Helix-issued credential bundle. Bulk-loads existing rows once before the per-workspace loop to avoid an N+1 query. """ client = await self._get_client(integration_id) workspaces = (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == integration_id, BailianWorkspace.deleted_at.is_(None), ) )).scalars().all() ws_pk_by_external = {ws.workspace_id: ws.id for ws in workspaces} # Complete every remote pagination walk before touching ORM rows. If # one workspace fails, no earlier workspace's partial snapshot is # staged in the session and no drift sweep can run. discovered_by_workspace: list[ tuple[BailianWorkspace, list[dict[str, Any]]] ] = [] for ws in workspaces: # **必须**按该 workspace 自己的区域取:打错区域返回 # HTTP 200 + 空列表(不报错),下面的漂移清扫就会把这个 # 区域的 Key 全部软删。``ws.region`` 由 sync_workspaces 用 # 端点区域回填,老行可能为 NULL → 回落默认(新加坡),与 # 改动前行为一致。 discovered_by_workspace.append(( ws, await client.list_api_keys( workspace_id=ws.workspace_id, region=ws.region, ), )) # Bulk-load every existing key for this integration once; # per-row SELECT inside the inner loop was N+1. existing_by_id: dict[str, BailianApiKey] = { r.api_key_id: r for r in (await self.db.execute( select(BailianApiKey).where( BailianApiKey.integration_id == integration_id, ) )).scalars().all() } seen: set[str] = set() now = datetime.now(UTC) total_discovered = 0 total_upserted = 0 for ws, discovered in discovered_by_workspace: total_discovered += len(discovered) for item in discovered: api_key_id = item["api_key_id"] seen.add(api_key_id) row = existing_by_id.get(api_key_id) workspace_pk = ws_pk_by_external.get(item["workspace_id"]) if row is None: new_row = BailianApiKey( integration_id=integration_id, workspace_pk=workspace_pk, api_key_id=api_key_id, display_name=item["display_name"], ram_user_name=item.get("created_by"), status="disabled" if item.get("disabled") else "active", raw_data=item.get("raw"), last_synced_at=now, ) self.db.add(new_row) existing_by_id[api_key_id] = new_row elif row.deleted_at is not None and row.assignment_id is not None: # Revoke-retired ADOPTED key. mark_api_key_deleted_on_revoke # is the ONLY path that soft-deletes an assignment-linked # row (the drift sweep below skips assignment_id IS NOT # NULL). Aliyun DeleteApiKey is eventually consistent, so a # just-revoked key can still surface in ListApiKeys for a # window; un-deleting it here would reopen the two-table # drift #1230 closed (row active again while its assignment # is revoked). Leave it retired; only refresh the sync ts. row.last_synced_at = now else: # Don't overwrite display_name if autogrant set a # human-friendly one ("Helix · APIKEY-12345 · …"). # Only update from upstream when the local row is # still the bare ``apikey-`` placeholder. if row.display_name.startswith("apikey-"): row.display_name = item["display_name"] row.workspace_pk = workspace_pk or row.workspace_pk row.status = ( "disabled" if item.get("disabled") else "active" ) row.raw_data = item.get("raw") or row.raw_data row.deleted_at = None row.last_synced_at = now total_upserted += 1 # Soft-delete only upstream-sourced rows we didn't see this run. # Autogrant rows (assignment_id IS NOT NULL) carry the RAM # AccessKey ID, NOT the ModelStudio numeric apiKeyId, so they # are intentionally absent from the upstream listing and must # never be wiped here. # # Same percentage-floor guard as ``sync_workspaces`` (P1 from # CR2): a partial-truncation response would otherwise mark # every upstream API key as deleted, leaving the cost # dashboard's per-key attribution empty. from app.scheduled_jobs._seat_sync_helpers import ( safe_to_drift_revoke, ) upstream_existing = [ r for r in existing_by_id.values() if r.deleted_at is None and r.assignment_id is None ] safe, refuse_reason = safe_to_drift_revoke( matched_count=len(seen), existing_count=len(upstream_existing), label=f"Bailian api_keys (integration={integration_id})", ) soft_deleted = 0 if not safe: logger.warning( "bailian.sync_api_keys: drift cleanup refused " "(integration=%s): %s", integration_id, refuse_reason, ) else: for api_key_id, row in existing_by_id.items(): if api_key_id in seen: continue if row.deleted_at is not None: continue if row.assignment_id is not None: continue row.deleted_at = now row.status = "deleted" soft_deleted += 1 await self.db.flush() return { "integration_id": integration_id, "workspaces_processed": len(workspaces), "discovered": total_discovered, "upserted": total_upserted, "soft_deleted": soft_deleted, "soft_delete_skipped_reason": ( refuse_reason if not safe else None ), } async def mark_api_key_deleted_on_revoke( self, *, integration_id: str, api_key_id: str, ) -> dict[str, Any]: """Retire the ``bailian_api_keys`` mirror row when an assignment revoke deletes its backing DashScope key on Aliyun. The revoke path (apikey domain) calls ``BailianClient.delete_api_key`` to delete the real key, then calls this to mirror that deletion into the local registry so the cost dashboard and admin list stop showing a key that no longer exists upstream. Keyed on ``(integration_id, api_key_id)`` — the SAME natural key ``sync_api_keys`` upserts on. This hook is load-bearing: ``sync_api_keys`` NEVER soft-deletes rows whose ``assignment_id`` is set (they may carry RAM AccessKey IDs that never appear in the ModelStudio listing), so without an explicit retire-on-revoke the row for an adopted key would linger ``status='active'`` / ``deleted_at IS NULL`` forever after its key was deleted. Idempotent: a missing or already soft-deleted row is a no-op. Sets the ORM attributes only — the caller's transaction flushes them alongside the assignment status flip, so the registry retire and the revoke commit atomically. """ row = (await self.db.execute( select(BailianApiKey).where( BailianApiKey.integration_id == integration_id, BailianApiKey.api_key_id == api_key_id, ) )).scalar_one_or_none() if row is None: return {"matched": False, "already_deleted": False} if row.deleted_at is not None: return {"matched": True, "already_deleted": True} row.deleted_at = datetime.now(UTC) row.status = "deleted" return {"matched": True, "already_deleted": False} async def sync_usage_daily( self, *, integration_id: str, days: int = 7, ) -> dict[str, Any]: """Pull last-N-days of per-(day × API key × model) Bailian cost. Production-grade source: Aliyun BSS ``DescribeInstanceBill`` (Granularity=DAILY) filtered to ``ProductCode=sfm``. Each row's ``InstanceID`` is a composite ``apiKeyId;workspaceId;model; tokenType;channel;0`` (verified prod 2026-06-17), so spend lands at real per-API-key / per-model granularity — superseding the old account-level monthly aggregate. ``OwnerID`` → ``aliyun_ account_id`` (the spending member account). When the API key is one of ours (synced into ``bailian_api_keys``) the row also gets ``api_key_pk`` + the owning workspace, so the dashboard attributes spend to a person; keys living in Resource-Directory member accounts that the integration AK can't enumerate keep ``api_key_pk=NULL`` and attribute by account. Re-runs are idempotent — composite unique on ``(integration_id, usage_date, aliyun_account_id, bailian_api_key_id, model_name)``. """ # Up to ~14 months so a one-off backfill can re-pull the whole # dashboard lookback (days selects how many trailing days are # queried — one DescribeInstanceBill DAILY call per day). if days < 1 or days > 430: raise AppException( code="INVALID_USAGE_DAYS", message="days must be between 1 and 430", status_code=400, ) client = await self._get_client(integration_id) rate = await self._get_cny_to_usd_rate() # Aliyun BSS billing keys on UTC dates. Using ``date.today()`` # would skew by one day if the worker runs in Asia/Shanghai # (prod EC2). Use UTC explicitly to match BSS's own clock. end = datetime.now(UTC).date() start = end - timedelta(days=days - 1) discovered = await client.query_instance_bill_daily( start_date=start.isoformat(), end_date=end.isoformat(), cny_to_usd_rate=rate, ) # Production-grade write strategy — diff-based, not # delete-then-insert. A transient empty BSS response (verified # to happen during regional failovers) must NOT silently wipe # cost history, so 0 rows → leave the table untouched and warn. if not discovered: logger.warning( "bailian.usage: BSS returned 0 rows for " "integration %s in window %s..%s; leaving the " "existing rows untouched. If this persists, " "investigate BSS connectivity — re-running sync " "won't help.", integration_id, start.isoformat(), end.isoformat(), ) return { "integration_id": integration_id, "discovered": 0, "upserted": 0, "deleted_stale": 0, "deleted_legacy": 0, "period_start": start.isoformat(), "period_end": end.isoformat(), } now = datetime.now(UTC) # Resolve raw ids → local PKs for attribution. The API-key map # keys on the ModelStudio numeric id (``api_key_id``); workspaces # on the workspace id string. api_key_pk_by_id = { r.api_key_id: r.id for r in (await self.db.execute( select(BailianApiKey).where( BailianApiKey.integration_id == integration_id, ) )).scalars().all() } workspace_pk_by_id = { r.workspace_id: r.id for r in (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == integration_id, ) )).scalars().all() } # Cutover safety: the previous ingest wrote ONE # ``model_name='aggregate'`` row per month (dated first-of-month). # Within a covered month those would double-count alongside the new # per-key rows (and sit outside the day-range sweep), so drop the # legacy shape — but ONLY for the MONTHS this run actually # re-ingests, so a 7-day sync doesn't wipe an older month's # aggregate that the per-key path isn't repopulating. Idempotent — # the new ingest never writes ``aggregate``. legacy_window_start = start.replace(day=1) _em = end.replace(day=1) legacy_window_end_excl = ( _em.replace(year=_em.year + 1, month=1) if _em.month == 12 else _em.replace(month=_em.month + 1) ) deleted_legacy = (await self.db.execute( delete(BailianUsageDaily).where( BailianUsageDaily.integration_id == integration_id, BailianUsageDaily.model_name == BAILIAN_AGGREGATE_MODEL, BailianUsageDaily.usage_date >= legacy_window_start, BailianUsageDaily.usage_date < legacy_window_end_excl, ) )).rowcount or 0 # Index existing rows by the new full unique key, over the literal # day window (DescribeInstanceBill is daily, so no month expansion). existing_rows = (await self.db.execute( select(BailianUsageDaily).where( BailianUsageDaily.integration_id == integration_id, BailianUsageDaily.usage_date >= start, BailianUsageDaily.usage_date <= end, BailianUsageDaily.model_name != BAILIAN_AGGREGATE_MODEL, ) )).scalars().all() existing_by_key: dict[ tuple[date, str | None, str | None, str], BailianUsageDaily, ] = { (r.usage_date, r.aliyun_account_id, r.bailian_api_key_id, r.model_name): r for r in existing_rows } def _add_opt(a: Decimal | None, b: Decimal | None) -> Decimal | None: if a is None and b is None: return None return (a or Decimal("0")) + (b or Decimal("0")) upserted = 0 seen_keys: set[tuple[date, str | None, str | None, str]] = set() # Holds the row (new or updated) already materialised for each grain # key THIS run, so a second source row at the same grain folds into it # instead of issuing a duplicate INSERT. pending_by_key: dict[ tuple[date, str | None, str | None, str], BailianUsageDaily, ] = {} for item in discovered: try: usage_dt = date.fromisoformat(item["usage_date"][:10]) except ValueError: continue account_id = item.get("aliyun_account_id") api_key_id_raw = item.get("bailian_api_key_id") workspace_id_raw = item.get("workspace_id") model_name = item["model_name"] key = (usage_dt, account_id, api_key_id_raw, model_name) seen_keys.add(key) api_key_pk = ( api_key_pk_by_id.get(api_key_id_raw) if api_key_id_raw else None ) workspace_pk = ( workspace_pk_by_id.get(workspace_id_raw) if workspace_id_raw else None ) cost_usd = Decimal(str(item["cost_usd"])) cost_cny = Decimal(str(item["cost_cny"])) cost_usd_gross = ( Decimal(str(item["cost_usd_gross"])) if item.get("cost_usd_gross") is not None else None ) cost_cny_gross = ( Decimal(str(item["cost_cny_gross"])) if item.get("cost_cny_gross") is not None else None ) cost_usd_ad = ( Decimal(str(item["cost_usd_after_discount"])) if item.get("cost_usd_after_discount") is not None else None ) cost_cny_ad = ( Decimal(str(item["cost_cny_after_discount"])) if item.get("cost_cny_after_discount") is not None else None ) raw = item.get("raw") folded = pending_by_key.get(key) if folded is not None: # Two source rows collapse onto the same DB chargeback grain # (the unique key is (date, account, api_key, model) — NO # workspace), e.g. one api-key's spend split across workspaces # or keyless account-level spend in >1 workspace on the same # day+model. SUM the native costs so the persisted row is the # TRUE per-grain total; a blind second INSERT would trip # uq_bailian_usage_integ_date_acct_key_model and crash the # whole integration's usage sync. Workspace becomes ambiguous # at this grain → null it rather than claim one of several. folded.input_tokens += item["input_tokens"] folded.output_tokens += item["output_tokens"] folded.request_count += item["request_count"] folded.cost_usd = folded.cost_usd + cost_usd folded.cost_cny = folded.cost_cny + cost_cny folded.cost_usd_gross = _add_opt(folded.cost_usd_gross, cost_usd_gross) folded.cost_cny_gross = _add_opt(folded.cost_cny_gross, cost_cny_gross) folded.cost_usd_after_discount = _add_opt( folded.cost_usd_after_discount, cost_usd_ad) folded.cost_cny_after_discount = _add_opt( folded.cost_cny_after_discount, cost_cny_ad) if folded.workspace_id_raw != workspace_id_raw: folded.workspace_id_raw = None folded.workspace_pk = None continue existing = existing_by_key.get(key) if existing is None: row = BailianUsageDaily( integration_id=integration_id, workspace_pk=workspace_pk, api_key_pk=api_key_pk, aliyun_account_id=account_id, bailian_api_key_id=api_key_id_raw, workspace_id_raw=workspace_id_raw, usage_date=usage_dt, model_name=model_name, input_tokens=item["input_tokens"], output_tokens=item["output_tokens"], request_count=item["request_count"], cost_usd=cost_usd, cost_cny=cost_cny, cost_usd_gross=cost_usd_gross, cost_cny_gross=cost_cny_gross, cost_usd_after_discount=cost_usd_ad, cost_cny_after_discount=cost_cny_ad, raw_data=raw, synced_at=now, ) self.db.add(row) else: existing.workspace_pk = workspace_pk existing.api_key_pk = api_key_pk existing.workspace_id_raw = workspace_id_raw existing.input_tokens = item["input_tokens"] existing.output_tokens = item["output_tokens"] existing.request_count = item["request_count"] existing.cost_usd = cost_usd existing.cost_cny = cost_cny existing.cost_usd_gross = cost_usd_gross existing.cost_cny_gross = cost_cny_gross existing.cost_usd_after_discount = cost_usd_ad existing.cost_cny_after_discount = cost_cny_ad existing.raw_data = raw existing.synced_at = now row = existing pending_by_key[key] = row upserted += 1 # Sweep stale rows — but ONLY within DAYS that actually returned # data this run, so a transient empty-day response (the failure # mode the 0-rows guard cites, but per-day) doesn't delete a good # historical row. seen_days = {k[0] for k in seen_keys} deleted_stale = 0 for key, row in existing_by_key.items(): if key not in seen_keys and row.usage_date in seen_days: await self.db.delete(row) deleted_stale += 1 await self.db.flush() return { "integration_id": integration_id, "discovered": len(discovered), "upserted": upserted, "deleted_stale": deleted_stale, "deleted_legacy": deleted_legacy, "period_start": start.isoformat(), "period_end": end.isoformat(), } async def _get_cny_to_usd_rate(self) -> float: """Read the AI-spending-managed CNY→USD rate. Used by the BSS billing sync to normalise PretaxAmount (always CNY) into the USD canonical form persisted on ``bailian_usage_daily.cost_usd``. Delegates to :class:`AiSpendingService._get_cny_rate` so the validation (raise on missing / corrupt / out-of-range) is the single source of truth. """ from app.domains.ai_spending.service import AiSpendingService return await AiSpendingService(self.db)._get_cny_rate() # ── Workspace owner reassignment ──────────────────────────── async def reassign_workspace_owner( self, *, workspace_pk: UUID, owner_person_id: UUID | None, actor_id: UUID, reason: str, trace_id: str, ) -> BailianWorkspaceResponse: ws = await self.db.get(BailianWorkspace, workspace_pk) if ws is None or ws.deleted_at is not None: raise NotFoundError("BailianWorkspace", str(workspace_pk)) previous = ws.owner_person_id if owner_person_id is not None: person = await self.db.get(IdentityPerson, owner_person_id) if person is None: raise NotFoundError("IdentityPerson", str(owner_person_id)) ws.owner_person_id = str(person.id) else: ws.owner_person_id = None await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="bailian.workspace.owner.update", resource_type="bailian_workspace", resource_id=str(ws.id), status="success", detail={ "workspace_id": ws.workspace_id, "previous_owner_person_id": previous, "new_owner_person_id": ws.owner_person_id, "reason": reason, }, ) owner_map = await self._load_owner_labels( [ws.owner_person_id] if ws.owner_person_id else [], ) integ = await self.db.get(Integration, ws.integration_id) return BailianWorkspaceResponse.model_validate(ws).model_copy( update={ "integration_name": integ.name if integ else None, "owner_label": ( owner_map.get(ws.owner_person_id) if ws.owner_person_id else None ), }, ) # ── Default-workspace selection (auto-grant target) ───────── async def set_default_workspace( self, *, workspace_pk: UUID, is_default: bool, actor_id: UUID, reason: str, trace_id: str, ) -> BailianWorkspaceResponse: """Pick the workspace new API keys are minted in. Exactly one workspace per integration may hold the flag, so setting it clears every sibling in the same transaction — two flagged workspaces read as ambiguous to the auto-grant resolver and would fail the grant just as loudly as zero. """ ws = await self.db.get(BailianWorkspace, workspace_pk) if ws is None or ws.deleted_at is not None: raise NotFoundError("BailianWorkspace", str(workspace_pk)) cleared: list[str] = [] if is_default: siblings = (await self.db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == ws.integration_id, BailianWorkspace.id != ws.id, BailianWorkspace.is_default.is_(True), ) )).scalars().all() for other in siblings: other.is_default = False cleared.append(other.workspace_id) ws.is_default = is_default await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="bailian.workspace.default.update", resource_type="bailian_workspace", resource_id=str(ws.id), status="success", detail={ "workspace_id": ws.workspace_id, "region": ws.region, "is_default": is_default, "cleared_workspace_ids": cleared, "reason": reason, }, ) owner_map = await self._load_owner_labels( [ws.owner_person_id] if ws.owner_person_id else [], ) integ = await self.db.get(Integration, ws.integration_id) return BailianWorkspaceResponse.model_validate(ws).model_copy( update={ "integration_name": integ.name if integ else None, "owner_label": ( owner_map.get(ws.owner_person_id) if ws.owner_person_id else None ), }, ) # ── API key owner reassignment ────────────────────────────── async def reassign_api_key_owner( self, *, api_key_pk: UUID, owner_person_id: UUID | None, actor_id: UUID, reason: str, trace_id: str, ) -> BailianApiKeyResponse: key = await self.db.get(BailianApiKey, api_key_pk) if key is None or key.deleted_at is not None: raise NotFoundError("BailianApiKey", str(api_key_pk)) previous = key.owner_person_id if owner_person_id is not None: person = await self.db.get(IdentityPerson, owner_person_id) if person is None: raise NotFoundError("IdentityPerson", str(owner_person_id)) key.owner_person_id = str(person.id) else: key.owner_person_id = None # 纳管'd key (linked to a registry assignment): the assignment is # the single source of truth for owner — cost attribution, AWS IAM # tags, the shared-access roster and offboarding force-transfer all # key off ``api_key_assignments.assigned_to``. Route the change # through the apikey owner coordinator so all of that moves with # it; the registry transfer's own owner-mirror sync writes # ``key.owner_person_id`` back, so both surfaces agree afterward. # Clearing the owner of a managed key would orphan the assignment # (whose ``assigned_to`` is NOT NULL), so we refuse it — the admin # must transfer to a person, release the key, or解除纳管 first. if key.assignment_id is not None: if owner_person_id is None: raise ConflictError( "此 API Key 已纳管，负责人不能清空。请转让给具体的人，" "或在 API Key 注册表中释放该 Key。", ) from app.core.coordinators.apikey_owner_coordinator import ( # noqa: PLC0415 reassign_linked_assignment_owner, ) await reassign_linked_assignment_owner( self.db, assignment_id=key.assignment_id, new_owner_id=owner_person_id, trace_id=trace_id, ) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="bailian.api_key.owner.update", resource_type="bailian_api_key", resource_id=str(key.id), status="success", detail={ "api_key_id": key.api_key_id, "previous_owner_person_id": previous, "new_owner_person_id": key.owner_person_id, "reason": reason, }, ) integ = await self.db.get(Integration, key.integration_id) ws_name: str | None = None if key.workspace_pk: ws = await self.db.get(BailianWorkspace, key.workspace_pk) ws_name = ws.name if ws else None owner_map = await self._load_owner_labels( [key.owner_person_id] if key.owner_person_id else [], ) return BailianApiKeyResponse.model_validate(key).model_copy( update={ "integration_name": integ.name if integ else None, "workspace_name": ws_name, "owner_label": ( owner_map.get(key.owner_person_id) if key.owner_person_id else None ), }, ) # ── helpers ───────────────────────────────────────────────── async def _load_integrations( self, integration_ids: set[str], ) -> dict[str, Integration]: 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 for r in rows} async def _load_owner_labels( self, person_ids: list[str], ) -> dict[str, str]: clean = [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: label = ( f"{r.display_name or r.work_email or ''} · " f"{r.work_email or ''}" ).strip(" ·") out[str(r.id)] = label return out