duxingchen/track
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 宏观状态统一(待仓库收货/已入库)废弃"在库"; WIP矩阵返回product_name并新增下钻明细

Browse Source
This commit is contained in:
duxingchen
2026-09-02 10:38:49 +08:00
parent e194b97649
commit 3019653dfa
6 changed files with 199 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
backend/app/services/dashboard_service.py
View File

@ -82,11 +82,28 @@ class OperationDetail(BaseModel):
class WipMatrixRow(BaseModel):
spec_model: str # 规格型号（Y 轴）
product_name: str = "" # 产品名称（物料名称，用于首列复合显示）
dimension_key: str # 人员姓名 或 工序名称（X 轴）
count: int = 0 # 该交叉点的设备数量
assignees: list[str] = [] # 该交叉点涉及的主负责人（中文名，去重）
class WipMatrixDetailRow(BaseModel):
"""WIP 矩阵单元格下钻 — 某规格型号 × 某工序下的设备明细"""
product_id: str
serial_number: str # 16位HEX身份证
external_serial: str | None = None # 业务序列号
material_name: str = "" # 产品名称
spec_model: str = ""
task_status: str = "" # 当前状态 WIP/PENDING/COMPLETED/ARCHIVED/OUTBOUND
assignee_id: str = ""
assignee: str = "" # 负责人中文名
duration_hours: float = 0.0 # 已在该工序滞留时长（小时）
total_hours: float = 0.0 # 整个项目总时长（自然小时，最早介入→最晚结束）
work_hours: float = 0.0 # 仅工作日有效时长（小时，排除周末/节假日）
received_at: str | None = None # 接手时间 MM-DD HH:mm
class PersonDevice(BaseModel):
product_id: str
serial_number: str # 16位HEX身份证
@ -631,8 +648,8 @@ async def get_wip_matrix(
1. 取该设备最新的一条主分支任务（parent_task_id IS NULL 或 TRANSFER/RECOVERY）
2. 设备当前工序 = 该最新主任务的工序名（task_name），不区分状态
- 「待确认」(PENDING) = 别人转给我但未接收
- 「在库」(COMPLETED) = 真正入库/生产完成
- 其他已完成工序（如「测试」完成）按原工序显示，不强制归「在库」
- 「已入库」= MOM 已扫码实收；「已完成」= 车间完工待实收
- 其他已完成工序（如「测试」完成）按原工序显示，不强制归仓库态
这样一台设备在「上一步已完成 + 下一步待确认」时只算一次（待确认），不会重复计数。
since/until 按设备最新主任务的创建时间过滤。
@ -650,6 +667,7 @@ async def get_wip_matrix(
select(
Product.id,
Product.spec_model,
Product.material_name,
Task.task_name,
Task.assignee_id,
Task.created_at,
@ -671,7 +689,7 @@ async def get_wip_matrix(
# 每台设备 → (spec, 当前工序/负责人, 当前负责人ID)
device_cur: dict[str, tuple] = {}
seen: set[str] = set()
for pid, spec, task_name, assignee, created, loc, overall_status, product_status in rows:
for pid, spec, product_name, task_name, assignee, created, loc, overall_status, product_status in rows:
if pid in seen:
continue
seen.add(pid)
@ -691,24 +709,27 @@ async def get_wip_matrix(
# 待仓库收货 = 完工已转交仓库、MOM 尚未扫码实收
# 否则按最新主任务工序名（活跃/完成态都归该工序）
if loc == "virtual_warehouse":
# 统一术语：待仓库收货(车间完工) → "已完成"；扫码实收 → "已入库"；已出库保持
if overall_status in ("已入库", "在库") or str(product_status).upper() == "ARCHIVED":
key = "已入库"
elif overall_status == "已出库" or str(product_status).upper() == "OUTBOUND":
key = "已出库"
else:
key = "待仓库收货"
key = "已完成"
else:
key = task_name or "—"
else:
key = assignee or "未分配"
device_cur[pid] = (spec or "未知型号", key, assignee or "")
device_cur[pid] = (spec or "未知型号", key, assignee or "", product_name or "")
# 聚合：规格 × 当前工序 → 设备数；同时收集负责人
# 聚合：规格 × 当前工序 → 设备数；同时收集负责人 与 产品名称
agg: dict[tuple, int] = {}
assignee_map: dict[tuple, set] = {}
for spec, key, assignee in device_cur.values():
product_name_map: dict[tuple, str] = {}
for spec, key, assignee, product_name in device_cur.values():
k = (spec, key)
agg[k] = agg.get(k, 0) + 1
product_name_map.setdefault(k, product_name)
if assignee:
assignee_map.setdefault(k, set()).add(assignee)
@ -727,6 +748,7 @@ async def get_wip_matrix(
assignees = [name_map.get(a, a) for a in assignee_map.get((spec, key), set())] or []
items.append(WipMatrixRow(
spec_model=spec,
product_name=product_name_map.get((spec, key), ""),
dimension_key=dim_display,
count=cnt,
assignees=assignees,
@ -735,6 +757,149 @@ async def get_wip_matrix(
return items
# ============================================================
# WIP 矩阵单元格下钻 — 某规格型号 × 某工序下的设备明细
# ============================================================
async def get_wip_matrix_detail(
db: AsyncSession,
spec_model: str,
process: str,
since: datetime | None = None,
until: datetime | None = None,
) -> list[WipMatrixDetailRow]:
"""WIP 矩阵单元格下钻：返回该 规格型号×工序 交叉点下的设备明细。
口径与 get_wip_matrix 完全一致：每台设备取最新一条主分支任务，
按其在库三态（已完成/已入库/已出库）或工序名归类到 process。
"""
from datetime import timezone as dt_timezone
from app.models.task import Task
from app.models.product import Product
from app.models.holiday import Holiday
from app.core.time_utils import get_beijing_time, to_beijing, working_duration_hours
hres = await db.execute(select(Holiday.day))
holidays = {r[0] for r in hres}
result = await db.execute(
select(
Product.id,
Product.serial_number,
Product.external_serial,
Product.material_name,
Product.spec_model,
Product.current_location_id,
Product.overall_status,
Product.status,
Task.task_name,
Task.assignee_id,
Task.status.label("task_status"),
Task.created_at,
Task.received_at,
)
.join(Task, Task.product_id == Product.id)
.where(
or_(
Task.parent_task_id.is_(None),
Task.task_type.in_(["TRANSFER", "RECOVERY"]),
)
)
.order_by(Product.id, Task.created_at.desc())
)
rows = result.all()
now = get_beijing_time()
seen: set[str] = set()
matched: list[WipMatrixDetailRow] = []
raw_names: set[str] = set()
for pid, serial, ext, mat_name, spec, loc, overall, pstatus, task_name, assignee, tstatus, created, received in rows:
if pid in seen:
continue
seen.add(pid)
if spec_model and (spec or "") != spec_model:
continue
# 时间筛选（设备最新主任务创建时间）
if created is not None and created.tzinfo is None:
created = created.replace(tzinfo=dt_timezone.utc)
if since and created is not None and created < since:
continue
if until and created is not None and created > until:
continue
# 当前工序归类（与 get_wip_matrix 一致）
if loc == "virtual_warehouse":
if overall in ("已入库", "在库") or str(pstatus).upper() == "ARCHIVED":
key = "已入库"
elif overall == "已出库" or str(pstatus).upper() == "OUTBOUND":
key = "已出库"
else:
key = "已完成"
else:
key = task_name or "—"
if process and key != process:
continue
# 滞留时长（当前工序接手时间 → 现在）
start = received or created
duration = 0.0
received_str = None
if start:
start_bj = to_beijing(start)
duration = working_duration_hours(start_bj, now, holidays)
received_str = start_bj.strftime("%m-%d %H:%M")
matched.append(WipMatrixDetailRow(
product_id=str(pid),
serial_number=serial or "",
external_serial=ext,
material_name=mat_name or "",
spec_model=spec or "",
task_status=tstatus or "",
assignee_id=assignee or "",
duration_hours=duration,
received_at=received_str,
))
if assignee:
raw_names.add(assignee)
# 负责人中文名映射
name_map: dict[str, str] = {}
if raw_names:
from app.services.mom_cache import get_display_names
name_map = get_display_names(list(raw_names))
for r in matched:
r.assignee = name_map.get(r.assignee_id, r.assignee_id)
# ── 整个项目总时长：设备最早介入 → 最晚结束（自然 + 工作日） ──
if matched:
import uuid as uuid_mod
from sqlalchemy import func as sa_func
uuid_list = [uuid_mod.UUID(m.product_id) for m in matched]
range_rows = (await db.execute(
select(
Task.product_id,
sa_func.min(sa_func.coalesce(Task.received_at, Task.created_at)).label("t0"),
sa_func.max(sa_func.coalesce(Task.completed_at, now)).label("tmax"),
)
.where(Task.product_id.in_(uuid_list))
.group_by(Task.product_id)
)).all()
row_by_pid = {m.product_id: m for m in matched}
for pid, t0, tmax in range_rows:
row = row_by_pid.get(str(pid))
if row is None or t0 is None or tmax is None:
continue
row.total_hours = round((tmax - t0).total_seconds() / 3600, 1)
row.work_hours = round(working_duration_hours(to_beijing(t0), to_beijing(tmax), holidays), 1)
return matched
# ============================================================
# 人员操作明细（点击数字下钻 — 接收/转交/上传备注）
# ============================================================