"""The half-yearly ownership confirmation round.

A register decays the moment nobody is asked to stand behind it. This
module is the asking: twice a year every named owner gets one card
listing the flows recorded against them, and answers each one with
「还是我」 or 「转给某某」.

Three design points that are easy to get wrong:

* **The round only opens inside a window after the boundary.** June 1
  and December 1, plus seven days. A deploy in the middle of a
  half-year must not retroactively open a round nobody was expecting —
  the same storm guard the半年度访问自查 uses, for the same reason: the
  first thing a mid-cycle deploy would otherwise do is push a card to
  every owner in the company.
* **One card per person, never one per flow.** An owner with nine flows
  gets one card listing nine, not nine cards. The card is the whole
  interaction for most people; making it nine is how it gets muted.
* **An unanswered item at close is ``lapsed``, not ``pending``.** At
  close time the difference between "still waiting" and "nobody ever
  answered" is the entire finding, and collapsing them loses it.
"""
from __future__ import annotations

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

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.core.exceptions import ConflictError, NotFoundError, ValidationError
from app.domains.approval_registry import constants as const
from app.domains.approval_registry import service as svc
from app.domains.approval_registry.models import (
    ApprovalFlow,
    FlowConfirmationItem,
    FlowConfirmationRound,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)


def _now() -> datetime:
    return datetime.now(UTC)


def _naive(dt: datetime) -> datetime:
    return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt


# ════════════════════════════════════════════════════════════════════
# Boundaries
# ════════════════════════════════════════════════════════════════════


def label_for(moment: datetime) -> str:
    """``2026-H1`` / ``2026-H2`` for the half-year ``moment`` falls in."""
    half = "H1" if moment.month < 7 else "H2"
    return f"{moment.year}-{half}"


def due_boundary(moment: datetime, *, window_days: int) -> tuple[str, datetime] | None:
    """The boundary this moment is inside the opening window of.

    Returns ``(label, boundary_datetime)`` or ``None``. Only the CURRENT
    year's boundaries are considered: a January run must not decide it
    is inside last December's window and open a round four weeks late.
    """
    for month, day, suffix in const.ROUND_BOUNDARIES:
        boundary = datetime(moment.year, month, day, tzinfo=UTC)
        if boundary <= moment < boundary + timedelta(days=window_days):
            return f"{moment.year}-{suffix}", boundary
    return None


# ════════════════════════════════════════════════════════════════════
# Reads
# ════════════════════════════════════════════════════════════════════


async def get_round(db: AsyncSession, round_id: uuid.UUID) -> FlowConfirmationRound:
    row = (
        await db.execute(
            select(FlowConfirmationRound).where(
                FlowConfirmationRound.id == round_id,
            ),
        )
    ).scalar_one_or_none()
    if row is None:
        raise NotFoundError("FlowConfirmationRound", str(round_id))
    return row


async def get_round_by_label(
    db: AsyncSession, label: str, *, lock: bool = False,
) -> FlowConfirmationRound | None:
    """The round for a label. ``lock=True`` reads the latest COMMITTED
    row rather than this transaction's snapshot — required in the
    conflict-recovery path below, where the winner committed after our
    snapshot was taken and a plain read would still return ``None``."""
    stmt = select(FlowConfirmationRound).where(
        FlowConfirmationRound.label == label,
    )
    if lock:
        stmt = stmt.with_for_update().execution_options(populate_existing=True)
    return (await db.execute(stmt)).scalar_one_or_none()


async def current_open_round(db: AsyncSession) -> FlowConfirmationRound | None:
    return (
        await db.execute(
            select(FlowConfirmationRound)
            .where(FlowConfirmationRound.status == const.ROUND_OPEN)
            .order_by(FlowConfirmationRound.opened_at.desc())
            .limit(1),
        )
    ).scalar_one_or_none()


async def list_rounds(
    db: AsyncSession, *, limit: int = 20,
) -> list[FlowConfirmationRound]:
    return list(
        (
            await db.execute(
                select(FlowConfirmationRound)
                .order_by(FlowConfirmationRound.opened_at.desc())
                .limit(limit),
            )
        ).scalars().all(),
    )


async def round_items(
    db: AsyncSession, round_id: uuid.UUID,
) -> list[FlowConfirmationItem]:
    return list(
        (
            await db.execute(
                select(FlowConfirmationItem)
                .where(FlowConfirmationItem.round_id == round_id)
                .order_by(FlowConfirmationItem.created_at.asc()),
            )
        ).scalars().all(),
    )


async def items_for_owner(
    db: AsyncSession, round_id: uuid.UUID, person_id: uuid.UUID,
) -> list[FlowConfirmationItem]:
    return list(
        (
            await db.execute(
                select(FlowConfirmationItem).where(
                    FlowConfirmationItem.round_id == round_id,
                    FlowConfirmationItem.owner_person_id == person_id,
                ),
            )
        ).scalars().all(),
    )


async def round_summary(
    db: AsyncSession, round_row: FlowConfirmationRound,
) -> dict[str, Any]:
    """Live counts for an open round; the frozen ones for a closed one.

    A closed round reports what it looked like at close. Recomputing it
    later would let a transfer made in March rewrite December's
    completion rate — the number is a record of an event, not a view.
    """
    if round_row.status == const.ROUND_CLOSED and round_row.final_stats:
        return dict(round_row.final_stats)
    rows = (
        await db.execute(
            select(FlowConfirmationItem.state, func.count())
            .where(FlowConfirmationItem.round_id == round_row.id)
            .group_by(FlowConfirmationItem.state),
        )
    ).all()
    by_state = {state: int(count) for state, count in rows}
    total = sum(by_state.values())
    in_scope = total - by_state.get(const.ITEM_DROPPED, 0)
    answered = sum(by_state.get(s, 0) for s in const.ANSWERED_STATES)
    return {
        "total": total,
        "in_scope": in_scope,
        "answered": answered,
        "pending": by_state.get(const.ITEM_PENDING, 0),
        "confirmed": by_state.get(const.ITEM_CONFIRMED, 0),
        "transferred": by_state.get(const.ITEM_TRANSFERRED, 0),
        "lapsed": by_state.get(const.ITEM_LAPSED, 0),
        "dropped": by_state.get(const.ITEM_DROPPED, 0),
        "completion_pct": (
            round(answered * 100 / in_scope, 1) if in_scope else 0.0
        ),
    }


# ════════════════════════════════════════════════════════════════════
# Opening
# ════════════════════════════════════════════════════════════════════


async def open_round(
    db: AsyncSession,
    *,
    label: str,
    due_days: int = const.ROUND_DUE_DAYS,
    actor: IdentityPerson | None = None,
) -> tuple[FlowConfirmationRound, bool]:
    """Create the round and snapshot one item per owned live flow.

    Returns ``(round, created)``. Idempotent on ``label``: a second call
    in the same window returns the existing round untouched, so a cron
    that runs daily inside a seven-day window opens one campaign.

    Flows with **no** owner are deliberately not snapshotted. There is
    nobody to ask; they are already counted by the coverage read, and
    putting them in the round would depress a completion rate that
    measures whether people answered.
    """
    existing = await get_round_by_label(db, label)
    if existing is not None:
        return existing, False

    now = _naive(_now())
    round_row = FlowConfirmationRound(
        label=label,
        status=const.ROUND_OPEN,
        opened_at=now,
        due_at=now + timedelta(days=due_days),
        opened_by=actor.id if actor else None,
    )
    try:
        # ``add`` INSIDE the savepoint, not before it: a row added
        # outside is still pending when the savepoint rolls back, so the
        # recovery SELECT re-flushes it and raises again instead of
        # replaying. (Documented platform trap; same shape as
        # ``ensure_run`` and the attendance roster override.)
        async with db.begin_nested():
            db.add(round_row)
            await db.flush()
    except IntegrityError:
        # ``label`` is UNIQUE and two operators can press 「立即开盘」 at
        # the same moment: both see "no round yet", both insert, and the
        # loser would get a 500 for doing something reasonable. Adopt the
        # round that won — the caller wanted a round for this window, and
        # there is one. The re-read MUST lock: under REPEATABLE READ this
        # transaction's snapshot predates the winner's commit, so a plain
        # consistent read returns ``None`` and we would re-raise the 500
        # we just caught.
        winner = await get_round_by_label(db, label, lock=True)
        if winner is None:
            raise
        return winner, False

    owner = selectinload(ApprovalFlow.owner)
    flows = (
        await db.execute(
            select(ApprovalFlow)
            .where(
                ApprovalFlow.deleted_at.is_(None),
                ApprovalFlow.status.in_(const.LIVE_STATUSES),
                ApprovalFlow.owner_person_id.is_not(None),
            )
            .options(owner),
        )
    ).scalars().all()

    for flow in flows:
        person = flow.owner
        # An owner who has already left is not asked — the offboarding
        # blocker owns that case, and a card into a dead mailbox is a
        # completion rate that can never reach 100%.
        if person is None or person.employment_status != "active":
            continue
        db.add(FlowConfirmationItem(
            round_id=round_row.id,
            flow_id=flow.id,
            owner_person_id=flow.owner_person_id,
            state=const.ITEM_PENDING,
        ))
    await db.flush()
    return round_row, True


async def auto_open_semiannual(
    db: AsyncSession,
    *,
    window_days: int = const.ROUND_WINDOW_DAYS,
    due_days: int = const.ROUND_DUE_DAYS,
) -> dict[str, Any]:
    """Daily cron: open the round only inside a boundary window."""
    now = _now()
    boundary = due_boundary(now, window_days=window_days)
    if boundary is None:
        return {"opened": False, "reason": "outside the boundary window"}
    label, _at = boundary
    round_row, created = await open_round(db, label=label, due_days=due_days)
    return {
        "opened": created,
        "label": label,
        "round_id": str(round_row.id),
        "items": len(await round_items(db, round_row.id)) if created else 0,
        "reason": None if created else "round already open for this window",
    }


# ════════════════════════════════════════════════════════════════════
# Answering
# ════════════════════════════════════════════════════════════════════


async def _flow_for_item(
    db: AsyncSession, item: FlowConfirmationItem,
) -> ApprovalFlow:
    flow = (
        await db.execute(
            select(ApprovalFlow)
            .where(ApprovalFlow.id == item.flow_id)
            .options(selectinload(ApprovalFlow.owner))
            .execution_options(populate_existing=True),
        )
    ).scalar_one_or_none()
    if flow is None:
        raise NotFoundError("ApprovalFlow", str(item.flow_id))
    return flow


async def answer_item(
    db: AsyncSession,
    item: FlowConfirmationItem,
    *,
    actor: IdentityPerson,
    keep: bool,
    transfer_to_person_id: uuid.UUID | None = None,
    note: str | None = None,
    via: str = "web",
) -> FlowConfirmationItem:
    """Record one owner's answer and apply it to the register.

    ``keep=True`` stamps the confirmation on the flow (the only thing
    that turns a name in a box into coverage). ``keep=False`` requires a
    successor: 「不是我」 without saying who leaves the register no better
    off than before it asked, and the register cannot guess.
    """
    if item.owner_person_id != actor.id:
        raise ConflictError(
            "this confirmation item was addressed to somebody else",
            code="APPROVAL_FLOW_ITEM_NOT_YOURS",
        )
    round_row = await get_round(db, item.round_id)
    if round_row.status != const.ROUND_OPEN:
        raise ConflictError(
            f"round {round_row.label} is closed",
            code="APPROVAL_FLOW_ROUND_CLOSED",
        )
    if item.state in const.ANSWERED_STATES:
        # Idempotent: the card button can be pressed twice, and the
        # second press must not look like an error to the person.
        return item

    flow = await _flow_for_item(db, item)
    if flow.owner_person_id != actor.id:
        # The flow moved on after the round opened — IT re-pointed it,
        # or an offboarding handover did. The item is still pending
        # against the person we ASKED, and confirming it would write
        # their name back onto a flow somebody else now owns, silently
        # taking it from them. The question 「这还是你的吗」 simply no
        # longer applies: drop the item rather than answer it.
        item.state = const.ITEM_DROPPED
        item.decided_at = _naive(_now())
        item.decided_via = via[:16]
        item.note = "流程已在本轮期间转给其他人"
        return item

    if keep:
        await svc.set_owner(
            db, flow, owner_person_id=actor.id, actor=actor,
            owner_source=const.OWNER_SOURCE_CONFIRMED, confirmed=True,
            note=note or f"{round_row.label} 确认仍由本人负责",
        )
        item.state = const.ITEM_CONFIRMED
    else:
        if transfer_to_person_id is None:
            raise ValidationError(
                "转交时必须指定接手人 —— 「不是我」而不说是谁，"
                "台账并不会因此更准",
            )
        successor = svc.assert_assignable(
            await svc.load_person(db, transfer_to_person_id),
        )
        await svc.set_owner(
            db, flow, owner_person_id=successor.id, actor=actor,
            owner_source=const.OWNER_SOURCE_MANUAL,
            note=note or f"{round_row.label} 由 {svc.person_label(actor)} 转交",
            announce=const.HANDOVER_TRANSFER,
        )
        item.state = const.ITEM_TRANSFERRED
        item.transferred_to_person_id = successor.id
    item.decided_at = _naive(_now())
    item.decided_via = via[:16]
    item.note = note
    return item


async def answer_all_for_owner(
    db: AsyncSession,
    round_row: FlowConfirmationRound,
    *,
    actor: IdentityPerson,
    via: str = "token",
) -> int:
    """「全部确认」 — the one-click path from the card.

    Only touches items still pending: a person who already handed two
    flows over and then pressed the bulk button must not have those two
    silently taken back.
    """
    answered = 0
    for item in await items_for_owner(db, round_row.id, actor.id):
        if item.state != const.ITEM_PENDING:
            continue
        await answer_item(db, item, actor=actor, keep=True, via=via)
        # Count what was actually answered, not what was visited: an
        # item whose flow moved on is dropped rather than confirmed, and
        # telling the person 「已确认 3 条」 when one of them was not
        # theirs to confirm is a small lie on the page they trust most.
        if item.state in const.ANSWERED_STATES:
            answered += 1
    return answered


# ════════════════════════════════════════════════════════════════════
# Housekeeping
# ════════════════════════════════════════════════════════════════════


async def drop_stale_items(db: AsyncSession, round_row: FlowConfirmationRound) -> int:
    """Retire items whose question stopped making sense mid-round.

    Two cases: the flow was archived, or the person we asked has left.
    Both are ``dropped``, not ``lapsed`` — chasing an answer about a
    retired flow, or from somebody who is gone, is not a finding, and
    counting it as one makes the completion rate meaningless.
    """
    dropped = 0
    rows = (
        await db.execute(
            select(FlowConfirmationItem, ApprovalFlow, IdentityPerson)
            .join(ApprovalFlow, ApprovalFlow.id == FlowConfirmationItem.flow_id)
            .outerjoin(
                IdentityPerson,
                IdentityPerson.id == FlowConfirmationItem.owner_person_id,
            )
            .where(
                FlowConfirmationItem.round_id == round_row.id,
                FlowConfirmationItem.state == const.ITEM_PENDING,
            ),
        )
    ).all()
    for item, flow, person in rows:
        gone = (
            person is None
            or person.employment_status != "active"
            or person.deleted_at is not None
        )
        retired = flow.deleted_at is not None or flow.status == const.STATUS_RETIRED
        if gone or retired:
            item.state = const.ITEM_DROPPED
            item.decided_at = _naive(_now())
            item.note = "流程已归档" if retired else "被询问的负责人已离职"
            dropped += 1
    return dropped


async def close_round(
    db: AsyncSession,
    round_row: FlowConfirmationRound,
    *,
    actor: IdentityPerson | None = None,
) -> dict[str, Any]:
    """Close it: drop what became irrelevant, lapse the rest, freeze the
    numbers."""
    if round_row.status == const.ROUND_CLOSED:
        return await round_summary(db, round_row)
    await drop_stale_items(db, round_row)
    for item in await round_items(db, round_row.id):
        if item.state == const.ITEM_PENDING:
            item.state = const.ITEM_LAPSED
    await db.flush()
    stats = await round_summary(db, round_row)
    round_row.status = const.ROUND_CLOSED
    round_row.closed_at = _naive(_now())
    round_row.final_stats = stats
    return stats


async def overdue_items(
    db: AsyncSession,
    round_row: FlowConfirmationRound,
    *,
    after_days: int = const.ESCALATE_AFTER_DAYS,
) -> list[FlowConfirmationItem]:
    """Pending items old enough to chase, not yet chased.

    The clock runs from ``notified_at``, not from when the round opened,
    and an item that was never delivered is never chased. Escalation
    copies the owner's MANAGER; doing that over a card the platform
    failed to send blames somebody for silence they were never given the
    chance to break. An undelivered ask is the platform's bug to fix,
    not the employee's to answer for.
    """
    cutoff = _naive(_now() - timedelta(days=after_days))
    return list(
        (
            await db.execute(
                select(FlowConfirmationItem).where(
                    FlowConfirmationItem.round_id == round_row.id,
                    FlowConfirmationItem.state == const.ITEM_PENDING,
                    FlowConfirmationItem.escalated_at.is_(None),
                    FlowConfirmationItem.notified_at.is_not(None),
                    FlowConfirmationItem.notified_at < cutoff,
                ),
            )
        ).scalars().all(),
    )


async def unnotified_items(
    db: AsyncSession, round_row: FlowConfirmationRound,
) -> list[FlowConfirmationItem]:
    """Pending items whose card never went out.

    The opening fan-out happens once, on the run that creates the round.
    If Feishu is unreachable in that minute — twice a year, for one
    minute — every owner's card is lost and nothing retries: escalation
    correctly refuses to chase an ask nobody received, so the round
    would sit unasked until it closed and lapsed in full. The daily
    escalator re-sends these, which turns a transient outage into a
    late card instead of a wasted half-year.
    """
    return list(
        (
            await db.execute(
                select(FlowConfirmationItem).where(
                    FlowConfirmationItem.round_id == round_row.id,
                    FlowConfirmationItem.state == const.ITEM_PENDING,
                    FlowConfirmationItem.notified_at.is_(None),
                ),
            )
        ).scalars().all(),
    )


def group_by_owner(
    items: list[FlowConfirmationItem],
) -> dict[uuid.UUID, list[FlowConfirmationItem]]:
    """One card per person, never one per flow."""
    grouped: dict[uuid.UUID, list[FlowConfirmationItem]] = defaultdict(list)
    for item in items:
        if item.owner_person_id is not None:
            grouped[item.owner_person_id].append(item)
    return dict(grouped)


async def record_self_answer(
    db: AsyncSession,
    flow: ApprovalFlow,
    *,
    actor: IdentityPerson,
    keep: bool,
    transfer_to_person_id: uuid.UUID | None = None,
    note: str | None = None,
    via: str = "web",
) -> dict[str, Any]:
    """The owner answers from their own page — routed through the open
    round when there is one.

    One path, not two. If the answer only wrote to the flow and skipped
    the round item, the round would report somebody as unresponsive
    while their answer sat in the register: the completion rate and the
    register would disagree about the same person on the same day.
    """
    round_row = await current_open_round(db)
    item = None
    if round_row is not None:
        item = (
            await db.execute(
                select(FlowConfirmationItem).where(
                    FlowConfirmationItem.round_id == round_row.id,
                    FlowConfirmationItem.flow_id == flow.id,
                    FlowConfirmationItem.owner_person_id == actor.id,
                ),
            )
        ).scalar_one_or_none()

    if item is not None:
        await answer_item(
            db, item, actor=actor, keep=keep,
            transfer_to_person_id=transfer_to_person_id, note=note, via=via,
        )
        return {"round": round_row.label if round_row else None, "item": item.state}

    # No open round (or this flow was registered after it opened): the
    # answer still stands on its own — an owner who confirms in March
    # should not be told to wait for June.
    if keep:
        await svc.confirm_owner(db, flow, actor=actor, note=note)
        return {"round": None, "item": const.ITEM_CONFIRMED}
    if transfer_to_person_id is None:
        raise ValidationError(
            "转交时必须指定接手人 —— 「不是我」而不说是谁，台账并不会因此更准",
        )
    successor = svc.assert_assignable(
        await svc.load_person(db, transfer_to_person_id),
    )
    await svc.set_owner(
        db, flow, owner_person_id=successor.id, actor=actor,
        owner_source=const.OWNER_SOURCE_MANUAL,
        note=note or f"由 {svc.person_label(actor)} 转交",
        announce=const.HANDOVER_TRANSFER,
    )
    return {"round": None, "item": const.ITEM_TRANSFERRED}