"""Dashboard summary endpoint.

One call, role-aware. Returns three optional sections — admin /
manager / employee — populated according to the caller's role set.
All three can be present (a global_admin who is also someone's
manager sees admin + manager + their own employee section).

The response is intentionally wide and shallow: UI can render
whatever sections it cares about without extra round-trips. Each
section is cheap to compute (indexed counts + a handful of
``LIMIT 20`` SELECTs) so the whole summary returns in one DB round.
"""
from __future__ import annotations

from datetime import date, timedelta
from decimal import Decimal
from typing import Any
from uuid import UUID

from fastapi import APIRouter, Depends, Request
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.core.person_scope import SubordinateScope, subordinate_person_ids
from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role
from app.core.response import success_response
from app.domains.apikey.models import ApiKeyAssignment
from app.domains.asset.models import (
    AssetItem,
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.feed.models import FeedBlessing, FeedCelebration
from app.domains.offboarding.models import (
    PROC_BLOCKED,
    RESO_PENDING,
    XFER_PENDING,
    AssetTransferRequest,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.domains.provisioning.models import (
    STATUS_IN_PROGRESS,
    STATUS_NOTIFIED,
    STATUS_PENDING,
    ProvisioningTask,
)
from app.identity.person_models import IdentityPerson

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


@router.get(
    "/cockpit",
    summary="运管驾驶舱 — full operational picture (leadership / admin only)",
)
async def helix_cockpit(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """The leadership/IT operational cockpit — people, assets, cost, cloud, ops
    backlog, compliance — as labelled metric sections. Gated to admin-read
    roles (``global_admin`` / ``management``); cost is included, hence the gate.
    Read-only assembly layer."""
    from app.core.exceptions import ForbiddenError
    from app.core.rbac import MANAGEMENT_ROLE
    roles = user.role_names or []
    if not (
        has_role(roles, GLOBAL_ADMIN_ROLE)
        or has_role(roles, "it_admin")
        or has_role(roles, MANAGEMENT_ROLE)
    ):
        raise ForbiddenError("运管驾驶舱 requires global_admin or management.")
    from app.dashboard.pulse import build_cockpit
    data = await build_cockpit(db)
    return success_response(data=data, request=request)


@router.get("/summary", summary="Role-aware dashboard summary")
async def dashboard_summary(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    today = date.today()
    person_id = user.id
    is_admin = has_role(user.role_names or [], GLOBAL_ADMIN_ROLE) or has_role(
        user.role_names or [], "it_admin",
    )

    admin_section: dict[str, Any] | None = None
    manager_section: dict[str, Any] | None = None
    employee_section = await _employee_section(db, person_id, today)

    # Manager = has at least one subordinate.
    subordinate_ids = await _subordinate_ids(db, person_id)
    if subordinate_ids:
        manager_section = await _manager_section(db, subordinate_ids, today)

    if is_admin:
        admin_section = await _admin_section(db, today)

    return success_response(
        data={
            "role": "admin" if is_admin else ("manager" if manager_section else "employee"),
            "admin": admin_section,
            "manager": manager_section,
            "employee": employee_section,
        },
        request=request,
    )


# ── Employee ──────────────────────────────────────────────────────


async def _employee_section(
    db: AsyncSession, person_id: UUID, today: date,
) -> dict[str, Any]:
    person = (
        await db.execute(select(IdentityPerson).where(IdentityPerson.id == person_id))
    ).scalar_one_or_none()

    my_assets = (
        await db.execute(
            select(func.count()).select_from(AssetItem).where(
                and_(
                    AssetItem.assigned_to == person_id,
                    AssetItem.status == "assigned",
                )
            )
        )
    ).scalar() or 0

    my_seats = (
        await db.execute(
            select(func.count()).select_from(SoftwareSeatAllocation).where(
                and_(
                    SoftwareSeatAllocation.user_id == person_id,
                    SoftwareSeatAllocation.status == "active",
                )
            )
        )
    ).scalar() or 0

    my_api_keys = (
        await db.execute(
            select(func.count()).select_from(ApiKeyAssignment).where(
                and_(
                    ApiKeyAssignment.assigned_to == person_id,
                    ApiKeyAssignment.status.in_(["active", "approved"]),
                )
            )
        )
    ).scalar() or 0

    pending_incoming_transfers = (
        await db.execute(
            select(func.count()).select_from(AssetTransferRequest).where(
                and_(
                    AssetTransferRequest.recipient_id == person_id,
                    AssetTransferRequest.status == XFER_PENDING,
                )
            )
        )
    ).scalar() or 0

    my_open_tasks = (
        await db.execute(
            select(func.count()).select_from(ProvisioningTask).where(
                and_(
                    ProvisioningTask.user_id == person_id,
                    ProvisioningTask.status.in_(
                        [STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED]
                    ),
                )
            )
        )
    ).scalar() or 0

    # NOTE (2026-09-05, checked): this block is emitted for ANY process
    # status, terminal included — a cancelled offboarding still yields
    # ``offboarding_info`` with a ``days_left`` and a ``pending_blockers``
    # count. That was left as-is rather than "fixed" because the payload
    # currently has NO consumer: ``offboarding_info`` / ``pending_blockers``
    # have zero hits across ``web/src``, so nothing renders it and no
    # employee is being told anything wrong today.
    #
    # That is a fact with a shelf life, and it is the dangerous half of the
    # pair: a field that does not exist fails LOUDLY when someone starts
    # reading it, while a field that exists and is wrong renders a
    # plausible-looking number in silence. **Before rendering any of this,
    # constrain the process to the open set** —
    # ``status.notin_([PROC_COMPLETED, PROC_CANCELLED])``, the same filter
    # `leaver_blocked/dashboard_service`, `health/service` and the two
    # mobile admin surfaces use (#3455) — because a terminal process's
    # pending blockers are frozen: ``cancel_process`` leaves them alone and
    # ``reconcile_blockers`` refuses terminal processes, so nothing will
    # ever resolve them and the count can only be wrong upward.
    offboarding_info: dict[str, Any] | None = None
    proc = (
        await db.execute(
            select(OffboardingProcess).where(OffboardingProcess.person_id == person_id)
        )
    ).scalar_one_or_none()
    if proc and person and person.termination_date:
        pending_blockers = (
            await db.execute(
                select(func.count()).select_from(OffboardingBlocker).where(
                    and_(
                        OffboardingBlocker.process_id == proc.id,
                        OffboardingBlocker.resolution == RESO_PENDING,
                    )
                )
            )
        ).scalar() or 0
        days_left = (person.termination_date - today).days
        offboarding_info = {
            "process_status": proc.status,
            "termination_date": person.termination_date.isoformat(),
            "days_left": days_left,
            "pending_blockers": pending_blockers,
        }

    # ── Cost snapshot (transparency: tell the employee what their tools cost) ──
    # We compute personal monthly software cost the same way the admin
    # report does — yearly plans amortised /12, one_time plans excluded.
    cost_rows = (
        await db.execute(
            select(SoftwareLicensePlan.price, SoftwareLicensePlan.billing_cycle)
            .select_from(SoftwareSeatAllocation)
            .join(SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id)
            .where(
                and_(
                    SoftwareSeatAllocation.user_id == person_id,
                    SoftwareSeatAllocation.status == "active",
                )
            )
        )
    ).all()
    monthly_software_cost = 0.0
    for price, cycle in cost_rows:
        if price is None:
            continue
        p = float(price) if isinstance(price, Decimal) else float(price)
        if cycle == "monthly":
            monthly_software_cost += p
        elif cycle == "yearly":
            monthly_software_cost += p / 12.0

    # Total purchase value of my assigned hardware (best-effort: NULLs ignored)
    asset_value = (
        await db.execute(
            select(func.coalesce(func.sum(AssetItem.purchase_cost), 0)).where(
                and_(
                    AssetItem.assigned_to == person_id,
                    AssetItem.status == "assigned",
                )
            )
        )
    ).scalar() or 0
    asset_value = float(asset_value) if isinstance(asset_value, Decimal) else float(asset_value)

    # ── Expiring soon: API keys within 30 days, assigned assets with
    # warranty expiring within 90 days. ──
    expiring_keys_rows = (
        await db.execute(
            select(
                ApiKeyAssignment.id,
                ApiKeyAssignment.request_number,
                ApiKeyAssignment.project_name,
                ApiKeyAssignment.valid_until,
            ).where(
                and_(
                    ApiKeyAssignment.assigned_to == person_id,
                    ApiKeyAssignment.status == "active",
                    ApiKeyAssignment.valid_until.is_not(None),
                    ApiKeyAssignment.valid_until >= today,
                    ApiKeyAssignment.valid_until <= today + timedelta(days=30),
                )
            )
            .order_by(ApiKeyAssignment.valid_until)
        )
    ).all()
    expiring_api_keys = [
        {
            "id": str(kid),
            "request_number": rn,
            "project_name": pn,
            "valid_until": vu.isoformat(),
            "days_left": (vu - today).days,
        }
        for kid, rn, pn, vu in expiring_keys_rows
    ]
    expiring_warranty_rows = (
        await db.execute(
            select(
                AssetItem.id,
                AssetItem.asset_tag,
                AssetItem.asset_name,
                AssetItem.warranty_until,
            ).where(
                and_(
                    AssetItem.assigned_to == person_id,
                    AssetItem.status == "assigned",
                    AssetItem.warranty_until.is_not(None),
                    AssetItem.warranty_until >= today,
                    AssetItem.warranty_until <= today + timedelta(days=90),
                )
            )
            .order_by(AssetItem.warranty_until)
        )
    ).all()
    expiring_warranty = [
        {
            "asset_id": str(aid),
            "asset_tag": tag,
            "asset_name": name,
            "warranty_until": wu.isoformat(),
            "days_left": (wu - today).days,
        }
        for aid, tag, name, wu in expiring_warranty_rows
    ]

    # ── My celebrations (last 7d) — surface when the home-page viewer
    # is themselves the celebrant so they can see how many blessings
    # have rolled in. The feed page is where they'd normally see this,
    # but on the dashboard it's a small "your moment" banner. Limit to
    # the rolling window the feed card stays visible for. ──
    #
    # ``FeedCelebration.person_id`` is ``String(36)`` — bind a stringified
    # UUID, never the UUID object (the asyncmy-on-CHAR bind quirk).
    my_celebrations_rows = (
        await db.execute(
            select(
                FeedCelebration.id,
                FeedCelebration.celebration_type,
                FeedCelebration.occasion_date,
                FeedCelebration.years,
                func.count(FeedBlessing.id).label("blessing_count"),
            )
            .outerjoin(
                FeedBlessing, FeedBlessing.celebration_id == FeedCelebration.id,
            )
            .where(
                and_(
                    FeedCelebration.person_id == str(person_id),
                    FeedCelebration.occasion_date <= today,
                    FeedCelebration.occasion_date >= today - timedelta(days=7),
                )
            )
            .group_by(
                FeedCelebration.id,
                FeedCelebration.celebration_type,
                FeedCelebration.occasion_date,
                FeedCelebration.years,
            )
            .order_by(FeedCelebration.occasion_date.desc())
        )
    ).all()
    my_celebrations = [
        {
            "celebration_id": str(cid),
            "celebration_type": ctype,
            "occasion_date": odate.isoformat(),
            "years": int(years or 0),
            "blessing_count": int(bcount or 0),
        }
        for cid, ctype, odate, years, bcount in my_celebrations_rows
    ]

    # ── Recent personal activity (last 10 audit_logs by me) ──
    recent_rows = (
        await db.execute(
            select(
                AuditLog.id,
                AuditLog.action,
                AuditLog.resource_type,
                AuditLog.resource_id,
                AuditLog.status,
                AuditLog.created_at,
            )
            .where(
                and_(
                    AuditLog.actor_user_id == person_id,
                    AuditLog.actor_type == "user",
                )
            )
            .order_by(AuditLog.created_at.desc())
            .limit(10)
        )
    ).all()
    recent_activity = [
        {
            "id": str(rid),
            "action": action,
            "resource_type": rtype,
            "resource_id": rid_,
            "status": status,
            "created_at": ts.isoformat() if ts else None,
        }
        for rid, action, rtype, rid_, status, ts in recent_rows
    ]

    return {
        "person_name": person.display_name if person else None,
        "my_assets_count": int(my_assets),
        "my_seats_count": int(my_seats),
        "my_api_keys_count": int(my_api_keys),
        "pending_incoming_transfers": int(pending_incoming_transfers),
        "my_open_tasks": int(my_open_tasks),
        "offboarding": offboarding_info,
        # New (Insights v1) — employee-facing personal panels.
        "cost_snapshot": {
            "monthly_software_cost": round(monthly_software_cost, 2),
            "assigned_asset_value": round(asset_value, 2),
        },
        "expiring_api_keys": expiring_api_keys,
        "expiring_warranty": expiring_warranty,
        "my_celebrations": my_celebrations,
        "recent_activity": recent_activity,
    }


# ── Manager ───────────────────────────────────────────────────────


async def _subordinate_ids(db: AsyncSession, person_id: UUID) -> list[UUID]:
    # A departed report (employment_status == "left", set once their last
    # day passes / offboarding completes) is no longer part of the
    # manager's *current* team — they must drop off the headcount, the
    # subordinate list, AND leaving_soon (all three derive from this set).
    # ``leaver_pending`` is intentionally KEPT — they are still employed
    # until their last day and are exactly who "即将离职" surfaces. That is
    # precisely the ``MANAGED`` scope of the canonical resolver. Direct
    # reports only (``max_depth=1``) — the dashboard "我的团队" card is a
    # one-level view.
    return await subordinate_person_ids(
        db, person_id, scope=SubordinateScope.MANAGED, max_depth=1,
    )


async def _manager_section(
    db: AsyncSession, sub_ids: list[UUID], today: date,
) -> dict[str, Any]:
    # Top-N subordinate snapshot
    subs = (
        await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(sub_ids))
            .order_by(IdentityPerson.display_name)
            .limit(100)
        )
    ).scalars().all()

    # For each subordinate, count open requests (provisioning tasks) and
    # whether they have an upcoming leaver_date. Single query each for
    # simplicity (N ≤ 100 is fine).
    open_tasks_by_user = dict(
        (
            await db.execute(
                select(ProvisioningTask.user_id, func.count())
                .where(
                    and_(
                        ProvisioningTask.user_id.in_(sub_ids),
                        ProvisioningTask.status.in_(
                            [STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED]
                        ),
                    )
                )
                .group_by(ProvisioningTask.user_id)
            )
        ).all()
    )

    proc_rows = (
        await db.execute(
            select(OffboardingProcess).where(OffboardingProcess.person_id.in_(sub_ids))
        )
    ).scalars().all()
    proc_by_person = {p.person_id: p for p in proc_rows}

    total_open_tasks = sum(open_tasks_by_user.values()) if open_tasks_by_user else 0

    # Bulk-load pending blocker counts for all processes in one query
    # (avoids N+1 when a manager has many leaving subordinates).
    proc_ids = [pr.id for pr in proc_rows]
    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()
        )

    leaving_soon: list[dict[str, Any]] = []
    for p in subs:
        # "即将离职" = a subordinate still on their way out. Once their last
        # day passes / offboarding completes, hr_lifecycle.advance_service
        # flips employment_status to ``left`` — at that point they have
        # already departed and must drop off this forward-looking panel.
        # Without the ``left`` guard a completed leaver lingers here with a
        # negative days_left (e.g. termination 8 days ago → days_left == -8,
        # rendered by the FE as "T--8 天") and keeps inflating the count.
        if (
            p.termination_date
            and p.employment_status != "left"
            and (p.termination_date - today).days <= 30
        ):
            proc = proc_by_person.get(p.id)
            pending_blockers = int(blocker_counts.get(proc.id, 0)) if proc else 0
            leaving_soon.append({
                "person_id": str(p.id),
                "display_name": p.display_name,
                "work_email": p.work_email,
                "termination_date": p.termination_date.isoformat(),
                "days_left": (p.termination_date - today).days,
                "pending_blockers": pending_blockers,
                "process_status": proc.status if proc else None,
            })

    # Pending transfer requests FROM subordinates TO me (escalations)
    incoming_escalations = (
        await db.execute(
            select(func.count()).select_from(AssetTransferRequest).where(
                and_(
                    AssetTransferRequest.initiator_id.in_(sub_ids),
                    AssetTransferRequest.status == XFER_PENDING,
                )
            )
        )
    ).scalar() or 0

    return {
        "subordinate_count": len(sub_ids),
        "subordinates": [
            {
                "person_id": str(p.id),
                "display_name": p.display_name,
                "work_email": p.work_email,
                "job_title": p.job_title,
                "employment_status": p.employment_status,
                "open_tasks": int(open_tasks_by_user.get(p.id, 0) or 0),
            }
            for p in subs
        ],
        "team_open_tasks": int(total_open_tasks),
        "leaving_soon": leaving_soon,
        "incoming_escalations": int(incoming_escalations),
    }


# ── Admin ─────────────────────────────────────────────────────────


async def _admin_section(db: AsyncSession, today: date) -> dict[str, Any]:
    # Provisioning task counts by status (today's backlog view)
    status_rows = (
        await db.execute(
            select(ProvisioningTask.status, func.count())
            .group_by(ProvisioningTask.status)
        )
    ).all()
    status_counts = {k: int(v) for k, v in status_rows}

    # By source (onboarding / offboarding / request / role_change) — open only
    source_rows = (
        await db.execute(
            select(ProvisioningTask.source, func.count())
            .where(
                ProvisioningTask.status.in_(
                    [STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED]
                )
            )
            .group_by(ProvisioningTask.source)
        )
    ).all()
    open_by_source = {k: int(v) for k, v in source_rows}

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

    # Leaver process status breakdown
    proc_rows = (
        await db.execute(
            select(OffboardingProcess.status, func.count())
            .group_by(OffboardingProcess.status)
        )
    ).all()
    process_counts = {k: int(v) for k, v in proc_rows}

    # Blocked leavers (past termination_date AND still held)
    blocked_past_due = (
        await db.execute(
            select(func.count())
            .select_from(OffboardingProcess)
            .join(IdentityPerson, IdentityPerson.id == OffboardingProcess.person_id)
            .where(
                and_(
                    OffboardingProcess.status == PROC_BLOCKED,
                    IdentityPerson.termination_date.is_not(None),
                    IdentityPerson.termination_date <= today,
                )
            )
        )
    ).scalar() or 0

    # This-month hires / leavers
    month_start = date(today.year, today.month, 1)
    month_end = (
        date(today.year + (today.month // 12), (today.month % 12) + 1, 1)
        if today.month < 12
        else date(today.year + 1, 1, 1)
    )
    this_month_hires = (
        await db.execute(
            select(func.count())
            .select_from(IdentityPerson)
            .where(
                and_(
                    IdentityPerson.hire_date.is_not(None),
                    IdentityPerson.hire_date >= month_start,
                    IdentityPerson.hire_date < month_end,
                )
            )
        )
    ).scalar() or 0
    this_month_leavers = (
        await db.execute(
            select(func.count())
            .select_from(IdentityPerson)
            .where(
                and_(
                    IdentityPerson.termination_date.is_not(None),
                    IdentityPerson.termination_date >= month_start,
                    IdentityPerson.termination_date < month_end,
                )
            )
        )
    ).scalar() or 0

    # Asset inventory snapshot
    asset_rows = (
        await db.execute(
            select(AssetItem.status, func.count())
            .group_by(AssetItem.status)
        )
    ).all()
    asset_by_status = {k: int(v) for k, v in asset_rows}

    # Software catalog & offboarding-policy health
    sw_policy_rows = (
        await db.execute(
            select(AssetSoftwareCatalog.offboarding_policy, func.count())
            .group_by(AssetSoftwareCatalog.offboarding_policy)
        )
    ).all()
    software_by_policy = {k: int(v) for k, v in sw_policy_rows}

    return {
        "provisioning": {
            "status_counts": status_counts,
            "open_by_source": open_by_source,
        },
        "open_transfers": int(open_transfers),
        "offboarding": {
            "process_counts": process_counts,
            "blocked_past_due": int(blocked_past_due),
        },
        "lifecycle": {
            "this_month_hires": int(this_month_hires),
            "this_month_leavers": int(this_month_leavers),
        },
        "assets": {"by_status": asset_by_status},
        "software": {"by_policy": software_by_policy},
    }