"""Aliyun Bailian (Model Studio) HTTP client. Wraps three Aliyun OpenAPI surfaces with the existing ACS3-HMAC-SHA256 signing helper from :class:`AliyunCloudDriver`: * ``bailian..aliyuncs.com`` — Bailian OpenAPI (workspaces, applications). API version is the public ``2023-12-29`` revision; if Aliyun adds new operations we can pin a newer version here. * ``ram.aliyuncs.com`` — RAM (CreateUser, AttachPolicyToUser, CreateAccessKey). Already heavily used by ``aliyun_cloud.py`` for per-employee access provisioning; reused here for Bailian RAM users. * ``business.aliyuncs.com`` — BSS billing query for cost data (``QueryAccountBalance``, ``QueryProductPrice``, etc.). The client never persists secrets to disk — credentials come from the ``Integration.secrets_encrypted`` row, decrypted in-memory via :func:`decrypt_fields` per-call. Why not import :class:`AliyunCloudDriver` and call its private ``_acs3_call`` directly: that driver registers itself via ``@register_driver`` and brings the whole CMDB sync surface into scope; we only need the signed-request helper. Duplicating the signer (≈40 lines) keeps the Bailian module standalone and avoids circular import risk between ``bailian`` and ``cmdb`` / ``integrations``. """ from __future__ import annotations import datetime as _dt import hashlib import hmac import json import logging import urllib.parse import uuid from typing import Any import httpx from app.core.exceptions import AppException from app.core.pagination import PaginationGuard from app.domains.cmdb.credential_crypto import decrypt_fields from app.integrations.models import Integration logger = logging.getLogger(__name__) # Bailian / Model Studio is consumed via the AliCloud International # (海外) tenant only. PingCAP's 国内阿里云 account has never enabled # Bailian, so the integration filter, scheduled jobs, autogrant and # UI all narrow to ``alicloud_intl``. The 国内 ``aliyun`` provider # stays in the system for CMDB / RAM / billing purposes — Bailian # just doesn't touch it. If 国内 Bailian is ever needed, add the # routing entry back here. _DEFAULT_REGION_BY_PROVIDER = { "alicloud_intl": "ap-southeast-1", } _BSS_HOST_BY_PROVIDER = { # International BSS only. Calling ``business.aliyuncs.com`` (CN BSS) # from an intl AK returns ``NotApplicable: caller site doesn't match # API region`` (verified 2026-04-25 against prod 海外 tenant). "alicloud_intl": "business.ap-southeast-1.aliyuncs.com", } _DASHSCOPE_ENDPOINT_BY_PROVIDER = { "alicloud_intl": "https://dashscope-intl.aliyuncs.com/api/v1", } _SUPPORTED_PROVIDERS = frozenset(_DEFAULT_REGION_BY_PROVIDER) _DEFAULT_PROVIDER = "alicloud_intl" _DEFAULT_REGION = _DEFAULT_REGION_BY_PROVIDER[_DEFAULT_PROVIDER] # International Bailian / Model Studio is a SEPARATE Aliyun OpenAPI # product (``ModelStudio``, version ``2026-02-10``) — not the CN # ``bailian/2023-12-29`` product. Different host (``modelstudio.*`` # instead of ``bailian.*``), ROA style (path-based GET) instead of # RPC, and a much smaller surface (only 7 actions: CreateWorkspace, # ListWorkspaces, CreateApiKey, GetApiKey, ListApiKeys, UpdateApiKey, # DeleteApiKey). Notably ``ListApplications`` does NOT exist on intl # — the application catalog lives only in the DashScope data plane, # so the application-sync path is a graceful no-op below. # Spec: https://api.aliyun.com/meta/v1/products/ModelStudio/versions/2026-02-10/api-docs.json _MODELSTUDIO_HOST = "modelstudio.ap-southeast-1.aliyuncs.com" # ModelStudio 是**分区域**的:每个区域有自己独立的「默认业务空间」,而 # ``ListWorkspaces`` 只返回**调用端点所在区域**的那一个。2026-08-15 生产 # 实测(同一把 AK,同一个账号 5962436752113096): # # ap-southeast-1 → llm-tbeflduouo79t4fv(28 把 Key) # cn-beijing → ws-glxk2cmse8jwzzgz(5 把 Key,含当月最大的 $4,960) # us-east-1 → ws-7tcent4dizu8z1s7(1 把 Key) # # 只打新加坡端点时,后两个区域的 Key 在 Helix 里根本不存在 —— 它们的费用 # 于是解析不到人、全部兜底记到账号 Owner 名下(用户看到「100% 都是我」的 # 由来)。拿 ``ws-`` 的 id 去新加坡端点查 ``ListApiKeys`` 返回 **HTTP 200 + # 0 把**,不报错,所以这个缺口是**静默**的。 # # 注意:``ListApiKeys`` / ``CreateApiKey`` / ``DeleteApiKey`` 也必须打 # **该 workspace 所属区域**的端点,否则同样是「200 但看不见」——而 # ``sync_api_keys`` 的漂移清扫会据此把该区域的 Key 全部软删。 _MODELSTUDIO_REGIONS: tuple[str, ...] = ( "ap-southeast-1", "cn-beijing", "us-east-1", ) def modelstudio_host(region: str | None) -> str: """区域 → ModelStudio 端点。``None``/空 回落到默认(新加坡)。""" reg = (region or "").strip() or _DEFAULT_REGION return f"modelstudio.{reg}.aliyuncs.com" _MODELSTUDIO_VERSION = "2026-02-10" _RAM_VERSION = "2015-05-01" _BSS_VERSION = "2017-12-14" # Aliyun BSS ProductCode for the Bailian / Model Studio product. _BSS_PRODUCT_CODE_BAILIAN = "bailian" # The intl ``DescribeInstanceBill`` / ``QueryBillOverview`` ProductCode # for Model Studio is ``sfm`` (verified prod 2026-06). ``ProductName`` = # "Alibaba Cloud Model Studio". _BSS_PRODUCT_CODE_SFM = "sfm" _MAX_PAGINATION_PAGES = 1000 _MAX_WORKSPACE_ITEMS = 10_000 _MAX_API_KEY_ITEMS = 100_000 _MAX_BILLING_ITEMS_PER_DAY = 1_000_000 def _is_bailian_product(product_code: str, product_name: str) -> bool: """Match a BSS bill row to the Bailian / Model Studio product. We match on BOTH ``ProductCode`` and ``ProductName`` because Aliyun's code for this product on the intl ``QueryBillOverview`` is **``sfm``** (verified prod 2026-06) while older docs / other endpoints use ``bailian`` / ``dashscope``, and the human-readable name is "Alibaba Cloud Model Studio" / 百炼. Name match is the reliable signal; the code set is a belt-and-braces fallback.""" pc = (product_code or "").strip().lower() pn = (product_name or "").strip().lower() if pc in {"sfm", "bailian", "dashscope", "modelstudio", "model_studio"}: return True return any( kw in pn for kw in ("model studio", "bailian", "百炼", "dashscope", "灵积") ) def _to_usd_cny(native: float, currency: str, rate: float) -> tuple[float, float]: """Convert one BSS native amount to ``(usd, cny)`` honouring the per-row ``Currency`` field. ``rate`` is USD-per-CNY (≈0.145; 1 CNY = ``rate`` USD), the same ``ai_spending.cny_to_usd_rate`` used elsewhere. The intl BSS endpoint bills in **USD by default**; the old code did ``usd = native * rate`` treating every amount as CNY, dividing real USD bills ~7×. Mirror :meth:`AlicloudIntlCloudDriver._convert_pretax_ amount`: ``USD`` (or empty) → ``usd=native, cny=native/rate``; ``CNY`` → ``cny=native, usd=native*rate``. Other currencies treated as USD (never silently /rate — the bug class removed).""" if (currency or "USD").strip().upper() == "CNY": return native * rate, native return native, (native / rate if rate else 0.0) # AliyunBailianFullAccess grants both management-layer (workspace mgmt, # member mgmt) AND data-layer (model invoke) permissions. Same scope # the Bailian docs recommend for "I want a RAM user that can manage # everything in this workspace". _BAILIAN_FULL_ACCESS_POLICY = "AliyunBailianFullAccess" _BAILIAN_DATA_FULL_ACCESS_POLICY = "AliyunBailianDataFullAccess" def resolve_credentials(integration: Integration) -> dict[str, Any]: """Decrypt + merge an Aliyun cloud integration's config. Returns the merged dict of ``config_public`` + decrypted secrets, plus a ``_provider`` key carrying the integration's provider so the client can route to the right control-plane host (国内 vs 海外). Raises ``AppException`` if the integration isn't an Aliyun-family cloud integration or has no credentials configured (saves the caller from doing per-field None checks). """ if integration.kind != "cloud" or integration.provider not in _SUPPORTED_PROVIDERS: raise AppException( code="BAILIAN_WRONG_INTEGRATION_KIND", message=( f"Bailian integrations require kind=cloud and provider in " f"{sorted(_SUPPORTED_PROVIDERS)} (got kind={integration.kind!r}, " f"provider={integration.provider!r})." ), status_code=400, ) fields: dict[str, Any] = {} if integration.secrets_encrypted: fields = decrypt_fields( integration.secrets_encrypted, integration.secrets_nonce, integration_id=str(integration.id), ) merged = { k: str(v) for k, v in (integration.config_public or {}).items() if v is not None } merged.update({k: str(v) for k, v in fields.items() if v is not None}) # Stamp the provider so :class:`BailianClient` can pick the right # control-plane host without re-reading the Integration row. The # underscore prefix keeps this out of the credential namespace # (no Aliyun credential field starts with ``_``). merged["_provider"] = integration.provider if not merged.get("access_key_id") or not merged.get("access_key_secret"): raise AppException( code="BAILIAN_NO_CREDENTIALS", message=( "Aliyun integration is missing access_key_id / " "access_key_secret. Configure them under " "/admin/integrations and try again." ), status_code=503, ) return merged class BailianClient: """Stateless Bailian HTTP client, one per call. Caller passes the already-decrypted credentials dict so this class has no awareness of the encryption format. Helps testability — tests can construct a client with literal AK/SK in-process. """ def __init__(self, credentials: dict[str, Any]): self._ak = str(credentials.get("access_key_id") or "").strip() self._sk = str(credentials.get("access_key_secret") or "").strip() # Bailian only supports the 海外 (alicloud_intl) tenant. The # provider field is plumbed through for consistency with other # callers; if a future 国内 Bailian rollout happens, add the # mapping entries above and this code keeps working. self._provider = str( credentials.get("_provider") or _DEFAULT_PROVIDER, ).strip() or _DEFAULT_PROVIDER default_region = _DEFAULT_REGION_BY_PROVIDER.get( self._provider, _DEFAULT_REGION, ) # Bailian region resolution must NOT fall back to the integration's # generic ``region`` field. That field is set to where the AliCloud # tenant's Cloud SSO directory lives (typically ``cn-shanghai`` even # for 海外 tenants, see :class:`AlicloudIntlCloudDriver._sso_directory_id`) # — but Bailian's Singapore control plane is at ``ap-southeast-1``, # and ``bailian.cn-shanghai.aliyuncs.com`` simply doesn't exist # (404). Same trap applies to 国内 setups where ``region=cn-beijing`` # for ECS doesn't mean Bailian is in cn-beijing (only cn-hangzhou # is hosted). Use the explicit ``bailian_region`` override or the # provider default — never the generic field. self._region = str( credentials.get("bailian_region") or default_region, ).strip() or default_region self._bss_host = _BSS_HOST_BY_PROVIDER.get( self._provider, _BSS_HOST_BY_PROVIDER[_DEFAULT_PROVIDER], ) if not self._ak or not self._sk: raise AppException( code="BAILIAN_NO_CREDENTIALS", message="Aliyun access key/secret not configured.", status_code=503, ) # ── ACS3 signer (HMAC-SHA256) ───────────────────────────────── async def _acs3_call( self, *, host: str, version: str, action: str, method: str = "POST", params: dict[str, Any] | None = None, body: dict[str, Any] | None = None, path: str = "/", ) -> dict[str, Any]: now_utc = _dt.datetime.now(tz=_dt.UTC) iso_time = now_utc.strftime("%Y-%m-%dT%H:%M:%SZ") nonce = uuid.uuid4().hex query = {k: str(v) for k, v in (params or {}).items()} body_bytes = ( json.dumps(body, separators=(",", ":")).encode() if body is not None else b"" ) content_sha = hashlib.sha256(body_bytes).hexdigest() canonical_query = "&".join( f"{urllib.parse.quote(k, safe='-_.~')}=" f"{urllib.parse.quote(v, safe='-_.~')}" for k, v in sorted(query.items()) ) headers = { "host": host, "x-acs-action": action, "x-acs-content-sha256": content_sha, "x-acs-date": iso_time, "x-acs-signature-nonce": nonce, "x-acs-version": version, } if body_bytes: headers["content-type"] = "application/json; charset=utf-8" canonical_headers = "".join( f"{k}:{v}\n" for k, v in sorted(headers.items()) ) signed_headers = ";".join(sorted(headers.keys())) canonical_request = ( f"{method}\n{path}\n{canonical_query}\n{canonical_headers}\n" f"{signed_headers}\n{content_sha}" ) # ACS3 requires an unkeyed SHA-256 canonical-request digest; the # secret is applied only in the HMAC-SHA-256 signature below. hashed_canonical = hashlib.sha256( # codeql[py/weak-sensitive-data-hashing] canonical_request.encode(), ).hexdigest() string_to_sign = f"ACS3-HMAC-SHA256\n{hashed_canonical}" signature = hmac.new( self._sk.encode(), string_to_sign.encode(), hashlib.sha256, ).hexdigest() auth = ( f"ACS3-HMAC-SHA256 Credential={self._ak}," f"SignedHeaders={signed_headers},Signature={signature}" ) headers["authorization"] = auth url = ( f"https://{host}{path}?{canonical_query}" if canonical_query else f"https://{host}{path}" ) # Pre-launch P2 fix (Round 1, SEC-005): trust_env=False to # match platform-wide outbound HTTP discipline. async with httpx.AsyncClient(timeout=30, trust_env=False) as client: resp = await client.request( method, url, headers={k: str(v) for k, v in headers.items()}, content=body_bytes or None, ) if resp.status_code >= 400: raise AppException( code="ALIYUN_API_ERROR", message=( f"Aliyun {action} {resp.status_code}: " f"{resp.text[:400]}" ), status_code=502, ) try: return resp.json() except json.JSONDecodeError: return {"raw": resp.text} # ── Bailian workspace + application discovery ──────────────── async def list_workspaces( self, *, region: str | None = None, ) -> list[dict[str, Any]]: """Enumerate ModelStudio workspaces in ONE region. ``region=None`` 打默认(新加坡)端点。**每个区域只返回它自己的 那个「默认业务空间」** —— 要拿全account 的空间必须逐区域调用, 见 ``_MODELSTUDIO_REGIONS`` 上方注释。 Calls ``ModelStudio/2026-02-10`` ``ListWorkspaces`` (ROA style, ``GET /modelstudio/workspaces``) on ``modelstudio.ap-southeast-1.aliyuncs.com``. Cursor pagination via ``maxResults`` (10–100, default 10) + ``nextToken`` — note the lower-case parameter names, which the intl product spec uses (the CN bailian product uses Pascal-case). """ out: list[dict[str, Any]] = [] next_token: str | None = None guard = PaginationGuard( integration="bailian", action="list_workspaces", max_pages=_MAX_PAGINATION_PAGES, max_items=_MAX_WORKSPACE_ITEMS, ) while True: params: dict[str, Any] = {"maxResults": 100} if next_token: params["nextToken"] = next_token data = await self._acs3_call( host=modelstudio_host(region or self._region), version=_MODELSTUDIO_VERSION, action="ListWorkspaces", method="GET", path="/modelstudio/workspaces", params=params, ) if not isinstance(data, dict): guard.reject_malformed_response() page_items = data.get("workspaces") or data.get("Workspaces") or [] if ( not isinstance(page_items, list) or any(not isinstance(item, dict) for item in page_items) ): guard.reject_malformed_response() guard.record_page(len(page_items)) for ws in page_items: out.append({ "workspace_id": str( ws.get("workspaceId") or ws.get("WorkspaceId") or "", ), "name": str( ws.get("workspaceName") or ws.get("WorkspaceName") or "", ), "description": ws.get("description") or ws.get("Description") or None, # 端点区域是权威值:响应里的 ``region`` 字段在实测中 # 可能缺失,而「这个空间属于哪个区域」决定了后续 # ListApiKeys / Create / Delete 该打哪个端点。 "region": ( region or ws.get("region") or ws.get("Region") or self._region ), "is_default": bool( ws.get("isDefault") or ws.get("IsDefault"), ), "raw": ws, }) next_token = data.get("nextToken") if next_token is None: next_token = data.get("NextToken") if next_token is not None and not isinstance(next_token, str): guard.reject_malformed_response() if not guard.require_safe_next(next_token): break return out async def create_api_key( self, *, workspace_id: str, description: str | None = None, region: str | None = None, ) -> dict[str, Any]: """Mint a brand-new Bailian (DashScope) API key under a workspace. ``region`` 必须是该 workspace 所属区域(见 ``_MODELSTUDIO_REGIONS``);打错区域会在错误的区域建 Key。 Calls ``ModelStudio/2026-02-10`` ``CreateApiKey`` (ROA style, ``POST /modelstudio/apikeys``). The returned ``apiKeyValue`` is the *plaintext* sk-... — it is only exposed once at creation time; subsequent ``ListApiKeys`` masks it (``sk-X****YYYY``). Callers MUST persist the value immediately if they want to hand it back to an employee. The ``description`` field is the only human-readable handle the Bailian console / ListApiKeys surface to operators, so callers should encode the Helix assignment / owner / RAM user there for downstream auditability (Aliyun caps it at ~256 chars; we trim defensively). Idempotency: Aliyun's CreateApiKey is NOT idempotent — repeated calls mint distinct sk-... values. The autogrant flow guards re-entry via ``detail.bailian.granted`` / ``in_flight`` so this method does not need its own dedup. """ body: dict[str, Any] = {"workspaceId": workspace_id} if description: body["description"] = description[:256] data = await self._acs3_call( host=modelstudio_host(region or self._region), version=_MODELSTUDIO_VERSION, action="CreateApiKey", method="POST", path="/modelstudio/apikeys", body=body, ) # Aliyun ModelStudio wraps the created key under ``apiKey`` on # the intl tenant; tolerate ``ApiKey`` for forward-compat in # case a future version flips the casing. block = data.get("apiKey") or data.get("ApiKey") or {} api_key_id = block.get("apiKeyId") or block.get("ApiKeyId") api_key_value = block.get("apiKeyValue") or block.get("ApiKeyValue") if not api_key_id or not api_key_value: # NEVER interpolate the raw response — if Aliyun returns a # populated apiKeyValue (live ``sk-...``) but a missing apiKeyId # (shape drift / partial response), ``{data!r}`` would leak the # plaintext secret into the error envelope AND the logs. Report # only the safe presence flags. raise AppException( code="BAILIAN_CREATE_APIKEY_FAILED", message=( f"ModelStudio CreateApiKey returned an incomplete apiKey " f"block (workspace={workspace_id!r}, " f"has_id={bool(api_key_id)}, has_value={bool(api_key_value)})." ), status_code=502, ) created_ms = block.get("gmtCreate") or block.get("GmtCreate") return { "api_key_id": str(api_key_id), "api_key_value": str(api_key_value), "workspace_id": str( block.get("workspaceId") or block.get("WorkspaceId") or workspace_id, ), "created_by": ( str(block.get("createdBy") or block.get("CreatedBy") or "") or None ), "disabled": bool(block.get("disabled") or block.get("Disabled")), "created_at_ms": int(created_ms) if created_ms else None, "description": ( block.get("description") or block.get("Description") or None ), "raw": block, } async def delete_api_key( self, *, api_key_id: str, workspace_id: str | None = None, region: str | None = None, ) -> dict[str, Any]: """Permanently invalidate a Bailian API key (sk-...). Calls ``ModelStudio/2026-02-10`` ``DeleteApiKey`` (ROA style, ``DELETE /modelstudio/apikeys/{apiKeyId}``). Idempotent on 404 / EntityNotExist so the revoke retry path doesn't bomb on a key already removed manually in the Bailian console. ``workspace_id`` is optional in Aliyun's spec but supplying it lets the server scope the lookup; we always pass it when known. """ path = ( f"/modelstudio/apikeys/" f"{urllib.parse.quote(api_key_id, safe='')}" ) params: dict[str, Any] | None = ( {"workspaceId": workspace_id} if workspace_id else None ) try: data = await self._acs3_call( host=modelstudio_host(region or self._region), version=_MODELSTUDIO_VERSION, action="DeleteApiKey", method="DELETE", path=path, params=params, ) return {"deleted": True, "already_absent": False, "raw": data} except AppException as exc: msg = exc.message or "" if ( "NotFound" in msg or "NotExist" in msg or " 404" in msg or "ApiKey.NotFound" in msg ): return {"deleted": True, "already_absent": True} raise async def list_api_keys( self, *, workspace_id: str, region: str | None = None, ) -> list[dict[str, Any]]: """Enumerate Bailian API keys (sk-xxx) under a single workspace. ``region`` 必须是**该 workspace 所属区域**:打错区域会返回 HTTP 200 + 空列表(不报错),而调用方的漂移清扫会据此把该区域 的 Key 全部软删。 Calls ``ModelStudio/2026-02-10`` ``ListApiKeys`` (ROA style, ``GET /modelstudio/apikeys``) — note the unhyphenated / all-lowercase path; the camelCase variants Aliyun uses for workspaces are NOT used here (verified empirically against prod intl tenant 2026-04-29). Returns the masked key value + ``apiKeyId`` numeric handle so the admin UI can show "this key still exists" without holding the secret. Unsupported on 国内 Bailian — only intl exposes ListApiKeys. """ out: list[dict[str, Any]] = [] next_token: str | None = None guard = PaginationGuard( integration="bailian", action="list_api_keys", max_pages=_MAX_PAGINATION_PAGES, max_items=_MAX_API_KEY_ITEMS, ) while True: params: dict[str, Any] = { "workspaceId": workspace_id, "maxResults": 100, } if next_token: params["nextToken"] = next_token data = await self._acs3_call( host=modelstudio_host(region or self._region), version=_MODELSTUDIO_VERSION, action="ListApiKeys", method="GET", path="/modelstudio/apikeys", params=params, ) if not isinstance(data, dict): guard.reject_malformed_response() page_items = data.get("apiKeys") or [] if ( not isinstance(page_items, list) or any(not isinstance(item, dict) for item in page_items) ): guard.reject_malformed_response() guard.record_page(len(page_items)) for k in page_items: api_key_id = k.get("apiKeyId") if api_key_id is None: continue # ``apiKeyValue`` is masked by Aliyun (sk-X****YYYY) — # treat it as a public display-name fragment, not a # secret. The full sk-... only ever leaves the Bailian # console at creation time. masked = str(k.get("apiKeyValue") or "") description = str(k.get("description") or "").strip() display_name = description or masked or f"apikey-{api_key_id}" created_ms = k.get("gmtCreate") out.append({ "api_key_id": str(api_key_id), "workspace_id": str( k.get("workspaceId") or workspace_id, ), "display_name": display_name, "masked_key": masked, "created_by": str(k.get("createdBy") or "") or None, "disabled": bool(k.get("disabled")), "created_at_ms": int(created_ms) if created_ms else None, "raw": k, }) next_token = data.get("nextToken") if next_token is not None and not isinstance(next_token, str): guard.reject_malformed_response() if not guard.require_safe_next(next_token): break return out # ── Bailian usage / billing ────────────────────────────────── async def query_instance_bill_daily( self, *, start_date: str, end_date: str, cny_to_usd_rate: float, ) -> list[dict[str, Any]]: """Query Bailian (Model Studio, ProductCode ``sfm``) cost at **per-(day × API key × model)** granularity via BSS ``DescribeInstanceBill`` (``Granularity=DAILY``). **Why DescribeInstanceBill** (verified against prod 2026-06-17): the instance-bill ``InstanceID`` for ``sfm`` is a composite ``apiKeyId;workspaceId;model;tokenType;channel;0`` — e.g. ``132245;ws-8a6e…;qwen3.6-plus;context_0-256k_input_token;;0``. The first field is the **global ModelStudio API-key id** (it matches ``bailian_api_keys.api_key_id``), so this is the source that finally yields real per-key / per-model chargeback — the long-standing "BSS only exposes account-level Bailian spend" assumption was wrong. It reconciles to the official bill to the cent (2026-05 / 2026-04 net matched ``QueryBillOverview`` exactly; 2026-06 within a sub-cent). The earlier "DescribeInstanceBill only sees ~$354" conclusion was a wrong-params artefact — under consolidated billing the master account's instance bill sees every Resource-Directory member's ``sfm`` usage. Only consumption rows (``PretaxGrossAmount`` > 0) are summed so the figure equals the console "消费" line; adjustment / pure- credit rows (gross ≤ 0, often negative net) are excluded. Amounts are currency-aware via :func:`_to_usd_cny` (intl BSS bills in USD). ``OwnerID`` (the spending member account) lands as ``aliyun_account_id``; ``BillAccountName`` as the account label. Returns one dict per ``(usage_date, aliyun_account_id, bailian_api_key_id, workspace_id, model_name)`` with three price tiers (原价 / 优惠后 / 应付) in USD + CNY. Args: start_date / end_date: ISO ``YYYY-MM-DD``. Every day in the inclusive range is queried (one ``DescribeInstanceBill`` DAILY call per day, paginated). cny_to_usd_rate: USD-per-CNY rate (≈0.145) for the derived non-native currency column. """ try: s = _dt.date.fromisoformat(start_date) e = _dt.date.fromisoformat(end_date) except ValueError as exc: raise AppException( code="BAILIAN_BAD_DATE_RANGE", message=( f"start_date / end_date must be ISO YYYY-MM-DD; " f"got start_date={start_date!r}, end_date={end_date!r}" ), status_code=400, ) from exc if s > e: raise AppException( code="BAILIAN_BAD_DATE_RANGE", message=( f"start_date {start_date} must be on or before " f"end_date {end_date}." ), status_code=400, ) rate = float(cny_to_usd_rate) if cny_to_usd_rate else 0.0 # Accumulate across (token-type × channel) into one row per # (date, account, api_key, workspace, model) — the chargeback # grain. The bill splits each model by token-type and channel; # those are detail we fold up (channels retained in ``raw``). agg: dict[tuple[str, str | None, str | None, str | None, str], dict[str, Any]] = {} cur = s while cur <= e: cycle = cur.strftime("%Y-%m") billing_date = cur.isoformat() next_token: str | None = None guard = PaginationGuard( integration="bailian", action="query_instance_bill_daily", max_pages=_MAX_PAGINATION_PAGES, max_items=_MAX_BILLING_ITEMS_PER_DAY, ) while True: params: dict[str, Any] = { "BillingCycle": cycle, "BillingDate": billing_date, "Granularity": "DAILY", "ProductCode": _BSS_PRODUCT_CODE_SFM, "MaxResults": 300, } if next_token: params["NextToken"] = next_token data = await self._acs3_call( host=self._bss_host, version=_BSS_VERSION, action="DescribeInstanceBill", params=params, ) if not isinstance(data, dict): guard.reject_malformed_response() payload = data.get("Data") or {} if not isinstance(payload, dict): guard.reject_malformed_response() items = payload.get("Items") items = (items.get("Item") if isinstance(items, dict) else items) or [] if ( not isinstance(items, list) or any(not isinstance(item, dict) for item in items) ): guard.reject_malformed_response() guard.record_page(len(items)) for it in items: if not _is_bailian_product( it.get("ProductCode"), it.get("ProductName"), ): continue g = float(it.get("PretaxGrossAmount") or 0) if g <= 0: continue # adjustment / pure-credit row — skip d = float(it.get("InvoiceDiscount") or 0) n = float(it.get("PretaxAmount") or 0) cur_it = str(it.get("Currency") or "USD") # InstanceID = apiKeyId;workspaceId;model;tokenType;channel;0 parts = str(it.get("InstanceID") or "").split(";") api_key_id = (parts[0].strip() if len(parts) > 0 else "") or None workspace_id = (parts[1].strip() if len(parts) > 1 else "") or None model = (parts[2].strip() if len(parts) > 2 else "") or "unknown" channel = (parts[4].strip() if len(parts) > 4 else "") or None account_id = (str(it.get("OwnerID") or "").strip() or None) account_name = it.get("BillAccountName") or None key = (billing_date, account_id, api_key_id, workspace_id, model) slot = agg.get(key) if slot is None: slot = { "gross": 0.0, "disc": 0.0, "net": 0.0, "currency": cur_it, "account_name": account_name, "channels": set(), } agg[key] = slot elif slot["currency"] != cur_it: # Native amounts are summed before a single currency # conversion at the end — a slot that mixes currencies # would convert the total with the first-seen currency # and silently misstate spend. Fail loudly (mirrors the # cost_facts pre-aggregation mixed-currency guard). raise RuntimeError( "bailian DescribeInstanceBill: slot " f"{key} spans currencies {slot['currency']} vs " f"{cur_it}; cannot sum natively before conversion." ) slot["gross"] += g slot["disc"] += d slot["net"] += n if channel: slot["channels"].add(channel) next_token = payload.get("NextToken") if next_token is not None and not isinstance(next_token, str): guard.reject_malformed_response() if not guard.require_safe_next(next_token): break cur = cur + _dt.timedelta(days=1) out: list[dict[str, Any]] = [] for (billing_date, account_id, api_key_id, workspace_id, model), slot in agg.items(): currency = slot["currency"] gross_n = slot["gross"] after_disc_n = gross_n - slot["disc"] net_n = slot["net"] usd_g, cny_g = _to_usd_cny(gross_n, currency, rate) usd_ad, cny_ad = _to_usd_cny(after_disc_n, currency, rate) usd, cny = _to_usd_cny(net_n, currency, rate) channels = sorted(slot["channels"]) out.append({ "usage_date": billing_date, "aliyun_account_id": account_id, "aliyun_account_name": slot["account_name"], "bailian_api_key_id": api_key_id, "workspace_id": workspace_id, "model_name": model, "channels": channels, "input_tokens": 0, "output_tokens": 0, "request_count": 0, "cost_cny": cny, "cost_usd": usd, "cost_cny_gross": cny_g, "cost_usd_gross": usd_g, "cost_cny_after_discount": cny_ad, "cost_usd_after_discount": usd_ad, "raw": {"currency": currency, "channels": channels}, }) return out # ── RAM user lifecycle (Bailian-tagged) ───────────────────── async def ensure_ram_user( self, *, user_name: str, display_name: str, email: str | None, comments: str = "Bailian auto-grant — managed by Helix", ) -> dict[str, Any]: """Idempotently create a RAM user. Returns the user dict on success (whether newly-created or pre-existing). Idempotent on ``EntityAlreadyExists.User``. """ try: data = await self._acs3_call( host="ram.aliyuncs.com", version=_RAM_VERSION, action="CreateUser", params={ "UserName": user_name, "DisplayName": display_name or user_name, "Email": email or "", "Comments": comments, }, ) return {"user": data.get("User") or {}, "created": True} except AppException as exc: if "EntityAlreadyExists.User" not in (exc.message or ""): raise # Already exists — fetch it for the return value. existing = await self._acs3_call( host="ram.aliyuncs.com", version=_RAM_VERSION, action="GetUser", params={"UserName": user_name}, ) return {"user": existing.get("User") or {}, "created": False} async def attach_policy( self, *, user_name: str, policy_name: str = _BAILIAN_FULL_ACCESS_POLICY, policy_type: str = "System", ) -> dict[str, Any]: """Attach a RAM policy to a user. Idempotent on duplicate.""" try: await self._acs3_call( host="ram.aliyuncs.com", version=_RAM_VERSION, action="AttachPolicyToUser", params={ "PolicyType": policy_type, "PolicyName": policy_name, "UserName": user_name, }, ) return {"already_attached": False} except AppException as exc: if "EntityAlreadyExists" in (exc.message or ""): return {"already_attached": True} raise async def create_access_key( self, *, user_name: str, ) -> dict[str, Any]: """Mint a fresh AccessKey/Secret pair for a RAM user.""" data = await self._acs3_call( host="ram.aliyuncs.com", version=_RAM_VERSION, action="CreateAccessKey", params={"UserName": user_name}, ) ak_block = (data.get("AccessKey") or {}) ak_id = ak_block.get("AccessKeyId") or "" ak_sec = ak_block.get("AccessKeySecret") or "" if not ak_id or not ak_sec: raise AppException( code="RAM_AK_MINT_FAILED", message=( "Aliyun CreateAccessKey returned an empty access " f"key pair for {user_name!r}." ), status_code=502, ) return { "access_key_id": ak_id, "access_key_secret": ak_sec, "raw": ak_block, } # International ModelStudio has no ``AddWorkspaceMembers`` action; # auto-grant relies on the ``AliyunBailianFullAccess`` RAM policy # for tenant-wide invoke and skips per-workspace membership # entirely. The CN ``bailian/2023-12-29`` product had it; if Helix # ever supports CN Bailian, restore this method behind the same # provider routing as the rest of the client. def dashscope_endpoint_for_provider(provider: str | None) -> str: """Resolve provider → DashScope (RAM-mode) endpoint URL. Only ``alicloud_intl`` is supported today; the helper still takes a provider argument for forward-compat. The auto-grant flow embeds this URL in the credential bundle so employees can copy-paste a working DashScope SDK config. """ return _DASHSCOPE_ENDPOINT_BY_PROVIDER.get( (provider or _DEFAULT_PROVIDER), _DASHSCOPE_ENDPOINT_BY_PROVIDER[_DEFAULT_PROVIDER], ) def default_region_for_provider(provider: str | None) -> str: return _DEFAULT_REGION_BY_PROVIDER.get( (provider or _DEFAULT_PROVIDER), _DEFAULT_REGION, ) __all__ = [ "BailianClient", "resolve_credentials", "dashscope_endpoint_for_provider", "default_region_for_provider", "_BAILIAN_FULL_ACCESS_POLICY", "_BAILIAN_DATA_FULL_ACCESS_POLICY", "_SUPPORTED_PROVIDERS", ]