"""Monthly compliance-posture summary — the one place a model narrates.

Everything else in this feature is deterministic on purpose. This is the
exception, and it is deliberately narrow: the numbers are computed here,
frozen into a snapshot, and the model is handed **counts and anonymous
references only** so it can turn them into two paragraphs an auditor can
read on a Monday morning. It cannot change a number, name a person, or
introduce a fact — the platform's existing grounding, anonymous-reference
and low-confidence guards all still apply.

What the month's facts are:

* control pass/fail counts from the GRC catalogue;
* evidence coverage — how many company-wide questions actually got
  pre-compiled for the closed month, and which controls that covers;
* open access-drift findings by severity;
* the decision backlog: findings waiting on a human.

Why a snapshot rather than a fresh computation each time: the same
month must narrate identically on re-read. The snapshot's facts hash is
what makes "regenerate" a no-op when nothing moved, so re-opening last
month's summary costs nothing and says the same thing.
"""
from __future__ import annotations

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

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

from app.intelligence.privacy import EvidenceObject, build_reference_map

POSTURE_SUBJECT_TYPE = "compliance_posture"
POSTURE_SCORE_VERSION = "compliance-posture-v1"
POSTURE_ANALYSIS_KIND = "compliance_posture_monthly"
POSTURE_AUDIENCE_KEY = "auditor_shared"


async def build_posture_facts(
    db: AsyncSession, *, period_start: str, period_end: str,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    """Compute the month's facts. Returns ``(health, facts, reference_map)``.

    Every value is a count, a ratio or an anonymous reference. Nothing
    here is free text, a name, or an identifier — the privacy gate would
    refuse the package, and that refusal is the design working rather
    than an obstacle to route around.
    """
    from app.domains.auditor import compliance_catalog as cat  # noqa: PLC0415
    from app.domains.auditor.access_drift import CONTROL_KEY as DRIFT_CONTROL
    from app.domains.auditor.models import (
        AuditorEvidenceBundle,
        AuditorFinding,
        ComplianceControl,
        ComplianceControlStatus,
    )

    # LEFT OUTER JOIN, and the denominator is the enabled catalogue —
    # not the set of controls that happen to have a status row. An inner
    # join dropped never-evaluated controls from BOTH sides, so ten
    # enabled controls of which only two had ever run reported a 100%
    # pass ratio and grade A. "We have not looked" is not "we passed".
    control_rows = (await db.execute(
        select(ComplianceControl.key, ComplianceControlStatus.status)
        .outerjoin(
            ComplianceControlStatus,
            ComplianceControlStatus.control_id == ComplianceControl.id,
        )
        .where(ComplianceControl.enabled.is_(True))
    )).all()
    by_status: dict[str, int] = {}
    for _key, status in control_rows:
        label = str(status) if status else "not_evaluated"
        by_status[label] = by_status.get(label, 0) + 1
    total_controls = len(control_rows)
    passing = by_status.get(cat.STATUS_PASS, 0)

    # Scoped to THIS month's window. Counting every scheduled bundle ever
    # produced would report full coverage forever after the first
    # successful month — including a month whose pre-compile failed
    # outright, which is precisely the false assurance this summary
    # exists to prevent.
    window_start = datetime.fromisoformat(period_start).replace(tzinfo=UTC)
    window_end = datetime.fromisoformat(period_end).replace(
        hour=23, minute=59, second=59, tzinfo=UTC,
    )
    bundles = (await db.execute(
        select(AuditorEvidenceBundle.query_key)
        .where(
            AuditorEvidenceBundle.generated_via == "scheduled",
            AuditorEvidenceBundle.period_start.is_not(None),
            AuditorEvidenceBundle.period_start >= window_start,
            AuditorEvidenceBundle.period_start <= window_end,
        )
    )).all()
    precompiled_queries = {str(key) for (key,) in bundles}

    findings = (await db.execute(
        select(AuditorFinding.severity, func.count())
        .where(
            AuditorFinding.control_key == DRIFT_CONTROL,
            AuditorFinding.status == "open",
        )
        .group_by(AuditorFinding.severity)
    )).all()
    open_by_severity = {str(sev): int(count) for sev, count in findings}

    awaiting = int((await db.execute(
        select(func.count())
        .select_from(AuditorFinding)
        .where(AuditorFinding.status == "open")
    )).scalar_one())

    # Anonymous references stand in for the controls that are failing, so
    # the narrative can point at "three of these" without the model ever
    # seeing a control key. Sorted before numbering so the same failing
    # set always produces the same package — and therefore the same
    # facts hash, and therefore no needless regeneration.
    # Same population as the counts above — enabled controls only, and
    # never-evaluated ones included. Drawing the refs from a different
    # set than the denominator let disabled controls drive a narrative
    # whose own numbers never counted them.
    # Only genuinely failing controls. ``status is None`` means the
    # control has never been evaluated — filing that as "failing" would
    # have the fact package tell the model eight controls are failing
    # while ``controls_by_status`` in the same dict calls them
    # ``not_evaluated``, and would contradict the answer-freshness rule
    # that "nobody has looked" is not "the claim is wrong".
    failing_keys = sorted(
        str(key) for key, status in control_rows if status == cat.STATUS_FAIL
    )
    refs, reference_map = build_reference_map(
        [EvidenceObject(kind="control", object_id=key) for key in failing_keys]
    )

    facts: dict[str, Any] = {
        "period_start": period_start,
        "period_end": period_end,
        "controls_total": total_controls,
        "controls_passing": passing,
        "controls_by_status": by_status,
        "control_pass_ratio": (
            round(passing / total_controls, 4) if total_controls else 0
        ),
        # The COUNT, not the list. Facts are scalars a narrative cites by
        # pointer; the anonymous refs themselves travel on the separate
        # ``reference_map`` channel, which is what ``allowed_evidence_refs``
        # is built from and where the model is meant to use them. Exposing
        # the list here invited the model to cite it as if it were a
        # scalar, and the contract refused the citation — correctly, but
        # for a temptation this package should never have offered.
        "failing_control_count": len(refs),
        "evidence_queries_precompiled": len(precompiled_queries),
        "open_drift_findings": sum(open_by_severity.values()),
        "open_drift_by_severity": open_by_severity,
        "findings_awaiting_decision": awaiting,
    }

    # Confidence is deterministic, not a model opinion: a month with no
    # controls or no pre-compiled evidence cannot support a confident
    # narrative, and the platform's low-confidence rule then forces the
    # model to name the gap instead of writing around it.
    confidence = 90
    if not total_controls:
        confidence = 40
    elif not precompiled_queries:
        confidence = 60
    # The scored view belongs in the FACT package too, not only in
    # ``health``. The model is shown the score and naturally cites it as
    # ``/facts/overall_score`` — a pointer that resolved nowhere, so the
    # grounding check refused the whole narrative with "Unknown fact
    # reference" and the monthly summary could never be produced. These
    # are derived from ``control_pass_ratio``, which is already a fact;
    # publishing the derivation makes it citable instead of inventable.
    facts["overall_score"] = int(round(facts["control_pass_ratio"] * 100))
    # The SAME ratio in the form a narrator actually writes. The guard
    # compares literal number tokens and understands no arithmetic, so a
    # model handed 0.4545 and writing the obvious "45.45%" is refused for
    # an "uncited number" it in fact derived correctly.
    facts["control_pass_percent"] = round(facts["control_pass_ratio"] * 100, 2)
    # Both sides of the ratio. A posture letter naturally reports what is
    # failing, not only what passes — production showed the model writing
    # the complements (54.55%, 6) which no fact carried, so the guard
    # refused figures that were arithmetically right and editorially
    # obvious. "How many controls are failing" is a fact an auditor wants
    # in its own right, so it is computed here rather than left to be
    # derived.
    failing = sum(
        n for st, n in by_status.items() if st == cat.STATUS_FAIL
    )
    facts["controls_failing"] = failing
    facts["control_fail_percent"] = (
        round(failing / total_controls * 100, 2) if total_controls else 0
    )
    facts["controls_not_evaluated"] = by_status.get("not_evaluated", 0)

    # Flatten the two status maps into scalars. A fact package is a set of
    # citable scalars: the model cites by JSON Pointer, and the contract
    # refuses a pointer that lands on a dict or a list. Leaving nested
    # maps in here means the model can pick one and have its whole
    # narrative refused for a shape this package chose to offer — which
    # is exactly what happened with the refs list. Prevent the class, not
    # the instance.
    for status, count in by_status.items():
        facts[f"controls_status_{status}"] = count
    for severity, count in open_by_severity.items():
        facts[f"open_drift_{severity}"] = count
    facts.pop("controls_by_status", None)
    facts.pop("open_drift_by_severity", None)
    facts["confidence_score"] = confidence
    facts["grade"] = _grade(facts["control_pass_ratio"])

    # Dimensions built once, then published into the fact package under
    # their own keys. The model is shown this breakdown and cites it the
    # way it is labelled — ``/facts/decision_backlog`` — so a dimension
    # that exists only inside ``health`` makes the grounding guard refuse
    # the whole narrative with "Unknown fact reference". Fixing these one
    # pointer at a time is whack-a-mole; anything the model can SEE has
    # to be citable.
    dimensions = [
        {
            "key": "control_coverage",
            "score": facts["overall_score"],
            "available": bool(total_controls),
        },
        {
            "key": "evidence_coverage",
            "score": min(100, facts["evidence_queries_precompiled"] * 12),
            "available": bool(precompiled_queries),
        },
        {
            # Inverted on purpose: fewer open drift findings is a better
            # posture, so the raw count would score backwards.
            "key": "access_drift",
            "score": max(0, 100 - facts["open_drift_findings"] * 5),
            "available": True,
        },
        {
            "key": "decision_backlog",
            "score": max(0, 100 - facts["findings_awaiting_decision"] * 5),
            "available": True,
        },
    ]
    for dim in dimensions:
        facts[dim["key"]] = dim["score"]

    health = {
        # ``persist_compiled_snapshot`` cross-checks this against the
        # version it is told to persist under, so a compiled package can
        # never be filed against the wrong scoring generation.
        "score_version": POSTURE_SCORE_VERSION,
        "overall_score": facts["overall_score"],
        "confidence_score": facts["confidence_score"],
        "grade": facts["grade"],
        # The same list the facts were published from, so a dimension
        # score can never differ between what the model may cite and
        # what the read layer renders.
        "dimensions": dimensions,
    }
    return health, facts, reference_map


def _grade(ratio: float) -> str:
    if ratio >= 0.95:
        return "A"
    if ratio >= 0.85:
        return "B"
    if ratio >= 0.70:
        return "C"
    return "D"


async def build_posture_snapshot(
    db: AsyncSession, *, period_start: str, period_end: str,
) -> Any:
    """Freeze the month's facts as an immutable snapshot."""
    from app.intelligence.snapshot_service import (
        SnapshotSubject,
        persist_compiled_snapshot,
    )

    health, facts, reference_map = await build_posture_facts(
        db, period_start=period_start, period_end=period_end,
    )
    return await persist_compiled_snapshot(
        db,
        subject=SnapshotSubject(
            POSTURE_SUBJECT_TYPE, period_start[:7], POSTURE_AUDIENCE_KEY,
        ),
        score_version=POSTURE_SCORE_VERSION,
        health=health,
        aggregate_facts=facts,
        reference_map=reference_map,
        data_as_of=datetime.now(UTC),
    )


async def notify_posture_summary(
    db: AsyncSession, *, analysis_id: uuid.UUID, period_label: str,
) -> int:
    """Queue the summary to auditors and global admins. Never employees.

    Writes pending Outbox rows and leaves delivery to the existing retry
    worker — the same path every other platform notification uses, so
    this inherits its receipt-before-deliver ordering rather than
    inventing a second one.
    """
    from app.auth.models import AuthRole, AuthUserRole
    from app.identity.person_models import IdentityPerson
    from app.notifications.service import (
        NotificationEnvelope,
        NotificationService,
    )

    recipients = (await db.execute(
        select(IdentityPerson)
        .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(
            AuthRole.name.in_(("auditor", "global_admin")),
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status == "active",
        )
        .distinct()
    )).scalars().all()

    service = NotificationService(db)
    queued = 0
    for person in recipients:
        await service.dispatch(NotificationEnvelope(
            recipient=person,
            title=f"合规态势月报 {period_label}",
            body_text=(
                "本月控制项通过率、新增访问漂移与待决策事项已汇总。"
                "点开查看结论、数据缺口与证据下钻。"
            ),
            url=f"/auditor/evidence-query?posture={analysis_id}",
            related_resource_type="intelligence_analysis",
            related_resource_id=analysis_id,
        ))
        queued += 1
    return queued


__all__ = [
    "POSTURE_ANALYSIS_KIND",
    "POSTURE_AUDIENCE_KEY",
    "POSTURE_SCORE_VERSION",
    "POSTURE_SUBJECT_TYPE",
    "build_posture_facts",
    "build_posture_snapshot",
    "notify_posture_summary",
]