492 lines
29 KiB
Python
492 lines
29 KiB
Python
"""Complete planning workbooks: canonical business roles plus validated JSON adapters."""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import hashlib
|
||
import io
|
||
import math
|
||
from datetime import date, datetime
|
||
from typing import Any
|
||
|
||
from server.importers.workbook_profiles import ROLE_FIELDS, get_profile, select_profile
|
||
|
||
ROLE_KINDS = {role: "materials" if role == "products" else role for role in ROLE_FIELDS}
|
||
|
||
|
||
def _enum(profile: dict, enum: str, value: Any) -> Any:
|
||
key = _text(value)
|
||
aliases = profile["enums"][enum]
|
||
if key not in aliases:
|
||
raise ValueError(f"{enum}的值“{key}”没有配置标准映射")
|
||
return aliases[key]
|
||
|
||
|
||
def _list(profile: dict, value: Any) -> list[str]:
|
||
return [item.strip() for item in _text(value).split(profile["separators"]["list"]) if item.strip()]
|
||
|
||
|
||
def _canonical_row(role: str, raw: dict, profile: dict) -> dict:
|
||
"""Physical columns are resolved before this function; only roles occur here."""
|
||
row = dict(raw)
|
||
if role in ("sourceNotes", "planningParameters", "sandboxScenario"):
|
||
row["key"] = profile["metadataKeys"][role].get(_text(row["key"]), _text(row["key"]))
|
||
if role == "factoryResources":
|
||
row["resourceType"] = row["resourceKind"]
|
||
row["resourceKind"] = _enum(profile, "resourceKind", row["resourceKind"])
|
||
row["status"] = _enum(profile, "resourceStatus", row["status"])
|
||
row["timeZone"] = profile["planning"]["timeZone"]
|
||
row["isBottleneck"] = any(marker in _text(row["description"])
|
||
for marker in profile["provenance"]["bottleneckMarkers"])
|
||
elif role == "equipment":
|
||
row["status"] = _enum(profile, "equipmentStatus", row["status"])
|
||
row["availabilityRate"] = _number(row["availabilityRate"], "可用率", positive=True)
|
||
if row["availabilityRate"] > 1:
|
||
raise ValueError("可用率不能大于1")
|
||
row["capabilities"] = _list(profile, row["capabilities"])
|
||
if not row["capabilities"]:
|
||
raise ValueError("设备能力没有填写")
|
||
elif role == "personnel":
|
||
row["skills"] = _list(profile, row["skills"])
|
||
if not row["skills"] or row["skillLevel"] not in profile["planning"]["skillLevelOrder"]:
|
||
raise ValueError("人员技能缺失或技能等级未配置")
|
||
elif role == "calendar":
|
||
event = profile["enums"]["calendarEvent"].get(_text(row["eventCode"]))
|
||
if event == "MAINTENANCE":
|
||
start, end = _instant(row["start"], "维护开始"), _instant(row["end"], "维护结束")
|
||
if end <= start:
|
||
raise ValueError("维护结束必须晚于开始")
|
||
return {"entityType": "maintenance", "code": f"{row['workdaysOrEquipment']}:{start}",
|
||
"name": row["name"], "equipmentCode": row["workdaysOrEquipment"],
|
||
"start": start, "end": end, "reason": row["statusOrReason"]}
|
||
start, end = _text(row["start"]), _text(row["end"])
|
||
datetime.strptime(start, "%H:%M") # noqa: DTZ007 - source local shift clock
|
||
datetime.strptime(end, "%H:%M") # noqa: DTZ007 - source local shift clock
|
||
days = [int(day) for day in _list(profile, row["workdaysOrEquipment"])]
|
||
if not days or any(day not in range(1, 8) for day in days):
|
||
raise ValueError("工作日必须为1至7")
|
||
breaks = []
|
||
for interval in _list(profile, row["breaks"]):
|
||
bs, be = interval.split(profile["separators"]["interval"])
|
||
datetime.strptime(bs, "%H:%M") # noqa: DTZ007 - source local break clock
|
||
datetime.strptime(be, "%H:%M") # noqa: DTZ007 - source local break clock
|
||
breaks.append({"start": bs, "end": be})
|
||
return {"entityType": "shift", "shiftCode": row["eventCode"], "name": row["name"],
|
||
"startTime": start, "endTime": end, "breaks": breaks, "workdays": days,
|
||
"sourceStatus": row["statusOrReason"], "enabled": _enum(profile, "enabled", row["statusOrReason"])}
|
||
elif role in ("products", "materials"):
|
||
row["type"] = _enum(profile, "materialType", row["type"])
|
||
row["sourcingType"] = _enum(profile, "sourcingType", row["sourcingType"])
|
||
if not _text(row["unit"]):
|
||
raise ValueError("物料单位缺失")
|
||
for key in ("stock", "safetyStock", "procurementLeadTime"):
|
||
if key in row:
|
||
row[key] = _number(row[key], key)
|
||
elif role == "inventory":
|
||
# A snapshot never carries authority to replace the material name/unit.
|
||
row.pop("name")
|
||
for key in ("stock", "inTransit", "safetyStock", "procurementLeadTime"):
|
||
row[key] = _number(row[key], key)
|
||
row["expectedArrivalDate"] = _date(row["expectedArrivalDate"], "预计到货", optional=True)
|
||
row["inventorySource"] = profile["provenance"]["inventorySource"]
|
||
elif role == "bom":
|
||
row["quantity"] = _number(row["quantity"], "单套用量", positive=True)
|
||
row["lossRate"] = _number(row["lossRate"], "损耗率")
|
||
if row["lossRate"] >= 1:
|
||
raise ValueError("损耗率必须小于1")
|
||
row["isKey"] = _enum(profile, "yesNo", row["isKey"])
|
||
elif role == "routing":
|
||
seq = _number(row["seq"], "工序顺序", positive=True)
|
||
if seq != int(seq):
|
||
raise ValueError("工序顺序必须是整数")
|
||
row["seq"] = int(seq)
|
||
row["stdTimePerUnit"] = _number(row["stdTimePerUnit"], "标准工时", positive=True)
|
||
row["isExternal"] = _enum(profile, "yesNo", row["isExternal"])
|
||
row["stdTimeSource"] = "demo" if any(marker in _text(row["sourceText"])
|
||
for marker in profile["provenance"]["demoMarkers"]) else profile["provenance"]["defaultTimeSource"]
|
||
row["timeConfirmed"] = False
|
||
elif role == "partners":
|
||
row["partnerType"] = _enum(profile, "partnerType", row["partnerType"])
|
||
elif role == "orders":
|
||
row["orderType"] = _enum(profile, "orderType", row["orderType"])
|
||
row["sourceStatus"] = row["status"]
|
||
row["status"] = _enum(profile, "orderStatus", row["status"])
|
||
row["isSandbox"] = row["orderType"] == "SANDBOX"
|
||
expected_status = "PENDING_EVALUATION" if row["isSandbox"] else "RELEASED"
|
||
if row["status"] != expected_status:
|
||
raise ValueError("订单状态与正式/沙盒分类不一致")
|
||
row["quantity"] = _number(row["quantity"], "订单数量", positive=True)
|
||
row["priority"] = _number(row["priority"], "优先级", positive=True)
|
||
row["orderDate"] = _date(row["orderDate"], "订单日期")
|
||
row["dueDate"] = _date(row["dueDate"], "交期")
|
||
row["deliveryDate"] = row["dueDate"]
|
||
elif role == "wip":
|
||
row["sourceStatus"] = row["status"]
|
||
row["status"] = _enum(profile, "wipStatus", row["status"])
|
||
row["completedQuantity"] = _number(row["completedQuantity"], "完成数量")
|
||
row["completionTime"] = _instant(row["completionTime"], "实际/预计完成", optional=True)
|
||
row["completionTimeMeaning"] = "actual_or_expected_unspecified"
|
||
elif role == "sourceValidation":
|
||
row["verification"] = "source_reported_only"
|
||
return row
|
||
|
||
|
||
def _build_context(rows: dict, profile: dict, digest: str) -> dict:
|
||
notes = {r["key"]: r["value"] for r in rows["sourceNotes"]}
|
||
params = {r["key"]: r for r in rows["planningParameters"]}
|
||
data_date = _date(notes.get("dataDate"), "资料日期")
|
||
if not _text(notes.get("boundary")):
|
||
raise ValueError("缺少资料来源边界说明")
|
||
day_shifts = [r for r in rows["calendar"] if r["entityType"] == "shift" and r["enabled"]]
|
||
if not day_shifts:
|
||
raise ValueError("没有启用的班次，不能创建计划起点")
|
||
for key in ("horizonDays", "freezeHours", "nightShiftEnabled"):
|
||
if params[key]["delivery"] != params[key]["bottleneck"]:
|
||
raise ValueError(f"两种方案的{key}不同，需分别配置")
|
||
modes = profile["planning"]["sortModeAliases"]
|
||
strategies = [params["sortMode"][key] for key in ("delivery", "bottleneck")]
|
||
if any(strategy not in modes for strategy in strategies):
|
||
raise ValueError("资料策略没有配置标准排序模式")
|
||
return {"sourceProfile": profile["id"], "sourceSha256": digest, "profileDigest": profile["profileDigest"],
|
||
"capabilities": profile["capabilities"], "dataDate": data_date,
|
||
"planStart": f"{data_date}T{min(r['startTime'] for r in day_shifts)}:00",
|
||
"planStartSource": "data_date_and_first_enabled_shift",
|
||
"horizonDays": _number(params["horizonDays"]["delivery"], "计划周期", positive=True),
|
||
"freezeHours": _number(params["freezeHours"]["delivery"], "冻结窗口"),
|
||
"nightShiftEnabled": _enum(profile, "enabled", params["nightShiftEnabled"]["delivery"]),
|
||
"strategies": strategies, "defaultSortMode": modes[strategies[0]],
|
||
"comparisonMetrics": params["comparisonMetrics"]["delivery"],
|
||
"trialOnly": True, "boundary": notes["boundary"],
|
||
"timeZone": profile["planning"]["timeZone"], "skillLevelOrder": profile["planning"]["skillLevelOrder"],
|
||
"planningDefaultsSource": "profile-config",
|
||
"sourceRef": {"sha256": digest, "sheet": profile["sheets"]["planningParameters"]["name"],
|
||
"role": "planningParameters"}}
|
||
|
||
|
||
def preview_planning_workbook(filename: str, raw: bytes) -> dict | None:
|
||
import openpyxl
|
||
|
||
workbook = openpyxl.load_workbook(io.BytesIO(raw), read_only=True, data_only=True)
|
||
try:
|
||
physical_headers = {sheet.title: [_text(v) for v in next(sheet.values, ())] for sheet in workbook}
|
||
profile = select_profile(physical_headers)
|
||
if profile is None:
|
||
return None
|
||
digest = hashlib.sha256(raw).hexdigest()
|
||
batches, diagnostics = [], []
|
||
physical_roles = {spec["name"]: role for role, spec in profile["sheets"].items()}
|
||
|
||
def issue(sheet, row, message, code="INVALID_PROFILE_DATA", severity="blocking"):
|
||
role = physical_roles.get(sheet)
|
||
diagnostics.append({"sheet": sheet, "excelRow": row, "role": role,
|
||
"kind": ROLE_KINDS.get(role), "code": code, "severity": severity, "message": message})
|
||
|
||
for role, spec in profile["sheets"].items():
|
||
sheet, columns = spec["name"], spec["columns"]
|
||
batch = {"sheet": sheet, "physicalSheet": sheet, "role": role, "kind": ROLE_KINDS[role],
|
||
"sourceProfile": profile["id"], "profileDigest": profile["profileDigest"],
|
||
"capabilities": profile["capabilities"], "sourceFile": filename, "sourceSha256": digest,
|
||
"contractSheets": list(physical_roles), "contractRoles": list(ROLE_FIELDS),
|
||
"okRows": [], "headersRaw": list(columns.values()),
|
||
"fieldMap": [{"source": v, "target": k} for k, v in columns.items()]}
|
||
batches.append(batch)
|
||
if sheet not in physical_headers:
|
||
issue(sheet, None, f"缺少工作表“{sheet}”，整份资料暂不导入")
|
||
continue
|
||
headers = physical_headers[sheet]
|
||
required = list(columns.values())
|
||
if any(key not in headers for key in required) or len(headers) != len(set(headers)):
|
||
issue(sheet, 1, "必要列缺失或表头重复：" + "、".join(key for key in required if key not in headers))
|
||
continue
|
||
for number, line in enumerate(list(workbook[sheet].values)[1:], 2):
|
||
if all(v is None or _text(v) == "" for v in line):
|
||
continue
|
||
source = {key: _json_value(line[i]) if i < len(line) else None for i, key in enumerate(headers) if key}
|
||
try:
|
||
row = _canonical_row(role, {key: source[physical] for key, physical in columns.items()}, profile)
|
||
identity = next((row[k] for k in ("code", "orderNo", "taskNo", "productCode", "key", "shiftCode") if k in row), None)
|
||
if not _text(identity):
|
||
raise ValueError("业务编码或字段名缺失")
|
||
row.update(sourceProfile=profile["id"],
|
||
sourceRef={"sha256": digest, "sheet": sheet, "role": role, "excelRow": number},
|
||
sourceValues=source, provenance={"confirmation": "unconfirmed", "trialOnly": True})
|
||
batch["okRows"].append(row)
|
||
except (ValueError, TypeError, KeyError) as exc:
|
||
issue(sheet, number, str(exc))
|
||
for name in set(physical_headers) - set(physical_roles):
|
||
issue(name, None, f"工作表“{name}”未配置用途，不能宣称完整采用", "UNMAPPED_WORKSHEET")
|
||
by_role = {b["role"]: b["okRows"] for b in batches}
|
||
_validate_references(by_role, issue)
|
||
context = None
|
||
try:
|
||
context = _build_context(by_role, profile, digest)
|
||
except (ValueError, TypeError, KeyError) as exc:
|
||
issue(profile["sheets"]["planningParameters"]["name"], None, str(exc))
|
||
note = next((r for r in by_role["sourceNotes"] if r["key"] == "boundary"), {})
|
||
boundary = _text(note.get("value"))
|
||
issue(profile["sheets"]["sourceNotes"]["name"], note.get("sourceRef", {}).get("excelRow"),
|
||
boundary + "；本次只用于试排，工时、库存和订单仍需现场确认。", "UNCONFIRMED_DEMO_VALUES", "warning")
|
||
counts = _entity_counts(by_role)
|
||
blocking = sum(d["severity"] == "blocking" for d in diagnostics)
|
||
for batch in batches:
|
||
ds = [d for d in diagnostics if d["sheet"] == batch["sheet"]]
|
||
batch.update(diagnostics=ds, diagnosticCounts={s: sum(d["severity"] == s for d in ds)
|
||
for s in ("blocking", "warning", "ignored")}, errors=[d["message"] for d in ds if d["severity"] == "blocking"],
|
||
warnings=[d["message"] for d in ds if d["severity"] == "warning"], okCount=len(batch["okRows"]),
|
||
errorCount=sum(d["severity"] == "blocking" for d in ds), canCommit=blocking == 0)
|
||
batch["entityCount"] = counts[batch["role"]] if batch["role"] in ("orders", "calendar") else len(batch["okRows"])
|
||
return {"filename": filename, "profile": profile["id"], "profileDigest": profile["profileDigest"],
|
||
"capabilities": profile["capabilities"], "batches": batches, "planningContext": context,
|
||
"source": {"sha256": digest, "profileDigest": profile["profileDigest"],
|
||
"dataDate": (context or {}).get("dataDate"), "boundary": boundary, "trialOnly": True},
|
||
"entityCounts": counts, "totalOk": sum(b["okCount"] for b in batches), "totalErrors": blocking,
|
||
"canCommit": blocking == 0, "diagnostics": diagnostics, "diagnosticsVersion": 1,
|
||
"canSchedule": blocking == 0 and not any(d["code"] in ("NO_PERSONNEL_SKILL", "UNKNOWN_WIP_EQUIPMENT",
|
||
"UNCONFIRMED_WIP_HISTORY") for d in diagnostics),
|
||
"diagnosticCounts": {s: sum(d["severity"] == s for d in diagnostics) for s in ("blocking", "warning", "ignored")},
|
||
"sheetSummary": [{"physicalSheet": b["sheet"], "role": b["role"], "count": b["okCount"],
|
||
"usable": b["errorCount"] == 0, "unsupported": False, "needsReview": bool(b["diagnostics"])} for b in batches]}
|
||
finally:
|
||
workbook.close()
|
||
|
||
|
||
def apply_planning_batches(world: dict, batches: list[dict]) -> dict:
|
||
"""Only a reviewed complete snapshot with its current mapping can be adopted."""
|
||
ids = {b.get("sourceProfile") for b in batches}
|
||
if len(ids) != 1 or not next(iter(ids)):
|
||
raise ValueError("完整资料必须来自同一已注册格式")
|
||
profile = get_profile(next(iter(ids)))
|
||
sheets = {b.get("role"): b for b in batches}
|
||
if set(sheets) != set(ROLE_FIELDS) or len(batches) != len(ROLE_FIELDS):
|
||
raise ValueError(f"必须一次确认完整的{len(ROLE_FIELDS)}张资料表，不能拆分导入")
|
||
if any(b.get("profileDigest") != profile["profileDigest"] for b in batches):
|
||
raise ValueError("工作簿映射配置已变化或确认缺少配置版本，请重新检查并发起确认；本次未写入")
|
||
if any(b.get("physicalSheet") != profile["sheets"][b["role"]]["name"] for b in batches):
|
||
raise ValueError("确认批次的物理工作表与注册配置不一致")
|
||
if any(b.get("canCommit") is not True or b.get("errorCount", 0) for b in batches):
|
||
raise ValueError("资料检查未通过，整份工作簿没有写入")
|
||
hashes = {b.get("sourceSha256") for b in batches}
|
||
if len(hashes) != 1 or not next(iter(hashes)):
|
||
raise ValueError("资料表必须来自同一份已检查文件")
|
||
digest = next(iter(hashes))
|
||
adopted = [(s.get("profile"), s.get("sha256"), s.get("profileDigest")) for s in world.get("intakeSources", [])]
|
||
context = world.get("planningContext") or {}
|
||
if context.get("sourceProfile"):
|
||
adopted.append((context["sourceProfile"], context.get("sourceSha256") or (context.get("sourceRef") or {}).get("sha256"),
|
||
context.get("profileDigest")))
|
||
if any(sha and (pid != profile["id"] or sha != digest) for pid, sha, _ in adopted):
|
||
raise ValueError("当前项目已采用另一版本的完整排产资料。新版需在独立项目核对或经过差异审核后替换；本次未写入。")
|
||
if any(pid == profile["id"] and sha == digest and pd != profile["profileDigest"] for pid, sha, pd in adopted):
|
||
raise ValueError("已采用资料未记录当前映射版本或配置已变化，请先核对映射差异；原数据保持不变")
|
||
by_role = {role: copy.deepcopy(b["okRows"]) for role, b in sheets.items()}
|
||
counts = _entity_counts(by_role)
|
||
if any(pid == profile["id"] and sha == digest for pid, sha, _ in adopted):
|
||
return {"summary": counts, "total": 0, "unchanged": True, "sourceSha256": digest, "profile": profile["id"]}
|
||
candidate = copy.deepcopy(world)
|
||
tables = {"factoryResources": ("flexFactoryResources", ("code",)), "equipment": ("flexEquipment", ("code",)),
|
||
"personnel": ("flexPersonnel", ("code",)), "products": ("flexMaterials", ("code",)),
|
||
"materials": ("flexMaterials", ("code",)), "bom": ("flexBom", ("productCode", "materialCode")),
|
||
"routing": ("flexRoutings", ("productCode", "seq")), "partners": ("flexPartners", ("code",)),
|
||
"wip": ("flexWip", ("taskNo",))}
|
||
for role, (table, keys) in tables.items():
|
||
for row in by_role[role]:
|
||
_upsert(candidate, table, row, keys)
|
||
for row in by_role["inventory"]:
|
||
inventory = {k: v for k, v in row.items() if k not in ("sourceRef", "sourceValues", "provenance")}
|
||
inventory.update(inventorySourceRef=row["sourceRef"], inventorySourceValues=row["sourceValues"])
|
||
_upsert(candidate, "flexMaterials", inventory, ("code",))
|
||
for row in by_role["orders"]:
|
||
if row["isSandbox"] and any(o.get("orderNo") == row["orderNo"] for o in candidate.get("flexOrders", [])):
|
||
raise ValueError(f"{row['orderNo']}已存在正式订单，不能静默当作沙盒重复导入，请核对订单来源")
|
||
_upsert(candidate, "flexSandboxOrders" if row["isSandbox"] else "flexOrders", row, ("orderNo",))
|
||
for row in by_role["calendar"]:
|
||
maintenance = row["entityType"] == "maintenance"
|
||
_upsert(candidate, "flexMaintenance" if maintenance else "flexCalendar", row, ("code",) if maintenance else ("shiftCode",))
|
||
for row in by_role["factoryResources"]:
|
||
if row["resourceKind"] == "OPERATION":
|
||
_upsert(candidate, "flexOperations", row, ("code",))
|
||
elif row["resourceKind"] == "ZONE":
|
||
_upsert(candidate, "flexZones", row, ("code",))
|
||
groups = {}
|
||
for row in by_role["personnel"]:
|
||
groups.setdefault((row["teamName"], tuple(sorted(row["skills"])), row["shiftCode"]), []).append(row)
|
||
for (name, skills, shift), people in groups.items():
|
||
code = "TEAM-" + hashlib.sha256(f"{name}|{','.join(skills)}|{shift}".encode()).hexdigest()[:12]
|
||
_upsert(candidate, "flexTeams", {"code": code, "name": name, "supportOps": list(skills), "memberCount": len(people),
|
||
"personCodes": [p["code"] for p in people], "shiftCode": shift, "capacitySource": "distinct_personnel",
|
||
"sourceProfile": profile["id"]}, ("code",))
|
||
candidate["planningContext"] = _build_context(by_role, profile, digest)
|
||
scenario = {r["key"]: r["value"] for r in by_role["sandboxScenario"]}
|
||
if scenario:
|
||
_upsert(candidate, "flexScenarios", {**scenario, "sourceProfile": profile["id"],
|
||
"status": "PENDING_EVALUATION", "sourceSha256": digest}, ("orderNo",))
|
||
candidate.setdefault("flexParams", {}).update(siteProfile=profile["id"], demoDataCleared=True)
|
||
from server.aps_domain.orders import sync_flex_orders_to_sales
|
||
from server.importers.sql_pack import sync_flex_to_classic_master
|
||
|
||
sync_flex_to_classic_master(candidate)
|
||
sync_flex_orders_to_sales(candidate)
|
||
diagnostics = [copy.deepcopy(d) for b in batches for d in b.get("diagnostics", [])]
|
||
candidate.setdefault("intakeSources", []).append({"profile": profile["id"], "profileDigest": profile["profileDigest"],
|
||
"capabilities": profile["capabilities"], "sha256": digest, "filename": batches[0].get("sourceFile"),
|
||
"dataDate": candidate["planningContext"]["dataDate"], "trialOnly": True, "entityCounts": counts,
|
||
"diagnostics": diagnostics, "notes": by_role["sourceNotes"], "parameters": by_role["planningParameters"],
|
||
"sourceValidation": by_role["sourceValidation"]})
|
||
world.clear()
|
||
world.update(candidate)
|
||
return {"summary": counts, "total": sum(counts.values()), "profile": profile["id"],
|
||
"unchanged": False, "sourceSha256": digest, "profileDigest": profile["profileDigest"], "diagnostics": diagnostics}
|
||
|
||
def _text(value: Any) -> str:
|
||
return str(value).strip() if value is not None else ""
|
||
|
||
|
||
def _number(value: Any, label: str, *, positive: bool = False) -> float:
|
||
if value is None or isinstance(value, bool):
|
||
raise ValueError(f"{label}缺失或不是数字")
|
||
try:
|
||
number = float(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError(f"{label}不是数字") from exc
|
||
if not math.isfinite(number) or number < 0 or (positive and number == 0):
|
||
raise ValueError(f"{label}必须是{'正' if positive else '非负'}有限数字")
|
||
return number
|
||
|
||
|
||
def _date(value: Any, label: str, *, optional: bool = False) -> str | None:
|
||
if not _text(value) and optional:
|
||
return None
|
||
try:
|
||
return datetime.fromisoformat(_text(value)).date().isoformat()
|
||
except ValueError as exc:
|
||
raise ValueError(f"{label}不是有效日期") from exc
|
||
|
||
|
||
def _instant(value: Any, label: str, *, optional: bool = False) -> str | None:
|
||
if not _text(value) and optional:
|
||
return None
|
||
try:
|
||
return datetime.fromisoformat(_text(value)).isoformat(timespec="seconds")
|
||
except ValueError as exc:
|
||
raise ValueError(f"{label}不是有效日期时间") from exc
|
||
|
||
|
||
def _json_value(value: Any) -> Any:
|
||
return value.isoformat() if isinstance(value, (date, datetime)) else value
|
||
|
||
|
||
def _iso_date(value: Any) -> str:
|
||
"""日期字段归一化：Excel 日期单元格序列化成 ISO 时间戳时仍按日期比较。"""
|
||
if isinstance(value, datetime):
|
||
return value.date().isoformat()
|
||
if isinstance(value, date):
|
||
return value.isoformat()
|
||
text = str(value or "").strip()
|
||
return text[:10] if ("T" in text or " " in text) else text
|
||
|
||
|
||
def _validate_references(sheets: dict, issue) -> None:
|
||
def rows(name):
|
||
return sheets[name]
|
||
|
||
def warn(row, message, code, severity="blocking"):
|
||
ref = row["sourceRef"]
|
||
issue(ref["sheet"], ref["excelRow"], message, code, severity)
|
||
|
||
unique_keys = {"factoryResources": ("code",), "equipment": ("code",), "personnel": ("code",),
|
||
"products": ("code",), "materials": ("code",), "bom": ("productCode", "materialCode"),
|
||
"routing": ("productCode", "seq"), "inventory": ("code",),
|
||
"orders": ("orderNo",), "wip": ("taskNo",),
|
||
"planningParameters": ("key",), "sandboxScenario": ("key",), "sourceNotes": ("key",)}
|
||
for sheet, keys in unique_keys.items():
|
||
seen = set()
|
||
for row in rows(sheet):
|
||
key = tuple(row[k] for k in keys)
|
||
if key in seen:
|
||
warn(row, f"重复业务编码：{key}", "DUPLICATE_BUSINESS_KEY")
|
||
seen.add(key)
|
||
materials = {r["code"] for s in ("products", "materials") for r in rows(s)}
|
||
products = {r["code"] for r in rows("products")}
|
||
equipment = {r["code"] for r in rows("equipment")}
|
||
orders = {r["orderNo"]: r for r in rows("orders")}
|
||
for row in rows("materials"):
|
||
if row["code"] in products:
|
||
warn(row, "同一编码同时出现在产品和物料档案", "DUPLICATE_MATERIAL")
|
||
for sheet in ("bom", "routing", "orders"):
|
||
for row in rows(sheet):
|
||
if row["productCode"] not in products:
|
||
warn(row, f"产品{row['productCode']}没有档案", "UNKNOWN_PRODUCT")
|
||
for row in rows("bom"):
|
||
if row["materialCode"] not in materials:
|
||
warn(row, f"材料{row['materialCode']}没有档案", "UNKNOWN_MATERIAL")
|
||
for row in rows("inventory"):
|
||
if row["code"] not in materials:
|
||
warn(row, f"库存材料{row['code']}没有档案", "UNKNOWN_MATERIAL")
|
||
if row["inTransit"] > 0 and not row["expectedArrivalDate"]:
|
||
warn(row, "有在途数量，但没有预计到货日期；排产不能假定已经到货", "UNKNOWN_ARRIVAL", "warning")
|
||
skills = {skill for row in rows("personnel") for skill in row["skills"]}
|
||
for row in rows("routing"):
|
||
if row["operationCode"] not in skills and not row["isExternal"]:
|
||
warn(row, f"{row['operationName']}没有对应技能人员，需核对才能安排", "NO_PERSONNEL_SKILL", "warning")
|
||
for row in rows("wip"):
|
||
candidates = [r for r in rows("routing") if r["productCode"] == row["productCode"]
|
||
and r["operationName"] == row["operationName"]]
|
||
if len(candidates) != 1:
|
||
warn(row, "在制任务的工序不能唯一关联到产品工艺", "INVALID_WIP_OPERATION")
|
||
else:
|
||
row.update(operationCode=candidates[0]["operationCode"], seq=candidates[0]["seq"])
|
||
order = orders.get(row["orderNo"])
|
||
if not order or order["productCode"] != row["productCode"] or order["isSandbox"]:
|
||
warn(row, "在制任务不能关联到对应正式订单", "INVALID_WIP_ORDER")
|
||
elif row["completedQuantity"] > order["quantity"]:
|
||
warn(row, "在制完成数量超过订单数量", "INVALID_WIP_QUANTITY")
|
||
if row["equipmentCode"] not in equipment:
|
||
warn(row, f"在制任务引用设备{row['equipmentCode']}，设备档案未提供，需核对",
|
||
"UNKNOWN_WIP_EQUIPMENT", "warning")
|
||
if row["status"] == "RUNNING":
|
||
row.update(expectedEnd=row["completionTime"], endTimeSource="status_and_source_completion_column")
|
||
warn(row, "执行中任务按原表时间保留预计结束；实际开工和前序完成事实仍需确认",
|
||
"UNCONFIRMED_WIP_HISTORY", "warning")
|
||
elif row["status"] == "DONE":
|
||
row.update(actualEnd=row["completionTime"], endTimeSource="status_and_source_completion_column")
|
||
scenario = {r["key"]: r["value"] for r in rows("sandboxScenario")}
|
||
if rows("sandboxScenario"):
|
||
order = orders.get(scenario.get("orderNo"))
|
||
if not order or not order["isSandbox"]:
|
||
warn(rows("sandboxScenario")[0], "插单场景必须关联到一张沙盒订单", "INVALID_SANDBOX")
|
||
else:
|
||
for key in ("productCode", "dueDate", "quantity", "priority", "customerName", "customerLevel"):
|
||
value = scenario.get(key)
|
||
try:
|
||
if key in ("quantity", "priority"):
|
||
equal = float(value) == float(order[key])
|
||
elif key == "dueDate":
|
||
equal = _iso_date(value) == _iso_date(order[key])
|
||
else:
|
||
equal = value == order[key]
|
||
except (ValueError, TypeError):
|
||
equal = False
|
||
if not equal:
|
||
warn(rows("sandboxScenario")[0], f"插单场景的{key}与销售订单不一致", "SANDBOX_CONFLICT")
|
||
|
||
|
||
def _entity_counts(sheets: dict) -> dict:
|
||
return {"materials": len({r["code"] for s in ("products", "materials") for r in sheets[s]}),
|
||
"equipment": len(sheets["equipment"]), "bom": len(sheets["bom"]), "routing": len(sheets["routing"]),
|
||
"orders": sum(not r["isSandbox"] for r in sheets["orders"]),
|
||
"sandboxOrders": sum(r["isSandbox"] for r in sheets["orders"]),
|
||
"personnel": len(sheets["personnel"]), "wip": len(sheets["wip"]),
|
||
"calendar": sum(r["entityType"] == "shift" for r in sheets["calendar"]),
|
||
"maintenance": sum(r["entityType"] == "maintenance" for r in sheets["calendar"]),
|
||
"inventory": len(sheets["inventory"])}
|
||
|
||
|
||
def _upsert(world: dict, table: str, row: dict, keys: tuple[str, ...]) -> dict:
|
||
values = world.setdefault(table, [])
|
||
existing = next((r for r in values if all(r.get(k) == row.get(k) for k in keys)), None)
|
||
if existing is None:
|
||
identifier = max((r.get("id", 0) for r in values if isinstance(r.get("id"), int)), default=0) + 1
|
||
existing = {"id": identifier}
|
||
values.append(existing)
|
||
existing.update(copy.deepcopy(row))
|
||
return existing
|