# ============================================================ # 计划追溯（moduleId: domain-trace, 可重生 ✅） # 对齐聚制云 §4.9：销售订单 → 主数据依据(BOM/路线/产线绑定) → 分解(采购/委外) # → 排产产物(PO/WO) → 负荷占用 → 库存齐套；缺项显式列出。 # 见 docs/product/master-data-and-plan-chain.md # ============================================================ from __future__ import annotations from typing import Any from server.engines.queries import find_bom_items, find_product_lines, find_routing_steps from server.timeutil import parse_dt World = dict[str, Any] def _latest_version(world: World) -> dict | None: versions = world.get("scheduleVersions") or [] return versions[-1] if versions else None def _product(world: World, product_id: int) -> dict | None: return next((m for m in world.get("materials", []) if m["id"] == product_id), None) def plan_trace(world: World, order_no: str | None = None, track: str | None = None) -> dict[str, Any]: """计划追溯入口：按 track 或订单号前缀路由固定/柔性链（OR-06）。""" tr = (track or "").lower().strip() if not tr or tr == "auto": if order_no and re_flex_order(order_no): tr = "flex" else: tr = "fixed" if tr == "flex": return flex_plan_trace(world, order_no) if tr == "both": fixed = plan_trace_fixed(world, None if (order_no and re_flex_order(order_no)) else order_no) flex = flex_plan_trace(world, order_no if (not order_no or re_flex_order(order_no)) else None) return { "track": "both", "fixed": fixed, "flex": flex, "count": fixed.get("count", 0) + flex.get("count", 0), } return plan_trace_fixed(world, order_no) def re_flex_order(order_no: str) -> bool: no = (order_no or "").upper() return no.startswith("FO-") or no.startswith("RUSH-") or "SAP" in no def plan_trace_fixed(world: World, order_no: str | None = None) -> dict[str, Any]: """对一张（或全部活跃）销售订单给出完整计划追溯投影（P0 只读）。""" orders = [so for so in world.get("salesOrders", []) if so.get("status") not in ("CANCELLED",)] if order_no: orders = [so for so in orders if so["orderNo"].lower() == order_no.lower()] if not orders: raise ValueError(f"没找到订单：{order_no}") latest = _latest_version(world) latest_id = latest["id"] if latest else None op_name = {o["id"]: o["name"] for o in world.get("operations", [])} line_name = {ln["id"]: ln["name"] for ln in world.get("lines", [])} mat_by_id = {m["id"]: m for m in world.get("materials", [])} rows: list[dict[str, Any]] = [] for so in orders: items = [it for it in so.get("items", []) if it.get("status") not in ("CANCELLED", "COMPLETED")] if not items: items = so.get("items", [])[:1] item = items[0] if items else {} pid = item.get("productId") product = _product(world, pid) if pid else None qty = item.get("quantity") or 0 # ---- 主数据依据 ---- missing: list[str] = [] bom = next((b for b in world.get("boms", []) if b.get("productId") == pid and b.get("isDefault")), None) routing = next((r for r in world.get("routings", []) if r.get("productId") == pid and r.get("isDefault")), None) if not product: missing.append("成品物料不存在") if not bom: missing.append("无默认 BOM") if not routing: missing.append("无默认工艺路线") line_bindings = find_product_lines(world, pid) if pid else [] if pid and not line_bindings: missing.append("未做产线-产品绑定（排产前必需）") bom_lines = [] for bi in (find_bom_items(world, pid) if pid else []): mat = mat_by_id.get(bi["materialId"], {}) gross = float(bi.get("quantity", 0)) * qty stock = float(mat.get("stock", 0)) transit = float(mat.get("inTransit", 0)) shortage = max(0.0, gross - stock - transit) bom_lines.append({ "materialId": bi["materialId"], "materialCode": mat.get("code", ""), "materialName": mat.get("name", f"#{bi['materialId']}"), "unitQty": bi.get("quantity"), "grossQty": round(gross, 3), "stock": stock, "inTransit": transit, "shortage": round(shortage, 3), "isKeyMaterial": bi.get("isKeyMaterial", False), "procurementLeadTime": mat.get("procurementLeadTime", 0), }) if any(b["shortage"] > 0 for b in bom_lines): missing.append("库存不足以覆盖 BOM 毛需求（见物料缺口）") route_steps = [] for s in (find_routing_steps(world, pid) if pid else []): route_steps.append({ "sequenceNo": s["sequenceNo"], "operationName": op_name.get(s["operationId"], f"OP#{s['operationId']}"), "setupTime": s.get("setupTime"), "runTimePerUnit": s.get("runTimePerUnit"), "isExternal": s.get("isExternal", False), }) # ---- 分解（采购/委外） ---- purchases = [p for p in world.get("purchaseOrders", []) if p.get("salesOrderId") == so["id"]] outsources = [o for o in world.get("outsourceOrders", []) if o.get("salesOrderId") == so["id"]] # ---- 排产产物 ---- pos = [] wos = [] load_by_line: dict[str, float] = {} conflicts = [] if latest_id: pos = [p for p in world.get("productionOrders", []) if p.get("salesOrderId") == so["id"] and p.get("schedulingVersionId") == latest_id] po_ids = {p["id"] for p in pos} for wo in world.get("workOrders", []): if wo.get("productionOrderId") not in po_ids: continue wos.append({ "id": wo["id"], "orderNo": wo.get("orderNo"), "operationName": wo.get("operationName"), "lineId": wo.get("lineId"), "lineName": line_name.get(wo.get("lineId"), ""), "workstationName": wo.get("workstationName"), "start": wo.get("plannedStartTime"), "end": wo.get("plannedEndTime"), "kitStatus": wo.get("kitStatus"), "conflictCount": wo.get("conflictCount", 0), }) try: mins = (parse_dt(wo["plannedEndTime"]) - parse_dt(wo["plannedStartTime"])).total_seconds() / 60 except Exception: mins = 0 ln = line_name.get(wo.get("lineId"), f"#{wo.get('lineId')}") load_by_line[ln] = load_by_line.get(ln, 0.0) + mins conflicts = [c for c in world.get("conflicts", []) if c.get("versionId") == latest_id and ( c.get("orderNo") == so["orderNo"] or c.get("salesOrderId") == so["id"] or any(str(c.get("resourceName", "")).startswith(so["orderNo"]) for _ in [0]) )] # 冲突若未挂订单号，用 PO 关联兜底 if not conflicts and pos: po_nos = {p.get("orderNo") for p in pos} conflicts = [c for c in world.get("conflicts", []) if c.get("versionId") == latest_id and ( c.get("orderNo") in po_nos or c.get("productionOrderId") in po_ids )] if not latest_id: missing.append("尚无排产版本（请先试排）") elif not pos: missing.append("最新版本未包含本订单的生产订单") rows.append({ "order": { "id": so["id"], "orderNo": so["orderNo"], "customerName": so.get("customerName"), "customerLevel": so.get("customerLevel"), "deliveryDate": so.get("deliveryDate"), "priority": so.get("priority"), "status": so.get("status"), "isRush": so.get("isRush", False), "productId": pid, "productCode": item.get("productCode") or (product or {}).get("code"), "productName": item.get("productName") or (product or {}).get("name"), "quantity": qty, "unit": item.get("unit") or (product or {}).get("unit", "件"), }, "master": { "bomVersion": bom.get("version") if bom else None, "routingVersion": routing.get("version") if routing else None, "lineBindings": [{"lineId": lp["lineId"], "lineName": line_name.get(lp["lineId"], ""), "priority": lp.get("priority")} for lp in line_bindings], "bom": bom_lines, "routing": route_steps, }, "decompose": { "purchase": [{ "orderNo": p.get("orderNo"), "materialName": p.get("materialName"), "quantity": p.get("quantity"), "unit": p.get("unit"), "suggestedOrderDate": p.get("suggestedOrderDate"), "requiredDate": p.get("requiredDate"), "status": p.get("status"), "isKeyMaterial": p.get("isKeyMaterial", False), } for p in purchases], "outsource": [{ "orderNo": o.get("orderNo"), "operationName": o.get("operationName"), "quantity": o.get("quantity"), "requiredDate": o.get("requiredDate"), "status": o.get("status"), } for o in outsources], }, "schedule": { "versionNo": latest.get("versionNo") if latest else None, "versionStatus": latest.get("status") if latest else None, "productionOrders": [{ "id": p["id"], "orderNo": p.get("orderNo"), "status": p.get("status"), "plannedStartDate": p.get("plannedStartDate"), "plannedEndDate": p.get("plannedEndDate"), "lineId": p.get("lineId"), "lineName": line_name.get(p.get("lineId"), ""), "conflictCount": p.get("conflictCount", 0), } for p in pos], "workOrders": wos, "loadMinutesByLine": {k: round(v, 1) for k, v in load_by_line.items()}, "conflicts": [{ "type": c.get("conflictType"), "severity": c.get("severity"), "description": c.get("description"), "resourceName": c.get("resourceName"), } for c in conflicts], }, "missing": missing, "readyToSchedule": len([m for m in missing if m not in ( "尚无排产版本（请先试排）", "最新版本未包含本订单的生产订单", "库存不足以覆盖 BOM 毛需求（见物料缺口）", )]) == 0, }) return { "track": "fixed", "latestVersion": { "versionNo": latest.get("versionNo"), "status": latest.get("status"), "woCount": latest.get("woCount"), "poCount": latest.get("poCount"), "conflictCount": latest.get("conflictCount"), } if latest else None, "orders": rows, "count": len(rows), } def flex_plan_trace(world: World, order_no: str | None = None) -> dict[str, Any]: """柔性轨钉扎：FO → flexBom/flexRouting → 虚拟产线 → 工单/设备负荷（OR-06）。""" from server.state.seed import ensure_flex_seed ensure_flex_seed(world) orders = [o for o in world.get("flexOrders", []) if o.get("status") not in ("CANCELLED",)] if order_no: orders = [o for o in orders if o["orderNo"].lower() == order_no.lower()] if not orders: raise ValueError(f"没找到柔性订单：{order_no}") versions = world.get("flexScheduleVersions") or [] latest = versions[-1] if versions else None latest_id = latest["id"] if latest else None mats = {m["code"]: m for m in world.get("flexMaterials", [])} ops = {o["code"]: o for o in world.get("flexOperations", [])} rows: list[dict[str, Any]] = [] for fo in orders: code = fo.get("productCode") or "" qty = int(fo.get("quantity") or 0) missing: list[str] = [] bom_src = [b for b in world.get("flexBom", []) if b.get("productCode") == code] route_src = [r for r in world.get("flexRoutings", []) if r.get("productCode") == code] route_src = sorted(route_src, key=lambda r: r.get("seq") or r.get("sequenceNo") or 0) if not mats.get(code): missing.append("成品不在柔性物料表") if not bom_src: missing.append("无柔性 BOM") if not route_src: missing.append("无柔性工艺路线") bom_lines = [] for bi in bom_src: mat = mats.get(bi["materialCode"], {}) gross = float(bi.get("quantity") or 0) * qty stock = float(mat.get("stock") or 0) transit = float(mat.get("inTransit") or 0) shortage = max(0.0, gross - stock - transit) bom_lines.append({ "materialCode": bi["materialCode"], "materialName": mat.get("name", bi["materialCode"]), "unitQty": bi.get("quantity"), "grossQty": round(gross, 3), "stock": stock, "inTransit": transit, "shortage": round(shortage, 3), "isKeyMaterial": bool(bi.get("isKey")), "consumeOp": bi.get("consumeOp"), "procurementLeadTime": mat.get("procurementLeadTime", 0), }) if any(b["shortage"] > 0 for b in bom_lines): missing.append("库存不足以覆盖柔性 BOM 毛需求") route_steps = [] for s in route_src: op = ops.get(s.get("operationCode") or "", {}) route_steps.append({ "sequenceNo": s.get("seq") or s.get("sequenceNo"), "operationCode": s.get("operationCode"), "operationName": op.get("name") or s.get("operationCode"), "setupTime": s.get("setupTime") or op.get("setupTime") or 0, "runTimePerUnit": s.get("stdTimePerUnit") or s.get("runTimePerUnit") or op.get("stdTimePerUnit") or op.get("runTimePerUnit") or 0, "requireMold": bool(s.get("requireMold")), "isExternal": False, }) # 虚拟产线 + 工单 vls, wos = [], [] load_by_eq: dict[str, float] = {} conflicts = [] if latest_id: vls = [v for v in world.get("flexVirtualLines", []) if v.get("versionId") == latest_id and v.get("orderNo") == fo["orderNo"]] for wo in world.get("flexWorkOrders", []): if wo.get("versionId") != latest_id or wo.get("flexOrderNo") != fo["orderNo"]: continue eq = wo.get("equipmentCode") or "?" wos.append({ "id": wo["id"], "orderNo": fo["orderNo"], "operationCode": wo.get("operationCode"), "operationName": wo.get("operationName") or wo.get("operationCode"), "equipmentCode": eq, "moldCode": wo.get("moldCode"), "start": wo.get("plannedStartTime"), "end": wo.get("plannedEndTime"), "status": wo.get("status"), "progressPct": wo.get("progressPct", 0), "mesExternalId": wo.get("mesExternalId"), "frozen": bool(wo.get("frozen")), }) try: mins = (parse_dt(wo["plannedEndTime"]) - parse_dt(wo["plannedStartTime"])).total_seconds() / 60 except Exception: mins = 0 load_by_eq[eq] = load_by_eq.get(eq, 0.0) + mins conflicts = [c for c in world.get("flexConflicts", []) if c.get("versionId") == latest_id and c.get("orderNo") == fo["orderNo"] and not c.get("isResolved")] if not latest_id: missing.append("尚无柔性排产版本（请先柔性排产）") elif not vls and not wos: missing.append("最新柔性版本未包含本订单") # SAP 链接 sap_link = next((l for l in world.get("sapLinks", []) if l.get("kind") == "order" and l.get("orderNo") == fo["orderNo"]), None) rows.append({ "order": { "id": fo["id"], "orderNo": fo["orderNo"], "customerName": fo.get("productionController") or fo.get("source") or "柔性订单", "customerLevel": "FLEX", "deliveryDate": fo.get("dueDate"), "priority": fo.get("priority"), "status": fo.get("status"), "isRush": bool(fo.get("isRush")), "productId": 0, "productCode": code, "productName": fo.get("productName") or (mats.get(code) or {}).get("name") or code, "quantity": qty, "unit": (mats.get(code) or {}).get("unit", "套"), "externalAufnr": fo.get("externalAufnr"), "source": fo.get("source"), "kitOk": fo.get("kitOk"), }, "master": { "bomVersion": "flex", "routingVersion": "flex", "lineBindings": [], # 柔性无固定产线绑定 "capabilityHint": "按工序能力池选设备（非固定产线）", "bom": bom_lines, "routing": route_steps, }, "decompose": {"purchase": [], "outsource": []}, # 柔性轨暂无 MRP 分解 "schedule": { "versionNo": latest.get("versionNo") if latest else None, "versionStatus": latest.get("status") if latest else None, "sortMode": latest.get("sortMode") if latest else None, "virtualLines": [{ "id": v["id"], "vlNo": v.get("vlNo"), "orderNo": v.get("orderNo"), "plannedStart": v.get("plannedStart"), "plannedEnd": v.get("plannedEnd"), "onTime": v.get("onTime"), } for v in vls], "productionOrders": [], # 柔性无 PO "workOrders": wos, "loadMinutesByEquipment": {k: round(v, 1) for k, v in load_by_eq.items()}, "loadMinutesByLine": {k: round(v, 1) for k, v in load_by_eq.items()}, # UI 复用字段 "conflicts": [{ "type": c.get("conflictType"), "severity": c.get("severity"), "description": c.get("description"), "resourceName": c.get("resourceName"), } for c in conflicts], }, "integration": { "sapAufnr": (sap_link or {}).get("aufnr") or fo.get("externalAufnr"), "mesDispatched": sum(1 for w in wos if w.get("mesExternalId")), "mesCompleted": sum(1 for w in wos if w.get("status") == "COMPLETED" or (w.get("progressPct") or 0) >= 100), }, "missing": missing, "readyToSchedule": len([m for m in missing if m not in ( "尚无柔性排产版本（请先柔性排产）", "最新柔性版本未包含本订单", "库存不足以覆盖柔性 BOM 毛需求", )]) == 0, }) return { "track": "flex", "latestVersion": { "versionNo": latest.get("versionNo"), "status": latest.get("status"), "woCount": latest.get("woCount"), "vlCount": latest.get("vlCount"), "conflictCount": latest.get("conflictCount"), "sortMode": latest.get("sortMode"), } if latest else None, "orders": rows, "count": len(rows), } def summarize_trace(result: dict[str, Any]) -> str: """对话短摘要。""" if result.get("track") == "both": a = summarize_trace(result.get("fixed") or {"orders": [], "count": 0, "track": "fixed"}) b = summarize_trace(result.get("flex") or {"orders": [], "count": 0, "track": "flex"}) return f"【固定轨】\n{a}\n\n【柔性轨】\n{b}" track = result.get("track") or "fixed" track_cn = "柔性钉扎" if track == "flex" else "固定追溯" if not result.get("orders"): return f"{track_cn}：没有可追溯的订单。" ver = result.get("latestVersion") lines = [f"{track_cn}：共 {result['count']} 张订单" + ( f"（最新版本 {ver['versionNo']}）" if ver else "（尚无排产版本）" )] for row in result["orders"][:5]: o = row["order"] miss = "；缺：" + "、".join(row["missing"][:3]) if row["missing"] else "；主数据齐全" wo_n = len(row["schedule"]["workOrders"]) if track == "flex": vl_n = len(row["schedule"].get("virtualLines") or []) integ = row.get("integration") or {} lines.append( f"· {o['orderNo']} {o['productName']}×{o['quantity']} → " f"虚拟产线 {vl_n} / 工单 {wo_n}" + (f" / MES {integ.get('mesDispatched', 0)}" if integ.get("mesDispatched") else "") + miss ) else: pur_n = len(row["decompose"]["purchase"]) lines.append( f"· {o['orderNo']} {o['productName']}×{o['quantity']} → " f"工单 {wo_n} / 采购建议 {pur_n}{miss}" ) if result["count"] > 5: hint = "柔性工作台「钉扎」" if track == "flex" else "订单面板「追溯」" lines.append(f"· …其余 {result['count'] - 5} 张见{hint}") return "\n".join(lines)