"""Coverage — the number this whole domain exists to be able to state.

Everything upstream (a hosted library, a per-person target ledger, an
append-only exposure log) was built so that one question has an answer:
**what fraction of the workforce confirmed the security material we
required them to read, and can we show the per-person evidence?**

## What counts, and what deliberately does not

* **Required campaigns only.** A voluntary notice asks nothing of
  anybody; letting a comic nobody had to read carry the control would
  make the number meaningless in the one direction that matters.
* **``completion_mode='ack'`` only.** An "opened it" campaign is a real
  thing to run, but it is not the claim an auditor is asking about.
* **The denominator is who we ACTUALLY sent it to**, from the target
  ledger — not "everybody employed today". A campaign sent in March to
  the 480 people who worked here then is a March fact; recomputing its
  denominator against today's headcount would silently restate it every
  time somebody joins.
* **Nothing is rounded up.** A closed campaign keeps its unacknowledged
  rows. 87% is 87%.

## Why status is derived from the LATEST campaign, not from all of them

Averaging every campaign ever run rewards volume: five easy notices
would drown one badly-answered mandatory. The control asks "are we
currently doing this well", so it reads the most recent required
campaign and separately checks that one is recent enough to count.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.domains.security_awareness.catalog import (
    CAMPAIGN_CLOSED,
    CAMPAIGN_SENT,
    COMPLETION_ACK,
    STATUS_PUBLISHED,
    SURFACE_BUBBLE,
    SURFACE_DETAIL,
)
from app.domains.security_awareness.models import (
    SecawCampaign,
    SecawCampaignTarget,
    SecawMaterial,
)

logger = logging.getLogger(__name__)

#: Statuses whose numbers count. A draft / scheduled / cancelled
#: campaign reached nobody, and ``sending`` is mid-flight.
COUNTED_STATUSES: tuple[str, ...] = (CAMPAIGN_SENT, CAMPAIGN_CLOSED)

DEFAULT_MIN_COVERAGE_PCT = 90
DEFAULT_STALE_AFTER_DAYS = 200


def _count_map(rows: Any) -> dict[uuid.UUID, int]:
    """``[(material_id, n), ...]`` → ``{material_id: n}``.

    ``dict(rows)`` does the same at runtime, but a SQLAlchemy ``Row`` is
    not a typed 2-tuple as far as mypy is concerned. Unpacking makes the
    shape explicit at the one place it is asserted.
    """
    return {mid: int(n) for mid, n in rows}


async def _latest_required_campaign(db: AsyncSession) -> SecawCampaign | None:
    return (await db.execute(
        select(SecawCampaign)
        .where(
            SecawCampaign.mandatory.is_(True),
            SecawCampaign.completion_mode == COMPLETION_ACK,
            SecawCampaign.status.in_(COUNTED_STATUSES),
            SecawCampaign.sent_at.is_not(None),
        )
        .order_by(SecawCampaign.sent_at.desc())
        .limit(1)
    )).scalar_one_or_none()


async def campaign_coverage(
    db: AsyncSession, campaign_id: uuid.UUID,
) -> dict[str, Any]:
    """``{targeted, acknowledged, opened, coverage_pct}`` for one campaign."""
    row = (await db.execute(
        select(
            func.count(),
            func.count(SecawCampaignTarget.opened_at),
            func.count(SecawCampaignTarget.acknowledged_at),
        ).where(SecawCampaignTarget.campaign_id == campaign_id)
    )).one()
    targeted, opened, acked = int(row[0]), int(row[1]), int(row[2])
    return {
        "targeted": targeted,
        "opened": opened,
        "acknowledged": acked,
        # Integer percent, floored — a coverage report that rounds 89.6
        # up to 90 against a 90 threshold is the one rounding nobody
        # wants to discover afterwards.
        "coverage_pct": int(acked * 100 // targeted) if targeted else 0,
    }


async def evaluate_awareness_control(db: AsyncSession) -> dict[str, Any]:
    """Compute the ``security_awareness_training`` GRC control status.

    Returns the shape ``compliance_service.recompute_control_status``
    expects: ``{status, score, summary, detail}``.
    """
    from app.domains.auditor import compliance_catalog as cat

    spec = next(
        (c for c in cat.CONTROLS if c["key"] == "security_awareness_training"),
        {},
    )
    config = spec.get("config") or {}
    min_pct = int(config.get("min_coverage_pct", DEFAULT_MIN_COVERAGE_PCT))
    stale_days = int(config.get("stale_after_days", DEFAULT_STALE_AFTER_DAYS))

    campaign = await _latest_required_campaign(db)
    if campaign is None:
        return {
            "status": cat.STATUS_NOT_CONFIGURED,
            "score": None,
            "summary": "还没有跑过任何必修安全宣贯",
            "detail": {},
        }

    stats = await campaign_coverage(db, campaign.id)
    material = await db.get(SecawMaterial, campaign.material_id)
    detail = {
        "campaign_id": str(campaign.id),
        "campaign_name": campaign.name,
        "sent_at": campaign.sent_at.isoformat() if campaign.sent_at else None,
        "material_version": campaign.material_version,
        "material_kind": material.kind if material else None,
        "min_coverage_pct": min_pct,
        **stats,
    }

    sent_at = campaign.sent_at
    if sent_at is not None:
        # Stored DATETIMEs come back naive; compare in the same space
        # rather than crashing on an offset-naive/aware subtraction.
        now = datetime.now(UTC) if sent_at.tzinfo else datetime.now(UTC).replace(
            tzinfo=None,
        )
        if now - sent_at > timedelta(days=stale_days):
            return {
                "status": cat.STATUS_STALE,
                "score": stats["coverage_pct"],
                "summary": (
                    f"最近一次必修宣贯「{campaign.name}」已经是 {stale_days} "
                    "天前的事了"
                ),
                "detail": detail,
            }

    if stats["targeted"] == 0:
        # Sent to nobody. Not a pass: a campaign with an empty audience
        # evidences nothing, and reporting 0/0 as 100% would be the
        # single most misleading thing this module could do.
        return {
            "status": cat.STATUS_FAIL,
            "score": 0,
            "summary": f"「{campaign.name}」一个人都没发到",
            "detail": detail,
        }

    if stats["coverage_pct"] >= min_pct:
        return {
            "status": cat.STATUS_PASS,
            "score": stats["coverage_pct"],
            "summary": (
                f"「{campaign.name}」确认率 {stats['coverage_pct']}%"
                f"({stats['acknowledged']}/{stats['targeted']})"
            ),
            "detail": detail,
        }
    return {
        "status": cat.STATUS_FAIL,
        "score": stats["coverage_pct"],
        "summary": (
            f"「{campaign.name}」确认率 {stats['coverage_pct']}%,低于 "
            f"{min_pct}% 的要求({stats['acknowledged']}/{stats['targeted']})"
        ),
        "detail": detail,
    }


async def coverage_report(
    db: AsyncSession, *, window_days: int = 365,
) -> dict[str, Any]:
    """Every counted required campaign in the window, plus a roll-up.

    The roll-up is **people-weighted** (total acknowledged / total
    targeted), not an average of per-campaign percentages: averaging
    percentages lets a campaign sent to 3 people move the number as much
    as one sent to 480, which is exactly backwards.
    """
    since = datetime.now(UTC) - timedelta(days=max(1, window_days))
    rows = (await db.execute(
        select(SecawCampaign)
        .where(
            SecawCampaign.mandatory.is_(True),
            SecawCampaign.completion_mode == COMPLETION_ACK,
            SecawCampaign.status.in_(COUNTED_STATUSES),
            SecawCampaign.sent_at.is_not(None),
        )
        .order_by(SecawCampaign.sent_at.desc())
    )).scalars().all()
    # Filter in Python: ``sent_at`` comes back naive from MySQL/TiDB and
    # a tz-aware bound in the WHERE would compare wrong rather than
    # loudly.
    naive_since = since.replace(tzinfo=None)
    in_window = [
        c for c in rows
        if c.sent_at is not None
        and (c.sent_at if c.sent_at.tzinfo is None else c.sent_at.replace(
            tzinfo=None,
        )) >= naive_since
    ]

    campaigns: list[dict[str, Any]] = []
    total_targeted = 0
    total_acked = 0
    for c in in_window:
        stats = await campaign_coverage(db, c.id)
        total_targeted += stats["targeted"]
        total_acked += stats["acknowledged"]
        campaigns.append({
            "campaign_id": str(c.id),
            "name": c.name,
            "status": c.status,
            "sent_at": c.sent_at.isoformat() if c.sent_at else None,
            "due_at": c.due_at.isoformat() if c.due_at else None,
            "material_id": str(c.material_id),
            "material_version": c.material_version,
            **stats,
        })

    return {
        "window_days": window_days,
        "campaigns": campaigns,
        "rollup": {
            "campaigns": len(campaigns),
            "targeted": total_targeted,
            "acknowledged": total_acked,
            # People-weighted — see the docstring.
            "coverage_pct": (
                int(total_acked * 100 // total_targeted)
                if total_targeted else 0
            ),
        },
    }


async def evidence_rows(
    db: AsyncSession, campaign_id: uuid.UUID,
) -> list[dict[str, Any]]:
    """The per-person evidence for one campaign, ready for CSV.

    Reads the SNAPSHOT columns rather than joining live person rows:
    somebody who transferred out last month was reached as a member of
    their old team, and an export that re-derived their department would
    quietly restate a past department's coverage.
    """
    rows = (await db.execute(
        select(SecawCampaignTarget)
        .where(SecawCampaignTarget.campaign_id == campaign_id)
        .order_by(SecawCampaignTarget.created_at.asc())
    )).scalars().all()
    return [
        {
            # Name and email stay separate columns (#3417).
            "name": (r.person_snapshot or {}).get("display_name") or "",
            "email": (r.person_snapshot or {}).get("work_email") or "",
            "employee_no": (r.person_snapshot or {}).get("employee_no") or "",
            "department": (r.person_snapshot or {}).get("department") or "",
            "country": (r.person_snapshot or {}).get("country") or "",
            "language": r.lang,
            "material_version": r.material_version,
            "sent_at": r.queued_at.isoformat() if r.queued_at else "",
            "opened_at": r.opened_at.isoformat() if r.opened_at else "",
            "acknowledged_at": (
                r.acknowledged_at.isoformat() if r.acknowledged_at else ""
            ),
            "acknowledged_via": r.ack_source or "",
            # Every follow-up we sent this person about this campaign —
            # the automatic ladder AND an admin-triggered resend both
            # bump it. "Did we chase them?" is one of the questions this
            # file exists to answer, and counting only the manual pushes
            # would print a 0 next to somebody the cron nudged four
            # times.
            "reminders_sent": r.resend_count,
        }
        for r in rows
    ]


#: CSV column order. Fixed here rather than derived from the dict so a
#: later field addition cannot silently reorder somebody's saved
#: spreadsheet template.
EVIDENCE_COLUMNS: tuple[str, ...] = (
    "name", "email", "employee_no", "department", "country", "language",
    "material_version", "sent_at", "opened_at", "acknowledged_at",
    "acknowledged_via", "reminders_sent",
)


async def participation_report(
    db: AsyncSession, *, window_days: int = 90,
) -> dict[str, Any]:
    """Who is engaging with the library at all — the OTHER half.

    Coverage answers "did the people we required to read confirm". This
    answers "is anybody reading the 95% we did not require", which is the
    thing that decides whether the ambient surface and the library are
    worth running at all. They are deliberately separate numbers: rolling
    voluntary reading into the compliance figure is how a comic nobody
    had to read ends up carrying an audit claim.

    Distinct PEOPLE per material on the ``detail`` surface, not
    impressions — "shown 900 times" and "reached 300 people" answer
    different questions, and only the second one is engagement.
    """
    from app.domains.security_awareness.models import (
        SecawImpression,
        SecawQuizAttempt,
    )

    since = datetime.now(UTC) - timedelta(days=max(1, window_days))
    naive_since = since.replace(tzinfo=None)

    materials = (await db.execute(
        select(SecawMaterial).where(SecawMaterial.status == STATUS_PUBLISHED)
    )).scalars().all()
    if not materials:
        return {"window_days": window_days, "materials": [], "totals": {}}

    ids = [m.id for m in materials]
    readers = _count_map((await db.execute(
        select(
            SecawImpression.material_id,
            func.count(func.distinct(SecawImpression.person_id)),
        )
        .where(
            SecawImpression.material_id.in_(ids),
            SecawImpression.surface == SURFACE_DETAIL,
            SecawImpression.occurred_at >= naive_since,
        )
        .group_by(SecawImpression.material_id)
    )).all())
    shown = _count_map((await db.execute(
        select(
            SecawImpression.material_id,
            func.count(func.distinct(SecawImpression.person_id)),
        )
        .where(
            SecawImpression.material_id.in_(ids),
            SecawImpression.surface == SURFACE_BUBBLE,
            SecawImpression.occurred_at >= naive_since,
        )
        .group_by(SecawImpression.material_id)
    )).all())
    # The open-rate NUMERATOR: people the bubble showed it to who then
    # opened it. NOT ``readers`` — somebody can open a material straight
    # from the library without the bubble ever offering it, so dividing
    # every opener by the bubble-reached would produce rates over 100%
    # (2 shown, 10 openers → "500%") and a progress bar that runs off the
    # end. ``EXISTS`` rather than a join: a person may hold several rows
    # on each surface, and a join would multiply them before the count.
    _opened = aliased(SecawImpression)
    converted = _count_map((await db.execute(
        select(
            SecawImpression.material_id,
            func.count(func.distinct(SecawImpression.person_id)),
        )
        .where(
            SecawImpression.material_id.in_(ids),
            SecawImpression.surface == SURFACE_BUBBLE,
            SecawImpression.occurred_at >= naive_since,
            select(_opened.id).where(
                _opened.material_id == SecawImpression.material_id,
                _opened.person_id == SecawImpression.person_id,
                _opened.surface == SURFACE_DETAIL,
                _opened.occurred_at >= naive_since,
            ).exists(),
        )
        .group_by(SecawImpression.material_id)
    )).all())
    quiz_passed = _count_map((await db.execute(
        select(
            SecawQuizAttempt.material_id,
            func.count(func.distinct(SecawQuizAttempt.person_id)),
        )
        .where(
            SecawQuizAttempt.material_id.in_(ids),
            SecawQuizAttempt.passed.is_(True),
        )
        .group_by(SecawQuizAttempt.material_id)
    )).all())
    quiz_tried = _count_map((await db.execute(
        select(
            SecawQuizAttempt.material_id,
            func.count(func.distinct(SecawQuizAttempt.person_id)),
        )
        .where(SecawQuizAttempt.material_id.in_(ids))
        .group_by(SecawQuizAttempt.material_id)
    )).all())

    rows: list[dict[str, Any]] = []
    for m in materials:
        block = next(
            (b for b in (m.localized or {}).values()
             if isinstance(b, dict) and b.get("title")),
            {},
        )
        bubble_reach = int(shown.get(m.id, 0))
        opened = int(readers.get(m.id, 0))
        opened_after_bubble = int(converted.get(m.id, 0))
        rows.append({
            "material_id": str(m.id),
            "title": block.get("title"),
            "kind": m.kind,
            "category": m.category,
            "bubble_reach": bubble_reach,
            "opened": opened,
            # People who came to it from the ambient surface, i.e. the
            # numerator of the open rate. Exposed as its own field
            # because "10 people opened this, 3 of them from the bubble"
            # is the sentence the rate is short for.
            "opened_after_bubble": opened_after_bubble,
            # Of the people the ambient surface actually showed it to,
            # how many opened it. This is the number that says whether a
            # piece of material is worth its slot — and it is NOT
            # computable from impressions alone, which is why the ledger
            # records distinct people. Bounded by construction: both
            # sides are counted over the same bubble-reached set.
            "open_rate_pct": (
                int(opened_after_bubble * 100 // bubble_reach)
                if bubble_reach else 0
            ),
            "quiz_attempted": int(quiz_tried.get(m.id, 0)),
            "quiz_passed": int(quiz_passed.get(m.id, 0)),
        })
    rows.sort(key=lambda r: -r["opened"])

    return {
        "window_days": window_days,
        "materials": rows,
        "totals": {
            "published": len(rows),
            "opened_total": sum(r["opened"] for r in rows),
            "quiz_passed_total": sum(r["quiz_passed"] for r in rows),
        },
    }