wecom_it_smart_desk/backend/app/api/statistics.py

# =============================================================================
# 企微IT智能服务台 — 数据看板 API
# =============================================================================
# 说明：数据统计接口，为管理后台数据看板提供数据支持
#   1. GET /api/admin/stats/overview — 获取整体统计概览
#   2. GET /api/admin/stats/conversations — 会话趋势统计
#   3. GET /api/admin/stats/agents — 坐席绩效统计
#   4. GET /api/admin/stats/satisfaction — 满意度统计
# =============================================================================

import logging
from datetime import datetime, timedelta
from typing import Optional

from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select, and_, or_
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models.agent import Agent
from app.models.conversation import Conversation
from app.models.conversation_evaluation import ConversationEvaluation
from app.models.conversation_annotation import ConversationAnnotation
from app.models.message import Message
from app.utils.response import success_response

from app.api.agents import get_current_agent

logger = logging.getLogger(__name__)

# 创建路由器
router = APIRouter()


# --------------------------------------------------------------------------
# 辅助函数
# --------------------------------------------------------------------------

async def get_date_range(
    start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
) -> tuple[datetime, datetime]:
    """解析日期范围参数。

    Args:
        start_date: 开始日期
        end_date: 结束日期

    Returns:
        tuple: (开始时间, 结束时间)
    """
    if end_date:
        end_dt = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
    else:
        end_dt = datetime.now() + timedelta(days=1)

    if start_date:
        start_dt = datetime.strptime(start_date, "%Y-%m-%d")
    else:
        start_dt = end_dt - timedelta(days=30)  # 默认30天

    return start_dt, end_dt


# --------------------------------------------------------------------------
# GET /api/admin/stats/overview — 整体统计概览
# --------------------------------------------------------------------------
@router.get("/admin/stats/overview")
async def get_overview_stats(
    start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
    db: AsyncSession = Depends(get_db),
    admin: Agent = Depends(get_current_agent),
):
    """获取整体统计概览。

    包含：总会话数、待处理会话数、已解决会话数、平均响应时间、满意度等。

    Args:
        start_date: 开始日期
        end_date: 结束日期
        db: 数据库会话
        admin: 当前管理员

    Returns:
        Dict: 整体统计数据
    """
    start_dt, end_dt = await get_date_range(start_date, end_date)

    # 总会话数
    stmt_total = select(func.count(Conversation.id)).where(
        and_(
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_total)
    total_conversations = result.scalar() or 0

    # 待处理会话数（状态为 queued 或 serving）
    stmt_pending = select(func.count(Conversation.id)).where(
        and_(
            Conversation.status.in_(["queued", "serving"]),
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_pending)
    pending_conversations = result.scalar() or 0

    # 已解决会话数（状态为 resolved）
    stmt_resolved = select(func.count(Conversation.id)).where(
        and_(
            Conversation.status == "resolved",
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_resolved)
    resolved_conversations = result.scalar() or 0

    # 计算满意度（已评价会话的平均评分）
    stmt_satisfaction = select(
        func.avg(ConversationEvaluation.score),
        func.count(ConversationEvaluation.id),
    ).join(
        Conversation,
        ConversationEvaluation.conversation_id == Conversation.id,
    ).where(
        and_(
            ConversationEvaluation.created_at >= start_dt,
            ConversationEvaluation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_satisfaction)
    satisfaction_row = result.first()
    avg_satisfaction = float(satisfaction_row[0]) if satisfaction_row[0] else 0.0
    evaluated_count = satisfaction_row[1] or 0

    # 计算平均响应时间（第一条坐席消息与第一条消息的时间差）
    # 简化计算：resolved会话的平均解决时长
    stmt_duration = select(func.avg(
        func.extract('epoch', Conversation.updated_at) - func.extract('epoch', Conversation.created_at)
    )).where(
        and_(
            Conversation.status == "resolved",
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_duration)
    avg_duration_seconds = result.scalar() or 0
    avg_duration_minutes = avg_duration_seconds / 60 if avg_duration_seconds else 0

    data = {
        "total_conversations": total_conversations,
        "pending_conversations": pending_conversations,
        "resolved_conversations": resolved_conversations,
        "resolution_rate": round(resolved_conversations / total_conversations * 100, 1) if total_conversations > 0 else 0,
        "avg_satisfaction": round(avg_satisfaction, 2),
        "evaluated_count": evaluated_count,
        "avg_duration_minutes": round(avg_duration_minutes, 1),
    }

    return success_response(data=data)


# --------------------------------------------------------------------------
# GET /api/admin/stats/conversations — 会话趋势统计
# --------------------------------------------------------------------------
@router.get("/admin/stats/conversations")
async def get_conversation_stats(
    start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
    db: AsyncSession = Depends(get_db),
    admin: Agent = Depends(get_current_agent),
):
    """获取会话趋势统计。

    按天统计每日会话数、解决数。

    Args:
        start_date: 开始日期
        end_date: 结束日期
        db: 数据库会话
        admin: 当前管理员

    Returns:
        Dict: 趋势数据列表
    """
    start_dt, end_dt = await get_date_range(start_date, end_date)

    # 按天统计会话数
    stmt = select(
        func.date(Conversation.created_at).label("date"),
        func.count(Conversation.id).label("total"),
    ).where(
        and_(
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    ).group_by(
        func.date(Conversation.created_at)
    ).order_by(
        func.date(Conversation.created_at)
    )

    result = await db.execute(stmt)
    rows = result.all()

    # 转换为日期+统计的格式
    trend_data = []
    for row in rows:
        date_val = row.date
        if isinstance(date_val, datetime):
            date_str = date_val.strftime("%Y-%m-%d")
        else:
            date_str = str(date_val)

        trend_data.append({
            "date": date_str,
            "total": row.total,
        })

    return success_response(data={"items": trend_data})


# --------------------------------------------------------------------------
# GET /api/admin/stats/agents — 坐席绩效统计
# --------------------------------------------------------------------------
@router.get("/admin/stats/agents")
async def get_agent_stats(
    start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
    db: AsyncSession = Depends(get_db),
    admin: Agent = Depends(get_current_agent),
):
    """获取坐席绩效统计。

    统计各坐席的处理会话数、解决数、平均响应时间。

    Args:
        start_date: 开始日期
        end_date: 结束日期
        db: 数据库会话
        admin: 当前管理员

    Returns:
        Dict: 坐席绩效列表
    """
    start_dt, end_dt = await get_date_range(start_date, end_date)

    # 统计各坐席的会话数
    stmt = select(
        Conversation.assigned_agent_id,
        func.count(Conversation.id).label("total"),
        func.sum(
            func.case((Conversation.status == "resolved", 1), else_=0)
        ).label("resolved"),
    ).where(
        and_(
            Conversation.assigned_agent_id.isnot(None),
            Conversation.created_at >= start_dt,
            Conversation.created_at < end_dt,
        )
    ).group_by(
        Conversation.assigned_agent_id
    )

    result = await db.execute(stmt)
    rows = result.all()

    # 获取坐席信息
    agent_ids = [row[0] for row in rows if row[0]]
    agent_stmt = select(Agent.id, Agent.name).where(Agent.id.in_(agent_ids))
    agent_result = await db.execute(agent_stmt)
    agent_map = {a.id: a.name for a in agent_result.scalars().all()}

    # 转换为坐席绩效数据
    agent_data = []
    for row in rows:
        if not row[0]:
            continue
        agent_id = row[0]
        agent_data.append({
            "agent_id": agent_id,
            "agent_name": agent_map.get(agent_id, "未知"),
            "total_conversations": row[1],
            "resolved_conversations": row[2] or 0,
            "resolution_rate": round((row[2] or 0) / row[1] * 100, 1) if row[1] > 0 else 0,
        })

    # 按处理数排序
    agent_data.sort(key=lambda x: x["total_conversations"], reverse=True)

    return success_response(data={"items": agent_data})


# --------------------------------------------------------------------------
# GET /api/admin/stats/satisfaction — 满意度统计
# --------------------------------------------------------------------------
@router.get("/admin/stats/satisfaction")
async def get_satisfaction_stats(
    start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"),
    db: AsyncSession = Depends(get_db),
    admin: Agent = Depends(get_current_agent),
):
    """获取满意度统计。

    统计评分分布、各表情占比。

    Args:
        start_date: 开始日期
        end_date: 结束日期
        db: 数据库会话
        admin: 当前管理员

    Returns:
        Dict: 满意度统计数据
    """
    start_dt, end_dt = await get_date_range(start_date, end_date)

    # 评分分布统计
    stmt = select(
        ConversationEvaluation.score,
        func.count(ConversationEvaluation.id).label("count"),
    ).join(
        Conversation,
        ConversationEvaluation.conversation_id == Conversation.id,
    ).where(
        and_(
            ConversationEvaluation.created_at >= start_dt,
            ConversationEvaluation.created_at < end_dt,
        )
    ).group_by(
        ConversationEvaluation.score
    )

    result = await db.execute(stmt)
    rows = result.all()

    # 评分分布
    score_distribution = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
    for row in rows:
        if row[0] in score_distribution:
            score_distribution[row[0]] = row[1]

    # 表情分布
    stmt_emoji = select(
        ConversationEvaluation.emoji,
        func.count(ConversationEvaluation.id).label("count"),
    ).join(
        Conversation,
        ConversationEvaluation.conversation_id == Conversation.id,
    ).where(
        and_(
            ConversationEvaluation.created_at >= start_dt,
            ConversationEvaluation.created_at < end_dt,
            ConversationEvaluation.emoji.isnot(None),
        )
    ).group_by(
        ConversationEvaluation.emoji
    )

    result = await db.execute(stmt_emoji)
    emoji_rows = result.all()

    emoji_distribution = {}
    for row in emoji_rows:
        if row[0]:
            emoji_distribution[row[0]] = row[1]

    # 计算平均分
    stmt_avg = select(func.avg(ConversationEvaluation.score)).join(
        Conversation,
        ConversationEvaluation.conversation_id == Conversation.id,
    ).where(
        and_(
            ConversationEvaluation.created_at >= start_dt,
            ConversationEvaluation.created_at < end_dt,
        )
    )
    result = await db.execute(stmt_avg)
    avg_score = result.scalar() or 0

    data = {
        "avg_score": round(float(avg_score), 2),
        "total_evaluated": sum(score_distribution.values()),
        "score_distribution": [
            {"score": k, "count": v} for k, v in sorted(score_distribution.items())
        ],
        "emoji_distribution": [
            {"emoji": k, "count": v} for k, v in emoji_distribution.items()
        ],
    }

    return success_response(data=data)