"""Admin Insights — unified analytics endpoints.

One file, eight categories, all admin-only:

    /api/v1/dashboard/insights/overview     — executive KPI snapshot
    /api/v1/dashboard/insights/people       — headcount / org analytics
    /api/v1/dashboard/insights/assets       — hardware lifecycle
    /api/v1/dashboard/insights/software     — SaaS seat cost / utilization
    /api/v1/dashboard/insights/operations   — provisioning SLA / throughput
    /api/v1/dashboard/insights/lifecycle    — onboarding / offboarding health
    /api/v1/dashboard/insights/security     — audit / privileged-access risk
    /api/v1/dashboard/insights/cloud        — CMDB billing / inventory
    /api/v1/dashboard/insights/impact       — cumulative platform scorecard

Every endpoint takes a ``range`` query param (``7d`` / ``30d`` / ``90d`` /
``12m`` / ``24m``) and returns a wide, shallow JSON payload. The frontend
builds every chart from a single round-trip per tab — never N+1.

There are no fallbacks for missing rows: if the underlying table is empty,
the field is ``[]`` or ``0``. The frontend renders an Empty state. We
never invent numbers.
"""
from __future__ import annotations

import logging
from collections.abc import Sequence
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any, Literal
from uuid import UUID

from fastapi import APIRouter, Depends, Query, Request
from sqlalchemy import and_, case, distinct, func, or_, select
from sqlalchemy import true as sa_true
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from app.approval.models import ApprovalRequest
from app.audit.models import AuditLog
from app.auth.models import AuthRole, AuthUser, AuthUserRole
from app.core.database import get_db
from app.core.dependencies import (
    check_sensitive_endpoint_rate_limit,
    get_current_user,
)
from app.core.exceptions import ForbiddenError
from app.core.fx_rate import get_usd_per_cny
from app.core.geo import normalize_country_code
from app.core.rbac import (
    ASSET_ADMIN_ROLE,
    AUDITOR_ROLE,
    CLOUD_ADMIN_ROLE,
    FINANCE_ROLE,
    GLOBAL_ADMIN_ROLE,
    HR_ROLE,
    has_role,
)
from app.core.response import success_response
from app.core.tenure import completed_years
from app.core.workflow_journal.constants import RUN_STATUS_SUCCEEDED
from app.core.workflow_journal.models import WorkflowRun
from app.domains.apikey.models import ApiKeyAssignment
from app.domains.asset.models import (
    ASSET_IN_USE_STATUSES,
    AssetItem,
    AssetOperation,
    AssetSoftwareCatalog,
    EmployeeRequest,
    SoftwareExternalHolder,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.domains.auditor.models import (
    AccessCertification,
    AccessCertificationItem,
    AuditorMfaStatus,
    AuditorPublicExposureFinding,
    AuditorSigninEvent,
)
from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    GRANT_STATUSES_RECLAIMED,
    SCOPE_STATUS_ACTIVE,
    CmdbBillingMonthly,
    CmdbCdnDomain,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbCloudScopeRequest,
    CmdbDnsRecord,
    CmdbDnsRecordRequest,
    CmdbDomain,
    CmdbIamUser,
    CmdbLoadBalancer,
    CmdbServer,
)
from app.domains.email_group.models import EmailGroup
from app.domains.google_workspace.models import GoogleOrphanAccount
from app.domains.identity_sync.models import IdentitySyncRun
from app.domains.leaver_blocked.models import LeaverBlockAudit, LeaverDualSignOverride
from app.domains.offboarding.models import (
    PROC_BLOCKED,
    PROC_COMPLETED,
    RESO_IDENTITY_SYNC_MANAGED,
    RESO_PENDING,
    RESO_REVOKED_AUTO,
    RESO_REVOKED_MANUAL,
    RESO_SSO_CASCADE,
    RESO_TRANSFERRED,
    XFER_PENDING,
    AssetTransferRequest,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.domains.onboarding.models import OnboardingProcess
from app.domains.permission.models import PERM_REVOKED, UserPermission
from app.domains.procurement.models import (
    ProcurementContract,
    ProcurementOrder,
    ProcurementSupplier,
    SoftwareProcurementMeta,
)
from app.domains.provisioning.models import (
    EXEC_AUTO_API,
    EXEC_MANUAL_NOTIFY,
    EXEC_SSO_AUTO,
    STATUS_COMPLETED,
    STATUS_FAILED,
    STATUS_IN_PROGRESS,
    STATUS_NOTIFIED,
    STATUS_PENDING,
    ProvisioningTask,
    SoftwareConnector,
)
from app.domains.sso.models import SSO_GRANT_STATUS_REVOKED, SsoGrant
from app.identity.org_hierarchy import rollup_counts
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)
from app.integrations.models import Integration
from app.notifications.models import NotificationOutbox
from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun

router = APIRouter(prefix="/api/v1/dashboard/insights", tags=["dashboard-insights"])

logger = logging.getLogger(__name__)

RangeKey = Literal["7d", "30d", "90d", "12m", "24m"]


# ─── Helpers ────────────────────────────────────────────────────────


def _require_admin(user: AuthUser) -> None:
    """Read gate for the insights surface.

    Open to ``global_admin`` (and its legacy ``it_admin`` alias) and
    to the ``management`` role — the leadership view's whole point is
    to see operational dashboards without being a full IT admin.
    Mutating endpoints (none today on the insights surface) would
    use a separate write gate.
    """
    from app.core.rbac import MANAGEMENT_ROLE
    role_names = user.role_names or []
    if (
        has_role(role_names, GLOBAL_ADMIN_ROLE)
        or has_role(role_names, "it_admin")
        or has_role(role_names, MANAGEMENT_ROLE)
    ):
        return
    raise ForbiddenError("Admin required")


# Per-section read gates. The cockpit (运营洞察) shows EVERY section to
# admin / management. Each section is ALSO surfaced as a standalone domain
# page for the role that owns that domain (2026-05-27, nav-split PR3), so
# the read gate is widened per section. Sections absent from the map
# (overview / operations) stay admin/management-only — they're the
# cross-domain executive snapshot + provisioning-ops view.
_INSIGHTS_SECTION_EXTRA_ROLES: dict[str, tuple[str, ...]] = {
    "people": (HR_ROLE,),
    "lifecycle": (HR_ROLE,),
    "assets": (ASSET_ADMIN_ROLE,),
    "software": (ASSET_ADMIN_ROLE, FINANCE_ROLE),
    "software-catalog": (ASSET_ADMIN_ROLE, FINANCE_ROLE),
    "renewals": (ASSET_ADMIN_ROLE, FINANCE_ROLE),
    "cloud": (CLOUD_ADMIN_ROLE, FINANCE_ROLE),
    "hardware-finance": (ASSET_ADMIN_ROLE, FINANCE_ROLE),
    "security": (AUDITOR_ROLE,),
}


def _require_insights_section(user: AuthUser, section: str) -> None:
    """Read gate for a single insights section.

    Admin / it_admin / management always pass (same as ``_require_admin``).
    Beyond that, each section admits the role that owns its domain so the
    standalone domain page is reachable by that role without full admin
    rights: people / lifecycle → hr, assets → asset_admin, software (+
    catalog) → asset_admin + finance, cloud → cloud_admin + finance,
    security → auditor.
    """
    from app.core.rbac import MANAGEMENT_ROLE
    role_names = user.role_names or []
    if (
        has_role(role_names, GLOBAL_ADMIN_ROLE)
        or has_role(role_names, "it_admin")
        or has_role(role_names, MANAGEMENT_ROLE)
    ):
        return
    for role in _INSIGHTS_SECTION_EXTRA_ROLES.get(section, ()):
        if has_role(role_names, role):
            return
    raise ForbiddenError("Not authorized for this insights section")


def _range_to_window(rng: str) -> tuple[date, int, str]:
    """Resolve the requested window.

    Returns ``(start_date, granularity_buckets, granularity_unit)``:
        * ``7d``  → 7 daily buckets
        * ``30d`` → 30 daily buckets
        * ``90d`` → 13 weekly buckets (~90 days)
        * ``12m`` → 12 monthly buckets
        * ``24m`` → 24 monthly buckets
    """
    today = date.today()
    if rng == "7d":
        return today - timedelta(days=6), 7, "day"
    if rng == "30d":
        return today - timedelta(days=29), 30, "day"
    if rng == "90d":
        return today - timedelta(days=89), 90, "day"
    if rng == "24m":
        return _month_floor(today, 23), 24, "month"
    # 12m default
    return _month_floor(today, 11), 12, "month"


def _month_floor(d: date, months_back: int) -> date:
    y, m = d.year, d.month - months_back
    while m <= 0:
        m += 12
        y -= 1
    return date(y, m, 1)


def _months_axis(start: date, count: int) -> list[str]:
    out: list[str] = []
    y, m = start.year, start.month
    for _ in range(count):
        out.append(f"{y:04d}-{m:02d}")
        m += 1
        if m > 12:
            m = 1
            y += 1
    return out


def _days_axis(start: date, count: int) -> list[str]:
    return [(start + timedelta(days=i)).isoformat() for i in range(count)]


def _start_of_day(d: date) -> datetime:
    return datetime.combine(d, datetime.min.time(), tzinfo=UTC)


def _to_float(v: Any) -> float:
    if v is None:
        return 0.0
    if isinstance(v, Decimal):
        return float(v)
    return float(v)


def _aware_utc(dt: datetime | None) -> datetime | None:
    """Coerce a possibly-naive datetime to UTC-aware.

    MySQL/TiDB do not actually store timezone information even for
    ``DateTime(timezone=True)`` columns, so SQLAlchemy returns naive
    datetimes for some installs. Subtracting a naive datetime from an
    aware one (``today_dt`` built with ``tzinfo=UTC``) raises
    ``TypeError: can't subtract offset-naive and offset-aware datetimes``
    and 500s every Insights tab that does any age math. Normalize before
    arithmetic instead of crashing.
    """
    if dt is None:
        return None
    if dt.tzinfo is None:
        return dt.replace(tzinfo=UTC)
    return dt


def _str_str_int(rows: Sequence[Any]) -> dict[str, int]:
    """Coerce ``[(label_or_None, count)]`` into ``{str(label): int(count)}``.

    Buckets ``NULL`` labels into the literal string ``"unknown"`` so the
    frontend can show them under one consistent slice.

    Takes ``Sequence``, not ``list``: every one of the ~27 call sites passes
    a SQLAlchemy ``.all()`` result, which is a ``Sequence[Row[...]]`` and not
    a ``list`` — the narrower annotation made each of those an ``arg-type``
    error, and one more of them tipped the mypy ratchet over on main.
    """
    out: dict[str, int] = {}
    for k, n in rows:
        out[str(k) if k is not None else "unknown"] = int(n)
    return out


# ─── 1. Overview ────────────────────────────────────────────────────


@router.get("/overview", summary="Executive KPI snapshot")
async def insights_overview(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_admin(user)
    today = date.today()
    last_30 = today - timedelta(days=30)
    last_60 = today - timedelta(days=60)
    month_start = date(today.year, today.month, 1)
    last_month_start = _month_floor(today, 1)

    # Headcount (active + leaver_pending)
    headcount = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status.in_(["active", "leaver_pending"]),
            )
        )
    )).scalar() or 0

    # 30d hires/leavers — vs prior 30d
    hires_30 = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(IdentityPerson.hire_date.is_not(None), IdentityPerson.hire_date >= last_30)
        )
    )).scalar() or 0
    hires_prev_30 = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(
                IdentityPerson.hire_date.is_not(None),
                IdentityPerson.hire_date >= last_60,
                IdentityPerson.hire_date < last_30,
            )
        )
    )).scalar() or 0
    leavers_30 = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date >= last_30)
        )
    )).scalar() or 0
    leavers_prev_30 = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(
                IdentityPerson.termination_date.is_not(None),
                IdentityPerson.termination_date >= last_60,
                IdentityPerson.termination_date < last_30,
            )
        )
    )).scalar() or 0

    # Active asset value (sum of purchase_cost where assigned)
    asset_value_row = (await db.execute(
        select(
            func.count(),
            func.coalesce(func.sum(AssetItem.purchase_cost), 0),
        ).where(AssetItem.status == "assigned")
    )).one()
    assigned_assets = int(asset_value_row[0] or 0)
    assigned_value = _to_float(asset_value_row[1])

    available_assets = (await db.execute(
        select(func.count()).select_from(AssetItem).where(
            AssetItem.status.in_(["available", "idle"])
        )
    )).scalar() or 0

    # Software seats + monthly cost
    sw_rows = (await db.execute(
        select(
            SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle,
            func.count(SoftwareSeatAllocation.id),
        )
        .join(SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id)
        .where(SoftwareSeatAllocation.status == "active")
        .group_by(SoftwareLicensePlan.id, SoftwareLicensePlan.price, SoftwareLicensePlan.billing_cycle)
    )).all()
    total_active_seats = 0
    total_monthly_cost = 0.0
    for price, cycle, cnt in sw_rows:
        n = int(cnt)
        total_active_seats += n
        if cycle == "monthly":
            total_monthly_cost += _to_float(price) * n
        elif cycle == "yearly":
            total_monthly_cost += _to_float(price) * n / 12.0

    # Provisioning backlog
    prov_status = _str_str_int((await db.execute(
        select(ProvisioningTask.status, func.count())
        .group_by(ProvisioningTask.status)
    )).all())
    open_tasks = (
        prov_status.get(STATUS_PENDING, 0)
        + prov_status.get(STATUS_IN_PROGRESS, 0)
        + prov_status.get(STATUS_NOTIFIED, 0)
    )
    failed_tasks = prov_status.get(STATUS_FAILED, 0)

    # Open transfers
    open_transfers = (await db.execute(
        select(func.count()).select_from(AssetTransferRequest).where(
            AssetTransferRequest.status == XFER_PENDING
        )
    )).scalar() or 0

    # Blocked offboardings past leaver_date
    blocked_past_due = (await db.execute(
        select(func.count())
        .select_from(OffboardingProcess)
        .where(
            and_(
                OffboardingProcess.status == PROC_BLOCKED,
                OffboardingProcess.leaver_date <= today,
            )
        )
    )).scalar() or 0

    # 30d trend — daily series for requests + tasks
    request_daily_rows = (await db.execute(
        select(func.date(EmployeeRequest.created_at), func.count())
        .where(EmployeeRequest.created_at >= _start_of_day(last_30))
        .group_by(func.date(EmployeeRequest.created_at))
    )).all()
    task_daily_rows = (await db.execute(
        select(func.date(ProvisioningTask.created_at), func.count())
        .where(ProvisioningTask.created_at >= _start_of_day(last_30))
        .group_by(func.date(ProvisioningTask.created_at))
    )).all()
    days = _days_axis(last_30, 30)
    req_map = {str(d): int(n) for d, n in request_daily_rows}
    task_map = {str(d): int(n) for d, n in task_daily_rows}
    daily_trend = [
        {"day": d, "requests": req_map.get(d, 0), "tasks": task_map.get(d, 0)}
        for d in days
    ]

    # MoM cost & seat delta — compare last full month vs current MTD just for awareness
    # (we compare current MTD with same-day prior month — too noisy; instead show
    # number of newly-allocated seats this month vs prior month)
    seats_grants_mtd = (await db.execute(
        select(func.count()).select_from(SoftwareSeatAllocation).where(
            SoftwareSeatAllocation.allocated_at >= _start_of_day(month_start)
        )
    )).scalar() or 0
    seats_grants_last_month = (await db.execute(
        select(func.count()).select_from(SoftwareSeatAllocation).where(
            and_(
                SoftwareSeatAllocation.allocated_at >= _start_of_day(last_month_start),
                SoftwareSeatAllocation.allocated_at < _start_of_day(month_start),
            )
        )
    )).scalar() or 0

    # Top alerts (synthesized — only real conditions, no decoration)
    alerts: list[dict[str, Any]] = []
    if blocked_past_due > 0:
        alerts.append({
            "level": "error",
            "key": "blocked_offboarding_past_due",
            "count": int(blocked_past_due),
            "link": "/admin/offboarding",
        })
    if failed_tasks > 0:
        alerts.append({
            "level": "warning",
            "key": "provisioning_failed",
            "count": int(failed_tasks),
            "link": "/admin/provisioning/tasks?status=failed",
        })
    # Stale in_progress > 7 days
    stale_in_progress = (await db.execute(
        select(func.count()).select_from(ProvisioningTask).where(
            and_(
                ProvisioningTask.status == STATUS_IN_PROGRESS,
                ProvisioningTask.started_at.is_not(None),
                ProvisioningTask.started_at <= _start_of_day(today - timedelta(days=7)),
            )
        )
    )).scalar() or 0
    if stale_in_progress > 0:
        alerts.append({
            "level": "warning",
            "key": "provisioning_stale",
            "count": int(stale_in_progress),
            "link": "/admin/provisioning/tasks?status=in_progress",
        })
    # API keys expiring within 7 days
    expiring_keys = (await db.execute(
        select(func.count()).select_from(ApiKeyAssignment).where(
            and_(
                ApiKeyAssignment.status == "active",
                ApiKeyAssignment.valid_until.is_not(None),
                ApiKeyAssignment.valid_until >= today,
                ApiKeyAssignment.valid_until <= today + timedelta(days=7),
            )
        )
    )).scalar() or 0
    if expiring_keys > 0:
        alerts.append({
            "level": "info",
            "key": "api_keys_expiring",
            "count": int(expiring_keys),
            "link": "/admin/assets/api-keys",
        })
    # Domains expiring within 30 days
    expiring_domains = (await db.execute(
        select(func.count()).select_from(CmdbDomain).where(
            and_(
                CmdbDomain.deleted_at.is_(None),
                CmdbDomain.expiry_date.is_not(None),
                CmdbDomain.expiry_date >= today,
                CmdbDomain.expiry_date <= today + timedelta(days=30),
            )
        )
    )).scalar() or 0
    if expiring_domains > 0:
        alerts.append({
            "level": "warning",
            "key": "domains_expiring",
            "count": int(expiring_domains),
            "link": "/admin/cloud/domains",
        })

    return success_response(
        data={
            "as_of": today.isoformat(),
            "kpis": {
                "headcount": int(headcount),
                "hires_30d": int(hires_30),
                "hires_prev_30d": int(hires_prev_30),
                "leavers_30d": int(leavers_30),
                "leavers_prev_30d": int(leavers_prev_30),
                "assigned_assets": assigned_assets,
                "available_assets": int(available_assets),
                "assigned_asset_value": assigned_value,
                "active_seats": total_active_seats,
                "monthly_software_cost": round(total_monthly_cost, 2),
                "open_tasks": int(open_tasks),
                "failed_tasks": int(failed_tasks),
                "open_transfers": int(open_transfers),
                "blocked_past_due": int(blocked_past_due),
                "seats_grants_mtd": int(seats_grants_mtd),
                "seats_grants_last_month": int(seats_grants_last_month),
            },
            "daily_trend": daily_trend,
            "alerts": alerts,
        },
        request=request,
    )


# ─── 2. People ──────────────────────────────────────────────────────

#: 司龄分布 bands, in display order. Year boundaries sit on the SAME milestones
#: the 成就墙's 同行时光 badges are earned on (1/3/5/7/10 年) so the two
#: surfaces can never disagree on who counts as 满 N 年.
TENURE_BUCKETS: tuple[str, ...] = (
    "<1m", "1-3m", "3-6m", "6-12m",
    "1-2y", "2-3y", "3-5y", "5-7y", "7-10y", "10y+",
)


def _tenure_bucket(hire: date, today: date) -> str:
    """The 司龄 band ``hire`` falls in as of ``today``.

    Year boundaries use the platform tenure caliber (calendar-aware
    ``completed_years``, core/tenure.py), not ``N × 365`` days — the day
    caliber drifted 1-2 days across leap years and disagreed with the badge
    wall's anniversary milestones. Sub-year bands stay day-based (a "month"
    band has no calendar anchor worth arguing about).
    """
    years = completed_years(hire, today)
    if years >= 10:
        return "10y+"
    if years >= 7:
        return "7-10y"
    if years >= 5:
        return "5-7y"
    if years >= 3:
        return "3-5y"
    if years >= 2:
        return "2-3y"
    if years >= 1:
        return "1-2y"
    days = (today - hire).days
    if days < 30:
        return "<1m"
    if days < 90:
        return "1-3m"
    if days < 180:
        return "3-6m"
    return "6-12m"


async def build_people_payload(db: AsyncSession, rng: str) -> dict[str, Any]:
    """Aggregate the 人员洞察 (People) payload for one time window.

    Extracted from the ``GET /people`` handler (2026-07-08) so the AI
    diagnosis generator (``insights_hr_ai``) reads EXACTLY what the tab
    renders — one source of truth for the numbers on screen and the
    numbers the LLM narrates.
    """
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()

    headcount = (await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status.in_(["active", "leaver_pending"]),
            )
        )
    )).scalar() or 0

    by_status = _str_str_int((await db.execute(
        select(IdentityPerson.employment_status, func.count())
        .where(IdentityPerson.deleted_at.is_(None))
        .group_by(IdentityPerson.employment_status)
    )).all())

    # Department rollup
    dept_rows = (await db.execute(
        select(
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_name,
            func.count(),
        )
        .join(IdentityPerson, IdentityPerson.id == IdentityOrgMembership.person_id)
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            )
        )
        .group_by(IdentityOrgMembership.org_path, IdentityOrgMembership.org_node_name)
    )).all()
    by_department_tree = rollup_counts((op, name, int(cnt)) for op, name, cnt in dept_rows)

    by_location = _str_str_int((await db.execute(
        select(IdentityPerson.work_location_name, func.count())
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
        )
        .group_by(IdentityPerson.work_location_name)
    )).all())

    by_country = _str_str_int((await db.execute(
        select(IdentityPerson.country, func.count())
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
        )
        .group_by(IdentityPerson.country)
    )).all())

    title_rows = (await db.execute(
        select(IdentityPerson.job_title, func.count())
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
                IdentityPerson.job_title.is_not(None),
            )
        )
        .group_by(IdentityPerson.job_title)
        .order_by(func.count().desc())
        .limit(20)
    )).all()
    by_job_title = _str_str_int(title_rows)

    # Tenure buckets (active employees only, hire_date present)
    tenure_rows = (await db.execute(
        select(IdentityPerson.hire_date)
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
                IdentityPerson.hire_date.is_not(None),
            )
        )
    )).scalars().all()
    buckets = dict.fromkeys(TENURE_BUCKETS, 0)
    for hd in tenure_rows:
        buckets[_tenure_bucket(hd, today)] += 1
    tenure_distribution = buckets

    # Hires/leavers monthly trend
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        hires_rows = (await db.execute(
            select(func.date_format(IdentityPerson.hire_date, "%Y-%m"), func.count())
            .where(
                and_(
                    IdentityPerson.hire_date.is_not(None),
                    IdentityPerson.hire_date >= start_date,
                )
            )
            .group_by(func.date_format(IdentityPerson.hire_date, "%Y-%m"))
        )).all()
        leavers_rows = (await db.execute(
            select(func.date_format(IdentityPerson.termination_date, "%Y-%m"), func.count())
            .where(
                and_(
                    IdentityPerson.termination_date.is_not(None),
                    IdentityPerson.termination_date >= start_date,
                )
            )
            .group_by(func.date_format(IdentityPerson.termination_date, "%Y-%m"))
        )).all()
        h_map = {str(k): int(v) for k, v in hires_rows}
        l_map = {str(k): int(v) for k, v in leavers_rows}
        trend = [
            {"bucket": b, "hires": h_map.get(b, 0), "leavers": l_map.get(b, 0),
             "net": h_map.get(b, 0) - l_map.get(b, 0)}
            for b in axis
        ]
    else:
        axis = _days_axis(start_date, _bucket_count)
        hires_rows = (await db.execute(
            select(IdentityPerson.hire_date, func.count())
            .where(
                and_(
                    IdentityPerson.hire_date.is_not(None),
                    IdentityPerson.hire_date >= start_date,
                )
            )
            .group_by(IdentityPerson.hire_date)
        )).all()
        leavers_rows = (await db.execute(
            select(IdentityPerson.termination_date, func.count())
            .where(
                and_(
                    IdentityPerson.termination_date.is_not(None),
                    IdentityPerson.termination_date >= start_date,
                )
            )
            .group_by(IdentityPerson.termination_date)
        )).all()
        h_map = {d.isoformat(): int(n) for d, n in hires_rows}
        l_map = {d.isoformat(): int(n) for d, n in leavers_rows}
        trend = [
            {"bucket": b, "hires": h_map.get(b, 0), "leavers": l_map.get(b, 0),
             "net": h_map.get(b, 0) - l_map.get(b, 0)}
            for b in axis
        ]

    # Manager span of control
    span_rows = (await db.execute(
        select(IdentityPerson.manager_person_id, func.count())
        .where(
            and_(
                IdentityPerson.manager_person_id.is_not(None),
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
        )
        .group_by(IdentityPerson.manager_person_id)
    )).all()
    spans = sorted(int(n) for _, n in span_rows)
    if spans:
        p50 = spans[len(spans) // 2]
        p90 = spans[min(int(len(spans) * 0.9), len(spans) - 1)]
        max_span = spans[-1]
    else:
        p50 = p90 = max_span = 0

    # Span buckets for histogram
    span_buckets = {"1-2": 0, "3-5": 0, "6-9": 0, "10-15": 0, "16+": 0}
    for n in spans:
        if n <= 2:
            span_buckets["1-2"] += 1
        elif n <= 5:
            span_buckets["3-5"] += 1
        elif n <= 9:
            span_buckets["6-9"] += 1
        elif n <= 15:
            span_buckets["10-15"] += 1
        else:
            span_buckets["16+"] += 1

    # International assignment / 外派 — active staff on a host-country
    # assignment (host country set AND != home). Supplementary HRIS insight
    # (migration 392): count + host-country breakdown + the assignee roster.
    assignment_rows = (await db.execute(
        select(
            IdentityPerson.display_name,
            IdentityPerson.assignment_home_country,
            IdentityPerson.assignment_host_country,
            IdentityPerson.assignment_start_date,
            IdentityPerson.assignment_end_date,
        ).where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status.in_(["active", "leaver_pending"]),
                IdentityPerson.assignment_host_country.is_not(None),
                IdentityPerson.assignment_host_country != "",
            )
        )
    )).all()
    assignees = [
        {
            "display_name": dn,
            "home_country": home,
            "host_country": host,
            "start_date": sd.isoformat() if sd else None,
            "end_date": ed.isoformat() if ed else None,
        }
        for dn, home, host, sd, ed in assignment_rows
        # Mirror IdentityPerson.is_on_international_assignment: host != home
        # AND the assignment window hasn't ended (an ended assignment whose
        # historical host country lingers must not count as active).
        if (host or "").strip().upper() != (home or "").strip().upper()
        and (ed is None or ed >= today)
    ]
    by_host_country: dict[str, int] = {}
    for a in assignees:
        key = a["host_country"] or "?"
        by_host_country[key] = by_host_country.get(key, 0) + 1

    return {
        "headcount": int(headcount),
        "by_status": by_status,
        "by_department_tree": by_department_tree,
        "by_location": by_location,
        "by_country": by_country,
        "by_job_title": by_job_title,
        "tenure_distribution": tenure_distribution,
        "trend": trend,
        "axis_unit": unit,
        "manager_span": {
            "managers_count": len(spans),
            "p50": p50,
            "p90": p90,
            "max": max_span,
            "buckets": span_buckets,
        },
        "international_assignment": {
            "count": len(assignees),
            "by_host_country": by_host_country,
            "assignees": sorted(
                assignees,
                key=lambda a: (a["host_country"] or "", a["display_name"] or ""),
            )[:100],
        },
    }


@router.get("/people", summary="People & headcount analytics")
async def insights_people(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_insights_section(user, "people")
    return success_response(
        data=await build_people_payload(db, rng), request=request,
    )


@router.get("/people-geo", summary="People geographic distribution")
async def insights_people_geo(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Headcount by country (ISO alpha-2) + by city within China.

    Point-in-time snapshot of the *active* population — no range applies. The
    world map keys every country on the alpha-2 code; the China drill-down
    keys on the raw city name. ``by_country['CN']`` equals the sum of
    ``by_china_city`` so the world view and the drill-down stay consistent.
    """
    _require_insights_section(user, "people")
    rows = (await db.execute(
        select(
            IdentityPerson.country,
            IdentityPerson.city,
            func.count(),
        )
        .where(
            and_(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
        )
        .group_by(IdentityPerson.country, IdentityPerson.city)
    )).all()

    by_country: dict[str, int] = {}
    by_china_city: dict[str, int] = {}
    for country_raw, city_raw, cnt in rows:
        n = int(cnt)
        code = normalize_country_code(country_raw)
        if code:
            by_country[code] = by_country.get(code, 0) + n
        if code == "CN":
            city = (city_raw or "").strip() or "unknown"
            by_china_city[city] = by_china_city.get(city, 0) + n

    return success_response(
        data={"by_country": by_country, "by_china_city": by_china_city},
        request=request,
    )


# ─── 3. Assets ──────────────────────────────────────────────────────


@router.get("/assets", summary="Hardware lifecycle analytics")
async def insights_assets(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_insights_section(user, "assets")
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()

    # Inventory-shaped counts are LIVE inventory: ``deleted_at IS NULL``
    # everywhere, matching the asset admin list and the impact
    # scorecard's ``hardware_assets`` — without it this tab quietly
    # reports soft-deleted rows that no other surface shows (caught in
    # the 2026-06-12 cross-surface parity audit). The procurement spend
    # trend below intentionally stays unfiltered: purchases are
    # historical flow and deleting an asset row must not erase spend.
    total = (await db.execute(
        select(func.count()).select_from(AssetItem)
        .where(AssetItem.deleted_at.is_(None))
    )).scalar() or 0
    by_status = _str_str_int((await db.execute(
        select(AssetItem.status, func.count())
        .where(AssetItem.deleted_at.is_(None))
        .group_by(AssetItem.status)
    )).all())
    by_category = _str_str_int((await db.execute(
        select(AssetItem.category, func.count())
        .where(AssetItem.deleted_at.is_(None))
        .group_by(AssetItem.category)
    )).all())
    by_brand = _str_str_int((await db.execute(
        select(AssetItem.brand, func.count())
        .where(AssetItem.brand.is_not(None), AssetItem.deleted_at.is_(None))
        .group_by(AssetItem.brand)
        .order_by(func.count().desc())
        .limit(15)
    )).all())

    # Procurement trend
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        proc_rows = (await db.execute(
            select(
                func.date_format(AssetItem.purchase_date, "%Y-%m"),
                func.count(),
                func.coalesce(func.sum(AssetItem.purchase_cost), 0),
            )
            .where(
                and_(
                    AssetItem.purchase_date.is_not(None),
                    AssetItem.purchase_date >= start_date,
                )
            )
            .group_by(func.date_format(AssetItem.purchase_date, "%Y-%m"))
        )).all()
        p_map = {str(k): (int(c), _to_float(s)) for k, c, s in proc_rows}
        trend = [
            {"bucket": b, "count": p_map.get(b, (0, 0.0))[0],
             "spend": p_map.get(b, (0, 0.0))[1]}
            for b in axis
        ]
    else:
        axis = _days_axis(start_date, _bucket_count)
        proc_rows = (await db.execute(
            select(
                AssetItem.purchase_date,
                func.count(),
                func.coalesce(func.sum(AssetItem.purchase_cost), 0),
            )
            .where(
                and_(
                    AssetItem.purchase_date.is_not(None),
                    AssetItem.purchase_date >= start_date,
                )
            )
            .group_by(AssetItem.purchase_date)
        )).all()
        p_map = {d.isoformat(): (int(c), _to_float(s)) for d, c, s in proc_rows}
        trend = [
            {"bucket": b, "count": p_map.get(b, (0, 0.0))[0],
             "spend": p_map.get(b, (0, 0.0))[1]}
            for b in axis
        ]

    # Aging buckets (assigned only)
    age_rows = (await db.execute(
        select(AssetItem.purchase_date, AssetItem.purchase_cost)
        .where(
            and_(
                AssetItem.status == "assigned",
                AssetItem.purchase_date.is_not(None),
                AssetItem.deleted_at.is_(None),
            )
        )
    )).all()
    age_buckets = {"<1y": 0, "1-2y": 0, "2-3y": 0, "3-4y": 0, "4y+": 0}
    age_value_buckets = {k: 0.0 for k in age_buckets}
    ages: list[int] = []
    for d, c in age_rows:
        days = (today - d).days
        ages.append(days)
        v = _to_float(c)
        if days < 365:
            age_buckets["<1y"] += 1
            age_value_buckets["<1y"] += v
        elif days < 730:
            age_buckets["1-2y"] += 1
            age_value_buckets["1-2y"] += v
        elif days < 1095:
            age_buckets["2-3y"] += 1
            age_value_buckets["2-3y"] += v
        elif days < 1460:
            age_buckets["3-4y"] += 1
            age_value_buckets["3-4y"] += v
        else:
            age_buckets["4y+"] += 1
            age_value_buckets["4y+"] += v
    avg_age_days = int(sum(ages) / len(ages)) if ages else 0

    # Warranty buckets
    warranty_rows = (await db.execute(
        select(AssetItem.warranty_until)
        .where(
            and_(
                AssetItem.status.in_(["assigned", "available", "idle"]),
                AssetItem.warranty_until.is_not(None),
                AssetItem.deleted_at.is_(None),
            )
        )
    )).scalars().all()
    warranty_buckets = {"expired": 0, "30d": 0, "90d": 0, "1y": 0, ">1y": 0}
    for wu in warranty_rows:
        if wu < today:
            warranty_buckets["expired"] += 1
        elif wu <= today + timedelta(days=30):
            warranty_buckets["30d"] += 1
        elif wu <= today + timedelta(days=90):
            warranty_buckets["90d"] += 1
        elif wu <= today + timedelta(days=365):
            warranty_buckets["1y"] += 1
        else:
            warranty_buckets[">1y"] += 1

    # Department holding (assigned)
    dept_rows = (await db.execute(
        select(
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_name,
            func.count(),
        )
        .select_from(AssetItem)
        .join(IdentityPerson, IdentityPerson.id == AssetItem.assigned_to)
        .join(
            IdentityOrgMembership,
            and_(
                IdentityOrgMembership.person_id == IdentityPerson.id,
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            ),
        )
        .where(AssetItem.status == "assigned", AssetItem.deleted_at.is_(None))
        .group_by(IdentityOrgMembership.org_path, IdentityOrgMembership.org_node_name)
    )).all()
    by_department_tree = rollup_counts((op, name, int(cnt)) for op, name, cnt in dept_rows)

    # Total assigned value + idle/available value
    val_row = (await db.execute(
        select(
            AssetItem.status,
            func.count(),
            func.coalesce(func.sum(AssetItem.purchase_cost), 0),
        ).where(AssetItem.deleted_at.is_(None)).group_by(AssetItem.status)
    )).all()
    value_by_status = {
        str(s): {"count": int(c), "value": _to_float(v)} for s, c, v in val_row
    }
    total_value = sum(v["value"] for v in value_by_status.values())
    assigned_value = value_by_status.get("assigned", {}).get("value", 0.0)
    idle_value = (
        value_by_status.get("available", {}).get("value", 0.0)
        + value_by_status.get("idle", {}).get("value", 0.0)
    )

    return success_response(
        data={
            "total": int(total),
            "by_status": by_status,
            "by_category": by_category,
            "by_brand": by_brand,
            "trend": trend,
            "axis_unit": unit,
            "age_buckets": age_buckets,
            "age_value_buckets": {k: round(v, 2) for k, v in age_value_buckets.items()},
            "avg_assigned_age_days": avg_age_days,
            "warranty_buckets": warranty_buckets,
            "by_department_tree": by_department_tree,
            "value_by_status": value_by_status,
            "total_value": round(total_value, 2),
            "assigned_value": round(assigned_value, 2),
            "idle_value": round(idle_value, 2),
        },
        request=request,
    )


# ─── 4. Software ────────────────────────────────────────────────────


@router.get("/hardware-finance", summary="Hardware asset book value & depreciation analytics")
async def insights_hardware_finance(
    request: Request,
    rng: RangeKey = Query("24m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Finance/admin hardware financial analytics — net book value by
    category / department / office, disposal P&L, and the CNY-normalised
    depreciation trend. Closes the cost-axis parity gap (software / cloud
    / AI already have an insights tab; hardware did not)."""
    _require_insights_section(user, "hardware-finance")
    from app.domains.asset import financial_service
    from app.domains.asset.schemas import (
        BookValueTrendResponse,
        FinancialStatsResponse,
    )
    months = {"3m": 3, "6m": 6, "12m": 12, "24m": 24}.get(str(rng), 24)
    stats = await financial_service.compute_financial_stats(db, scope="all")
    trend = await financial_service.book_value_trend(db, months=months)
    return success_response(
        data={
            "summary": FinancialStatsResponse.model_validate(stats).model_dump(mode="json"),
            "trend": BookValueTrendResponse.model_validate(trend).model_dump(mode="json"),
        },
        request=request,
    )


@router.get("/software", summary="SaaS seat cost & utilization analytics")
async def insights_software(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_insights_section(user, "software")
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()

    # Per-software roll-up + monthly cost
    rows = (await db.execute(
        select(
            AssetSoftwareCatalog.id,
            AssetSoftwareCatalog.name,
            AssetSoftwareCatalog.status,
            SoftwareLicensePlan.id,
            SoftwareLicensePlan.name,
            SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle,
            SoftwareLicensePlan.max_seats,
            func.count(SoftwareSeatAllocation.id),
        )
        .join(SoftwareLicensePlan, SoftwareLicensePlan.software_id == AssetSoftwareCatalog.id)
        .outerjoin(
            SoftwareSeatAllocation,
            and_(
                SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id,
                SoftwareSeatAllocation.status == "active",
            ),
        )
        .group_by(
            AssetSoftwareCatalog.id, AssetSoftwareCatalog.name, AssetSoftwareCatalog.status,
            SoftwareLicensePlan.id, SoftwareLicensePlan.name, SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle, SoftwareLicensePlan.max_seats,
        )
    )).all()

    per_software: dict[UUID, dict[str, Any]] = {}
    for sw_id, sw_name, sw_status, plan_id, plan_name, price, cycle, max_seats, cnt in rows:
        n = int(cnt)
        if cycle == "monthly":
            mp = _to_float(price) * n
        elif cycle == "yearly":
            mp = _to_float(price) * n / 12.0
        else:
            mp = 0.0
        s = per_software.setdefault(sw_id, {
            "software_id": str(sw_id),
            "software_name": sw_name,
            "software_status": sw_status,
            "active_seats": 0,
            "monthly_cost": 0.0,
            "max_seats_total": 0,
            "plans": [],
        })
        s["active_seats"] += n
        s["monthly_cost"] = round(s["monthly_cost"] + mp, 2)
        if max_seats is not None:
            s["max_seats_total"] += int(max_seats)
        s["plans"].append({
            "plan_id": str(plan_id),
            "plan_name": plan_name,
            "price": _to_float(price),
            "billing_cycle": cycle,
            "max_seats": int(max_seats) if max_seats is not None else None,
            "active_seats": n,
        })
    sw_list = sorted(per_software.values(), key=lambda x: -x["monthly_cost"])

    total_active_seats = sum(s["active_seats"] for s in sw_list)
    total_monthly_cost = round(sum(s["monthly_cost"] for s in sw_list), 2)

    # Trend — grants/revokes
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        grant_rows = (await db.execute(
            select(
                func.date_format(SoftwareSeatAllocation.allocated_at, "%Y-%m"),
                func.count(),
            )
            .where(SoftwareSeatAllocation.allocated_at >= _start_of_day(start_date))
            .group_by(func.date_format(SoftwareSeatAllocation.allocated_at, "%Y-%m"))
        )).all()
        revoke_rows = (await db.execute(
            select(
                func.date_format(SoftwareSeatAllocation.revoked_at, "%Y-%m"),
                func.count(),
            )
            .where(
                and_(
                    SoftwareSeatAllocation.status == "revoked",
                    SoftwareSeatAllocation.revoked_at.is_not(None),
                    SoftwareSeatAllocation.revoked_at >= _start_of_day(start_date),
                )
            )
            .group_by(func.date_format(SoftwareSeatAllocation.revoked_at, "%Y-%m"))
        )).all()
        g_map = {str(k): int(v) for k, v in grant_rows}
        r_map = {str(k): int(v) for k, v in revoke_rows}
    else:
        axis = _days_axis(start_date, _bucket_count)
        grant_rows = (await db.execute(
            select(func.date(SoftwareSeatAllocation.allocated_at), func.count())
            .where(SoftwareSeatAllocation.allocated_at >= _start_of_day(start_date))
            .group_by(func.date(SoftwareSeatAllocation.allocated_at))
        )).all()
        revoke_rows = (await db.execute(
            select(func.date(SoftwareSeatAllocation.revoked_at), func.count())
            .where(
                and_(
                    SoftwareSeatAllocation.status == "revoked",
                    SoftwareSeatAllocation.revoked_at.is_not(None),
                    SoftwareSeatAllocation.revoked_at >= _start_of_day(start_date),
                )
            )
            .group_by(func.date(SoftwareSeatAllocation.revoked_at))
        )).all()
        g_map = {str(k): int(v) for k, v in grant_rows}
        r_map = {str(k): int(v) for k, v in revoke_rows}

    trend = [
        {"bucket": b, "grants": g_map.get(b, 0), "revokes": r_map.get(b, 0),
         "net": g_map.get(b, 0) - r_map.get(b, 0)}
        for b in axis
    ]

    # Allocation age buckets — proxy for stale seats (we don't have last_used)
    age_rows = (await db.execute(
        select(SoftwareSeatAllocation.allocated_at)
        .where(SoftwareSeatAllocation.status == "active")
    )).scalars().all()
    age_buckets = {"<30d": 0, "30-90d": 0, "90-180d": 0, "180-365d": 0, "1y+": 0}
    today_dt = datetime.combine(today, datetime.min.time(), tzinfo=UTC)
    for ts in age_rows:
        ts = _aware_utc(ts)
        days = (today_dt - ts).days if ts else 0
        if days < 30:
            age_buckets["<30d"] += 1
        elif days < 90:
            age_buckets["30-90d"] += 1
        elif days < 180:
            age_buckets["90-180d"] += 1
        elif days < 365:
            age_buckets["180-365d"] += 1
        else:
            age_buckets["1y+"] += 1

    # Cost by department
    cost_dept_rows = (await db.execute(
        select(
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_name,
            SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle,
            func.count(SoftwareSeatAllocation.id),
        )
        .select_from(SoftwareSeatAllocation)
        .join(SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id)
        .join(IdentityPerson, IdentityPerson.id == SoftwareSeatAllocation.user_id)
        .join(
            IdentityOrgMembership,
            and_(
                IdentityOrgMembership.person_id == IdentityPerson.id,
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            ),
        )
        .where(SoftwareSeatAllocation.status == "active")
        .group_by(
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_name,
            SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle,
        )
    )).all()
    # Sum cost per department leaf, then roll up.
    leaf_costs: dict[tuple[str | None, str], float] = {}
    for org_path, leaf_name, price, cycle, cnt in cost_dept_rows:
        n = int(cnt)
        if cycle == "monthly":
            mp = _to_float(price) * n
        elif cycle == "yearly":
            mp = _to_float(price) * n / 12.0
        else:
            mp = 0.0
        key = (org_path, leaf_name)
        leaf_costs[key] = leaf_costs.get(key, 0.0) + mp
    # Convert to (org_path, leaf, count_int) form for rollup_counts. We
    # multiply cost by 100 to keep integer arithmetic, then divide back.
    rollup_input = [
        (op, leaf, int(round(v * 100))) for (op, leaf), v in leaf_costs.items()
    ]
    cost_tree_raw = rollup_counts(rollup_input)
    # rollup_counts returns ints; convert back to float dollars
    cost_by_department_tree = [
        {
            **r,
            "monthly_cost": round(r["total_count"] / 100.0, 2),
            "direct_monthly_cost": round(r["direct_count"] / 100.0, 2),
        }
        for r in cost_tree_raw
    ]
    for r in cost_by_department_tree:
        r.pop("total_count", None)
        r.pop("direct_count", None)

    # Top spenders by software (already in sw_list, top 10)
    top_software = sw_list[:10]

    # Plan utilization — for plans with max_seats, show fill %
    plan_utilization: list[dict[str, Any]] = []
    for s in sw_list:
        for p in s["plans"]:
            if p["max_seats"] is not None and p["max_seats"] > 0:
                plan_utilization.append({
                    "software_name": s["software_name"],
                    "plan_name": p["plan_name"],
                    "active_seats": p["active_seats"],
                    "max_seats": p["max_seats"],
                    "utilization_pct": round(p["active_seats"] / p["max_seats"] * 100, 1),
                })
    plan_utilization.sort(key=lambda x: -x["utilization_pct"])

    # Software with expiry_date — renewal calendar
    renewal_rows = (await db.execute(
        select(
            AssetSoftwareCatalog.id,
            AssetSoftwareCatalog.name,
            AssetSoftwareCatalog.expiry_date,
        )
        .where(
            and_(
                AssetSoftwareCatalog.status == "active",
                AssetSoftwareCatalog.expiry_date.is_not(None),
            )
        )
        .order_by(AssetSoftwareCatalog.expiry_date)
    )).all()
    renewal_calendar = [
        {
            "software_id": str(sid),
            "software_name": name,
            "expiry_date": ed.isoformat(),
            "days_left": (ed - today).days,
        }
        for sid, name, ed in renewal_rows
    ]

    return success_response(
        data={
            "total_active_seats": total_active_seats,
            "total_monthly_cost": total_monthly_cost,
            "by_software": sw_list,
            "top_software": top_software,
            "trend": trend,
            "axis_unit": unit,
            "allocation_age_buckets": age_buckets,
            "cost_by_department_tree": cost_by_department_tree,
            "plan_utilization": plan_utilization,
            "renewal_calendar": renewal_calendar,
        },
        request=request,
    )


# ─── 4b. Software catalog (governance / coverage report) ───────────


@router.get(
    "/renewals",
    summary="Software contract renewal-decision panel",
)
async def insights_renewals(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Software contracts expiring within the next 365 days, each with
    the seat/cost context an admin needs to decide whether (and at what
    size) to renew. Honest by design: shows the 12-month seat MAX +
    ``under_licensed`` flag, NOT a computed "recommended renewal seats".

    Source: ``procurement_contracts.expiry_date`` (status active/signed)
    linked to a software via ``software_procurement_meta.current_contract_id``.
    Software with NO current contract is excluded (nothing to renew).
    Per row: vendor / product / max_seats / current vendor holders /
    12-month seat sparkline / 12-month cost trend / expiry_date /
    days_to_expiry / colour tier / under_licensed.
    """
    _require_insights_section(user, "renewals")
    today = date.today()
    horizon = today + timedelta(days=365)

    # ── Contracts (with software link) expiring within the horizon ──
    contract_rows = (await db.execute(
        select(
            AssetSoftwareCatalog.id,
            AssetSoftwareCatalog.name,
            ProcurementContract.id,
            ProcurementContract.expiry_date,
            ProcurementContract.auto_renew,
            ProcurementSupplier.name,
        )
        .join(
            SoftwareProcurementMeta,
            SoftwareProcurementMeta.software_id == AssetSoftwareCatalog.id,
        )
        .join(
            ProcurementContract,
            ProcurementContract.id == SoftwareProcurementMeta.current_contract_id,
        )
        .outerjoin(
            ProcurementSupplier,
            ProcurementSupplier.id == ProcurementContract.supplier_id,
        )
        .where(
            ProcurementContract.expiry_date.is_not(None),
            ProcurementContract.expiry_date <= horizon,
            ProcurementContract.status.in_(["active", "signed"]),
        )
    )).all()

    if not contract_rows:
        return success_response(data={"rows": [], "as_of": today.isoformat()},
                                request=request)

    software_ids = [r[0] for r in contract_rows]

    # ── Plans (max_seats / price / cycle) + current active seats ──
    plan_rows = (await db.execute(
        select(
            SoftwareLicensePlan.software_id,
            SoftwareLicensePlan.id,
            SoftwareLicensePlan.price,
            SoftwareLicensePlan.billing_cycle,
            SoftwareLicensePlan.max_seats,
            func.count(SoftwareSeatAllocation.id),
        )
        .outerjoin(
            SoftwareSeatAllocation,
            and_(
                SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id,
                SoftwareSeatAllocation.status == "active",
            ),
        )
        .where(SoftwareLicensePlan.software_id.in_(software_ids))
        .group_by(
            SoftwareLicensePlan.software_id, SoftwareLicensePlan.id,
            SoftwareLicensePlan.price, SoftwareLicensePlan.billing_cycle,
            SoftwareLicensePlan.max_seats,
        )
    )).all()

    plan_agg: dict[UUID, dict[str, Any]] = {}
    for sw_id, _plan_id, price, cycle, max_seats, cnt in plan_rows:
        a = plan_agg.setdefault(sw_id, {
            "max_seats_total": 0, "active_seats": 0,
            "monthly_cost": 0.0, "has_max": False, "has_uncapped": False,
        })
        n = int(cnt)
        a["active_seats"] += n
        if max_seats is not None:
            a["max_seats_total"] += int(max_seats)
            a["has_max"] = True
        else:
            # An uncapped plan means no contract ceiling on file — the
            # software can't be "under-licensed" against a finite cap.
            a["has_uncapped"] = True
        if cycle == "monthly":
            a["monthly_cost"] += _to_float(price) * n
        elif cycle == "yearly":
            a["monthly_cost"] += _to_float(price) * n / 12.0

    # ── Vendor license holders (RAW, deduped by product family) ──
    # ``vendor_license_holders`` is a family-keyed total replicated on
    # every plan row of the same ``product_key`` (one stat row per
    # plan). Naively SUM-ing over plans double-counts a multi-plan
    # software (memory: seat count = vendor truth, not bill lines). Take
    # ONE representative value per family (``max``) then sum families.
    # NB: raw holders, NOT ``_effective_seats`` (which clamps to
    # max_seats) — the renewals panel WANTS the raw access count so the
    # ``under_licensed`` over-allocation signal survives.
    from app.domains.asset.billing_service import _stat_family_key
    stat_rows = (await db.execute(
        select(
            SoftwareSeatSyncStat.software_id,
            SoftwareSeatSyncStat.product_key,
            SoftwareSeatSyncStat.plan_id,
            SoftwareSeatSyncStat.vendor_license_holders,
        )
        .where(SoftwareSeatSyncStat.software_id.in_(software_ids))
    )).all()
    fam_by_sw: dict[UUID, dict[str, int]] = {}
    for sw_id, product_key, plan_id, vlh in stat_rows:
        fam = fam_by_sw.setdefault(sw_id, {})
        fkey = _stat_family_key(product_key, plan_id)
        fam[fkey] = max(fam.get(fkey, 0), int(vlh or 0))
    holders_by_sw = {sw: sum(f.values()) for sw, f in fam_by_sw.items()}

    # ── 12-month month-end active-seat snapshots (true point-in-time) ──
    month_ends: list[date] = []
    y, m = today.year, today.month
    for _ in range(12):
        # last day of month (y, m): first of next month minus 1 day
        if m == 12:
            nxt = date(y + 1, 1, 1)
        else:
            nxt = date(y, m + 1, 1)
        month_ends.append(nxt - timedelta(days=1))
        m -= 1
        if m == 0:
            m = 12
            y -= 1
    month_ends.reverse()  # oldest → newest

    # software_id -> [seat_count per month]
    seat_series: dict[UUID, list[int]] = {sw: [0] * 12 for sw in software_ids}
    for idx, me in enumerate(month_ends):
        snap = (await db.execute(
            select(
                SoftwareLicensePlan.software_id,
                func.count(SoftwareSeatAllocation.id),
            )
            .join(
                SoftwareSeatAllocation,
                SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id,
            )
            .where(
                SoftwareLicensePlan.software_id.in_(software_ids),
                func.date(SoftwareSeatAllocation.allocated_at) <= me,
                or_(
                    SoftwareSeatAllocation.revoked_at.is_(None),
                    func.date(SoftwareSeatAllocation.revoked_at) > me,
                ),
            )
            .group_by(SoftwareLicensePlan.software_id)
        )).all()
        for sw_id, cnt in snap:
            if sw_id in seat_series:
                seat_series[sw_id][idx] = int(cnt)

    def _tier(days: int) -> str:
        if days <= 7:
            return "red"
        if days <= 30:
            return "orange"
        if days < 90:
            return "yellow"
        return "green"

    rows_out: list[dict[str, Any]] = []
    for sw_id, sw_name, contract_id, expiry, auto_renew, vendor in contract_rows:
        agg = plan_agg.get(sw_id, {})
        max_seats_total = agg.get("max_seats_total", 0)
        holders = holders_by_sw.get(sw_id, 0)
        days_to_expiry = (expiry - today).days
        seats = seat_series.get(sw_id, [0] * 12)
        per_seat = (
            agg.get("monthly_cost", 0.0) / agg["active_seats"]
            if agg.get("active_seats")
            else 0.0
        )
        rows_out.append({
            "software_id": str(sw_id),
            "software_name": sw_name,
            "vendor": vendor,
            "contract_id": str(contract_id),
            "expiry_date": expiry.isoformat(),
            "days_to_expiry": days_to_expiry,
            "auto_renew": bool(auto_renew),
            "color_tier": _tier(days_to_expiry),
            "max_seats": max_seats_total if agg.get("has_max") else None,
            "current_holders": holders,
            "active_seats": agg.get("active_seats", 0),
            # Honest: the 12-month MAX, not a recommended renewal size.
            "seats_12m_max": max(seats) if seats else 0,
            "seat_sparkline": seats,
            "cost_sparkline": [round(s * per_seat, 2) for s in seats],
            # under-licensed only when every plan has a finite ceiling
            # (an uncapped/unlimited plan can't be under-licensed) and
            # the raw vendor holders exceed the summed contract cap.
            "under_licensed": (
                agg.get("has_max", False)
                and not agg.get("has_uncapped", False)
                and holders > max_seats_total
            ),
        })

    rows_out.sort(key=lambda r: r["days_to_expiry"])
    return success_response(
        data={"rows": rows_out, "as_of": today.isoformat()},
        request=request,
    )


@router.get(
    "/software-catalog",
    summary="Per-software governance report (catalog facts, no seats)",
)
async def insights_software_catalog(
    request: Request,
    rng: RangeKey = Query("30d", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Catalog-level governance + coverage report.

    Complementary to ``/software`` (which is seat-cost focused). This
    endpoint returns:

      * KPIs — total / active / inactive software count, owner-coverage,
        IT-contact coverage, SSO-policy adoption (``offboarding_policy
        == 'sso_cascade'``), connector-binding adoption, and the count
        of catalog rows added inside the requested ``range`` window
        (the only place ``range`` matters here — every other field is
        a point-in-time snapshot).
      * ``by_category`` — distribution of catalog rows by ``category``.
      * ``by_offboarding_policy`` — split between sso_cascade vs
        auto_revoke (helpful as the "do we have a plan when someone
        leaves?" health signal).
      * ``rows`` — full per-software list with name, category, status,
        asset_type, owner / IT contact (display name + email), owner's
        department, ``offboarding_policy``, ``has_connector`` flag,
        ``created_at`` / ``updated_at``.

    Read gate: ``_require_admin`` (admin or management role).
    """
    _require_insights_section(user, "software-catalog")
    start_date, _bucket_count, _unit = _range_to_window(rng)
    window_start = _start_of_day(start_date)

    # Aliases let us join the same identity_persons table twice without
    # collision (owner + IT contact + owner's department membership all
    # in one go) and pick the matching department label without a
    # second round trip per row.
    owner_alias = aliased(IdentityPerson)
    it_alias = aliased(IdentityPerson)
    owner_dept_alias = aliased(IdentityOrgMembership)

    rows = (await db.execute(
        select(
            AssetSoftwareCatalog.id,
            AssetSoftwareCatalog.name,
            AssetSoftwareCatalog.asset_type,
            AssetSoftwareCatalog.category,
            AssetSoftwareCatalog.status,
            AssetSoftwareCatalog.offboarding_policy,
            AssetSoftwareCatalog.department.label("dept_override"),
            AssetSoftwareCatalog.created_at,
            AssetSoftwareCatalog.updated_at,
            AssetSoftwareCatalog.owner_person_id,
            AssetSoftwareCatalog.it_contact_person_id,
            owner_alias.display_name.label("owner_name"),
            owner_alias.work_email.label("owner_email"),
            it_alias.display_name.label("it_name"),
            it_alias.work_email.label("it_email"),
            owner_dept_alias.org_node_name.label("owner_department"),
            SoftwareConnector.id.label("connector_id"),
        )
        .select_from(AssetSoftwareCatalog)
        .outerjoin(owner_alias, AssetSoftwareCatalog.owner_person_id == owner_alias.id)
        .outerjoin(it_alias, AssetSoftwareCatalog.it_contact_person_id == it_alias.id)
        .outerjoin(
            owner_dept_alias,
            and_(
                owner_dept_alias.person_id == AssetSoftwareCatalog.owner_person_id,
                owner_dept_alias.is_primary.is_(True),
                owner_dept_alias.deleted_at.is_(None),
            ),
        )
        .outerjoin(
            SoftwareConnector,
            and_(
                SoftwareConnector.software_id == AssetSoftwareCatalog.id,
                SoftwareConnector.is_active.is_(True),
            ),
        )
        .order_by(AssetSoftwareCatalog.name)
    )).all()

    # Deduplicate by software id — the org-membership join can return
    # multiple rows per software when the owner happens to hold more
    # than one ``is_primary=True`` membership across source systems
    # (HRIS sync can occasionally land that state during a re-org).
    # Without this dedupe, KPI totals would double-count those rows.
    seen_ids: set[Any] = set()
    deduped: list[Any] = []
    for r in rows:
        if r.id in seen_ids:
            continue
        seen_ids.add(r.id)
        deduped.append(r)
    rows = deduped

    total = len(rows)
    active = sum(1 for r in rows if r.status == "active")
    inactive = total - active
    has_owner = sum(1 for r in rows if r.owner_person_id is not None)
    has_it = sum(1 for r in rows if r.it_contact_person_id is not None)
    sso_count = sum(1 for r in rows if r.offboarding_policy == "sso_cascade")
    connector_count = sum(1 for r in rows if r.connector_id is not None)
    recently_added = sum(
        1 for r in rows
        if _aware_utc(r.created_at) is not None
        and _aware_utc(r.created_at) >= window_start  # type: ignore[operator]
    )

    by_category: dict[str, int] = {}
    for r in rows:
        cat = (r.category or "other").lower()
        by_category[cat] = by_category.get(cat, 0) + 1

    by_policy: dict[str, int] = {}
    for r in rows:
        pol = r.offboarding_policy or "sso_cascade"
        by_policy[pol] = by_policy.get(pol, 0) + 1

    def pct(num: int, denom: int) -> float:
        return round(num * 100.0 / denom, 1) if denom else 0.0

    table_rows = [
        {
            "id": str(r.id),
            "name": r.name,
            "asset_type": r.asset_type,
            "category": r.category or "other",
            "status": r.status,
            "owner_person_id": str(r.owner_person_id) if r.owner_person_id else None,
            "owner_name": r.owner_name,
            "owner_email": r.owner_email,
            "it_contact_person_id": (
                str(r.it_contact_person_id) if r.it_contact_person_id else None
            ),
            "it_contact_name": r.it_name,
            "it_contact_email": r.it_email,
            "owner_department": r.owner_department or r.dept_override,
            "offboarding_policy": r.offboarding_policy or "sso_cascade",
            "has_connector": bool(r.connector_id),
            "created_at": (
                _aware_utc(r.created_at).isoformat()  # type: ignore[union-attr]
                if r.created_at else None
            ),
            "updated_at": (
                _aware_utc(r.updated_at).isoformat()  # type: ignore[union-attr]
                if r.updated_at else None
            ),
        }
        for r in rows
    ]

    return success_response(
        data={
            "kpis": {
                "total": total,
                "active": active,
                "inactive": inactive,
                "has_owner": has_owner,
                "has_owner_pct": pct(has_owner, total),
                "has_it_contact": has_it,
                "has_it_contact_pct": pct(has_it, total),
                "sso_cascade": sso_count,
                "sso_cascade_pct": pct(sso_count, total),
                "with_connector": connector_count,
                "with_connector_pct": pct(connector_count, total),
                "recently_added": recently_added,
                "range": rng,
            },
            "by_category": by_category,
            "by_offboarding_policy": by_policy,
            "rows": table_rows,
        },
        request=request,
    )


# ─── 5. Operations ──────────────────────────────────────────────────


@router.get("/operations", summary="Provisioning SLA & throughput analytics")
async def insights_operations(
    request: Request,
    rng: RangeKey = Query("30d", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_admin(user)
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()
    window_start = _start_of_day(start_date)

    # Status breakdown — all-time (live backlog matters for ops)
    by_status = _str_str_int((await db.execute(
        select(ProvisioningTask.status, func.count())
        .group_by(ProvisioningTask.status)
    )).all())

    # Source breakdown over window
    by_source = _str_str_int((await db.execute(
        select(ProvisioningTask.source, func.count())
        .where(ProvisioningTask.created_at >= window_start)
        .group_by(ProvisioningTask.source)
    )).all())

    # Execution-mode breakdown (auto vs manual ratio)
    by_exec_mode = _str_str_int((await db.execute(
        select(ProvisioningTask.execution_mode, func.count())
        .where(ProvisioningTask.created_at >= window_start)
        .group_by(ProvisioningTask.execution_mode)
    )).all())
    total_in_window = sum(by_exec_mode.values())
    auto_count = by_exec_mode.get(EXEC_AUTO_API, 0) + by_exec_mode.get(EXEC_SSO_AUTO, 0)
    automation_rate = round(auto_count / total_in_window * 100, 1) if total_in_window else 0.0

    # Failure rate
    by_status_window = _str_str_int((await db.execute(
        select(ProvisioningTask.status, func.count())
        .where(ProvisioningTask.created_at >= window_start)
        .group_by(ProvisioningTask.status)
    )).all())
    failed_count = by_status_window.get(STATUS_FAILED, 0)
    failure_rate = round(failed_count / total_in_window * 100, 2) if total_in_window else 0.0

    # Avg completion time (created_at → completed_at) for completed tasks in window
    duration_rows = (await db.execute(
        select(
            ProvisioningTask.execution_mode,
            ProvisioningTask.created_at,
            ProvisioningTask.completed_at,
        )
        .where(
            and_(
                ProvisioningTask.status == STATUS_COMPLETED,
                ProvisioningTask.completed_at.is_not(None),
                ProvisioningTask.created_at >= window_start,
            )
        )
    )).all()
    overall_durations: list[float] = []
    auto_durations: list[float] = []
    manual_durations: list[float] = []
    for em, c, comp in duration_rows:
        if not c or not comp:
            continue
        secs = (_aware_utc(comp) - _aware_utc(c)).total_seconds()
        if secs < 0:
            continue
        overall_durations.append(secs)
        if em == EXEC_MANUAL_NOTIFY:
            manual_durations.append(secs)
        else:
            auto_durations.append(secs)
    overall_avg_hours = round(sum(overall_durations) / len(overall_durations) / 3600, 2) if overall_durations else 0.0
    auto_avg_hours = round(sum(auto_durations) / len(auto_durations) / 3600, 2) if auto_durations else 0.0
    manual_avg_hours = round(sum(manual_durations) / len(manual_durations) / 3600, 2) if manual_durations else 0.0

    # Throughput trend
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        rows = (await db.execute(
            select(
                func.date_format(ProvisioningTask.created_at, "%Y-%m"),
                ProvisioningTask.source,
                func.count(),
            )
            .where(ProvisioningTask.created_at >= window_start)
            .group_by(
                func.date_format(ProvisioningTask.created_at, "%Y-%m"),
                ProvisioningTask.source,
            )
        )).all()
        m_by: dict[str, dict[str, int]] = {}
        for k, src, n in rows:
            m_by.setdefault(str(k), {})[str(src)] = int(n)
        trend = [
            {"bucket": b, "by_source": m_by.get(b, {}),
             "total": sum(m_by.get(b, {}).values())}
            for b in axis
        ]
    else:
        axis = _days_axis(start_date, _bucket_count)
        rows = (await db.execute(
            select(
                func.date(ProvisioningTask.created_at),
                ProvisioningTask.source,
                func.count(),
            )
            .where(ProvisioningTask.created_at >= window_start)
            .group_by(
                func.date(ProvisioningTask.created_at),
                ProvisioningTask.source,
            )
        )).all()
        m_by = {}
        for k, src, n in rows:
            m_by.setdefault(str(k), {})[str(src)] = int(n)
        trend = [
            {"bucket": b, "by_source": m_by.get(b, {}),
             "total": sum(m_by.get(b, {}).values())}
            for b in axis
        ]

    # Stale in_progress
    stale_threshold = _start_of_day(today - timedelta(days=7))
    stale_rows = (await db.execute(
        select(
            ProvisioningTask.target_type,
            func.count(),
        )
        .where(
            and_(
                ProvisioningTask.status == STATUS_IN_PROGRESS,
                ProvisioningTask.started_at.is_not(None),
                ProvisioningTask.started_at <= stale_threshold,
            )
        )
        .group_by(ProvisioningTask.target_type)
    )).all()
    stale_by_target = _str_str_int(stale_rows)
    stale_total = sum(stale_by_target.values())

    # Top failing target types in window
    failing_target_rows = (await db.execute(
        select(ProvisioningTask.target_type, func.count())
        .where(
            and_(
                ProvisioningTask.status == STATUS_FAILED,
                ProvisioningTask.created_at >= window_start,
            )
        )
        .group_by(ProvisioningTask.target_type)
        .order_by(func.count().desc())
        .limit(10)
    )).all()
    top_failing_targets = _str_str_int(failing_target_rows)

    # Pending transfers age
    transfer_rows = (await db.execute(
        select(AssetTransferRequest.created_at)
        .where(AssetTransferRequest.status == XFER_PENDING)
    )).scalars().all()
    today_dt = _start_of_day(today)
    transfer_age_buckets = {"<1d": 0, "1-3d": 0, "3-7d": 0, "7-14d": 0, "14d+": 0}
    for c in transfer_rows:
        days = (today_dt - _aware_utc(c)).days
        if days < 1:
            transfer_age_buckets["<1d"] += 1
        elif days < 3:
            transfer_age_buckets["1-3d"] += 1
        elif days < 7:
            transfer_age_buckets["3-7d"] += 1
        elif days < 14:
            transfer_age_buckets["7-14d"] += 1
        else:
            transfer_age_buckets["14d+"] += 1
    open_transfers = sum(transfer_age_buckets.values())

    return success_response(
        data={
            "by_status": by_status,
            "by_status_window": by_status_window,
            "by_source": by_source,
            "by_execution_mode": by_exec_mode,
            "automation_rate_pct": automation_rate,
            "failure_rate_pct": failure_rate,
            "completion_avg_hours": {
                "overall": overall_avg_hours,
                "auto": auto_avg_hours,
                "manual": manual_avg_hours,
            },
            "trend": trend,
            "axis_unit": unit,
            "stale_by_target": stale_by_target,
            "stale_total": stale_total,
            "top_failing_targets": top_failing_targets,
            "transfer_age_buckets": transfer_age_buckets,
            "open_transfers": open_transfers,
        },
        request=request,
    )


# ─── 5b. Automation health (execution-plane PR5, 2026-06-10) ────────


@router.get(
    "/automation-health",
    summary="Execution-plane health: did the automation actually work?",
)
async def insights_automation_health(
    request: Request,
    days: int = Query(7, ge=1, le=90),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """One card answering "did the platform's automated actions
    actually succeed?" — the question the 2026-06-10 prod audit had to
    answer by hand-querying audit_logs / notification_outbox / TiDB
    (1.4% notification delivery, a 463-attempt revoke storm and 23
    expected-failure provisioning rows all sat invisible for weeks).

    Four lanes over a rolling window (default 7d):

    * **notifications** — outbox rows created in-window by status +
      live ``critical_dead`` (guaranteed-delivery templates that died,
      all-time: a dead critical alert stays a problem until handled).
    * **cloud_revoke** — ``cloud.grant.revoke`` audit outcomes +
      failed retry attempts in-window, plus the LIVE stuck/escalated
      backlog (requires_manual_cloud_revoke).
    * **provisioning** — tasks created in-window by status +
      transient-retry / dead-letter audit counts.
    * **cron** — scheduled_job_runs in-window: failure rate + the top
      failing jobs by name.

    Read-only composition of existing telemetry; no new state.
    """
    _require_admin(user)
    window_start = datetime.now(UTC) - timedelta(days=days)

    # ── Notifications ────────────────────────────────────────────
    from app.notifications.models import NotificationOutbox
    from app.notifications.service import _CRITICAL_TEMPLATE_PREFIXES

    notif_by_status = _str_str_int((await db.execute(
        select(NotificationOutbox.status, func.count())
        .where(NotificationOutbox.created_at >= window_start)
        .group_by(NotificationOutbox.status)
    )).all())
    notif_total = sum(notif_by_status.values())
    notif_delivered = notif_by_status.get("delivered", 0)
    # Acknowledged rows ('cancelled by admin' — consciously closed)
    # are excluded, mirroring the outbox-summary counter (2026-06-10
    # dead-letter ack).
    critical_dead = (await db.execute(
        select(func.count(NotificationOutbox.id)).where(
            NotificationOutbox.status == "dead",
            or_(
                NotificationOutbox.last_error.is_(None),
                ~NotificationOutbox.last_error.like("cancelled by admin%"),
            ),
            or_(*[
                NotificationOutbox.template_key.like(f"{p}%")
                for p in _CRITICAL_TEMPLATE_PREFIXES
            ]),
        )
    )).scalar() or 0

    # ── Cloud revoke ─────────────────────────────────────────────
    from app.domains.cmdb.models import CmdbCloudAccessGrant

    revoke_rows = (await db.execute(
        select(AuditLog.status, func.count())
        .where(
            AuditLog.action == "cloud.grant.revoke",
            AuditLog.created_at >= window_start,
        )
        .group_by(AuditLog.status)
    )).all()
    revoke_by_status = {str(s): int(n) for s, n in revoke_rows}
    revoke_retry_failures = (await db.execute(
        select(func.count(AuditLog.id)).where(
            AuditLog.action.in_((
                "cloud.grant.revoke_retry",
                "cloud.grant.revoke_attempt_failed",
            )),
            AuditLog.status == "failure",
            AuditLog.created_at >= window_start,
        )
    )).scalar() or 0
    stuck_pending = (await db.execute(
        select(func.count(CmdbCloudAccessGrant.id)).where(
            CmdbCloudAccessGrant.status == "revoked",
            func.json_extract(
                CmdbCloudAccessGrant.detail,
                "$.requires_manual_cloud_revoke",
            ) == sa_true(),
        )
    )).scalar() or 0
    escalated_pending = (await db.execute(
        select(func.count(CmdbCloudAccessGrant.id)).where(
            CmdbCloudAccessGrant.status == "revoked",
            func.json_extract(
                CmdbCloudAccessGrant.detail,
                "$.requires_manual_cloud_revoke",
            ) == sa_true(),
            func.json_extract(
                CmdbCloudAccessGrant.detail,
                "$.manual_revoke_escalated_at",
            ).isnot(None),
        )
    )).scalar() or 0
    # Post-review fix (2026-06-10): the "云侧未回收 N" tile previously
    # deep-linked to the scopes LIST, which has no stuck indicator —
    # the admin had to open every scope's 已结束 tab to hunt for the
    # rows. Hand the card the top stuck scopes so each renders as a
    # direct link to /admin/cloud/scopes/{id}?tab=history. Two queries
    # (group by the String(36) scope_id, then name lookup) — the
    # Uuid-PK ↔ String-FK pair doesn't join cleanly in SQL.
    stuck_scopes: list[dict[str, Any]] = []
    if stuck_pending:
        from app.domains.cmdb.models import CmdbCloudScope

        per_scope = (await db.execute(
            select(
                CmdbCloudAccessGrant.scope_id,
                func.count(CmdbCloudAccessGrant.id),
            )
            .where(
                CmdbCloudAccessGrant.status == "revoked",
                func.json_extract(
                    CmdbCloudAccessGrant.detail,
                    "$.requires_manual_cloud_revoke",
                ) == sa_true(),
            )
            .group_by(CmdbCloudAccessGrant.scope_id)
            .order_by(func.count(CmdbCloudAccessGrant.id).desc())
            .limit(5)
        )).all()
        scope_ids = [UUID(str(sid)) for sid, _n in per_scope]
        name_rows = (await db.execute(
            select(CmdbCloudScope.id, CmdbCloudScope.display_name)
            .where(CmdbCloudScope.id.in_(scope_ids))
        )).all() if scope_ids else []
        names = {str(i): n for i, n in name_rows}
        stuck_scopes = [
            {
                "scope_id": str(sid),
                "display_name": names.get(str(sid)) or str(sid)[:8],
                "count": int(n),
            }
            for sid, n in per_scope
        ]

    # ── Provisioning ─────────────────────────────────────────────
    prov_by_status = _str_str_int((await db.execute(
        select(ProvisioningTask.status, func.count())
        .where(ProvisioningTask.created_at >= window_start)
        .group_by(ProvisioningTask.status)
    )).all())
    prov_total = sum(prov_by_status.values())
    transient_retries = (await db.execute(
        select(func.count(AuditLog.id)).where(
            AuditLog.action == "provisioning.auto_api_transient_retry",
            AuditLog.created_at >= window_start,
        )
    )).scalar() or 0
    dead_lettered = (await db.execute(
        select(func.count(AuditLog.id)).where(
            AuditLog.action == "provisioning.dead_lettered",
            AuditLog.created_at >= window_start,
        )
    )).scalar() or 0

    # ── Cron ─────────────────────────────────────────────────────
    from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun

    run_rows = (await db.execute(
        select(ScheduledJobRun.status, func.count())
        .where(ScheduledJobRun.started_at >= window_start)
        .group_by(ScheduledJobRun.status)
    )).all()
    runs_by_status = {str(s): int(n) for s, n in run_rows}
    runs_total = sum(runs_by_status.values())
    runs_failed = runs_by_status.get("failed", 0)
    failing_jobs_rows = (await db.execute(
        select(ScheduledJob.name, func.count())
        .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
        .where(
            ScheduledJobRun.started_at >= window_start,
            ScheduledJobRun.status == "failed",
        )
        .group_by(ScheduledJob.name)
        .order_by(func.count().desc())
        .limit(5)
    )).all()

    def _pct(num: int, denom: int) -> float:
        return round(num / denom * 100, 1) if denom else 100.0

    # ── Cloud sync (per-capability failures) ─────────────────────
    from app.domains.cmdb.sync_status import build_cloud_sync_health
    cloud_sync = await build_cloud_sync_health(db, since=window_start)

    return success_response(
        data={
            "window_days": days,
            "cloud_sync": cloud_sync,
            "notifications": {
                "total": notif_total,
                "by_status": notif_by_status,
                "delivery_rate": _pct(notif_delivered, notif_total),
                "critical_dead": int(critical_dead),
            },
            "cloud_revoke": {
                "succeeded": revoke_by_status.get("success", 0),
                "failed": revoke_by_status.get("failure", 0),
                "retry_failures": int(revoke_retry_failures),
                "stuck_pending": int(stuck_pending),
                "escalated_pending": int(escalated_pending),
                "stuck_scopes": stuck_scopes,
            },
            "provisioning": {
                "total": prov_total,
                "by_status": prov_by_status,
                "success_rate": _pct(
                    prov_by_status.get(STATUS_COMPLETED, 0), prov_total,
                ),
                "transient_retries": int(transient_retries),
                "dead_lettered": int(dead_lettered),
            },
            "cron": {
                "runs": runs_total,
                "failed": runs_failed,
                "failure_rate": _pct(runs_failed, runs_total) if runs_total else 0.0,
                "failing_jobs": [
                    {"name": str(n), "failures": int(c)}
                    for n, c in failing_jobs_rows
                ],
            },
        },
        request=request,
    )


# ─── 6. Lifecycle ───────────────────────────────────────────────────


async def build_lifecycle_payload(
    db: AsyncSession, rng: str,
) -> dict[str, Any]:
    """Aggregate the 入离职洞察 (Lifecycle) payload for one time window.

    Extracted from the ``GET /lifecycle`` handler (2026-07-08) so the AI
    diagnosis generator (``insights_hr_ai``) reads EXACTLY what the tab
    renders.
    """
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()
    window_start = _start_of_day(start_date)

    # Onboarding throughput — count of hires per bucket
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        hires_rows = (await db.execute(
            select(func.date_format(IdentityPerson.hire_date, "%Y-%m"), func.count())
            .where(
                and_(
                    IdentityPerson.hire_date.is_not(None),
                    IdentityPerson.hire_date >= start_date,
                )
            )
            .group_by(func.date_format(IdentityPerson.hire_date, "%Y-%m"))
        )).all()
        h_map = {str(k): int(v) for k, v in hires_rows}
    else:
        axis = _days_axis(start_date, _bucket_count)
        hires_rows = (await db.execute(
            select(IdentityPerson.hire_date, func.count())
            .where(
                and_(
                    IdentityPerson.hire_date.is_not(None),
                    IdentityPerson.hire_date >= start_date,
                )
            )
            .group_by(IdentityPerson.hire_date)
        )).all()
        h_map = {d.isoformat(): int(n) for d, n in hires_rows}

    onboarding_trend = [{"bucket": b, "hires": h_map.get(b, 0)} for b in axis]

    # Avg onboarding duration (hire_date -> last completed onboarding task)
    onb_rows = (await db.execute(
        select(
            ProvisioningTask.user_id,
            func.max(ProvisioningTask.completed_at),
        )
        .where(
            and_(
                ProvisioningTask.source == "onboarding",
                ProvisioningTask.status == STATUS_COMPLETED,
                ProvisioningTask.completed_at.is_not(None),
            )
        )
        .group_by(ProvisioningTask.user_id)
    )).all()
    last_onb: dict[UUID, datetime] = {uid: ts for uid, ts in onb_rows}
    hired_in_window = (await db.execute(
        select(IdentityPerson.id, IdentityPerson.hire_date).where(
            and_(
                IdentityPerson.hire_date.is_not(None),
                IdentityPerson.hire_date >= start_date,
            )
        )
    )).all()
    durations: list[int] = []
    for uid, hd in hired_in_window:
        last = last_onb.get(uid)
        if last and hd:
            d = (last.date() - hd).days
            if d >= 0:
                durations.append(d)
    avg_onboarding_days = round(sum(durations) / len(durations), 1) if durations else 0.0
    onboarding_buckets = {"same_day": 0, "1-3d": 0, "3-7d": 0, "7-14d": 0, "14d+": 0}
    for d in durations:
        if d == 0:
            onboarding_buckets["same_day"] += 1
        elif d <= 3:
            onboarding_buckets["1-3d"] += 1
        elif d <= 7:
            onboarding_buckets["3-7d"] += 1
        elif d <= 14:
            onboarding_buckets["7-14d"] += 1
        else:
            onboarding_buckets["14d+"] += 1

    # Offboarding pipeline — process status counts (live)
    proc_status_counts = _str_str_int((await db.execute(
        select(OffboardingProcess.status, func.count())
        .group_by(OffboardingProcess.status)
    )).all())

    blocked_past_due = (await db.execute(
        select(func.count())
        .select_from(OffboardingProcess)
        .where(
            and_(
                OffboardingProcess.status == PROC_BLOCKED,
                OffboardingProcess.leaver_date <= today,
            )
        )
    )).scalar() or 0

    # Offboarding completion duration (created_at → completed_at)
    off_dur_rows = (await db.execute(
        select(OffboardingProcess.created_at, OffboardingProcess.completed_at)
        .where(
            and_(
                OffboardingProcess.status == PROC_COMPLETED,
                OffboardingProcess.completed_at.is_not(None),
                OffboardingProcess.created_at >= window_start,
            )
        )
    )).all()
    off_durations = [
        (_aware_utc(comp) - _aware_utc(c)).days
        for c, comp in off_dur_rows
        if c and comp
    ]
    avg_offboarding_days = round(sum(off_durations) / len(off_durations), 1) if off_durations else 0.0

    # Blocker breakdown by resource_type (pending only)
    blocker_rows = (await db.execute(
        select(OffboardingBlocker.resource_type, func.count())
        .where(OffboardingBlocker.resolution == RESO_PENDING)
        .group_by(OffboardingBlocker.resource_type)
    )).all()
    blockers_by_resource = _str_str_int(blocker_rows)

    # Blocker aging (pending only)
    pending_blocker_ages = (await db.execute(
        select(OffboardingBlocker.created_at)
        .where(OffboardingBlocker.resolution == RESO_PENDING)
    )).scalars().all()
    today_dt = _start_of_day(today)
    blocker_age_buckets = {"<1d": 0, "1-3d": 0, "3-7d": 0, "7-14d": 0, "14d+": 0}
    for c in pending_blocker_ages:
        days = (today_dt - _aware_utc(c)).days
        if days < 1:
            blocker_age_buckets["<1d"] += 1
        elif days < 3:
            blocker_age_buckets["1-3d"] += 1
        elif days < 7:
            blocker_age_buckets["3-7d"] += 1
        elif days < 14:
            blocker_age_buckets["7-14d"] += 1
        else:
            blocker_age_buckets["14d+"] += 1

    # Recovered asset value (transfers from leavers, completed in window)
    recovered_value_rows = (await db.execute(
        select(func.coalesce(func.sum(AssetItem.purchase_cost), 0), func.count())
        .select_from(AssetTransferRequest)
        .join(
            AssetItem,
            and_(
                AssetItem.id == AssetTransferRequest.asset_ref_id,
                AssetTransferRequest.asset_type == "hardware",
            ),
        )
        .where(
            and_(
                AssetTransferRequest.status == "completed",
                AssetTransferRequest.completed_at.is_not(None),
                AssetTransferRequest.completed_at >= window_start,
                AssetTransferRequest.offboarding_process_id.is_not(None),
            )
        )
    )).one()
    recovered_value = _to_float(recovered_value_rows[0])
    recovered_count = int(recovered_value_rows[1] or 0)

    # Upcoming leavers — next 30 days, with their pending blockers
    upcoming = (await db.execute(
        select(
            OffboardingProcess.id,
            OffboardingProcess.person_id,
            OffboardingProcess.leaver_date,
            OffboardingProcess.status,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
        )
        .join(IdentityPerson, IdentityPerson.id == OffboardingProcess.person_id)
        .where(
            and_(
                OffboardingProcess.leaver_date >= today,
                OffboardingProcess.leaver_date <= today + timedelta(days=30),
                OffboardingProcess.status.in_([PROC_BLOCKED, "ready", "scanning"]),
            )
        )
        .order_by(OffboardingProcess.leaver_date)
    )).all()
    proc_ids = [r[0] for r in upcoming]
    blocker_counts: dict[UUID, int] = {}
    if proc_ids:
        blocker_counts = dict((await db.execute(
            select(OffboardingBlocker.process_id, func.count())
            .where(
                and_(
                    OffboardingBlocker.process_id.in_(proc_ids),
                    OffboardingBlocker.resolution == RESO_PENDING,
                )
            )
            .group_by(OffboardingBlocker.process_id)
        )).all())
    upcoming_leavers = [
        {
            "process_id": str(pid),
            "person_id": str(personid),
            "display_name": name,
            "work_email": email,
            "leaver_date": ld.isoformat(),
            "days_left": (ld - today).days,
            "status": st,
            "pending_blockers": int(blocker_counts.get(pid, 0)),
        }
        for pid, personid, ld, st, name, email in upcoming
    ]

    return {
        "onboarding_trend": onboarding_trend,
        "axis_unit": unit,
        "avg_onboarding_days": avg_onboarding_days,
        "onboarding_buckets": onboarding_buckets,
        "offboarding_status_counts": proc_status_counts,
        "blocked_past_due": int(blocked_past_due),
        "avg_offboarding_days": avg_offboarding_days,
        "blockers_by_resource": blockers_by_resource,
        "blocker_age_buckets": blocker_age_buckets,
        "recovered_asset_value": round(recovered_value, 2),
        "recovered_asset_count": recovered_count,
        "upcoming_leavers": upcoming_leavers,
    }


@router.get("/lifecycle", summary="Onboarding & offboarding health analytics")
async def insights_lifecycle(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_insights_section(user, "lifecycle")
    return success_response(
        data=await build_lifecycle_payload(db, rng), request=request,
    )


# ─── 7. Security ────────────────────────────────────────────────────


@router.get("/security", summary="Audit-trail & privileged-access analytics")
async def insights_security(
    request: Request,
    rng: RangeKey = Query("30d", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_insights_section(user, "security")
    start_date, _bucket_count, unit = _range_to_window(rng)
    today = date.today()
    window_start = _start_of_day(start_date)

    # Top actions in window
    action_rows = (await db.execute(
        select(AuditLog.action, func.count())
        .where(AuditLog.created_at >= window_start)
        .group_by(AuditLog.action)
        .order_by(func.count().desc())
        .limit(15)
    )).all()
    top_actions = _str_str_int(action_rows)

    # Top actors (resolve display names)
    actor_rows = (await db.execute(
        select(AuditLog.actor_user_id, func.count())
        .where(
            and_(
                AuditLog.created_at >= window_start,
                AuditLog.actor_user_id.is_not(None),
                AuditLog.actor_type == "user",
            )
        )
        .group_by(AuditLog.actor_user_id)
        .order_by(func.count().desc())
        .limit(10)
    )).all()
    actor_ids = [a for a, _ in actor_rows if a is not None]
    name_map: dict[UUID, str] = {}
    if actor_ids:
        person_rows = (await db.execute(
            select(IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email)
            .where(IdentityPerson.id.in_(actor_ids))
        )).all()
        for pid, name, email in person_rows:
            name_map[pid] = name or email or str(pid)
    top_actors = [
        {
            "actor_id": str(aid),
            "label": name_map.get(aid, str(aid)),
            "count": int(c),
        }
        for aid, c in actor_rows
        if aid is not None
    ]

    # Status breakdown (success/failure)
    status_rows = (await db.execute(
        select(AuditLog.status, func.count())
        .where(AuditLog.created_at >= window_start)
        .group_by(AuditLog.status)
    )).all()
    by_status = _str_str_int(status_rows)
    total_in_window = sum(by_status.values())
    failure_count = by_status.get("failure", 0)
    failure_rate = round(failure_count / total_in_window * 100, 2) if total_in_window else 0.0

    # Audit volume trend
    if unit == "month":
        axis = _months_axis(start_date, _bucket_count)
        vol_rows = (await db.execute(
            select(func.date_format(AuditLog.created_at, "%Y-%m"), func.count())
            .where(AuditLog.created_at >= window_start)
            .group_by(func.date_format(AuditLog.created_at, "%Y-%m"))
        )).all()
        v_map = {str(k): int(n) for k, n in vol_rows}
    else:
        axis = _days_axis(start_date, _bucket_count)
        vol_rows = (await db.execute(
            select(func.date(AuditLog.created_at), func.count())
            .where(AuditLog.created_at >= window_start)
            .group_by(func.date(AuditLog.created_at))
        )).all()
        v_map = {str(k): int(n) for k, n in vol_rows}
    volume_trend = [{"bucket": b, "count": v_map.get(b, 0)} for b in axis]

    # Privileged role holders (admin / auditor / asset_admin / hr / finance).
    # AuthUserRole stores ``role_id`` only; the human-readable name lives on
    # AuthRole, so we have to join.
    privileged_role_names = [
        "global_admin", "it_admin", "asset_admin", "asset_global_admin",
        "security_auditor", "auditor", "hr", "finance",
    ]
    priv_rows = (await db.execute(
        select(AuthRole.name, func.count(distinct(AuthUserRole.user_id)))
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(AuthRole.name.in_(privileged_role_names))
        .group_by(AuthRole.name)
    )).all()
    privileged_role_counts = _str_str_int(priv_rows)
    # Distinct privileged users (any of the listed roles)
    priv_users_count = (await db.execute(
        select(func.count(distinct(AuthUserRole.user_id)))
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(AuthRole.name.in_(privileged_role_names))
    )).scalar() or 0

    # API keys — active count + expiring
    active_keys = (await db.execute(
        select(func.count()).select_from(ApiKeyAssignment).where(
            ApiKeyAssignment.status == "active"
        )
    )).scalar() or 0
    expiring_keys = (await db.execute(
        select(func.count()).select_from(ApiKeyAssignment).where(
            and_(
                ApiKeyAssignment.status == "active",
                ApiKeyAssignment.valid_until.is_not(None),
                ApiKeyAssignment.valid_until >= today,
                ApiKeyAssignment.valid_until <= today + timedelta(days=30),
            )
        )
    )).scalar() or 0
    expired_keys = (await db.execute(
        select(func.count()).select_from(ApiKeyAssignment).where(
            and_(
                ApiKeyAssignment.status == "active",
                ApiKeyAssignment.valid_until.is_not(None),
                ApiKeyAssignment.valid_until < today,
            )
        )
    )).scalar() or 0

    # Post-leaver activity — audit_logs from people whose employment_status='left'
    # AND created_at > their termination_date. Real risk indicator.
    post_leaver_rows = (await db.execute(
        select(
            AuditLog.actor_user_id,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
            IdentityPerson.termination_date,
            func.count(AuditLog.id),
            func.max(AuditLog.created_at),
        )
        .join(IdentityPerson, IdentityPerson.id == AuditLog.actor_user_id)
        .where(
            and_(
                IdentityPerson.employment_status == "left",
                IdentityPerson.termination_date.is_not(None),
                AuditLog.created_at > IdentityPerson.termination_date,
                AuditLog.actor_type == "user",
            )
        )
        .group_by(
            AuditLog.actor_user_id,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
            IdentityPerson.termination_date,
        )
        .order_by(func.count(AuditLog.id).desc())
        .limit(20)
    )).all()
    post_leaver_activity = [
        {
            "person_id": str(pid),
            "display_name": name,
            "work_email": email,
            "termination_date": td.isoformat() if td else None,
            "audit_count": int(c),
            "last_seen": last.isoformat() if last else None,
        }
        for pid, name, email, td, c, last in post_leaver_rows
    ]

    return success_response(
        data={
            "top_actions": top_actions,
            "top_actors": top_actors,
            "by_status": by_status,
            "failure_rate_pct": failure_rate,
            "volume_trend": volume_trend,
            "axis_unit": unit,
            "privileged_role_counts": privileged_role_counts,
            "privileged_users_total": int(priv_users_count),
            "api_keys": {
                "active": int(active_keys),
                "expiring_30d": int(expiring_keys),
                "already_expired": int(expired_keys),
            },
            "post_leaver_activity": post_leaver_activity,
        },
        request=request,
    )


# ─── 8. Cloud / CMDB ────────────────────────────────────────────────


@router.get("/cloud", summary="Cloud inventory & spend analytics")
async def insights_cloud(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径)或 "
            "``invoice``(实付,仅 finance / cloud_admin / global_admin)。"
            "只影响 ``billing`` 子对象。"
        ),
    ),
):
    """Cloud inventory + spend for the 云洞察 tab.

    口径路由（2026-07 保密红线补口）：``_require_insights_section`` 对
    ``cloud`` 放行的角色里有 **management**（领导视图，保密圈外），而
    ``billing`` 此前裸读 ``cmdb_billing_monthly``（纯实付）——同一批人
    在 ``/cmdb/billing/monthly`` 拿到的是同粒度的 **list** 数，两个面
    一相除就是公司折扣率。默认改为 list（按月聚合
    ``cmdb_cloud_billing_daily``，服务榜从 ``cmdb_cost_facts`` 的 list
    列独立聚合），``invoice`` 仅保密圈可请求（无权限 → 403，绝不静默
    降级）。xlsx 导出走同一个 loader，口径写进工作簿（见
    ``_flatten_into_sheets``）——下载下来的表不能对口径无标注。
    """
    _require_insights_section(user, "cloud")
    from app.core.exceptions import ValidationError  # noqa: PLC0415
    from app.domains.cmdb.cost_view import (  # noqa: PLC0415
        COST_MEASURE_INVOICE,
        resolve_measure,
    )

    try:
        resolved_measure = resolve_measure(measure, user.role_names or [])
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError(
            "measure=invoice (实付) 仅限 finance / cloud_admin / "
            "global_admin。"
        )
    wants_invoice = resolved_measure == COST_MEASURE_INVOICE
    start_date, bucket_count, unit = _range_to_window(rng)
    today = date.today()

    # Servers — by status / region (live counts)
    server_status = _str_str_int((await db.execute(
        select(CmdbServer.status, func.count())
        .where(CmdbServer.deleted_at.is_(None))
        .group_by(CmdbServer.status)
    )).all())
    server_region = _str_str_int((await db.execute(
        select(CmdbServer.region, func.count())
        .where(CmdbServer.deleted_at.is_(None))
        .group_by(CmdbServer.region)
        .order_by(func.count().desc())
        .limit(20)
    )).all())
    total_servers = sum(server_status.values())

    # Domains — counts, expiring buckets
    total_domains = (await db.execute(
        select(func.count()).select_from(CmdbDomain).where(CmdbDomain.deleted_at.is_(None))
    )).scalar() or 0
    domain_rows = (await db.execute(
        select(CmdbDomain.expiry_date)
        .where(
            and_(
                CmdbDomain.deleted_at.is_(None),
                CmdbDomain.expiry_date.is_not(None),
            )
        )
    )).scalars().all()
    domain_expiry_buckets = {"expired": 0, "30d": 0, "90d": 0, "1y": 0, ">1y": 0}
    upcoming_domains: list[date] = []
    for ed in domain_rows:
        if ed < today:
            domain_expiry_buckets["expired"] += 1
        elif ed <= today + timedelta(days=30):
            domain_expiry_buckets["30d"] += 1
            upcoming_domains.append(ed)
        elif ed <= today + timedelta(days=90):
            domain_expiry_buckets["90d"] += 1
        elif ed <= today + timedelta(days=365):
            domain_expiry_buckets["1y"] += 1
        else:
            domain_expiry_buckets[">1y"] += 1
    # Pull up to 20 named upcoming domains for the calendar widget
    upcoming_domain_rows = (await db.execute(
        select(CmdbDomain.domain_name, CmdbDomain.expiry_date, CmdbDomain.registrar)
        .where(
            and_(
                CmdbDomain.deleted_at.is_(None),
                CmdbDomain.expiry_date.is_not(None),
                CmdbDomain.expiry_date >= today,
                CmdbDomain.expiry_date <= today + timedelta(days=90),
            )
        )
        .order_by(CmdbDomain.expiry_date)
        .limit(20)
    )).all()
    upcoming_domain_calendar = [
        {
            "domain_name": dn,
            "expiry_date": ed.isoformat(),
            "registrar": reg,
            "days_left": (ed - today).days,
        }
        for dn, ed, reg in upcoming_domain_rows
    ]

    # Cloud monthly billing — last N months
    if unit != "month":
        # Billing is monthly-only data — coerce to 12m view
        bill_start = _month_floor(today, 11)
        bill_axis = _months_axis(bill_start, 12)
    else:
        bill_start = start_date
        bill_axis = _months_axis(start_date, bucket_count)
    # Convert CNY-native rows (total stored as native CNY with
    # currency='CNY') to USD per the integration's FX rate before any
    # aggregation, so this headline + trend + by-service + by-integration
    # all reconcile with the range-correct /billing/overview and the
    # /billing/summary card. Pulling at (month, integration, currency)
    # grain is what makes the per-row conversion possible.
    #
    # 口径:list 聚合 ``cmdb_cloud_billing_daily``(有 list 列),
    # invoice 读 ``cmdb_billing_monthly``(只有实付)。见函数 docstring。
    from app.core.fx_rate import CloudCostUsdConverter  # noqa: PLC0415
    from app.domains.cmdb.cost_view import (  # noqa: PLC0415
        daily_engineering_amount,
        fact_amount_expr,
    )
    from app.domains.cmdb.models import (  # noqa: PLC0415
        CmdbCloudBillingDaily,
        CmdbCostFact,
    )

    cost_conv = CloudCostUsdConverter(db)
    if wants_invoice:
        bill_rows = (await db.execute(
            select(
                func.date_format(CmdbBillingMonthly.month, "%Y-%m"),
                CmdbBillingMonthly.integration_id,
                func.coalesce(func.sum(CmdbBillingMonthly.total_cost_usd), 0),
                CmdbBillingMonthly.currency,
            )
            .where(CmdbBillingMonthly.month >= bill_start)
            .group_by(
                func.date_format(CmdbBillingMonthly.month, "%Y-%m"),
                CmdbBillingMonthly.integration_id,
                CmdbBillingMonthly.currency,
            )
        )).all()
    else:
        bill_rows = (await db.execute(
            select(
                func.date_format(CmdbCloudBillingDaily.date, "%Y-%m"),
                CmdbCloudBillingDaily.integration_id,
                func.coalesce(func.sum(daily_engineering_amount()), 0),
                CmdbCloudBillingDaily.currency,
            )
            .where(CmdbCloudBillingDaily.date >= bill_start)
            .group_by(
                func.date_format(CmdbCloudBillingDaily.date, "%Y-%m"),
                CmdbCloudBillingDaily.integration_id,
                CmdbCloudBillingDaily.currency,
            )
        )).all()
    b_map: dict[str, float] = {}
    int_acc: dict[str, float] = {}
    for ym, int_id, raw, currency in bill_rows:
        # 折不出汇率的组不计入(``to_usd_checked`` 的第二个返回值),
        # 绝不把未折算额当 0 美元折进趋势。
        usd, converted = await cost_conv.to_usd_checked(raw, currency, int_id)
        if not converted:
            continue
        b_map[str(ym)] = b_map.get(str(ym), 0.0) + usd
        int_acc[str(int_id)] = int_acc.get(str(int_id), 0.0) + usd
    billing_trend = [
        {"bucket": b, "cost_usd": round(b_map.get(b, 0.0), 2)}
        for b in bill_axis
    ]
    total_billing_period = round(sum(b_map.values()), 2)

    # Top services across the window — **两个口径都**从
    # ``cmdb_cost_facts`` 按 ``fact_amount_expr`` 聚合。
    #   * list —— ``cmdb_billing_monthly.cost_by_service`` 是实付拆分,
    #     挂在 list 合计旁边就把 (服务 × 月) 粒度的折扣率漏了。
    #   * invoice —— 那个 JSON 列的币种在读侧不可判定:写侧 rebase 只折
    #     ``total_cost_usd`` 并把 ``currency`` 标 USD,拆分留在原生 CNY,
    #     于是 factor=1 会把 ¥ 数字打着 $ 标发出。facts 的 ``amount_usd*``
    #     是入库时折好的真 USD,没有这个歧义。
    # 限制在 ``int_acc``(实际进了合计的租户)上 —— 榜单与合计必须覆盖
    # 同一批云。未接账单明细的云没有 facts 行,榜单比合计窄:这是覆盖
    # 差,不是折扣。
    service_acc: dict[str, float] = {}
    if int_acc:
        service_rows = (await db.execute(
            select(
                CmdbCostFact.service,
                func.coalesce(func.sum(fact_amount_expr(resolved_measure)), 0),
            )
            .where(
                and_(
                    CmdbCostFact.date >= bill_start,
                    CmdbCostFact.integration_id.in_(set(int_acc)),
                )
            )
            .group_by(CmdbCostFact.service)
        )).all()
        for svc, amt in service_rows:
            if not svc:
                continue
            service_acc[str(svc)] = service_acc.get(str(svc), 0.0) + _to_float(amt)
    top_services = sorted(
        service_acc.items(), key=lambda kv: -kv[1]
    )[:15]
    top_services_dict = {k: round(v, 2) for k, v in top_services}

    # Cost by integration (cloud tenant) — already accumulated in USD above.
    cost_by_integration = [
        {"integration_id": iid, "cost_usd": round(v, 2)}
        for iid, v in sorted(int_acc.items(), key=lambda kv: -kv[1])[:20]
    ]
    if cost_conv.unconverted_by_currency:
        # Don't silently fold un-converted native amounts into the USD
        # total as 0 — surface so a missing FX rate is visible.
        logger.warning(
            "insights_cloud: billing rows had no FX rate and were excluded "
            "from the USD total: %s", cost_conv.unconverted_by_currency,
        )

    # Cloud scopes & grants
    scope_status = _str_str_int((await db.execute(
        select(CmdbCloudScope.status, func.count())
        .where(CmdbCloudScope.deleted_at.is_(None))
        .group_by(CmdbCloudScope.status)
    )).all())
    # Provider lives on the parent Integration row, not on the scope itself.
    scope_provider = _str_str_int((await db.execute(
        select(Integration.provider, func.count())
        .select_from(CmdbCloudScope)
        .join(Integration, Integration.id == CmdbCloudScope.integration_id)
        .where(CmdbCloudScope.deleted_at.is_(None))
        .group_by(Integration.provider)
    )).all())
    grant_status = _str_str_int((await db.execute(
        select(CmdbCloudAccessGrant.status, func.count())
        .group_by(CmdbCloudAccessGrant.status)
    )).all())
    expiring_grants = (await db.execute(
        select(func.count()).select_from(CmdbCloudAccessGrant).where(
            and_(
                CmdbCloudAccessGrant.status == "active",
                CmdbCloudAccessGrant.expires_at.is_not(None),
                CmdbCloudAccessGrant.expires_at <= datetime.now(UTC) + timedelta(days=14),
                CmdbCloudAccessGrant.expires_at >= datetime.now(UTC),
            )
        )
    )).scalar() or 0

    return success_response(
        data={
            "servers": {
                "total": int(total_servers),
                "by_status": server_status,
                "by_region": server_region,
            },
            "domains": {
                "total": int(total_domains),
                "expiry_buckets": domain_expiry_buckets,
                "upcoming_calendar": upcoming_domain_calendar,
            },
            "billing": {
                "trend": billing_trend,
                "axis_unit": "month",
                "total_period_usd": total_billing_period,
                "by_service_top": top_services_dict,
                "by_integration_top": cost_by_integration,
                # 口径标识 —— FE 徽标 + xlsx 导出都读它。
                "measure": resolved_measure,
                "aggregate_basis": (
                    "tenant_monthly" if wants_invoice else "billing_daily_sum"
                ),
                "service_basis": (
                    "cost_facts_invoice" if wants_invoice
                    else "cost_facts_list"
                ),
                "unconverted_by_currency": (
                    dict(cost_conv.unconverted_by_currency)
                    if cost_conv.unconverted_by_currency else None
                ),
            },
            "scopes": {
                "by_status": scope_status,
                "by_provider": scope_provider,
            },
            "grants": {
                "by_status": grant_status,
                "expiring_14d": int(expiring_grants),
            },
        },
        request=request,
    )


# ─── 9. Impact — cumulative platform scorecard ─────────────────────


async def _bucket_counts(
    db: AsyncSession,
    col: Any,
    start_dt: datetime,
    unit: str,
    *extra_where: Any,
) -> dict[str, int]:
    """COUNT(*) grouped by day/month bucket of ``col``.

    Always filters ``col IS NOT NULL AND col >= start_dt``; callers pass
    status predicates via ``extra_where``. Returns ``{bucket: count}``
    keyed the same way ``_months_axis`` / ``_days_axis`` label buckets.
    """
    bucket = func.date_format(col, "%Y-%m") if unit == "month" else func.date(col)
    rows = (await db.execute(
        select(bucket, func.count())
        .where(and_(col.is_not(None), col >= start_dt, *extra_where))
        .group_by(bucket)
    )).all()
    return {str(k): int(v) for k, v in rows}


async def _count(db: AsyncSession, model: Any, *where: Any) -> int:
    q = select(func.count()).select_from(model)
    if where:
        q = q.where(and_(*where))
    return int((await db.execute(q)).scalar() or 0)


async def _agg_counts(
    db: AsyncSession,
    model: Any,
    conds: list[Any],
    *base_where: Any,
) -> list[int]:
    """N conditional counts over ONE table scan.

    ``SELECT SUM(cond1), SUM(cond2), …`` instead of N separate
    ``COUNT(*)`` round-trips — the impact scorecard reads several
    append-only tables more than once, and on TiDB each full COUNT is a
    real scan. Pass ``sa_true()`` as a condition to count every row.
    """
    cols = [
        func.coalesce(func.sum(case((cond, 1), else_=0)), 0) for cond in conds
    ]
    q = select(*cols).select_from(model)
    if base_where:
        q = q.where(and_(*base_where))
    row = (await db.execute(q)).one()
    return [int(v or 0) for v in row]


# Per-process TTL cache for the impact payload, keyed by range. The
# aggregation is ~60 COUNTs (some full-table, audit_logs included); the
# projection board polls every 60 s and a meeting room may have several
# screens open — without a cap an admin token can be scripted into a
# DB-load amplifier, and the sensitive-endpoint rate limiter (10/h) is
# the wrong tool for a surface that legitimately polls. Read-only data,
# so short staleness is harmless. TTL comes from
# ``settings.INSIGHTS_IMPACT_CACHE_SECONDS`` (0 disables; tests set 0).
_IMPACT_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}


@router.get("/impact", summary="Cumulative platform scorecard")
async def insights_impact(
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """All-time platform achievements for the leadership view.

    Counting discipline (deliberate, defensible under challenge):

    * **Reclaim counters are timestamp-gated wherever a legacy-import
      path exists** (seats / permissions / cloud grants / API keys /
      SSO / asset returns: ``revoked_at`` / ``confirmed_at`` IS NOT
      NULL) — rows migrated in without an action timestamp are NOT
      claimed as platform work. The remaining reclaim counters
      (external-holder resolution, offboarding blocker resolutions)
      come from platform-native state machines with no import path.
    * **Scale counters** are live inventory: ``deleted_at IS NULL``
      (plus the domain's "active" status where one exists).
    * **Money is an estimate** and is flagged ``is_estimate`` — seat
      savings = revoked allocations × their plan price annualized;
      API-key savings = the monthly budget caps released. The payload
      carries the FX rate used so the number can be re-derived.

    The ``range`` param only drives the trend / cumulative series at the
    bottom; every headline number is all-time.
    """
    _require_admin(user)

    import time  # noqa: PLC0415

    from app.core.config import settings  # noqa: PLC0415

    cache_ttl = settings.INSIGHTS_IMPACT_CACHE_SECONDS
    if cache_ttl > 0:
        hit = _IMPACT_CACHE.get(rng)
        if hit is not None and (time.monotonic() - hit[0]) < cache_ttl:
            return success_response(data=hit[1], request=request)

    payload = await build_impact_payload(db, rng)
    if cache_ttl > 0:
        _IMPACT_CACHE[rng] = (time.monotonic(), payload)
    return success_response(data=payload, request=request)


async def build_impact_payload(
    db: AsyncSession, rng: RangeKey = "12m",
) -> dict[str, Any]:
    """Assemble the all-time impact payload.

    No RBAC, no cache — shared by the HTTP endpoint above (which adds
    both) and the AI-summary generator
    (:mod:`app.dashboard.insights_impact_summary`), so the narrative is
    produced from exactly the numbers the page shows.
    """
    today = date.today()
    start_date, bucket_count, unit = _range_to_window(rng)
    start_dt = _start_of_day(start_date)
    axis = (
        _months_axis(start_date, bucket_count)
        if unit == "month"
        else _days_axis(start_date, bucket_count)
    )

    # ── Scale + reclaim — single-scan conditional aggregates where a
    # table is read more than once (see _agg_counts); one-shot tables
    # keep the plain COUNT.
    people_active, people_total = await _agg_counts(
        db, IdentityPerson,
        [
            IdentityPerson.employment_status.in_(["active", "leaver_pending"]),
            sa_true(),
        ],
        IdentityPerson.deleted_at.is_(None),
    )
    hardware_in_use, hardware_assets = await _agg_counts(
        db, AssetItem,
        [AssetItem.status.in_(list(ASSET_IN_USE_STATUSES)), sa_true()],
        AssetItem.deleted_at.is_(None),
    )
    sw_products_active, sw_products_total = await _agg_counts(
        db, AssetSoftwareCatalog,
        [AssetSoftwareCatalog.status == "active", sa_true()],
    )
    seats_active, seats_revoked = await _agg_counts(
        db, SoftwareSeatAllocation,
        [
            SoftwareSeatAllocation.status == "active",
            and_(
                SoftwareSeatAllocation.status == "revoked",
                SoftwareSeatAllocation.revoked_at.is_not(None),
            ),
        ],
    )
    api_keys_active, api_keys_revoked, api_keys_auto_revoked = await _agg_counts(
        db, ApiKeyAssignment,
        [
            ApiKeyAssignment.status == "active",
            and_(
                ApiKeyAssignment.status == "revoked",
                ApiKeyAssignment.revoked_at.is_not(None),
            ),
            and_(
                ApiKeyAssignment.status == "revoked",
                ApiKeyAssignment.revoked_at.is_not(None),
                ApiKeyAssignment.revoke_actor_role == "system_offboarding",
            ),
        ],
    )
    # Cloud-grant counters mirror the cloud overview's definition
    # (admin_overview_service._permissions_section) exactly: reclaim =
    # every closure-of-access status (GRANT_STATUSES_RECLAIMED) with a
    # stamped ``revoked_at``, and grants on soft-deleted scopes drop
    # out of every count. Any other definition makes this page disagree
    # with 云资产总览 in front of the same audience.
    cloud_scope_alive = CmdbCloudAccessGrant.scope_id.in_(
        select(CmdbCloudScope.id).where(CmdbCloudScope.deleted_at.is_(None))
    )
    cloud_grant_reclaimed = and_(
        CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED),
        CmdbCloudAccessGrant.revoked_at.is_not(None),
    )
    cloud_grants_active, cloud_grants_revoked = await _agg_counts(
        db, CmdbCloudAccessGrant,
        [
            CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE,
            cloud_grant_reclaimed,
        ],
        cloud_scope_alive,
    )

    scale = {
        "people_active": people_active,
        "people_total": people_total,
        "departments": await _count(
            db, IdentityOrgNode,
            IdentityOrgNode.deleted_at.is_(None),
            IdentityOrgNode.node_status == "active",
        ),
        "hardware_assets": hardware_assets,
        "hardware_in_use": hardware_in_use,
        "software_products": sw_products_total,
        "software_products_active": sw_products_active,
        "software_seats_active": seats_active,
        "api_keys_active": api_keys_active,
        "servers": await _count(db, CmdbServer, CmdbServer.deleted_at.is_(None)),
        "domains": await _count(db, CmdbDomain, CmdbDomain.deleted_at.is_(None)),
        "dns_records": await _count(
            db, CmdbDnsRecord, CmdbDnsRecord.deleted_at.is_(None),
        ),
        "load_balancers": await _count(
            db, CmdbLoadBalancer, CmdbLoadBalancer.deleted_at.is_(None),
        ),
        "cdn_domains": await _count(
            db, CmdbCdnDomain, CmdbCdnDomain.deleted_at.is_(None),
        ),
        "cloud_scopes_active": await _count(
            db, CmdbCloudScope,
            CmdbCloudScope.status == SCOPE_STATUS_ACTIVE,
            CmdbCloudScope.deleted_at.is_(None),
        ),
        "iam_users": await _count(
            db, CmdbIamUser, CmdbIamUser.deleted_at.is_(None),
        ),
        "cloud_grants_active": cloud_grants_active,
        "email_groups": await _count(
            db, EmailGroup, EmailGroup.status == "active",
        ),
        "suppliers": await _count(
            db, ProcurementSupplier, ProcurementSupplier.status == "active",
        ),
        "contracts": await _count(db, ProcurementContract),
        "purchase_orders": await _count(db, ProcurementOrder),
        "integrations_enabled": await _count(
            db, Integration, Integration.enabled.is_(True),
        ),
    }

    # ── Reclaim — what the platform took back (all-time) ───────────
    leaver_residual_resolved, external_holders_resolved = await _agg_counts(
        db, SoftwareExternalHolder,
        [
            SoftwareExternalHolder.category == "leaver_residual",
            sa_true(),
        ],
        SoftwareExternalHolder.resolved.is_(True),
    )
    offb_items_resolved, offb_items_auto = await _agg_counts(
        db, OffboardingBlocker,
        [
            OffboardingBlocker.resolution.in_([
                RESO_TRANSFERRED, RESO_REVOKED_AUTO, RESO_REVOKED_MANUAL,
                RESO_SSO_CASCADE, RESO_IDENTITY_SYNC_MANAGED,
            ]),
            OffboardingBlocker.resolution.in_([
                RESO_REVOKED_AUTO, RESO_SSO_CASCADE, RESO_IDENTITY_SYNC_MANAGED,
            ]),
        ],
    )
    reclaim = {
        "seats_revoked": seats_revoked,
        "permissions_revoked": await _count(
            db, UserPermission,
            UserPermission.status == PERM_REVOKED,
            UserPermission.revoked_at.is_not(None),
        ),
        "cloud_grants_revoked": cloud_grants_revoked,
        "api_keys_revoked": api_keys_revoked,
        "api_keys_auto_revoked": api_keys_auto_revoked,
        "sso_grants_revoked": await _count(
            db, SsoGrant,
            SsoGrant.status == SSO_GRANT_STATUS_REVOKED,
            SsoGrant.revoked_at.is_not(None),
        ),
        "leaver_residual_resolved": leaver_residual_resolved,
        "external_holders_resolved": external_holders_resolved,
        "assets_returned": await _count(
            db, AssetOperation,
            AssetOperation.operation_type.in_(["return", "borrow_return"]),
            AssetOperation.status == "confirmed",
            AssetOperation.confirmed_at.is_not(None),
        ),
        "offboarding_items_resolved": offb_items_resolved,
        "offboarding_items_auto_resolved": offb_items_auto,
    }
    # 30-day momentum — same WHERE as the all-time counters above,
    # windowed on the platform-stamped action timestamp. Surfaces as
    # "+N (30d)" chips so the page shows movement, not just stock.
    cutoff_30 = _start_of_day(today - timedelta(days=30))
    reclaim["recent_30d"] = {
        "seats_revoked": await _count(
            db, SoftwareSeatAllocation,
            SoftwareSeatAllocation.status == "revoked",
            SoftwareSeatAllocation.revoked_at.is_not(None),
            SoftwareSeatAllocation.revoked_at >= cutoff_30,
        ),
        "permissions_revoked": await _count(
            db, UserPermission,
            UserPermission.status == PERM_REVOKED,
            UserPermission.revoked_at.is_not(None),
            UserPermission.revoked_at >= cutoff_30,
        ),
        "cloud_grants_revoked": await _count(
            db, CmdbCloudAccessGrant,
            CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED),
            CmdbCloudAccessGrant.revoked_at.is_not(None),
            CmdbCloudAccessGrant.revoked_at >= cutoff_30,
            cloud_scope_alive,
        ),
        "api_keys_revoked": await _count(
            db, ApiKeyAssignment,
            ApiKeyAssignment.status == "revoked",
            ApiKeyAssignment.revoked_at.is_not(None),
            ApiKeyAssignment.revoked_at >= cutoff_30,
        ),
        "sso_grants_revoked": await _count(
            db, SsoGrant,
            SsoGrant.status == SSO_GRANT_STATUS_REVOKED,
            SsoGrant.revoked_at.is_not(None),
            SsoGrant.revoked_at >= cutoff_30,
        ),
    }
    reclaim["cloud_grants_by_reason"] = _str_str_int((await db.execute(
        select(CmdbCloudAccessGrant.revoke_reason, func.count())
        .where(and_(cloud_grant_reclaimed, cloud_scope_alive))
        .group_by(CmdbCloudAccessGrant.revoke_reason)
    )).all())

    # Assets recovered from leavers — same join the lifecycle section
    # uses, without the window (all-time).
    recovered_row = (await db.execute(
        select(func.count(), func.coalesce(func.sum(AssetItem.purchase_cost), 0))
        .select_from(AssetTransferRequest)
        .join(
            AssetItem,
            and_(
                AssetItem.id == AssetTransferRequest.asset_ref_id,
                AssetTransferRequest.asset_type == "hardware",
            ),
        )
        .where(
            and_(
                AssetTransferRequest.status == "completed",
                AssetTransferRequest.completed_at.is_not(None),
                AssetTransferRequest.offboarding_process_id.is_not(None),
            )
        )
    )).one()
    reclaim["assets_recovered_from_leavers"] = int(recovered_row[0] or 0)
    reclaim["assets_recovered_value"] = round(_to_float(recovered_row[1]), 2)

    # ── Flows — how much work the platform carried (all-time) ──────
    completed_tasks, auto_completed_tasks, grant_tasks, revoke_tasks = (
        await _agg_counts(
            db, ProvisioningTask,
            [
                sa_true(),
                ProvisioningTask.execution_mode.in_([EXEC_AUTO_API, EXEC_SSO_AUTO]),
                ProvisioningTask.action == "grant",
                ProvisioningTask.action == "revoke",
            ],
            ProvisioningTask.status == STATUS_COMPLETED,
        )
    )
    onboarding_completed, onboarding_total = await _agg_counts(
        db, OnboardingProcess,
        [OnboardingProcess.status == "completed", sa_true()],
    )
    offboarding_completed, offboarding_total = await _agg_counts(
        db, OffboardingProcess,
        [OffboardingProcess.status == PROC_COMPLETED, sa_true()],
    )
    requests_processed, requests_total = await _agg_counts(
        db, EmployeeRequest,
        [EmployeeRequest.processed_at.is_not(None), sa_true()],
    )
    workflow_succeeded, workflow_total = await _agg_counts(
        db, WorkflowRun,
        [WorkflowRun.status == RUN_STATUS_SUCCEEDED, sa_true()],
    )
    flows = {
        "onboarding_completed": onboarding_completed,
        "onboarding_total": onboarding_total,
        "offboarding_completed": offboarding_completed,
        "offboarding_total": offboarding_total,
        "approvals_processed": await _count(
            db, ApprovalRequest,
            ApprovalRequest.status.in_(["approved", "rejected"]),
        ),
        "employee_requests_processed": requests_processed,
        "employee_requests_total": requests_total,
        "dns_requests_executed": await _count(
            db, CmdbDnsRecordRequest, CmdbDnsRecordRequest.status == "executed",
        ),
        "cloud_scope_requests_provisioned": await _count(
            db, CmdbCloudScopeRequest,
            CmdbCloudScopeRequest.status == "provisioned",
        ),
        "provisioning_completed": completed_tasks,
        "provisioning_auto_completed": auto_completed_tasks,
        "provisioning_grants_completed": grant_tasks,
        "provisioning_revokes_completed": revoke_tasks,
        "hris_sync_runs": await _count(db, IdentitySyncRun),
        "workflow_runs_total": workflow_total,
        "workflow_runs_succeeded": workflow_succeeded,
    }

    # ── Security & governance (all-time) ───────────────────────────
    # Both legs restricted to observations a provider actually answered:
    # an unreadable org 2FA policy is not an un-enrolled identity, and
    # this widget is a headline number (see AuditorMfaStatus column docs).
    mfa_enrolled, mfa_total = await _agg_counts(
        db, AuditorMfaStatus,
        [
            and_(
                AuditorMfaStatus.enrollment_observed.is_(True),
                AuditorMfaStatus.is_enrolled.is_(True),
            ),
            AuditorMfaStatus.enrollment_observed.is_(True),
        ],
    )
    cert_items_reviewed, cert_revocations = await _agg_counts(
        db, AccessCertificationItem,
        [
            AccessCertificationItem.decision.in_(["keep", "revoke"]),
            AccessCertificationItem.revocation_status == "completed",
        ],
    )
    findings_resolved, findings_open = await _agg_counts(
        db, AuditorPublicExposureFinding,
        [
            AuditorPublicExposureFinding.status == "resolved",
            AuditorPublicExposureFinding.status == "open",
        ],
    )
    signin_monitored, signin_anomalies = await _agg_counts(
        db, AuditorSigninEvent,
        [sa_true(), AuditorSigninEvent.anomaly_level != "normal"],
    )
    security = {
        "leaver_blocks_intercepted": await _count(db, LeaverBlockAudit),
        "dual_sign_overrides_executed": await _count(
            db, LeaverDualSignOverride, LeaverDualSignOverride.status == "executed",
        ),
        "access_certs_completed": await _count(
            db, AccessCertification, AccessCertification.status == "completed",
        ),
        "access_cert_items_reviewed": cert_items_reviewed,
        "access_cert_revocations_completed": cert_revocations,
        "exposure_findings_resolved": findings_resolved,
        "exposure_findings_open": findings_open,
        "mfa_total": mfa_total,
        "mfa_enrolled": mfa_enrolled,
        "mfa_coverage_pct": (
            round(mfa_enrolled * 100.0 / mfa_total, 1) if mfa_total else 0.0
        ),
        "signin_events_monitored": signin_monitored,
        "signin_anomalies_flagged": signin_anomalies,
        "google_orphans_tracked": await _count(db, GoogleOrphanAccount),
        "audit_logs_total": await _count(db, AuditLog),
    }

    # ── Automation (all-time) ──────────────────────────────────────
    cron_ok, cron_failed, cron_total = await _agg_counts(
        db, ScheduledJobRun,
        [
            ScheduledJobRun.status == "ok",
            ScheduledJobRun.status == "failed",
            sa_true(),
        ],
    )
    automation = {
        "cron_jobs": await _count(db, ScheduledJob),
        "cron_runs_total": cron_total,
        "cron_runs_ok": cron_ok,
        "cron_success_rate_pct": (
            round(cron_ok * 100.0 / (cron_ok + cron_failed), 1)
            if (cron_ok + cron_failed)
            else 0.0
        ),
        "notifications_delivered": await _count(
            db, NotificationOutbox, NotificationOutbox.status == "delivered",
        ),
        "provisioning_automation_rate_pct": (
            round(auto_completed_tasks * 100.0 / completed_tasks, 1)
            if completed_tasks
            else 0.0
        ),
    }
    # MCP tool count — imported lazily: app.mcp.tools registers every
    # tool at import time, which the API process has already paid via
    # main.py, but unit contexts that import only this router shouldn't.
    from app.mcp.tools import _REGISTERED_TOOLS_COUNT_HINT  # noqa: PLC0415
    automation["mcp_tools"] = int(_REGISTERED_TOOLS_COUNT_HINT)

    # ── Savings — estimate, flagged as such ─────────────────────────
    seat_saving_rows = (await db.execute(
        select(
            SoftwareLicensePlan.currency,
            SoftwareLicensePlan.billing_cycle,
            SoftwareLicensePlan.price,
            func.count(SoftwareSeatAllocation.id),
        )
        .join(
            SoftwareLicensePlan,
            SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id,
        )
        .where(
            and_(
                SoftwareSeatAllocation.status == "revoked",
                SoftwareSeatAllocation.revoked_at.is_not(None),
            )
        )
        .group_by(
            SoftwareLicensePlan.id,
            SoftwareLicensePlan.currency,
            SoftwareLicensePlan.billing_cycle,
            SoftwareLicensePlan.price,
        )
    )).all()
    seats_by_currency: dict[str, float] = {}
    for currency, cycle, price, cnt in seat_saving_rows:
        if cycle == "monthly":
            annual = _to_float(price) * 12.0 * int(cnt)
        elif cycle == "yearly":
            annual = _to_float(price) * int(cnt)
        else:  # one_time — releasing the seat saves no recurring spend
            continue
        cur = (currency or "CNY").upper()
        seats_by_currency[cur] = round(seats_by_currency.get(cur, 0.0) + annual, 2)

    api_key_monthly_usd = _to_float((await db.execute(
        select(func.coalesce(func.sum(ApiKeyAssignment.monthly_budget_usd), 0))
        .where(
            and_(
                ApiKeyAssignment.status == "revoked",
                ApiKeyAssignment.revoked_at.is_not(None),
            )
        )
    )).scalar())

    usd_per_cny = await get_usd_per_cny(db)
    total_annualized_cny: float | None = None
    # A grand total is only honest when every currency in the book is
    # convertible. Plans allow CNY/USD/HKD — the platform FX setting
    # only covers CNY↔USD, so an HKD bucket forces the total to None
    # (the per-currency breakdown in the payload still shows it) rather
    # than silently under-reporting.
    convertible = set(seats_by_currency) <= {"CNY", "USD"}
    if convertible and usd_per_cny and usd_per_cny > 0:
        cny_per_usd = 1.0 / usd_per_cny
        total = seats_by_currency.get("CNY", 0.0)
        total += seats_by_currency.get("USD", 0.0) * cny_per_usd
        total += api_key_monthly_usd * 12.0 * cny_per_usd
        total_annualized_cny = round(total, 2)
    elif not any(c != "CNY" for c in seats_by_currency) and api_key_monthly_usd == 0:
        # Pure-CNY book — no FX needed for an exact total.
        total_annualized_cny = round(seats_by_currency.get("CNY", 0.0), 2)

    savings = {
        "is_estimate": True,
        "fx_rate_available": bool(usd_per_cny),
        "usd_per_cny": usd_per_cny,
        "seats_annualized_by_currency": seats_by_currency,
        "api_keys_monthly_budget_usd": round(api_key_monthly_usd, 2),
        "total_annualized_cny": total_annualized_cny,
    }

    # ── Trend + cumulative series (window) ──────────────────────────
    seats_b = await _bucket_counts(
        db, SoftwareSeatAllocation.revoked_at, start_dt, unit,
        SoftwareSeatAllocation.status == "revoked",
    )
    perms_b = await _bucket_counts(
        db, UserPermission.revoked_at, start_dt, unit,
        UserPermission.status == PERM_REVOKED,
    )
    cloud_b = await _bucket_counts(
        db, CmdbCloudAccessGrant.revoked_at, start_dt, unit,
        CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED),
        cloud_scope_alive,
    )
    keys_b = await _bucket_counts(
        db, ApiKeyAssignment.revoked_at, start_dt, unit,
        ApiKeyAssignment.status == "revoked",
    )
    onb_b = await _bucket_counts(
        db, OnboardingProcess.completed_at, start_dt, unit,
        OnboardingProcess.status == "completed",
    )
    offb_b = await _bucket_counts(
        db, OffboardingProcess.completed_at, start_dt, unit,
        OffboardingProcess.status == PROC_COMPLETED,
    )
    reqs_b = await _bucket_counts(
        db, EmployeeRequest.processed_at, start_dt, unit,
    )
    tasks_b = await _bucket_counts(
        db, ProvisioningTask.completed_at, start_dt, unit,
        ProvisioningTask.status == STATUS_COMPLETED,
    )
    notif_b = await _bucket_counts(
        db, NotificationOutbox.delivered_at, start_dt, unit,
        NotificationOutbox.status == "delivered",
    )
    trend = [
        {
            "bucket": b,
            "seats_revoked": seats_b.get(b, 0),
            "permissions_revoked": perms_b.get(b, 0),
            "cloud_grants_revoked": cloud_b.get(b, 0),
            "api_keys_revoked": keys_b.get(b, 0),
            "onboarding_completed": onb_b.get(b, 0),
            "offboarding_completed": offb_b.get(b, 0),
            "requests_processed": reqs_b.get(b, 0),
            "tasks_completed": tasks_b.get(b, 0),
            "notifications_delivered": notif_b.get(b, 0),
        }
        for b in axis
    ]

    # Cumulative reclaim curves — base = everything before the window,
    # then a running sum across buckets. This is the up-and-to-the-right
    # line leadership reads as "the platform keeps compounding".
    async def _base(model: Any, ts_col: Any, *where: Any) -> int:
        return await _count(
            db, model, ts_col.is_not(None), ts_col < start_dt, *where,
        )

    bases = {
        "seats_revoked": await _base(
            SoftwareSeatAllocation, SoftwareSeatAllocation.revoked_at,
            SoftwareSeatAllocation.status == "revoked",
        ),
        "permissions_revoked": await _base(
            UserPermission, UserPermission.revoked_at,
            UserPermission.status == PERM_REVOKED,
        ),
        "cloud_grants_revoked": await _base(
            CmdbCloudAccessGrant, CmdbCloudAccessGrant.revoked_at,
            CmdbCloudAccessGrant.status.in_(GRANT_STATUSES_RECLAIMED),
            cloud_scope_alive,
        ),
        "api_keys_revoked": await _base(
            ApiKeyAssignment, ApiKeyAssignment.revoked_at,
            ApiKeyAssignment.status == "revoked",
        ),
    }
    running = dict(bases)
    cumulative = []
    for row in trend:
        for key in bases:
            running[key] += row[key]
        cumulative.append({
            "bucket": row["bucket"],
            **{key: running[key] for key in bases},
            "total_reclaimed": sum(running[key] for key in bases),
        })

    return {
        "as_of": today.isoformat(),
        "axis_unit": unit,
        "scale": scale,
        "reclaim": reclaim,
        "flows": flows,
        "security": security,
        "automation": automation,
        "savings": savings,
        "trend": trend,
        "cumulative": cumulative,
    }


# ─── Impact AI summary ──────────────────────────────────────────────


@router.get("/impact/summary", summary="Cached AI executive summary")
async def impact_summary(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Newest cached bilingual narrative for the impact page.

    Same read gate as the page itself (admin + management). Never
    triggers generation — an empty result renders as "not generated
    yet" and only a global_admin can (re)generate.
    """
    _require_admin(user)
    from app.dashboard.insights_impact_summary import (  # noqa: PLC0415
        get_impact_summary,
    )
    data = await get_impact_summary(db)
    return success_response(data=data, request=request)


@router.post("/impact/summary/regenerate", summary="Regenerate AI summary")
async def impact_summary_regenerate(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Manual regenerate — global_admin only (NOT management: the
    leadership role reads dashboards, it doesn't spend LLM quota), and
    per-user-per-day rate-limited inside the generator."""
    role_names = user.role_names or []
    if not (
        has_role(role_names, GLOBAL_ADMIN_ROLE) or has_role(role_names, "it_admin")
    ):
        raise ForbiddenError("Admin required")
    from app.dashboard.insights_impact_summary import (  # noqa: PLC0415
        generate_impact_summary,
    )
    data = await generate_impact_summary(
        db,
        actor_id=user.id,
        trace_id=request.state.trace_id,
        enforce_quota=True,
    )
    await db.commit()
    # The summary panel sits on a cached page — drop the impact cache
    # so the very next poll (e.g. the projection board's 60 s tick)
    # can't show a payload older than the narrative beside it.
    _IMPACT_CACHE.clear()
    return success_response(data=data, request=request)


# ─── HR insights AI diagnosis (人员洞察 / 入离职洞察) ────────────────


@router.get(
    "/{section}/ai-summary",
    summary="Cached AI diagnosis for a HR insights tab",
)
async def hr_ai_summary(
    section: Literal["people", "lifecycle"],
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Newest cached bilingual diagnosis for ``(section, range)``.

    Same read gate as the tab itself (``_require_insights_section``: admin
    / management + hr for people & lifecycle). Never triggers generation —
    an empty result renders as "not generated yet" and only a
    global_admin / hr can generate.
    """
    _require_insights_section(user, section)
    from app.dashboard.insights_hr_ai import get_hr_ai_summary  # noqa: PLC0415

    data = await get_hr_ai_summary(db, section, rng)
    return success_response(data=data, request=request)


@router.post(
    "/{section}/ai-summary/regenerate",
    summary="Regenerate the AI diagnosis for a HR insights tab",
)
async def hr_ai_summary_regenerate(
    section: Literal["people", "lifecycle"],
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Manual regenerate — global_admin OR hr (this is HR's own page, so
    unlike the leadership impact summary, hr may spend LLM quota here).
    auditor / management can read but not generate. Per-user-per-day
    rate-limited inside the generator."""
    from app.core.rbac import HR_ROLE  # noqa: PLC0415
    role_names = user.role_names or []
    if not (
        has_role(role_names, GLOBAL_ADMIN_ROLE)
        or has_role(role_names, "it_admin")
        or has_role(role_names, HR_ROLE)
    ):
        raise ForbiddenError("Global admin or HR required")
    from app.dashboard.insights_hr_ai import (  # noqa: PLC0415
        generate_hr_ai_summary,
    )

    data = await generate_hr_ai_summary(
        db,
        section=section,
        range_key=rng,
        actor_id=user.id,
        trace_id=request.state.trace_id,
        enforce_quota=True,
    )
    await db.commit()
    return success_response(data=data, request=request)


# ─── Excel export ───────────────────────────────────────────────────


@router.get("/export/{section}", summary="Export an insights tab as Excel (.xlsx)")
async def insights_export(
    section: Literal[
        "overview", "people", "assets", "software", "software-catalog",
        "operations", "lifecycle", "security", "cloud", "impact",
        "hardware-finance",
    ],
    request: Request,
    rng: RangeKey = Query("12m", alias="range"),
    user: AuthUser = Depends(get_current_user),
    _rate: None = Depends(check_sensitive_endpoint_rate_limit),
    db: AsyncSession = Depends(get_db),
    measure: str | None = Query(
        default=None,
        description=(
            "``cloud`` section only — 成本口径,与 ``GET /cloud`` 同语义"
            "(默认 ``list``;``invoice`` 仅保密圈)。其它 section 忽略。"
        ),
    ),
):
    """Return the same payload the matching ``GET /<section>`` returns,
    but rendered into a multi-sheet ``.xlsx`` workbook so admins can
    archive / share. We call the same handlers internally so the
    exported numbers match the on-screen view exactly.

    ``cloud`` 的口径闸由 :func:`insights_cloud` 自己执行（同一个
    loader），并把口径写进工作簿 —— 一份下载下来的表必须自己说清
    它是业务折前还是实付。
    """
    _require_insights_section(user, section)
    from io import BytesIO

    from fastapi.responses import StreamingResponse
    from openpyxl import Workbook
    from openpyxl.styles import Alignment, Font, PatternFill

    # Dispatch to the right loader. We re-use the same handlers so the
    # numbers are identical to the on-screen view.
    if section == "overview":
        resp = await insights_overview(request, user, db)
    elif section == "people":
        resp = await insights_people(request, rng, user, db)
    elif section == "assets":
        resp = await insights_assets(request, rng, user, db)
    elif section == "software":
        resp = await insights_software(request, rng, user, db)
    elif section == "software-catalog":
        resp = await insights_software_catalog(request, rng, user, db)
    elif section == "operations":
        resp = await insights_operations(request, rng, user, db)
    elif section == "lifecycle":
        resp = await insights_lifecycle(request, rng, user, db)
    elif section == "security":
        resp = await insights_security(request, rng, user, db)
    elif section == "impact":
        resp = await insights_impact(request, rng, user, db)
    elif section == "hardware-finance":
        resp = await insights_hardware_finance(request, rng, user, db)
    else:  # cloud
        resp = await insights_cloud(request, rng, user, db, measure)

    # ``success_response`` returns the standard envelope as a plain
    # dict — it is not yet a Response, so we just unwrap ``data``.
    data = (resp or {}).get("data") or {}

    wb = Workbook()
    # Remove default sheet
    wb.remove(wb.active)

    header_font = Font(bold=True, color="FFFFFF")
    header_fill = PatternFill("solid", fgColor="4F46E5")
    centered = Alignment(horizontal="center")

    def _add_sheet(title: str, rows: list[list[Any]]) -> None:
        # Excel sheet titles cap at 31 chars + can't contain certain chars
        clean = title[:31].replace(":", "-").replace("/", "-").replace("\\", "-")
        ws = wb.create_sheet(clean)
        for row_idx, row in enumerate(rows, start=1):
            for col_idx, val in enumerate(row, start=1):
                cell = ws.cell(row=row_idx, column=col_idx, value=val)
                if row_idx == 1:
                    cell.font = header_font
                    cell.fill = header_fill
                    cell.alignment = centered
        # Auto-size columns (rough)
        for col_idx in range(1, len(rows[0]) + 1 if rows else 1):
            ws.column_dimensions[ws.cell(row=1, column=col_idx).column_letter].width = 22

    # Each section has its own sheet layout — kept human-readable.
    _flatten_into_sheets(_add_sheet, section, data)

    out = BytesIO()
    wb.save(out)
    out.seek(0)

    filename = f"helix-insights-{section}-{rng}-{date.today().isoformat()}.xlsx"
    return StreamingResponse(
        out,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


def _flatten_into_sheets(
    add_sheet: Any,
    section: str,
    data: dict[str, Any],
) -> None:
    """Section-specific Excel layout — keeps the export readable.

    Each section gets a Summary sheet with KPIs + one sheet per
    chart/table that matters. Only fields with rows produce a sheet —
    empty data does NOT produce an empty sheet.
    """
    if section == "overview":
        kpis = data.get("kpis") or {}
        add_sheet("Summary", [["KPI", "Value"], *[[k, v] for k, v in kpis.items()]])
        trend = data.get("daily_trend") or []
        if trend:
            add_sheet("Daily Trend", [["Day", "Requests", "Tasks"],
                                      *[[r["day"], r["requests"], r["tasks"]] for r in trend]])
        alerts = data.get("alerts") or []
        if alerts:
            add_sheet("Alerts", [["Level", "Key", "Count", "Link"],
                                 *[[a["level"], a["key"], a["count"], a["link"]] for a in alerts]])
        return

    if section == "people":
        add_sheet("Summary", [
            ["Headcount", data.get("headcount", 0)],
            ["Avg manager span p50", (data.get("manager_span") or {}).get("p50", 0)],
            ["Avg manager span p90", (data.get("manager_span") or {}).get("p90", 0)],
            ["Manager span max", (data.get("manager_span") or {}).get("max", 0)],
        ])
        if data.get("by_status"):
            add_sheet("By Status", [["Status", "Count"], *[[k, v] for k, v in data["by_status"].items()]])
        if data.get("by_location"):
            add_sheet("By Location", [["Location", "Count"], *[[k, v] for k, v in data["by_location"].items()]])
        if data.get("by_country"):
            add_sheet("By Country", [["Country", "Count"], *[[k, v] for k, v in data["by_country"].items()]])
        if data.get("by_job_title"):
            add_sheet("Top Job Titles", [["Title", "Count"], *[[k, v] for k, v in data["by_job_title"].items()]])
        if data.get("by_department_tree"):
            add_sheet("By Department", [
                ["Department", "Direct", "Total (incl. subtree)", "Parent"],
                *[[r["department"], r.get("direct_count", 0), r.get("total_count", 0), r.get("parent_department")]
                  for r in data["by_department_tree"]],
            ])
        if data.get("trend"):
            add_sheet("Trend", [["Bucket", "Hires", "Leavers", "Net"],
                                *[[r["bucket"], r["hires"], r["leavers"], r["net"]] for r in data["trend"]]])
        if data.get("tenure_distribution"):
            add_sheet("Tenure", [["Bucket", "Count"], *[[k, v] for k, v in data["tenure_distribution"].items()]])
        return

    if section == "assets":
        add_sheet("Summary", [
            ["Total", data.get("total", 0)],
            ["Total value", data.get("total_value", 0)],
            ["Assigned value", data.get("assigned_value", 0)],
            ["Idle value", data.get("idle_value", 0)],
            ["Avg assigned age (days)", data.get("avg_assigned_age_days", 0)],
        ])
        if data.get("by_status"):
            add_sheet("By Status", [["Status", "Count"], *[[k, v] for k, v in data["by_status"].items()]])
        if data.get("by_category"):
            add_sheet("By Category", [["Category", "Count"], *[[k, v] for k, v in data["by_category"].items()]])
        if data.get("by_brand"):
            add_sheet("By Brand", [["Brand", "Count"], *[[k, v] for k, v in data["by_brand"].items()]])
        if data.get("trend"):
            add_sheet("Procurement Trend", [["Bucket", "Count", "Spend"],
                                            *[[r["bucket"], r["count"], r["spend"]] for r in data["trend"]]])
        if data.get("age_buckets"):
            avb = data.get("age_value_buckets") or {}
            add_sheet("Aging", [
                ["Bucket", "Count", "Value"],
                *[[k, data["age_buckets"][k], avb.get(k, 0)] for k in data["age_buckets"]],
            ])
        if data.get("warranty_buckets"):
            add_sheet("Warranty", [["Bucket", "Count"], *[[k, v] for k, v in data["warranty_buckets"].items()]])
        if data.get("by_department_tree"):
            add_sheet("By Department", [
                ["Department", "Direct", "Total", "Parent"],
                *[[r["department"], r.get("direct_count", 0), r.get("total_count", 0), r.get("parent_department")]
                  for r in data["by_department_tree"]],
            ])
        return

    if section == "hardware-finance":
        summary = data.get("summary") or {}
        trend = data.get("trend") or {}
        pnl = summary.get("disposal_pnl") or {}
        add_sheet("Summary", [
            ["Metric", "Value"],
            ["As of", summary.get("as_of")],
            ["Asset count", summary.get("asset_count", 0)],
            ["Original cost", summary.get("total_original_cost")],
            ["Accumulated depreciation", summary.get("total_accumulated_depreciation")],
            ["Net book value", summary.get("total_net_book_value")],
            ["Repair cost (TCO)", summary.get("total_repair_cost")],
            ["Disposed count", pnl.get("disposed_count", 0)],
            ["Disposal proceeds", pnl.get("total_proceeds")],
            ["Disposal book value", pnl.get("total_book_value_at_disposal")],
            ["Disposal gain/loss", pnl.get("total_gain_loss")],
        ])
        for bucket_key, title in (
            ("by_category", "By Category"),
            ("by_department", "By Department"),
            ("by_office", "By Office"),
        ):
            rows = summary.get(bucket_key) or []
            if rows:
                add_sheet(title, [
                    ["Key", "Label", "Assets", "Original Cost", "Accum. Dep.", "Net Book Value", "Repair Cost"],
                    *[[r.get("key"), r.get("label"), r.get("asset_count", 0),
                       r.get("original_cost"), r.get("accumulated_depreciation"),
                       r.get("net_book_value"), r.get("repair_total_cost")]
                      for r in rows],
                ])
        points = trend.get("points") or []
        if points:
            add_sheet("Book Value Trend (CNY)", [
                ["Period", "Original Cost", "Accum. Dep.", "Net Book Value", "Assets"],
                *[[p.get("period"), p.get("original_cost_cny"),
                   p.get("accumulated_depreciation_cny"), p.get("net_book_value_cny"),
                   p.get("asset_count", 0)] for p in points],
            ])
        return

    if section == "software":
        add_sheet("Summary", [
            ["Total active seats", data.get("total_active_seats", 0)],
            ["Monthly cost", data.get("total_monthly_cost", 0)],
        ])
        sw = data.get("by_software") or []
        if sw:
            add_sheet("By Software", [
                ["Software", "Active seats", "Monthly cost", "Status"],
                *[[s["software_name"], s["active_seats"], s["monthly_cost"], s.get("software_status")] for s in sw],
            ])
        if data.get("trend"):
            add_sheet("Grant/Revoke Trend", [["Bucket", "Grants", "Revokes", "Net"],
                                             *[[r["bucket"], r["grants"], r["revokes"], r["net"]] for r in data["trend"]]])
        if data.get("allocation_age_buckets"):
            add_sheet("Allocation Age", [["Bucket", "Count"],
                                         *[[k, v] for k, v in data["allocation_age_buckets"].items()]])
        if data.get("plan_utilization"):
            add_sheet("Plan Utilization", [
                ["Software", "Plan", "Active", "Max", "Utilization %"],
                *[[r["software_name"], r["plan_name"], r["active_seats"], r["max_seats"], r["utilization_pct"]]
                  for r in data["plan_utilization"]],
            ])
        if data.get("renewal_calendar"):
            add_sheet("Renewals", [
                ["Software", "Expiry date", "Days left"],
                *[[r["software_name"], r["expiry_date"], r["days_left"]] for r in data["renewal_calendar"]],
            ])
        if data.get("cost_by_department_tree"):
            add_sheet("Cost by Department", [
                ["Department", "Direct cost", "Total cost", "Parent"],
                *[[r["department"], r.get("direct_monthly_cost", 0), r.get("monthly_cost", 0),
                   r.get("parent_department")]
                  for r in data["cost_by_department_tree"]],
            ])
        return

    if section == "software-catalog":
        kpis = data.get("kpis") or {}
        add_sheet("Summary", [
            ["KPI", "Value"],
            ["Total", kpis.get("total", 0)],
            ["Active", kpis.get("active", 0)],
            ["Inactive", kpis.get("inactive", 0)],
            ["Has owner", kpis.get("has_owner", 0)],
            ["Owner coverage %", kpis.get("has_owner_pct", 0)],
            ["Has IT contact", kpis.get("has_it_contact", 0)],
            ["IT contact coverage %", kpis.get("has_it_contact_pct", 0)],
            ["SSO cascade", kpis.get("sso_cascade", 0)],
            ["SSO cascade %", kpis.get("sso_cascade_pct", 0)],
            ["With connector", kpis.get("with_connector", 0)],
            ["With connector %", kpis.get("with_connector_pct", 0)],
            ["Recently added (range)", kpis.get("recently_added", 0)],
        ])
        if data.get("by_category"):
            add_sheet("By Category",
                      [["Category", "Count"],
                       *[[k, v] for k, v in data["by_category"].items()]])
        if data.get("by_offboarding_policy"):
            add_sheet("By Offboarding Policy",
                      [["Policy", "Count"],
                       *[[k, v] for k, v in data["by_offboarding_policy"].items()]])
        rows = data.get("rows") or []
        if rows:
            add_sheet("Software", [
                ["Name", "Category", "Status", "Asset type",
                 "Owner", "Owner email", "IT contact", "IT contact email",
                 "Department", "Offboarding policy", "Has connector",
                 "Created", "Updated"],
                *[
                    [
                        r["name"], r["category"], r["status"], r.get("asset_type") or "",
                        r.get("owner_name") or "", r.get("owner_email") or "",
                        r.get("it_contact_name") or "", r.get("it_contact_email") or "",
                        r.get("owner_department") or "",
                        r.get("offboarding_policy"), "yes" if r.get("has_connector") else "no",
                        r.get("created_at") or "", r.get("updated_at") or "",
                    ]
                    for r in rows
                ],
            ])
        return

    if section == "operations":
        add_sheet("Summary", [
            ["Automation rate %", data.get("automation_rate_pct", 0)],
            ["Failure rate %", data.get("failure_rate_pct", 0)],
            ["Avg completion (h, overall)", (data.get("completion_avg_hours") or {}).get("overall", 0)],
            ["Avg completion (h, auto)", (data.get("completion_avg_hours") or {}).get("auto", 0)],
            ["Avg completion (h, manual)", (data.get("completion_avg_hours") or {}).get("manual", 0)],
            ["Stale in_progress total", data.get("stale_total", 0)],
            ["Open transfers", data.get("open_transfers", 0)],
        ])
        for sheet_name, key in [
            ("Status (window)", "by_status_window"),
            ("Status (live)", "by_status"),
            ("By Source", "by_source"),
            ("By Exec Mode", "by_execution_mode"),
            ("Stale by Target", "stale_by_target"),
            ("Top Failing Targets", "top_failing_targets"),
            ("Transfer Age", "transfer_age_buckets"),
        ]:
            d = data.get(key)
            if d:
                add_sheet(sheet_name, [["Key", "Count"], *[[k, v] for k, v in d.items()]])
        if data.get("trend"):
            add_sheet("Throughput Trend", [
                ["Bucket", "Total", "By source"],
                *[[r["bucket"], r["total"], ", ".join(f"{k}={v}" for k, v in r["by_source"].items())]
                  for r in data["trend"]],
            ])
        return

    if section == "lifecycle":
        add_sheet("Summary", [
            ["Avg onboarding days", data.get("avg_onboarding_days", 0)],
            ["Avg offboarding days", data.get("avg_offboarding_days", 0)],
            ["Blocked past due", data.get("blocked_past_due", 0)],
            ["Recovered asset value", data.get("recovered_asset_value", 0)],
            ["Recovered asset count", data.get("recovered_asset_count", 0)],
        ])
        if data.get("onboarding_buckets"):
            add_sheet("Onboarding Buckets", [["Bucket", "Count"],
                                             *[[k, v] for k, v in data["onboarding_buckets"].items()]])
        if data.get("blockers_by_resource"):
            add_sheet("Blockers by Resource", [["Resource", "Count"],
                                               *[[k, v] for k, v in data["blockers_by_resource"].items()]])
        if data.get("blocker_age_buckets"):
            add_sheet("Blocker Age", [["Bucket", "Count"],
                                      *[[k, v] for k, v in data["blocker_age_buckets"].items()]])
        if data.get("offboarding_status_counts"):
            add_sheet("Offboarding Status", [["Status", "Count"],
                                             *[[k, v] for k, v in data["offboarding_status_counts"].items()]])
        if data.get("upcoming_leavers"):
            add_sheet("Upcoming Leavers", [
                ["Name", "Email", "Leaver date", "Days left", "Status", "Pending blockers"],
                *[[r["display_name"], r["work_email"], r["leaver_date"], r["days_left"], r["status"],
                   r["pending_blockers"]] for r in data["upcoming_leavers"]],
            ])
        if data.get("onboarding_trend"):
            add_sheet("Onboarding Trend", [["Bucket", "Hires"],
                                           *[[r["bucket"], r["hires"]] for r in data["onboarding_trend"]]])
        return

    if section == "security":
        add_sheet("Summary", [
            ["Privileged users total", data.get("privileged_users_total", 0)],
            ["Failure rate %", data.get("failure_rate_pct", 0)],
            ["API keys active", (data.get("api_keys") or {}).get("active", 0)],
            ["API keys expiring 30d", (data.get("api_keys") or {}).get("expiring_30d", 0)],
            ["API keys expired", (data.get("api_keys") or {}).get("already_expired", 0)],
        ])
        if data.get("top_actions"):
            add_sheet("Top Actions", [["Action", "Count"], *[[k, v] for k, v in data["top_actions"].items()]])
        if data.get("top_actors"):
            add_sheet("Top Actors", [
                ["Actor", "Actor ID", "Count"],
                *[[r["label"], r["actor_id"], r["count"]] for r in data["top_actors"]],
            ])
        if data.get("by_status"):
            add_sheet("By Status", [["Status", "Count"], *[[k, v] for k, v in data["by_status"].items()]])
        if data.get("privileged_role_counts"):
            add_sheet("Privileged Roles", [["Role", "Users"],
                                           *[[k, v] for k, v in data["privileged_role_counts"].items()]])
        if data.get("post_leaver_activity"):
            add_sheet("Post-Leaver Activity", [
                ["Name", "Email", "Termination", "Audit count", "Last seen"],
                *[[r["display_name"], r["work_email"], r["termination_date"],
                   r["audit_count"], r["last_seen"]] for r in data["post_leaver_activity"]],
            ])
        if data.get("volume_trend"):
            add_sheet("Volume Trend", [["Bucket", "Count"],
                                       *[[r["bucket"], r["count"]] for r in data["volume_trend"]]])
        return

    if section == "impact":
        for sheet_name, key in [
            ("Scale", "scale"),
            ("Reclaim", "reclaim"),
            ("Flows", "flows"),
            ("Security", "security"),
            ("Automation", "automation"),
        ]:
            d = data.get(key) or {}
            rows = [
                [k, v] for k, v in d.items()
                if not isinstance(v, dict)
            ]
            if rows:
                add_sheet(sheet_name, [["Metric", "Value"], *rows])
        by_reason = (data.get("reclaim") or {}).get("cloud_grants_by_reason") or {}
        if by_reason:
            add_sheet("Cloud Revoke Reasons", [
                ["Reason", "Count"], *[[k, v] for k, v in by_reason.items()],
            ])
        sav = data.get("savings") or {}
        if sav:
            add_sheet("Savings (estimate)", [
                ["Metric", "Value"],
                ["Is estimate", "yes" if sav.get("is_estimate") else "no"],
                ["FX rate available", "yes" if sav.get("fx_rate_available") else "no"],
                ["USD per CNY", sav.get("usd_per_cny") or ""],
                *[
                    [f"Seats annualized ({cur})", amt]
                    for cur, amt in (sav.get("seats_annualized_by_currency") or {}).items()
                ],
                ["API-key monthly budget released (USD)", sav.get("api_keys_monthly_budget_usd", 0)],
                # A legitimate 0.0 (fresh deployment, nothing reclaimed
                # yet) must stay distinguishable from None (no honest
                # total exists, e.g. an HKD bucket without FX coverage).
                [
                    "Total annualized (CNY)",
                    sav.get("total_annualized_cny")
                    if sav.get("total_annualized_cny") is not None
                    else "",
                ],
            ])
        if data.get("trend"):
            add_sheet("Trend", [
                ["Bucket", "Seats revoked", "Permissions revoked",
                 "Cloud grants revoked", "API keys revoked",
                 "Onboarding done", "Offboarding done",
                 "Requests processed", "Tasks completed",
                 "Notifications delivered"],
                *[[r["bucket"], r["seats_revoked"], r["permissions_revoked"],
                   r["cloud_grants_revoked"], r["api_keys_revoked"],
                   r["onboarding_completed"], r["offboarding_completed"],
                   r["requests_processed"], r["tasks_completed"],
                   r["notifications_delivered"]] for r in data["trend"]],
            ])
        if data.get("cumulative"):
            add_sheet("Cumulative", [
                ["Bucket", "Seats", "Permissions", "Cloud grants",
                 "API keys", "Total reclaimed"],
                *[[r["bucket"], r["seats_revoked"], r["permissions_revoked"],
                   r["cloud_grants_revoked"], r["api_keys_revoked"],
                   r["total_reclaimed"]] for r in data["cumulative"]],
            ])
        return

    if section == "cloud":
        servers = data.get("servers") or {}
        domains = data.get("domains") or {}
        billing = data.get("billing") or {}
        scopes = data.get("scopes") or {}
        grants = data.get("grants") or {}
        # 口径必须写进工作簿:一份离线的表格没有任何 UI 徽标可依赖,
        # 而「业务折前(list)」与「实付(invoice)」的同名 USD 列相差
        # 一个公司折扣率。basis 一并写出,因为服务榜与合计不同源
        # (合计来自日表/月表,服务榜来自 cost_facts)。
        b_measure = billing.get("measure") or "list"
        add_sheet("Summary", [
            ["Servers total", servers.get("total", 0)],
            ["Domains total", domains.get("total", 0)],
            [
                f"Billing total period (USD, measure={b_measure})",
                billing.get("total_period_usd", 0),
            ],
            [
                "Billing measure",
                "list = pre-discount engineering cost"
                if b_measure == "list"
                else "invoice = net paid (vendor invoice)",
            ],
            ["Billing aggregate basis", billing.get("aggregate_basis") or ""],
            ["Billing service basis", billing.get("service_basis") or ""],
            [
                "Amounts excluded (no FX rate)",
                ", ".join(
                    f"{cur} {amt:.4f}"
                    for cur, amt in sorted(
                        (billing.get("unconverted_by_currency") or {}).items()
                    )
                ) or "none",
            ],
            ["Grants expiring 14d", grants.get("expiring_14d", 0)],
        ])
        if servers.get("by_status"):
            add_sheet("Server Status", [["Status", "Count"],
                                        *[[k, v] for k, v in servers["by_status"].items()]])
        if servers.get("by_region"):
            add_sheet("Server Region", [["Region", "Count"],
                                        *[[k, v] for k, v in servers["by_region"].items()]])
        if domains.get("expiry_buckets"):
            add_sheet("Domain Expiry", [["Bucket", "Count"],
                                        *[[k, v] for k, v in domains["expiry_buckets"].items()]])
        if domains.get("upcoming_calendar"):
            add_sheet("Domain Calendar", [
                ["Domain", "Expiry", "Registrar", "Days left"],
                *[[r["domain_name"], r["expiry_date"], r["registrar"], r["days_left"]]
                  for r in domains["upcoming_calendar"]],
            ])
        # 每个金额表头都带上口径 —— 单独抽出一张 sheet 传给别人时,
        # Summary 里的口径行不会跟着走。
        if billing.get("trend"):
            add_sheet("Billing Trend", [["Bucket", f"Cost (USD, {b_measure})"],
                                        *[[r["bucket"], r["cost_usd"]] for r in billing["trend"]]])
        if billing.get("by_service_top"):
            add_sheet("Top Services", [["Service", f"Cost (USD, {b_measure})"],
                                       *[[k, v] for k, v in billing["by_service_top"].items()]])
        if billing.get("by_integration_top"):
            add_sheet("By Integration", [["Integration ID", f"Cost (USD, {b_measure})"],
                                         *[[r["integration_id"], r["cost_usd"]]
                                           for r in billing["by_integration_top"]]])
        if scopes.get("by_status"):
            add_sheet("Scope Status", [["Status", "Count"],
                                       *[[k, v] for k, v in scopes["by_status"].items()]])
        if scopes.get("by_provider"):
            add_sheet("Scope Provider", [["Provider", "Count"],
                                         *[[k, v] for k, v in scopes["by_provider"].items()]])
        if grants.get("by_status"):
            add_sheet("Grant Status", [["Status", "Count"],
                                       *[[k, v] for k, v in grants["by_status"].items()]])
        return