"""服务台分析 read service — KPI / trend / breakdown aggregations + data list.

All aggregations window on ``created_at_feishu`` (the ticket's creation
day). Durations average over non-NULL rows only (the sync already mapped
the API's ``-1`` sentinel to NULL), so unfinished tickets don't drag an
SLA mean toward zero.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.domains.feishu_helpdesk.models import FeishuHelpdeskTicket
from app.identity.person_models import IdentityOrgMembership
from app.integrations.models import Integration

# status enum (Feishu helpdesk, real production codes confirmed in #1331):
#   1=待处理 2=处理中 (open/unresolved) · 50=已解决 51/52=已关闭 (terminal).
# "Open" = not-yet-terminal = pending(1) or in-progress(2).
OPEN_STATUSES = (1, 2)
DEFAULT_WINDOW_DAYS = 30
_MAX_WINDOW_DAYS = 366


def _window(
    start: datetime | None, end: datetime | None, days: int | None,
) -> tuple[datetime, datetime]:
    now = datetime.now(UTC)
    if end is None:
        end = now
    if start is None:
        span = min(days or DEFAULT_WINDOW_DAYS, _MAX_WINDOW_DAYS)
        start = end - timedelta(days=span)
    return start, end


def _base_filters(
    integration_id: uuid.UUID | None, start: datetime, end: datetime,
) -> list[Any]:
    conds: list[Any] = [
        FeishuHelpdeskTicket.created_at_feishu >= start,
        FeishuHelpdeskTicket.created_at_feishu <= end,
    ]
    if integration_id is not None:
        conds.append(FeishuHelpdeskTicket.integration_id == integration_id)
    return conds


async def list_helpdesks(db: AsyncSession) -> list[dict[str, Any]]:
    """Configured 服务台 (integration rows) for the selector, with the
    mirrored ticket count each currently holds."""
    rows = (
        await db.execute(
            select(Integration).where(
                Integration.kind == "helpdesk",
                Integration.provider == "feishu",
            ).order_by(Integration.name)
        )
    ).scalars().all()
    counts: dict[uuid.UUID, int] = {
        iid: int(n)
        for iid, n in (
            await db.execute(
                select(
                    FeishuHelpdeskTicket.integration_id,
                    func.count().label("n"),
                ).group_by(FeishuHelpdeskTicket.integration_id)
            )
        ).all()
    }
    return [
        {
            "integration_id": str(r.id),
            "name": r.name,
            "slug": r.slug,
            "enabled": r.enabled,
            "helpdesk_id": (r.config_public or {}).get("helpdesk_id"),
            "ticket_count": int(counts.get(r.id, 0)),
        }
        for r in rows
    ]


async def get_analytics(
    db: AsyncSession,
    *,
    integration_id: uuid.UUID | None,
    start: datetime | None,
    end: datetime | None,
    days: int | None,
) -> dict[str, Any]:
    start, end = _window(start, end, days)
    base = _base_filters(integration_id, start, end)

    # ── KPIs ──
    kpi_row = (
        await db.execute(
            select(
                func.count().label("total"),
                func.sum(
                    case((FeishuHelpdeskTicket.status.in_(OPEN_STATUSES), 1), else_=0)
                ).label("open"),
                func.sum(
                    case((FeishuHelpdeskTicket.solve == 1, 1), else_=0)
                ).label("solved"),
                func.sum(
                    case((FeishuHelpdeskTicket.solve.in_((1, 2)), 1), else_=0)
                ).label("solve_known"),
                func.avg(FeishuHelpdeskTicket.first_response_seconds).label("avg_frt"),
                func.avg(FeishuHelpdeskTicket.resolution_seconds).label("avg_res"),
                # Feishu uses score=0 for 未评价 (un-rated). Only ratings
                # >=1 count toward satisfaction (3-point CSAT: 1=不满意 /
                # 2=一般 / 3=满意) — case→NULL otherwise so AVG/COUNT ignore
                # the un-rated rows.
                func.avg(
                    case((FeishuHelpdeskTicket.score >= 1, FeishuHelpdeskTicket.score))
                ).label("avg_score"),
                func.sum(
                    case((FeishuHelpdeskTicket.score >= 1, 1), else_=0)
                ).label("scored"),
                func.sum(
                    case((FeishuHelpdeskTicket.score == 3, 1), else_=0)
                ).label("satisfied"),
            ).where(and_(*base))
        )
    ).one()
    total = int(kpi_row.total or 0)
    solve_known = int(kpi_row.solve_known or 0)
    kpis = {
        "total": total,
        "open": int(kpi_row.open or 0),
        "solved": int(kpi_row.solved or 0),
        "solve_rate": (round(int(kpi_row.solved or 0) / solve_known, 4) if solve_known else None),
        "avg_first_response_seconds": (
            round(float(kpi_row.avg_frt), 1) if kpi_row.avg_frt is not None else None
        ),
        "avg_resolution_seconds": (
            round(float(kpi_row.avg_res), 1) if kpi_row.avg_res is not None else None
        ),
        "satisfaction_avg": (
            round(float(kpi_row.avg_score), 2) if kpi_row.avg_score is not None else None
        ),
        "scored_count": int(kpi_row.scored or 0),
        # 满意率 = 满意(score=3) / 已评价(score>=1) — clearer than a 3-point
        # average for the KPI tile.
        "satisfaction_rate": (
            round(int(kpi_row.satisfied or 0) / int(kpi_row.scored), 4)
            if kpi_row.scored else None
        ),
    }

    # ── Daily trend (by created day) ──
    day = func.date(FeishuHelpdeskTicket.created_at_feishu)
    trend_rows = (
        await db.execute(
            select(day.label("d"), func.count().label("n"))
            .where(and_(*base))
            .group_by(day)
            .order_by(day)
        )
    ).all()
    trend = [{"date": str(d), "count": int(n)} for d, n in trend_rows if d is not None]

    async def _breakdown(col) -> list[dict[str, Any]]:
        rows = (
            await db.execute(
                select(col.label("k"), func.count().label("n"))
                .where(and_(*base))
                .group_by(col)
                .order_by(func.count().desc())
            )
        ).all()
        return [{"key": k, "count": int(n)} for k, n in rows]

    by_status = await _breakdown(FeishuHelpdeskTicket.status)
    by_channel = await _breakdown(FeishuHelpdeskTicket.channel)
    by_type = await _breakdown(FeishuHelpdeskTicket.ticket_type)
    by_score = await _breakdown(FeishuHelpdeskTicket.score)
    by_solve = await _breakdown(FeishuHelpdeskTicket.solve)

    # ── Daily resolved trend (by closed day) — pairs with the created
    # trend so the report shows backlog build-up vs burn-down. ──
    cday = func.date(FeishuHelpdeskTicket.closed_at_feishu)
    resolved_rows = (
        await db.execute(
            select(cday.label("d"), func.count().label("n"))
            .where(and_(
                FeishuHelpdeskTicket.closed_at_feishu >= start,
                FeishuHelpdeskTicket.closed_at_feishu <= end,
                *([FeishuHelpdeskTicket.integration_id == integration_id]
                  if integration_id is not None else []),
            ))
            .group_by(cday)
            .order_by(cday)
        )
    ).all()
    resolved_trend = [
        {"date": str(d), "count": int(n)} for d, n in resolved_rows if d is not None
    ]

    # ── Agent workload (top 20) ──
    agent_rows = (
        await db.execute(
            select(
                FeishuHelpdeskTicket.agent_owner_name.label("name"),
                FeishuHelpdeskTicket.agent_owner_email.label("email"),
                func.count().label("n"),
                func.avg(FeishuHelpdeskTicket.first_response_seconds).label("avg_frt"),
            )
            .where(and_(*base, FeishuHelpdeskTicket.agent_owner_name.is_not(None)))
            .group_by(
                FeishuHelpdeskTicket.agent_owner_name,
                FeishuHelpdeskTicket.agent_owner_email,
            )
            .order_by(func.count().desc())
            .limit(20)
        )
    ).all()
    agents = [
        {
            "name": name,
            "email": email,
            "count": int(n),
            "avg_first_response_seconds": (round(float(avg_frt), 1) if avg_frt is not None else None),
        }
        for name, email, n, avg_frt in agent_rows
    ]

    # ── Department breakdown (top 20, via resolved guest person's primary
    # org membership — ``IdentityPerson.department`` is a Python property
    # derived from org_memberships, not a queryable column). ──
    dept_rows = (
        await db.execute(
            select(
                IdentityOrgMembership.org_node_name.label("dept"),
                func.count().label("n"),
            )
            .select_from(FeishuHelpdeskTicket)
            .join(
                IdentityOrgMembership,
                and_(
                    IdentityOrgMembership.person_id
                    == FeishuHelpdeskTicket.guest_person_id,
                    IdentityOrgMembership.is_primary.is_(True),
                    IdentityOrgMembership.deleted_at.is_(None),
                ),
            )
            .where(and_(*base, FeishuHelpdeskTicket.guest_person_id.is_not(None)))
            .group_by(IdentityOrgMembership.org_node_name)
            .order_by(func.count().desc())
            .limit(20)
        )
    ).all()
    by_department = [
        {"department": dept or "(未知)", "count": int(n)} for dept, n in dept_rows
    ]

    # Resolution coverage: how many tickets could be attributed to a Helix person.
    attributed = (
        await db.execute(
            select(func.count()).where(
                and_(*base, FeishuHelpdeskTicket.guest_person_id.is_not(None))
            )
        )
    ).scalar() or 0

    return {
        "window": {"start": start.isoformat(), "end": end.isoformat()},
        "kpis": kpis,
        "trend": trend,
        "resolved_trend": resolved_trend,
        "by_status": by_status,
        "by_channel": by_channel,
        "by_type": by_type,
        "by_score": by_score,
        "by_solve": by_solve,
        "agents": agents,
        "by_department": by_department,
        "attribution": {"attributed": int(attributed), "total": total},
    }


def _data_query(
    integration_id: uuid.UUID | None,
    status: int | None,
    search: str | None,
    start: datetime,
    end: datetime,
) -> Select:
    conds = _base_filters(integration_id, start, end)
    if status is not None:
        conds.append(FeishuHelpdeskTicket.status == status)
    if search:
        like = f"%{search.strip()}%"
        conds.append(
            func.lower(FeishuHelpdeskTicket.guest_name).like(func.lower(like))
            | func.lower(FeishuHelpdeskTicket.guest_email).like(func.lower(like))
            | (FeishuHelpdeskTicket.ticket_id == search.strip())
        )
    return select(FeishuHelpdeskTicket).where(and_(*conds))


def serialize_ticket(t: FeishuHelpdeskTicket) -> dict[str, Any]:
    def _iso(dt: datetime | None) -> str | None:
        return dt.isoformat() if dt else None

    return {
        "ticket_id": t.ticket_id,
        "helpdesk_id": t.helpdesk_id,
        "helpdesk_name": t.helpdesk_name,
        "status": t.status,
        "channel": t.channel,
        "ticket_type": t.ticket_type,
        "solve": t.solve,
        "score": t.score,
        "language": t.language,
        "guest_name": t.guest_name,
        "guest_email": t.guest_email,
        "guest_person_id": str(t.guest_person_id) if t.guest_person_id else None,
        "agent_owner_name": t.agent_owner_name,
        "agent_owner_email": t.agent_owner_email,
        "agent_group": t.agent_group,
        "first_response_seconds": t.first_response_seconds,
        "resolution_seconds": t.resolution_seconds,
        "agent_service_minutes": t.agent_service_minutes,
        "created_at": _iso(t.created_at_feishu),
        "updated_at": _iso(t.updated_at_feishu),
        "closed_at": _iso(t.closed_at_feishu),
    }


async def list_tickets(
    db: AsyncSession,
    *,
    integration_id: uuid.UUID | None,
    status: int | None,
    search: str | None,
    start: datetime | None,
    end: datetime | None,
    days: int | None,
    page: int,
    page_size: int,
) -> tuple[list[dict[str, Any]], int]:
    start, end = _window(start, end, days)
    q = _data_query(integration_id, status, search, start, end)
    total = (
        await db.execute(select(func.count()).select_from(q.subquery()))
    ).scalar() or 0
    rows = (
        await db.execute(
            q.order_by(FeishuHelpdeskTicket.created_at_feishu.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
    ).scalars().all()
    return [serialize_ticket(t) for t in rows], int(total)


# Export-only enum→label maps (verified prod codes; mirror the analytics
# page's statusLabel/typeLabel/solveLabel/scoreLabel so the xlsx reads as
# human labels instead of bare numeric codes). Kept out of serialize_ticket
# so the on-screen table — which does its own i18n mapping — stays raw.
_STATUS_EXPORT_LABELS = {1: "待处理", 2: "处理中", 50: "已解决", 51: "已关闭·未解决", 52: "已关闭·未解决"}
_TYPE_EXPORT_LABELS = {1: "机器人", 2: "人工"}
_SOLVE_EXPORT_LABELS = {1: "已解决", 2: "未解决"}
_SCORE_EXPORT_LABELS = {1: "不满意", 2: "一般", 3: "满意"}


def _labeled_export_row(d: dict[str, Any]) -> dict[str, Any]:
    st = d.get("status")
    d["status"] = _STATUS_EXPORT_LABELS.get(st, f"状态 {st}" if st is not None else "")
    ch = d.get("channel")
    d["channel"] = f"渠道 {ch}" if ch is not None else ""
    tp = d.get("ticket_type")
    d["ticket_type"] = _TYPE_EXPORT_LABELS.get(tp, f"类型 {tp}" if tp is not None else "")
    sv = d.get("solve")
    d["solve"] = _SOLVE_EXPORT_LABELS.get(sv, "")  # 1/2 only; open/unknown → blank
    sc = d.get("score")
    d["score"] = (
        "未评价" if sc in (0, None) else _SCORE_EXPORT_LABELS.get(sc, f"{sc} 分")
    )
    return d


async def all_tickets_for_export(
    db: AsyncSession,
    *,
    integration_id: uuid.UUID | None,
    status: int | None,
    search: str | None,
    start: datetime | None,
    end: datetime | None,
    days: int | None,
    cap: int = 20000,
) -> tuple[list[dict[str, Any]], int]:
    start, end = _window(start, end, days)
    q = _data_query(integration_id, status, search, start, end)
    total = (
        await db.execute(select(func.count()).select_from(q.subquery()))
    ).scalar() or 0
    rows = (
        await db.execute(
            q.order_by(FeishuHelpdeskTicket.created_at_feishu.desc()).limit(cap)
        )
    ).scalars().all()
    return [_labeled_export_row(serialize_ticket(t)) for t in rows], int(total)