"""Helix 此刻 · 运管驾驶舱 — the leadership/IT operational pulse of the platform.

``build_cockpit`` returns the full operational picture (people / assets / cost /
cloud / ops backlog / compliance) as labelled metric sections, gated to
admin-read roles since it includes cost.

(The former employee-facing 广场 ambient pulse — ``build_employee_pulse`` — was
removed 2026-06-22 per product decision; "Helix 此刻" is now leadership-only.)

Read-only assembly layer (same pattern as ``me_universe`` / the dashboard
summary): it reaches across domain models to count, never mutates.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.fx_rate import CloudCostUsdConverter
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)


_TERMINAL_PROC = ("completed", "cancelled")


def _metric(key: str, value: Any, tone: str = "neutral") -> dict[str, Any]:
    """One labelled metric. ``label_key`` is rendered frontend-side
    (``cockpit.metric.<key>``) so every number carries its 口径 inline — no bare
    number grids. ``tone`` drives the accent (neutral / good / warn / bad)."""
    return {"key": key, "label_key": f"cockpit.metric.{key}", "value": value, "tone": tone}


async def build_cockpit(db: AsyncSession) -> dict[str, Any]:
    """运管驾驶舱 — the full operational picture for leadership / IT.

    Admin-gated (the router enforces ``can read admin``). Grouped into equal-
    weight sections, every metric labelled with its 口径; cost is included
    (admin-only surface). Read-only assembly layer."""
    from app.compliance.models import ComplianceIncident
    from app.domains.asset.models import (
        AssetItem,
        SoftwareLicensePlan,
        SoftwareSeatAllocation,
    )
    from app.domains.cmdb.models import CmdbDomain, CmdbServer
    from app.domains.offboarding.models import (
        XFER_PENDING,
        AssetTransferRequest,
        OffboardingProcess,
    )
    from app.domains.onboarding.models import OnboardingProcess
    from app.domains.provisioning.models import (
        STATUS_IN_PROGRESS,
        STATUS_NOTIFIED,
        STATUS_PENDING,
        ProvisioningTask,
    )
    from app.domains.support.models import TERMINAL_STATUSES, SupportTicket

    today = datetime.now(UTC).date()
    month_start = today.replace(day=1)
    PROC_BLOCKED = "blocked"

    async def _count(stmt) -> int:
        return int(await db.scalar(stmt) or 0)

    # ── People ──
    headcount = await _count(
        select(func.count()).select_from(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )
    new_this_month = await _count(
        select(func.count()).select_from(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
            IdentityPerson.hire_date >= month_start,
            IdentityPerson.hire_date <= today,
        )
    )
    onboarding_open = await _count(
        select(func.count()).select_from(OnboardingProcess).where(
            OnboardingProcess.status.notin_(_TERMINAL_PROC),
        )
    )
    offboarding_open = await _count(
        select(func.count()).select_from(OffboardingProcess).where(
            OffboardingProcess.status.notin_(_TERMINAL_PROC),
        )
    )
    leavers_blocked = await _count(
        select(func.count()).select_from(OffboardingProcess).where(
            OffboardingProcess.status == PROC_BLOCKED,
        )
    )

    # ── Assets ──
    asset_rows = (await db.execute(
        select(AssetItem.status, func.count())
        .where(AssetItem.deleted_at.is_(None))
        .group_by(AssetItem.status)
    )).all()
    asset_by_status = {k: int(v) for k, v in asset_rows}
    asset_total = sum(asset_by_status.values())
    warranty_expiring = await _count(
        select(func.count()).select_from(AssetItem).where(
            AssetItem.deleted_at.is_(None),
            AssetItem.warranty_until.is_not(None),
            AssetItem.warranty_until >= today,
            AssetItem.warranty_until <= today + timedelta(days=30),
        )
    )

    # ── Software cost (monthly-equiv across active seats) ──
    seat_rows = (await db.execute(
        select(SoftwareSeatAllocation.plan_id, func.count())
        .where(SoftwareSeatAllocation.status == "active")
        .group_by(SoftwareSeatAllocation.plan_id)
    )).all()
    active_seats = sum(int(c) for _, c in seat_rows)
    plan_ids = [pid for pid, _ in seat_rows if pid]
    # Plan prices are plan-level native currency (``SoftwareLicensePlan.currency``
    # is NOT NULL default 'CNY'). The cockpit renders this metric as USD, so every
    # plan's monthly-equiv amount is normalised to USD via the FX converter before
    # summing — otherwise raw CNY would be stamped as USD (~7x overstatement) on a
    # leadership dashboard. Matches billing_service's per-plan currency handling.
    converter = CloudCostUsdConverter(db)
    monthly_software = Decimal("0")
    if plan_ids:
        plans = (await db.execute(
            select(SoftwareLicensePlan).where(SoftwareLicensePlan.id.in_(plan_ids))
        )).scalars().all()
        plan_by_id = {p.id: p for p in plans}
        for pid, cnt in seat_rows:
            plan = plan_by_id.get(pid) if pid else None
            if plan is None or plan.price is None:
                continue
            cycle = (plan.billing_cycle or "").lower()
            if cycle == "monthly":
                native = Decimal(str(plan.price)) * int(cnt)
            elif cycle == "yearly":
                native = (Decimal(str(plan.price)) / Decimal("12")) * int(cnt)
            else:
                continue
            # ``to_usd_checked``, never ``to_usd``: a plan in a currency no FX
            # rate covers (or CNY with the platform rate missing) converts to
            # ``0.0``, and a fabricated $0 is indistinguishable from "this plan
            # costs nothing" on a leadership dashboard. Unconvertible plans are
            # excluded from the total and disclosed below, same contract as
            # every other cost read path (``/billing/overview``,
            # ``/dashboard/insights``, the MCP cost tools).
            usd, converted = await converter.to_usd_checked(
                native, plan.currency, None,
            )
            if not converted:
                continue
            monthly_software += Decimal(str(usd))
    # 折不出汇率的原生额(按币种)。非空 ⇒ ``monthly_software_cost`` 是**下
    # 界**,不是全量 —— 响应顶层披露 + 该指标 tone 转 warn,驾驶舱不许把一个
    # 不完整的合计当成完整的报出来。
    software_unconverted = dict(converter.unconverted_by_currency)
    if software_unconverted:
        logger.warning(
            "cockpit monthly_software_cost excludes %d currency(ies) with no "
            "resolvable FX rate: %s — the reported total is a lower bound",
            len(software_unconverted), software_unconverted,
        )

    # ── Cloud ──
    cloud_servers = await _count(
        select(func.count()).select_from(CmdbServer).where(CmdbServer.deleted_at.is_(None))
    )
    cloud_domains = await _count(
        select(func.count()).select_from(CmdbDomain).where(CmdbDomain.deleted_at.is_(None))
    )

    # ── Ops backlog ──
    provisioning_open = await _count(
        select(func.count()).select_from(ProvisioningTask).where(
            ProvisioningTask.status.in_((STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED)),
        )
    )
    open_transfers = await _count(
        select(func.count()).select_from(AssetTransferRequest).where(
            AssetTransferRequest.status == XFER_PENDING,
        )
    )
    support_open = await _count(
        select(func.count()).select_from(SupportTicket).where(
            SupportTicket.status.notin_(TERMINAL_STATUSES),
        )
    )

    # ── Compliance ──
    open_incidents = await _count(
        select(func.count()).select_from(ComplianceIncident).where(
            ComplianceIncident.status == "open",
        )
    )

    return {
        "generated_at": datetime.now(UTC).isoformat(),
        # 与 ``/cmdb/billing/overview`` / ``/dashboard/insights`` 同一披露键:
        # 非 None ⇒ 有支出因为折不出汇率而**没进** ``monthly_software_cost``,
        # 值是按币种的原生额。渲染方是 ``web/src/pages/admin/CockpitPage.tsx``
        # 的 warning Alert(``cockpit.fxIncomplete`` + 原生额列表,形状同
        # ``CloudBillingPage``),由 ``CockpitPage.test.tsx`` 盯着 —— 只发这个键
        # 而页面不渲染,等于把一个橙色的 $0.00 摆在领导面前,和这个改造要消灭
        # 的那句谎没有区别。MCP 双胞胎(``helix_cockpit``)拿到同一份 dict。
        "unconverted_by_currency": software_unconverted or None,
        "sections": [
            {
                "key": "people",
                "title_key": "cockpit.section.people",
                "metrics": [
                    _metric("headcount", headcount),
                    _metric("new_this_month", new_this_month, "good"),
                    _metric("onboarding_open", onboarding_open),
                    _metric("offboarding_open", offboarding_open),
                    _metric("leavers_blocked", leavers_blocked, "bad" if leavers_blocked else "good"),
                ],
            },
            {
                "key": "assets",
                "title_key": "cockpit.section.assets",
                "metrics": [
                    _metric("asset_total", asset_total),
                    _metric("asset_assigned", asset_by_status.get("assigned", 0)),
                    _metric("asset_idle", asset_by_status.get("idle", 0)),
                    _metric("asset_held_by_leaver", asset_by_status.get("held_by_leaver", 0),
                            "warn" if asset_by_status.get("held_by_leaver", 0) else "good"),
                    _metric("warranty_expiring", warranty_expiring,
                            "warn" if warranty_expiring else "neutral"),
                ],
            },
            {
                "key": "cost",
                "title_key": "cockpit.section.cost",
                "metrics": [
                    _metric("active_seats", active_seats),
                    _metric("monthly_software_cost",
                            float(monthly_software.quantize(Decimal("0.01"))),
                            "warn" if software_unconverted else "neutral"),
                ],
            },
            {
                "key": "cloud",
                "title_key": "cockpit.section.cloud",
                "metrics": [
                    _metric("cloud_servers", cloud_servers),
                    _metric("cloud_domains", cloud_domains),
                ],
            },
            {
                "key": "ops",
                "title_key": "cockpit.section.ops",
                "metrics": [
                    _metric("provisioning_open", provisioning_open,
                            "warn" if provisioning_open else "good"),
                    _metric("open_transfers", open_transfers),
                    _metric("support_open", support_open,
                            "warn" if support_open else "good"),
                ],
            },
            {
                "key": "compliance",
                "title_key": "cockpit.section.compliance",
                "metrics": [
                    _metric("open_incidents", open_incidents,
                            "bad" if open_incidents else "good"),
                ],
            },
        ],
    }


__all__ = ["build_cockpit"]