"""Per-person fact bag the badge criteria are evaluated against.

Everything here is **set-based**: each fact is one grouped query over a set of
person ids, returning ``{person_id: value}``. The single-person read path calls
the same functions with a one-element id set (so it stays an indexed point
lookup), and the nightly org-wide scan calls them once for the whole company
instead of running ~18 queries per person × 500 people.

That shape is deliberate. The scan runs inside one transaction; a per-person
query fan-out would hold it open for minutes, and this platform has already
paid for long-running cron transactions twice (an outbox fan-out that FK-locked
``identity_persons`` and browned out login; a 03:00 cron collision that
surfaced as TiDB 1205 on an unrelated table). A fixed set of grouped queries
keeps the scan's transaction short regardless of headcount.

**Person-id column types are not uniform across domains.** ``identity_persons``
ids are hyphenated CHAR(36) GUIDs, but several tables store the reference as a
plain ``String(36)`` (cmdb cloud grants, every feed table, ai_tool_applications)
rather than the GUID type. Comparing those against a ``uuid.UUID`` silently
matches zero rows — that is exactly how ``cloud_explorer`` shipped unearnable
for everyone. ``_str_ids`` / ``_as_uuid`` make the two worlds explicit at every
call site; do not "simplify" them away.

Adding a fact = one query here + its key in ``FACT_KEYS`` + a badge in
``catalog.py``. Facts default to 0 for people with no rows: badge criteria read
a missing key as 0, and a locked badge at 0 progress is the correct rendering
for "never did this".
"""
from __future__ import annotations

import uuid
from collections.abc import Sequence
from datetime import date, datetime
from typing import Any

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

from app.core.tenure import completed_years, observed_anniversary
from app.core.working_hours import resolve_timezone
from app.identity.person_models import IdentityPerson

#: Every fact key the catalog may reference. Kept explicit so a typo in a
#: badge's ``metric`` is caught by ``test_catalog_metrics_are_known_facts``
#: rather than silently pinning that badge's progress at 0 forever.
FACT_KEYS: tuple[str, ...] = (
    "tenure_years",
    "onboarding_ack_complete",
    "longest_streak",
    "total_correct",
    "cloud_grants",
    "devices_ever",
    "devices_returned",
    "inventory_confirmations",
    "arcade_plays",
    "knowledge_correct",
    "requests_submitted",
    "approvals_handled",
    "feed_posts",
    "feed_likes_received",
    "blessings_sent",
    "attestations_done",
    "awareness_materials_read",
    "awareness_quizzes_passed",
    "support_deflections",
    "seats_active",
    "ai_tools_approved",
    "api_keys_held",
)

#: ``asset_operations.operation_type`` values that mean "the holder handed the
#: device back". Mirrors the platform's existing return caliber in
#: dashboard/insights_router.py. Offboarding recovery is deliberately NOT here —
#: hardware clawed back by IT is not the person returning it.
RETURN_OPERATIONS: tuple[str, ...] = ("return", "borrow_return")

#: ``asset_history.action`` values that put a device in someone's hands.
#: Both spellings of transfer are written by different call sites (``transfer``
#: by the self-service handover path, ``transferred`` by the admin path) and the
#: original fact query matched only one of them, undercounting anyone whose
#: device arrived by handover.
ASSIGN_ACTIONS: tuple[str, ...] = (
    "assigned", "transferred", "transfer", "force_transfer",
)


def _str_ids(ids: Sequence[uuid.UUID]) -> list[str]:
    """String(36) form of the id set, for tables that store the reference as a
    plain string rather than the platform GUID type."""
    return [str(i) for i in ids]


def _as_uuid(value: object) -> uuid.UUID | None:
    """Normalise a person-id cell (GUID or String(36)) back to ``UUID`` so
    results from both column types key into the same facts dict."""
    if isinstance(value, uuid.UUID):
        return value
    if isinstance(value, str):
        try:
            return uuid.UUID(value)
        except ValueError:
            return None
    return None


def _tenure_years_asof(hire: date, as_of: date) -> float:
    """满 N 年 (calendar-aware, via ``core.tenure``) plus fractional progress
    toward the next anniversary — so 「满 5 年」 unlocks on the actual 5th
    anniversary, not at ``5 × 365`` days (which lands 1-2 days early across
    leap years and disagrees with every other tenure surface).

    The integer part is ``completed_years``; the fraction is days elapsed
    since the last observed anniversary over the length of the current
    anniversary year, TRUNCATED to two decimals — rounding could lift 4.999
    to 5.0 and unlock a day early, the exact drift this fact exists to kill.
    """
    years = completed_years(hire, as_of)
    last = observed_anniversary(hire, hire.year + years)
    nxt = observed_anniversary(hire, hire.year + years + 1)
    span = (nxt - last).days
    frac = (as_of - last).days / span if span > 0 else 0.0
    return years + int(frac * 100) / 100


def _tenure_years(person: IdentityPerson) -> float | None:
    """Calendar-aware tenure in years for ``person`` (no query).

    Local-calendar rather than UTC because a Tokyo joiner's 周年 must land on
    their own date — the same rule the rest of the platform follows via
    ``resolve_timezone``.
    """
    if person.hire_date is None:
        return None
    today_local: date = datetime.now(resolve_timezone(person)).date()
    if person.hire_date > today_local:
        # A pre-hire is in the nightly scan's population (status != 'left').
        # Treating their tenure as 0 would light 启航 ("your first day on
        # Helix") before they have had one.
        return None
    return _tenure_years_asof(person.hire_date, today_local)


async def _grouped(
    db: AsyncSession, stmt: Select, ids: Sequence[uuid.UUID],
) -> dict[uuid.UUID, int]:
    """Run a ``(person_id, aggregate)`` grouped select into ``{UUID: int}``.

    ``stmt`` must already carry its own WHERE clauses (including the person-id
    restriction) and GROUP BY. Rows whose person id does not parse as a UUID
    are dropped rather than raising — a malformed string in someone else's
    table must not take down the whole badge wall.
    """
    if not ids:
        return {}
    out: dict[uuid.UUID, int] = {}
    for person_id, value in (await db.execute(stmt)).all():
        key = _as_uuid(person_id)
        if key is not None:
            out[key] = int(value or 0)
    return out


async def _quiz_stats(
    db: AsyncSession, ids: Sequence[uuid.UUID],
) -> dict[uuid.UUID, tuple[int, int]]:
    """``{person_id: (total_correct, longest_streak)}`` from secquiz_answers.

    One query for the whole set; the streak is the longest run of consecutive
    calendar dates answered correctly — the same definition as
    ``security_quiz.service.compute_stats``, which stays the authority for that
    domain's own surfaces.
    """
    if not ids:
        return {}
    from app.domains.security_quiz.models import SecQuizAnswer

    rows = (await db.execute(
        select(SecQuizAnswer.person_id, SecQuizAnswer.quiz_date)
        .where(SecQuizAnswer.person_id.in_(ids), SecQuizAnswer.is_correct.is_(True))
        .order_by(SecQuizAnswer.person_id, SecQuizAnswer.quiz_date)
    )).all()

    by_person: dict[uuid.UUID, list[date]] = {}
    for person_id, quiz_date in rows:
        key = _as_uuid(person_id)
        if key is not None:
            by_person.setdefault(key, []).append(quiz_date)

    out: dict[uuid.UUID, tuple[int, int]] = {}
    for person_id, dates in by_person.items():
        # De-duplicate defensively: the (person, date) unique key makes repeats
        # impossible today, but the streak maths must not depend on that.
        ordered = sorted(set(dates))
        longest = 0
        run = 0
        prev: date | None = None
        for d in ordered:
            run = run + 1 if (prev is not None and (d - prev).days == 1) else 1
            longest = max(longest, run)
            prev = d
        out[person_id] = (len(ordered), longest)
    return out


async def _devices_ever(
    db: AsyncSession, ids: Sequence[uuid.UUID],
) -> dict[uuid.UUID, int]:
    """Distinct devices ever held = history-assigned ∪ currently-assigned.

    A set union rather than two counts: anything currently held that also has a
    transfer trail appears in both sources, and counting it twice would hand
    out 资产大亨 for a single laptop.
    """
    from app.domains.asset.models import AssetHistory, AssetItem

    hist = (await db.execute(
        select(AssetHistory.to_user, AssetHistory.asset_id).where(
            AssetHistory.to_user.in_(ids),
            AssetHistory.action.in_(ASSIGN_ACTIONS),
        )
    )).all()
    current = (await db.execute(
        select(AssetItem.assigned_to, AssetItem.id).where(AssetItem.assigned_to.in_(ids))
    )).all()

    ever: dict[uuid.UUID, set[uuid.UUID]] = {}
    for person_id, asset_id in [*hist, *current]:
        key = _as_uuid(person_id)
        if key is not None:
            ever.setdefault(key, set()).add(asset_id)
    return {pid: len(assets) for pid, assets in ever.items()}


async def _onboarding_ack_complete(
    db: AsyncSession, ids: Sequence[uuid.UUID],
) -> dict[uuid.UUID, int]:
    """1 for people who have acknowledged every active+required onboarding item.

    Mirrors the onboarding scanner's own predicate ("the dimension is satisfied
    when every ``is_active AND required`` item has an acknowledgement row"), so
    the badge and the onboarding cockpit can never disagree. When the org has
    no required items configured at all the badge is *not* granted — an empty
    checklist is not an accomplishment.
    """
    from app.domains.onboarding.models import OnboardingAckItem, OnboardingAcknowledgement

    required = int((await db.scalar(
        select(func.count()).select_from(OnboardingAckItem).where(
            OnboardingAckItem.is_active.is_(True),
            OnboardingAckItem.required.is_(True),
        )
    )) or 0)
    if required == 0:
        return {}

    acked = await _grouped(db, (
        select(
            OnboardingAcknowledgement.person_id,
            func.count(func.distinct(OnboardingAcknowledgement.item_id)),
        )
        .join(
            OnboardingAckItem,
            and_(
                OnboardingAckItem.id == OnboardingAcknowledgement.item_id,
                OnboardingAckItem.is_active.is_(True),
                OnboardingAckItem.required.is_(True),
            ),
        )
        .where(OnboardingAcknowledgement.person_id.in_(ids))
        .group_by(OnboardingAcknowledgement.person_id)
    ), ids)
    return {pid: 1 for pid, count in acked.items() if count >= required}


async def _approvals_handled(
    db: AsyncSession, ids: Sequence[uuid.UUID],
) -> dict[uuid.UUID, int]:
    """Decisions this person made as an approver, summed across the surfaces
    that actually record an in-Helix decider.

    Deliberately NOT ``approval_steps.approver_id``: that column is never
    populated (approvals are decided in the Feishu app and only the outcome
    comes back), so counting it would score every manager 0 forever.
    """
    from app.domains.asset.models import EmployeeRequest
    from app.domains.auditor.models import AccessCertification, AccessCertificationItem
    from app.domains.cmdb.models import CmdbCloudScopeRequest
    from app.domains.email_group.models import EmailGroupJoinRequest

    totals: dict[uuid.UUID, int] = {}

    def merge(part: dict[uuid.UUID, int]) -> None:
        for pid, value in part.items():
            totals[pid] = totals.get(pid, 0) + value

    merge(await _grouped(db, (
        select(EmployeeRequest.processed_by, func.count())
        .where(
            EmployeeRequest.processed_by.in_(ids),
            EmployeeRequest.status.in_(("approved", "rejected", "completed")),
        )
        .group_by(EmployeeRequest.processed_by)
    ), ids))

    merge(await _grouped(db, (
        select(EmailGroupJoinRequest.decided_by_id, func.count())
        .where(EmailGroupJoinRequest.decided_by_id.in_(ids))
        .group_by(EmailGroupJoinRequest.decided_by_id)
    ), ids))

    # Access certifications, EXCLUDING self-attestation. A self-check writes
    # ``decided_by_person_id = the holder`` — so counting those rows made every
    # employee who did their own half-yearly access review an "approver"
    # (10 entitlements → ⚖️ 审批能手, 50 → 🏛 审批中坚), for approving nothing
    # for nobody. It also double-counted one action as both an approval and an
    # attestation. Reviewer-decided items still count; own-decided never do.
    merge(await _grouped(db, (
        select(AccessCertificationItem.decided_by_person_id, func.count())
        .join(
            AccessCertification,
            AccessCertification.id == AccessCertificationItem.certification_id,
        )
        .where(
            AccessCertificationItem.decided_by_person_id.in_(ids),
            AccessCertificationItem.decision.in_(("keep", "revoke")),
            AccessCertification.campaign_type != "self_attestation",
            AccessCertificationItem.decided_by_person_id
            != AccessCertificationItem.person_id,
        )
        .group_by(AccessCertificationItem.decided_by_person_id)
    ), ids))

    # String(36) person column — must compare against the string form.
    merge(await _grouped(db, (
        select(CmdbCloudScopeRequest.approver_person_id, func.count())
        .where(
            CmdbCloudScopeRequest.approver_person_id.in_(_str_ids(ids)),
            # NOT "approved" — that value does not exist in this model's state
            # machine, so the whole leg scored 0 for every cloud admin. The
            # real post-decision states are feishu_approved → provisioning →
            # provisioned (an approval that has moved on is still an approval).
            CmdbCloudScopeRequest.status.in_(
                ("feishu_approved", "provisioning", "provisioned", "rejected"),
            ),
        )
        .group_by(CmdbCloudScopeRequest.approver_person_id)
    ), ids))

    return totals


async def _counters(
    db: AsyncSession, ids: Sequence[uuid.UUID],
) -> dict[str, dict[uuid.UUID, int]]:
    """The count-shaped facts — one grouped query each."""
    from app.domains.ai_tool_application.models import AiToolApplication
    from app.domains.apikey.models import ApiKeyAssignment
    from app.domains.arcade.models import ArcadeGameSession
    from app.domains.asset.models import (
        AssetInventoryItem,
        AssetOperation,
        EmployeeRequest,
        SoftwareSeatAllocation,
    )
    from app.domains.auditor.models import AccessCertification, AccessCertificationItem
    from app.domains.cmdb.models import CmdbCloudAccessGrant
    from app.domains.feed.models import POST_TYPE_USER, FeedBlessing, FeedPost
    from app.domains.security_awareness.models import (
        SecawImpression,
        SecawQuizAttempt,
    )
    from app.domains.support.models import SupportDeflectionEvent

    str_ids = _str_ids(ids)
    out: dict[str, dict[uuid.UUID, int]] = {}

    # ── 云端 ─────────────────────────────────────────────────────────
    # ``person_id`` here is String(36), NOT the GUID type — see module docstring.
    out["cloud_grants"] = await _grouped(db, (
        select(CmdbCloudAccessGrant.person_id, func.count())
        .where(
            CmdbCloudAccessGrant.person_id.in_(str_ids),
            CmdbCloudAccessGrant.status == "active",
        )
        .group_by(CmdbCloudAccessGrant.person_id)
    ), ids)

    # ── 装备 ─────────────────────────────────────────────────────────
    out["devices_ever"] = await _devices_ever(db, ids)
    out["devices_returned"] = await _grouped(db, (
        select(AssetOperation.from_user_id, func.count())
        .where(
            AssetOperation.from_user_id.in_(ids),
            AssetOperation.operation_type.in_(RETURN_OPERATIONS),
            AssetOperation.status == "confirmed",
        )
        .group_by(AssetOperation.from_user_id)
    ), ids)
    out["inventory_confirmations"] = await _grouped(db, (
        select(AssetInventoryItem.holder_person_id, func.count())
        .where(
            AssetInventoryItem.holder_person_id.in_(ids),
            AssetInventoryItem.status == "confirmed",
            # ``location_admin`` rows are idle stock routed to a regional
            # administrator, who confirms hundreds per cycle — a completely
            # different act from an employee confirming the laptop on their
            # desk. Filtering to ``assignee`` needs no legacy alias: the
            # pre-543 token was never ``assignee`` either.
            AssetInventoryItem.holder_kind == "assignee",
        )
        .group_by(AssetInventoryItem.holder_person_id)
    ), ids)

    # ── 放松一下 ─────────────────────────────────────────────────────
    out["arcade_plays"] = await _grouped(db, (
        select(ArcadeGameSession.person_id, func.count())
        .where(
            ArcadeGameSession.person_id.in_(ids),
            ArcadeGameSession.status == "completed",
        )
        .group_by(ArcadeGameSession.person_id)
    ), ids)
    out["knowledge_correct"] = await _grouped(db, (
        select(
            ArcadeGameSession.person_id,
            func.sum(func.coalesce(ArcadeGameSession.correct_count, 0)),
        )
        .where(
            ArcadeGameSession.person_id.in_(ids),
            ArcadeGameSession.game_key == "quiz_rush",
            ArcadeGameSession.status == "completed",
        )
        .group_by(ArcadeGameSession.person_id)
    ), ids)

    # ── 自助与审批 ───────────────────────────────────────────────────
    # "Submitted" counts every request they raised, whatever the outcome — a
    # rejected request is still an errand they ran themselves. Cancelled ones
    # are excluded so a mis-click doesn't count.
    out["requests_submitted"] = await _grouped(db, (
        select(EmployeeRequest.requester_id, func.count())
        .where(
            EmployeeRequest.requester_id.in_(ids),
            EmployeeRequest.status != "cancelled",
        )
        .group_by(EmployeeRequest.requester_id)
    ), ids)
    out["approvals_handled"] = await _approvals_handled(db, ids)

    # ── 社区与互助 ───────────────────────────────────────────────────
    # All feed person columns are String(36).
    out["feed_posts"] = await _grouped(db, (
        select(FeedPost.author_person_id, func.count())
        .where(
            FeedPost.author_person_id.in_(str_ids),
            FeedPost.post_type == POST_TYPE_USER,
            FeedPost.deleted_at.is_(None),
        )
        .group_by(FeedPost.author_person_id)
    ), ids)
    out["feed_likes_received"] = await _grouped(db, (
        select(FeedPost.author_person_id, func.sum(func.coalesce(FeedPost.like_count, 0)))
        .where(
            FeedPost.author_person_id.in_(str_ids),
            FeedPost.deleted_at.is_(None),
        )
        .group_by(FeedPost.author_person_id)
    ), ids)
    out["blessings_sent"] = await _grouped(db, (
        select(FeedBlessing.sender_person_id, func.count())
        .where(FeedBlessing.sender_person_id.in_(str_ids))
        .group_by(FeedBlessing.sender_person_id)
    ), ids)
    out["support_deflections"] = await _grouped(db, (
        select(SupportDeflectionEvent.person_id, func.count())
        .where(
            SupportDeflectionEvent.person_id.in_(ids),
            SupportDeflectionEvent.outcome == "deflected",
        )
        .group_by(SupportDeflectionEvent.person_id)
    ), ids)

    # ── 合规 ─────────────────────────────────────────────────────────
    # Distinct self-attestation campaigns the person actually answered.
    # ``no_response`` is the cron's terminal disposition for people who never
    # replied — it must not count as "you did your review".
    out["attestations_done"] = await _grouped(db, (
        select(
            AccessCertificationItem.person_id,
            func.count(func.distinct(AccessCertificationItem.certification_id)),
        )
        .join(
            AccessCertification,
            AccessCertification.id == AccessCertificationItem.certification_id,
        )
        .where(
            AccessCertificationItem.person_id.in_(ids),
            AccessCertification.campaign_type == "self_attestation",
            AccessCertificationItem.decision.in_(("keep", "revoke")),
        )
        .group_by(AccessCertificationItem.person_id)
    ), ids)

    # 安全宣贯:读过的素材数 / 通过的素材小测数。
    # ``awareness_materials_read`` counts DISTINCT materials on the
    # ``detail`` surface only — being shown a card in the ambient bubble
    # is not having read the thing, and a badge that rewarded passive
    # exposure would reward leaving a tab open.
    out["awareness_materials_read"] = await _grouped(db, (
        select(
            SecawImpression.person_id,
            func.count(func.distinct(SecawImpression.material_id)),
        )
        .where(
            SecawImpression.person_id.in_(ids),
            SecawImpression.surface == "detail",
        )
        .group_by(SecawImpression.person_id)
    ), ids)
    out["awareness_quizzes_passed"] = await _grouped(db, (
        select(
            SecawQuizAttempt.person_id,
            func.count(func.distinct(SecawQuizAttempt.material_id)),
        )
        .where(
            SecawQuizAttempt.person_id.in_(ids),
            SecawQuizAttempt.passed.is_(True),
        )
        .group_by(SecawQuizAttempt.person_id)
    ), ids)

    # ── 软件与工具 ───────────────────────────────────────────────────
    out["seats_active"] = await _grouped(db, (
        select(SoftwareSeatAllocation.user_id, func.count())
        .where(
            SoftwareSeatAllocation.user_id.in_(ids),
            SoftwareSeatAllocation.status == "active",
        )
        .group_by(SoftwareSeatAllocation.user_id)
    ), ids)
    out["ai_tools_approved"] = await _grouped(db, (
        select(AiToolApplication.person_id, func.count())
        .where(
            AiToolApplication.person_id.in_(str_ids),
            AiToolApplication.status == "approved",
        )
        .group_by(AiToolApplication.person_id)
    ), ids)
    out["api_keys_held"] = await _grouped(db, (
        select(ApiKeyAssignment.assigned_to, func.count())
        .where(
            ApiKeyAssignment.assigned_to.in_(ids),
            ApiKeyAssignment.status == "active",
        )
        .group_by(ApiKeyAssignment.assigned_to)
    ), ids)

    return out


async def compute_facts_for(
    db: AsyncSession, persons: Sequence[IdentityPerson],
) -> dict[uuid.UUID, dict[str, Any]]:
    """Build the fact bag for every person in ``persons``.

    Returns ``{person_id: facts}`` with an entry for every requested person
    (zero-filled when they have no rows anywhere), so callers never guard on a
    missing key.
    """
    facts: dict[uuid.UUID, dict[str, Any]] = {
        p.id: dict.fromkeys(FACT_KEYS, 0) for p in persons
    }
    ids = list(facts.keys())
    if not ids:
        return facts

    for p in persons:
        facts[p.id]["tenure_years"] = _tenure_years(p)

    for pid, (total_correct, longest) in (await _quiz_stats(db, ids)).items():
        if pid in facts:
            facts[pid]["total_correct"] = total_correct
            facts[pid]["longest_streak"] = longest

    for pid, value in (await _onboarding_ack_complete(db, ids)).items():
        if pid in facts:
            facts[pid]["onboarding_ack_complete"] = value

    for key, values in (await _counters(db, ids)).items():
        for pid, value in values.items():
            if pid in facts:
                facts[pid][key] = value

    return facts


async def compute_facts(
    db: AsyncSession, person: IdentityPerson,
) -> dict[str, Any]:
    """Single-person convenience wrapper over :func:`compute_facts_for`."""
    return (await compute_facts_for(db, [person]))[person.id]


__all__ = [
    "ASSIGN_ACTIONS",
    "FACT_KEYS",
    "RETURN_OPERATIONS",
    "compute_facts",
    "compute_facts_for",
]