aps-agent/server/aps_domain/intake_views.py

46 lines
2.9 KiB
Python

"""Safe, complete inspection snapshots for the four planning-data counts."""
from __future__ import annotations
import copy
COLUMNS = {
"orders": [("orderNo", "订单号"), ("productCode", "产品编码"), ("productName", "产品名称"), ("quantity", "数量"),
("dueDate", "交货日期"), ("customerName", "客户"), ("status", "状态")],
"materials": [("code", "编码"), ("name", "名称"), ("spec", "规格"), ("type", "类别"), ("unit", "单位"),
("stock", "库存"), ("inTransit", "在途"), ("expectedArrivalDate", "预计到货")],
"routing": [("productCode", "产品编码"), ("seq", "顺序"), ("operationCode", "工序编码"), ("operationName", "加工步骤"),
("stdTimePerUnit", "单件工时(分钟)"), ("stdTimeSource", "工时来源"), ("isExternal", "是否外协")],
"equipment": [("code", "设备编码"), ("name", "设备名称"), ("capabilities", "可加工工序"), ("zone", "区域"),
("status", "状态"), ("availabilityRate", "可用率")],
}
LABELS = {"orders": "订单明细", "materials": "产品和材料明细", "routing": "加工步骤明细", "equipment": "设备明细"}
def planning_data_views(world: dict, batches: list[dict], *, adopted: bool) -> dict:
if adopted:
records = {key: copy.deepcopy(world.get(table, [])) for key, table in (
("orders", "flexOrders"), ("materials", "flexMaterials"), ("routing", "flexRoutings"), ("equipment", "flexEquipment"))}
else:
roles = {batch.get("role"): batch.get("okRows", []) for batch in batches}
materials = {row.get("code"): copy.deepcopy(row) for role in ("products", "materials") for row in roles.get(role, [])}
for row in roles.get("inventory", []):
target = materials.get(row.get("code"))
if target is not None:
target.update({key: row[key] for key in ("stock", "inTransit", "expectedArrivalDate") if key in row})
records = {"orders": [row for row in roles.get("orders", []) if not row.get("isSandbox")],
"materials": list(materials.values()), "routing": roles.get("routing", []), "equipment": roles.get("equipment", [])}
views = {}
for key, columns in COLUMNS.items():
projected = []
for row in records[key]:
item = {}
for field, _label in columns:
value = row.get(field)
item[field] = "、".join(str(x) for x in value) if isinstance(value, list) else value
ref = row.get("sourceRef") or {}
item["sourceLocation"] = f"{ref.get('sheet', '')} {ref.get('excelRow', '')}".strip() or "已有项目数据"
projected.append(item)
views[key] = {"label": LABELS[key], "columns": [{"key": field, "label": label} for field, label in columns]
+ [{"key": "sourceLocation", "label": "来源行"}], "rows": projected, "total": len(projected)}
return views