"""「转人工与工单群」运营台的读模型。

## 这一页要回答的问题

它不是一张「所有群」的表。它回答的是四个每天会被问到的问题:

1. **转人工到底在把人拉进来吗?** —— 转过多少次、其中多少次真的拉起了群、剩下的
   为什么没拉起来(降级分布)。降级是正式路径,但它**长期偏高**就是一个要处理的
   事实(通常是某一类还没指派负责人)。
2. **模型分得准吗?** —— 每一类被判中多少次、平均置信度、走兜底的比例。分类是
   这套东西唯一值得优化的指标。
3. **这些房间现在什么样?** —— 开着的有多少、其中多少还挂着一张没关的工单、多少
   已经没人管了(长期无活动)。
4. **过程有没有留下来?** —— 群会被解散,而聊天记录跟着它一起消失。所以「这个房
   间的讨论有没有落到工单上」是这一页最重要的一列:**它是可恢复性的唯一指标**。

## 为什么是一个聚合端点而不是让前端拼

四块数各自要一到三次聚合查询,而它们必须**同一时刻**的:一个页面上「共 40 个群」
和下面列出 37 行,读的人会以为丢了三行。一次请求一次快照。
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import Select, case, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from app.assistant import ticket_chat
from app.assistant.models import (
    CHAT_STATE_CLOSED,
    CHAT_STATE_OPEN,
    AssistantHandoff,
    AssistantTicketChat,
)
from app.domains.support.models import (
    MSG_NOTE,
    OPEN_STATUSES,
    SupportTicket,
    SupportTicketMessage,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

MAX_PAGE_SIZE = 100
#: 概览的默认窗口。30 天是这个平台其它运营面的统一口径。
DEFAULT_WINDOW_DAYS = 30
#: 多久没动静算「没人管」。与收尾任务的默认阈值一致 —— 两个数不一致的话,页面
#: 上标红的那些房间和任务实际去问的那些房间不是同一批。
IDLE_DAYS = 14


@dataclass(frozen=True, slots=True)
class ChatFilters:
    state: str | None = None
    ticket_open: bool | None = None
    opened_reason: str | None = None
    #: 只看「工单上还没有过程记录」的房间 —— 那是解散之后再也找不回来的那些。
    missing_record: bool | None = None


def _note_exists() -> Any:
    """这张工单上有没有过程记录(相关子查询)。

    EXISTS 而不是 join:一张工单可能有三条记录,而 join 会把那一行变成三行。
    """
    return (
        select(SupportTicketMessage.id)
        .where(
            SupportTicketMessage.ticket_id == AssistantTicketChat.ticket_id,
            SupportTicketMessage.kind == MSG_NOTE,
        )
        .exists()
    )


def _apply(stmt: Select, filters: ChatFilters) -> Select:
    if filters.state:
        stmt = stmt.where(AssistantTicketChat.state == filters.state)
    if filters.opened_reason:
        stmt = stmt.where(AssistantTicketChat.opened_reason == filters.opened_reason)
    if filters.ticket_open is not None:
        # 🔴 子查询里必须用**别名**。清单那条语句已经把 ``support_tickets``
        # outer-join 进来了,而一个引用同一张表的相关子查询会被 SQLAlchemy
        # 自动关联掉整个 FROM —— 报的是
        # 「returned no FROM clauses due to auto-correlation」,而这个筛选
        # 在计数那条语句上(没有 join)是好的。同一个 ``_apply`` 要在两处都
        # 成立,所以别名不是可选项。
        other = aliased(SupportTicket)
        inner = select(other.id).where(
            other.id == AssistantTicketChat.ticket_id,
            other.status.in_(OPEN_STATUSES),
        )
        stmt = stmt.where(inner.exists() if filters.ticket_open else ~inner.exists())
    if filters.missing_record is not None:
        stmt = stmt.where(
            ~_note_exists() if filters.missing_record else _note_exists(),
        )
    return stmt


async def overview(
    db: AsyncSession, *, window_days: int = DEFAULT_WINDOW_DAYS,
) -> dict[str, Any]:
    """概览:转人工的成色、分类质量、房间现状、过程记录覆盖率。"""
    since = datetime.now(UTC) - timedelta(days=window_days)

    handoffs = (await db.execute(
        select(
            func.count(AssistantHandoff.id),
            # ``sa.case`` 不是 ``func.case`` —— 后者会被当成一个叫 case 的 SQL
            # 函数,而它不接受 ``else_``。
            func.sum(
                case((AssistantHandoff.feishu_chat_id.is_not(None), 1), else_=0),
            ),
            func.avg(AssistantHandoff.classified_confidence),
        ).where(AssistantHandoff.created_at >= since)
    )).one()
    total_handoffs = int(handoffs[0] or 0)
    with_chat = int(handoffs[1] or 0)

    degraded = {
        (row[0] or "ok"): int(row[1] or 0)
        for row in (await db.execute(
            select(AssistantHandoff.degraded_reason, func.count(AssistantHandoff.id))
            .where(AssistantHandoff.created_at >= since)
            .group_by(AssistantHandoff.degraded_reason)
        )).all()
    }

    categories = [
        {
            "category": row[0] or "unclassified",
            "count": int(row[1] or 0),
            "avg_confidence": float(row[2]) if row[2] is not None else None,
        }
        for row in (await db.execute(
            select(
                AssistantHandoff.classified_category,
                func.count(AssistantHandoff.id),
                func.avg(AssistantHandoff.classified_confidence),
            )
            .where(AssistantHandoff.created_at >= since)
            .group_by(AssistantHandoff.classified_category)
            .order_by(func.count(AssistantHandoff.id).desc())
        )).all()
    ]

    states = {
        row[0]: int(row[1] or 0)
        for row in (await db.execute(
            select(AssistantTicketChat.state, func.count(AssistantTicketChat.id))
            .group_by(AssistantTicketChat.state)
        )).all()
    }

    open_chats = states.get(CHAT_STATE_OPEN, 0)
    idle_cutoff = datetime.now(UTC) - timedelta(days=IDLE_DAYS)
    idle_open = int((await db.execute(
        select(func.count(AssistantTicketChat.id)).where(
            AssistantTicketChat.state == CHAT_STATE_OPEN,
            func.coalesce(
                AssistantTicketChat.last_activity_at,
                AssistantTicketChat.created_at,
            ) < idle_cutoff,
        )
    )).scalar() or 0)

    # 🔴 过程记录覆盖率。群会被解散,而聊天记录跟着它一起消失 —— 所以「有多少
    # 开着的房间的讨论还没有落到工单上」是这一页唯一关于**可恢复性**的数。
    without_record = int((await db.execute(
        _apply(
            select(func.count(AssistantTicketChat.id)),
            ChatFilters(state=CHAT_STATE_OPEN, missing_record=True),
        )
    )).scalar() or 0)

    settled_open = int((await db.execute(
        select(func.count(AssistantTicketChat.id)).where(
            AssistantTicketChat.state == CHAT_STATE_OPEN,
            ~select(SupportTicket.id).where(
                SupportTicket.id == AssistantTicketChat.ticket_id,
                SupportTicket.status.in_(OPEN_STATUSES),
            ).exists(),
        )
    )).scalar() or 0)

    return {
        "window_days": window_days,
        "handoffs": total_handoffs,
        "handoffs_with_chat": with_chat,
        # 比例在服务端算一次:两个面(页面与 MCP)各自算会漂,而这个数会被
        # 拿去和上周比。
        "chat_rate": round(with_chat / total_handoffs, 3) if total_handoffs else None,
        "avg_confidence": (
            round(float(handoffs[2]), 3) if handoffs[2] is not None else None
        ),
        "degraded": degraded,
        "categories": categories,
        "chats_open": open_chats,
        "chats_closed": states.get(CHAT_STATE_CLOSED, 0),
        "chats_idle": idle_open,
        #: 开着的房间里,工单已经结了的那些 —— 收尾任务的待办队列。
        "chats_awaiting_cleanup": settled_open,
        #: 开着的房间里,讨论还没落到工单上的那些。解散就再也找不回来。
        "chats_without_record": without_record,
        "idle_days": IDLE_DAYS,
    }


async def list_chats(
    db: AsyncSession,
    *,
    filters: ChatFilters,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    """房间清单,每行带上「这个房间此刻的全部处境」。

    一行里同时有工单状态、群状态、最后活动、有没有过程记录 —— 因为运营台上
    真正的动作(该催了 / 该清了 / 别清)都要同时看这四样。
    """
    page_size = max(1, min(page_size, MAX_PAGE_SIZE))
    page = max(1, page)

    total = int((await db.execute(
        _apply(select(func.count(AssistantTicketChat.id)), filters)
    )).scalar() or 0)

    stmt = _apply(
        select(AssistantTicketChat, SupportTicket, _note_exists().label("has_record"))
        .outerjoin(
            SupportTicket, SupportTicket.id == AssistantTicketChat.ticket_id,
        ),
        filters,
    ).order_by(
        # 开着的排在前面(那些才需要动作),同档里最近活动的在前。
        AssistantTicketChat.state.asc(),
        func.coalesce(
            AssistantTicketChat.last_activity_at, AssistantTicketChat.created_at,
        ).desc(),
    ).offset((page - 1) * page_size).limit(page_size)

    rows = (await db.execute(stmt)).all()
    people_ids = {
        pid
        for _chat, ticket, _has in rows
        if ticket is not None
        for pid in (ticket.requester_id, ticket.assignee_id)
        if pid is not None
    }
    people: dict[Any, str] = {}
    if people_ids:
        people = {
            p.id: (p.display_name or p.work_email or "")
            for p in (await db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(people_ids)),
            )).scalars().all()
        }

    now = datetime.now(UTC)
    items: list[dict[str, Any]] = []
    for chat, ticket, has_record in rows:
        last = chat.last_activity_at or chat.created_at
        idle_days = None
        if last is not None:
            idle_days = max(
                0, (now - (last if last.tzinfo else last.replace(tzinfo=UTC))).days,
            )
        items.append({
            "id": str(chat.id),
            "chat_id": chat.feishu_chat_id,
            # 复用唯一那个 applink 构造器:两处拼同一个 URL 会漂,而漂的
            # 表现是运营台上的「进群」按钮点开一个空白页(``openId`` 是单聊的
            # 形式,``openChatId`` 才是群的)。
            "chat_url": ticket_chat.chat_applink(chat.feishu_chat_id),
            "state": chat.state,
            "opened_reason": chat.opened_reason,
            "opened_at": chat.created_at.isoformat() if chat.created_at else None,
            "last_activity_at": (
                chat.last_activity_at.isoformat() if chat.last_activity_at else None
            ),
            "idle_days": idle_days,
            "closed_at": chat.closed_at.isoformat() if chat.closed_at else None,
            "close_reason": chat.close_reason,
            "has_record": bool(has_record),
            "ticket_id": str(chat.ticket_id),
            "ticket_number": ticket.ticket_number if ticket else None,
            "ticket_title": ticket.title if ticket else None,
            "ticket_status": ticket.status if ticket else None,
            "ticket_priority": ticket.priority if ticket else None,
            "requester_name": people.get(ticket.requester_id) if ticket else None,
            "assignee_name": (
                people.get(ticket.assignee_id)
                if ticket and ticket.assignee_id else None
            ),
        })
    return items, total


async def unassigned_routes(db: AsyncSession) -> list[str]:
    """启用了但没有负责人的路由。

    它的后果很具体:那一类的「转人工」此刻降级成只建工单 —— 员工按下去拿到的是
    一个工单号而不是一个人。所以它在页面顶部报警,而不是等人逐行去看。
    """
    from app.assistant.models import AssistantEscalationRoute

    rows = (await db.execute(
        select(AssistantEscalationRoute.category_key)
        .where(
            AssistantEscalationRoute.enabled.is_(True),
            AssistantEscalationRoute.owner_person_id.is_(None),
        )
        .order_by(AssistantEscalationRoute.sort_order)
    )).scalars().all()
    return [str(r) for r in rows]


__all__ = [
    "DEFAULT_WINDOW_DAYS",
    "IDLE_DAYS",
    "MAX_PAGE_SIZE",
    "ChatFilters",
    "list_chats",
    "overview",
    "unassigned_routes",
]