duxingchen
ff7c791d49
语义修正: Product.serial_number = 16位HEX系统追溯码(如000000000000000D) Product.external_serial = 业务产品序列号(如25022) 后端: - ProductMessageItem 新增 external_serial 字段 - search_product_messages 关联查询 Product.external_serial - 关键词搜索也支持按 external_serial 匹配 前端显示: 样品升降台V1J · 序列号: 25022 ← 醒目 追溯码: 000000000000000D ← 10px浅灰小字
254 lines
9.2 KiB
Python
254 lines
9.2 KiB
Python
"""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
|
||
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)
|
||
.join(Product, Task.product_id == Product.id)
|
||
.where(Task.status.in_([TASK_STATUS_PENDING, TASK_STATUS_WIP]))
|
||
.limit(limit * 2) # 多取一些，后面 Python 统一排序
|
||
)
|
||
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 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 "",
|
||
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)
|