151 lines
10 KiB
Python
151 lines
10 KiB
Python
"""Explicit, fingerprint-bound recovery of legacy import classification conflicts."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
|
|
def _fingerprint(value: Any) -> str:
|
|
return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":"), default=str).encode()).hexdigest()
|
|
|
|
|
|
def _same_order(existing: dict, incoming: dict) -> bool:
|
|
try:
|
|
return (existing.get("productCode") == incoming.get("productCode")
|
|
and float(existing.get("quantity")) == float(incoming.get("quantity"))
|
|
and str(existing.get("dueDate") or existing.get("deliveryDate") or "")[:10]
|
|
== str(incoming.get("dueDate") or incoming.get("deliveryDate") or "")[:10])
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def _protected_reason(world: dict, order: dict, sales: list[dict]) -> str | None:
|
|
number = order.get("orderNo")
|
|
if order.get("status") not in {"CREATED", "RELEASED", "DRAFT"}:
|
|
return "已有订单状态不属于可整理的未执行记录"
|
|
if order.get("source") not in {None, "", "IMPORT", "FILE", "EXCEL"} or order.get("sourceProfile"):
|
|
return "已有订单有独立系统或已采用资料来源，不能自动改变分类"
|
|
if any(row.get("source") != "FLEX" or row.get("status") not in {"APPROVED", "DRAFT", "CONFIRMED"}
|
|
for row in sales):
|
|
return "订单台账中有独立维护或执行记录，需要先核对"
|
|
if any(row.get("changes") or row.get("masterOverrides") or row.get("manualPriority") is not None
|
|
or row.get("reviewNote") or row.get("isRush") or row.get("rushStrategy") for row in [order, *sales]):
|
|
return "已有订单包含人工维护或审核记录，需要先核对，不能由原表覆盖"
|
|
if any(row.get("orderNo") == number for row in world.get("flexWip", [])):
|
|
return "该订单已有在制记录，不能转回待评估插单"
|
|
sale_ids = {row.get("id") for row in sales}
|
|
if any(row is not order and (row.get("salesOrderNo") == number or row.get("salesOrderId") in sale_ids)
|
|
for row in world.get("flexOrders", [])):
|
|
return "该订单已有分解后的生产需求，需先核对关联关系，不能直接改变分类"
|
|
for table in ("purchaseOrders", "outsourceOrders"):
|
|
if any((row.get("salesOrderId") in sale_ids or row.get("salesOrderNo") == number)
|
|
and row.get("status") not in {"DRAFT", "CANCELLED", "REJECTED"}
|
|
for row in world.get(table, [])):
|
|
return "该订单已有下达的采购或委外记录，不能自动改变分类"
|
|
production = [row for row in world.get("productionOrders", [])
|
|
if row.get("salesOrderId") in sale_ids or row.get("salesOrderNo") == number or row.get("orderNo") == number]
|
|
production_ids = {row.get("id") for row in production}
|
|
work = [row for row in world.get("workOrders", []) if row.get("productionOrderId") in production_ids]
|
|
flexible = [row for row in world.get("flexWorkOrders", []) if row.get("flexOrderNo") == number or row.get("orderNo") == number]
|
|
versions = {row.get("id"): row for row in world.get("flexScheduleVersions", [])}
|
|
fixed_versions = {row.get("id"): row for row in world.get("scheduleVersions", [])}
|
|
protected_statuses = {"APPROVED", "RELEASED", "PUBLISHED", "DISPATCHED", "FROZEN", "RUNNING", "STARTED", "COMPLETED", "DONE"}
|
|
for row in flexible + work + production:
|
|
if (row.get("status") in protected_statuses or row.get("frozen") or row.get("locked")
|
|
or row.get("mesExternalId") or row.get("externalWoId")
|
|
or row.get("actualStartTime") or row.get("actualStart")
|
|
or float(row.get("progressPct") or 0) > 0 or float(row.get("qtyDone") or 0) > 0):
|
|
return "该订单已有发布、下发或生产进度，不能自动整理"
|
|
for rows, version_map in ((flexible, versions), (work + production, fixed_versions)):
|
|
if any(version_map.get(row.get("versionId"), {}).get("status") in protected_statuses for row in rows):
|
|
return "该订单已进入发布或冻结方案，不能自动整理"
|
|
for line in world.get("flexVirtualLines", []):
|
|
if line.get("orderNo") == number and versions.get(line.get("versionId"), {}).get("status") in protected_statuses:
|
|
return "该订单已进入发布或冻结方案，不能自动整理"
|
|
flex_ids, work_ids = {r.get("id") for r in flexible}, {r.get("id") for r in work}
|
|
for link in world.get("mesLinks", []):
|
|
if (link.get("orderNo") == number or link.get("flexOrderNo") == number
|
|
or (link.get("track") == "flex" and link.get("woId") in flex_ids)
|
|
or (link.get("track") == "fixed" and link.get("woId") in work_ids)):
|
|
return "该订单存在车间系统回执，不能自动整理"
|
|
return None
|
|
|
|
|
|
def review_intake_recovery(world: dict, batches: list[dict]) -> dict:
|
|
from server.aps_domain.folder_pack import folder_schedule_world_fingerprint
|
|
|
|
conflicts = []
|
|
for batch in batches:
|
|
if batch.get("role") != "orders":
|
|
continue
|
|
for incoming in batch.get("okRows", []):
|
|
number = incoming.get("orderNo")
|
|
existing = [row for row in world.get("flexOrders", []) if row.get("orderNo") == number]
|
|
sales = [row for row in world.get("salesOrders", []) if row.get("orderNo") == number]
|
|
if not existing and not sales:
|
|
continue
|
|
reason = None
|
|
if len(existing) != 1 or not _same_order(existing[0], incoming):
|
|
reason = "已有订单与文件中的产品、数量或交期不一致，需要先核对来源"
|
|
else:
|
|
reason = _protected_reason(world, existing[0], sales)
|
|
if existing[0].get("customerName") != incoming.get("customerName") or any(
|
|
row.get("customerName") != incoming.get("customerName") for row in sales
|
|
):
|
|
reason = "已有客户信息与原文件不同，需要先核对人工维护内容"
|
|
if any(row.get("priority") is not None and float(row["priority"]) != float(incoming.get("priority") or 0)
|
|
or row.get("customerLevel") and row["customerLevel"] != incoming.get("customerLevel")
|
|
for row in [existing[0], *sales]):
|
|
reason = "已有订单优先级或客户等级经过修改，需要先核对"
|
|
# Even a FLEX projection must refer to the same order content.
|
|
if any(len(row.get("items", [])) != 1 or not _same_order(
|
|
{**row["items"][0], "deliveryDate": row.get("deliveryDate")}, incoming) for row in sales):
|
|
reason = "订单台账的明细与本次资料不一致，需要先核对"
|
|
if not incoming.get("isSandbox") and reason is None:
|
|
continue
|
|
conflicts.append({"orderNo": number, "canRecover": reason is None and bool(incoming.get("isSandbox")),
|
|
"reason": reason or "旧导入列为正式待排，原文件标为待评估插单；可确认后恢复分类，未下达建议一并归档，历史方案保留",
|
|
"existingOrder": {key: (existing[0] if existing else {}).get(key) for key in ("productCode", "quantity", "dueDate", "customerName", "status")},
|
|
"incomingOrder": {key: incoming.get(key) for key in ("productCode", "quantity", "dueDate", "customerName", "status")},
|
|
"existingFingerprint": _fingerprint({"flexOrders": existing, "salesOrders": sales}),
|
|
"sourceRef": incoming.get("sourceRef")})
|
|
plan = {"version": 1, "conflicts": conflicts, "worldFingerprint": folder_schedule_world_fingerprint(world)}
|
|
return {**plan, "fingerprint": _fingerprint(plan)}
|
|
|
|
|
|
def apply_reconciled_intake(world: dict, next_id, batches: list[dict], recovery: dict) -> dict:
|
|
from server.aps_domain.importers import apply_import_commit
|
|
|
|
current = review_intake_recovery(world, batches)
|
|
if current != recovery or not all(row["canRecover"] for row in current["conflicts"]):
|
|
raise ValueError("已有订单状态或整理范围已变化，请重新检查资料；本次未采用")
|
|
candidate = copy.deepcopy(world)
|
|
numbers = {row["orderNo"] for row in current["conflicts"]}
|
|
archived = {"flexOrders": [copy.deepcopy(row) for row in candidate.get("flexOrders", []) if row.get("orderNo") in numbers],
|
|
"salesOrders": [copy.deepcopy(row) for row in candidate.get("salesOrders", []) if row.get("orderNo") in numbers]}
|
|
sales_ids = {row["id"] for row in archived["salesOrders"]}
|
|
for table in ("purchaseOrders", "outsourceOrders", "makeSuggestions"):
|
|
archived[table] = [copy.deepcopy(row) for row in candidate.get(table, [])
|
|
if row.get("status") == "DRAFT" and (row.get("salesOrderNo") in numbers or row.get("salesOrderId") in sales_ids)]
|
|
candidate[table] = [row for row in candidate.get(table, []) if row not in archived[table]]
|
|
candidate["flexOrders"] = [row for row in candidate.get("flexOrders", []) if row.get("orderNo") not in numbers]
|
|
# Keep a non-schedulable tombstone so legacy foreign keys/IDs cannot be
|
|
# silently reused after restart. The original rows are also archived above.
|
|
candidate["salesOrders"] = [
|
|
{**row, "source": "LEGACY_INTAKE_ARCHIVE", "status": "CANCELLED", "reclassifiedAs": "sandbox"}
|
|
if row.get("orderNo") in numbers else row for row in candidate.get("salesOrders", [])
|
|
]
|
|
result = apply_import_commit(candidate, next_id, batches)
|
|
source = next((item for item in candidate.get("intakeSources", []) if item.get("sha256") == result.get("sourceSha256")), None)
|
|
if source is None:
|
|
raise ValueError("资料采用未返回来源记录，本次整理未保存")
|
|
if numbers:
|
|
source["legacyReconciliation"] = {"plan": current, "archivedRecords": archived,
|
|
"meaning": "explicitly-approved sandbox reclassification; schedule history retained"}
|
|
world.clear()
|
|
world.update(candidate)
|
|
return {**result, "recoveredOrderNos": sorted(numbers)}
|