610 lines
39 KiB
Python
610 lines
39 KiB
Python
"""Closed-loop Scheduling Kernel v1: pure order-to-requirement/supply builder.
|
|
|
|
``build_closed_loop_problem`` reads a world mapping and emits a versioned
|
|
manufacturing requirement graph. It never creates or mutates PO/WO records.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import asdict, dataclass, field
|
|
from hashlib import sha256
|
|
import json
|
|
from typing import Any, Literal, Mapping, MutableMapping, Sequence
|
|
|
|
from server.aps_domain.supply_netting import (
|
|
DemandBalance, NettingRequirement, SupplyAllocation, SupplyCandidate,
|
|
SupplyEvent, SourcingType, net_supply_requirements, normalize_date,
|
|
normalize_quantity,
|
|
)
|
|
|
|
RoutingStatus = Literal["READY", "TEMPLATE", "MISSING", "NOT_APPLICABLE"]
|
|
ResourceStatus = Literal["READY", "MISSING", "NOT_APPLICABLE"]
|
|
Severity = Literal["HARD", "WARNING"]
|
|
_SCHEMA_VERSION = "closed-loop-scheduling/1.0"
|
|
_INACTIVE = {"CANCELLED", "CANCELED", "COMPLETED", "CLOSED", "REJECTED", "VOID"}
|
|
_TRUSTED_SUPPLY = {"APPROVED", "CONFIRMED", "RELEASED", "IN_PROGRESS", "PARTIAL", "OPEN", "RECEIVED", "COMPLETED"}
|
|
_DRAFT_SUPPLY = {"DRAFT", "SUGGESTED", "PROPOSED", "PENDING_APPROVAL"}
|
|
_TEMPLATE = {"模板", "TEMPLATE", "DEFAULT", "INFERRED", "RAG_CANDIDATE"}
|
|
_EXTERNAL = {"OUTSOURCE", "SUBCONTRACT", "EXTERNAL", "外协", "委外"}
|
|
_MAKE = {"MAKE", "PRODUCTION", "INTERNAL", "自制"}
|
|
_BUY = {"BUY", "PURCHASE", "采购"}
|
|
_RESOURCE_INACTIVE = {"DOWN", "FAULT", "MAINTENANCE", "INACTIVE", "DISABLED", "SCRAPPED"}
|
|
_RELEVANT_KEYS = (
|
|
"salesOrders", "materials", "flexMaterials", "boms", "bomItems", "flexBom",
|
|
"routings", "routingSteps", "routingOperations", "operations", "flexRoutings",
|
|
"workstations", "workstationOperations", "equipment", "flexEquipment",
|
|
"inventory", "inventories", "inventoryBalances", "inventoryItems", "stock",
|
|
"inTransit", "inTransitItems", "inventoryInTransit", "purchaseOrders", "outsourceOrders",
|
|
)
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PeggingRef:
|
|
parent_requirement_id: str
|
|
quantity: float
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Blocker:
|
|
blocker_id: str
|
|
code: str
|
|
severity: Severity
|
|
entity_type: str
|
|
entity_id: str
|
|
message: str
|
|
evidence: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Requirement:
|
|
requirement_id: str
|
|
sales_order_id: str
|
|
sales_order_no: str
|
|
sales_order_line_id: str
|
|
material_id: str | int | None
|
|
material_key: str
|
|
material_code: str
|
|
material_name: str
|
|
quantity: float
|
|
unit: str
|
|
required_at: str
|
|
sourcing_type: SourcingType
|
|
bom_depth: int
|
|
parent_requirement_ids: tuple[str, ...]
|
|
child_requirement_ids: tuple[str, ...]
|
|
pegging_refs: tuple[PeggingRef, ...]
|
|
operation_code: str | None
|
|
routing_status: RoutingStatus
|
|
resource_status: ResourceStatus
|
|
blocker_codes: tuple[str, ...]
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ManufacturingDemand:
|
|
demand_id: str
|
|
requirement_id: str
|
|
sales_order_id: str
|
|
sales_order_no: str
|
|
sales_order_line_id: str
|
|
product_id: str | int | None
|
|
product_code: str
|
|
product_name: str
|
|
quantity: float
|
|
unit: str
|
|
required_at: str
|
|
bom_depth: int
|
|
parent_demand_ids: tuple[str, ...]
|
|
child_requirement_ids: tuple[str, ...]
|
|
routing_status: RoutingStatus
|
|
resource_status: ResourceStatus
|
|
material_status: str
|
|
release_status: str
|
|
blocker_codes: tuple[str, ...]
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ClosedLoopProblem:
|
|
schema_version: str
|
|
problem_id: str
|
|
source_revision: str
|
|
business_date: str
|
|
requirements: tuple[Requirement, ...]
|
|
manufacturing_demands: tuple[ManufacturingDemand, ...]
|
|
supply_events: tuple[SupplyEvent, ...]
|
|
demand_balances: tuple[DemandBalance, ...]
|
|
blockers: tuple[Blocker, ...]
|
|
stats: Mapping[str, Any]
|
|
|
|
@property
|
|
def problemId(self) -> str:
|
|
return self.problem_id
|
|
|
|
@property
|
|
def sourceHash(self) -> str:
|
|
return self.source_revision
|
|
|
|
@property
|
|
def manufacturingDemands(self) -> tuple[ManufacturingDemand, ...]:
|
|
return self.manufacturing_demands
|
|
|
|
@property
|
|
def supplyEvents(self) -> tuple[SupplyEvent, ...]:
|
|
return self.supply_events
|
|
|
|
@property
|
|
def summary(self) -> Mapping[str, Any]:
|
|
return self.stats
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schemaVersion": self.schema_version,
|
|
"problemId": self.problem_id,
|
|
"sourceHash": self.source_revision,
|
|
"businessDate": self.business_date,
|
|
"requirements": [asdict(row) for row in self.requirements],
|
|
"manufacturingDemands": [asdict(row) for row in self.manufacturing_demands],
|
|
"supplyEvents": [asdict(row) for row in self.supply_events],
|
|
"demandBalances": [asdict(row) for row in self.demand_balances],
|
|
"blockers": [asdict(row) for row in self.blockers],
|
|
"summary": dict(self.stats),
|
|
}
|
|
|
|
@dataclass(slots=True)
|
|
class _RequirementDraft:
|
|
requirement_id: str
|
|
sales_order_id: str
|
|
sales_order_no: str
|
|
sales_order_line_id: str
|
|
material_id: str | int | None
|
|
material_key: str
|
|
material_code: str
|
|
material_name: str
|
|
quantity: float
|
|
unit: str
|
|
required_at: str
|
|
sourcing_type: SourcingType
|
|
bom_depth: int
|
|
operation_code: str | None = None
|
|
routing_status: RoutingStatus = "NOT_APPLICABLE"
|
|
resource_status: ResourceStatus = "NOT_APPLICABLE"
|
|
parent_quantities: MutableMapping[str, float] = field(default_factory=dict)
|
|
child_ids: set[str] = field(default_factory=set)
|
|
blocker_codes: set[str] = field(default_factory=set)
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _RouteAnalysis:
|
|
status: RoutingStatus
|
|
resource_status: ResourceStatus
|
|
missing_resource_codes: tuple[str, ...]
|
|
external_steps: tuple[Mapping[str, Any], ...]
|
|
|
|
def _stable_id(prefix: str, *parts: Any) -> str:
|
|
payload = json.dumps(parts, ensure_ascii=False, sort_keys=True, default=str, separators=(",", ":"))
|
|
return f"{prefix}-{sha256(payload.encode('utf-8')).hexdigest()[:20]}"
|
|
|
|
def _status(value: Any) -> str:
|
|
return str(value or "").strip().upper()
|
|
|
|
def _material_key(material: Mapping[str, Any] | None, fallback: Any = "") -> str:
|
|
material = material or {}
|
|
code = str(material.get("code") or material.get("materialCode") or fallback or "").strip()
|
|
if code:
|
|
return f"CODE:{code.upper()}"
|
|
mid = material.get("id") or material.get("materialId")
|
|
return f"ID:{mid}" if mid not in (None, "") else "UNKNOWN"
|
|
|
|
def _catalog(world: Mapping[str, Any]) -> tuple[dict[Any, dict[str, Any]], dict[str, dict[str, Any]]]:
|
|
by_id: dict[Any, dict[str, Any]] = {}
|
|
by_code: dict[str, dict[str, Any]] = {}
|
|
for table in ("flexMaterials", "materials"):
|
|
rows = world.get(table) or ()
|
|
if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes, bytearray)):
|
|
continue
|
|
for source in rows:
|
|
if not isinstance(source, Mapping):
|
|
continue
|
|
row = dict(source)
|
|
code = str(row.get("code") or row.get("materialCode") or "").strip().upper()
|
|
if code and code in by_code:
|
|
by_code[code].update(row)
|
|
row = by_code[code]
|
|
elif code:
|
|
by_code[code] = row
|
|
mid = row.get("id") or row.get("materialId")
|
|
if mid not in (None, ""):
|
|
by_id[mid] = row
|
|
return by_id, by_code
|
|
|
|
def _resolve_material(by_id: Mapping[Any, Mapping[str, Any]], by_code: Mapping[str, Mapping[str, Any]], *, material_id: Any = None, material_code: Any = None, fallback: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
|
row = by_id.get(material_id) if material_id not in (None, "") else None
|
|
if row is None and material_code not in (None, ""):
|
|
row = by_code.get(str(material_code).strip().upper())
|
|
merged = dict(row or {})
|
|
if fallback:
|
|
for key, value in fallback.items():
|
|
if value not in (None, "") and merged.get(key) in (None, ""):
|
|
merged[key] = value
|
|
if material_id not in (None, "") and merged.get("id") in (None, ""):
|
|
merged["id"] = material_id
|
|
if material_code not in (None, "") and merged.get("code") in (None, ""):
|
|
merged["code"] = material_code
|
|
return merged
|
|
|
|
def _has_bom(world: Mapping[str, Any], material: Mapping[str, Any]) -> bool:
|
|
pid = material.get("id") or material.get("productId")
|
|
code = str(material.get("code") or material.get("productCode") or "").strip().upper()
|
|
return any(isinstance(row, Mapping) and row.get("productId") == pid and bool(row.get("isDefault", True)) for row in (world.get("boms") or ())) or bool(code and any(isinstance(row, Mapping) and str(row.get("productCode") or "").strip().upper() == code for row in (world.get("flexBom") or ())))
|
|
|
|
def _infer_sourcing(world: Mapping[str, Any], material: Mapping[str, Any]) -> SourcingType:
|
|
procurement = _status(material.get("procurementType"))
|
|
sourcing = _status(material.get("sourcingType"))
|
|
source = _status(material.get("sourcingTypeSource"))
|
|
explicit = procurement or (sourcing if source in {"EXPLICIT", "MANUAL", "IMPORT"} else "")
|
|
if explicit in _EXTERNAL: return "SUBCONTRACT"
|
|
if explicit in _BUY: return "BUY"
|
|
if explicit in _MAKE: return "MAKE"
|
|
mtype = _status(material.get("type"))
|
|
if _route_rows(world, material)[0] or _route_rows(world, material)[1] or _has_bom(world, material) or mtype in {"FINISHED_PRODUCT", "SEMI_FINISHED"}: return "MAKE"
|
|
if sourcing in _EXTERNAL or str(material.get("code") or "").upper().startswith("WZ"): return "SUBCONTRACT"
|
|
if sourcing in _MAKE: return "MAKE"
|
|
return "BUY"
|
|
def _bom_children(world: Mapping[str, Any], material: Mapping[str, Any], by_id: Mapping[Any, Mapping[str, Any]], by_code: Mapping[str, Mapping[str, Any]]) -> list[tuple[dict[str, Any], float, str]]:
|
|
pid = material.get("id") or material.get("productId")
|
|
boms = [row for row in (world.get("boms") or ()) if isinstance(row, Mapping) and pid not in (None, "") and row.get("productId") == pid and bool(row.get("isDefault", True))]
|
|
if boms:
|
|
boms.sort(key=lambda row: (0 if _status(row.get("status")) in {"ACTIVE", "APPROVED"} else 1, str(row.get("id"))))
|
|
bom_id = boms[0].get("id")
|
|
result = []
|
|
for item in world.get("bomItems") or ():
|
|
if not isinstance(item, Mapping) or item.get("bomId") != bom_id: continue
|
|
child = _resolve_material(by_id, by_code, material_id=item.get("materialId"), material_code=item.get("materialCode"))
|
|
result.append((child, normalize_quantity(item.get("quantity")), str(item.get("id") or "")))
|
|
return result
|
|
code = str(material.get("code") or material.get("productCode") or "").strip().upper()
|
|
result = []
|
|
for item in world.get("flexBom") or ():
|
|
if not isinstance(item, Mapping) or str(item.get("productCode") or "").strip().upper() != code: continue
|
|
child = _resolve_material(by_id, by_code, material_id=item.get("materialId"), material_code=item.get("materialCode"))
|
|
result.append((child, normalize_quantity(item.get("quantity")), str(item.get("id") or "")))
|
|
return result
|
|
|
|
def _route_rows(world: Mapping[str, Any], material: Mapping[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
pid = material.get("id") or material.get("productId")
|
|
code = str(material.get("code") or material.get("productCode") or "").strip().upper()
|
|
operations = {row.get("id"): row for row in (world.get("operations") or ()) if isinstance(row, Mapping)}
|
|
routes = [row for row in (world.get("routings") or ()) if isinstance(row, Mapping) and pid not in (None, "") and row.get("productId") == pid and bool(row.get("isDefault", True))]
|
|
fixed: list[dict[str, Any]] = []
|
|
if routes:
|
|
routes.sort(key=lambda row: (0 if _status(row.get("status")) in {"ACTIVE", "APPROVED"} else 1, str(row.get("id"))))
|
|
routing_id = routes[0].get("id")
|
|
for step in list(world.get("routingSteps") or ()) + list(world.get("routingOperations") or ()):
|
|
if not isinstance(step, Mapping) or step.get("routingId") != routing_id: continue
|
|
op = operations.get(step.get("operationId")) or {}
|
|
normalized = dict(step)
|
|
normalized.setdefault("operationCode", op.get("code"))
|
|
normalized.setdefault("operationName", op.get("name"))
|
|
normalized.setdefault("operationType", op.get("type"))
|
|
normalized.setdefault("sourcingType", op.get("sourcingType"))
|
|
fixed.append(normalized)
|
|
fixed.sort(key=lambda row: (int(row.get("sequenceNo") or row.get("seq") or 0), str(row.get("id") or "")))
|
|
flex = [dict(row) for row in (world.get("flexRoutings") or ()) if isinstance(row, Mapping) and code and str(row.get("productCode") or "").strip().upper() == code]
|
|
flex.sort(key=lambda row: (int(row.get("seq") or row.get("sequenceNo") or 0), str(row.get("id") or "")))
|
|
return fixed, flex
|
|
|
|
def _is_external_step(step: Mapping[str, Any]) -> bool:
|
|
if "isExternal" in step: return bool(step.get("isExternal"))
|
|
marker = _status(step.get("sourcingType") or step.get("operationType") or step.get("opType"))
|
|
code = str(step.get("operationCode") or "").strip().upper()
|
|
name = str(step.get("operationName") or "")
|
|
return marker in _EXTERNAL or code.startswith("WZ") or "委外" in name or "外协" in name
|
|
|
|
def _available_capabilities(world: Mapping[str, Any]) -> set[str]:
|
|
capabilities: set[str] = set()
|
|
for row in list(world.get("flexEquipment") or ()) + list(world.get("equipment") or ()):
|
|
if not isinstance(row, Mapping) or _status(row.get("status")) in _RESOURCE_INACTIVE: continue
|
|
raw = row.get("capabilities") or row.get("capabilityCodes") or ()
|
|
if isinstance(raw, str): raw = [part.strip() for part in raw.replace(";", ",").split(",") if part.strip()]
|
|
capabilities.update(str(value).strip().upper() for value in raw)
|
|
active_ws = {row.get("id") for row in (world.get("workstations") or ()) if isinstance(row, Mapping) and _status(row.get("status")) not in _RESOURCE_INACTIVE}
|
|
op_codes = {row.get("id"): str(row.get("code") or "").strip().upper() for row in (world.get("operations") or ()) if isinstance(row, Mapping)}
|
|
for link in world.get("workstationOperations") or ():
|
|
if isinstance(link, Mapping) and link.get("workstationId") in active_ws:
|
|
code = op_codes.get(link.get("operationId")) or str(link.get("operationCode") or "").strip().upper()
|
|
if code: capabilities.add(code)
|
|
return capabilities
|
|
|
|
def _analyze_route(world: Mapping[str, Any], material: Mapping[str, Any]) -> _RouteAnalysis:
|
|
fixed, flex = _route_rows(world, material)
|
|
steps = fixed or flex
|
|
if not steps: return _RouteAnalysis("MISSING", "MISSING", (), ())
|
|
route_status: RoutingStatus = "TEMPLATE" if flex and all(_status(step.get("stdTimeSource")) in _TEMPLATE for step in flex) else "READY"
|
|
internal_codes, external = [], []
|
|
for step in steps:
|
|
if _is_external_step(step): external.append(dict(step))
|
|
else:
|
|
code = str(step.get("operationCode") or "").strip().upper()
|
|
if code: internal_codes.append(code)
|
|
capabilities = _available_capabilities(world)
|
|
missing = tuple(sorted({code for code in internal_codes if code not in capabilities}))
|
|
return _RouteAnalysis(route_status, "MISSING" if missing else "READY", missing, tuple(external))
|
|
|
|
_DRAFT_SUPPLY_TABLES = frozenset({"purchaseOrders", "outsourceOrders"})
|
|
_DRAFT_SUPPLY_TRANSIENT_FIELDS = frozenset({"id", "orderNo", "createdAt", "updatedAt"})
|
|
|
|
|
|
def _canonical_source_value(value: Any, *, collection: str | None = None) -> Any:
|
|
"""Return a deterministic planning-source representation.
|
|
|
|
Generated DRAFT supply rows are recreated by every MRP decomposition. Their
|
|
persistence identifiers, display numbers, and audit timestamps are technical
|
|
provenance only; material, quantity, dates, sourcing, and status remain in
|
|
the digest. Lists are sorted by their canonical JSON so record ordering does
|
|
not make the same business snapshot look different.
|
|
"""
|
|
if isinstance(value, Mapping):
|
|
is_generated_draft_supply = (
|
|
collection in _DRAFT_SUPPLY_TABLES
|
|
and _status(value.get("status")) in _DRAFT_SUPPLY
|
|
)
|
|
normalized_mapping: dict[str, Any] = {}
|
|
for raw_key in sorted(value, key=lambda item: str(item)):
|
|
key = str(raw_key)
|
|
if is_generated_draft_supply and key in _DRAFT_SUPPLY_TRANSIENT_FIELDS:
|
|
continue
|
|
normalized_mapping[key] = _canonical_source_value(value[raw_key])
|
|
return normalized_mapping
|
|
if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
normalized_sequence = [
|
|
_canonical_source_value(item, collection=collection)
|
|
for item in value
|
|
]
|
|
return sorted(
|
|
normalized_sequence,
|
|
key=lambda item: json.dumps(
|
|
item,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
default=str,
|
|
separators=(",", ":"),
|
|
),
|
|
)
|
|
if isinstance(value, (set, frozenset)):
|
|
return sorted(
|
|
(_canonical_source_value(item) for item in value),
|
|
key=lambda item: json.dumps(
|
|
item,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
default=str,
|
|
separators=(",", ":"),
|
|
),
|
|
)
|
|
return value
|
|
|
|
def _source_revision(world: Mapping[str, Any], business_date: str) -> str:
|
|
snapshot = {
|
|
key: _canonical_source_value(world.get(key), collection=key)
|
|
for key in _RELEVANT_KEYS
|
|
if key in world
|
|
}
|
|
payload = json.dumps(
|
|
{"businessDate": business_date, "world": snapshot},
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
default=str,
|
|
separators=(",", ":"),
|
|
)
|
|
return sha256(payload.encode("utf-8")).hexdigest()
|
|
|
|
def _add_blocker(blockers: dict[str, Blocker], codes: MutableMapping[str, set[str]], *, code: str, entity_id: str, message: str, evidence: Mapping[str, Any] | None = None, entity_type: str = "REQUIREMENT", severity: Severity = "HARD") -> None:
|
|
evidence_dict = dict(evidence or {})
|
|
blocker_id = _stable_id("BLK", code, entity_type, entity_id, evidence_dict)
|
|
blockers.setdefault(blocker_id, Blocker(blocker_id, code, severity, entity_type, entity_id, message, evidence_dict))
|
|
if entity_type == "REQUIREMENT": codes.setdefault(entity_id, set()).add(code)
|
|
|
|
def _supply_date(row: Mapping[str, Any]) -> str | None:
|
|
for name in ("availableAt", "expectedDate", "arrivalDate", "promisedDate", "expectedArrivalDate", "requiredDate", "dueDate"):
|
|
value = normalize_date(row.get(name))
|
|
if value: return value
|
|
return None
|
|
|
|
def _row_quantity(row: Mapping[str, Any]) -> float:
|
|
for name in ("availableQuantity", "quantity", "qty", "onHand", "stock", "balance"):
|
|
if row.get(name) not in (None, ""): return normalize_quantity(row.get(name))
|
|
return 0.0
|
|
|
|
def _row_material_key(row: Mapping[str, Any], by_id: Mapping[Any, Mapping[str, Any]], by_code: Mapping[str, Mapping[str, Any]]) -> str:
|
|
material = _resolve_material(by_id, by_code, material_id=row.get("materialId") or row.get("productId"), material_code=row.get("materialCode") or row.get("productCode") or row.get("code"))
|
|
return _material_key(material, row.get("materialCode") or row.get("productCode") or row.get("code"))
|
|
def _collect_supply_candidates(world: Mapping[str, Any], business_date: str, requirements: Mapping[str, _RequirementDraft], by_id: Mapping[Any, Mapping[str, Any]], by_code: Mapping[str, Mapping[str, Any]]) -> list[SupplyCandidate]:
|
|
external_keys = {req.material_key for req in requirements.values() if req.sourcing_type in {"BUY", "SUBCONTRACT"}}
|
|
buy_keys = {req.material_key for req in requirements.values() if req.sourcing_type == "BUY"}
|
|
candidates: list[SupplyCandidate] = []
|
|
inventory_tables = [world.get(name) for name in ("inventoryBalances", "inventories", "inventory", "inventoryItems", "stock") if isinstance(world.get(name), list)]
|
|
explicit_stock: set[str] = set()
|
|
for table in inventory_tables:
|
|
for row in table or ():
|
|
if not isinstance(row, Mapping): continue
|
|
key, quantity = _row_material_key(row, by_id, by_code), _row_quantity(row)
|
|
if key not in buy_keys or quantity <= 0: continue
|
|
explicit_stock.add(key)
|
|
candidates.append(SupplyCandidate(_stable_id("STOCK", key, row.get("id") or row.get("locationId") or len(candidates)), key, quantity, business_date, "STOCK", "AVAILABLE", True, str(row.get("id") or row.get("locationId") or key), {"sourceTable": "inventory"}))
|
|
for material in by_code.values():
|
|
key = _material_key(material)
|
|
if key not in buy_keys or key in explicit_stock: continue
|
|
gross, safety = normalize_quantity(material.get("stock")), normalize_quantity(material.get("safetyStock"))
|
|
available = normalize_quantity(max(gross - safety, 0.0))
|
|
if available > 0:
|
|
candidates.append(SupplyCandidate(f"STOCK:{key}", key, available, business_date, "STOCK", "AVAILABLE", True, key, {"grossStock": gross, "safetyStock": safety, "sourceTable": "materials"}))
|
|
transit_tables = [world.get(name) for name in ("inTransit", "inTransitItems", "inventoryInTransit") if isinstance(world.get(name), list)]
|
|
explicit_transit: set[str] = set()
|
|
for table in transit_tables:
|
|
for row in table or ():
|
|
if not isinstance(row, Mapping): continue
|
|
key, quantity = _row_material_key(row, by_id, by_code), _row_quantity(row)
|
|
if key not in buy_keys or quantity <= 0: continue
|
|
explicit_transit.add(key)
|
|
available_at, status = _supply_date(row), _status(row.get("status")) or "IN_TRANSIT"
|
|
candidates.append(SupplyCandidate(_stable_id("INTRANSIT", key, row.get("id") or len(candidates)), key, quantity, available_at, "IN_TRANSIT", status, available_at is not None and status not in _DRAFT_SUPPLY, str(row.get("id") or ""), {"sourceTable": "inTransit"}))
|
|
for material in by_code.values():
|
|
key = _material_key(material)
|
|
if key not in buy_keys or key in explicit_transit: continue
|
|
quantity = normalize_quantity(material.get("inTransit"))
|
|
if quantity <= 0: continue
|
|
available_at = _supply_date(material)
|
|
candidates.append(SupplyCandidate(f"INTRANSIT:{key}", key, quantity, available_at, "IN_TRANSIT", "IN_TRANSIT", available_at is not None, key, {"sourceTable": "materials"}))
|
|
for row in world.get("purchaseOrders") or ():
|
|
if not isinstance(row, Mapping): continue
|
|
key, quantity = _row_material_key(row, by_id, by_code), _row_quantity(row)
|
|
if key not in external_keys or quantity <= 0: continue
|
|
status, available_at = _status(row.get("status")) or "UNKNOWN", _supply_date(row)
|
|
candidates.append(SupplyCandidate(f"PO:{row.get('id') or row.get('orderNo') or _stable_id('REF', key, quantity)}", key, quantity, available_at, "PURCHASE", status, status in _TRUSTED_SUPPLY and available_at is not None, str(row.get("id") or row.get("orderNo") or ""), {"orderNo": row.get("orderNo"), "salesOrderId": row.get("salesOrderId"), "sourceTable": "purchaseOrders"}))
|
|
for row in world.get("outsourceOrders") or ():
|
|
if not isinstance(row, Mapping): continue
|
|
product_code = str(row.get("productCode") or "").strip().upper()
|
|
operation_code = str(row.get("operationCode") or "").strip().upper()
|
|
key = f"OP:{product_code}:{operation_code}:SEQ:{int(row.get('sequenceNo') or 0)}" if product_code and operation_code else _row_material_key(row, by_id, by_code)
|
|
quantity = _row_quantity(row)
|
|
if key not in external_keys or quantity <= 0: continue
|
|
status, available_at = _status(row.get("status")) or "UNKNOWN", _supply_date(row)
|
|
candidates.append(SupplyCandidate(f"OUT:{row.get('id') or row.get('orderNo') or _stable_id('REF', key, quantity)}", key, quantity, available_at, "SUBCONTRACT", status, status in _TRUSTED_SUPPLY and available_at is not None, str(row.get("id") or row.get("orderNo") or ""), {"orderNo": row.get("orderNo"), "salesOrderId": row.get("salesOrderId"), "sourceTable": "outsourceOrders"}))
|
|
return candidates
|
|
|
|
def build_closed_loop_problem(world: Mapping[str, Any], *, business_date: str = "2026-08-02", order_nos: Sequence[str] | None = None, strict: bool = True) -> ClosedLoopProblem:
|
|
"""Build the v1 closed-loop problem without mutating ``world``."""
|
|
business_day = normalize_date(business_date)
|
|
if business_day is None: raise ValueError(f"invalid business_date: {business_date!r}")
|
|
selected_orders = {str(value) for value in order_nos} if order_nos else None
|
|
by_id, by_code = _catalog(world)
|
|
requirements: dict[str, _RequirementDraft] = {}
|
|
blockers: dict[str, Blocker] = {}
|
|
blocker_codes: dict[str, set[str]] = defaultdict(set)
|
|
route_cache: dict[str, _RouteAnalysis] = {}
|
|
order_priorities: dict[str, int] = {}
|
|
order_count = order_line_count = 0
|
|
|
|
def upsert(*, root_token: str, sales_order_id: str, sales_order_no: str, sales_order_line_id: str, material: Mapping[str, Any], quantity: float, unit: str, required_at: str, sourcing_type: SourcingType, bom_depth: int, parent_id: str | None, operation_code: str | None = None, synthetic_key: str | None = None) -> _RequirementDraft:
|
|
key = synthetic_key or _material_key(material)
|
|
req_id = _stable_id("REQ", root_token, sourcing_type, key, operation_code or "")
|
|
qty = normalize_quantity(quantity)
|
|
req = requirements.get(req_id)
|
|
if req is None:
|
|
req = _RequirementDraft(req_id, sales_order_id, sales_order_no, sales_order_line_id, material.get("id") or material.get("materialId"), key, str(material.get("code") or material.get("materialCode") or ""), str(material.get("name") or material.get("materialName") or material.get("code") or key), 0.0, str(unit or material.get("unit") or "PCS"), required_at, sourcing_type, bom_depth, operation_code)
|
|
requirements[req_id] = req
|
|
req.quantity = normalize_quantity(req.quantity + qty)
|
|
req.bom_depth = min(req.bom_depth, bom_depth)
|
|
if parent_id:
|
|
req.parent_quantities[parent_id] = normalize_quantity(req.parent_quantities.get(parent_id, 0.0) + qty)
|
|
requirements[parent_id].child_ids.add(req_id)
|
|
return req
|
|
|
|
def analyze_make(req: _RequirementDraft, material: Mapping[str, Any]) -> _RouteAnalysis:
|
|
analysis = route_cache.setdefault(req.material_key, _analyze_route(world, material))
|
|
req.routing_status, req.resource_status = analysis.status, analysis.resource_status
|
|
if analysis.status == "MISSING": _add_blocker(blockers, blocker_codes, code="MISSING_ROUTING", entity_id=req.requirement_id, message=f"{req.material_code or req.material_name} 缺少正式工艺路线", evidence={"materialKey": req.material_key})
|
|
elif analysis.status == "TEMPLATE": _add_blocker(blockers, blocker_codes, code="TEMPLATE_ROUTING_UNCONFIRMED", entity_id=req.requirement_id, message=f"{req.material_code or req.material_name} 仅有模板工艺，确认前不可正式排产", evidence={"materialKey": req.material_key}, severity="HARD" if strict else "WARNING")
|
|
if analysis.resource_status == "MISSING" and analysis.missing_resource_codes: _add_blocker(blockers, blocker_codes, code="MISSING_RESOURCE_CAPABILITY", entity_id=req.requirement_id, message=f"{req.material_code or req.material_name} 缺少工序能力资源", evidence={"operationCodes": list(analysis.missing_resource_codes)})
|
|
return analysis
|
|
def walk(*, root_token: str, sales_order_id: str, sales_order_no: str, sales_order_line_id: str, material: Mapping[str, Any], quantity: float, unit: str, required_at: str, bom_depth: int, parent_id: str | None, ancestors: tuple[str, ...]) -> None:
|
|
key = _material_key(material)
|
|
sourcing = _infer_sourcing(world, material)
|
|
req = upsert(root_token=root_token, sales_order_id=sales_order_id, sales_order_no=sales_order_no, sales_order_line_id=sales_order_line_id, material=material, quantity=quantity, unit=unit, required_at=required_at, sourcing_type=sourcing, bom_depth=bom_depth, parent_id=parent_id)
|
|
if key == "UNKNOWN": _add_blocker(blockers, blocker_codes, code="MISSING_MATERIAL_MASTER", entity_id=req.requirement_id, message="BOM/订单需求无法解析到物料主数据", evidence={"depth": bom_depth})
|
|
if sourcing != "MAKE": return
|
|
analysis = analyze_make(req, material)
|
|
for step in analysis.external_steps:
|
|
product_code = str(material.get("code") or material.get("productCode") or "").strip().upper()
|
|
operation_code = str(step.get("operationCode") or "").strip().upper()
|
|
sequence = int(step.get("sequenceNo") or step.get("seq") or 0)
|
|
op_key = f"OP:{product_code}:{operation_code}:SEQ:{sequence}"
|
|
op_material = {"code": op_key, "name": str(step.get("operationName") or operation_code or "委外工序"), "unit": unit, "sourcingType": "SUBCONTRACT"}
|
|
upsert(root_token=root_token, sales_order_id=sales_order_id, sales_order_no=sales_order_no, sales_order_line_id=sales_order_line_id, material=op_material, quantity=quantity, unit=unit, required_at=required_at, sourcing_type="SUBCONTRACT", bom_depth=bom_depth + 1, parent_id=req.requirement_id, operation_code=operation_code, synthetic_key=op_key)
|
|
next_ancestors = ancestors + (key,)
|
|
for child, per_qty, edge_id in _bom_children(world, material, by_id, by_code):
|
|
if per_qty <= 0:
|
|
_add_blocker(blockers, blocker_codes, code="INVALID_BOM_QUANTITY", entity_id=req.requirement_id, message="BOM 用量必须大于 0", evidence={"bomItemId": edge_id, "quantity": per_qty})
|
|
continue
|
|
child_key = _material_key(child)
|
|
if child_key in next_ancestors:
|
|
_add_blocker(blockers, blocker_codes, code="BOM_CYCLE", entity_id=req.requirement_id, message="检测到 BOM 环，已停止继续展开", evidence={"path": list(next_ancestors + (child_key,)), "bomItemId": edge_id})
|
|
continue
|
|
walk(root_token=root_token, sales_order_id=sales_order_id, sales_order_no=sales_order_no, sales_order_line_id=sales_order_line_id, material=child, quantity=normalize_quantity(quantity * per_qty), unit=str(child.get("unit") or unit), required_at=required_at, bom_depth=bom_depth + 1, parent_id=req.requirement_id, ancestors=next_ancestors)
|
|
|
|
for order in world.get("salesOrders") or ():
|
|
if not isinstance(order, Mapping) or _status(order.get("status")) in _INACTIVE: continue
|
|
order_id = str(order.get("id") or order.get("orderNo") or "")
|
|
order_no = str(order.get("orderNo") or order_id)
|
|
if selected_orders is not None and order_no not in selected_orders: continue
|
|
priority = int(order.get("priority") or 0) + (100 if bool(order.get("isRush")) else 0)
|
|
raw_due = order.get("deliveryDate") or order.get("dueDate")
|
|
required_at = normalize_date(raw_due) or business_day
|
|
order_priorities[order_id] = priority
|
|
order_count += 1
|
|
for index, item in enumerate(order.get("items") or (), start=1):
|
|
if not isinstance(item, Mapping) or _status(item.get("status")) in _INACTIVE: continue
|
|
quantity = normalize_quantity(item.get("quantity"))
|
|
if quantity <= 0: continue
|
|
line_id = str(item.get("id") or item.get("lineId") or index)
|
|
material = _resolve_material(by_id, by_code, material_id=item.get("productId") or item.get("materialId"), material_code=item.get("productCode") or item.get("materialCode"), fallback={"id": item.get("productId") or item.get("materialId"), "code": item.get("productCode") or item.get("materialCode"), "name": item.get("productName") or item.get("materialName"), "unit": item.get("unit"), "type": "FINISHED_PRODUCT"})
|
|
before = set(requirements)
|
|
root_token = f"{order_id}:{line_id}"
|
|
order_line_count += 1
|
|
walk(root_token=root_token, sales_order_id=order_id, sales_order_no=order_no, sales_order_line_id=line_id, material=material, quantity=quantity, unit=str(item.get("unit") or material.get("unit") or "PCS"), required_at=required_at, bom_depth=0, parent_id=None, ancestors=())
|
|
if normalize_date(raw_due) is None:
|
|
root_ids = sorted(set(requirements) - before)
|
|
if root_ids: _add_blocker(blockers, blocker_codes, code="INVALID_REQUIRED_DATE", entity_id=root_ids[0], message="订单交期缺失或无效，暂以业务日期作为占位", evidence={"businessDate": business_day, "orderNo": order_no})
|
|
|
|
candidates = _collect_supply_candidates(world, business_day, requirements, by_id, by_code)
|
|
external = [NettingRequirement(req.requirement_id, req.material_key, req.quantity, req.required_at, req.sourcing_type, order_priorities.get(req.sales_order_id, 0), req.sales_order_no) for req in requirements.values() if req.sourcing_type in {"BUY", "SUBCONTRACT"}]
|
|
netting = net_supply_requirements(external, candidates, business_date=business_day)
|
|
balances_by_id = {balance.requirement_id: balance for balance in netting.demand_balances}
|
|
for candidate in candidates:
|
|
matching = [req for req in requirements.values() if req.material_key == candidate.material_key and req.sourcing_type in {"BUY", "SUBCONTRACT"} and (not candidate.metadata.get("salesOrderId") or str(candidate.metadata.get("salesOrderId")) == req.sales_order_id)]
|
|
if _status(candidate.status) in _DRAFT_SUPPLY:
|
|
for req in matching: _add_blocker(blockers, blocker_codes, code="UNTRUSTED_DRAFT_SUPPLY", entity_id=req.requirement_id, message="DRAFT 采购/委外记录仅是建议，不可作为可信供应", evidence={"eventId": candidate.event_id, "status": candidate.status}, severity="HARD" if strict else "WARNING")
|
|
elif candidate.available_at is None and candidate.quantity > 0:
|
|
for req in matching: _add_blocker(blockers, blocker_codes, code="UNTRUSTED_SUPPLY_DATE", entity_id=req.requirement_id, message="在途/采购/委外供应缺少可信到达日期", evidence={"eventId": candidate.event_id, "status": candidate.status})
|
|
for req in requirements.values():
|
|
balance = balances_by_id.get(req.requirement_id)
|
|
if balance and balance.shortage_quantity > 0:
|
|
_add_blocker(blockers, blocker_codes, code="SUPPLY_SHORTAGE", entity_id=req.requirement_id, message="截止需求日期的可信供应不足", evidence={"requiredQuantity": balance.required_quantity, "allocatedQuantity": balance.allocated_quantity, "shortageQuantity": balance.shortage_quantity, "requiredAt": req.required_at})
|
|
for req_id, codes in blocker_codes.items():
|
|
if req_id in requirements: requirements[req_id].blocker_codes.update(codes)
|
|
|
|
frozen_requirements = tuple(Requirement(req.requirement_id, req.sales_order_id, req.sales_order_no, req.sales_order_line_id, req.material_id, req.material_key, req.material_code, req.material_name, req.quantity, req.unit, req.required_at, req.sourcing_type, req.bom_depth, tuple(sorted(req.parent_quantities)), tuple(sorted(req.child_ids)), tuple(PeggingRef(parent_id, qty) for parent_id, qty in sorted(req.parent_quantities.items())), req.operation_code, req.routing_status, req.resource_status, tuple(sorted(req.blocker_codes))) for req in sorted(requirements.values(), key=lambda row: row.requirement_id))
|
|
frozen_by_id = {req.requirement_id: req for req in frozen_requirements}
|
|
demand_id_by_req = {req.requirement_id: _stable_id("MFG", req.requirement_id) for req in frozen_requirements if req.sourcing_type == "MAKE"}
|
|
|
|
def descendant_codes(requirement_id: str, seen: set[str] | None = None) -> set[str]:
|
|
seen = set(seen or ())
|
|
if requirement_id in seen: return {"BOM_CYCLE"}
|
|
seen.add(requirement_id)
|
|
req = frozen_by_id[requirement_id]
|
|
result = set(req.blocker_codes)
|
|
for child_id in req.child_requirement_ids: result.update(descendant_codes(child_id, seen))
|
|
return result
|
|
manufacturing_demands: list[ManufacturingDemand] = []
|
|
make_events: list[SupplyEvent] = []
|
|
make_balances: list[DemandBalance] = []
|
|
for req in frozen_requirements:
|
|
if req.sourcing_type != "MAKE": continue
|
|
all_codes = tuple(sorted(descendant_codes(req.requirement_id)))
|
|
child_rows = [frozen_by_id[child_id] for child_id in req.child_requirement_ids]
|
|
material_status = "BLOCKED" if any("SUPPLY_SHORTAGE" in child.blocker_codes for child in child_rows) else ("PENDING_DEPENDENCIES" if child_rows else "READY")
|
|
manufacturing_demands.append(ManufacturingDemand(demand_id_by_req[req.requirement_id], req.requirement_id, req.sales_order_id, req.sales_order_no, req.sales_order_line_id, req.material_id, req.material_code, req.material_name, req.quantity, req.unit, req.required_at, req.bom_depth, tuple(sorted(demand_id_by_req[parent] for parent in req.parent_requirement_ids if parent in demand_id_by_req)), req.child_requirement_ids, req.routing_status, req.resource_status, material_status, "BLOCKED" if all_codes else "READY_FOR_SCHEDULING", all_codes))
|
|
make_events.append(SupplyEvent(f"MAKE:{req.requirement_id}", req.material_key, req.quantity, None, "MAKE", "PLANNED", False, req.requirement_id, (SupplyAllocation(req.requirement_id, req.quantity),), 0.0, {"requiresScheduling": True}))
|
|
make_balances.append(DemandBalance(req.requirement_id, req.quantity, req.quantity, 0.0))
|
|
all_events = tuple(sorted(tuple(netting.supply_events) + tuple(make_events), key=lambda event: (event.material_key, event.event_id)))
|
|
all_balances = tuple(sorted(tuple(netting.demand_balances) + tuple(make_balances), key=lambda balance: balance.requirement_id))
|
|
frozen_blockers = tuple(sorted(blockers.values(), key=lambda row: row.blocker_id))
|
|
counts = Counter(blocker.code for blocker in frozen_blockers)
|
|
source_hash = _source_revision(world, business_day)
|
|
summary = {
|
|
"businessDate": business_day, "strict": bool(strict), "orderCount": order_count,
|
|
"orderLineCount": order_line_count, "requirementCount": len(frozen_requirements),
|
|
"makeCount": sum(req.sourcing_type == "MAKE" for req in frozen_requirements),
|
|
"buyCount": sum(req.sourcing_type == "BUY" for req in frozen_requirements),
|
|
"subcontractCount": sum(req.sourcing_type == "SUBCONTRACT" for req in frozen_requirements),
|
|
"blockedRequirementCount": sum(bool(req.blocker_codes) for req in frozen_requirements),
|
|
"supplyEventCount": len(all_events), "trustedSupplyEventCount": sum(event.trusted for event in all_events),
|
|
"blockerCounts": dict(sorted(counts.items())),
|
|
}
|
|
return ClosedLoopProblem(_SCHEMA_VERSION, f"CLP-{business_day.replace('-', '')}-{source_hash[:16]}", source_hash, business_day, frozen_requirements, tuple(sorted(manufacturing_demands, key=lambda row: row.demand_id)), all_events, all_balances, frozen_blockers, summary)
|
|
|
|
__all__ = ["Blocker", "ClosedLoopProblem", "ManufacturingDemand", "PeggingRef", "Requirement", "build_closed_loop_problem"]
|