"""我的已办 — the read-only history complement to ``MyTodosPage`` (待办).

The todos page surfaces items waiting for ME to *act* (asset transfers to
confirm, provisioning tasks to fulfil, cloud-grant approvals, api-key
release decisions, inventory items to respond). Once acted on, those items
drop out of the pending queues and the employee had NO way to see what
they'd already handled — the "已办看不到" gap. This aggregates, across the
same five domains, the items the CALLER personally acted on, newest first.

Mirrors the provider-registry shape of ``my_requests`` /
``mobile.action_inbox``: each provider filters its domain table by
(actor == caller) AND (terminal state), normalising to one ``DoneItem``.
Read-only; every row carries a ``deep_link`` back to the owning surface.

Actor columns are CHAR(36)-backed (``app.core.sql_types.GUID`` TypeDecorator,
or plain VARCHAR(36)); ``GUID.process_bind_param`` accepts a ``str``, so every
provider binds ``== str(person_id)`` and round-trips correctly.
"""
from __future__ import annotations

from collections import Counter
from dataclasses import dataclass
from datetime import UTC, datetime
from uuid import UUID

from pydantic import BaseModel
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.working_hours import resolve_timezone
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyReleaseRequest
from app.domains.asset.models import AssetInventoryItem
from app.domains.cmdb.models import CmdbCloudAccessGrant
from app.domains.offboarding.models import AssetTransferRequest
from app.domains.provisioning.models import (
    STATUS_CANCELLED as PROV_CANCELLED,
)
from app.domains.provisioning.models import (
    STATUS_COMPLETED as PROV_COMPLETED,
)
from app.domains.provisioning.models import (
    STATUS_FAILED as PROV_FAILED,
)
from app.domains.provisioning.models import (
    ProvisioningTask,
)
from app.identity.person_models import IdentityPerson

TYPE_TRANSFER = "asset_transfer"
TYPE_PROVISIONING = "provisioning"
TYPE_CLOUD_GRANT = "cloud_grant"
TYPE_APIKEY_RELEASE = "apikey_release"
TYPE_INVENTORY = "inventory"

_PER_TYPE_CAP = 200
# The report tab aggregates the *full* personal history (counts, status
# mix, 12-month trend) rather than one page, so it pulls with a much
# higher per-type cap. Still bounded — a single person's lifetime of
# handled items is small, and the cap protects a pathological
# software-owner with tens of thousands of provisioning tasks. When any
# provider hits this cap the stats response flags ``truncated=true`` so
# the frontend can say "showing your most recent N".
_STATS_CAP = 2000


class DoneField(BaseModel):
    """One extra label/value pair for the 已办 detail drawer.

    ``key`` is an i18n label key (``myTodos.done.field.<key>`` with raw
    fallback); ``value`` is a display-ready string. Text only — no
    timestamps, which the frontend can't locale-format from a bare
    string (``acted_at`` already covers the one timestamp that matters).
    Mirrors ``my_requests``' ``RequestContentField`` shape so the detail
    surfaces stay consistent across 我的申请 and 我的待办.
    """

    key: str
    value: str


class DoneItem(BaseModel):
    """One item the caller already handled."""

    id: str  # f"{type}:{pk}"
    type: str
    title: str
    subtitle: str | None = None
    status: str  # source-native terminal status (frontend localises)
    acted_at: str | None  # ISO 8601 — when the caller acted
    deep_link: str
    # Rich detail for the per-row drawer (the "已办看不了详情" fix). Built
    # from columns already loaded on the source row — zero extra queries.
    fields: list[DoneField] = []


class TodoHistoryResponse(BaseModel):
    items: list[DoneItem]
    counts_by_type: dict[str, int]
    total: int
    page: int
    page_size: int


@dataclass
class _Cand:
    item: DoneItem
    sort_dt: datetime


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


def _iso(dt: datetime | None) -> str | None:
    return dt.isoformat() if dt else None


def _fields(*pairs: tuple[str, object | None]) -> list[DoneField]:
    """Build the detail field list, dropping empty values."""
    out: list[DoneField] = []
    for key, value in pairs:
        if value is None:
            continue
        text = str(value).strip()
        if not text:
            continue
        out.append(DoneField(key=key, value=text))
    return out


def _mk(
    *, type_: str, pk: object, title: str, status: str, acted_at: datetime | None,
    deep_link: str, subtitle: str | None = None,
    fields: list[DoneField] | None = None,
) -> _Cand:
    return _Cand(
        item=DoneItem(
            id=f"{type_}:{pk}",
            type=type_,
            title=title,
            subtitle=subtitle,
            status=status,
            acted_at=_iso(acted_at),
            deep_link=deep_link,
            fields=fields or [],
        ),
        sort_dt=_naive(acted_at),
    )


async def _provider_transfers(
    db: AsyncSession, pid: str, cap: int = _PER_TYPE_CAP
) -> list[_Cand]:
    # Incoming transfers I responded to (recipient acted → responded_at set).
    rows = (
        await db.execute(
            select(AssetTransferRequest)
            .where(
                AssetTransferRequest.recipient_id == pid,
                AssetTransferRequest.responded_at.isnot(None),
            )
            .order_by(AssetTransferRequest.responded_at.desc())
            .limit(cap)
        )
    ).scalars().all()
    return [
        _mk(
            type_=TYPE_TRANSFER,
            pk=r.id,
            title=r.asset_label or r.asset_type,
            subtitle=r.response_note or r.reason,
            status=r.status,
            acted_at=r.responded_at,
            deep_link="/my/todos",
            fields=_fields(
                ("asset", r.asset_label),
                ("assetType", r.asset_type),
                ("reason", r.reason),
                ("note", r.response_note),
            ),
        )
        for r in rows
    ]


async def _provider_provisioning(
    db: AsyncSession, pid: str, cap: int = _PER_TYPE_CAP
) -> list[_Cand]:
    rows = (
        await db.execute(
            select(ProvisioningTask)
            .where(
                ProvisioningTask.assignee_id == pid,
                ProvisioningTask.status.in_(
                    [PROV_COMPLETED, PROV_FAILED, PROV_CANCELLED]
                ),
            )
            .order_by(ProvisioningTask.updated_at.desc())
            .limit(cap)
        )
    ).scalars().all()
    return [
        _mk(
            type_=TYPE_PROVISIONING,
            pk=r.id,
            title=f"{r.action} · {r.target_type}",
            subtitle=r.source,
            status=r.status,
            acted_at=r.completed_at or r.updated_at,
            deep_link="/my/todos",
            fields=_fields(
                ("action", r.action),
                ("target", r.target_type),
                ("source", r.source),
                ("error", r.error_message),
            ),
        )
        for r in rows
    ]


async def _provider_apikey(
    db: AsyncSession, pid: str, cap: int = _PER_TYPE_CAP
) -> list[_Cand]:
    rows = (
        await db.execute(
            select(ApiKeyReleaseRequest)
            .where(
                ApiKeyReleaseRequest.approver_id == pid,
                ApiKeyReleaseRequest.status.in_(["approved", "rejected"]),
            )
            .order_by(ApiKeyReleaseRequest.responded_at.desc())
            .limit(cap)
        )
    ).scalars().all()
    # project_name / request_number live on the parent ApiKeyAssignment,
    # NOT on the release request — batch-load them by assignment_id so the
    # detail drawer can show the recognisable request number without an
    # N+1 (and without an AttributeError from reading the wrong model).
    assignment_ids = {r.assignment_id for r in rows}
    assignments: dict[object, ApiKeyAssignment] = {}
    if assignment_ids:
        assignments = {
            a.id: a
            for a in (
                await db.execute(
                    select(ApiKeyAssignment).where(
                        ApiKeyAssignment.id.in_(assignment_ids)
                    )
                )
            ).scalars().all()
        }
    out: list[_Cand] = []
    for r in rows:
        a = assignments.get(r.assignment_id)
        out.append(
            _mk(
                type_=TYPE_APIKEY_RELEASE,
                pk=r.id,
                title=r.reason_code,  # frontend localises
                subtitle=r.reason_note or r.decision_note,
                status=r.status,
                acted_at=r.responded_at,
                deep_link="/my/todos",
                fields=_fields(
                    ("project", a.project_name if a else None),
                    ("requestNumber", a.request_number if a else None),
                    ("reasonCode", r.reason_code),
                    ("reasonNote", r.reason_note),
                    ("decisionNote", r.decision_note),
                ),
            )
        )
    return out


async def _provider_cloud_grants(
    db: AsyncSession, pid: str, cap: int = _PER_TYPE_CAP
) -> list[_Cand]:
    # Grants I personally granted (approved) or revoked.
    rows = (
        await db.execute(
            select(CmdbCloudAccessGrant)
            .where(
                or_(
                    CmdbCloudAccessGrant.granted_by_person_id == pid,
                    CmdbCloudAccessGrant.revoked_by_person_id == pid,
                )
            )
            .order_by(CmdbCloudAccessGrant.updated_at.desc())
            .limit(cap)
        )
    ).scalars().all()
    out: list[_Cand] = []
    for r in rows:
        i_revoked = r.revoked_by_person_id == pid
        out.append(
            _mk(
                type_=TYPE_CLOUD_GRANT,
                pk=r.id,
                title=r.role_display_name or r.role_key,
                subtitle=r.external_principal_email,
                status="revoked" if i_revoked else "granted",
                acted_at=(r.revoked_at if i_revoked else r.granted_at) or r.updated_at,
                deep_link=f"/cloud/scopes/{r.scope_id}" if r.scope_id else "/my/todos",
                fields=_fields(
                    ("role", r.role_display_name or r.role_key),
                    ("roleKey", r.role_key),
                    ("principal", r.external_principal_email),
                ),
            )
        )
    return out


async def _provider_inventory(
    db: AsyncSession, pid: str, cap: int = _PER_TYPE_CAP
) -> list[_Cand]:
    rows = (
        await db.execute(
            select(AssetInventoryItem)
            .where(
                AssetInventoryItem.responded_by == pid,
                AssetInventoryItem.responded_at.isnot(None),
            )
            .order_by(AssetInventoryItem.responded_at.desc())
            .limit(cap)
        )
    ).scalars().all()
    return [
        _mk(
            type_=TYPE_INVENTORY,
            pk=r.id,
            title=r.asset_name_snapshot or r.asset_tag_snapshot or "asset",
            subtitle=r.asset_tag_snapshot,
            status=r.status,
            acted_at=r.responded_at,
            deep_link=f"/me/inventory/{r.cycle_id}",
            fields=_fields(
                ("asset", r.asset_name_snapshot),
                ("assetTag", r.asset_tag_snapshot),
                ("note", r.response_notes),
            ),
        )
        for r in rows
    ]


_PROVIDERS = [
    _provider_transfers,
    _provider_provisioning,
    _provider_apikey,
    _provider_cloud_grants,
    _provider_inventory,
]


async def build_todo_history(
    db: AsyncSession,
    *,
    person_id: UUID,
    type_filter: str | None = None,
    page: int = 1,
    page_size: int = 20,
) -> TodoHistoryResponse:
    """Assemble the caller's 已办 history across all five todo domains.

    ``type_filter`` narrows to one ``DoneItem.type``; ``counts_by_type``
    always reflects the unfiltered set so the frontend chips stay stable.
    """
    pid = str(person_id)
    candidates: list[_Cand] = []
    for provider in _PROVIDERS:
        candidates.extend(await provider(db, pid))

    counts_by_type: dict[str, int] = {}
    for c in candidates:
        counts_by_type[c.item.type] = counts_by_type.get(c.item.type, 0) + 1

    if type_filter:
        candidates = [c for c in candidates if c.item.type == type_filter]

    candidates.sort(key=lambda c: c.sort_dt, reverse=True)
    total = len(candidates)
    start = (page - 1) * page_size
    page_items = [c.item for c in candidates[start : start + page_size]]

    return TodoHistoryResponse(
        items=page_items,
        counts_by_type=counts_by_type,
        total=total,
        page=page,
        page_size=page_size,
    )


# ───────────────────────────────────────────────────────────────────
# 已办报表 — aggregate analytics over the caller's full handled history.
#
# Powers the 报表 tab of ``MyDonePage`` (the list tab reuses
# ``build_todo_history``). Reuses the SAME five providers as the single
# source of truth — no parallel SQL aggregates to drift — but pulls with
# ``_STATS_CAP`` so the counts/trend reflect (effectively) the whole
# personal history rather than one page. Month bucketing honours the
# caller's local timezone so "this month" matches what they'd expect on
# their own calendar (cf. the cron-local-TZ convention).
# ───────────────────────────────────────────────────────────────────

_TREND_MONTHS = 12


class MonthBucket(BaseModel):
    """One month of handled-item activity for the trend chart."""

    month: str  # "YYYY-MM" in the caller's local timezone
    total: int
    counts: dict[str, int]  # DoneItem.type -> count in this month


class TodoHistoryStats(BaseModel):
    """Aggregate analytics over the caller's handled history."""

    total: int
    by_type: dict[str, int]
    by_status: dict[str, int]
    monthly: list[MonthBucket]
    this_month: int
    prev_month: int
    active_days: int  # distinct local dates on which the caller acted
    first_acted_at: str | None
    last_acted_at: str | None
    # True when any provider hit ``_STATS_CAP`` — the stats then cover
    # only the caller's most-recent _STATS_CAP items of that type.
    truncated: bool


def _recent_month_keys(now_local: datetime, n: int) -> list[str]:
    """The last ``n`` "YYYY-MM" keys, oldest first, ending at now_local."""
    keys: list[str] = []
    y, m = now_local.year, now_local.month
    for _ in range(n):
        keys.append(f"{y:04d}-{m:02d}")
        m -= 1
        if m == 0:
            m = 12
            y -= 1
    return list(reversed(keys))


async def build_todo_history_stats(
    db: AsyncSession, *, person: IdentityPerson
) -> TodoHistoryStats:
    """Assemble the 已办 report for ``person`` across all five domains."""
    pid = str(person.id)
    tz = resolve_timezone(person)

    candidates: list[_Cand] = []
    truncated = False
    for provider in _PROVIDERS:
        rows = await provider(db, pid, _STATS_CAP)
        if len(rows) >= _STATS_CAP:
            truncated = True
        candidates.extend(rows)

    by_type: Counter[str] = Counter()
    by_status: Counter[str] = Counter()
    # month "YYYY-MM" -> (type -> count)
    month_type: dict[str, Counter[str]] = {}
    active_dates: set[str] = set()
    sort_dts: list[datetime] = []

    for c in candidates:
        by_type[c.item.type] += 1
        by_status[c.item.status] += 1
        # ``sort_dt`` is naive (DB DATETIME loaded as naive UTC); attach
        # UTC then convert to the caller's local zone for bucketing.
        if c.sort_dt != datetime.min:
            local = c.sort_dt.replace(tzinfo=UTC).astimezone(tz)
            mkey = local.strftime("%Y-%m")
            month_type.setdefault(mkey, Counter())[c.item.type] += 1
            active_dates.add(local.strftime("%Y-%m-%d"))
            sort_dts.append(c.sort_dt)

    now_local = datetime.now(UTC).astimezone(tz)
    month_keys = _recent_month_keys(now_local, _TREND_MONTHS)
    monthly = [
        MonthBucket(
            month=mk,
            total=int(sum(month_type.get(mk, Counter()).values())),
            counts=dict(month_type.get(mk, Counter())),
        )
        for mk in month_keys
    ]

    this_key = month_keys[-1]
    prev_key = month_keys[-2] if len(month_keys) >= 2 else None
    this_month = int(sum(month_type.get(this_key, Counter()).values()))
    prev_month = (
        int(sum(month_type.get(prev_key, Counter()).values())) if prev_key else 0
    )

    return TodoHistoryStats(
        total=len(candidates),
        by_type=dict(by_type),
        by_status=dict(by_status),
        monthly=monthly,
        this_month=this_month,
        prev_month=prev_month,
        active_days=len(active_dates),
        first_acted_at=_iso(min(sort_dts)) if sort_dts else None,
        last_acted_at=_iso(max(sort_dts)) if sort_dts else None,
        truncated=truncated,
    )