"""部门擂台 — security-quiz department standings (team battle).

Aggregates the daily-quiz answers by the answerer's PRIMARY department over a
rolling window, ranking departments by participation (how many of the team
played) then accuracy. The live read powers the leaderboard; a weekly cron
freezes the standings into ``secquiz_team_snapshots`` so the page can show
"last week's winner" and a week-over-week trend without recomputing history.

Department is resolved from ``identity_org_memberships`` (primary + active) —
there is no denormalised column on the person row. People with no primary
membership fall into a single "未分配" bucket so the totals reconcile.
"""
from __future__ import annotations

from datetime import date, datetime, timedelta
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy import and_, case, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.security_quiz.models import SecQuizAnswer, SecQuizTeamSnapshot
from app.identity.person_models import IdentityOrgMembership, IdentityPerson

_HQ_TZ = ZoneInfo("Asia/Shanghai")
_UNASSIGNED = "—"


def _hq_today() -> date:
    return datetime.now(_HQ_TZ).date()


def _rank_key(s: dict[str, Any]) -> tuple[float, float, int]:
    """Single source of truth for 部门擂台 ordering — participation first
    (engagement is the point), then accuracy, then raw correct as a stable
    tiebreaker. Operates on the ROUNDED display pcts so the live board and the
    frozen weekly snapshot never disagree on order for a rounding tie."""
    return (-s["participation_pct"], -s["accuracy_pct"], -s["correct"])


def _primary_dept_subquery():
    """One primary, active org-node name per person. ``MAX`` collapses the rare
    multi-primary row to a stable single value."""
    return (
        select(
            IdentityOrgMembership.person_id.label("pid"),
            func.max(IdentityOrgMembership.org_node_name).label("dept"),
        )
        .where(
            IdentityOrgMembership.is_primary.is_(True),
            IdentityOrgMembership.membership_status == "active",
        )
        .group_by(IdentityOrgMembership.person_id)
        .subquery()
    )


async def compute_department_standings(
    db: AsyncSession, *, window_days: int = 30,
) -> dict[str, Any]:
    """Live department standings over the trailing ``window_days``.

    Ranked by participation_pct (share of the department that played at least
    once in the window) then accuracy_pct. Departments with zero active
    headcount are omitted; an "未分配" bucket collects membership-less people.
    """
    today = _hq_today()
    since = today - timedelta(days=max(1, window_days))
    dept = _primary_dept_subquery()

    # Active headcount per department (the denominator for participation).
    head_rows = (await db.execute(
        select(
            func.coalesce(dept.c.dept, _UNASSIGNED).label("dept"),
            func.count(func.distinct(IdentityPerson.id)).label("headcount"),
        )
        .select_from(IdentityPerson)
        .join(dept, dept.c.pid == IdentityPerson.id, isouter=True)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
        .group_by(func.coalesce(dept.c.dept, _UNASSIGNED))
    )).all()
    headcount_by_dept = {d: int(h or 0) for d, h in head_rows}

    # Activity per department within the window.
    correct_expr = func.sum(case((SecQuizAnswer.is_correct.is_(True), 1), else_=0))
    act_rows = (await db.execute(
        select(
            func.coalesce(dept.c.dept, _UNASSIGNED).label("dept"),
            func.count(func.distinct(SecQuizAnswer.person_id)).label("participants"),
            func.count().label("answered"),
            correct_expr.label("correct"),
        )
        .select_from(SecQuizAnswer)
        .join(IdentityPerson, and_(
            IdentityPerson.id == SecQuizAnswer.person_id,
            IdentityPerson.deleted_at.is_(None),
            # Same population as the headcount denominator above — without
            # ``!= 'left'`` a departed employee's answers inflate a dept's
            # participants / answered / correct, pushing participation_pct
            # (min(participants, headcount)/headcount) and accuracy above the
            # true active-staff rate, and that skew is frozen into the weekly
            # snapshot.
            IdentityPerson.employment_status != "left",
        ))
        .join(dept, dept.c.pid == SecQuizAnswer.person_id, isouter=True)
        .where(SecQuizAnswer.quiz_date >= since)
        .group_by(func.coalesce(dept.c.dept, _UNASSIGNED))
    )).all()
    activity_by_dept = {
        d: {"participants": int(p or 0), "answered": int(a or 0), "correct": int(c or 0)}
        for d, p, a, c in act_rows
    }

    standings: list[dict[str, Any]] = []
    for d, headcount in headcount_by_dept.items():
        if headcount <= 0:
            continue
        act = activity_by_dept.get(d, {"participants": 0, "answered": 0, "correct": 0})
        participants = min(act["participants"], headcount)
        answered = act["answered"]
        correct = act["correct"]
        standings.append({
            "department": d,
            "headcount": headcount,
            "participants": participants,
            "participation_pct": round(100.0 * participants / headcount, 1),
            "answered": answered,
            "correct": correct,
            "accuracy_pct": round(100.0 * correct / answered, 1) if answered else 0.0,
        })

    standings.sort(key=_rank_key)
    for i, s in enumerate(standings):
        s["rank"] = i + 1

    return {
        "window_days": window_days,
        "since": since.isoformat(),
        "generated_at": datetime.now(_HQ_TZ).isoformat(),
        "standings": standings,
    }


def _week_start(d: date) -> date:
    """Monday of the ISO week containing ``d``."""
    return d - timedelta(days=d.weekday())


async def snapshot_last_week(db: AsyncSession) -> dict[str, Any]:
    """Freeze the just-finished ISO week's standings into
    ``secquiz_team_snapshots``. Idempotent per (week_start, department) via the
    unique key; safe to re-run. Returns a summary for the cron result."""
    today = _hq_today()
    this_monday = _week_start(today)
    last_monday = this_monday - timedelta(days=7)
    # Standings over exactly last week's 7 days [last_monday, this_monday).
    rows = await _standings_for_window(db, last_monday, this_monday)
    written = 0
    for s in rows:
        existing = (await db.execute(
            select(SecQuizTeamSnapshot).where(
                SecQuizTeamSnapshot.week_start == last_monday,
                SecQuizTeamSnapshot.department == s["department"],
            )
        )).scalar_one_or_none()
        if existing is None:
            # SAVEPOINT-isolate the INSERT: a concurrent run (or scheduler
            # retry) racing on the same (week, dept) would otherwise trip the
            # unique key and poison the whole session flush. On collision we
            # roll back just this insert and fall through to an update.
            try:
                async with db.begin_nested():
                    db.add(SecQuizTeamSnapshot(
                        week_start=last_monday,
                        department=s["department"],
                        headcount=s["headcount"],
                        participants=s["participants"],
                        answered=s["answered"],
                        correct=s["correct"],
                        rank=s["rank"],
                    ))
                written += 1
                continue
            except IntegrityError:
                existing = (await db.execute(
                    select(SecQuizTeamSnapshot).where(
                        SecQuizTeamSnapshot.week_start == last_monday,
                        SecQuizTeamSnapshot.department == s["department"],
                    )
                )).scalar_one_or_none()
        if existing is not None:
            existing.headcount = s["headcount"]
            existing.participants = s["participants"]
            existing.answered = s["answered"]
            existing.correct = s["correct"]
            existing.rank = s["rank"]
    await db.flush()
    return {
        "week_start": last_monday.isoformat(),
        "departments": len(rows),
        "written": written,
        "winner": rows[0]["department"] if rows else None,
    }


async def _standings_for_window(
    db: AsyncSession, start: date, end: date,
) -> list[dict[str, Any]]:
    """Standings for the explicit half-open window [start, end). Shared shape
    with the live read but anchored to fixed dates (used by the snapshot)."""
    dept = _primary_dept_subquery()
    head_rows = (await db.execute(
        select(
            func.coalesce(dept.c.dept, _UNASSIGNED).label("dept"),
            func.count(func.distinct(IdentityPerson.id)).label("headcount"),
        )
        .select_from(IdentityPerson)
        .join(dept, dept.c.pid == IdentityPerson.id, isouter=True)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
        .group_by(func.coalesce(dept.c.dept, _UNASSIGNED))
    )).all()
    headcount_by_dept = {d: int(h or 0) for d, h in head_rows}

    correct_expr = func.sum(case((SecQuizAnswer.is_correct.is_(True), 1), else_=0))
    act_rows = (await db.execute(
        select(
            func.coalesce(dept.c.dept, _UNASSIGNED).label("dept"),
            func.count(func.distinct(SecQuizAnswer.person_id)).label("participants"),
            func.count().label("answered"),
            correct_expr.label("correct"),
        )
        .select_from(SecQuizAnswer)
        .join(IdentityPerson, and_(
            IdentityPerson.id == SecQuizAnswer.person_id,
            IdentityPerson.deleted_at.is_(None),
            # Mirror the headcount denominator (active, non-left) so a departed
            # employee's answers don't inflate this window's dept standings —
            # same fix as compute_department_standings above.
            IdentityPerson.employment_status != "left",
        ))
        .join(dept, dept.c.pid == SecQuizAnswer.person_id, isouter=True)
        .where(SecQuizAnswer.quiz_date >= start, SecQuizAnswer.quiz_date < end)
        .group_by(func.coalesce(dept.c.dept, _UNASSIGNED))
    )).all()
    activity_by_dept = {
        d: {"participants": int(p or 0), "answered": int(a or 0), "correct": int(c or 0)}
        for d, p, a, c in act_rows
    }
    out: list[dict[str, Any]] = []
    for d, headcount in headcount_by_dept.items():
        if headcount <= 0:
            continue
        act = activity_by_dept.get(d, {"participants": 0, "answered": 0, "correct": 0})
        participants = min(act["participants"], headcount)
        answered = act["answered"]
        correct = act["correct"]
        out.append({
            "department": d,
            "headcount": headcount,
            "participants": participants,
            "answered": answered,
            "correct": correct,
            # Rounded display pcts so the snapshot orders identically to the
            # live board via the shared ``_rank_key``.
            "participation_pct": round(100.0 * participants / headcount, 1),
            "accuracy_pct": round(100.0 * correct / answered, 1) if answered else 0.0,
        })
    out.sort(key=_rank_key)
    for i, s in enumerate(out):
        s["rank"] = i + 1
    return out


async def latest_snapshot(db: AsyncSession) -> dict[str, Any] | None:
    """The most recently frozen week's standings, or None if never snapshotted."""
    week = (await db.execute(
        select(func.max(SecQuizTeamSnapshot.week_start))
    )).scalar()
    if week is None:
        return None
    rows = (await db.execute(
        select(SecQuizTeamSnapshot)
        .where(SecQuizTeamSnapshot.week_start == week)
        .order_by(SecQuizTeamSnapshot.rank)
    )).scalars().all()
    return {
        "week_start": week.isoformat(),
        "standings": [
            {
                "rank": r.rank,
                "department": r.department,
                "headcount": r.headcount,
                "participants": r.participants,
                "answered": r.answered,
                "correct": r.correct,
            }
            for r in rows
        ],
    }


__all__ = [
    "compute_department_standings",
    "snapshot_last_week",
    "latest_snapshot",
]