"""信息安全自评估 — real-data aggregators per report section.

Each aggregator pulls the structured numbers a section's table needs from
the LIVE Helix tables (assets, RBAC, MFA, offboarding, access review,
exposure, change/provisioning, secrets, incidents, cloud, periodic
self-check). This is the "能自动生成的就自动生成" half: the auditor never
hand-counts servers or admins again — they only write/confirm the narrative
prose and the 评估结论.

Returns per source ``{"table": [row,...], "summary": str, "numbers": {...}}``.
Point-in-time sources (inventory, RBAC, MFA) ignore the period window;
flow sources (change mgmt, incidents, offboarding-in-period) honour it.

Datetime discipline: naive-UTC throughout (DateTime columns load naive).
"""
from __future__ import annotations

from datetime import date, datetime, time
from typing import Any

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

# zh labels for the known asset categories (seed in migration 205).
_CATEGORY_ZH: dict[str, str] = {
    "laptop": "笔记本",
    "desktop": "台式机",
    "monitor": "显示器",
    "printer": "打印机",
    "server": "服务器",
    "network_device": "网络设备",
    "accessory": "配件",
    "conference_display": "会议电视",
}
_DEVICE_CATEGORIES = {"laptop", "desktop", "monitor", "conference_display"}


def _window(ps: date, pe: date) -> tuple[datetime, datetime]:
    return datetime.combine(ps, time.min), datetime.combine(pe, time.max)


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


# ════════════════════════════════════════════════════════════════════
# Aggregators
# ════════════════════════════════════════════════════════════════════


async def _agg_asset_inventory(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.asset.models import AssetItem

    rows = (await db.execute(
        select(AssetItem.category, func.count())
        .where(AssetItem.deleted_at.is_(None))
        .group_by(AssetItem.category)
        .order_by(func.count().desc())
    )).all()
    table = [
        {"asset_type": _CATEGORY_ZH.get(cat, cat or "未分类"),
         "count": int(n), "note": ""}
        for cat, n in rows
    ]
    total = sum(r["count"] for r in table)
    return {
        "table": table,
        "summary": f"在册资产 {total} 件,覆盖 {len(table)} 个类别",
        "numbers": {"total": total},
    }


async def _agg_device_inventory(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.asset.models import AssetItem

    rows = (await db.execute(
        select(AssetItem.category, func.count())
        .where(AssetItem.deleted_at.is_(None),
               AssetItem.category.in_(_DEVICE_CATEGORIES))
        .group_by(AssetItem.category)
        .order_by(func.count().desc())
    )).all()
    table = [
        {"asset_type": _CATEGORY_ZH.get(cat, cat), "count": int(n), "note": ""}
        for cat, n in rows
    ]
    total = sum(r["count"] for r in table)
    return {
        "table": table,
        "summary": f"终端设备 {total} 台",
        "numbers": {"total": total},
    }


async def _agg_headcount(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.identity.person_models import IdentityPerson

    # Tombstones stay out of a signed compliance artefact. A soft-deleted
    # row keeps employment_status='active' in this codebase, so without
    # the deleted_at filter every purged duplicate inflated the
    # "在职员工 N 人" headline of the ISMS self-assessment.
    rows = (await db.execute(
        select(IdentityPerson.employment_status, func.count())
        .where(IdentityPerson.deleted_at.is_(None))
        .group_by(IdentityPerson.employment_status)
    )).all()
    by_status = {s: int(n) for s, n in rows}
    active = by_status.get("active", 0)
    table = [
        {"metric": "在职员工", "value": active},
        {"metric": "全部人员记录", "value": sum(by_status.values())},
    ]
    return {
        "table": table,
        "summary": f"在职员工 {active} 人",
        "numbers": {"active": active, "by_status": by_status},
    }


async def _agg_offboarding_flow(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.offboarding.models import (
        PROC_COMPLETED,
        OffboardingProcess,
    )

    start, end = _window(ps, pe)
    total = await _count(
        db,
        select(func.count()).select_from(OffboardingProcess)
        .where(OffboardingProcess.created_at >= start,
               OffboardingProcess.created_at <= end),
    )
    completed = await _count(
        db,
        select(func.count()).select_from(OffboardingProcess)
        .where(OffboardingProcess.created_at >= start,
               OffboardingProcess.created_at <= end,
               OffboardingProcess.status == PROC_COMPLETED),
    )
    table = [
        {"metric": "本期离职流程", "value": total},
        {"metric": "已完成", "value": completed},
        {"metric": "离职处理流程", "value": "通知 → 权限撤销 → 资产归还 → 安全检查"},
    ]
    return {
        "table": table,
        "summary": f"本期离职流程 {total} 例,已完成 {completed} 例",
        "numbers": {"total": total, "completed": completed},
    }


async def _agg_rbac_access(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.auth.models import AuthRole, AuthUserRole
    from app.identity.person_models import IdentityPerson

    # Count only ACTIVE people. AuthUserRole rows persist for leavers
    # (RBAC cleanup is a separate best-effort daily cron), so an
    # unfiltered count would over-report "特权角色授予 N 人次" in a signed
    # ISMS report with people who have already left.
    rows = (await db.execute(
        select(AuthRole.name, func.count(func.distinct(AuthUserRole.user_id)))
        .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id)
        .join(IdentityPerson, IdentityPerson.id == AuthUserRole.user_id)
        .where(
            IdentityPerson.employment_status == "active",
            # …and not a tombstone: soft-deleted rows keep
            # employment_status='active', so the status filter alone
            # still let purged duplicates into the signed report.
            IdentityPerson.deleted_at.is_(None),
        )
        .group_by(AuthRole.name)
        .order_by(func.count(func.distinct(AuthUserRole.user_id)).desc())
    )).all()
    privileged = {
        "global_admin", "auditor", "cloud_admin", "asset_admin",
        "asset_global_admin", "hr", "finance", "management",
    }
    table: list[dict[str, Any]] = []
    admin_total = 0
    for name, n in rows:
        n = int(n)
        if name in privileged:
            admin_total += n
        table.append({"level": name, "count": n, "note": (
            "特权角色" if name in privileged else "")})
    return {
        "table": table,
        "summary": f"特权角色授予合计 {admin_total} 人次,基于角色的最小权限分配",
        "numbers": {"privileged_grants": admin_total},
    }


async def _agg_mfa_coverage(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.auditor.models import AuditorMfaStatus

    # Denominator = identities whose provider actually reported an
    # enrollment state. An Atlassian row with an unreadable org 2FA policy
    # is not an un-enrolled account, and this table is ISMS *evidence* —
    # shipping "MFA 覆盖率 0.0% (0/533)" into an assessment pack because a
    # provider could not answer is the most expensive version of this bug.
    _observed = AuditorMfaStatus.enrollment_observed.is_(True)
    total = await _count(
        db,
        select(func.count(func.distinct(AuditorMfaStatus.person_id)))
        .where(_observed),
    )
    enrolled = await _count(
        db,
        select(func.count(func.distinct(AuditorMfaStatus.person_id)))
        .where(_observed, AuditorMfaStatus.is_enrolled.is_(True)),
    )
    pct = round(enrolled / total * 100, 1) if total else 0.0
    table = [
        {"metric": "纳入 MFA 审计的账号", "value": total},
        {"metric": "已启用 MFA", "value": enrolled},
        {"metric": "覆盖率", "value": f"{pct}%"},
    ]
    return {
        "table": table,
        "summary": (
            f"MFA 覆盖率 {pct}%（{enrolled}/{total}）" if total
            else "尚未同步 MFA 数据"
        ),
        "numbers": {"total": total, "enrolled": enrolled, "pct": pct},
    }


async def _agg_access_review(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.auditor.models import AccessCertification

    latest = (await db.execute(
        select(AccessCertification)
        .order_by(AccessCertification.started_at.desc())
        .limit(1)
    )).scalar_one_or_none()
    if latest is None:
        return {
            "table": [{"metric": "访问复核活动", "value": "尚未开展"}],
            "summary": "尚未开展访问复核活动",
            "numbers": {},
        }
    table = [
        {"metric": "最近复核活动", "value": latest.label},
        {"metric": "状态", "value": latest.status},
        {"metric": "复核项", "value": latest.snapshot_item_count},
        {"metric": "已决策", "value": latest.decided_count},
        {"metric": "已回收", "value": latest.revoked_count},
    ]
    return {
        "table": table,
        "summary": f"最近复核活动「{latest.label}」状态 {latest.status}",
        "numbers": {
            "items": latest.snapshot_item_count,
            "decided": latest.decided_count,
            "revoked": latest.revoked_count,
        },
    }


async def _agg_exposure_findings(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.auditor.models import AuditorPublicExposureFinding

    open_ct = await _count(
        db,
        select(func.count()).select_from(AuditorPublicExposureFinding)
        .where(AuditorPublicExposureFinding.status == "open"),
    )
    table = [{"metric": "当前公网暴露未处置项", "value": open_ct}]
    return {
        "table": table,
        "summary": (
            "当前无公网暴露未处置项" if open_ct == 0
            else f"当前 {open_ct} 项公网暴露未处置"
        ),
        "numbers": {"open": open_ct},
    }


async def _agg_change_mgmt(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.provisioning.models import (
        STATUS_COMPLETED,
        STATUS_FAILED,
        ProvisioningTask,
    )

    start, end = _window(ps, pe)
    completed = await _count(
        db,
        select(func.count()).select_from(ProvisioningTask)
        .where(ProvisioningTask.created_at >= start,
               ProvisioningTask.created_at <= end,
               ProvisioningTask.status == STATUS_COMPLETED),
    )
    failed = await _count(
        db,
        select(func.count()).select_from(ProvisioningTask)
        .where(ProvisioningTask.created_at >= start,
               ProvisioningTask.created_at <= end,
               ProvisioningTask.status == STATUS_FAILED),
    )
    closed = completed + failed
    rate = f"{round(completed / closed * 100)}%" if closed else "—"
    table = [{
        "change_type": "权限/账号配置变更(自动化)",
        "count": closed,
        "success_rate": rate,
    }]
    return {
        "table": table,
        "summary": f"本期自动化变更 {closed} 项,成功率 {rate}",
        "numbers": {"completed": completed, "failed": failed},
    }


async def _agg_secret_rotation(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.auditor.models import AuditorSecretRotationRecord

    total = await _count(
        db, select(func.count()).select_from(AuditorSecretRotationRecord))
    overdue = await _count(
        db,
        select(func.count()).select_from(AuditorSecretRotationRecord)
        .where(AuditorSecretRotationRecord.status == "overdue"),
    )
    table = [
        {"metric": "纳管密钥/凭证", "value": total},
        {"metric": "超期未轮换", "value": overdue},
    ]
    return {
        "table": table,
        "summary": (
            f"纳管密钥 {total} 项,超期 {overdue} 项" if total
            else "尚未同步密钥轮换台账"
        ),
        "numbers": {"total": total, "overdue": overdue},
    }


async def _agg_incident_stats(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.compliance.models import ComplianceIncident

    start, end = _window(ps, pe)
    rows = (await db.execute(
        select(ComplianceIncident.severity, func.count())
        .where(ComplianceIncident.created_at >= start,
               ComplianceIncident.created_at <= end)
        .group_by(ComplianceIncident.severity)
    )).all()
    by_sev = {s: int(n) for s, n in rows}
    total = sum(by_sev.values())
    table = [{"metric": "本期安全事件", "value": total}]
    table += [{"metric": f"  {sev}", "value": n} for sev, n in by_sev.items()]
    return {
        "table": table,
        "summary": (
            "本期无登记的安全事件" if total == 0
            else f"本期登记安全事件 {total} 起"
        ),
        "numbers": {"total": total, "by_severity": by_sev},
    }


async def _agg_cloud_security(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope

    scopes = await _count(db, select(func.count()).select_from(CmdbCloudScope))
    active_grants = await _count(
        db,
        select(func.count()).select_from(CmdbCloudAccessGrant)
        .where(CmdbCloudAccessGrant.status == "active"),
    )
    table = [
        {"metric": "纳管云账号/范围", "value": scopes},
        {"metric": "活跃云访问授权", "value": active_grants},
    ]
    return {
        "table": table,
        "summary": f"纳管云范围 {scopes} 个,活跃授权 {active_grants} 项",
        "numbers": {"scopes": scopes, "active_grants": active_grants},
    }


async def _agg_self_check_summary(db: AsyncSession, ps: date, pe: date) -> dict[str, Any]:
    """Ties the two pillars: periodic self-check execution in the period."""
    from app.domains.auditor.models import (
        AuditorSelfCheckCategory,
        AuditorSelfCheckRecord,
    )

    start, end = _window(ps, pe)
    categories = await _count(
        db,
        select(func.count()).select_from(AuditorSelfCheckCategory)
        .where(AuditorSelfCheckCategory.enabled.is_(True)),
    )
    completed = await _count(
        db,
        select(func.count()).select_from(AuditorSelfCheckRecord)
        .where(AuditorSelfCheckRecord.completed_at >= start,
               AuditorSelfCheckRecord.completed_at <= end,
               AuditorSelfCheckRecord.status == "completed"),
    )
    overdue = await _count(
        db,
        select(func.count()).select_from(AuditorSelfCheckRecord)
        .where(AuditorSelfCheckRecord.status == "pending",
               AuditorSelfCheckRecord.due_at < datetime.combine(pe, time.max)),
    )
    table = [
        {"metric": "启用的自检类目", "value": categories},
        {"metric": "本期完成检查", "value": completed},
        {"metric": "当前逾期未检查", "value": overdue},
    ]
    return {
        "table": table,
        "summary": (
            f"本期完成周期自检 {completed} 次,当前逾期 {overdue} 项"
        ),
        "numbers": {
            "categories": categories, "completed": completed, "overdue": overdue,
        },
    }


AGGREGATORS = {
    "asset_inventory": _agg_asset_inventory,
    "device_inventory": _agg_device_inventory,
    "headcount": _agg_headcount,
    "offboarding_flow": _agg_offboarding_flow,
    "rbac_access": _agg_rbac_access,
    "mfa_coverage": _agg_mfa_coverage,
    "access_review": _agg_access_review,
    "exposure_findings": _agg_exposure_findings,
    "change_mgmt": _agg_change_mgmt,
    "secret_rotation": _agg_secret_rotation,
    "incident_stats": _agg_incident_stats,
    "cloud_security": _agg_cloud_security,
    "self_check_summary": _agg_self_check_summary,
}


async def aggregate(
    db: AsyncSession, source: str | None, ps: date, pe: date,
) -> dict[str, Any] | None:
    """Run the aggregator for ``source`` over the period; None if no source."""
    if not source:
        return None
    fn = AGGREGATORS.get(source)
    if fn is None:
        return None
    return await fn(db, ps, pe)