From 5f983fb2c733006b7d60997901b2ab6b495a9e65 Mon Sep 17 00:00:00 2001 From: "z.zhang" Date: Fri, 9 Oct 2026 15:33:33 +0800 Subject: [PATCH] =?UTF-8?q?fix(flex):=20=E5=AF=B9=E8=AF=9D=E6=96=B0?= =?UTF-8?q?=E5=BB=BA=E6=80=A5=E5=8D=95=E6=8E=92=E4=B8=8D=E8=BF=9B=E6=9F=94?= =?UTF-8?q?=E6=80=A7=E6=8E=92=E4=BA=A7=E2=80=94=E2=80=94=E4=B8=BB=E5=B9=B2?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=E6=A1=A5=E6=8E=A5=20+=20=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E8=AE=A1=E5=88=92=E5=91=98=E5=8D=95=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 根因1：data.import/order.upsert 只写主干 salesOrders，PoolEngine 只读 flexOrders，导致计划员对话新建的急单在重排时完全不可见 - 根因2：新建订单强制使用 SO+日期 自动单号，计划员指定的 RUSH-0007 被丢弃，后续按单号查询/引用全部落空 - 新增 sync_sales_orders_to_flex：试排世界每次柔性排产前把可排主干单 （APPROVED/CONFIRMED）幂等投影为 RELEASED 柔性单；主干取消/完成/ 退审同步回撤出可排集合；source=SALES 标记防止 flex→sales 回投重复 - intake/order.upsert 接受并保留指定 orderNo，双池查重 fail-closed - 作用域收敛：仅 trial 试排世界桥接，演示厂维持经典/柔性双轨隔离 - 验证：锐扬资料世界五场景（排入/查重/幂等/回撤/兼容）+ golden 全量 2378 通过 0 失败 --- server/aps_domain/flex.py | 6 ++- server/aps_domain/intake.py | 7 ++++ server/aps_domain/orders.py | 79 ++++++++++++++++++++++++++++++++++++- 3 files changed, 89 insertions(+), 3 deletions(-) diff --git a/server/aps_domain/flex.py b/server/aps_domain/flex.py index be37b8a..a463408 100644 --- a/server/aps_domain/flex.py +++ b/server/aps_domain/flex.py @@ -249,7 +249,7 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int] from server.agent_core.audit import write_audit from server.aps_domain.closed_loop_runtime import run_closed_loop_candidate from server.aps_domain.mrp import decompose_orders - from server.aps_domain.orders import sync_flex_orders_to_sales + from server.aps_domain.orders import sync_flex_orders_to_sales, sync_sales_orders_to_flex from server.aps_domain.sourcing import annotate_world_sourcing from server.state.seed import ensure_flex_seed @@ -279,6 +279,10 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int] def execute_pipeline() -> dict: if explicit_demo or trial: + # 试排世界：主干对话/手工单（含紧急插单）桥接进柔性订单池，否则排产看不到它们； + # 演示厂保持经典轨/柔性轨双轨隔离，不桥接 + if trial: + sync_sales_orders_to_flex(store.data) result = PoolEngine().solve( store.data, store.next_id, diff --git a/server/aps_domain/intake.py b/server/aps_domain/intake.py index 9407f88..97d365e 100644 --- a/server/aps_domain/intake.py +++ b/server/aps_domain/intake.py @@ -62,6 +62,12 @@ def _normalize_order_row(world: World, row: Any, index: int) -> tuple[dict[str, if level not in CUSTOMER_LEVELS: return None, f"第{index}行 customerLevel 必须是 VIP/A/B/C" is_rush = bool(row.get("isRush", False)) + order_no = str(row.get("orderNo") or "").strip() + if order_no: + folded = order_no.casefold() + for pool in ("salesOrders", "flexOrders"): + if any(str(o.get("orderNo") or "").casefold() == folded for o in (world.get(pool) or [])): + return None, f"第{index}行 orderNo「{order_no}」已存在" priority = row.get("priority") if priority is None: priority = 1 if is_rush else 5 @@ -73,6 +79,7 @@ def _normalize_order_row(world: World, row: Any, index: int) -> tuple[dict[str, return { "customerName": customer, "customerLevel": level, + "orderNo": order_no, "productId": int(product["id"]), "productCode": product["code"], "productName": product["name"], diff --git a/server/aps_domain/orders.py b/server/aps_domain/orders.py index acc50ab..fa83188 100644 --- a/server/aps_domain/orders.py +++ b/server/aps_domain/orders.py @@ -120,7 +120,8 @@ def sync_flex_orders_to_sales(world: World) -> int: } flex: list[dict[str, Any]] = [] for order in world.get("flexOrders") or []: - if str(order.get("source") or "").upper() in {"CLOSED_LOOP", "MRP"}: + # SALES 源柔性单来自主干销售订单投影，不反向回投，避免重复建单 + if str(order.get("source") or "").upper() in {"CLOSED_LOOP", "MRP", "SALES"}: continue if str(order.get("status") or "").upper() in {"DONE", "CANCELLED", "COMPLETED"}: continue @@ -257,6 +258,66 @@ def sync_flex_orders_to_sales(world: World) -> int: world["salesOrders"] = next_rows return changed + +SALES_FLEX_SOURCE = "SALES" # 主干销售订单 → 柔性订单的桥接来源标记 + + +def sync_sales_orders_to_flex(world: World) -> int: + """把主干可排销售订单幂等投影为柔性订单（salesOrders → flexOrders）。 + + 对话/手工新建的订单（含紧急插单）只落在主干 salesOrders；柔性 PoolEngine + 只读 flexOrders，导致「订单已导入却排不进去」。每次柔性排产前调用本投影： + APPROVED/CONFIRMED 的主干单桥接为 RELEASED 柔性单；主干状态回撤 + （取消/完成/退回审核）同步把柔性单移出可排集合。文件/SQL/闭环来源的 + 柔性单仍由各自通道维护，本函数只认领 source=SALES 的行。 + """ + flex_orders = world.setdefault("flexOrders", []) + by_no = {str(o.get("orderNo") or ""): o for o in flex_orders} + next_fid = max( + (int(o.get("id")) for o in flex_orders if isinstance(o.get("id"), int)), + default=0, + ) + 1 + changed = 0 + for so in world.get("salesOrders") or []: + order_no = str(so.get("orderNo") or "").strip() + if not order_no: + continue + existing = by_no.get(order_no) + if existing is not None and str(existing.get("source") or "").upper() != SALES_FLEX_SOURCE: + continue # 同名柔性单由文件/SQL/闭环通道维护，不覆盖 + status = str(so.get("status") or "").upper() + if is_schedulable(status): + item = (so.get("items") or [{}])[0] + if not item.get("productCode"): + continue + values = { + "productCode": item.get("productCode"), + "productName": item.get("productName"), + "quantity": item.get("quantity"), + "dueDate": str(so.get("deliveryDate") or "")[:10], + "priority": int(so.get("priority") or 5), + "customerName": so.get("customerName"), + "customerLevel": so.get("customerLevel") or "B", + "isRush": bool(so.get("isRush")), + "status": "RELEASED", + "source": SALES_FLEX_SOURCE, + "salesOrderNo": order_no, + } + if existing is None: + flex_orders.append({"id": next_fid, "orderNo": order_no, **values}) + next_fid += 1 + changed += 1 + elif any(existing.get(k) != v for k, v in values.items()): + existing.update(values) + changed += 1 + elif existing is not None: + # 主干状态回撤：柔性投影同步退出可排集合 + target = {"CANCELLED": "CANCELLED", "COMPLETED": "DONE"}.get(status, "ON_HOLD") + if existing.get("status") != target: + existing["status"] = target + changed += 1 + return changed + def list_orders(world: World) -> list[dict[str, Any]]: """销售订单列表投影（含已投影的 SQL/柔性单；只读兜底不写盘）。""" rows: list[dict[str, Any]] = [] @@ -450,6 +511,7 @@ def normalize_order_payload(world: World, payload: dict[str, Any]) -> dict[str, out["customerName"] = str(out.get("customerName") or "").strip() if not out["customerName"]: raise ValueError("客户名称不能为空") + out["orderNo"] = str(out.get("orderNo") or "").strip() out["customerLevel"] = str(out.get("customerLevel") or "A").upper() if out["customerLevel"] not in CUSTOMER_LEVELS: raise ValueError("客户等级必须是 VIP/A/B/C") @@ -649,9 +711,22 @@ def apply_order_action(world: World, next_id, action: str, payload: dict[str, An order_id = next_id("salesOrder") order_date = today0().strftime("%Y-%m-%d") product = _product(world, p["productId"]) + order_no = p.get("orderNo") or "" + if order_no: + # 计划员指定单号（如急单 RUSH-0007）：保留并查重，失败即拒绝 + folded = order_no.casefold() + clash = any( + str(o.get("orderNo") or "").casefold() == folded + for pool in ("salesOrders", "flexOrders") + for o in (world.get(pool) or []) + ) + if clash: + raise ValueError(f"订单号 {order_no} 已存在") + else: + order_no = "SO" + today0().strftime("%Y%m%d") + f"{order_id:03d}" order = { "id": order_id, - "orderNo": "SO" + today0().strftime("%Y%m%d") + f"{order_id:03d}", + "orderNo": order_no, "customerId": f"CUST{order_id:03d}", "customerName": p["customerName"], "customerLevel": p["customerLevel"],