# ============================================================ # 批次/战役合并（moduleId: domain-campaign, SC-08，可重生 ✅） # 同产品 + 交期窗口内合并为一场战役：一次准备、累计数量、钉扎原销售单 # ============================================================ from __future__ import annotations from datetime import datetime from typing import Any from server.aps_domain.changeover import product_family from server.timeutil import parse_dt World = dict[str, Any] DEFAULT_CAMPAIGN_WINDOW_DAYS = 7 def _due(so: dict[str, Any]) -> datetime: return parse_dt(str(so.get("deliveryDate") or "")[:10] + " 00:00") def _can_join_campaign(entries: list[dict[str, Any]], candidate: dict[str, Any], window_days: int) -> bool: """候选能否并入当前战役：同产品已在组外校验；窗口看最早~最晚交期跨度。""" dues = [_due(e["so"]) for e in entries] + [_due(candidate["so"])] return (max(dues) - min(dues)).days <= window_days def _build_campaign_entry(members: list[dict[str, Any]]) -> dict[str, Any]: """多订单项合成一个排产入口（数量相加；主单取最早交期）。""" members = sorted(members, key=lambda e: (_due(e["so"]), e["so"].get("priority", 99), e["so"].get("orderNo") or "")) primary = members[0] item0 = primary["item"] total_qty = sum(float(m["item"].get("quantity") or 0) for m in members) sources = [] for m in members: so, it = m["so"], m["item"] sources.append({ "orderNo": so.get("orderNo"), "salesOrderId": so.get("id"), "itemId": it.get("id"), "quantity": it.get("quantity"), "deliveryDate": so.get("deliveryDate"), "customerName": so.get("customerName"), "priority": so.get("priority"), }) merged_item = dict(item0) merged_item["quantity"] = total_qty # 虚拟明细 id：保留主项 id，附加战役标记 merged_item["campaignSources"] = sources dues = [s["deliveryDate"] for s in sources] return { "so": primary["so"], "item": merged_item, "isCampaign": len(members) > 1, "campaign": { "productId": item0.get("productId"), "productCode": item0.get("productCode"), "productName": item0.get("productName"), "memberCount": len(members), "totalQty": total_qty, "earliestDue": min(dues) if dues else None, "latestDue": max(dues) if dues else None, "sourceOrderNos": [s["orderNo"] for s in sources], "windowDays": None, # 由调用方回填 }, } def merge_entries_to_campaigns( world: World, entries: list[dict[str, Any]], *, window_days: int = DEFAULT_CAMPAIGN_WINDOW_DAYS, ) -> list[dict[str, Any]]: """ 将待排入口按「同产品 + 交期窗口」合并。 插单 / 预测单独成战役，不与普通确定单合并。 """ window_days = max(0, int(window_days)) singles: list[dict[str, Any]] = [] buckets: dict[int, list[dict[str, Any]]] = {} for e in entries: so = e["so"] if so.get("isRush") or so.get("isForecast"): singles.append(_build_campaign_entry([e])) continue pid = int(e["item"]["productId"]) buckets.setdefault(pid, []).append(e) merged: list[dict[str, Any]] = list(singles) for _pid, group in buckets.items(): group = sorted(group, key=lambda e: (_due(e["so"]), e["so"].get("priority", 99), e["so"].get("orderNo") or "")) current: list[dict[str, Any]] = [] for e in group: if not current: current = [e] continue if _can_join_campaign(current, e, window_days): current.append(e) else: camp = _build_campaign_entry(current) camp["campaign"]["windowDays"] = window_days merged.append(camp) current = [e] if current: camp = _build_campaign_entry(current) camp["campaign"]["windowDays"] = window_days merged.append(camp) # 按最早交期排出场顺序（再交给 CHANGEOVER / 其他策略二次排序亦可） merged.sort(key=lambda e: ( 1 if e["so"].get("isForecast") else 0, 0 if e["so"].get("isRush") else 1, e["campaign"].get("earliestDue") or e["so"].get("deliveryDate") or "", e["so"].get("priority", 99), )) return merged def preview_campaigns( world: World, *, window_days: int = DEFAULT_CAMPAIGN_WINDOW_DAYS, include_unapproved: bool = False, include_forecast: bool = False, ) -> dict[str, Any]: """P0：预览战役合并方案（不写主干）。""" from server.aps_domain.orders import is_schedulable entries: list[dict[str, Any]] = [] for so in world.get("salesOrders") or []: if so.get("status") in ("CANCELLED", "COMPLETED"): continue if not is_schedulable(so.get("status"), include_unapproved=include_unapproved): continue for item in so.get("items") or []: if item.get("status") == "COMPLETED": continue entries.append({"so": so, "item": item}) if include_forecast: from server.aps_domain.forecast import forecasts_as_schedule_entries entries.extend(forecasts_as_schedule_entries(world)) before = len(entries) camps = merge_entries_to_campaigns(world, entries, window_days=window_days) rows = [] merged_orders = 0 for c in camps: meta = c.get("campaign") or {} members = meta.get("memberCount") or 1 if members > 1: merged_orders += members mat = next((m for m in world.get("materials", []) if m["id"] == meta.get("productId")), None) rows.append({ "productCode": meta.get("productCode") or (c["item"].get("productCode")), "productName": meta.get("productName") or (c["item"].get("productName")), "productFamily": product_family(mat) if mat else None, "memberCount": members, "totalQty": meta.get("totalQty") or c["item"].get("quantity"), "earliestDue": meta.get("earliestDue"), "latestDue": meta.get("latestDue"), "sourceOrderNos": meta.get("sourceOrderNos") or [c["so"].get("orderNo")], "isCampaign": bool(c.get("isCampaign")), "isRush": bool(c["so"].get("isRush")), "isForecast": bool(c["so"].get("isForecast")), }) saved_pos = before - len(camps) return { "windowDays": window_days, "includeForecast": include_forecast, "summary": { "inputOrders": before, "campaignCount": len(camps), "mergedCampaigns": sum(1 for r in rows if r["isCampaign"]), "ordersInMerged": merged_orders, "poSaved": max(0, saved_pos), }, "campaigns": rows, "hint": ( f"交期窗口 {window_days} 天内同产品合并为一场战役；插单/预测单独保留。" "试排请用策略 CAMPAIGN 或口令「战役合并试排」。" ), }