# ============================================================ # 排产参数（moduleId: domain-params, 可重生 ✅） # OR-02：客户等级权重 / 目标权重；变更走 P2，只影响后续新版本 # ============================================================ from __future__ import annotations from copy import deepcopy from typing import Any World = dict[str, Any] DEFAULT_LEVEL_WEIGHTS = {"VIP": 3.0, "A": 2.0, "B": 1.0, "C": 1.0} DEFAULT_OBJECTIVE_WEIGHTS = {"tardiness": 0.4, "cost": 0.3, "utilization": 0.2, "balance": 0.1} def default_schedule_params() -> dict[str, Any]: return { "defaultEngine": "RULE", "planningHorizonDays": 14, "timeGranularityMinutes": 15, "deliveryBufferRatio": 0.95, "weights": dict(DEFAULT_OBJECTIVE_WEIGHTS), "customerLevelWeights": dict(DEFAULT_LEVEL_WEIGHTS), "freezeWindowHours": 24, "cpTimeLimitSeconds": 8, # SC-03 CP-SAT 时限 } def get_schedule_params(world: World) -> dict[str, Any]: """合并默认值后的排产参数投影（P0）。""" base = default_schedule_params() cur = world.get("scheduleParams") or {} out = {**base, **cur} out["weights"] = {**DEFAULT_OBJECTIVE_WEIGHTS, **(cur.get("weights") or {})} out["customerLevelWeights"] = { **DEFAULT_LEVEL_WEIGHTS, **(cur.get("customerLevelWeights") or {}), } return out def level_weight(world: World, level: str | None) -> float: """取客户等级排序权重（越高越优先排）。""" weights = get_schedule_params(world)["customerLevelWeights"] return float(weights.get(str(level or "C").upper(), 1.0)) def normalize_params_payload(payload: dict[str, Any]) -> dict[str, Any]: """校验并规范化 params.update 载荷。""" out: dict[str, Any] = {} if payload.get("resetDefaults"): out["resetDefaults"] = True return out if "customerLevelWeights" in payload and payload["customerLevelWeights"] is not None: raw = payload["customerLevelWeights"] if not isinstance(raw, dict): raise ValueError("customerLevelWeights 必须是对象") levels: dict[str, float] = {} for k, v in raw.items(): key = str(k).upper() if key not in DEFAULT_LEVEL_WEIGHTS: raise ValueError(f"不支持的客户等级：{k}") try: val = float(v) except (TypeError, ValueError) as exc: raise ValueError(f"{key} 权重必须是数字") from exc if val < 0 or val > 100: raise ValueError(f"{key} 权重须在 0~100") levels[key] = val out["customerLevelWeights"] = levels if "weights" in payload and payload["weights"] is not None: raw = payload["weights"] if not isinstance(raw, dict): raise ValueError("weights 必须是对象") obj: dict[str, float] = {} for k, v in raw.items(): key = str(k) if key not in DEFAULT_OBJECTIVE_WEIGHTS: raise ValueError(f"不支持的目标权重键：{k}") try: val = float(v) except (TypeError, ValueError) as exc: raise ValueError(f"{key} 权重必须是数字") from exc if val < 0 or val > 1: raise ValueError(f"{key} 目标权重须在 0~1") obj[key] = val out["weights"] = obj if "planningHorizonDays" in payload and payload["planningHorizonDays"] is not None: days = int(payload["planningHorizonDays"]) if days < 1 or days > 90: raise ValueError("planningHorizonDays 须在 1~90") out["planningHorizonDays"] = days if "defaultEngine" in payload and payload["defaultEngine"] is not None: eng = str(payload["defaultEngine"]).upper() if eng not in ("RULE", "CP", "GA", "HYBRID", "OPTIMIZE"): raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID/OPTIMIZE") out["defaultEngine"] = eng if "cpTimeLimitSeconds" in payload and payload["cpTimeLimitSeconds"] is not None: sec = float(payload["cpTimeLimitSeconds"]) if sec < 0.5 or sec > 120: raise ValueError("cpTimeLimitSeconds 须在 0.5~120") out["cpTimeLimitSeconds"] = sec if not out: raise ValueError("未提供可更新的排产参数字段") return out def confirmation_for_params_update(world: World, payload: dict[str, Any]) -> tuple[str, list[str]]: """生成 params.update 确认卡文案。""" norm = normalize_params_payload(payload) cur = get_schedule_params(world) if norm.get("resetDefaults"): return "恢复排产参数默认值", [ "客户等级权重恢复 VIP=3 / A=2 / B=1 / C=1", "目标权重恢复交期/成本/利用率/均衡默认比", "只影响后续新排产版本，不回写历史版本", ] lines: list[str] = [] if "customerLevelWeights" in norm: before = cur["customerLevelWeights"] after = {**before, **norm["customerLevelWeights"]} parts = [f"{k}:{before.get(k)}→{after[k]}" for k in ("VIP", "A", "B", "C")] lines.append("客户等级权重：" + " / ".join(parts)) if "weights" in norm: before = cur["weights"] after = {**before, **norm["weights"]} parts = [f"{k}:{before.get(k)}→{after[k]}" for k in after] lines.append("目标权重：" + " / ".join(parts)) if "planningHorizonDays" in norm: lines.append(f"展望期：{cur['planningHorizonDays']}→{norm['planningHorizonDays']} 天") if "defaultEngine" in norm: lines.append(f"默认引擎：{cur.get('defaultEngine')}→{norm['defaultEngine']}") if "cpTimeLimitSeconds" in norm: lines.append(f"CP 时限：{cur.get('cpTimeLimitSeconds')}→{norm['cpTimeLimitSeconds']} 秒") lines.append("只影响后续新排产版本，不回写历史版本") return "更新排产参数", lines def apply_params_update(world: World, payload: dict[str, Any]) -> dict[str, Any]: """写入 scheduleParams（P2 批准后调用）。""" norm = normalize_params_payload(payload) before = deepcopy(get_schedule_params(world)) if norm.get("resetDefaults"): world["scheduleParams"] = default_schedule_params() else: sp = world.setdefault("scheduleParams", default_schedule_params()) if "customerLevelWeights" in norm: sp["customerLevelWeights"] = { **DEFAULT_LEVEL_WEIGHTS, **(sp.get("customerLevelWeights") or {}), **norm["customerLevelWeights"], } if "weights" in norm: sp["weights"] = { **DEFAULT_OBJECTIVE_WEIGHTS, **(sp.get("weights") or {}), **norm["weights"], } if "planningHorizonDays" in norm: sp["planningHorizonDays"] = norm["planningHorizonDays"] if "defaultEngine" in norm: sp["defaultEngine"] = norm["defaultEngine"] if "cpTimeLimitSeconds" in norm: sp["cpTimeLimitSeconds"] = norm["cpTimeLimitSeconds"] after = get_schedule_params(world) return { "kind": "SCHEDULE_PARAMS", "id": "scheduleParams", "name": "排产参数", "before": before, "after": after, "reset": bool(norm.get("resetDefaults")), }