from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from datetime import date as date_type from typing import Any from sqlalchemy import and_, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.cost_view import ( COST_MEASURE_INVOICE, COST_MEASURE_LIST, daily_engineering_amount, ) from app.domains.cmdb.models import ( SCOPE_ACCOUNT_RAW_KEYS, SCOPE_RAW_PATHS, CmdbBillingMonthly, CmdbCdnDomain, CmdbCertificate, CmdbCloudBillingDaily, CmdbCloudScope, CmdbCostFact, CmdbDnsRecord, CmdbDomain, CmdbIamUser, CmdbLoadBalancer, CmdbObjectStorage, CmdbServer, CmdbSyncJob, ) from app.domains.cmdb.visibility import OwnerVisibility, visibility_clause from app.integrations.models import Integration logger = logging.getLogger(__name__) @dataclass(frozen=True) class ScopeServerResolution: """Servers attributable to one cloud scope, with the proof used. A cloud subscription must not appear to have no machines merely because its cost export has not emitted a resource-level line. Inventory drivers already stamp the provider account/project/subscription onto ``raw_data``; use that first, then retain the cost-fact relation as a fallback. The source counts are deliberately aggregate-only so callers can explain data coverage without exposing inventory outside their normal RBAC projection. """ servers: list[CmdbServer] direct_count: int cost_fallback_count: int def _server_account_stamp(server: CmdbServer) -> str | None: """The provider account / project / subscription every sync driver stamps onto a server's ``raw_data``. Each driver writes exactly one of these keys (AWS ``_aws_account_id``, GCP ``_project_id``, Azure ``_subscription_id``, AliCloud ``__sub_account_id``), and the value is the same string the owning scope carries in ``external_id`` — so this is the provider's own account membership, not an inference from names or tags. """ raw = server.raw_data or {} for key in SCOPE_ACCOUNT_RAW_KEYS: value = raw.get(key) if value: return str(value) return None def _scope_inventory_clause( scope: CmdbCloudScope, *, parent_subscription_id: str | None = None, ) -> Any | None: """Return the provider-native predicate which proves scope ownership. This stays in SQL so a scope analysis never needs to load every server from a large cloud integration. Azure resource-group names are only unique *within a subscription*, so that predicate deliberately requires both the ARM resource-group path and the parent subscription stamp. """ external_id = str(scope.external_id or "").strip() if not external_id: return None raw = CmdbServer.raw_data if scope.scope_type == "azure_resource_group": if not parent_subscription_id: return None group = _esc(external_id.lower()) return and_( func.lower(func.json_unquote(func.coalesce( func.json_extract(raw, "$.id"), func.json_extract(raw, "$.resourceId"), ))).like( f"%/resourcegroups/{group}/%", escape="\\", ), func.json_unquote(func.coalesce( func.json_extract(raw, "$._subscription_id"), func.json_extract(raw, "$.subscription_id"), )) == parent_subscription_id, ) raw_path = SCOPE_RAW_PATHS.get(scope.scope_type) if raw_path is None: return None return func.json_unquote(func.json_extract(raw, raw_path)) == external_id async def resolve_scope_servers( db: AsyncSession, scope: CmdbCloudScope, ) -> ScopeServerResolution: """Resolve every inventory server belonging to ``scope``. This is the scope-side counterpart to :func:`resolve_server_scope_ids`. It intentionally combines provider-stamped inventory attribution with the existing cost-fact relation instead of treating billing export coverage as a prerequisite for resource-efficiency analysis. """ parent_subscription_id: str | None = None if scope.scope_type == "azure_resource_group" and scope.parent_scope_id: parent_subscription_id = ( await db.execute( select(CmdbCloudScope.external_id).where( CmdbCloudScope.id == str(scope.parent_scope_id), CmdbCloudScope.integration_id == str(scope.integration_id), CmdbCloudScope.scope_type == "azure_subscription", CmdbCloudScope.deleted_at.is_(None), ) ) ).scalar_one_or_none() direct_clause = _scope_inventory_clause( scope, parent_subscription_id=parent_subscription_id, ) direct = [] if direct_clause is not None: direct = list( ( await db.execute( select(CmdbServer).where( CmdbServer.integration_id == str(scope.integration_id), CmdbServer.deleted_at.is_(None), direct_clause, ) ) ).scalars() ) direct_ids = {str(row.id) for row in direct} # Cost exports are an attribution fallback, not a condition for direct # inventory. RG cost facts are keyed by (integration, resource_group), # whereas tenant scopes use scope_id; keep that routing identical to the # billing API so the two views cannot disagree. from app.domains.cmdb.scope_billing_detail import scope_fact_filters fact_filters, _ = scope_fact_filters(scope) fallback = [] if fact_filters: resource_ids = ( select(CmdbCostFact.resource_id) .where(*fact_filters, CmdbCostFact.resource_id.isnot(None)) .distinct() ) fallback_stmt = select(CmdbServer).where( CmdbServer.integration_id == str(scope.integration_id), CmdbServer.deleted_at.is_(None), CmdbServer.instance_id.in_(resource_ids), ) # Billing facts for Azure RGs have no subscription discriminator. A # same-named group in another subscription must never be admitted by # the fallback relation; require the provider-stamped machine scope. if scope.scope_type == "azure_resource_group": if not parent_subscription_id: return ScopeServerResolution(direct, len(direct), 0) fallback_stmt = fallback_stmt.where( func.json_unquote(func.coalesce( func.json_extract(CmdbServer.raw_data, "$._subscription_id"), func.json_extract(CmdbServer.raw_data, "$.subscription_id"), )) == parent_subscription_id, ) if direct_ids: fallback_stmt = fallback_stmt.where(CmdbServer.id.not_in(direct_ids)) fallback = list((await db.execute(fallback_stmt)).scalars()) return ScopeServerResolution([*direct, *fallback], len(direct), len(fallback)) def _esc(s: str) -> str: """Escape LIKE metacharacters to prevent wildcard injection.""" return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") async def resolve_server_scope_ids( db: AsyncSession, servers: list[CmdbServer], ) -> dict[str, str]: """Map ``server_id → scope_id`` for the given servers, preferring the provider's own account stamp and falling back to billing attribution (same contract as :func:`resolve_inherited_server_owners`, which uses this helper): 1. the account / project / subscription each sync driver stamps onto ``raw_data`` (``SCOPE_ACCOUNT_RAW_KEYS``) → scope ``(integration_id, external_id)`` 2. any cloud: a ``cmdb_cost_facts`` line (``resource_id == instance_id`` within the integration) carrying a ``scope_id``. Step 1 used to read ``_project_id`` alone, so only GCP took the direct link and every other cloud fell through to billing — which silently resolves nothing for two of them: AWS cost lines are aggregated per account (``resource_id`` = ``aws-account::``, ``resource_type='aws-aggregated'``) and Azure's ``resource_id`` is the ARM path, neither of which equals an instance id. Production 2026-08-18: 5136 AWS + 11 Azure servers carried ``_aws_account_id`` / ``_subscription_id`` matching a live scope, yet resolved to nothing here, so every surface built on this helper (server list inherited owner, the efficiency page, the owner digest, savings attribution) showed "-" for them. Azure note: a server is attributed to its SUBSCRIPTION scope, never to an ``azure_resource_group`` child scope — the stamp is a subscription GUID, and resource-group names are only unique within a subscription (see ``_scope_inventory_clause``). Servers with neither link are simply absent from the result. Unlike the owner resolver this takes the caller's list as-is (owned servers included) — it powers scope-level attribute inheritance such as the owner-digest environment filter (a machine's environment is its subscription's environment). """ if not servers: return {} # server_id → scope_id, resolved from whichever link is available. server_scope: dict[str, str] = {} # (1) Direct link via the driver's account stamp → scope.external_id. proj_pairs: set[tuple[str, str]] = set() for s in servers: proj = _server_account_stamp(s) if proj: proj_pairs.add((str(s.integration_id), str(proj))) ext_to_scope: dict[tuple[str, str], str] = {} if proj_pairs: integ_ids = {p[0] for p in proj_pairs} exts = {p[1] for p in proj_pairs} rows = await db.execute( select( CmdbCloudScope.integration_id, CmdbCloudScope.external_id, CmdbCloudScope.id, ).where( CmdbCloudScope.integration_id.in_(integ_ids), CmdbCloudScope.external_id.in_(exts), CmdbCloudScope.deleted_at.is_(None), ), ) for integ, ext, sid in rows.all(): ext_to_scope[(str(integ), str(ext))] = str(sid) for s in servers: proj = _server_account_stamp(s) if proj: sid = ext_to_scope.get((str(s.integration_id), str(proj))) if sid: server_scope[str(s.id)] = sid # (2) Cost-facts attribution for anything still unresolved. unresolved = [s for s in servers if str(s.id) not in server_scope] if unresolved: by_integ: dict[str, list[str]] = {} for s in unresolved: by_integ.setdefault(str(s.integration_id), []).append(s.instance_id) # Map (integration_id, resource_id) → scope_id via cost facts. inst_to_scope: dict[tuple[str, str], str] = {} for integ, instance_ids in by_integ.items(): rows = await db.execute( select( CmdbCostFact.resource_id, CmdbCostFact.scope_id, ) .where( CmdbCostFact.integration_id == integ, CmdbCostFact.resource_id.in_(instance_ids), CmdbCostFact.scope_id.isnot(None), ) .distinct(), ) for resource_id, scope_id in rows.all(): if resource_id and scope_id: inst_to_scope[(integ, str(resource_id))] = str(scope_id) for s in unresolved: sid = inst_to_scope.get((str(s.integration_id), s.instance_id)) if sid: server_scope[str(s.id)] = sid return server_scope async def resolve_inherited_server_owners( db: AsyncSession, servers: list[CmdbServer], *, include_deleted: bool = False, ) -> dict[str, dict[str, str | None]]: """For servers with NO explicit ``owner_person_id``, resolve the owner they *inherit* from their cloud subscription (scope), live — never written back. Mirrors the cert-owner-resolver philosophy: the effective owner is always current, so a scope-owner change or reassignment is reflected immediately with nothing to backfill, and this stamps NO ``owner_person_id`` (so the semiannual owner-confirmation cron, which keys on a stored owner, is untouched — no notification wave). A server maps to its scope through :func:`resolve_server_scope_ids`: the provider account stamp the sync driver wrote onto ``raw_data`` first, then billing attribution as a fallback. Servers with neither (e.g. a UCloud box, whose driver stamps no account key, or a tenant with no cost ingest) simply resolve to nothing and keep showing "-". Returns ``{server_id: {person_id, scope_id, scope_name}}`` only for servers whose scope has an owner. """ # ``include_deleted`` exists for the savings tracker: the destroy # path soft-deletes the row BEFORE the audit hook fires, and the # snapshot still needs the owner the machine inherited while alive. targets = [ s for s in servers if not s.owner_person_id and (include_deleted or s.deleted_at is None) ] if not targets: return {} server_scope = await resolve_server_scope_ids(db, targets) if not server_scope: return {} # Load scope owner + display name for the resolved scopes (owner only). scope_ids = set(server_scope.values()) scope_rows = await db.execute( select( CmdbCloudScope.id, CmdbCloudScope.owner_person_id, CmdbCloudScope.display_name, CmdbCloudScope.external_id, ).where( CmdbCloudScope.id.in_(scope_ids), CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.deleted_at.is_(None), ), ) scope_info: dict[str, dict[str, str | None]] = {} for sid, owner, name, ext in scope_rows.all(): scope_info[str(sid)] = { "person_id": str(owner), "scope_name": name or ext, } out: dict[str, dict[str, str | None]] = {} for server_id, scope_id in server_scope.items(): info = scope_info.get(scope_id) if info: out[server_id] = { "person_id": info["person_id"], "scope_id": scope_id, "scope_name": info["scope_name"], } return out async def _active_cloud_integration_ids(db: AsyncSession) -> set[str]: """Return ids (as strings) of currently-enabled cloud integrations. Every CMDB cloud-resource list endpoint that does NOT pin a specific ``integration_id`` must scope to this set so that resources belonging to a disabled / "frozen" tenant don't leak into the user-facing "全部" view. When the caller explicitly picks one tenant via ``integration_id``, the caller's choice wins and this filter is bypassed — admins legitimately audit disabled tenants one at a time via the ```` "disabled" tag path. Rationale (real prod incident 2026-05-25): the 阿里云 integration was disabled but its 4 domain rows (pimgcap.com / chaosmatrix.cn / tidbcloud.cn / tidbcloud.com.cn) stayed in the table because the cloud_sync cron skips disabled tenants, which also skips the mark-and-sweep that would soft-delete them. The user saw each domain twice in the 域名管理 list (once from the disabled tenant, once from the live UCloud tenant). Same hidden gap exists for every other resource type — fixing it once here at the helper. Returns string-form ids (matching ``cmdb_*.integration_id``'s ``String(36)`` column) so ``in_(...)`` comparison is type-safe — binding ``uuid.UUID`` Python objects against a ``String(36)`` column via asyncmy silently returned 0 rows in a prior incident (see memory: reference_asyncmy_uuid_bind). """ rows = (await db.execute( select(Integration.id).where( Integration.kind == "cloud", Integration.enabled.is_(True), ) )).all() return {str(r[0]) for r in rows} class CmdbService: """Query helpers for CMDB resources (servers / domains / dns / …). Cloud-tenant credentials are NOT owned by this service — those live as ``Integration(kind='cloud')`` rows managed by :mod:`app.integrations.service`. Every resource query below accepts an ``integration_id`` as the tenant scope. """ @staticmethod async def create_sync_job( db: AsyncSession, integration_id: str, resource_types: list[str], triggered_by_user_id: str | None, triggered_by: str = "manual", ) -> CmdbSyncJob: job = CmdbSyncJob( integration_id=integration_id, triggered_by=triggered_by, triggered_by_user_id=triggered_by_user_id, resource_types=resource_types, status="queued", ) db.add(job) await db.commit() await db.refresh(job) return job @staticmethod async def list_sync_jobs( db: AsyncSession, integration_id: str | None = None, limit: int = 50 ) -> list[CmdbSyncJob]: q = select(CmdbSyncJob).order_by(CmdbSyncJob.queued_at.desc()).limit(limit) if integration_id: q = q.where(CmdbSyncJob.integration_id == integration_id) result = await db.execute(q) return list(result.scalars().all()) @staticmethod async def list_servers( db: AsyncSession, integration_id: str | None = None, integration_ids: list[str] | None = None, status: str | None = None, region: str | None = None, zone: str | None = None, project_id: str | None = None, instance_type: str | None = None, os_type: str | None = None, vcpu_min: int | None = None, vcpu_max: int | None = None, age_min_days: int | None = None, age_max_days: int | None = None, search: str | None = None, owner_person_id: str | None = None, owner_person_ids: list[str] | None = None, visibility: OwnerVisibility | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbServer], int]: from sqlalchemy import func, or_ q = select(CmdbServer).where(CmdbServer.deleted_at.is_(None)) # Self-service scoping: server-level owner ∪ servers attributed to # cloud subscriptions the reader owns (see visibility.py). Mutually # exclusive with the admin-only owner_person_id(s) filters — the # router picks one or the other. if visibility is not None: q = q.where(visibility_clause(visibility)) if integration_id: q = q.where(CmdbServer.integration_id == integration_id) elif integration_ids is not None: # Pre-resolved by caller (e.g. provider→integrations lookup). # Empty list means "no matching tenant" — short-circuit. if not integration_ids: return [], 0 q = q.where(CmdbServer.integration_id.in_(integration_ids)) else: # Scope to enabled cloud tenants only when no specific tenant # was requested. See _active_cloud_integration_ids for why. active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbServer.integration_id.in_(active)) if status: q = q.where(CmdbServer.status == status) if region: q = q.where(CmdbServer.region == region) if zone: q = q.where(CmdbServer.zone == zone) if project_id: q = q.where(CmdbServer.project_id == project_id) if instance_type: q = q.where(CmdbServer.instance_type == instance_type) if os_type: q = q.where(CmdbServer.os_type == os_type) if vcpu_min is not None: q = q.where(CmdbServer.vcpu >= vcpu_min) if vcpu_max is not None: q = q.where(CmdbServer.vcpu <= vcpu_max) # Age bucket — uses ``first_seen_at`` as proxy for "how long # Helix has known about this VM". We don't read cloud-side # ``created_at`` (lives in raw_data JSON, per-provider shape), # so age understates for VMs created before Helix's first sync # of that tenant. Acceptable for the half-yearly governance # ask (filter "已用超过 1 年" still works after Helix has been # syncing for >1y, which is the relevant horizon). now = datetime.now(UTC) if age_min_days is not None and age_min_days > 0: q = q.where(CmdbServer.first_seen_at <= now - timedelta(days=age_min_days)) if age_max_days is not None and age_max_days > 0: q = q.where(CmdbServer.first_seen_at >= now - timedelta(days=age_max_days)) if owner_person_id: q = q.where(CmdbServer.owner_person_id == owner_person_id) if owner_person_ids: q = q.where(CmdbServer.owner_person_id.in_(owner_person_ids)) if search: # IP search across public_ips / private_ips JSON columns # uses MySQL/TiDB ``JSON_SEARCH`` (returns the path of the # matched element, or NULL). The pattern is wildcard- # escaped via _esc — admins searching for a literal ``%`` # in an IP shouldn't over-match. Pre-fix the search column # set was (name, instance_id) only, so an admin looking # for ``10.9.84.114`` (the prod incident IP) had no way to # find it; today's golden-path search hit. # # JSON_SEARCH's ``one`` arg returns on first match, which # is what we want for an exists-check. The ``LIKE`` pattern # is built with ``%`` wraps so partial / suffix matches # work (admins paste an IP they remember the last octet of). pat = f"%{_esc(search)}%" q = q.where( or_( CmdbServer.name.ilike(pat), CmdbServer.instance_id.ilike(pat), text( "JSON_SEARCH(cmdb_servers.public_ips, 'one', :ip_pat) IS NOT NULL" ).bindparams(ip_pat=pat), text( "JSON_SEARCH(cmdb_servers.private_ips, 'one', :ip_pat) IS NOT NULL" ).bindparams(ip_pat=pat), ) ) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = q.order_by(CmdbServer.name).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def list_domains( db: AsyncSession, integration_id: str | None = None, expiring_within_days: int | None = None, owner_person_id: str | None = None, owner_person_ids: list[str] | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbDomain], int]: from datetime import timedelta from sqlalchemy import func, or_ q = select(CmdbDomain).where(CmdbDomain.deleted_at.is_(None)) if integration_id: q = q.where(CmdbDomain.integration_id == integration_id) else: # Disabled-tenant rows are stale (cloud_sync skips them) and # appear as cross-tenant duplicates in the "全部" view. See # _active_cloud_integration_ids for the prod incident. active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbDomain.integration_id.in_(active)) if owner_person_id: q = q.where(CmdbDomain.owner_person_id == owner_person_id) if owner_person_ids: q = q.where(CmdbDomain.owner_person_id.in_(owner_person_ids)) if expiring_within_days: if expiring_within_days <= 0: # Negative/zero values silently inverted the cutoff — bound here. expiring_within_days = 1 cutoff = datetime.now(UTC).date() + timedelta(days=expiring_within_days) q = q.where(CmdbDomain.expiry_date <= cutoff, CmdbDomain.expiry_date.is_not(None)) if search: # Apply the same LIKE-wildcard escape as list_servers / # list_dns_records — without _esc, a user searching for # ``%foo%`` would over-match every row containing ``foo``. like = f"%{_esc(search)}%" q = q.where(or_( CmdbDomain.domain_name.ilike(like), CmdbDomain.registrar.ilike(like), )) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 # MySQL/TiDB do not support ``NULLS LAST`` syntax; emulate via the # ``IS NULL`` sort key trick so rows with a NULL expiry sort to the end. q = ( q.order_by(CmdbDomain.expiry_date.is_(None), CmdbDomain.expiry_date.asc()) .offset((page - 1) * page_size) .limit(page_size) ) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def list_dns_records( db: AsyncSession, integration_id: str | None = None, zone_name: str | None = None, record_type: str | None = None, search: str | None = None, owner_person_id: str | None = None, owner_person_ids: list[str] | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbDnsRecord], int]: from sqlalchemy import func, or_ q = select(CmdbDnsRecord).where(CmdbDnsRecord.deleted_at.is_(None)) if integration_id: q = q.where(CmdbDnsRecord.integration_id == integration_id) else: active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbDnsRecord.integration_id.in_(active)) if zone_name: q = q.where(CmdbDnsRecord.zone_name == zone_name) if record_type: q = q.where(CmdbDnsRecord.record_type == record_type) if owner_person_id: q = q.where(CmdbDnsRecord.owner_person_id == owner_person_id) if owner_person_ids: q = q.where(CmdbDnsRecord.owner_person_id.in_(owner_person_ids)) if search: q = q.where( or_( CmdbDnsRecord.name.ilike(f"%{_esc(search)}%"), CmdbDnsRecord.value.ilike(f"%{_esc(search)}%"), CmdbDnsRecord.zone_name.ilike(f"%{_esc(search)}%"), ) ) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 # Order newest-first: a record's ``first_seen_at`` is stamped # ``func.now()`` on insert (admin direct-create AND cron # discovery both go through it), and the cron's ON DUPLICATE # KEY UPDATE deliberately doesn't touch it — so this column is # a stable "when did this row first appear" anchor. User # feedback 2026-05-29: brand-new records should float to the # top of the list. ``(zone_name, name)`` becomes the # tie-breaker so same-second imports stay readable. q = q.order_by( CmdbDnsRecord.first_seen_at.desc(), CmdbDnsRecord.zone_name, CmdbDnsRecord.name, ).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def resolve_dns_integration_for_zone( db: AsyncSession, zone_name: str, ) -> tuple[str, str, str] | None: """Resolve the authoritative DNS provider for ``zone_name``. Returns ``(integration_id, integration_name, provider)`` of the integration that owns the *most* non-deleted DNS records in this zone, or ``None`` if no record currently exists. Rationale (prod incident 2026-06-01): ``cmdb_domains.integration_id`` is the **registrar** (populated by ``CloudProvider.list_domains`` which calls registrar APIs — ``route53domains:ListDomains``, Tencent 域名注册, etc.) — NOT the authoritative DNS provider. Conflating the two led the admin picker to show "AliCloud (海外)" for ``tikv.io`` (a UCloud-registered, AWS-Route53-hosted zone) purely because UCloud was the registrar in ``cmdb_domains`` and the FE fallback then picked the first capability-DNS-enabled integration. The truth-source for "who hosts DNS for this zone" is ``cmdb_dns_records.integration_id``, populated by the sync's ``_sync_dns_records`` which calls authoritative DNS APIs (Route53 ``list_hosted_zones`` / DNSPod ``DescribeDomainList`` / Aliyun DNS ``DescribeDomains`` / UCloud ``UdnrDomainDNSQuery``). Tie-break: highest record count wins, ORDER BY integration_id asc as a deterministic secondary so the displayed winner does not flip between page refreshes when two tenants happen to carry equal record counts. Scope: a stale historical record on a **disabled** tenant must NOT silently become the routing target — same active-tenant gate every other cloud-resource list applies via ``_active_cloud_integration_ids``. """ from sqlalchemy import func as sa_func active = await _active_cloud_integration_ids(db) if not active: return None q = ( select( CmdbDnsRecord.integration_id, sa_func.count(CmdbDnsRecord.id).label("rc"), ) .where( CmdbDnsRecord.deleted_at.is_(None), CmdbDnsRecord.zone_name == zone_name, CmdbDnsRecord.integration_id.in_(active), ) .group_by(CmdbDnsRecord.integration_id) # Deterministic tie-break: same input → same winner across # calls. Without this, two tenants with equal record counts # would let MySQL pick whichever the storage engine happened # to surface first, and the FE "✓ 将使用账号" banner could # flip between page refreshes — exactly the kind of silent # ambiguity this PR set out to eliminate. .order_by( sa_func.count(CmdbDnsRecord.id).desc(), CmdbDnsRecord.integration_id.asc(), ) .limit(1) ) winner = (await db.execute(q)).first() if winner is None: return None integ_id = str(winner[0]) # Filter the metadata lookup on enabled=True too — defense in # depth even though ``active`` already excludes disabled # tenants. If a tenant flipped to disabled between the GROUP BY # and this lookup, we'd rather return None than show stale # metadata for a disabled tenant as the routing target. integ = await db.scalar( select(Integration).where( Integration.id == integ_id, Integration.enabled.is_(True), ), ) if integ is None: return None return (integ_id, integ.name, integ.provider) @staticmethod async def list_dns_zones( db: AsyncSession, integration_id: str | None = None, ) -> list[dict[str, Any]]: """Return zones for the admin DNS picker with resolved provider. Each entry is:: { "zone_name": "pingcap.net", "integration_id": "21832517-...", # may be None "integration_name": "腾讯云 DNS (DNSPod)", # may be None "provider": "tencent", # may be None "record_count": 336, # may be None } UNION of two sources (deduplicated by zone_name): 1. ``cmdb_dns_records`` — authoritative DNS provider data. For each zone, GROUP BY integration_id and pick the integration with the most records (see ``resolve_dns_integration_for_zone``). 2. ``cmdb_domains`` where the zone is admin-curated. These rows reflect the **registrar** (populated by registrar APIs), not the DNS provider, so the entry returns ``integration_id=None`` to signal "no synced DNS provider manages this zone yet — admin must pick one explicitly". Source-2 zones already present in Source 1 are skipped (Source 1's resolved provider wins). Both sources are scoped to enabled integrations via ``_active_cloud_integration_ids`` unless the caller pins ``integration_id`` (admin auditing path). Note: this method serves the admin/auditor path (``GET /cmdb/dns-zones``). The employee self-service path uses the separate ``/cmdb/domains/selectable-for-dns`` endpoint — also corrected in this PR to return the resolved DNS provider instead of the registrar. """ from sqlalchemy import distinct from sqlalchemy import func as sa_func if integration_id: integ_filter_records = CmdbDnsRecord.integration_id == integration_id integ_filter_domains = CmdbDomain.integration_id == integration_id else: active = await _active_cloud_integration_ids(db) if not active: return [] integ_filter_records = CmdbDnsRecord.integration_id.in_(active) integ_filter_domains = CmdbDomain.integration_id.in_(active) # Source 1: zones with at least one existing record, with # resolved DNS provider (tie-broken by record count). # SELECT zone_name, integration_id, COUNT(*) FROM cmdb_dns_records # GROUP BY zone_name, integration_id; then in-memory pick the # max-count integration per zone. Avoids a per-zone N+1 round-trip. records_q = ( select( CmdbDnsRecord.zone_name, CmdbDnsRecord.integration_id, sa_func.count(CmdbDnsRecord.id).label("rc"), ) .where( CmdbDnsRecord.deleted_at.is_(None), integ_filter_records, ) .group_by(CmdbDnsRecord.zone_name, CmdbDnsRecord.integration_id) ) # Sort the raw rows by (rc desc, integration_id asc) before # the in-memory reduce so ties resolve to the same winner # every call. Without this the dict-update order is whatever # MySQL's storage engine returned — usually stable, but not # *guaranteed* stable, and the resulting FE banner ("✓ 将使用 # 账号: X") would flip across refreshes for equal-count # zones. Same determinism guarantee that # ``resolve_dns_integration_for_zone`` makes. records_rows = sorted( (await db.execute(records_q)).all(), key=lambda r: (-int(r[2]), str(r[1])), ) zone_to_winner: dict[str, tuple[str, int]] = {} # zone → (integ_id, rc) for zone, integ, rc in records_rows: # First write wins because we pre-sorted. if zone not in zone_to_winner: zone_to_winner[zone] = (str(integ), int(rc)) # Resolve integration metadata in one shot. integration_ids = {w[0] for w in zone_to_winner.values()} integ_meta: dict[str, tuple[str, str]] = {} # id → (name, provider) if integration_ids: rows = ( await db.execute( select( Integration.id, Integration.name, Integration.provider, ).where(Integration.id.in_(integration_ids)), ) ).all() integ_meta = {str(r[0]): (r[1], r[2]) for r in rows} # Source 2: admin-curated zones (registrar metadata). Surfaced # so admins can still see brand-new zones in the picker even # before the DNS provider sync has run — but with explicit # ``integration_id=None`` so the FE knows it has to ask. domains_q = select(distinct(CmdbDomain.domain_name)).where( CmdbDomain.deleted_at.is_(None), integ_filter_domains, ) registrar_zones = set( (await db.execute(domains_q)).scalars().all(), ) results: list[dict[str, Any]] = [] for zone, (integ_id, rc) in zone_to_winner.items(): name, provider = integ_meta.get(integ_id, ("", "")) results.append({ "zone_name": zone, "integration_id": integ_id, "integration_name": name or None, "provider": provider or None, "record_count": rc, }) # Source-2 zones not yet in Source 1. for zone in registrar_zones - zone_to_winner.keys(): results.append({ "zone_name": zone, "integration_id": None, "integration_name": None, "provider": None, "record_count": 0, }) results.sort(key=lambda r: r["zone_name"]) return results @staticmethod async def create_dns_record_in_db( db: AsyncSession, integration_id: str, zone_name: str, name: str, record_type: str, value: str, ttl: int, line: str | None, provider_record_id: str, owner_person_id: str | None = None, last_owner_review_at: datetime | None = None, actor_person_id: str | None = None, ) -> CmdbDnsRecord: # Round-28 (2026-05-18): the previous internal ``await db.commit()`` # split the row insert from the caller's owner_history seed write # into two transactions. If the second commit failed or the # process died between them, the cloud-side record + Helix row # would land with NULL history. The helper now only flushes and # the caller (router or _execute_create) commits AFTER calling # ``record_owner_change`` so the seed row is atomic with the # insert. ``actor_person_id`` is stamped as the modifier so the # UI's "修改人" reflects who originated the create. record = CmdbDnsRecord( integration_id=integration_id, zone_name=zone_name, record_id=provider_record_id, name=name, record_type=record_type, value=value, ttl=ttl, line=line, status="enable", owner_person_id=owner_person_id, last_owner_review_at=last_owner_review_at, ) if actor_person_id is not None: CmdbService._stamp_modifier(record, actor_person_id) db.add(record) await db.flush() await db.refresh(record) return record @staticmethod async def update_dns_record_in_db( db: AsyncSession, record_id: str, data, # DnsRecordUpdate *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> CmdbDnsRecord | None: from app.domains.cmdb.owner_history import ( CHANGE_REASON_ADMIN_PATCH, RESOURCE_TYPE_DNS_RECORD, record_owner_change, ) # Round-28: SELECT FOR UPDATE — same rationale as annotate_* # paths. Two concurrent PATCHes on the same DNS record would # otherwise race the owner-history write and lose the # intermediate transition. result = await db.execute( select(CmdbDnsRecord).where( CmdbDnsRecord.id == uuid.UUID(record_id), CmdbDnsRecord.deleted_at.is_(None), ).with_for_update() ) record = result.scalar_one_or_none() if not record: return None fields = data.model_fields_set previous_owner = record.owner_person_id owner_touched = False new_owner: str | None = None changed = False # NB: ``name`` (host record / SubDomain) was removed from the # DnsRecordUpdate schema on 2026-05-28 — host is immutable on # update; the schema (extra='forbid') 422s any caller that tries # to send one. Rename happens via delete-then-create only. if data.value is not None: record.value = data.value changed = True if data.ttl is not None: record.ttl = data.ttl changed = True if data.line is not None: record.line = data.line changed = True if "remark" in fields: record.remark = data.remark or None changed = True if "owner_person_id" in fields: uid = data.owner_person_id if uid is None: record.owner_person_id = None new_owner = None else: # Verify the target exists AND is an active employee. # Mirror the guard in ``_resolve_owner_input`` — handing # the record to a leaver_pending / left person silently # orphans it once offboarding T+0 fires. from app.core.exceptions import ValidationError from app.identity.person_models import IdentityPerson person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == str(uid), IdentityPerson.deleted_at.is_(None), ), ) if person is None: raise ValidationError( f"owner_person_id {uid!s} does not match any active person.", ) if person.employment_status != "active": raise ValidationError( f"owner_person_id {uid!s} resolves to " f"{person.display_name or person.work_email or 'person'} " f"(employment_status={person.employment_status!r}); " "owner must be an active employee." ) record.owner_person_id = str(person.id) new_owner = str(person.id) owner_touched = True changed = True # Round-30: clear recertify dedup so the new owner gets the # next nag promptly. Without this the previous owner's # ``last_recertify_nag_at`` keeps the record inside the dedup # window (default 30d) and the cron silently skips the new # owner — they go an entire quarter without a confirm prompt # for records they just inherited. ``last_owner_review_at`` # is reset to None so the next quarterly cycle counts from # the transfer date, not the previous owner's stale review. record.last_recertify_nag_at = None record.last_owner_review_at = None if changed: CmdbService._stamp_modifier(record, actor_person_id) if owner_touched: await record_owner_change( db, resource_type=RESOURCE_TYPE_DNS_RECORD, resource_id=str(record.id), previous_owner_person_id=previous_owner, new_owner_person_id=new_owner, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_ADMIN_PATCH, trace_id=trace_id, ) await db.commit() await db.refresh(record) return record @staticmethod async def set_dns_record_status_in_db( db: AsyncSession, record_id: str, *, enable: bool, actor_person_id: str | None, ) -> CmdbDnsRecord | None: """DB-side companion to ``provider.set_dns_record_status``. Caller is responsible for calling the cloud API first; this only flips the row's ``status`` field + stamps the modifier. Idempotent — if the row is already in the desired state we still write last_modified to record the intent (the cloud may have been re-flipped externally and we're catching up). """ result = await db.execute( select(CmdbDnsRecord).where( CmdbDnsRecord.id == uuid.UUID(record_id), CmdbDnsRecord.deleted_at.is_(None), ) ) record = result.scalar_one_or_none() if not record: return None record.status = "enable" if enable else "disable" CmdbService._stamp_modifier(record, actor_person_id) await db.commit() await db.refresh(record) return record @staticmethod async def delete_dns_record_in_db( db: AsyncSession, record_id: str, *, actor_person_id: str | None = None, ) -> CmdbDnsRecord | None: result = await db.execute( select(CmdbDnsRecord).where( CmdbDnsRecord.id == uuid.UUID(record_id), CmdbDnsRecord.deleted_at.is_(None), ) ) record = result.scalar_one_or_none() if not record: return None record.deleted_at = datetime.now(UTC) # Stamp modifier on soft-delete so the audit + UI can attribute # the deletion to the actor (otherwise the audit_log is the only # source of "who deleted this" — useful but harder to surface # on the deleted-rows view than a column on the row itself). CmdbService._stamp_modifier(record, actor_person_id) # Round-28: flush-only; caller commits AFTER optionally writing # the owner_history TERMINAL row so the delete + history land # in the same transaction. Without this they split, and a # crash between commits leaves a deleted-record with no # ownership-timeline endpoint entry. await db.flush() return record @staticmethod async def list_load_balancers( db: AsyncSession, integration_id: str | None = None, region: str | None = None, network_type: str | None = None, search: str | None = None, owner_person_id: str | None = None, owner_person_ids: list[str] | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbLoadBalancer], int]: from sqlalchemy import func, or_ q = select(CmdbLoadBalancer).where(CmdbLoadBalancer.deleted_at.is_(None)) if integration_id: q = q.where(CmdbLoadBalancer.integration_id == integration_id) else: active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbLoadBalancer.integration_id.in_(active)) if region: q = q.where(CmdbLoadBalancer.region == region) if network_type: q = q.where(CmdbLoadBalancer.network_type == network_type) # Migration 173: LB carries owner_person_id now, so manager-scope # filters apply the same way as servers / domains / dns / cdn. if owner_person_id: q = q.where(CmdbLoadBalancer.owner_person_id == owner_person_id) if owner_person_ids: q = q.where(CmdbLoadBalancer.owner_person_id.in_(owner_person_ids)) if search: q = q.where( or_( CmdbLoadBalancer.name.ilike(f"%{_esc(search)}%"), CmdbLoadBalancer.ip_address.ilike(f"%{_esc(search)}%"), ) ) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = q.order_by(CmdbLoadBalancer.name).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def list_cdn_domains( db: AsyncSession, integration_id: str | None = None, status: str | None = None, search: str | None = None, owner_person_id: str | None = None, owner_person_ids: list[str] | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbCdnDomain], int]: from sqlalchemy import func, or_ q = select(CmdbCdnDomain).where(CmdbCdnDomain.deleted_at.is_(None)) if integration_id: q = q.where(CmdbCdnDomain.integration_id == integration_id) else: active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbCdnDomain.integration_id.in_(active)) if status: q = q.where(CmdbCdnDomain.status == status) if search: q = q.where( or_( CmdbCdnDomain.domain_name.ilike(f"%{_esc(search)}%"), CmdbCdnDomain.cname.ilike(f"%{_esc(search)}%"), ) ) if owner_person_id: q = q.where(CmdbCdnDomain.owner_person_id == owner_person_id) elif owner_person_ids: q = q.where(CmdbCdnDomain.owner_person_id.in_(owner_person_ids)) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = q.order_by(CmdbCdnDomain.domain_name).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def _resolve_owner_input( db: AsyncSession, data: Any, ) -> tuple[bool, str | None, str | None]: """Resolve the owner inputs on an annotation payload. Returns ``(touched, owner_person_id, owner_raw)`` where: * ``touched`` — True if EITHER ``owner_person_id`` or ``owner_raw`` was present in the request body (caller flips the FK accordingly). False when the body omits both, so the existing owner stays put. * ``owner_person_id`` — the canonical FK value to write (may be None when the caller wants to clear the owner). * ``owner_raw`` — the raw string to keep on tables that still carry ``owner_raw`` (LB / CDN / object_storage). Pass-through of the input; not derived from the FK lookup. Resolution priority: 1. ``owner_person_id`` field set → use the UUID directly (after verifying the person exists). This is the canonical input from the UserSelect frontend picker. 2. ``owner_person_id`` not set but ``owner_raw`` set → resolve via ``_resolve_person_id_from_raw`` (email → display_name). 3. Neither set → caller skips owner update. """ from app.core.exceptions import ValidationError from app.identity.person_models import IdentityPerson fields = getattr(data, "model_fields_set", set()) touched_uuid = "owner_person_id" in fields touched_raw = "owner_raw" in fields if not (touched_uuid or touched_raw): return (False, None, None) # UUID wins when present. if touched_uuid: uid = getattr(data, "owner_person_id", None) if uid is None: return (True, None, getattr(data, "owner_raw", None) if touched_raw else None) uid_str = str(uid) person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == uid_str, IdentityPerson.deleted_at.is_(None), ), ) if person is None: raise ValidationError( f"owner_person_id {uid_str!r} does not match any active person.", ) # Block transfer to leavers / leaver_pending — the # offboarding scan has already (or is about to) walk that # person's owned resources, so handing them a fresh row # would silently orphan it once T+0 fires. Surface to the # admin with the actual employment_status so they can pick # an active recipient instead. if person.employment_status != "active": raise ValidationError( f"owner_person_id {uid_str!r} resolves to " f"{person.display_name or person.work_email or 'person'} " f"(employment_status={person.employment_status!r}); " "owner must be an active employee. The offboarding " "force-transfer will not pick up a resource owned by " "a leaver_pending / left person." ) raw_out = getattr(data, "owner_raw", None) if touched_raw else (person.work_email or person.display_name) return (True, str(person.id), raw_out) # owner_raw only. raw_val = getattr(data, "owner_raw", None) resolved = await CmdbService._resolve_person_id_from_raw(db, raw_val) return (True, resolved, raw_val) @staticmethod def _stamp_modifier(row: Any, actor_id: str | None) -> None: """Set ``last_modified_by_person_id`` + ``last_modified_at`` on a row if the model carries those columns. Idempotent; passing ``None`` for ``actor_id`` writes a system-attributed modification (used by automated owner-confirmation flows). We deliberately bypass ``updated_at`` — that column advances on every sync upsert and exposing it as "修改时间" in the UI would conflate human edits with sync activity.""" if hasattr(row, "last_modified_by_person_id"): row.last_modified_by_person_id = actor_id if hasattr(row, "last_modified_at"): row.last_modified_at = datetime.now(UTC) @staticmethod async def _resolve_person_id_from_raw( db: AsyncSession, raw: str | None, ) -> str | None: """Map a raw owner hint (email / display_name) to an identity_persons.id. Returns None for empty input. Resolution policy (matches migration 039's backfill): 1. Case-insensitive ``work_email`` match (unique by DB constraint) — return. 2. Exact ``display_name`` match — return only if exactly ONE active person matches; raise ValidationError on ambiguity so the admin sees "two Johns matched" instead of silently clearing the owner FK to NULL. 3. No match — return None (caller decides whether to error). """ from sqlalchemy import func as _func if raw is None or not str(raw).strip(): return None from app.core.exceptions import ValidationError from app.identity.person_models import IdentityPerson raw = str(raw).strip() # Email match. Look up active employees first; if zero hits, # check whether the email belongs to a leaver_pending / left # person and raise a loud ValidationError so the admin sees # "tried to assign to a leaver" rather than the field silently # being cleared. Pre-fix the active-only filter would silently # return None → caller wrote NULL owner — the EXACT silent- # orphan failure mode this PR is meant to prevent (CR pass 1 # HIGH). Mirrors the UUID-input path in ``_resolve_owner_input``. person = await db.scalar( select(IdentityPerson).where( _func.lower(IdentityPerson.work_email) == raw.lower(), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ), ) if person is not None: return str(person.id) # Email matched but person is non-active — raise. inactive_match = await db.scalar( select(IdentityPerson).where( _func.lower(IdentityPerson.work_email) == raw.lower(), IdentityPerson.deleted_at.is_(None), ), ) if inactive_match is not None: raise ValidationError( f"owner_raw {raw!r} resolves to " f"{inactive_match.display_name or inactive_match.work_email} " f"(employment_status={inactive_match.employment_status!r}); " "owner must be an active employee." ) # Unique display_name match. Multi-match used to silently # return None — admin sees "saved" and the owner field is # quietly cleared. Surface ambiguity loudly so admins know to # disambiguate by typing the email instead. rows = await db.execute( select(IdentityPerson).where( IdentityPerson.display_name == raw, IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ), ) people = list(rows.scalars().all()) if len(people) == 1: return str(people[0].id) if len(people) > 1: emails = ", ".join( p.work_email or "(no email)" for p in people[:5] ) raise ValidationError( f"Ambiguous owner: {len(people)} active people named " f"'{raw}' (e.g. {emails}). Use the work email to " "disambiguate." ) # Same display-name handling for non-active match — raise # rather than silently NULL the owner. inactive_dn = await db.scalar( select(IdentityPerson).where( IdentityPerson.display_name == raw, IdentityPerson.deleted_at.is_(None), ), ) if inactive_dn is not None: raise ValidationError( f"owner_raw {raw!r} resolves to " f"{inactive_dn.display_name or inactive_dn.work_email} " f"(employment_status={inactive_dn.employment_status!r}); " "owner must be an active employee." ) return None @staticmethod async def annotate_server( db: AsyncSession, server_id: str, data, # ServerAnnotation *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> CmdbServer | None: # Pre-launch P1 fix (Round 45 follow-up, CR-319): SELECT FOR # UPDATE so two concurrent admin PATCHes serialise — same # guard the sibling annotate_domain / annotate_load_balancer # / annotate_cdn_domain already use. result = await db.execute( select(CmdbServer).where( CmdbServer.id == uuid.UUID(server_id), CmdbServer.deleted_at.is_(None), ).with_for_update() ) server = result.scalar_one_or_none() if not server: return None fields = data.model_fields_set previous_owner = server.owner_person_id touched_owner, owner_id, _raw = await CmdbService._resolve_owner_input(db, data) changed = False if touched_owner: server.owner_person_id = owner_id changed = True if "app" in fields: server.app = data.app or None changed = True if "remark" in fields: server.remark = data.remark or None changed = True if changed: CmdbService._stamp_modifier(server, actor_person_id) # Pre-launch P1 fix (Round 45 follow-up, CR-319): write a # per-server owner-history row when the owner is touched. The # Round-45 batch added cloud_server to the resource_type enum # but only wired it into the offboarding T+0 force-transfer. # The admin direct-PATCH path was the bigger surface and was # being silently skipped — every manual owner re-bind landed # only in the generic audit_logs (with `body.model_dump()` # but NO before-owner) which doesn't appear in the standard # owner-history admin view. if touched_owner and previous_owner != server.owner_person_id: from app.domains.cmdb.owner_history import ( # noqa: PLC0415 CHANGE_REASON_ADMIN_PATCH, RESOURCE_TYPE_CLOUD_SERVER, record_owner_change, ) await record_owner_change( db, resource_type=RESOURCE_TYPE_CLOUD_SERVER, resource_id=str(server.id), previous_owner_person_id=previous_owner, new_owner_person_id=server.owner_person_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_ADMIN_PATCH, trace_id=trace_id, context={ "label": server.name or str(server.id), "via": "annotate_server", }, ) await db.commit() await db.refresh(server) return server @staticmethod async def annotate_domain( db: AsyncSession, domain_id: str, data, # DomainAnnotation *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> CmdbDomain | None: from app.domains.cmdb.owner_history import ( CHANGE_REASON_ADMIN_PATCH, RESOURCE_TYPE_DOMAIN, record_owner_change, ) # Round-28: SELECT FOR UPDATE so two concurrent admin PATCHes # serialize. Without the lock both could read the same previous # owner, both write owner_history rows claiming "X→A" and "X→B" # in parallel — the intermediate transition is then absent from # the timeline. Same fix applied to annotate_load_balancer and # annotate_cdn_domain below. result = await db.execute( select(CmdbDomain).where( CmdbDomain.id == uuid.UUID(domain_id), CmdbDomain.deleted_at.is_(None), ).with_for_update() ) domain = result.scalar_one_or_none() if not domain: return None fields = data.model_fields_set previous_owner = domain.owner_person_id touched_owner, owner_id, _raw = await CmdbService._resolve_owner_input(db, data) changed = False if touched_owner: domain.owner_person_id = owner_id changed = True if "icp_filing_no" in fields: domain.icp_filing_no = data.icp_filing_no or None changed = True if "environment" in fields: domain.environment = data.environment or None changed = True if data.allow_dns_request is not None: domain.allow_dns_request = data.allow_dns_request changed = True if "grade" in fields: domain.grade = data.grade or None changed = True if "possessor" in fields: domain.possessor = data.possessor or None changed = True if "remark" in fields: domain.remark = data.remark or None changed = True # Migration 218: admin override of sync-owned columns. Touching # ANY of {registrar, expiry_date, auto_renew} flips data_source # to ``manual`` so a future driver upsert still wins (it re- # stamps ``synced``) and so the UI can render the override # provenance. Tracked separately from the modifier stamp because # this is a stronger signal than "admin edited the row" — it # means the row's freshness now depends on an admin, not a sync. sync_owned_touched = False if "registrar" in fields: # Empty string normalises to NULL — sync drivers write either # a real slug or skip the column; never an empty string. domain.registrar = data.registrar or None sync_owned_touched = True if "expiry_date" in fields: domain.expiry_date = data.expiry_date sync_owned_touched = True if "auto_renew" in fields: # ``bool | None`` Pydantic field — only override when the # admin actually sent a value. ``None`` here means the JSON # body carried a literal ``null``; reject explicitly so the # NOT NULL column can never be coerced to ``False`` by a # buggy client that sends ``{"auto_renew": null}`` thinking # it means "leave unchanged" (the omit-vs-null contract: # omit = no change, null = REJECTED for this field, value # = set). ``ValidationError(message)`` lands a 422 via the # core exception handler — visible at the HTTP boundary # and through MCP's ``tool_runtime`` envelope. if data.auto_renew is None: # Module-level import would normally live at the top; # kept local here because ``ValidationError`` is only # referenced via the existing ``_resolve_owner_input`` # local-import pattern in this file, and the other # call-sites (lines 668, 678, 727, …) all do the same # — moving just this one to top creates inconsistency. from app.core.exceptions import ValidationError raise ValidationError( "auto_renew cannot be null — send true / false, " "or omit the field to leave it unchanged.", ) domain.auto_renew = data.auto_renew sync_owned_touched = True if sync_owned_touched: domain.data_source = "manual" changed = True if changed: CmdbService._stamp_modifier(domain, actor_person_id) # Emit owner-history row INSIDE the same tx so a rollback rolls # back both the mutation and the history line — never lie. if touched_owner: await record_owner_change( db, resource_type=RESOURCE_TYPE_DOMAIN, resource_id=str(domain.id), previous_owner_person_id=previous_owner, new_owner_person_id=owner_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_ADMIN_PATCH, trace_id=trace_id, ) await db.commit() await db.refresh(domain) return domain @staticmethod async def annotate_load_balancer( db: AsyncSession, lb_id: str, data, # LoadBalancerAnnotation *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> CmdbLoadBalancer | None: from app.domains.cmdb.owner_history import ( CHANGE_REASON_ADMIN_PATCH, RESOURCE_TYPE_LOAD_BALANCER, record_owner_change, ) result = await db.execute( select(CmdbLoadBalancer).where( CmdbLoadBalancer.id == uuid.UUID(lb_id), CmdbLoadBalancer.deleted_at.is_(None), ).with_for_update() ) lb = result.scalar_one_or_none() if not lb: return None fields = data.model_fields_set previous_owner = lb.owner_person_id touched_owner, owner_id, raw = await CmdbService._resolve_owner_input(db, data) changed = False if touched_owner: lb.owner_person_id = owner_id if raw is not None: lb.owner_raw = raw changed = True if "app" in fields: lb.app = data.app or None changed = True if "remark" in fields: lb.remark = data.remark or None changed = True if changed: CmdbService._stamp_modifier(lb, actor_person_id) if touched_owner: await record_owner_change( db, resource_type=RESOURCE_TYPE_LOAD_BALANCER, resource_id=str(lb.id), previous_owner_person_id=previous_owner, new_owner_person_id=owner_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_ADMIN_PATCH, trace_id=trace_id, ) await db.commit() await db.refresh(lb) return lb @staticmethod async def annotate_cdn_domain( db: AsyncSession, cdn_id: str, data, # CdnDomainAnnotation *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> CmdbCdnDomain | None: from app.domains.cmdb.owner_history import ( CHANGE_REASON_ADMIN_PATCH, RESOURCE_TYPE_CDN_DOMAIN, record_owner_change, ) result = await db.execute( select(CmdbCdnDomain).where( CmdbCdnDomain.id == uuid.UUID(cdn_id), CmdbCdnDomain.deleted_at.is_(None), ).with_for_update() ) cdn = result.scalar_one_or_none() if not cdn: return None fields = data.model_fields_set previous_owner = cdn.owner_person_id touched_owner, owner_id, raw = await CmdbService._resolve_owner_input(db, data) changed = False if touched_owner: cdn.owner_person_id = owner_id if raw is not None: cdn.owner_raw = raw changed = True if "cert_id" in fields: cdn.cert_id = data.cert_id or None changed = True if "remark" in fields: cdn.remark = data.remark or None changed = True if changed: CmdbService._stamp_modifier(cdn, actor_person_id) if touched_owner: await record_owner_change( db, resource_type=RESOURCE_TYPE_CDN_DOMAIN, resource_id=str(cdn.id), previous_owner_person_id=previous_owner, new_owner_person_id=owner_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_ADMIN_PATCH, trace_id=trace_id, ) await db.commit() await db.refresh(cdn) return cdn # ─── Batch annotate ──────────────────────────────────────────── @staticmethod async def batch_annotate_servers( db: AsyncSession, server_ids: list[str], data, # ServerAnnotation *, actor_person_id: str | None = None, ) -> dict[str, Any]: """Apply the same annotation to every server in ``server_ids``. Returns a summary dict ``{requested, updated, not_found}``. Each row goes through the same path as ``annotate_server`` so owner resolution + modifier stamping stay consistent. We commit once at the end for throughput — partial failure cannot happen because the only DB action per row is field assignment (no cloud call). """ # Pre-launch P1 fix (Round 45 follow-up, CR-320): per-row # owner-history writes. Same family as annotate_server's # CR-319 fix — without this, a batch reassign to dozens of # servers leaves only the generic batch audit row with no # per-server prior-owner snapshot. Mirrors the sibling # batch_annotate_load_balancers below. from app.domains.cmdb.owner_history import ( # noqa: PLC0415 CHANGE_REASON_BATCH_OWNER, RESOURCE_TYPE_CLOUD_SERVER, record_owner_change, ) if not server_ids: return {"requested": 0, "updated": 0, "not_found": 0} fields = data.model_fields_set touched_owner, owner_id, _raw = await CmdbService._resolve_owner_input(db, data) rows = await db.execute( select(CmdbServer).where( CmdbServer.id.in_([uuid.UUID(s) for s in server_ids]), CmdbServer.deleted_at.is_(None), ).with_for_update(), ) updated = 0 seen: set[str] = set() changed_owner_rows: list[tuple[str, str | None, str | None, str]] = [] for server in rows.scalars().all(): seen.add(str(server.id)) previous_owner = server.owner_person_id if touched_owner: server.owner_person_id = owner_id if previous_owner != server.owner_person_id: changed_owner_rows.append(( str(server.id), previous_owner, server.owner_person_id, server.name or str(server.id), )) if "app" in fields: server.app = data.app or None if "remark" in fields: server.remark = data.remark or None CmdbService._stamp_modifier(server, actor_person_id) updated += 1 for rid, prev, new, label in changed_owner_rows: await record_owner_change( db, resource_type=RESOURCE_TYPE_CLOUD_SERVER, resource_id=rid, previous_owner_person_id=prev, new_owner_person_id=new, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_BATCH_OWNER, trace_id=None, context={"label": label, "via": "batch_annotate_servers"}, ) await db.commit() return { "requested": len(server_ids), "updated": updated, "not_found": len(server_ids) - len(seen), "owner_changed_count": len(changed_owner_rows), } @staticmethod async def batch_annotate_load_balancers( db: AsyncSession, lb_ids: list[str], data, # LoadBalancerAnnotation *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> dict[str, Any]: from app.domains.cmdb.owner_history import ( CHANGE_REASON_BATCH_OWNER, RESOURCE_TYPE_LOAD_BALANCER, record_owner_change, ) if not lb_ids: return {"requested": 0, "updated": 0, "not_found": 0} fields = data.model_fields_set touched_owner, owner_id, raw = await CmdbService._resolve_owner_input(db, data) # Round-28 follow-up: SELECT FOR UPDATE so the batch serializes # with concurrent single-row ``annotate_load_balancer`` PATCHes. # Without the lock, the batch reads stale previous_owner from # its Python snapshot while a parallel annotate already # committed an intermediate transition — the timeline loses # the middle hop. rows = await db.execute( select(CmdbLoadBalancer).where( CmdbLoadBalancer.id.in_([uuid.UUID(s) for s in lb_ids]), CmdbLoadBalancer.deleted_at.is_(None), ).with_for_update(), ) # Round-28: prefill new_label / actor_label ONCE so per-row # record_owner_change skips its IdentityPerson lookup. See # batch_update_dns_record_owner for the same pattern. from app.domains.cmdb.owner_history import _resolve_label new_label = await _resolve_label(db, owner_id) if touched_owner else None actor_label = await _resolve_label(db, actor_person_id) updated = 0 seen: set[str] = set() # Capture previous owners BEFORE mutation so per-row history is # accurate. Without this snapshot the loop body would compare # ``previous == new`` against the just-mutated value and # record_owner_change's no-op short-circuit would silently # drop every history row. for lb in rows.scalars().all(): previous_owner = lb.owner_person_id seen.add(str(lb.id)) if touched_owner: lb.owner_person_id = owner_id if raw is not None: lb.owner_raw = raw if "app" in fields: lb.app = data.app or None if "remark" in fields: lb.remark = data.remark or None CmdbService._stamp_modifier(lb, actor_person_id) if touched_owner: await record_owner_change( db, resource_type=RESOURCE_TYPE_LOAD_BALANCER, resource_id=str(lb.id), previous_owner_person_id=previous_owner, new_owner_person_id=owner_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_BATCH_OWNER, trace_id=trace_id, context={ "new_label": new_label, "actor_label": actor_label, }, ) updated += 1 await db.commit() return { "requested": len(lb_ids), "updated": updated, "not_found": len(lb_ids) - len(seen), } # ─── Recycle bin (list soft-deleted + restore) ──────────────── # # 2026-05-28 — added in response to the legacy-CMDB-migration # → UCloud-sync collision incident: 116 UCloud servers were # silently soft-deleted by a single sweep on 2026-04-24, the user # had no admin surface to find or restore them. The Alembic 212 # rescue brings the swept rows back; these endpoints give admins # a permanent escape hatch for future drift / accidents. # # Conservative scope: only the 5 resource types with admin-curated # fields (owner_person_id, app, remark, owner_raw) — servers, # domains, dns_records, load_balancers, cdn_domains. Object storage # / IAM users / cost_facts are sync-managed without admin # annotation, so a recycle-bin there has no concrete "preserve my # owner" use case yet. @staticmethod async def _list_deleted( db: AsyncSession, *, model, integration_id: str | None, search: str | None, like_columns: list, json_array_columns: list[str] | None, page: int, page_size: int, ) -> tuple[list, int]: """Generic ``deleted_at IS NOT NULL`` paginator for CMDB resources. ``like_columns`` are model column objects to LIKE-search; ``json_array_columns`` are model column NAMES (strings, because ``JSON_SEARCH`` is rendered as raw SQL) whose JSON-array values should be substring-searched. Empty / None either is fine. """ q = select(model).where(model.deleted_at.is_not(None)) if integration_id: q = q.where(model.integration_id == integration_id) if search and search.strip(): pat = f"%{_esc(search.strip())}%" clauses = [c.ilike(pat) for c in like_columns] for col_name in (json_array_columns or []): clauses.append( text( f"JSON_SEARCH({model.__tablename__}.{col_name}, " "'one', :ip_pat) IS NOT NULL" ).bindparams(ip_pat=pat), ) q = q.where(or_(*clauses)) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = ( q.order_by(model.deleted_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def list_deleted_servers( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbServer], int]: return await CmdbService._list_deleted( db, model=CmdbServer, integration_id=integration_id, search=search, like_columns=[CmdbServer.name, CmdbServer.instance_id], json_array_columns=["public_ips", "private_ips"], page=page, page_size=page_size, ) @staticmethod async def list_deleted_domains( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbDomain], int]: return await CmdbService._list_deleted( db, model=CmdbDomain, integration_id=integration_id, search=search, like_columns=[CmdbDomain.domain_name, CmdbDomain.registrar], json_array_columns=None, page=page, page_size=page_size, ) @staticmethod async def list_deleted_dns_records( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbDnsRecord], int]: return await CmdbService._list_deleted( db, model=CmdbDnsRecord, integration_id=integration_id, search=search, like_columns=[ CmdbDnsRecord.name, CmdbDnsRecord.value, CmdbDnsRecord.zone_name, ], json_array_columns=None, page=page, page_size=page_size, ) @staticmethod async def list_deleted_load_balancers( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbLoadBalancer], int]: return await CmdbService._list_deleted( db, model=CmdbLoadBalancer, integration_id=integration_id, search=search, like_columns=[ CmdbLoadBalancer.name, CmdbLoadBalancer.lb_id, CmdbLoadBalancer.ip_address, ], json_array_columns=None, page=page, page_size=page_size, ) @staticmethod async def list_deleted_cdn_domains( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbCdnDomain], int]: return await CmdbService._list_deleted( db, model=CmdbCdnDomain, integration_id=integration_id, search=search, like_columns=[CmdbCdnDomain.domain_name, CmdbCdnDomain.cname], json_array_columns=None, page=page, page_size=page_size, ) @staticmethod async def list_deleted_certificates( db: AsyncSession, *, integration_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbCertificate], int]: return await CmdbService._list_deleted( db, model=CmdbCertificate, integration_id=integration_id, search=search, like_columns=[ CmdbCertificate.common_name, CmdbCertificate.serial_number, CmdbCertificate.fingerprint_sha256, ], # ``sans`` is a JSON array of additional hostnames — let admins # find a swept cert by any SAN, not just the common name. json_array_columns=["sans"], page=page, page_size=page_size, ) @staticmethod async def restore_resource( db: AsyncSession, *, model, resource_id: str, actor_person_id: str | None = None, ) -> tuple[object, dict[str, Any]] | None: """Restore a single soft-deleted CMDB row by clearing ``deleted_at``. Returns ``(refreshed_row, forensics)`` where ``forensics`` is a small dict the caller folds into its audit ``detail`` JSON so the restore can be reconciled later (``prior_deleted_at`` + ``prior_last_seen_sync_job_id`` — the two fields a future investigator needs to answer "was this restore reversing a real cloud-side decommission, or a sweep accident?"). Returns ``None`` if the row isn't found. Raises a 409 ``AppException`` if the row exists but is not soft-deleted (idempotency guard against double-click / stale recycle-bin list). Stamps modifier (``last_modified_by_person_id`` / ``last_modified_at``) so the restore is attributable in audit / owner-history tabs. Does NOT touch ``last_seen_sync_job_id`` — that field is sync-managed and a restored row genuinely was-not-seen by the most recent sync (otherwise it wouldn't be deleted). On the next sync, one of two things happens: * The driver returns the row → ``last_seen_sync_job_id`` updates, ``deleted_at`` stays NULL (upsert path's ``deleted_at=None`` clause), all good. * The driver doesn't return the row → the new sweep's ``last_seen_sync_job_id IS NOT NULL`` guard (added in this same PR) protects the row IF it was never seen by ANY sync; otherwise the old job_id stamps it as orphan again and admin must investigate (real decommission vs. sync scope misconfig). """ result = await db.execute( select(model).where(model.id == uuid.UUID(resource_id)), ) row = result.scalar_one_or_none() if row is None: return None if row.deleted_at is None: from app.core.exceptions import AppException # noqa: PLC0415 raise AppException( code="CMDB_RESOURCE_NOT_DELETED", message=( "Resource is not soft-deleted; nothing to restore." ), status_code=409, ) # Capture forensic snapshot BEFORE mutating so the caller can # write an audit row that names the prior state. Without this # the audit detail collapses to "restored=True" and a future # investigator can't tell when the row was tombstoned or which # sync's job_id last touched it. forensics: dict[str, Any] = { "prior_deleted_at": row.deleted_at.isoformat() if row.deleted_at else None, "prior_last_seen_sync_job_id": ( str(row.last_seen_sync_job_id) if getattr(row, "last_seen_sync_job_id", None) is not None else None ), } row.deleted_at = None CmdbService._stamp_modifier(row, actor_person_id) await db.commit() await db.refresh(row) return row, forensics @staticmethod async def batch_update_dns_record_owner( db: AsyncSession, record_ids: list[str], owner_person_id: str | None, *, actor_person_id: str | None = None, trace_id: str | None = None, ) -> dict[str, Any]: """Bulk-reassign DNS record owners. DB-side only (no cloud call — DNS providers don't know about Helix owner FKs).""" from app.domains.cmdb.owner_history import ( CHANGE_REASON_BATCH_OWNER, RESOURCE_TYPE_DNS_RECORD, record_owner_change, ) if not record_ids: return {"requested": 0, "updated": 0, "not_found": 0} # Validate the target owner exists (or is being cleared). if owner_person_id is not None: from app.core.exceptions import ValidationError from app.identity.person_models import IdentityPerson person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == owner_person_id, IdentityPerson.deleted_at.is_(None), ), ) if person is None: raise ValidationError( f"owner_person_id {owner_person_id!r} does not match any active person.", ) # Reject leaver_pending / left — offboarding's force-transfer # has already / is about to walk that person's owned resources; # batch-assigning fresh rows to them now would silently orphan # them once T+0 fires. Mirror the guard in ``_resolve_owner_input``. if person.employment_status != "active": raise ValidationError( f"owner_person_id {owner_person_id!r} resolves to " f"{person.display_name or person.work_email or 'person'} " f"(employment_status={person.employment_status!r}); " "owner must be an active employee." ) # Round-28 follow-up: SELECT FOR UPDATE so the batch serializes # with concurrent single-row ``update_dns_record_in_db`` / # ``annotate_*`` PATCHes. Same rationale as the LB batch above. rows = await db.execute( select(CmdbDnsRecord).where( CmdbDnsRecord.id.in_([uuid.UUID(s) for s in record_ids]), CmdbDnsRecord.deleted_at.is_(None), ).with_for_update(), ) # Round-28: prefill new_label / actor_label ONCE before the # batch loop. Without this, ``record_owner_change`` would do # 2 IdentityPerson lookups per row (new + actor) — for a 500- # record batch that's 1000 extra round-trips. The setdefault() # behaviour inside ``record_owner_change`` accepts the prefill # so we don't pay for re-resolution per call. ``previous_label`` # still varies per row (each record's previous owner may differ) # so it stays as the helper's lookup. from app.domains.cmdb.owner_history import _resolve_label new_label = await _resolve_label(db, owner_person_id) actor_label = await _resolve_label(db, actor_person_id) updated = 0 seen: set[str] = set() for rec in rows.scalars().all(): previous_owner = rec.owner_person_id seen.add(str(rec.id)) # Skip the no-op rows so we don't reset the previous owner's # legitimate review timestamp when an admin re-runs the same # batch on an idempotent retry. if str(previous_owner or "") != str(owner_person_id or ""): rec.owner_person_id = owner_person_id # Round-30: clear recertify dedup so the new owner is # nagged promptly. Mirrors update_dns_record_in_db. rec.last_recertify_nag_at = None rec.last_owner_review_at = None CmdbService._stamp_modifier(rec, actor_person_id) await record_owner_change( db, resource_type=RESOURCE_TYPE_DNS_RECORD, resource_id=str(rec.id), previous_owner_person_id=previous_owner, new_owner_person_id=owner_person_id, actor_person_id=actor_person_id, change_reason=CHANGE_REASON_BATCH_OWNER, trace_id=trace_id, context={ "new_label": new_label, "actor_label": actor_label, }, ) updated += 1 await db.commit() return { "requested": len(record_ids), "updated": updated, "not_found": len(record_ids) - len(seen), } @staticmethod async def transfer_owned_dns_records( db: AsyncSession, *, record_ids: list[str], new_owner_person_id: str, caller_person_id: str, caller_is_admin: bool, reason: str | None = None, trace_id: str | None = None, ) -> dict[str, int]: """Self-service DNS owner handoff. Caller MUST own every selected record (admin bypass via ``caller_is_admin``). Differs from ``batch_update_dns_record_owner`` in three ways: * ownership enforcement on the input set (non-admins cannot reach across the fleet); * ``new_owner_person_id`` is required — no "clear owner" mode, because the only audience for owner-clear is admin cleanup, which already has the legacy batch-owner endpoint; * caller is responsible for the new-owner notification; this method returns the list of transferred record ids so the router can dispatch. Pre-flight raises: * ``ValidationError`` if ``new_owner_person_id`` does not resolve to an active employee; * ``ForbiddenError`` (raised by the router by inspecting the ``not_owned`` count) when a non-admin includes any record they don't own. """ from app.core.exceptions import ValidationError from app.domains.cmdb.owner_history import ( CHANGE_REASON_SELF_SERVICE_TRANSFER, RESOURCE_TYPE_DNS_RECORD, _resolve_label, record_owner_change, ) from app.identity.person_models import IdentityPerson if not record_ids: return { "requested": 0, "transferred": 0, "not_found": 0, "not_owned": 0, "_transferred_ids": [], } # Refuse calls from a leaver_pending / left employee. Their JWT # may still be valid between announcement and T+0, but the # whole point of CHANGE_REASON_SELF_SERVICE_TRANSFER is "active # employee handed off"; if the caller is mid-offboarding the # cleanup belongs to CHANGE_REASON_OFFBOARDING_FORCE_TRANSFER # and should NOT be bypassed. Admins still get through because # an it_admin running cleanup on behalf of a leaver is a # legitimate path — but they should use the legacy batch-owner # endpoint, not this self-service surface. caller_person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == caller_person_id, IdentityPerson.deleted_at.is_(None), ), ) if ( caller_person is not None and not caller_is_admin and caller_person.employment_status != "active" ): raise ValidationError( "Self-service transfer is only available to active " "employees. Mid-offboarding handoffs go through the " "automated force-transfer at T+0.", ) # 1. Validate the target owner exists and is active. Mirror the # same guard the admin batch-owner endpoint uses so that a # leaver_pending recipient cannot accept a handoff that the # T+0 force-transfer would immediately re-orphan. person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == new_owner_person_id, IdentityPerson.deleted_at.is_(None), ), ) if person is None: raise ValidationError( f"new_owner_person_id {new_owner_person_id!r} does " "not match any active person.", ) if person.employment_status != "active": raise ValidationError( f"new_owner_person_id {new_owner_person_id!r} resolves " f"to {person.display_name or person.work_email or 'person'} " f"(employment_status={person.employment_status!r}); " "owner must be an active employee.", ) if str(person.id) == str(caller_person_id): raise ValidationError( "Cannot transfer records to yourself.", ) new_label = await _resolve_label(db, new_owner_person_id) actor_label = await _resolve_label(db, caller_person_id) # 2. Lock + load every record by id (soft-deleted excluded). rows = (await db.execute( select(CmdbDnsRecord).where( CmdbDnsRecord.id.in_( [uuid.UUID(s) for s in record_ids], ), CmdbDnsRecord.deleted_at.is_(None), ).with_for_update(), )).scalars().all() seen = {str(r.id) for r in rows} not_owned = [ r for r in rows if str(r.owner_person_id or "") != str(caller_person_id) ] if not caller_is_admin and not_owned: # Refuse the entire batch — atomic semantics: an employee # selecting 10 rows of which 1 is not theirs gets a clean # rejection, never a partial transfer that's painful to # undo. The router converts this into a 403 with the count. return { "requested": len(record_ids), "transferred": 0, "not_found": len(record_ids) - len(seen), "not_owned": len(not_owned), "_transferred_ids": [], } transferred_ids: list[str] = [] for rec in rows: previous_owner = rec.owner_person_id if str(previous_owner or "") == str(new_owner_person_id): # Idempotent — already owned by the target. Skip # writing owner_history + clearing the review timestamp # so a re-click on the same batch doesn't reset the # quarterly cycle. continue rec.owner_person_id = new_owner_person_id # Mirror the dedup-clear semantics of the existing # update_dns_record_in_db / batch_update_dns_record_owner # paths: the new owner should be in the next recertify # nag pass, not silently skipped because the previous # owner's nag stamp is still inside the dedup window. rec.last_recertify_nag_at = None rec.last_owner_review_at = None CmdbService._stamp_modifier(rec, caller_person_id) await record_owner_change( db, resource_type=RESOURCE_TYPE_DNS_RECORD, resource_id=str(rec.id), previous_owner_person_id=previous_owner, new_owner_person_id=new_owner_person_id, actor_person_id=caller_person_id, # Distinct reason from the admin batch path so audits # can split self-service handoffs vs admin cleanups. change_reason=CHANGE_REASON_SELF_SERVICE_TRANSFER, trace_id=trace_id, context={ "new_label": new_label, "actor_label": actor_label, "via": "transfer_owned_dns_records", "reason": reason, }, ) transferred_ids.append(str(rec.id)) await db.commit() return { "requested": len(record_ids), "transferred": len(transferred_ids), "not_found": len(record_ids) - len(seen), "not_owned": len(not_owned), "_transferred_ids": transferred_ids, } @staticmethod async def transfer_owned_servers( db: AsyncSession, *, server_ids: list[str], new_owner_person_id: str, caller_person_id: str, caller_is_admin: bool, reason: str | None = None, trace_id: str | None = None, ) -> dict[str, int]: """Self-service server owner handoff. Mirrors ``transfer_owned_dns_records`` over ``cmdb_servers``.""" from app.core.exceptions import ValidationError from app.domains.cmdb.owner_history import ( CHANGE_REASON_SELF_SERVICE_TRANSFER, RESOURCE_TYPE_CLOUD_SERVER, _resolve_label, record_owner_change, ) from app.identity.person_models import IdentityPerson # Refuse leaver_pending / left callers — see the parallel # comment in ``transfer_owned_dns_records``. caller_person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == caller_person_id, IdentityPerson.deleted_at.is_(None), ), ) if ( caller_person is not None and not caller_is_admin and caller_person.employment_status != "active" ): raise ValidationError( "Self-service transfer is only available to active " "employees. Mid-offboarding handoffs go through the " "automated force-transfer at T+0.", ) if not server_ids: return { "requested": 0, "transferred": 0, "not_found": 0, "not_owned": 0, "unconfirmed": 0, "_transferred_ids": [], } person = await db.scalar( select(IdentityPerson).where( IdentityPerson.id == new_owner_person_id, IdentityPerson.deleted_at.is_(None), ), ) if person is None: raise ValidationError( f"new_owner_person_id {new_owner_person_id!r} does " "not match any active person.", ) if person.employment_status != "active": raise ValidationError( f"new_owner_person_id {new_owner_person_id!r} resolves " f"to {person.display_name or person.work_email or 'person'} " f"(employment_status={person.employment_status!r}); " "owner must be an active employee.", ) if str(person.id) == str(caller_person_id): raise ValidationError( "Cannot transfer servers to yourself.", ) new_label = await _resolve_label(db, new_owner_person_id) actor_label = await _resolve_label(db, caller_person_id) rows = (await db.execute( select(CmdbServer).where( CmdbServer.id.in_( [uuid.UUID(s) for s in server_ids], ), CmdbServer.deleted_at.is_(None), ).with_for_update(), )).scalars().all() seen = {str(r.id) for r in rows} # Never-confirmed rows (``last_seen_sync_job_id IS NULL``) are # sync-managed records that no live sync has ever vouched for and # that ``_safe_sweep`` deliberately refuses to touch (it only sweeps # rows it has actually seen — sync.py). Such a row may already be # gone from the cloud and linger only because a connector wasn't # scanning its project — exactly the 2026-07 Oncall_bot incident: a # released UCloud host survived as a stale row and was handed off # 37 minutes before the reconcile soft-deleted it, leaving the # recipient a dangling "you are now the owner" email pointing at a # resource that no longer existed. Refuse to transfer these for # everyone (admins included) — they must be reconciled / cleaned up, # not passed to a new owner. (All live rows are sync-confirmed and # there is no manual server-create path, so this never rejects a # legitimately-owned server.) confirmed = [r for r in rows if r.last_seen_sync_job_id is not None] unconfirmed = len(rows) - len(confirmed) not_owned = [ r for r in confirmed if str(r.owner_person_id or "") != str(caller_person_id) ] if not caller_is_admin and not_owned: return { "requested": len(server_ids), "transferred": 0, "not_found": len(server_ids) - len(seen), "not_owned": len(not_owned), "unconfirmed": unconfirmed, "_transferred_ids": [], } transferred_ids: list[str] = [] for server in confirmed: previous_owner = server.owner_person_id if str(previous_owner or "") == str(new_owner_person_id): continue server.owner_person_id = new_owner_person_id # Mirror the DNS path's recertify dedup-clear so the new # owner enters the next semiannual confirmation cycle. Without # this clear, the new owner inherits the previous owner's # stamp and the half-yearly cron skips them for up to 180d # (see ``cmdb_servers.last_owner_confirmed_at`` cmdb model). server.last_owner_confirmed_at = None CmdbService._stamp_modifier(server, caller_person_id) await record_owner_change( db, resource_type=RESOURCE_TYPE_CLOUD_SERVER, resource_id=str(server.id), previous_owner_person_id=previous_owner, new_owner_person_id=new_owner_person_id, actor_person_id=caller_person_id, change_reason=CHANGE_REASON_SELF_SERVICE_TRANSFER, trace_id=trace_id, context={ "label": server.name or str(server.id), "new_label": new_label, "actor_label": actor_label, "via": "transfer_owned_servers", "reason": reason, }, ) transferred_ids.append(str(server.id)) await db.commit() return { "requested": len(server_ids), "transferred": len(transferred_ids), "not_found": len(server_ids) - len(seen), "not_owned": len(not_owned), "unconfirmed": unconfirmed, "_transferred_ids": transferred_ids, } @staticmethod async def list_object_storage( db: AsyncSession, integration_id: str | None = None, region: str | None = None, status: str | None = None, search: str | None = None, owner_person_id: str | None = None, unowned: bool = False, page: int = 1, page_size: int = 50, ) -> tuple[list[CmdbObjectStorage], int]: from sqlalchemy import func q = select(CmdbObjectStorage).where(CmdbObjectStorage.deleted_at.is_(None)) if integration_id: q = q.where(CmdbObjectStorage.integration_id == integration_id) else: active = await _active_cloud_integration_ids(db) if not active: return [], 0 q = q.where(CmdbObjectStorage.integration_id.in_(active)) if region: q = q.where(CmdbObjectStorage.region == region) if status: q = q.where(CmdbObjectStorage.status == status) # Migration 366 gave OSS an ``owner_person_id``; the list has # rendered the owner column ever since but offered no way to # filter on it. ``unowned`` is the counterpart the ops question # actually needs ("which buckets have nobody on the hook"). if owner_person_id: q = q.where(CmdbObjectStorage.owner_person_id == owner_person_id) if unowned: q = q.where(CmdbObjectStorage.owner_person_id.is_(None)) if search: q = q.where(CmdbObjectStorage.bucket_name.ilike(f"%{_esc(search)}%")) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = q.order_by(CmdbObjectStorage.bucket_name).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) return list(result.scalars().all()), total @staticmethod async def annotate_object_storage( db: AsyncSession, oss_id: str, data, # ObjectStorageAnnotation *, actor_person_id: str | None = None, # noqa: ARG004 trace_id: str | None = None, # noqa: ARG004 ) -> CmdbObjectStorage | None: # actor_person_id / trace_id are accepted for parity with the # sibling annotate_load_balancer / annotate_cdn_domain services # so the router calling convention is uniform. They are not used # inside the body today because CmdbObjectStorage has no # last_modified_* columns and OSS is outside the network owner- # history scope. The router writes the audit_logs row. result = await db.execute( select(CmdbObjectStorage).where( CmdbObjectStorage.id == uuid.UUID(oss_id), CmdbObjectStorage.deleted_at.is_(None), ).with_for_update() ) oss = result.scalar_one_or_none() if not oss: return None # Migration 366: owner is the canonical FK (UUID picker) with # owner_raw as the cloud-tag fallback. Mirror annotate_load_balancer # — _resolve_owner_input handles the UUID-wins / raw-fallback / # clear-to-null cases and validates the person is active. touched_owner, owner_id, raw = await CmdbService._resolve_owner_input(db, data) if touched_owner: oss.owner_person_id = owner_id if raw is not None: oss.owner_raw = raw await db.commit() await db.refresh(oss) return oss @staticmethod async def get_resource_stats(db: AsyncSession) -> dict: """Return resource counts grouped by provider and type, plus domain expiry stats. Runs one COUNT+GROUP BY query per resource table — never N+1. """ # --- per-provider counts for each resource type --- table_map = [ ("servers", CmdbServer, "instance_id"), ("domains", CmdbDomain, "domain_name"), ("dns_records", CmdbDnsRecord, "record_id"), ("load_balancers", CmdbLoadBalancer, "lb_id"), ("cdn_domains", CmdbCdnDomain, "domain_name"), ("object_storage", CmdbObjectStorage, "bucket_name"), ] by_provider: dict[str, dict[str, int]] = {} totals: dict[str, int] = {} for key, model_cls, _ in table_map: rows = await db.execute( select(model_cls.integration_id, func.count().label("cnt")) .where(model_cls.deleted_at.is_(None)) .group_by(model_cls.integration_id) ) type_total = 0 for integration_id, cnt in rows.all(): pid = str(integration_id) by_provider.setdefault(pid, {})[key] = cnt type_total += cnt totals[f"total_{key}"] = type_total # --- domain expiry stats --- now_date = datetime.now(UTC).date() expiry_30 = ( await db.scalar( select(func.count(CmdbDomain.id)).where( CmdbDomain.deleted_at.is_(None), CmdbDomain.expiry_date.is_not(None), CmdbDomain.expiry_date <= now_date + timedelta(days=30), CmdbDomain.expiry_date >= now_date, ) ) ) or 0 expiry_7 = ( await db.scalar( select(func.count(CmdbDomain.id)).where( CmdbDomain.deleted_at.is_(None), CmdbDomain.expiry_date.is_not(None), CmdbDomain.expiry_date <= now_date + timedelta(days=7), CmdbDomain.expiry_date >= now_date, ) ) ) or 0 return { **totals, "expiring_domains_30d": expiry_30, "expiring_domains_7d": expiry_7, "by_provider": by_provider, } # ------------------------------------------------------------------ # Analytics (charts on /admin/cloud/overview and /my/cloud/overview) # ------------------------------------------------------------------ @staticmethod async def get_analytics( db: AsyncSession, owner_person_id: str | None = None, billing_months: int = 12, integration_id: str | None = None, owner_email: str | None = None, owner_person_ids: list[str] | None = None, department: str | None = None, server_visibility: OwnerVisibility | None = None, measure: str = COST_MEASURE_LIST, ) -> dict: """Aggregate CMDB stats for the cloud + network overview dashboards. Parameters ---------- owner_person_id When set, every resource aggregation filters rows by that owner. Employee view passes their own ID; admin view passes None. department When set, an ADDITIVE filter (composes AND with every other scope): every owner-attributable model (those carrying ``owner_person_id``) is narrowed to rows owned by people whose primary org membership sits in ``department``'s subtree (a parent node rolls up every descendant). Models without an ``owner_person_id`` column report zero under the filter — they cannot be attributed to a department. The monthly billing trend is tenant-level and stays department-agnostic. Subtree match, same semantic as the admin overview's department filter (the platform-wide ``DepartmentTreeSelect``, unified in #709). owner_person_ids When set, supersedes ``owner_person_id`` and filters via ``owner_person_id IN (...)``. Used by the manager-scope path: a manager calling ``/analytics/my`` sees aggregates across self + subordinates so the dashboard numbers reconcile with the corresponding ``/cmdb/{domains, dns-records, cdn-domains}`` list endpoints which already expand to the manager's team scope. Without this parameter the dashboard undercounted self-only against the list pages. billing_months How many months of billing history to return. Only admin scope (``owner_person_id is None``) gets billing data — bills are tenant-level, not per-owner. measure 成本口径。``list``（默认）= 业务折前成本，按月聚合 ``cmdb_cloud_billing_daily`` 的 ``daily_engineering_amount()``； ``invoice`` = 实付，读 ``cmdb_billing_monthly``。调用方 （router）必须已经执行过 :func:`resolve_measure` 的保密圈闸 —— 本函数不判权。两个口径都逐行走 :class:`CloudCostUsdConverter`（CN 云把原生 CNY 写进 ``total_cost_usd*`` 并标 ``currency='CNY'``）。 integration_id When set, every aggregation is scoped to ONE cloud tenant — used by the tenant drill-down mode on the cloud overview page. ``None`` = all tenants (still broken down by tenant in the ``by_integration`` arrays, but summed together in totals/expiry/billing). owner_email Used only for the ``cmdb_iam_users`` table when ``owner_person_id`` is set: an IAM row is also counted as "owned" if its ``user_name`` matches the caller's email or email-local-part. Mirrors the fallback used by the IAM listing endpoint, since most tenants haven't reconciled ``owner_person_id`` on IAM users. Every aggregation uses one ``GROUP BY`` — no N+1. """ # Normalize: an ``owner_person_ids=[single_id]`` is equivalent # to ``owner_person_id=single_id``. A list with multiple ids # turns on the IN-clause path. ``owner_person_id`` is preserved # as a hint for the IAM email-fallback (which needs ONE # canonical id for the local-part match). if owner_person_ids and len(owner_person_ids) > 1: owner_scope_ids: list[str] | None = list(owner_person_ids) else: owner_scope_ids = None if owner_person_ids: owner_person_id = owner_person_ids[0] # Department filter — resolve once to the owning person-id set. # ``None`` = no filter; an empty list (department with no members) # narrows owner-attributable models to zero via ``IN ()``. Reuses # the admin-overview resolver so headcount + cost filters share # one definition of "this department". from app.domains.cmdb.admin_overview_service import ( resolve_department_person_ids, ) dept_person_ids = await resolve_department_person_ids(db, department) # --- 1. Integration catalog (id → provider label + display name) --- int_rows = ( await db.execute( select( Integration.id, Integration.name, Integration.provider, Integration.enabled, Integration.slug, ) .where(Integration.kind == "cloud") ) ).all() integrations = [ { "id": str(r.id), "name": r.name, "slug": r.slug, "provider": r.provider, "enabled": bool(r.enabled), } for r in int_rows ] provider_by_int = {i["id"]: i["provider"] for i in integrations} name_by_int = {i["id"]: i["name"] for i in integrations} # Set of integration_ids that are currently enabled. Used to # exclude disabled tenants from totals + aggregations so the # admin's "disable tenant X" action visibly drops the counts # (previous behavior: counts unchanged, eroding admin trust). # Per-tenant cards (below) still iterate over ALL integrations # so a disabled tenant remains visible in the catalog UI. enabled_integration_ids = { i["id"] for i in integrations if i["enabled"] } # Owner-scope honesty: models without an ``owner_person_id`` # column (load balancers, CDN, object storage) cannot be # per-user filtered yet. In owner-scope calls we short-circuit # to 0 for those instead of silently returning the fleet count # — an employee should never see someone else's bucket list. def _can_owner_scope(model_cls) -> bool: return hasattr(model_cls, "owner_person_id") # IAM email-fallback: most tenants haven't reconciled # ``owner_person_id`` on cmdb_iam_users yet, so an employee # whose IAM user_name == their email (the common AWS / Tencent / # Aliyun convention) would otherwise see iam_users=0 even when # the IAM listing page correctly surfaces 2 rows for them. Match # the IAM listing's filter exactly. def _owner_clause(model_cls): if owner_scope_ids is not None: # Manager scope: aggregate across self + subordinates. # IAM email-fallback doesn't apply to multi-id scope # (the local-part match is single-person semantics). return model_cls.owner_person_id.in_(owner_scope_ids) base = model_cls.owner_person_id == owner_person_id if model_cls is CmdbIamUser and owner_email and "@" in owner_email: local = owner_email.split("@", 1)[0] return or_( base, model_cls.user_name == local, model_cls.user_name == owner_email, ) return base owner_scoped = ( owner_person_id is not None or owner_scope_ids is not None ) def _with_filters(q, model_cls): q = q.where(model_cls.deleted_at.is_(None)) if owner_scoped and _can_owner_scope(model_cls): # Servers use the shared visibility contract (owner ∪ # owned-subscription attribution) when the caller passed # one, so the overview server counts reconcile with the # servers list / efficiency pages. Other resource types # keep the plain owner scope. if model_cls is CmdbServer and server_visibility is not None: q = q.where(visibility_clause(server_visibility)) else: q = q.where(_owner_clause(model_cls)) if dept_person_ids is not None and _can_owner_scope(model_cls): # Additive department narrowing on top of any owner / # integration scope already applied. q = q.where( model_cls.owner_person_id.in_(dept_person_ids), ) if integration_id is not None: q = q.where(model_cls.integration_id == integration_id) elif enabled_integration_ids: # Exclude disabled tenants from global aggregates. # When ``integration_id`` is set the caller has # already chosen a specific tenant (which may be # disabled and that's fine — they explicitly asked # for it), so we skip the enabled filter then. q = q.where( model_cls.integration_id.in_(enabled_integration_ids), ) return q # Ownerless models (no ``owner_person_id`` column) cannot be # attributed to a department — short-circuit them to empty under # either an owner scope or a department filter, so they never leak # the fleet total into a narrowed view. def _unattributable(model_cls) -> bool: return ( (owner_scoped or dept_person_ids is not None) and not _can_owner_scope(model_cls) ) async def _group_count(model_cls, column): if _unattributable(model_cls): return [] q = select(column, func.count().label("cnt")) q = _with_filters(q, model_cls).group_by(column) rows = (await db.execute(q)).all() return [(v, c) for v, c in rows] async def _total(model_cls): if _unattributable(model_cls): return 0 q = select(func.count(model_cls.id)) q = _with_filters(q, model_cls) return (await db.scalar(q)) or 0 # --- 2. Totals (respects owner + integration filters) --- totals = { "servers": await _total(CmdbServer), "domains": await _total(CmdbDomain), "dns_records": await _total(CmdbDnsRecord), "load_balancers": await _total(CmdbLoadBalancer), "cdn_domains": await _total(CmdbCdnDomain), "object_storage": await _total(CmdbObjectStorage), "iam_users": await _total(CmdbIamUser), } def _rollup_by_provider(rows): out: dict[str, int] = {} for int_id, cnt in rows: p = provider_by_int.get(str(int_id), "unknown") out[p] = out.get(p, 0) + int(cnt) return [{"provider": k, "count": v} for k, v in sorted(out.items(), key=lambda x: -x[1])] def _rollup_by_integration(rows): """Group by integration — each tenant shows as its own row (so two AWS tenants aren't silently merged into one 'aws').""" out: list[dict] = [] for int_id, cnt in rows: sid = str(int_id) out.append({ "integration_id": sid, "integration_name": name_by_int.get(sid, sid), "provider": provider_by_int.get(sid, "unknown"), "count": int(cnt), }) out.sort(key=lambda x: -x["count"]) return out def _kv(rows, key_name): return [ {key_name: (v or "unknown"), "count": int(c)} for v, c in sorted(rows, key=lambda x: -int(x[1])) ] # --- 3. Per-resource breakdowns --- servers_by_int = await _group_count(CmdbServer, CmdbServer.integration_id) servers_by_status = await _group_count(CmdbServer, CmdbServer.status) servers_by_region = await _group_count(CmdbServer, CmdbServer.region) servers_by_os = await _group_count(CmdbServer, CmdbServer.os_type) domains_by_int = await _group_count(CmdbDomain, CmdbDomain.integration_id) domains_by_registrar = await _group_count(CmdbDomain, CmdbDomain.registrar) # ICP filing: rows with a non-empty icp_filing_no vs without. # Restrict to zh-CN-resident registrars — a Cloudflare .com # or Route53 .io domain CAN'T have an ICP filing (it's a PRC # regulatory requirement); counting them as "missing" drags # the compliance rate to a meaningless number. # The cmdb provider whitelist below is the set of registrars # that ACTUALLY register zh-CN-resident domains in this org. # Anything outside this list is automatically considered # not-applicable and excluded from BOTH numerator + denominator. zh_cn_registrar_providers = {"aliyun", "tencent", "huawei"} zh_cn_integration_ids = { iid for iid, prov in provider_by_int.items() if prov in zh_cn_registrar_providers } async def _count(cond): q = select(func.count(CmdbDomain.id)) q = _with_filters(q, CmdbDomain).where(cond) if zh_cn_integration_ids: q = q.where(CmdbDomain.integration_id.in_(zh_cn_integration_ids)) else: # No zh-CN registrars at all — ICP doesn't apply. q = q.where(False) return int((await db.scalar(q)) or 0) icp_filed = await _count( CmdbDomain.icp_filing_no.is_not(None) & (CmdbDomain.icp_filing_no != "") ) icp_missing = await _count( CmdbDomain.icp_filing_no.is_(None) | (CmdbDomain.icp_filing_no == "") ) lbs_by_int = await _group_count(CmdbLoadBalancer, CmdbLoadBalancer.integration_id) lbs_by_type = await _group_count(CmdbLoadBalancer, CmdbLoadBalancer.lb_type) lbs_by_network = await _group_count(CmdbLoadBalancer, CmdbLoadBalancer.network_type) cdn_by_int = await _group_count(CmdbCdnDomain, CmdbCdnDomain.integration_id) cdn_by_source = await _group_count(CmdbCdnDomain, CmdbCdnDomain.source_type) # HTTPS adoption (public-site security metric) cdn_total_q = select(func.count(CmdbCdnDomain.id)) cdn_total_q = _with_filters(cdn_total_q, CmdbCdnDomain) cdn_total = int((await db.scalar(cdn_total_q)) or 0) cdn_https_q = select(func.count(CmdbCdnDomain.id)) cdn_https_q = _with_filters(cdn_https_q, CmdbCdnDomain).where(CmdbCdnDomain.is_https.is_(True)) cdn_https = int((await db.scalar(cdn_https_q)) or 0) # DNS by record_type and top zones dns_by_int = await _group_count(CmdbDnsRecord, CmdbDnsRecord.integration_id) dns_by_type = await _group_count(CmdbDnsRecord, CmdbDnsRecord.record_type) dns_by_zone = await _group_count(CmdbDnsRecord, CmdbDnsRecord.zone_name) buckets_by_int = await _group_count(CmdbObjectStorage, CmdbObjectStorage.integration_id) buckets_by_class = await _group_count(CmdbObjectStorage, CmdbObjectStorage.storage_class) iam_by_int = await _group_count(CmdbIamUser, CmdbIamUser.integration_id) # --- 4. Domain expiry — buckets + 180-day monthly timeline --- now_date = datetime.now(UTC).date() async def _expiry_count(lo_days: int | None, hi_days: int | None, include_no_expiry: bool = False): q = select(func.count(CmdbDomain.id)) q = _with_filters(q, CmdbDomain) if include_no_expiry: q = q.where(CmdbDomain.expiry_date.is_(None)) else: q = q.where(CmdbDomain.expiry_date.is_not(None)) if lo_days is not None: q = q.where(CmdbDomain.expiry_date >= now_date + timedelta(days=lo_days)) if hi_days is not None: q = q.where(CmdbDomain.expiry_date <= now_date + timedelta(days=hi_days)) return int((await db.scalar(q)) or 0) domain_expiry_buckets = [ {"bucket": "expired", "count": await _expiry_count(None, -1)}, {"bucket": "<=7d", "count": await _expiry_count(0, 7)}, {"bucket": "8-30d", "count": await _expiry_count(8, 30)}, {"bucket": "31-90d", "count": await _expiry_count(31, 90)}, {"bucket": ">90d", "count": await _expiry_count(91, 36500)}, {"bucket": "no_expiry", "count": await _expiry_count(None, None, include_no_expiry=True)}, ] # 6-month renewal runway — upcoming expiries bucketed by month. # Helps IT plan renewals ahead of the 90-day bucket. # Always use .replace(day=1) on next_start so the final bucket # doesn't bleed into the following month (bug: +32d alone can # land on day 2/3 for variable-length months). domain_expiry_timeline: list[dict] = [] for i in range(6): month_start = (now_date.replace(day=1) + timedelta(days=32 * i)).replace(day=1) next_start = (month_start + timedelta(days=32)).replace(day=1) q = select(func.count(CmdbDomain.id)) q = _with_filters(q, CmdbDomain).where( CmdbDomain.expiry_date.is_not(None), CmdbDomain.expiry_date >= month_start, CmdbDomain.expiry_date < next_start, ) c = int((await db.scalar(q)) or 0) domain_expiry_timeline.append({"month": month_start.strftime("%Y-%m"), "count": c}) # --- 5. Billing trend (admin only — bills are tenant-level) --- # 口径路由（2026-07 保密红线补口）：``cmdb_billing_monthly`` 是 # **实付**（发票）表，没有 list 列。这个趋势的受众是 # ``can_read_cloud``（含 auditor / management，都在保密圈外）， # 与已切成 list 的 ``/billing/monthly`` 完全重叠、粒度也一样 # （月 × integration × provider）——两个面一相除就是每云每月的 # 公司折扣率。所以默认（list）改为按月聚合 # ``cmdb_cloud_billing_daily`` 的 ``daily_engineering_amount()``， # ``invoice`` 才读原表（router 已执行保密圈闸）。 # # CN 云把原生 CNY 写进 ``total_cost_usd*`` 并标 ``currency='CNY'`` # （日表没有写侧 rebase，月表只有 Round-27 之后的行有），所以 # 两条路径都逐行走 ``CloudCostUsdConverter``；折不出汇率的行不 # 计入（绝不伪造 0），落 warning。 from app.core.fx_rate import CloudCostUsdConverter # noqa: PLC0415 cost_conv = CloudCostUsdConverter(db) wants_invoice = measure == COST_MEASURE_INVOICE billing_trend: list[dict] = [] if not owner_scoped: cutoff = (now_date.replace(day=1) - timedelta(days=31 * billing_months)).replace(day=1) if wants_invoice: bq = ( select( CmdbBillingMonthly.month, CmdbBillingMonthly.integration_id, CmdbBillingMonthly.total_cost_usd, CmdbBillingMonthly.currency, ) .where(CmdbBillingMonthly.month >= cutoff) .order_by(CmdbBillingMonthly.month.asc()) ) if integration_id is not None: bq = bq.where(CmdbBillingMonthly.integration_id == integration_id) else: # currency 进分组键 —— 混币组必须逐行折算，一起 SUM 会把 # ¥ 当 $ 加总。月首日用 date_format 归一，与月表的 # ``month`` 列（月初）同键。 _month_expr = func.date_format( CmdbCloudBillingDaily.date, "%Y-%m-01", ) bq = ( select( _month_expr.label("month"), CmdbCloudBillingDaily.integration_id, func.coalesce( func.sum(daily_engineering_amount()), 0, ).label("total"), CmdbCloudBillingDaily.currency, ) .where(CmdbCloudBillingDaily.date >= cutoff) .group_by( _month_expr, CmdbCloudBillingDaily.integration_id, CmdbCloudBillingDaily.currency, ) .order_by(_month_expr.asc()) ) if integration_id is not None: bq = bq.where( CmdbCloudBillingDaily.integration_id == integration_id, ) brows = (await db.execute(bq)).all() by_month: dict[str, dict] = {} for m, int_id, cost, currency in brows: # invoice 路径的 ``month`` 是 date，list 路径是 # ``date_format`` 出来的 "YYYY-MM-01" 字符串。 key = m.strftime("%Y-%m") if hasattr(m, "strftime") else str(m)[:7] entry = by_month.setdefault( key, {"month": key, "total_usd": 0.0, "by_provider": {}, "by_integration": {}} ) prov = provider_by_int.get(str(int_id), "unknown") iname = name_by_int.get(str(int_id), str(int_id)) cost_f, converted = await cost_conv.to_usd_checked( cost, currency, int_id, ) if not converted: continue entry["total_usd"] += cost_f entry["by_provider"][prov] = entry["by_provider"].get(prov, 0.0) + cost_f entry["by_integration"][iname] = entry["by_integration"].get(iname, 0.0) + cost_f billing_trend = [by_month[k] for k in sorted(by_month.keys())] # --- 6. Per-tenant summary cards (skipped in owner scope — an # employee's view is already one tenant-agnostic aggregate) --- per_tenant_summary: list[dict] = [] if not owner_scoped: # Latest-month cost per tenant — 同一口径路由。invoice 读 # 月表，list 取 ``billing_trend`` 最后一个月的 per-integration # 值（同一次日表聚合，不再多扫一遍；也保证卡片数与趋势末柱 # 是同一个数）。 latest_cost_by_int: dict[str, float] = {} if wants_invoice: latest_month = await db.scalar( select(func.max(CmdbBillingMonthly.month)), ) if latest_month is not None: lrows = ( await db.execute( select( CmdbBillingMonthly.integration_id, CmdbBillingMonthly.total_cost_usd, CmdbBillingMonthly.currency, ).where(CmdbBillingMonthly.month == latest_month) ) ).all() for i, c, currency in lrows: usd, converted = await cost_conv.to_usd_checked( c, currency, i, ) if converted: latest_cost_by_int[str(i)] = usd elif billing_trend: # ``by_integration`` 的键是 integration 名字（趋势图的 # 图例），这里要 id → 用最近一个月重新按 id 聚合日表。 _latest_key = billing_trend[-1]["month"] _latest_start = date_type.fromisoformat(f"{_latest_key}-01") lq = ( select( CmdbCloudBillingDaily.integration_id, func.coalesce( func.sum(daily_engineering_amount()), 0, ).label("total"), CmdbCloudBillingDaily.currency, ) .where( CmdbCloudBillingDaily.date >= _latest_start, CmdbCloudBillingDaily.date < ( _latest_start + timedelta(days=32) ).replace(day=1), ) .group_by( CmdbCloudBillingDaily.integration_id, CmdbCloudBillingDaily.currency, ) ) if integration_id is not None: lq = lq.where( CmdbCloudBillingDaily.integration_id == integration_id, ) # 这一趟是对**已经扫过的钱**换一个维度重读(list 路径: # 趋势图窗口里最近一个月的同一批日表行;invoice 路径: # 趋势图最新那个点的同期日表)。因此它的「折不出汇率」 # 必须走独立 converter —— 与趋势图共用一个会把同一笔原生 # 金额累加两次,披露值凭空翻倍(实测 ¥5000 报成 ¥10000)。 latest_conv = CloudCostUsdConverter(db) for i, c, currency in (await db.execute(lq)).all(): usd, converted = await latest_conv.to_usd_checked( c, currency, i, ) if converted: latest_cost_by_int[str(i)] = ( latest_cost_by_int.get(str(i), 0.0) + usd ) # 两趟覆盖同一批钱,取**较大**者而不是求和:list 路径下 # 最近一月是趋势窗口的子集,invoice 路径下日表与月表是同 # 一笔钱的两种粒度。求和会双计,只取趋势会漏掉「月表折得 # 出、日表折不出」的币种。 for cur, native in latest_conv.unconverted_by_currency.items(): cost_conv.unconverted_by_currency[cur] = max( cost_conv.unconverted_by_currency.get(cur, 0.0), native, ) if cost_conv.unconverted_by_currency: # Don't silently fold un-converted native amounts into the # USD trend/total as 0 — surface a missing FX rate. logger.warning( "cmdb analytics billing_trend: rows had no FX rate and " "were excluded from USD totals: %s", cost_conv.unconverted_by_currency, ) # Last-synced-at per tenant (max across all resource tables) async def _last_sync(model_cls): rows = ( await db.execute( select(model_cls.integration_id, func.max(model_cls.last_synced_at)) .where(model_cls.deleted_at.is_(None)) .group_by(model_cls.integration_id) ) ).all() return {str(i): t for i, t in rows} ls_by_int: dict[str, datetime] = {} for mc in (CmdbServer, CmdbDomain, CmdbDnsRecord, CmdbLoadBalancer, CmdbCdnDomain, CmdbObjectStorage, CmdbIamUser): for i, t in (await _last_sync(mc)).items(): cur = ls_by_int.get(i) if t is not None and (cur is None or t > cur): ls_by_int[i] = t # Per-integration resource counts (one dict across all types) def _counts_from_int_rows(rows): return {str(i): int(c) for i, c in rows} counts = { "servers": _counts_from_int_rows(servers_by_int), "domains": _counts_from_int_rows(domains_by_int), "dns_records": _counts_from_int_rows(dns_by_int), "load_balancers": _counts_from_int_rows(lbs_by_int), "cdn_domains": _counts_from_int_rows(cdn_by_int), "object_storage": _counts_from_int_rows(buckets_by_int), "iam_users": _counts_from_int_rows(iam_by_int), } for itg in integrations: iid = itg["id"] last_sync = ls_by_int.get(iid) per_tenant_summary.append({ "integration_id": iid, "name": itg["name"], "slug": itg["slug"], "provider": itg["provider"], "enabled": itg["enabled"], "counts": {k: v.get(iid, 0) for k, v in counts.items()}, "latest_month_cost_usd": latest_cost_by_int.get(iid, 0.0), "last_synced_at": last_sync.isoformat() if last_sync else None, }) # Sort: enabled first, then by total resource count desc per_tenant_summary.sort( key=lambda x: (not x["enabled"], -sum(x["counts"].values())) ) return { "scope": "owner" if owner_person_id else "fleet", "selected_integration_id": integration_id, "totals": totals, "integrations": integrations, "per_tenant_summary": per_tenant_summary, "servers": { "by_provider": _rollup_by_provider(servers_by_int), "by_integration": _rollup_by_integration(servers_by_int), "by_status": _kv(servers_by_status, "status"), "by_region": _kv(servers_by_region, "region"), "by_os": _kv(servers_by_os, "os_type"), }, "domains": { "by_provider": _rollup_by_provider(domains_by_int), "by_integration": _rollup_by_integration(domains_by_int), "by_registrar": _kv(domains_by_registrar, "registrar"), "expiry_buckets": domain_expiry_buckets, "expiry_timeline": domain_expiry_timeline, "icp_filing": {"filed": icp_filed, "missing": icp_missing}, }, "dns_records": { "by_provider": _rollup_by_provider(dns_by_int), "by_integration": _rollup_by_integration(dns_by_int), "by_record_type": _kv(dns_by_type, "record_type"), "top_zones": _kv(dns_by_zone, "zone_name")[:10], }, "load_balancers": { "by_provider": _rollup_by_provider(lbs_by_int), "by_integration": _rollup_by_integration(lbs_by_int), "by_type": _kv(lbs_by_type, "lb_type"), "by_network": _kv(lbs_by_network, "network_type"), }, "cdn_domains": { "by_provider": _rollup_by_provider(cdn_by_int), "by_integration": _rollup_by_integration(cdn_by_int), "by_source_type": _kv(cdn_by_source, "source_type"), "https_coverage": { "total": cdn_total, "https": cdn_https, "pct": round(100.0 * cdn_https / cdn_total, 1) if cdn_total else 0.0, }, }, "object_storage": { "by_provider": _rollup_by_provider(buckets_by_int), "by_integration": _rollup_by_integration(buckets_by_int), "by_storage_class": _kv(buckets_by_class, "storage_class"), }, "iam_users": { "by_provider": _rollup_by_provider(iam_by_int), "by_integration": _rollup_by_integration(iam_by_int), }, "billing_trend": billing_trend, # 口径标识 + 覆盖面基准。FE 据此渲染徽标；财务切口径时看到的 # 差额含覆盖差（invoice 是租户级 provider 月汇总，list 是日表 # 逐 scope 行之和），不可整体当折扣读。与 ``/billing/monthly`` # 的 ``meta`` 同语义。 "billing_measure": measure, "billing_aggregate_basis": ( "tenant_monthly" if wants_invoice else "billing_daily_sum" ), "billing_unconverted_by_currency": ( dict(cost_conv.unconverted_by_currency) if cost_conv.unconverted_by_currency else None ), }