track/backend/app/services/dashboard_service.py

"""Dashboard 统计服务 — 上帝视角（全厂全系统数据，不按用户过滤）"""
from datetime import datetime
from sqlalchemy import select, func, or_
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel


# ============================================================
# Schemas
# ============================================================

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
    unread_messages: int = 0


class WipTask(BaseModel):
    task_id: str
    task_name: str
    assignee: str
    product_sn: str               # 16位HEX身份证
    external_serial: str | None   # 业务序列号
    material_name: str            # 设备名称
    spec_model: str               # 规格型号
    status: str
    received_at: str
    duration_hours: float


class ProductMessageItem(BaseModel):
    id: str
    content: str
    operator_name: str            # 留言人中文姓名
    product_sn: str               # 16位HEX系统追溯码
    external_serial: str | None   # 业务产品序列号（如 25022）
    material_name: str            # 物料名称
    created_at: str               # ISO时间字符串


class ProductMessageList(BaseModel):
    items: list[ProductMessageItem]
    total: int


# ============================================================
# 看板统计（时间快照语义）
# ============================================================

async def get_dashboard_stats(
    db: AsyncSession,
    since: datetime | None = None,
    until: datetime | None = None,
) -> DashboardStats:
    """
    上帝视角 — 全厂全系统统计。

    时间筛选规则：
      - PENDING / WIP / 总数：永远忽略时间筛选，返回实时快照。
      - COMPLETED / REJECTED / 完成率：严格按 since~until 过滤（用于时段报表）。
    """
    from app.models.product import Product
    from app.models.task import (
        Task, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED,
        TASK_STATUS_REJECTED,
    )
    from app.models.notification import Notification
    from app.models.message import ProductMessage

    # ── 产品（实时快照，不过滤） ──
    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"))

    # ── 任务实时快照（PENDING/WIP/返工 — 永远不过滤） ──
    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_rework = await db.scalar(select(func.count(Task.id)).where(Task.is_rework.is_(True)))

    # ── 任务已完成/驳回（时间可过滤） ──
    t_done_q = select(func.count(Task.id)).where(Task.status == TASK_STATUS_COMPLETED)
    t_rej_q = select(func.count(Task.id)).where(Task.status == TASK_STATUS_REJECTED)
    if since:
        t_done_q = t_done_q.where(Task.completed_at >= since)
        t_rej_q = t_rej_q.where(Task.completed_at >= since)
    if until:
        t_done_q = t_done_q.where(Task.completed_at <= until)
        t_rej_q = t_rej_q.where(Task.completed_at <= until)
    t_done = await db.scalar(t_done_q)
    t_rejected = await db.scalar(t_rej_q)

    # ── 任务总数 = 实时快照段 + 时间过滤段（确保进度条段总和=总数） ──
    t_total = (t_pending or 0) + (t_progress or 0) + (t_done or 0) + (t_rejected or 0)

    # ── 通知 & 留言（实时快照） ──
    unread_notif = await db.scalar(
        select(func.count(Notification.id)).where(Notification.is_read.is_(False))
    )
    unread_msg = await db.scalar(select(func.count(ProductMessage.id)))

    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_notif or 0,
        unread_messages=unread_msg or 0,
    )


# ============================================================
# 在制品看板（永远实时）
# ============================================================

async def get_wip_tasks(db: AsyncSession, limit: int = 20) -> list[WipTask]:
    from app.models.task import Task, TASK_STATUS_PENDING, TASK_STATUS_WIP
    from app.models.product import Product
    from app.core.time_utils import get_beijing_time, BEIJING_TZ

    stmt = (
        select(Task, Product.serial_number, Product.external_serial, Product.material_name, Product.spec_model)
        .join(Product, Task.product_id == Product.id)
        .where(Task.status.in_([TASK_STATUS_PENDING, TASK_STATUS_WIP]))
        .limit(limit * 2)
    )
    result = await db.execute(stmt)
    rows = result.all()

    raw_ids = list({t.assignee_id for t, *_ in rows if t.assignee_id})
    name_map: dict[str, str] = {}
    if raw_ids:
        from app.services.mom_cache import get_display_names
        name_map = get_display_names(raw_ids)

    now = get_beijing_time()
    wip_list: list[WipTask] = []
    for task, product_sn, ext_sn, mat_name, spec in rows:
        start = task.received_at or task.created_at
        if start:
            if start.tzinfo is None:
                start = start.replace(tzinfo=BEIJING_TZ)
            else:
                start = start.astimezone(BEIJING_TZ)
            hours = round((now - start).total_seconds() / 3600, 1)
            recv_str = start.strftime("%m-%d %H:%M")
        else:
            hours = 0
            recv_str = ""

        wip_list.append(WipTask(
            task_id=str(task.id),
            task_name=task.task_name,
            assignee=name_map.get(task.assignee_id or "", task.assignee_id or "未分配"),
            product_sn=product_sn or "",
            external_serial=ext_sn or None,
            material_name=mat_name or "",
            spec_model=spec or "",
            status=task.status,
            received_at=recv_str,
            duration_hours=hours,
        ))
    # 统一按滞留时间降序排列（无视 status，纯数值排序）
    wip_list.sort(key=lambda t: t.duration_hours, reverse=True)
    return wip_list[:limit]


# ============================================================
# 协同留言搜索（上帝视角 — 全厂）
# ============================================================

async def search_product_messages(
    db: AsyncSession,
    keyword: str = "",
    skip: int = 0,
    limit: int = 50,
) -> ProductMessageList:
    """
    上帝视角 — 全厂所有产品的协同留言。

    关联链: ProductMessage → Product → (material_name, serial_number)
    搜索支持: SN码、物料名称、留言人
    排序: created_at 倒序（最新在前）
    """
    from app.models.message import ProductMessage
    from app.models.product import Product
    from app.core.time_utils import BEIJING_TZ

    # 基础查询 — 同时取出 16位追溯码 + 业务序列号
    stmt = (
        select(ProductMessage, Product.serial_number, Product.material_name, Product.external_serial)
        .join(Product, ProductMessage.product_id == Product.id)
    )

    # 关键词搜索
    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        stmt = stmt.where(or_(
            Product.serial_number.ilike(kw),
            Product.external_serial.ilike(kw),
            Product.material_name.ilike(kw),
            ProductMessage.operator_id.ilike(kw),
            ProductMessage.content.ilike(kw),
        ))

    # 总数
    count_stmt = select(func.count()).select_from(stmt.subquery())
    total = await db.scalar(count_stmt) or 0

    # 分页 + 排序
    stmt = stmt.order_by(ProductMessage.created_at.desc()).offset(skip).limit(limit)
    result = await db.execute(stmt)
    rows = result.all()

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

    items: list[ProductMessageItem] = []
    for msg, sn, mat_name, ext_sn in rows:
        t = msg.created_at
        if t:
            if t.tzinfo is None:
                # 旧数据（datetime.utcnow）：naive UTC → 转北京时间
                from datetime import timezone as dt_timezone
                t = t.replace(tzinfo=dt_timezone.utc).astimezone(BEIJING_TZ)
            else:
                t = t.astimezone(BEIJING_TZ)
        time_str = t.isoformat() if t else ""

        items.append(ProductMessageItem(
            id=str(msg.id),
            content=msg.content,
            operator_name=name_map.get(msg.operator_id, msg.operator_id),
            product_sn=sn or "",
            external_serial=ext_sn or None,
            material_name=mat_name or "",
            created_at=time_str,
        ))

    return ProductMessageList(items=items, total=total)