track/backend/app/services/dashboard_service.py

"""Dashboard 统计服务"""
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel


class DashboardStats(BaseModel):
    # 产品
    products_total: int
    products_pending: int
    products_in_progress: int
    products_completed: int
    # 任务
    tasks_total: int
    tasks_pending: int
    tasks_in_progress: int
    tasks_completed: int
    tasks_rejected: int
    tasks_rework: int
    # 通知
    unread_notifications: int = 0


class RecentActivity(BaseModel):
    action: str
    task_name: str
    operator: str
    product_sn: str
    time: str
    remark: str | None = None


async def get_dashboard_stats(db: AsyncSession) -> DashboardStats:
    from app.models.product import Product
    from app.models.task import (
        Task, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED,
        TASK_STATUS_REJECTED, TASK_STATUS_ARCHIVED,
    )
    from app.models.notification import Notification

    p_total = await db.scalar(select(func.count(Product.id)))
    p_pending = await db.scalar(select(func.count(Product.id)).where(Product.status == "pending"))
    p_progress = await db.scalar(select(func.count(Product.id)).where(Product.status == "in_progress"))
    p_done = await db.scalar(select(func.count(Product.id)).where(Product.status == "completed"))

    t_total = await db.scalar(select(func.count(Task.id)))
    t_pending = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_PENDING))
    t_progress = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_WIP))
    t_done = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_COMPLETED))
    t_rejected = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_REJECTED))
    t_rework = await db.scalar(select(func.count(Task.id)).where(Task.is_rework.is_(True)))

    unread = await db.scalar(
        select(func.count(Notification.id)).where(Notification.is_read.is_(False))
    )

    return DashboardStats(
        products_total=p_total or 0,
        products_pending=p_pending or 0,
        products_in_progress=p_progress or 0,
        products_completed=p_done or 0,
        tasks_total=t_total or 0,
        tasks_pending=t_pending or 0,
        tasks_in_progress=t_progress or 0,
        tasks_completed=t_done or 0,
        tasks_rejected=t_rejected or 0,
        tasks_rework=t_rework or 0,
        unread_notifications=unread or 0,
    )


async def get_recent_activity(db: AsyncSession, limit: int = 10) -> list[RecentActivity]:
    from app.models.task_log import TaskLog
    from app.models.task import Task
    from app.models.product import Product
    from app.core.time_utils import BEIJING_TZ

    stmt = (
        select(TaskLog, Task.task_name, Product.serial_number)
        .join(Task, TaskLog.task_id == Task.id)
        .join(Product, Task.product_id == Product.id)
        .order_by(TaskLog.created_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    rows = result.all()

    # 收集 operator_id → 批量翻译中文姓名
    operator_ids = list({row[0].operator_id for row in rows if row[0].operator_id})
    name_map: dict[str, str] = {}
    if operator_ids:
        from app.services.mom_cache import get_display_names
        name_map = get_display_names(operator_ids)

    activities: list[RecentActivity] = []
    for log, task_name, product_sn in rows:
        action_label = _action_label(log.action_type)
        # 强制转北京时间显示
        t = log.created_at
        if t:
            if t.tzinfo is None:
                t = t.replace(tzinfo=BEIJING_TZ)
            else:
                t = t.astimezone(BEIJING_TZ)
            time_str = t.strftime("%m-%d %H:%M")
        else:
            time_str = ""
        # operator_id: 优先显示中文姓名 → 英文用户名兜底 → 无记录时为空
        op = log.operator_id
        op_display = name_map.get(op, op or "")
        activities.append(RecentActivity(
            action=action_label,
            task_name=task_name or "",
            operator=op_display,
            product_sn=product_sn or "",
            time=time_str,
            remark=log.remark,
        ))
    return activities


def _action_label(action_type: str) -> str:
    labels = {
        "create": "创建任务",
        "receive": "确认接收",
        "complete": "完成任务",
        "transfer": "完工转交",
        "reject": "品质驳回",
        "end": "结束分支",
        "recall": "撤回转交",
    }
    return labels.get(action_type, action_type)