"""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)