"""Admin Cloud Overview — read-only aggregations for the multi-tab fleet dashboard at ``/admin/cloud/overview``. Distinct from :func:`CmdbService.get_analytics` (which feeds both admin and employee analytics dashboards): this module is admin-only and focused on subscriptions / grants / recovered-permissions / governance signals — the kind of data an IT admin needs to answer "how many subscriptions, how many active grants, how many recovered permissions" in one place. Every aggregation respects ``integration_id`` filtering when present. A single transactional read snapshot — no DB writes, no cloud-side calls. Numbers come from in-DB rows that the existing sync / lifecycle workflows maintain. Security posture ---------------- * Every entry point is gated by :func:`can_manage_cloud` (admin-only) in the router. This module assumes the caller has been authorized. * All filters compose AND, never OR — narrowing scope to a tenant never accidentally widens results. * No raw user input is interpolated into LIKE patterns; the ``integration_id`` parameter is parsed through ``uuid.UUID`` at the router boundary. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.cost_view import fact_engineering_amount from app.domains.cmdb.models import ( GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, GRANT_STATUS_ACTIVE, GRANT_STATUS_DISABLED, GRANT_STATUS_EXPIRED, GRANT_STATUS_PENDING, GRANT_STATUS_REVOKED, GRANT_STATUSES_RECLAIMED, PROD_ENVIRONMENTS, SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_FEISHU_SENT, SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_PROVISIONING, SCOPE_STATUS_ACTIVE, SCOPE_STATUS_CLOSED, SCOPE_STATUS_SUSPENDED, SCOPE_TYPE_GCP_BILLING_ACCOUNT, CmdbCloudAccessGrant, CmdbCloudScope, CmdbCloudScopeRequest, CmdbCostFact, CmdbIamUser, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson from app.integrations.models import Integration async def resolve_department_person_ids( db: AsyncSession, department: str | None, ) -> list[str] | None: """Resolve a department name to the ``identity_persons.id`` set whose *primary* org membership sits in ``department``'s subtree (inclusive). Returns ``None`` when no department filter is requested (caller skips the filter entirely) and a ``list[str]`` otherwise — possibly empty when the department has no members, which correctly narrows every owner/holder aggregation to zero rows via ``IN ()``. Subtree match via :func:`app.core.org_scope.department_membership_subtree_clause` — picking a parent node in ``DepartmentTreeSelect`` rolls every descendant up, identical to the Users-list and software-cost filters. The cost side (``cmdb_cost_facts.cost_owner_department``) is narrowed with the matching leaf-name set so the headcount filter and the cost filter agree on what a parent department contains. IDs are returned as ``str`` to match the ``String(36)`` ``owner_person_id`` / ``person_id`` FK columns on the cmdb tables (binding a ``uuid.UUID`` against those columns silently matches zero rows under asyncmy). """ if not department: return None from app.core.org_scope import department_membership_subtree_clause rows = (await db.execute( select(IdentityOrgMembership.person_id) .where( department_membership_subtree_clause(department), IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) )).scalars().all() return [str(pid) for pid in rows] def _utc_now_naive() -> datetime: """MySQL/TiDB returns naive datetimes for ``DateTime(timezone=True)`` columns. Use a naive UTC ``now`` so arithmetic against them does not raise ``can't subtract offset-naive and offset-aware datetimes``. Mirrors the helper in :mod:`lifecycle_router`. """ return datetime.now(UTC).replace(tzinfo=None) def _kv_rows(rows: list[Any]) -> list[dict[str, Any]]: """Turn ``[(name, count), ...]`` into the standard ``[{name, count}]`` JSON shape the frontend dashboards expect. Coerces NULLs to ``"unknown"`` so downstream code never has to defend against a JSON ``null`` key. """ return [ {"name": (k if k is not None else "unknown"), "count": int(c or 0)} for k, c in rows ] async def get_admin_overview( db: AsyncSession, *, integration_id: str | None = None, month: str | None = None, department: str | None = None, ) -> dict[str, Any]: """Build the admin-only Cloud Overview payload. Returns a JSON-shaped dict with five top-level sections: * ``filter`` — echoes the active ``integration_id`` + ``month``. * ``subscriptions`` — scope distributions, recent decommissions, AND month-anchored per-scope cost ranking (top 20 by spend + fleet total). Drives the Subscriptions tab's resource ranking + the click-through into the per-scope billing-detail drawer. * ``permissions`` — grant counts, expiring buckets, recovered history (revoked / expired / orphan-swept), and revocation reasons. * ``iam_users`` — provider-side IAM users discovered by the IAM sync, owner-mapping health. * ``alerts`` — fleet-level governance signals (unowned scopes, prod-overdue grants, expired-active grants). When ``integration_id`` is set, every section is restricted to that cloud tenant. When ``None``, results span all cloud integrations. When ``department`` is set, every owner/holder-attributable section is additionally narrowed to resources whose owner (scopes / IAM users) or holder (grants) sits in that department, and the per-scope cost ranking is narrowed to that department's ``cost_owner_department`` facts. Filters compose AND with ``integration_id``. Tenant-level metrics with no owner (pending scope requests) are intentionally left department-agnostic. ``department`` matches the whole subtree under the chosen node (a parent node rolls up every descendant) on the person's primary org membership — same semantic as the platform-wide ``DepartmentTreeSelect`` filter (unified in #709). ``month`` (YYYY-MM) selects the billing month for the cost ranking. Defaults to the current UTC month (clamped to today for partial months). Past months show the settled total. Same parse / clamp rules as the per-scope billing-detail endpoint, so the admin tab and the drawer always agree on the month window. """ # Resolve the department filter once. ``None`` → no department filter. # An empty list (department with no members) correctly narrows every # owner/holder aggregation to zero via ``IN ()``. dept_pids = await resolve_department_person_ids(db, department) integ_filter_scope = ( [CmdbCloudScope.integration_id == integration_id] if integration_id else [] ) integ_filter_iam = ( [CmdbIamUser.integration_id == integration_id] if integration_id else [] ) integ_filter_request = ( [CmdbCloudScopeRequest.integration_id == integration_id] if integration_id else [] ) if dept_pids is not None: integ_filter_scope.append( CmdbCloudScope.owner_person_id.in_(dept_pids), ) integ_filter_iam.append( CmdbIamUser.owner_person_id.in_(dept_pids), ) return { "filter": { "integration_id": integration_id, "month": month, "department": department, }, "subscriptions": await _subscriptions_section( db, integ_filter_scope, integ_filter_request, integration_id, month=month, dept_pids=dept_pids, ), "permissions": await _permissions_section( db, integration_id, dept_pids=dept_pids, ), "iam_users": await _iam_users_section(db, integ_filter_iam), "alerts": await _alerts_section( db, integration_id, dept_pids=dept_pids, ), } # ── Subscriptions ─────────────────────────────────────────────────── async def _subscriptions_section( db: AsyncSession, integ_filter_scope: list, integ_filter_request: list, integration_id: str | None, *, month: str | None = None, dept_pids: list[str] | None = None, ) -> dict[str, Any]: """Scope (子帐号/订阅) distribution + decommission history + cost ranking. ``integ_filter_scope`` already carries the department owner filter (appended by the caller) so every scope aggregation below inherits it; the cost ranking is narrowed by the SAME live ``dept_pids`` set (scopes whose current owner sits in the department), so the count, the list and the ranking all agree on which subscriptions belong to the department — even immediately after an owner transfer, before the nightly allocation recompute re-stamps ``cost_owner_department``. """ base = and_( CmdbCloudScope.deleted_at.is_(None), *integ_filter_scope, ) # Totals — split by status so the UI can render "active / closed / # suspended" without a second roundtrip. by_status_rows = (await db.execute( select(CmdbCloudScope.status, func.count()) .where(base) .group_by(CmdbCloudScope.status) )).all() by_status_map = {str(s): int(c or 0) for s, c in by_status_rows} total_active = by_status_map.get(SCOPE_STATUS_ACTIVE, 0) total_closed = by_status_map.get(SCOPE_STATUS_CLOSED, 0) total_suspended = by_status_map.get(SCOPE_STATUS_SUSPENDED, 0) total_all = sum(by_status_map.values()) # Owner health — only meaningful for non-closed scopes (closed # scopes are intentionally orphaned by the decommission flow). live_filter = and_( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status != SCOPE_STATUS_CLOSED, *integ_filter_scope, ) owned = (await db.execute( select(func.count()).select_from(CmdbCloudScope).where( live_filter, CmdbCloudScope.owner_person_id.isnot(None), ) )).scalar() or 0 # GCP billing-account scopes are intentionally ownerless (approval # routes to the cloud-admin team, not a per-scope owner), so they must # NOT inflate the "unowned scopes" governance count / alert — otherwise # the alert stays permanently lit and drives admins toward # assign_scope_owner, which is a no-op for a billing account. Mirrors # the exclusion in the auto-owner-grant backfill cron + the # reconciliation owner-orphan audit. unowned = (await db.execute( select(func.count()).select_from(CmdbCloudScope).where( live_filter, CmdbCloudScope.owner_person_id.is_(None), CmdbCloudScope.scope_type != SCOPE_TYPE_GCP_BILLING_ACCOUNT, # Aliyun RD master (payer) is owner-exempt infra — see # reconciliation owner-orphan audit. Don't count it unowned. func.json_extract( CmdbCloudScope.raw_data, "$.rd_master", ).is_(None), ) )).scalar() or 0 # Owner-health signals — scope's owner exists but is no longer # ``active``. Split into two distinct buckets so admins can act on # the right signal: # # * ``owner_in_offboarding`` — owner is ``leaver_pending``. The # offboarding flow is in-flight; the T+0+24h force-transfer # will rewrite the FK to the manager fallback automatically. # Expected to clear within ~24-48h. Yellow chip; no IT action # unless the count stays high for >72h. # * ``owner_left`` — owner is anything other than ``active`` or # ``leaver_pending`` (typically ``left``, or future statuses). # Force-transfer SHOULD have already moved the FK. If we see # rows here, the safety net FAILED — either the worker is # dead, the fallback chain has no usable target, or the # scanner missed the row. Red chip; needs admin action. # # Pre-2026-05-12 these two buckets were collapsed into ``owner_left_org`` # with the filter ``employment_status != "active"``. The conflation # was the root surface of the "huohao 4 个云订阅卡 21 天" report: # admins couldn't tell that "已离职" was actually "in-flight, waiting # on a dead scheduler" vs "truly past offboarding, safety net broke". in_offboarding_rows = await db.execute( select(func.count()) .select_from(CmdbCloudScope) .join( IdentityPerson, IdentityPerson.id == CmdbCloudScope.owner_person_id, ) .where( live_filter, CmdbCloudScope.owner_person_id.isnot(None), IdentityPerson.employment_status == "leaver_pending", ) ) owner_in_offboarding = in_offboarding_rows.scalar() or 0 # Explicit allow-list of statuses NOT counted as "transfer failed": # * ``active`` — the owner is fine, no signal. # * ``leaver_pending`` — counted in the in_offboarding bucket above. # * ``pre_hire`` — legitimate cloud-scope owner (hire starts next # week, scope pre-provisioned for them); landing pre_hire in the # red ``owner_left`` chip would falsely accuse the transfer # pipeline of failing. # Anything else (today only ``left``; any future enum value too) # lands in ``owner_left`` — fail-loud by construction. Pre-this-PR # the filter used double `!=` exclusions, which would silently bucket # a freshly introduced enum into red without surfacing it; the # ``not_in`` form keeps the contract crisp. left_rows = await db.execute( select(func.count()) .select_from(CmdbCloudScope) .join( IdentityPerson, IdentityPerson.id == CmdbCloudScope.owner_person_id, ) .where( live_filter, CmdbCloudScope.owner_person_id.isnot(None), IdentityPerson.employment_status.notin_( ("active", "leaver_pending", "pre_hire"), ), ) ) owner_left = left_rows.scalar() or 0 # By environment (dev/staging/prod) — live scopes only. by_env_rows = (await db.execute( select(CmdbCloudScope.environment, func.count()) .where(live_filter) .group_by(CmdbCloudScope.environment) )).all() # By scope_type (aws_account / azure_subscription / gcp_project / # aliyun_account / standalone_iam). by_type_rows = (await db.execute( select(CmdbCloudScope.scope_type, func.count()) .where(live_filter) .group_by(CmdbCloudScope.scope_type) )).all() # By provider — provider lives on the parent Integration row. by_provider_rows = (await db.execute( select(Integration.provider, func.count()) .select_from(CmdbCloudScope) .join(Integration, Integration.id == CmdbCloudScope.integration_id) .where(live_filter) .group_by(Integration.provider) )).all() # By integration (top-N tenants by scope count). by_integration_rows = (await db.execute( select( Integration.id, Integration.name, Integration.provider, func.count(CmdbCloudScope.id), ) .select_from(CmdbCloudScope) .join(Integration, Integration.id == CmdbCloudScope.integration_id) .where(live_filter) .group_by(Integration.id, Integration.name, Integration.provider) .order_by(func.count(CmdbCloudScope.id).desc()) .limit(20) )).all() # Recent decommissions — last 90 days, top 30. Matches the # ``status='closed'`` history path. cutoff = _utc_now_naive() - timedelta(days=90) recent_decom_rows = (await db.execute( select( CmdbCloudScope.id, CmdbCloudScope.display_name, CmdbCloudScope.external_id, Integration.name.label("integration_name"), Integration.provider, CmdbCloudScope.decommissioned_at, CmdbCloudScope.decommission_reason, CmdbCloudScope.decommission_provider_state, CmdbCloudScope.decommissioned_by, ) .select_from(CmdbCloudScope) .join(Integration, Integration.id == CmdbCloudScope.integration_id) .where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_CLOSED, CmdbCloudScope.decommissioned_at.isnot(None), CmdbCloudScope.decommissioned_at >= cutoff, *integ_filter_scope, ) .order_by(CmdbCloudScope.decommissioned_at.desc()) .limit(30) )).all() # Resolve actor email/display once per UUID — the same admin # frequently runs many decommissions in a sweep. actor_ids = { r.decommissioned_by for r in recent_decom_rows if r.decommissioned_by } actor_lookup: dict[str, dict[str, str]] = {} if actor_ids: actor_rows = (await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ) .where(IdentityPerson.id.in_([ uuid.UUID(a) for a in actor_ids if isinstance(a, str) and len(a) == 36 ])) )).all() actor_lookup = { str(a.id): { "display_name": a.display_name or "", "email": a.work_email or "", } for a in actor_rows } recent_decommissions = [ { "scope_id": str(r.id), "display_name": r.display_name, "external_id": r.external_id, "integration_name": r.integration_name, "provider": r.provider, "decommissioned_at": ( r.decommissioned_at.isoformat() if r.decommissioned_at else None ), "reason": r.decommission_reason, "provider_state": r.decommission_provider_state, "actor_display_name": ( actor_lookup.get(str(r.decommissioned_by), {}).get("display_name") if r.decommissioned_by else None ), "actor_email": ( actor_lookup.get(str(r.decommissioned_by), {}).get("email") if r.decommissioned_by else None ), } for r in recent_decom_rows ] # Pending self-service scope requests — count only (drill-down # lives at ``/admin/cloud/scope-requests``). pending_request_count = (await db.execute( select(func.count()).select_from(CmdbCloudScopeRequest).where( CmdbCloudScopeRequest.status.in_(( SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_FEISHU_SENT, SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_PROVISIONING, )), *integ_filter_request, ) )).scalar() or 0 # Month-anchored per-scope cost ranking. Powers the Subscriptions # tab's "top spenders" chart + the click-through into the per-scope # billing-detail drawer. Computed once here so the page doesn't # round-trip per scope. Pulls from ``cmdb_cost_facts`` (the same # ground truth the drawer reads) — admin total + the drawer's # scope_total will agree to the cent. cost_ranking = await _cost_by_scope_ranking( db, integration_id=integration_id, month=month, dept_pids=dept_pids, ) return { "totals": { "all": int(total_all), "active": int(total_active), "closed": int(total_closed), "suspended": int(total_suspended), }, "cost_ranking": cost_ranking, "owner_health": { "owned": int(owned), "unowned": int(unowned), # Two distinct leaver-state buckets (see _subscriptions_section # for the rationale on splitting). Frontends MUST render # both — collapsing them back into a single chip recreates # the conflation that masked the 2026-05 incident. "owner_in_offboarding": int(owner_in_offboarding), "owner_left": int(owner_left), # Backwards-compatible aggregate: kept as the historical # union (in_offboarding + left) so older API consumers / # MCP clients that haven't migrated yet keep working. New # widgets MUST use the two split keys instead. "owner_left_org": int(owner_in_offboarding) + int(owner_left), }, "by_status": _kv_rows(by_status_rows), "by_environment": _kv_rows(by_env_rows), "by_scope_type": _kv_rows(by_type_rows), "by_provider": _kv_rows(by_provider_rows), "by_integration": [ { "integration_id": str(r[0]), "integration_name": r[1], "provider": r[2], "count": int(r[3] or 0), } for r in by_integration_rows ], "recent_decommissions": recent_decommissions, "pending_scope_requests": int(pending_request_count), } # ── Permissions (grants) ──────────────────────────────────────────── async def _permissions_section( db: AsyncSession, integration_id: str | None, *, dept_pids: list[str] | None = None, ) -> dict[str, Any]: """Grant counts, expiring buckets, recovered (回收) history. Most aggregations join through :class:`CmdbCloudScope` so the integration filter can be applied — grants do not carry an ``integration_id`` column directly. When ``dept_pids`` is set the section is narrowed to grants *held by* people in the selected department (``CmdbCloudAccessGrant.person_id``) — the holder's department, not the scope owner's. Orphan-history rows (``person_id IS NULL``) carry no department and therefore drop out of every count under a department filter, which is the correct behaviour (a swept orphan principal has no Helix person to attribute). """ # Helper: a base ``WHERE`` that scopes grants to the selected # integration AND drops grants whose parent scope is soft-deleted. # Always include ``CmdbCloudScope.deleted_at.is_(None)`` — without # it, grants on a scope that the decommission flow soft-deleted # leak into every count, and the Permissions tab's totals diverge # from the Subscriptions tab's totals (which already filter # ``deleted_at``). This was a P0 finding from the launch-readiness # logic-scan pass. integ_clause: list = [CmdbCloudScope.deleted_at.is_(None)] if integration_id: integ_clause.append(CmdbCloudScope.integration_id == integration_id) if dept_pids is not None: # Holder-based department filter. Applied to every grant query # below (all of which select_from CmdbCloudAccessGrant), so the # totals / expiring / recovered slices stay mutually consistent. integ_clause.append(CmdbCloudAccessGrant.person_id.in_(dept_pids)) # Per-status totals (active/pending/expired/revoked/disabled). by_status_rows = (await db.execute( select(CmdbCloudAccessGrant.status, func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where(*integ_clause) .group_by(CmdbCloudAccessGrant.status) )).all() status_map = {str(s): int(c or 0) for s, c in by_status_rows} # Distinct holders (active grants only) — "how many people have # active cloud access". Excludes orphan history rows (person_id # NULL). distinct_holders = (await db.execute( select(func.count(func.distinct(CmdbCloudAccessGrant.person_id))) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.person_id.isnot(None), *integ_clause, ) )).scalar() or 0 # Active grants split by provider / environment (live scopes only). by_provider_active = (await db.execute( select(Integration.provider, func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .join(Integration, Integration.id == CmdbCloudScope.integration_id) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, *integ_clause, ) .group_by(Integration.provider) )).all() by_environment_active = (await db.execute( select(CmdbCloudScope.environment, func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, *integ_clause, ) .group_by(CmdbCloudScope.environment) )).all() # Expiring buckets — counted from ``active`` grants whose # ``expires_at`` falls within the named horizon. Mutually # exclusive: a t1 grant is NOT also counted in t7. The order of # checks (most-urgent first) is what gives that property. now = _utc_now_naive() expiring_rows = (await db.execute( select(CmdbCloudAccessGrant.expires_at) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.expires_at.isnot(None), CmdbCloudAccessGrant.expires_at >= now, CmdbCloudAccessGrant.expires_at <= now + timedelta(days=30), *integ_clause, ) )).scalars().all() bucket_t1 = bucket_t7 = bucket_t14 = bucket_t30 = 0 for ea in expiring_rows: if ea is None: continue delta = (ea - now).total_seconds() / 86400.0 if delta <= 1: bucket_t1 += 1 elif delta <= 7: bucket_t7 += 1 elif delta <= 14: bucket_t14 += 1 else: bucket_t30 += 1 # Already-expired but still ``status='active'`` — and ACTIONABLE. # # This used to count every past-due active row and render a red # "N 条权限已超过到期时间但仍处于生效状态" alert. On 2026-07-29 that N # was 460, of which **zero** were actionable: 320 were production # (deliberately unmanaged — no reminders, no auto-revoke, the R&D # security team owns prod end to end) and 140 were scope owners on # their own scope (exempt by design). A permanently-red alert whose # every row is exempt does not inform anyone; it trains admins to # ignore that corner of the page. # # So the headline number now excludes both by-design populations, and # they are reported as their own counters (``prod_overdue`` below / # ``owner_exempt_past_due``) so the information is not lost — just no # longer disguised as a problem. active_past_due = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.expires_at.isnot(None), CmdbCloudAccessGrant.expires_at < now, CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), or_( CmdbCloudScope.owner_person_id.is_(None), CmdbCloudScope.owner_person_id != CmdbCloudAccessGrant.person_id, ), *integ_clause, ) )).scalar() or 0 # By-design exempt populations, reported separately rather than # inflating the alert above. owner_exempt_past_due = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.expires_at.isnot(None), CmdbCloudAccessGrant.expires_at < now, CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), CmdbCloudScope.owner_person_id == CmdbCloudAccessGrant.person_id, *integ_clause, ) )).scalar() or 0 # Recovered (回收) — the user-asked-for "回收了多少个" metric. # Groups every closure-of-access path: revoked / expired / disabled. # Requires ``revoked_at IS NOT NULL`` so the total is consistent # with the 30d/90d slices below — without it, a grant transitioned # to ``expired`` purely by ``expires_at`` passing (sweeper hasn't # run, ``revoked_at`` never stamped) would inflate ``total`` while # being absent from the 30d/90d windows. P0 finding from the # launch-readiness review pass. cutoff_30 = now - timedelta(days=30) cutoff_90 = now - timedelta(days=90) recovered_total = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED), CmdbCloudAccessGrant.revoked_at.isnot(None), *integ_clause, ) )).scalar() or 0 recovered_30d = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED), CmdbCloudAccessGrant.revoked_at.isnot(None), CmdbCloudAccessGrant.revoked_at >= cutoff_30, *integ_clause, ) )).scalar() or 0 recovered_90d = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED), CmdbCloudAccessGrant.revoked_at.isnot(None), CmdbCloudAccessGrant.revoked_at >= cutoff_90, *integ_clause, ) )).scalar() or 0 # Reason breakdown for the recent (30d) recovered set. Mirrors the # constants in :mod:`models` (user_released / expired / inactive / # offboarding / owner_removed / admin_removed / drift_removed / # iam_user_disowned / orphan_principal_swept). by_reason_30d_rows = (await db.execute( select(CmdbCloudAccessGrant.revoke_reason, func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED), CmdbCloudAccessGrant.revoked_at.isnot(None), CmdbCloudAccessGrant.revoked_at >= cutoff_30, *integ_clause, ) .group_by(CmdbCloudAccessGrant.revoke_reason) )).all() # Orphan-swept counters (the cloud-side "user not found" cleanups). orphan_total = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.revoke_reason == GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, *integ_clause, ) )).scalar() or 0 orphan_30d = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.revoke_reason == GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, CmdbCloudAccessGrant.revoked_at.isnot(None), CmdbCloudAccessGrant.revoked_at >= cutoff_30, *integ_clause, ) )).scalar() or 0 # Recent recovered grants — most recent 30 across all reasons. The # frontend table renders person + scope + reason + when. We # explicitly select columns instead of full ORM rows so the JSON # shape stays cheap to serialize. recent_recovered_rows = (await db.execute( select( CmdbCloudAccessGrant.id, CmdbCloudAccessGrant.person_id, CmdbCloudAccessGrant.role_key, CmdbCloudAccessGrant.role_display_name, CmdbCloudAccessGrant.status, CmdbCloudAccessGrant.revoke_reason, CmdbCloudAccessGrant.revoked_at, CmdbCloudAccessGrant.expires_at, CmdbCloudAccessGrant.external_principal_email, CmdbCloudAccessGrant.scope_id, CmdbCloudScope.display_name.label("scope_display_name"), CmdbCloudScope.external_id.label("scope_external_id"), CmdbCloudScope.environment, Integration.name.label("integration_name"), Integration.provider, ) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .join(Integration, Integration.id == CmdbCloudScope.integration_id) .where( CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED), CmdbCloudAccessGrant.revoked_at.isnot(None), *integ_clause, ) .order_by(CmdbCloudAccessGrant.revoked_at.desc()) .limit(30) )).all() # Resolve person rows for the recent set so the table can show # display_name + work_email instead of an opaque UUID. Orphan rows # carry person_id NULL — fall back to ``external_principal_email``. person_ids = { str(r.person_id) for r in recent_recovered_rows if r.person_id } person_lookup: dict[str, dict[str, str]] = {} if person_ids: person_rows = (await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employee_no, IdentityPerson.employment_status, ) .where(IdentityPerson.id.in_([ uuid.UUID(p) for p in person_ids if isinstance(p, str) and len(p) == 36 ])) )).all() person_lookup = { str(p.id): { "display_name": p.display_name or "", "email": p.work_email or "", # Leaver visibility for PersonTag (工号 + 在职状态). "employee_no": p.employee_no or "", "employment_status": p.employment_status or "", } for p in person_rows } recent_recovered = [] for r in recent_recovered_rows: person = ( person_lookup.get(str(r.person_id)) if r.person_id else None ) recent_recovered.append({ "grant_id": str(r.id), "scope_id": str(r.scope_id), "scope_display_name": r.scope_display_name, "scope_external_id": r.scope_external_id, "integration_name": r.integration_name, "provider": r.provider, "environment": r.environment, "role_key": r.role_key, "role_display_name": r.role_display_name, "status": r.status, "revoke_reason": r.revoke_reason, "revoked_at": ( r.revoked_at.isoformat() if r.revoked_at else None ), "person_display_name": person["display_name"] if person else None, "person_email": ( person["email"] if person else r.external_principal_email ), "person_employee_no": ( person["employee_no"] if person else None ), "person_employment_status": ( person["employment_status"] if person else None ), "is_orphan_history": r.person_id is None, }) return { "totals": { "active": int(status_map.get(GRANT_STATUS_ACTIVE, 0)), "pending": int(status_map.get(GRANT_STATUS_PENDING, 0)), "expired": int(status_map.get(GRANT_STATUS_EXPIRED, 0)), "revoked": int(status_map.get(GRANT_STATUS_REVOKED, 0)), "disabled": int(status_map.get(GRANT_STATUS_DISABLED, 0)), "all": int(sum(status_map.values())), "distinct_active_holders": int(distinct_holders), }, "by_status": _kv_rows(by_status_rows), "by_provider_active": _kv_rows(by_provider_active), "by_environment_active": _kv_rows(by_environment_active), "expiring": { "t1": int(bucket_t1), "t7": int(bucket_t7), "t14": int(bucket_t14), "t30": int(bucket_t30), "active_past_due": int(active_past_due), # By-design exempt, reported separately so the actionable # count above stays trustworthy. "owner_exempt_past_due": int(owner_exempt_past_due), }, "recovered": { "total": int(recovered_total), "last_30d": int(recovered_30d), "last_90d": int(recovered_90d), "by_reason_30d": _kv_rows(by_reason_30d_rows), "orphan_swept_total": int(orphan_total), "orphan_swept_30d": int(orphan_30d), "recent": recent_recovered, }, } # ── IAM users ─────────────────────────────────────────────────────── async def _iam_users_section( db: AsyncSession, integ_filter_iam: list, ) -> dict[str, Any]: """IAM users discovered by the IAM sync — counts + owner mapping.""" # ``deleted_at`` IS NULL filters out soft-deleted rows. ``revoked_at`` # rows are intentionally surfaced (admins want to see "we cleaned up # 12 IAM users last month"). base = and_( CmdbIamUser.deleted_at.is_(None), *integ_filter_iam, ) total = (await db.execute( select(func.count()).select_from(CmdbIamUser).where(base) )).scalar() or 0 revoked = (await db.execute( select(func.count()).select_from(CmdbIamUser).where( base, CmdbIamUser.revoked_at.isnot(None), ) )).scalar() or 0 owned = (await db.execute( select(func.count()).select_from(CmdbIamUser).where( base, CmdbIamUser.revoked_at.is_(None), CmdbIamUser.owner_person_id.isnot(None), ) )).scalar() or 0 unowned = (await db.execute( select(func.count()).select_from(CmdbIamUser).where( base, CmdbIamUser.revoked_at.is_(None), CmdbIamUser.owner_person_id.is_(None), # Same definition of "orphan" the list endpoint uses # (``/cmdb/iam-users?has_owner=false``): a declared service # account has no owner by design and is not a finding. Without # this the KPI and the list it links to disagree forever — # marking the two Tencent robots as service accounts would # empty the list while this tile still showed 未匹配 = 2. or_( CmdbIamUser.account_kind.is_(None), CmdbIamUser.account_kind != "service", ), ) )).scalar() or 0 by_provider_rows = (await db.execute( select(Integration.provider, func.count()) .select_from(CmdbIamUser) .join(Integration, Integration.id == CmdbIamUser.integration_id) .where(base, CmdbIamUser.revoked_at.is_(None)) .group_by(Integration.provider) )).all() mfa_disabled = (await db.execute( select(func.count()).select_from(CmdbIamUser).where( base, CmdbIamUser.revoked_at.is_(None), CmdbIamUser.mfa_enabled.is_(False), ) )).scalar() or 0 return { "total": int(total), "active": int(total - revoked), "revoked_lifetime": int(revoked), "owned": int(owned), "unowned": int(unowned), "mfa_disabled": int(mfa_disabled), "by_provider": _kv_rows(by_provider_rows), } # ── Alerts (governance) ───────────────────────────────────────────── async def _alerts_section( db: AsyncSession, integration_id: str | None, *, dept_pids: list[str] | None = None, ) -> dict[str, Any]: """Fleet-level governance signals an admin should see at a glance. Each metric is a count, never a list — drill-down lives on the matching admin page (``/admin/cloud/scopes`` filters, etc.). Under a department filter the grant-based alerts narrow to grants *held by* that department (consistent with the Permissions section), and ``unowned_scopes`` narrows to that department's owned scopes — which is always zero by construction (an unowned scope has no owner and therefore no department), correctly signalling "this department has no un-owned scopes" rather than leaking the fleet total. """ # Drop soft-deleted scopes from every grant join — same correctness # requirement as ``_permissions_section``. P0 finding from the # launch-readiness review. grant_scope_clause: list = [CmdbCloudScope.deleted_at.is_(None)] if integration_id: grant_scope_clause.append( CmdbCloudScope.integration_id == integration_id, ) if dept_pids is not None: grant_scope_clause.append( CmdbCloudAccessGrant.person_id.in_(dept_pids), ) live_scope_filter: list = [ CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status != SCOPE_STATUS_CLOSED, ] if integration_id: live_scope_filter.append( CmdbCloudScope.integration_id == integration_id, ) if dept_pids is not None: live_scope_filter.append( CmdbCloudScope.owner_person_id.in_(dept_pids), ) live_scope = and_(*live_scope_filter) # Exclude intentionally-ownerless GCP billing-account scopes from the # fleet "unowned scopes" governance alert — same rationale as the # _subscriptions_section owner_health.unowned count + the reconciliation # owner-orphan audit. Without this the alert stays permanently lit on # the unfiltered overview (the dept-scoped path is naturally 0 since an # ownerless scope can't be attributed to a department). unowned_scopes = (await db.execute( select(func.count()).select_from(CmdbCloudScope).where( live_scope, CmdbCloudScope.owner_person_id.is_(None), CmdbCloudScope.scope_type != SCOPE_TYPE_GCP_BILLING_ACCOUNT, # Aliyun RD master (payer) is owner-exempt infra — see # reconciliation owner-orphan audit. Don't light the alert. func.json_extract( CmdbCloudScope.raw_data, "$.rd_master", ).is_(None), ) )).scalar() or 0 now = _utc_now_naive() # ACTIONABLE past-due grants only — prod (unmanaged by design) and # scope-owner-on-own-scope (exempt by design) are excluded and counted # separately. Same predicate as # ``permissions.expiring.active_past_due``; the two must stay # byte-identical or the overview and the permissions tab disagree # about the same number on the same page. See the long note there: # the unfiltered version rendered a red alert reading 460 of which 0 # were actionable. active_past_due = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.expires_at.isnot(None), CmdbCloudAccessGrant.expires_at < now, CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), or_( CmdbCloudScope.owner_person_id.is_(None), CmdbCloudScope.owner_person_id != CmdbCloudAccessGrant.person_id, ), *grant_scope_clause, ) )).scalar() or 0 # Prod scopes — past-due grants that are intentionally NOT # auto-revoked. NB: the sweeper does not notify anyone about these # either (PR #1961 removed the owner ping — prod cloud access is # unmanaged by Helix end to end, see ``_is_prod_grant``). This number # exists purely so an admin can see "X prod grants are dragging" and # hand it to the R&D security team. prod_overdue = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, CmdbCloudAccessGrant.expires_at.isnot(None), CmdbCloudAccessGrant.expires_at < now, CmdbCloudScope.environment == "prod", *grant_scope_clause, ) )).scalar() or 0 # Pending grants (someone requested, owner hasn't decided). pending_grants = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ) .where( CmdbCloudAccessGrant.status == GRANT_STATUS_PENDING, *grant_scope_clause, ) )).scalar() or 0 return { "unowned_scopes": int(unowned_scopes), "active_past_due_grants": int(active_past_due), "prod_overdue_grants": int(prod_overdue), "pending_grants": int(pending_grants), } # ── Per-scope cost ranking (Subscriptions tab) ──────────────────── # Top-N for the cost ranking. The Subscriptions tab is dense enough # that >20 rows starts to feel like a leaderboard rather than a # governance signal; 20 covers virtually every PingCAP-prod single- # tenant ranking we care about. _COST_RANKING_TOPN = 20 async def _cost_by_scope_ranking( db: AsyncSession, *, integration_id: str | None, month: str | None, dept_pids: list[str] | None = None, ) -> dict[str, Any]: """Top-N scopes by spend within the requested month + fleet total. Read from ``cmdb_cost_facts`` so the admin tab agrees with the per-scope billing-detail drawer to the cent (the drawer reads the same table). RG-typed scopes are counted under their parent subscription's scope_id — cost_facts for an RG carry the parent subscription's scope_id, not the RG-scope's own; reverse-mapping to "show the RG inside the ranking" would mean a second pass through ``resource_group`` and is intentionally deferred to the drawer's resource-level breakdown. When ``integration_id`` filter is set, the ranking is narrowed to that tenant — same contract as every other section here. Past months show the settled total; the current month is clamped to today (UTC). Mirrors the parse / clamp rules of the per-scope billing-detail endpoint so a click on a ranking row opens a drawer that totals to the same number this list shows. """ # Late import — scope_billing_detail imports from this module's # tree indirectly via lifecycle_router; lazy import dodges any # accidental circular at app boot. from app.domains.cmdb.scope_billing_detail import ( _month_bounds, parse_month_param, ) year, month_num = parse_month_param(month) month_start, month_end, is_partial = _month_bounds(year, month_num) where_clauses: list = [ CmdbCostFact.date >= month_start, CmdbCostFact.date <= month_end, # Drop NULL scope_id rows — those are tenant-aggregate facts # (e.g. an Aliyun BSS row that couldn't be mapped to a scope) # and would dominate the ranking under "(unknown)" if left in. CmdbCostFact.scope_id.is_not(None), ] if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) if dept_pids is not None: # Narrow to scopes whose CURRENT owner sits in the department — # the SAME live ``owner_person_id`` predicate the active-count and # subscription-list sections use, so all three sections of the tab # agree on which subscriptions belong to the department. # # The previous implementation filtered on the denormalised # ``cmdb_cost_facts.cost_owner_department`` (the owner's dept # stamped at fact-build time). That column does NOT move when a # scope's ``owner_person_id`` is reassigned — it only re-stamps on # the next cost ingest or the nightly ``recompute_allocation`` — so # in the window after an owner transfer the cost ranking kept # showing a scope under the OLD owner's department while the count # and list (live owner) had already dropped it. Prod repro # (2026-06-16): ``security-center`` transferred to 李卓然 (Security # dept) still ranked under "Business Operations" because its May # facts carried ``cost_owner_department='IT Services & Security'`` # (a BO-subtree leaf). Keying on the live scope owner removes the # staleness window entirely. ``dept_pids`` empty → no scopes → # empty ranking (correct: the department owns no scope with spend). scope_subq = select(CmdbCloudScope.id).where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.owner_person_id.in_(dept_pids), ) if integration_id: scope_subq = scope_subq.where( CmdbCloudScope.integration_id == integration_id, ) where_clauses.append(CmdbCostFact.scope_id.in_(scope_subq)) # First pass: top-N scopes by spend. top_rows = (await db.execute( select( CmdbCostFact.scope_id, func.coalesce(func.sum(fact_engineering_amount()), 0).label("usd"), ) .where(and_(*where_clauses)) .group_by(CmdbCostFact.scope_id) .order_by(func.sum(fact_engineering_amount()).desc()) .limit(_COST_RANKING_TOPN) )).all() # Fleet total — same WHERE, no GROUP BY. Powers the "Σ across N # scopes" headline. Past months: this equals the settled fleet # spend; current month: it's MTD. fleet_total = (await db.execute( select( func.coalesce(func.sum(fact_engineering_amount()), 0), ).where(and_(*where_clauses)) )).scalar() or 0 if not top_rows: return { "month": f"{year:04d}-{month_num:02d}", "is_partial": is_partial, "fleet_total_usd": float(fleet_total), "top": [], } # Resolve scope display info in one shot. # Pre-launch hardening 2026-05-16 (LS1 P2): also pull # ``deleted_at`` for each row so the ranking label can render # "(deleted)" instead of pretending a soft-deleted scope is # still alive. Without this filter the soft-delete UX leaked # through: a closed scope still showed up in admin top-spenders # under its live display_name with the dollars attributed # correctly but no indication that it's been retired. scope_ids = [r.scope_id for r in top_rows if r.scope_id] scopes = list((await db.execute( select(CmdbCloudScope).where( CmdbCloudScope.id.in_(scope_ids), ), )).scalars().all()) scope_map = {str(s.id): s for s in scopes} # Pre-compute a "this scope is soft-deleted" set so the loop below # can render the (deleted) label without inferring from another # column. Cost rows survive the scope's soft-delete (cost_facts # has ON DELETE SET NULL, see PR #126 — and scopes are only ever # soft-deleted today anyway, so the FK never fires). soft_deleted_scope_ids = { str(s.id) for s in scopes if s.deleted_at is not None } integ_ids = {str(s.integration_id) for s in scopes} integ_map: dict[str, Any] = {} if integ_ids: from app.integrations.models import Integration as _Integration integs = list((await db.execute( select(_Integration).where( _Integration.id.in_(integ_ids), ), )).scalars().all()) integ_map = {str(i.id): i for i in integs} out_top: list[dict[str, Any]] = [] fleet_total_d = float(fleet_total) for r in top_rows: sid = r.scope_id s = scope_map.get(sid) if s is None: # Scope deleted since the fact landed — keep the row but # render a "(deleted)" label so the spend isn't silently # dropped from the fleet total displayed alongside. # Round-1 review follow-up (Review-C P1): keep the dict # shape identical to the live-scope branch below so the FE # doesn't see heterogeneous rows (missing ``deleted`` / # ``has_rg_breakdown`` would render ``undefined`` for any # strict consumer / chart column). out_top.append({ "scope_id": sid, "display_name": None, "external_id": None, "scope_type": None, "provider": None, "integration_name": None, "owner_person_id": None, "budget_usd_monthly": None, "deleted": True, "has_rg_breakdown": False, "usd": float(r.usd or 0), "share": ( float(r.usd or 0) / fleet_total_d if fleet_total_d > 0 else 0.0 ), }) continue integ = integ_map.get(str(s.integration_id)) is_soft_deleted = str(s.id) in soft_deleted_scope_ids display_name = s.display_name if is_soft_deleted and display_name: display_name = f"{display_name} (deleted)" # Pre-launch hardening 2026-05-16 follow-up (LS1 P2): Azure # resource-group scopes' cost_facts carry the PARENT # subscription's scope_id (verified at ingest), so RG spend # rolls into the parent's ranking row. Expose ``has_rg_breakdown`` # so the FE can show a "包含 RG 细分,点开查看" affordance — the # per-scope drawer's resource-level breakdown remains the # canonical place to see the RG split. has_rg_breakdown = ( s.scope_type == "azure_subscription" and (integ.provider == "azure" if integ else False) ) out_top.append({ "scope_id": sid, "display_name": display_name, "external_id": s.external_id, "scope_type": s.scope_type, "provider": integ.provider if integ else None, "integration_name": integ.name if integ else None, "owner_person_id": s.owner_person_id, "budget_usd_monthly": ( float(s.budget_usd_monthly) if s.budget_usd_monthly is not None else None ), "deleted": is_soft_deleted, "has_rg_breakdown": has_rg_breakdown, "usd": float(r.usd or 0), "share": ( float(r.usd or 0) / fleet_total_d if fleet_total_d > 0 else 0.0 ), }) return { "month": f"{year:04d}-{month_num:02d}", "is_partial": is_partial, "fleet_total_usd": fleet_total_d, "top": out_top, } async def department_cost_reconciliation( db: AsyncSession, *, department: str, month: str | None = None, integration_id: str | None = None, ) -> dict[str, Any]: """Side-by-side BOTH-口径 cloud-cost view for a department + month. "How much did department X spend on the cloud last month, by cloud account and by provider, and what's the total?" has TWO legitimate answers that the platform exposes in different tabs, and they differ materially when an account's cost serves more than one team: * **Allocation 口径** (``allocation_*``) — spend *attributed* to the department's people via the per-fact ``cost_owner_department`` (allocation rules → scope-owner fallback, nightly recomputed). A single shared cloud account is SPLIT across departments. This is "what did the department actually consume." Mirrors the 账单 tab's drilldown. * **Ownership 口径** (``ownership_*``) — the WHOLE spend of every cloud account whose CURRENT owner sits in the department. A shared account is attributed in full to the owner's department. This is "what do the department's accounts cost." Mirrors the 云订阅 tab's cost ranking (which is keyed on the live scope owner). Both are returned together — by provider, by account, and as totals — so the reader can see the gap and its cause in one place (e.g. an account owned by the department but whose resources are charged out to other teams shows a large ownership amount and a small allocation amount on the same row). Department matches the whole subtree under the chosen node (parent rolls up every descendant), consistent with every other department filter. ``integration_id`` optionally narrows to one cloud tenant. """ from app.core.org_scope import resolve_department_subtree_names from app.domains.cmdb.scope_billing_detail import ( _month_bounds, parse_month_param, ) year, month_num = parse_month_param(month) month_start, month_end, is_partial = _month_bounds(year, month_num) leaf_names = await resolve_department_subtree_names(db, department) dept_pids = await resolve_department_person_ids(db, department) date_window = [ CmdbCostFact.date >= month_start, CmdbCostFact.date <= month_end, ] if integration_id: date_window.append(CmdbCostFact.integration_id == integration_id) # ── Allocation 口径: facts whose cost_owner_department ∈ subtree ── alloc_where = [*date_window] if leaf_names is not None: alloc_where.append( CmdbCostFact.cost_owner_department.in_(leaf_names), ) alloc_by_provider_rows = (await db.execute( select( Integration.provider, func.coalesce(func.sum(fact_engineering_amount()), 0).label("usd"), ) .join(Integration, Integration.id == CmdbCostFact.integration_id) .where(and_(*alloc_where)) .group_by(Integration.provider) )).all() alloc_by_scope_rows = (await db.execute( select( CmdbCostFact.scope_id, func.coalesce(func.sum(fact_engineering_amount()), 0).label("usd"), ) .where(and_(*alloc_where, CmdbCostFact.scope_id.is_not(None))) .group_by(CmdbCostFact.scope_id) )).all() alloc_total = float((await db.execute( select(func.coalesce(func.sum(fact_engineering_amount()), 0)) .where(and_(*alloc_where)) )).scalar() or 0) # ── Ownership 口径: every fact of a scope the department CURRENTLY # owns. ``dept_pids`` empty → no scopes → zero rows. ── own_by_provider_rows = [] own_by_scope_rows = [] own_total = 0.0 if dept_pids: owned_scope_subq = select(CmdbCloudScope.id).where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.owner_person_id.in_(dept_pids), ) if integration_id: owned_scope_subq = owned_scope_subq.where( CmdbCloudScope.integration_id == integration_id, ) own_where = [*date_window, CmdbCostFact.scope_id.in_(owned_scope_subq)] own_by_provider_rows = (await db.execute( select( Integration.provider, func.coalesce(func.sum(fact_engineering_amount()), 0).label("usd"), ) .join(Integration, Integration.id == CmdbCostFact.integration_id) .where(and_(*own_where)) .group_by(Integration.provider) )).all() own_by_scope_rows = (await db.execute( select( CmdbCostFact.scope_id, func.coalesce(func.sum(fact_engineering_amount()), 0).label("usd"), ) .where(and_(*own_where, CmdbCostFact.scope_id.is_not(None))) .group_by(CmdbCostFact.scope_id) )).all() own_total = float((await db.execute( select(func.coalesce(func.sum(fact_engineering_amount()), 0)) .where(and_(*own_where)) )).scalar() or 0) # ── Merge by provider ── alloc_prov = {r.provider: float(r.usd or 0) for r in alloc_by_provider_rows} own_prov = {r.provider: float(r.usd or 0) for r in own_by_provider_rows} by_provider = [ { "provider": prov, "allocation_usd": round(alloc_prov.get(prov, 0.0), 2), "ownership_usd": round(own_prov.get(prov, 0.0), 2), } for prov in sorted( set(alloc_prov) | set(own_prov), key=lambda p: max(alloc_prov.get(p, 0), own_prov.get(p, 0)), reverse=True, ) ] # ── Merge by account (scope) ── alloc_scope = { str(r.scope_id): float(r.usd or 0) for r in alloc_by_scope_rows } own_scope = { str(r.scope_id): float(r.usd or 0) for r in own_by_scope_rows } all_scope_ids = set(alloc_scope) | set(own_scope) scope_meta: dict[str, Any] = {} owner_names: dict[str, str] = {} if all_scope_ids: scopes = list((await db.execute( select(CmdbCloudScope).where(CmdbCloudScope.id.in_(all_scope_ids)) )).scalars().all()) scope_meta = {str(s.id): s for s in scopes} integ_ids = {str(s.integration_id) for s in scopes} integ_map = { str(i.id): i for i in (await db.execute( select(Integration).where(Integration.id.in_(integ_ids)) )).scalars().all() } if integ_ids else {} owner_ids = { str(s.owner_person_id) for s in scopes if s.owner_person_id } if owner_ids: owner_names = { str(p.id): (p.display_name or p.work_email or str(p.id)) for p in (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(owner_ids), ) )).scalars().all() } dept_pid_set = set(dept_pids or []) by_account = [] for sid in all_scope_ids: s = scope_meta.get(sid) integ = integ_map.get(str(s.integration_id)) if s else None owner_pid = str(s.owner_person_id) if s and s.owner_person_id else None by_account.append({ "scope_id": sid, "provider": integ.provider if integ else None, "account": (s.display_name if s else None), "external_id": (s.external_id if s else None), "owner_person_id": owner_pid, "owner_name": owner_names.get(owner_pid) if owner_pid else None, "owner_in_department": ( owner_pid in dept_pid_set if owner_pid else False ), "allocation_usd": round(alloc_scope.get(sid, 0.0), 2), "ownership_usd": round(own_scope.get(sid, 0.0), 2), }) by_account.sort( key=lambda r: max(r["allocation_usd"], r["ownership_usd"]), reverse=True, ) return { "department": department, "month": f"{year:04d}-{month_num:02d}", "is_partial": is_partial, "allocation_total_usd": round(alloc_total, 2), "ownership_total_usd": round(own_total, 2), "by_provider": by_provider, "by_account": by_account, }