179 lines
7.2 KiB
Python
179 lines
7.2 KiB
Python
# ============================================================
|
||
# 结构化数据导入（moduleId: domain-intake, 可重生 ✅）
|
||
# Pi 负责把自然语言规范化为 kind/rows；本模块只做字段校验与确定性归一化。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
from typing import Any
|
||
|
||
World = dict[str, Any]
|
||
|
||
IMPORT_KINDS = frozenset({"orders", "materials"})
|
||
MATERIAL_TYPES = frozenset({"FINISHED_PRODUCT", "SEMI_FINISHED", "RAW_MATERIAL"})
|
||
CUSTOMER_LEVELS = frozenset({"VIP", "A", "B", "C"})
|
||
|
||
|
||
def _import_error(kind: str | None, message: str) -> dict[str, Any]:
|
||
return {"kind": kind, "rows": [], "errors": [message]}
|
||
|
||
|
||
def _as_positive_int(value: Any) -> int | None:
|
||
if isinstance(value, bool):
|
||
return None
|
||
try:
|
||
number = int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return number if number > 0 else None
|
||
|
||
|
||
def _exact_product(world: World, product_code: str) -> dict[str, Any] | None:
|
||
lookup = product_code.casefold()
|
||
for material in world.get("materials") or []:
|
||
if material.get("type") != "FINISHED_PRODUCT":
|
||
continue
|
||
if str(material.get("code") or "").casefold() == lookup:
|
||
return material
|
||
return None
|
||
|
||
|
||
def _normalize_order_row(world: World, row: Any, index: int) -> tuple[dict[str, Any] | None, str | None]:
|
||
if not isinstance(row, dict):
|
||
return None, f"第{index}行必须是 JSON 对象"
|
||
customer = str(row.get("customerName") or "").strip()
|
||
product_code = str(row.get("productCode") or "").strip()
|
||
quantity = _as_positive_int(row.get("quantity"))
|
||
delivery_date = str(row.get("deliveryDate") or "").strip()
|
||
if not customer:
|
||
return None, f"第{index}行缺少 customerName"
|
||
if not product_code:
|
||
return None, f"第{index}行缺少 productCode"
|
||
if quantity is None:
|
||
return None, f"第{index}行 quantity 必须是正整数"
|
||
try:
|
||
delivery_date = dt.date.fromisoformat(delivery_date).isoformat()
|
||
except ValueError:
|
||
return None, f"第{index}行 deliveryDate 必须是 YYYY-MM-DD"
|
||
product = _exact_product(world, product_code)
|
||
if product is None:
|
||
return None, f"第{index}行 productCode「{product_code}」未找到精确匹配的成品"
|
||
level = str(row.get("customerLevel") or "A").strip().upper()
|
||
if level not in CUSTOMER_LEVELS:
|
||
return None, f"第{index}行 customerLevel 必须是 VIP/A/B/C"
|
||
is_rush = bool(row.get("isRush", False))
|
||
priority = row.get("priority")
|
||
if priority is None:
|
||
priority = 1 if is_rush else 5
|
||
if isinstance(priority, bool) or not isinstance(priority, int):
|
||
try:
|
||
priority = int(priority)
|
||
except (TypeError, ValueError):
|
||
return None, f"第{index}行 priority 必须是整数"
|
||
return {
|
||
"customerName": customer,
|
||
"customerLevel": level,
|
||
"productId": int(product["id"]),
|
||
"productCode": product["code"],
|
||
"productName": product["name"],
|
||
"quantity": quantity,
|
||
"deliveryDate": delivery_date,
|
||
"priority": max(1, min(9, priority)),
|
||
"isRush": is_rush,
|
||
"status": "CONFIRMED",
|
||
}, None
|
||
|
||
|
||
def _normalize_material_row(row: Any, index: int) -> tuple[dict[str, Any] | None, str | None]:
|
||
if not isinstance(row, dict):
|
||
return None, f"第{index}行必须是 JSON 对象"
|
||
code = str(row.get("code") or "").strip()
|
||
name = str(row.get("name") or "").strip()
|
||
material_type = str(row.get("type") or "").strip().upper()
|
||
if not code:
|
||
return None, f"第{index}行缺少 code"
|
||
if not name:
|
||
return None, f"第{index}行缺少 name"
|
||
if material_type not in MATERIAL_TYPES:
|
||
return None, f"第{index}行 type 必须是 FINISHED_PRODUCT/SEMI_FINISHED/RAW_MATERIAL"
|
||
numeric: dict[str, float] = {}
|
||
for field in ("stock", "inTransit", "safetyStock", "procurementLeadTime"):
|
||
value = row.get(field, 0)
|
||
if isinstance(value, bool):
|
||
return None, f"第{index}行 {field} 必须是数字"
|
||
try:
|
||
numeric[field] = float(value)
|
||
except (TypeError, ValueError):
|
||
return None, f"第{index}行 {field} 必须是数字"
|
||
return {
|
||
"code": code,
|
||
"name": name,
|
||
"type": material_type,
|
||
"unit": str(row.get("unit") or "件").strip() or "件",
|
||
"spec": str(row.get("spec") or "").strip(),
|
||
**numeric,
|
||
}, None
|
||
|
||
|
||
def parse_import(kind: str | None, world: World | None = None, *,
|
||
rows: list[Any] | None = None) -> dict[str, Any]:
|
||
"""校验 Pi 已规范化的导入参数，不解析任何原始自然语言文本。"""
|
||
normalized_kind = str(kind or "").strip().lower()
|
||
if normalized_kind not in IMPORT_KINDS:
|
||
return _import_error(None, "请明确导入类型：kind 必须为 orders 或 materials")
|
||
if not isinstance(rows, list) or not rows:
|
||
return _import_error(normalized_kind, "请提供结构化导入行：rows 必须是非空数组")
|
||
normalized: list[dict[str, Any]] = []
|
||
errors: list[str] = []
|
||
for index, row in enumerate(rows, 1):
|
||
if normalized_kind == "orders":
|
||
if world is None:
|
||
return _import_error(normalized_kind, "订单导入需要 world 用于精确校验 productCode")
|
||
item, error = _normalize_order_row(world, row, index)
|
||
else:
|
||
item, error = _normalize_material_row(row, index)
|
||
if error:
|
||
errors.append(error)
|
||
elif item is not None:
|
||
normalized.append(item)
|
||
return {"kind": normalized_kind, "rows": normalized, "errors": errors}
|
||
|
||
|
||
def confirmation_for_import(parsed: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""批量导入确认卡。"""
|
||
kind = parsed["kind"]
|
||
rows = parsed["rows"]
|
||
label = "订单" if kind == "orders" else "物料"
|
||
title = f"批量导入{label} {len(rows)} 条"
|
||
lines = [f"将新建 {len(rows)} 条{label}写入主干（P2）"]
|
||
for r in rows[:8]:
|
||
if kind == "orders":
|
||
lines.append(
|
||
f"· {r['customerName']} · {r.get('productName', '')} × {r['quantity']} · 交期 {r['deliveryDate']}"
|
||
)
|
||
else:
|
||
lines.append(f"· {r['code']} {r['name']} · {r['type']} · 库存 {r.get('stock', 0)}")
|
||
if len(rows) > 8:
|
||
lines.append(f"…另有 {len(rows) - 8} 条")
|
||
if parsed.get("errors"):
|
||
lines.append(f"跳过无法解析 {len(parsed['errors'])} 行")
|
||
lines.append("批准后执行前自动建档，可回滚")
|
||
return title, lines
|
||
|
||
|
||
def apply_import(world: World, next_id, parsed: dict[str, Any]) -> dict[str, Any]:
|
||
"""执行批量导入。"""
|
||
from server.aps_domain.masterdata import apply_master_action
|
||
from server.aps_domain.orders import apply_order_action
|
||
|
||
created: list[str] = []
|
||
kind = parsed["kind"]
|
||
for row in parsed["rows"]:
|
||
if kind == "orders":
|
||
applied = apply_order_action(world, next_id, "order.upsert", row)
|
||
created.append(applied["order"]["orderNo"])
|
||
else:
|
||
applied = apply_master_action(world, next_id, "master.material.upsert", row)
|
||
created.append(row["code"])
|
||
return {"kind": kind, "count": len(created), "created": created}
|