fix(透视表): 每台设备只按当前工序统计一次,修复待确认重复计数

- 取每台设备最新主任务作为当前状态
- 活跃任务(WIP/PENDING)→归属对应工序(待确认=转交未接收)
- 完成态任务→归属「在库」(生产完成)
- 上一步已完成+下一步待确认时只算一次,不再重复
- 时间筛选按设备最新主任务创建时间
This commit is contained in:
2026-08-28 16:54:43 +08:00
parent 0dcdb5dad7
commit b57ae19b44

View File

@ -612,65 +612,95 @@ async def get_wip_matrix(
) -> list[WipMatrixRow]:
"""生产分布透视表Y=规格型号X=人员 或 工序,单元格=设备数量。
覆盖全部任务状态(含已完成/在库),工序分布能看到「在库」(生产完成)。
只统计主分支parent_task_id IS NULL 或 TRANSFER/RECOVERY,避免协助分支/返工重复计数。
since/until 按任务接手/创建时间过滤。
核心口径:**每台设备只统计一次**,按它「当前所处工序」归属——
1. 取该设备最新的一条主分支任务parent_task_id IS NULL 或 TRANSFER/RECOVERY
2. 若该任务是活跃的WIP/PENDING→ 归属对应工序(如「待确认」= 别人转给我但未接收)
3. 若最新任务是完成态COMPLETED/ARCHIVED→ 归属「在库」(生产完成)
这样一台设备在「上一步已完成 + 下一步待确认」时只算一次(待确认),不会重复计数。
since/until 按设备最新主任务的创建时间过滤。
dimension:
- assignee: 按任务负责人聚合dimension_key 为中文姓名)
- task_name: 按工序聚合dimension_key 为工序名,附主负责人)
- assignee: 按当前任务负责人聚合dimension_key 为中文姓名)
- task_name: 按当前工序聚合dimension_key 为工序名,附主负责人)
"""
from app.models.task import Task
from datetime import timezone as dt_timezone
from app.models.task import Task, TASK_STATUS_WIP, TASK_STATUS_PENDING
from app.models.product import Product
dim_expr = Task.task_name if dimension == "task_name" else Task.assignee_id
stmt = (
# 每台设备按主任务创建时间倒序,取第一条即「最新主任务」
result = await db.execute(
select(
Product.id,
Product.spec_model,
dim_expr,
func.count(Product.id),
func.array_agg(func.distinct(Task.assignee_id)),
Task.task_name,
Task.status,
Task.assignee_id,
Task.created_at,
)
.join(Task, Task.product_id == Product.id)
# 🔧 只统计主分支(主线任务),排除 SPAWN 协助分支
.where(
or_(
Task.parent_task_id.is_(None),
Task.task_type.in_(["TRANSFER", "RECOVERY"]),
)
)
.group_by(Product.spec_model, dim_expr)
.order_by(Product.spec_model, dim_expr)
.order_by(Product.id, Task.created_at.desc())
)
if since:
stmt = stmt.where(func.coalesce(Task.received_at, Task.created_at) >= since)
if until:
stmt = stmt.where(func.coalesce(Task.received_at, Task.created_at) <= until)
result = await db.execute(stmt)
rows = result.all()
# 收集负责人 ID → 中文名
# 每台设备 → (spec, 当前工序/负责人, 当前负责人ID)
device_cur: dict[str, tuple] = {}
seen: set[str] = set()
for pid, spec, task_name, status, assignee, created in rows:
if pid in seen:
continue
seen.add(pid)
# 时间筛选:设备最新主任务的创建时间
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
if dimension == "task_name":
# 活跃 → 当前工序;完成态 → 在库(生产完成)
key = task_name if status in (TASK_STATUS_WIP, TASK_STATUS_PENDING) else "在库"
else:
key = assignee or "未分配"
device_cur[pid] = (spec or "未知型号", key, assignee or "")
# 聚合:规格 × 当前工序 → 设备数;同时收集负责人
agg: dict[tuple, int] = {}
assignee_map: dict[tuple, set] = {}
for spec, key, assignee in device_cur.values():
k = (spec, key)
agg[k] = agg.get(k, 0) + 1
if assignee:
assignee_map.setdefault(k, set()).add(assignee)
# 负责人 ID → 中文名
raw_ids: set[str] = set()
for r in rows:
for aid in (r[3] or []):
if aid:
raw_ids.add(aid)
for s in assignee_map.values():
raw_ids |= s
name_map: dict[str, str] = {}
if raw_ids:
from app.services.mom_cache import get_display_names
name_map = get_display_names(list(raw_ids))
items: list[WipMatrixRow] = []
for spec, dim_key, cnt, assignee_ids in rows:
dim_display = name_map.get(dim_key or "", dim_key or "未分配") if dimension == "assignee" else (dim_key or "")
assignees = [name_map.get(a, a) for a in (assignee_ids or []) if a] or []
for (spec, key), cnt in agg.items():
dim_display = name_map.get(key, key) if dimension == "assignee" else key
assignees = [name_map.get(a, a) for a in assignee_map.get((spec, key), set())] or []
items.append(WipMatrixRow(
spec_model=spec or "未知型号",
spec_model=spec,
dimension_key=dim_display,
count=cnt or 0,
count=cnt,
assignees=assignees,
))
items.sort(key=lambda x: (x.spec_model, x.dimension_key))
return items