aps-agent/server/aps_domain/importers.py

1095 lines
51 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# ============================================================
# Excel/CSV 导入管线(moduleId: domain-importers, 可重生 ✅)
# 对齐 docs/product/demand-data-intake.md · features MD-04
# preview(P1 校验)→ commit(P2 确认卡落库)
# ============================================================
from __future__ import annotations
import csv
import io
import re
from typing import Any
World = dict[str, Any]
_HEADER_ALIASES: dict[str, str] = {
"编码": "code", "code": "code", "设备编码": "code", "模具编码": "code",
"设备编号": "code", "工装模具编号": "code", "模具编号": "code",
"物料编码": "code", "物料代码": "code", "物料编号": "code",
"ERP品号": "code", "erp品号": "code", "财务编号": "code",
"固定资产编号": "code", "厂内编号": "code",
"物品编码": "code", "存货编码": "code", "零件编码": "code",
"产品编码": "productCode", "productcode": "productCode",
"成品编码": "productCode", "料号": "productCode", "产品料号": "productCode",
"父件编码": "productCode", "父项编码": "productCode", "父项产品编码": "productCode", "母件编码": "productCode",
"名称": "name", "name": "name", "设备名称": "name", "物料简称": "name",
"模具名称": "name", "工装模具名称": "name", "物料名称": "name",
"产品名称": "productName", "物料描述": "productName", "productname": "productName",
"类型": "type", "type": "type", "物料类型": "type", "物料分类": "type", "物料组名称": "type",
"单位": "unit", "unit": "unit", "计量单位": "unit",
"库存": "stock", "stock": "stock", "当前库存": "stock", "现有库存": "stock",
"在途": "inTransit", "intransit": "inTransit", "安全库存": "safetyStock",
"前置期": "procurementLeadTime", "采购前置期": "procurementLeadTime",
"预计到货日期": "expectedArrivalDate", "预计到货": "expectedArrivalDate", "expectedarrivaldate": "expectedArrivalDate",
"采购方式": "sourcingType", "供应类型": "sourcingType", "sourcingtype": "sourcingType",
"客户": "customerName", "customername": "customerName", "客户名称": "customerName",
"项目名称": "customerName", "项目": "customerName",
"数量": "quantity", "quantity": "quantity", "单套用量": "quantity",
"订单数量": "quantity", "计划订单数量": "quantity", "单位用量": "quantity", "用量": "quantity",
"交期": "deliveryDate", "duedate": "deliveryDate", "交货期": "deliveryDate",
"交货日期": "deliveryDate", "计划交期": "deliveryDate", "需求日期": "deliveryDate",
"计划结束时间": "deliveryDate", "计划结束": "deliveryDate", "完成日期": "deliveryDate",
"订单号": "orderNo", "orderno": "orderNo", "订单代码": "orderNo",
"生产订单": "orderNo", "工单编号": "orderNo", "工单号": "orderNo",
"优先级": "priority", "priority": "priority", "加急": "isRush", "isrush": "isRush",
"等级": "customerLevel", "客户等级": "customerLevel",
"工序编码": "operationCode", "operationcode": "operationCode",
"工序编号": "operationCode", "工序": "operationCode",
"工序名称": "name", "operationname": "name",
"序号": "seq", "seq": "seq", "顺序": "seq", "排序号": "seq",
"单件工时": "stdTimePerUnit", "stdtimeperunit": "stdTimePerUnit",
"工时": "stdTimePerUnit", "标准工时": "stdTimePerUnit", "标准工时/分钟": "stdTimePerUnit",
"需模具": "requireMold", "requiremold": "requireMold",
"是否瓶颈": "isBottleneck", "isbottleneck": "isBottleneck", "瓶颈": "isBottleneck",
"换型": "changeoverMin", "换型时间": "changeoverMin", "changeovermin": "changeoverMin",
"能力": "capabilities", "capabilities": "capabilities", "可执行工序": "capabilities",
"单件工时映射": "opStdTime", "opstdtime": "opStdTime",
"可移动": "movable", "movable": "movable", "移动耗时": "moveTimeMin",
"movetimemin": "moveTimeMin", "区域": "zone", "zone": "zone",
"适配模具": "adaptableMolds", "adaptablemolds": "adaptableMolds",
"适配设备": "adaptableEquipment", "adaptableequipment": "adaptableEquipment",
"可动率": "availabilityRate", "availabilityrate": "availabilityRate",
"状态": "status", "status": "status", "寿命上限": "lifeTotal", "lifetotal": "lifeTotal",
"已用寿命": "lifeUsed", "lifeused": "lifeUsed",
"子件编码": "materialCode", "子项编码": "materialCode", "子项物料编码": "materialCode", "组件编码": "materialCode",
"materialcode": "materialCode",
"关键料": "isKey", "iskey": "isKey", "是否关键件": "isKey",
"消耗工序": "consumeOp", "consumeop": "consumeOp",
"mes工序编号": "consumeOp", "MES工序编号": "consumeOp",
"规格型号": "spec",
"班次编码": "shiftCode", "shiftcode": "shiftCode", "班次名称": "name",
"开始时间": "startTime", "starttime": "startTime", "结束时间": "endTime", "endtime": "endTime",
"工作日": "workdays", "workdays": "workdays", "休息时段": "breaks", "breaks": "breaks",
"启用": "enabled", "enabled": "enabled",
}
# 各类型「排产关键字段」:源表头别名 → 标准字段(用于动态识别展示)
_KIND_KEY_FIELDS: dict[str, tuple[tuple[str, ...], ...]] = {
"orders": (
("productCode", "料号", "产品编码", "成品编码"),
("quantity", "订单数量", "计划订单数量", "数量"),
("deliveryDate", "计划结束时间", "完成日期", "交期"),
("customerName", "项目名称", "客户", "客户名称"),
("orderNo", "订单代码", "订单号", "生产订单"),
),
"calendar": (
("shiftCode", "班次编码"),
("startTime", "开始时间"),
("endTime", "结束时间"),
),
"materials": (
("code", "物料代码", "物料编码", "物料编号", "编码"),
("name", "物料名称", "名称"),
),
"routing": (
("operationCode", "工序编号", "工序编码", "工序"),
("seq", "排序号", "序号", "顺序"),
("stdTimePerUnit", "标准工时/分钟", "标准工时", "工时", "单件工时"),
("productCode", "产品编码", "料号", "成品编码"),
),
"equipment": (
("code", "设备编号", "设备编码", "编码"),
("name", "设备名称", "名称"),
("capabilities", "可执行工序", "能力"),
),
"molds": (
("code", "工装模具编号", "模具编号", "模具编码", "编码"),
("name", "工装模具名称", "模具名称", "名称"),
("operationCode", "工序编号", "适用工序", "工序"),
),
"bom": (
("materialCode", "物料编号", "物料编码", "子件编码"),
("quantity", "单位用量", "用量", "数量"),
("productCode", "成品编码", "产品编码", "料号"),
("orderNo", "工单编号", "订单代码"),
),
"operations": (
("code", "工序编号", "工序编码", "编码"),
("name", "工序名称", "名称"),
),
}
_KIND_HINTS: dict[str, tuple[str, ...]] = {
"orders": ("订单", "order", "销售订单"),
"calendar": ("班历", "班次", "calendar", "shift"),
"materials": ("物料", "material", "库存"),
"bom": ("bom", "物料清单", "产品bom"),
"routing": ("工艺路线", "routing", "路线"),
"equipment": ("设备", "equipment", "机器"),
"molds": ("模具", "mold", "工装"),
"operations": ("工序", "operation", "工序库"),
"zones": ("区域", "zone", "布局"),
}
IMPORT_KINDS = tuple(_KIND_HINTS.keys())
def _norm_header(h: str) -> str:
raw = str(h or "").strip()
key = re.sub(r"[\s\u3000]+", "", raw.lower())
direct = _HEADER_ALIASES.get(raw) or _HEADER_ALIASES.get(key)
if direct:
return direct
# 现场模板常在列名后追加「(必填)」「[分钟]」或星号;先去掉说明再匹配。
cleaned = re.sub(r"[((【\[].*?[))】\]]", "", key)
cleaned = re.sub(r"[**#::_\-]+", "", cleaned)
cleaned = re.sub(r"(?:必填|字段|列名)$", "", cleaned)
return _HEADER_ALIASES.get(cleaned) or cleaned
def _truthy(v: Any) -> bool:
if isinstance(v, bool):
return v
s = str(v or "").strip().lower()
return s in ("1", "true", "yes", "y", "是", "有")
def _split_multi(v: Any) -> list[str]:
if v is None or v == "":
return []
if isinstance(v, (list, tuple)):
return [str(x).strip() for x in v if str(x).strip()]
return [p.strip() for p in re.split(r"[,,;/|、]", str(v)) if p.strip()]
def _parse_op_std(v: Any) -> dict[str, float]:
if isinstance(v, dict):
return {str(k): float(val) for k, val in v.items()}
out: dict[str, float] = {}
for part in _split_multi(v):
if ":" in part:
k, t = part.split(":", 1)
try:
out[k.strip()] = float(t)
except ValueError:
continue
return out
def detect_kind(name: str, headers: list[str]) -> str | None:
blob = (name or "").lower()
for kind, hints in _KIND_HINTS.items():
if any(h.lower() in blob for h in hints):
return kind
hs = {h.lower() for h in headers}
hdrs = {str(h or "").strip() for h in headers}
mapped = {_norm_header(h) for h in headers if h}
if (
{"customername", "客户", "交期", "deliverydate", "订单代码", "料号", "订单数量"} & (hs | hdrs)
or ({"productcode", "quantity"} <= mapped and ("deliverydate" in mapped or "customername" in mapped))
or ("orderno" in mapped and "productcode" in mapped)
):
return "orders"
if {"starttime", "endtime"} <= mapped or {"开始时间", "结束时间"} <= hdrs:
return "calendar"
if "capabilities" in mapped or "可执行工序" in hdrs or "movable" in mapped or "设备编号" in hdrs:
return "equipment"
if "lifetotal" in mapped or "寿命上限" in hdrs or "工装模具编号" in hdrs:
return "molds"
if "isbottleneck" in mapped or "是否瓶颈" in hdrs:
return "operations"
if "operationcode" in mapped and ("seq" in mapped or "stdtimeperunit" in mapped):
return "routing"
if ("materialcode" in mapped or "物料编号" in hdrs) and ("quantity" in mapped or "单位用量" in hdrs):
return "bom"
if "stock" in mapped or "库存" in hdrs or "物料代码" in hdrs or "物料编码" in hdrs:
return "materials"
if "zone" in mapped and len(headers) <= 4:
return "zones"
return None
def detect_field_map(kind: str, headers_raw: list[str]) -> list[dict[str, str]]:
"""根据原始表头动态识别关键字段映射:源列 → 标准字段。"""
specs = _KIND_KEY_FIELDS.get(kind) or ()
found: list[dict[str, str]] = []
seen_targets: set[str] = set()
raw_list = [str(h or "").strip() for h in headers_raw if h]
raw_set = set(raw_list)
for group in specs:
target = group[0]
if target in seen_targets:
continue
hit = None
# 1) 优先精确命中中文/英文别名(跳过 target 自身,避免 序号 抢在 排序号 前)
for alias in group[1:]:
if alias in raw_set:
hit = alias
break
# 2) 再按规范化列名匹配
if not hit:
for h in raw_list:
if _norm_header(h) == target:
hit = h
break
if hit:
seen_targets.add(target)
found.append({"source": hit, "target": target})
return found
def _sheet_as_code(sheet: str | None) -> str:
s = str(sheet or "").strip()
if not s or s.lower().startswith("sheet") or s in ("订单导出", "物料", "Equipment"):
return ""
return s
def _rows_from_csv(raw: bytes) -> tuple[list[str], list[dict[str, Any]]]:
text = raw.decode("utf-8-sig", errors="replace")
reader = csv.DictReader(io.StringIO(text))
headers = [_norm_header(h) for h in (reader.fieldnames or [])]
rows = []
for r in reader:
item = {_norm_header(k): (v.strip() if isinstance(v, str) else v) for k, v in r.items() if k}
if any(str(v or "").strip() for v in item.values()):
rows.append(item)
return headers, rows
_XLSX_SOURCE_KEY = "__source__"
_XLSX_DIAGNOSTIC_KEY = "__diagnostic__"
_EMBEDDED_EQUIPMENT_SECTIONS = {"备品备件管理", "设备部件管理", "设备类型"}
_RAW_SUMMARY_PRIORITY = (
"财务编号", "固定资产编号", "厂内编号", "设备名称",
"设备型号", "出厂编号", "可执行工序", "工序名称",
)
def _is_blank(value: Any) -> bool:
return value is None or (isinstance(value, str) and not value.strip())
def _jsonish_value(value: Any) -> Any:
if value is None or isinstance(value, (bool, int, float)):
return value
if hasattr(value, "isoformat"):
try:
return value.isoformat()
except (TypeError, ValueError):
pass
text = str(value).strip()
return text if len(text) <= 120 else f"{text[:117]}..."
def _raw_row_summary(headers_raw: list[str], line: tuple[Any, ...], limit: int = 8) -> dict[str, Any]:
"""返回有限、可 JSON 序列化的原始行摘要;关键编码字段即使为空也保留。"""
by_header: dict[str, Any] = {}
for idx, header in enumerate(headers_raw):
if not header or idx >= len(line):
continue
value = line[idx]
if header not in by_header or (not _is_blank(value) and _is_blank(by_header[header])):
by_header[header] = value
result: dict[str, Any] = {}
for header in _RAW_SUMMARY_PRIORITY:
if header in by_header and len(result) < limit:
result[header] = _jsonish_value(by_header[header])
for header, value in by_header.items():
if len(result) >= limit:
break
if header in result or _is_blank(value):
continue
result[header] = _jsonish_value(value)
return result
def _set_mapped_value(item: dict[str, Any], key: str, value: Any) -> bool:
"""写入规范字段;返回 True 表示空的重复别名被安全忽略。"""
if key not in item:
item[key] = value
return False
if _is_blank(value):
return not _is_blank(item[key])
item[key] = value
return False
def _diagnostic_counts(diagnostics: list[dict[str, Any]]) -> dict[str, int]:
counts = {"ignored": 0, "warning": 0, "blocking": 0}
for item in diagnostics:
severity = str(item.get("severity") or "")
if severity in counts:
counts[severity] += 1
return counts
def _row_source(sheet: str, excel_row: int, headers_raw: list[str], line: tuple[Any, ...],
section: str = "") -> dict[str, Any]:
return {
"sheet": sheet,
"excelRow": excel_row,
"rawSummary": _raw_row_summary(headers_raw, line),
"section": section,
}
def _rows_from_xlsx(raw: bytes) -> list[dict[str, Any]]:
from openpyxl import load_workbook
wb = load_workbook(io.BytesIO(raw), read_only=True, data_only=True)
sheets = []
known_targets = set(_HEADER_ALIASES.values())
for ws in wb.worksheets:
data = list(ws.iter_rows(values_only=True))
if not data:
continue
# 现场模板常在首行放标题/合并单元格(如 MOM 收集表的「物料清单」、
# 分块「物料组 + 物料主数据」):扫描规范化后命中已知字段映射最多
# 的行作为表头(序号/编码/名称/单位/数量…),跳过标题行;
# 平铺表第 1 行即表头时行为不变;无任何命中退回第 1 行。
best_idx, best_score = 0, -1
for idx, row in enumerate(data):
if not row:
continue
score = sum(
1 for cell in row
if cell is not None and str(cell).strip() and _norm_header(cell) in known_targets
)
if score > best_score:
best_idx, best_score = idx, score
headers_raw = [str(cell).strip() if cell is not None else "" for cell in data[best_idx]]
headers = [_norm_header(header) for header in headers_raw]
flow_idx = next(
(idx for idx, header in enumerate(headers_raw)
if header and ("制造流程" in header or "工艺代码" in header or "工艺流程" in header)),
None,
)
active_section = ""
for prior in data[:best_idx + 1]:
if prior and not _is_blank(prior[0]):
active_section = str(prior[0]).strip()
rows: list[dict[str, Any]] = []
for excel_row, line in enumerate(data[best_idx + 1:], start=best_idx + 2):
if not line or all(_is_blank(cell) for cell in line):
continue
if not _is_blank(line[0]):
active_section = str(line[0]).strip()
source = _row_source(ws.title, excel_row, headers_raw, line, active_section)
item: dict[str, Any] = {}
normalizations: list[dict[str, Any]] = []
for idx, header in enumerate(headers):
if not header or idx >= len(line):
continue
value = line[idx]
if isinstance(value, str):
value = value.strip()
ignored_empty_alias = _set_mapped_value(item, header, value)
if ignored_empty_alias and header == "code":
normalizations.append({
"code": "DUPLICATE_ALIAS_EMPTY_IGNORED",
"message": f"空的重复编码列「{headers_raw[idx]}」未覆盖已有编码",
})
if normalizations:
source["normalizations"] = normalizations
sequence_values = []
if flow_idx is not None:
sequence_values = [
str(line[idx]).strip()
for idx in range(flow_idx, len(line))
if idx < len(line) and not _is_blank(line[idx])
]
is_secondary_header = (
flow_idx is not None
and excel_row == best_idx + 2
and len(sequence_values) >= 2
and all(re.fullmatch(r"\d{1,2}", value) for value in sequence_values)
and _is_blank(item.get("code"))
and _is_blank(item.get("name"))
)
if is_secondary_header:
source["rawSummary"] = {
**source.get("rawSummary", {}),
"制造流程序号": ",".join(sequence_values),
}
rows.append({
_XLSX_SOURCE_KEY: source,
_XLSX_DIAGNOSTIC_KEY: {
"kind": "routing",
"code": "SECONDARY_SEQUENCE_HEADER",
"severity": "ignored",
"message": "制造流程二级序号表头,已忽略",
},
})
continue
is_equipment_sheet = "设备" in ws.title or "equipment" in ws.title.lower()
if is_equipment_sheet and active_section in _EMBEDDED_EQUIPMENT_SECTIONS:
rows.append({
_XLSX_SOURCE_KEY: source,
_XLSX_DIAGNOSTIC_KEY: {
"kind": "equipment",
"code": "EMBEDDED_SECTION_ROW",
"severity": "ignored",
"message": f"{active_section}为嵌入式非设备实例数据,已忽略",
},
})
continue
if any(not _is_blank(value) for value in item.values()):
item[_XLSX_SOURCE_KEY] = source
rows.append(item)
# 制造流程列(锐扬 MOM 02-工艺模型):表头行后次行为 01..12 序号,
# 数据行该列起每格 = 工艺代码(如 ZP01组装 / LM01POP拉铆)→ 合成
# routing 视图,供 preview_file 按 routing 批校验/入库。
if flow_idx is not None:
flow_rows: list[dict[str, Any]] = []
seq_map: dict[int, int] = {}
sub = data[best_idx + 1] if best_idx + 1 < len(data) else None
if sub is not None:
for idx in range(flow_idx, len(sub)):
value = sub[idx]
if value is not None and str(value).strip().isdigit():
seq_map[idx] = int(str(value).strip())
for excel_row, line in enumerate(data[best_idx + 2:], start=best_idx + 3):
if not line or all(_is_blank(cell) for cell in line):
continue
item: dict[str, Any] = {}
for idx, header in enumerate(headers):
if not header or idx >= len(line):
continue
value = line[idx]
if isinstance(value, str):
value = value.strip()
_set_mapped_value(item, header, value)
product_code = str(item.get("code") or item.get("productCode") or "").strip()
if not product_code:
continue
source = _row_source(ws.title, excel_row, headers_raw, line, active_section)
for idx in range(flow_idx, len(headers)):
value = line[idx] if idx < len(line) else None
if _is_blank(value):
continue
raw_op = str(value).strip()
match = re.match(r"^([A-Za-z]{1,12}\d{0,6}[A-Za-z]{0,8})\s*(.*)$", raw_op)
operation_code = match.group(1) if match else raw_op
operation_name = (
match.group(2) if match and match.group(2) else raw_op
).strip() or operation_code
sequence = seq_map.get(idx) or (idx - flow_idx + 1)
flow_rows.append({
"productCode": product_code,
"seq": sequence,
"operationCode": operation_code,
"operationName": operation_name,
_XLSX_SOURCE_KEY: source,
})
if flow_rows:
sheets.append({
"sheet": f"{ws.title}#routing",
"physicalSheet": ws.title,
"kind": "routing",
"headers": ["productCode", "seq", "operationCode", "operationName"],
"headersRaw": ["产品编码", "序号", "工序编码", "工序名称"],
"rows": flow_rows,
})
sheets.append({
"sheet": ws.title,
"physicalSheet": ws.title,
"headers": headers,
"headersRaw": headers_raw,
"rows": rows,
})
return sheets
def _due_str(v: Any) -> str:
if v is None:
return ""
if hasattr(v, "strftime"):
return v.strftime("%Y-%m-%d")
return str(v).strip()[:10]
def validate_batch(
kind: str,
rows: list[dict[str, Any]],
world: World,
*,
soft: bool = False,
sheet: str | None = None,
product_by_order: dict[str, str] | None = None,
) -> dict[str, Any]:
"""校验一批导入行。
soft=True:工程目录分析模式——按现场表头容错,不要求成品已入库;
缺可执行工序/适用工序时给默认值并记 warning,便于动态识别关键字段。
"""
ok: list[dict[str, Any]] = []
errors: list[str] = []
warnings: list[str] = []
diagnostics: list[dict[str, Any]] = []
sheet_code = _sheet_as_code(sheet)
pmap = product_by_order or {}
def _source_for(raw: dict[str, Any], parser_index: int) -> dict[str, Any]:
source = raw.get(_XLSX_SOURCE_KEY)
if isinstance(source, dict):
return source
summary: dict[str, Any] = {}
for key, value in raw.items():
if str(key).startswith("__") or _is_blank(value):
continue
summary[str(key)] = _jsonish_value(value)
if len(summary) >= 8:
break
return {"sheet": sheet, "excelRow": None, "rawSummary": summary, "section": ""}
def _make_diagnostic(
raw: dict[str, Any], parser_index: int, *, code: str,
severity: str, message: str, diagnostic_kind: str | None = None,
classification: str | None = None,
) -> dict[str, Any]:
source = _source_for(raw, parser_index)
excel_row = source.get("excelRow")
inferred_classification = classification
if not inferred_classification:
if code == "DUPLICATE_ALIAS_EMPTY_IGNORED":
inferred_classification = "parser_noise"
elif code in {"SECONDARY_SEQUENCE_HEADER", "EMBEDDED_SECTION_ROW"}:
inferred_classification = "ignorable_section"
else:
inferred_classification = severity
return {
"sheet": source.get("sheet") or sheet or "",
"excelRow": int(excel_row) if isinstance(excel_row, (int, float)) else None,
"parserIndex": parser_index,
"kind": diagnostic_kind or kind,
"code": code,
"severity": severity,
"classification": inferred_classification,
"message": message,
"rawSummary": source.get("rawSummary") or {},
}
def _add_warning(raw: dict[str, Any], parser_index: int, code: str, message: str) -> None:
warnings.append(f"第{parser_index}行:{message}")
diagnostics.append(_make_diagnostic(
raw, parser_index, code=code, severity="warning", message=message,
))
def _blocking_code(raw: dict[str, Any], message: str) -> tuple[str, str]:
if kind == "equipment":
if _is_blank(raw.get("code")):
return (
"EQUIPMENT_CODE_MISSING",
"设备编码缺失:财务编号、固定资产编号和厂内编号均为空",
)
return "EQUIPMENT_CAPABILITIES_MISSING", "设备缺少可执行工序"
codes = {
"orders": "ORDER_REQUIRED_FIELDS_MISSING",
"materials": "MATERIAL_REQUIRED_FIELDS_MISSING",
"calendar": "CALENDAR_REQUIRED_FIELDS_MISSING",
"molds": "MOLD_REQUIRED_FIELDS_MISSING",
"operations": "OPERATION_CODE_MISSING",
"zones": "ZONE_CODE_MISSING",
"routing": "ROUTING_REQUIRED_FIELDS_MISSING",
"bom": "BOM_REQUIRED_FIELDS_MISSING",
}
return codes.get(kind, "ROW_VALIDATION_ERROR"), message
def _seq_val(v: Any) -> int:
if v is None or v == "":
return 0
try:
return int(float(v))
except (TypeError, ValueError):
s = str(v).strip()
digits = re.sub(r"\D", "", s)
return int(digits) if digits else 0
for i, raw in enumerate(rows, 1):
source = _source_for(raw, i)
parser_diagnostic = raw.get(_XLSX_DIAGNOSTIC_KEY)
if isinstance(parser_diagnostic, dict):
diagnostics.append(_make_diagnostic(
raw,
i,
code=str(parser_diagnostic.get("code") or "PARSER_IGNORED_ROW"),
severity=str(parser_diagnostic.get("severity") or "ignored"),
message=str(parser_diagnostic.get("message") or "结构性行已忽略"),
diagnostic_kind=str(parser_diagnostic.get("kind") or kind),
classification=str(parser_diagnostic.get("classification") or "") or None,
))
continue
for normalization in source.get("normalizations") or []:
if not isinstance(normalization, dict):
continue
diagnostics.append(_make_diagnostic(
raw, i,
code=str(normalization.get("code") or "DUPLICATE_ALIAS_NORMALIZED"),
severity="warning",
message=str(normalization.get("message") or "重复别名已规范化"),
))
try:
if kind == "orders":
from server.aps_domain.orders import find_product_by_hint
cust = str(raw.get("customerName") or "").strip()
hint = str(raw.get("productCode") or raw.get("productName") or "").strip()
# 料号形如 200042588[282000009 乘致电机] → 取括号内编码优先,否则整段
m_bracket = re.search(r"\[([^\]]+)\]", hint)
if m_bracket:
inner = m_bracket.group(1).strip()
code_part = re.split(r"\s+", inner, maxsplit=1)[0]
if code_part:
hint = code_part
if not str(raw.get("productName") or "").strip() and " " in inner:
raw = {**raw, "productName": inner.split(" ", 1)[1]}
try:
qty = int(float(raw.get("quantity") or 0))
except (TypeError, ValueError):
qty = 0
due = _due_str(raw.get("deliveryDate"))
if soft:
if not hint or qty <= 0 or not due:
raise ValueError("需产品(料号)/数量/交期(计划结束)")
if not cust:
cust = "现场项目"
_add_warning(
raw, i, "ORDER_CUSTOMER_DEFAULTED",
"无客户列,暂用「现场项目」(可用项目名称)",
)
elif not cust or not hint or qty <= 0 or not due:
raise ValueError("需客户/产品/数量/交期")
prod = find_product_by_hint(world, hint)
if prod is None:
prod = next((m for m in world.get("flexMaterials", [])
if m.get("type") == "FINISHED_PRODUCT"
and (m["code"].lower() == hint.lower() or hint in m.get("name", ""))), None)
if prod is None and not soft:
raise ValueError(f"找不到成品「{hint}」")
row_out = {
"customerName": cust,
"customerLevel": str(raw.get("customerLevel") or "A").upper(),
"productId": (prod or {}).get("id"),
"productCode": (prod or {}).get("code") or hint,
"productName": (prod or {}).get("name") or str(raw.get("productName") or hint),
"quantity": qty, "deliveryDate": due,
"priority": int(float(raw.get("priority") or 5)),
"isRush": _truthy(raw.get("isRush")), "status": "CONFIRMED",
}
ono = str(raw.get("orderNo") or "").strip()
if ono:
row_out["orderNo"] = ono
ok.append(row_out)
elif kind == "materials":
code = str(raw.get("code") or raw.get("materialCode") or raw.get("productCode") or "").strip()
name = str(raw.get("name") or raw.get("productName") or "").strip()
if not code or not name:
raise ValueError("需编码与名称")
type_map = {"成品": "FINISHED_PRODUCT", "半成品": "SEMI_FINISHED",
"原料": "RAW_MATERIAL", "原材料": "RAW_MATERIAL"}
raw_type = str(raw.get("type") or "").strip()
mtype = type_map.get(raw_type, raw_type.upper() if raw_type else "RAW_MATERIAL")
if soft and mtype not in ("FINISHED_PRODUCT", "SEMI_FINISHED", "RAW_MATERIAL"):
mtype = "RAW_MATERIAL"
ok.append({
"code": code, "name": name, "type": mtype,
"unit": str(raw.get("unit") or "件").strip('"') or "件",
"stock": float(raw.get("stock") or 0),
"inTransit": float(raw.get("inTransit") or 0),
"safetyStock": float(raw.get("safetyStock") or 0),
"procurementLeadTime": int(float(raw.get("procurementLeadTime") or 0)),
"spec": str(raw.get("spec") or ""),
"expectedArrivalDate": str(raw.get("expectedArrivalDate") or ""),
"sourcingType": str(raw.get("sourcingType") or ""),
})
elif kind == "calendar":
shift_code = str(raw.get("shiftCode") or raw.get("code") or "").strip()
start_time = str(raw.get("startTime") or "").strip()
end_time = str(raw.get("endTime") or "").strip()
if not shift_code or not re.fullmatch(r"\d{2}:\d{2}", start_time) or not re.fullmatch(r"\d{2}:\d{2}", end_time):
raise ValueError("需班次编码、开始时间与结束时间(HH:MM)")
workdays: list[int] = []
for value in _split_multi(raw.get("workdays")) or ["1", "2", "3", "4", "5", "6", "7"]:
try:
day = int(float(value))
except (TypeError, ValueError):
continue
if 1 <= day <= 7 and day not in workdays:
workdays.append(day)
breaks: list[dict[str, str]] = []
for value in _split_multi(raw.get("breaks")):
pair = re.split(r"\s*[-~~至]\s*", value, maxsplit=1)
if len(pair) == 2 and all(re.fullmatch(r"\d{2}:\d{2}", part) for part in pair):
breaks.append({"start": pair[0], "end": pair[1]})
enabled_raw = raw.get("enabled")
ok.append({
"shiftCode": shift_code,
"name": str(raw.get("name") or shift_code),
"startTime": start_time,
"endTime": end_time,
"breaks": breaks,
"workdays": workdays or [1, 2, 3, 4, 5, 6, 7],
"enabled": True if enabled_raw in (None, "") else _truthy(enabled_raw),
"demoOnly": _truthy(raw.get("demoOnly")),
})
elif kind == "equipment":
code = str(raw.get("code") or "").strip()
name = str(raw.get("name") or code).strip()
caps = _split_multi(raw.get("capabilities"))
op_std = _parse_op_std(raw.get("opStdTime"))
if not op_std and caps:
op_std = {c: 1.0 for c in caps}
if not caps and op_std:
caps = list(op_std.keys())
if soft and code and not caps:
caps = ["GENERAL"]
op_std = {"GENERAL": 1.0}
_add_warning(
raw, i, "EQUIPMENT_CAPABILITIES_DEFAULTED",
"设备无「可执行工序」,暂记 GENERAL(可按工艺推断)",
)
if not code or not caps:
raise ValueError("需设备编码与可执行工序")
ok.append({
"code": code, "name": name, "capabilities": caps, "opStdTime": op_std,
"movable": _truthy(raw.get("movable")),
"moveTimeMin": float(raw.get("moveTimeMin") or 0),
"zone": str(raw.get("zone") or "ZONE-A"),
"adaptableMolds": _split_multi(raw.get("adaptableMolds")),
"availabilityRate": float(raw.get("availabilityRate") or 0.95),
"status": str(raw.get("status") or "RUNNING").upper(),
})
elif kind == "molds":
code = str(raw.get("code") or "").strip()
op = str(raw.get("operationCode") or raw.get("type") or "").strip()
if soft and code and not op:
op = "GENERAL"
_add_warning(
raw, i, "MOLD_OPERATION_DEFAULTED",
"模具无适用工序,暂记 GENERAL",
)
if not code or not op:
raise ValueError("需模具编码与适用工序")
ok.append({
"code": code, "name": str(raw.get("name") or code),
"operationCode": op,
"adaptableEquipment": _split_multi(raw.get("adaptableEquipment")),
"lifeTotal": int(float(raw.get("lifeTotal") or 50000)),
"lifeUsed": int(float(raw.get("lifeUsed") or 0)),
"zone": str(raw.get("zone") or "ZONE-A"),
"changeoverMin": float(raw.get("changeoverMin") or 20),
"status": str(raw.get("status") or "AVAILABLE").upper(),
})
elif kind == "operations":
code = str(raw.get("code") or raw.get("operationCode") or "").strip()
if not code:
raise ValueError("需工序编码")
ok.append({
"code": code, "name": str(raw.get("name") or code),
"isBottleneck": _truthy(raw.get("isBottleneck")),
"changeoverMin": float(raw.get("changeoverMin") or 0),
})
elif kind == "zones":
code = str(raw.get("code") or "").strip()
if not code:
raise ValueError("需区域编码")
ok.append({"code": code, "name": str(raw.get("name") or code)})
elif kind == "routing":
pc = str(raw.get("productCode") or "").strip()
if not pc and soft:
# sheet 名常为订单号 → 映射到料号;否则直接用 sheet
pc = pmap.get(sheet_code) or sheet_code
seq = _seq_val(raw.get("seq"))
op = str(raw.get("operationCode") or "").strip()
if not op or seq <= 0:
raise ValueError("需产品编码/序号/工序")
if not pc:
raise ValueError("需产品编码/序号/工序")
std = raw.get("stdTimePerUnit")
try:
std_f = float(std) if std not in (None, "") else None
except (TypeError, ValueError):
std_f = None
ok.append({
"productCode": pc,
"productName": str(raw.get("productName") or raw.get("name") or pc),
"seq": seq, "operationCode": op,
"operationName": str(raw.get("name") or raw.get("operationName") or op),
"requireMold": _truthy(raw.get("requireMold")),
"stdTimePerUnit": std_f,
})
elif kind == "bom":
pc = str(raw.get("productCode") or "").strip()
if not pc and soft:
ono = str(raw.get("orderNo") or sheet_code).strip()
pc = pmap.get(ono) or ono
mc = str(raw.get("materialCode") or raw.get("code") or "").strip()
qty = float(raw.get("quantity") or 0)
if not pc or not mc or qty <= 0:
raise ValueError("需成品/子件/用量")
ok.append({
"productCode": pc, "materialCode": mc, "quantity": qty,
"consumeOp": str(raw.get("consumeOp") or ""),
"isKey": _truthy(raw.get("isKey")),
})
else:
raise ValueError(f"不支持的导入类型:{kind}")
except Exception as exc: # noqa: BLE001
message = str(exc)
errors.append(f"第{i}行:{message}")
code, diagnostic_message = _blocking_code(raw, message)
diagnostics.append(_make_diagnostic(
raw, i, code=code, severity="blocking",
message=diagnostic_message,
))
# soft 模式下同类 warning 去重截断
if warnings:
uniq: list[str] = []
seen_w: set[str] = set()
for w in warnings:
key = re.sub(r"第\d+行:", "第N行:", w)
if key in seen_w:
continue
seen_w.add(key)
uniq.append(w)
if len(uniq) >= 5:
break
warnings = uniq
counts = _diagnostic_counts(diagnostics)
return {
"kind": kind, "okRows": ok, "errors": errors, "warnings": warnings,
"okCount": len(ok), "errorCount": counts["blocking"],
"diagnostics": diagnostics, "diagnosticCounts": counts,
}
def preview_file(
filename: str,
raw: bytes,
world: World,
*,
soft: bool = False,
product_by_order: dict[str, str] | None = None,
kind_hint: str | None = None,
) -> dict[str, Any]:
name = filename or "upload"
lower = name.lower()
batches = []
if lower.endswith((".csv", ".txt")):
headers, rows = _rows_from_csv(raw)
kind = kind_hint or detect_kind(name, headers) or "materials"
fmap = detect_field_map(kind, headers)
batch = validate_batch(
kind, rows, world, soft=soft, sheet=name, product_by_order=product_by_order)
batches.append({
"sheet": name, "physicalSheet": name,
"kind": kind, "fieldMap": fmap, "headersRaw": headers, **batch,
})
elif lower.endswith((".xlsx", ".xlsm")):
from server.importers.planning_workbook import preview_planning_workbook
structured = preview_planning_workbook(name, raw)
if structured is not None:
return structured
for sheet_data in _rows_from_xlsx(raw):
kind = (
kind_hint
or sheet_data.get("kind")
or detect_kind(f"{name}:{sheet_data['sheet']}", sheet_data["headersRaw"])
or detect_kind(name, sheet_data["headers"])
or detect_kind(name, sheet_data["headersRaw"])
or "materials"
)
field_map = detect_field_map(kind, sheet_data["headersRaw"])
physical_sheet = sheet_data.get("physicalSheet") or sheet_data["sheet"]
batch = validate_batch(
kind, sheet_data["rows"], world, soft=soft,
sheet=physical_sheet, product_by_order=product_by_order,
)
batches.append({
"sheet": sheet_data["sheet"],
"physicalSheet": physical_sheet,
"kind": kind,
"fieldMap": field_map,
"headersRaw": sheet_data["headersRaw"],
**batch,
})
else:
raise ValueError("仅支持 .xlsx / .csv")
diagnostics = [
diagnostic
for batch in batches
for diagnostic in (batch.get("diagnostics") or [])
]
diagnostic_counts = _diagnostic_counts(diagnostics)
total_ok = sum(batch["okCount"] for batch in batches)
total_err = sum(batch["errorCount"] for batch in batches)
return {
"filename": name,
"batches": batches,
"totalOk": total_ok,
"totalErrors": total_err,
"canCommit": total_ok > 0,
"diagnostics": diagnostics,
"diagnosticCounts": diagnostic_counts,
"diagnosticsVersion": 1,
}
def confirmation_for_import_commit(preview: dict[str, Any]) -> tuple[str, list[str]]:
title = f"导入入库 · {preview.get('filename', '')}"
lines = [f"有效 {preview.get('totalOk', 0)} 行 · 错误 {preview.get('totalErrors', 0)} 行(错误行不入库)"]
for b in preview.get("batches", []):
lines.append(f"· [{b.get('sheet')}] {b['kind']} 通过 {b['okCount']} / 失败 {b['errorCount']}")
lines.append("批准后写入主干(P2);执行前自动建档可回滚")
return title, lines
def apply_import_commit(world: World, next_id, batches: list[dict[str, Any]]) -> dict[str, Any]:
from server.importers.planning_workbook import apply_planning_batches
profile_batches = [b for b in batches if b.get("sourceProfile")]
if profile_batches:
if len(profile_batches) != len(batches):
raise ValueError("完整工作簿与其他导入批次不能混合确认,请分别导入")
return apply_planning_batches(world, profile_batches)
from server.aps_domain.masterdata import apply_master_action
from server.aps_domain.orders import apply_order_action
from server.state.seed import ensure_flex_seed
ensure_flex_seed(world)
summary: dict[str, int] = {}
def _next_table_id(table: str) -> int:
items = world.get(table, [])
return max((x.get("id", 0) for x in items if isinstance(x.get("id"), int)), default=0) + 1
for batch in batches:
kind = batch["kind"]
for row in batch.get("okRows") or []:
if kind == "orders":
items = world.setdefault("flexOrders", [])
order_no = row.get("orderNo")
existing = next((item for item in items if order_no and item.get("orderNo") == order_no), None)
if row.get("productId") and existing is None and not order_no:
apply_order_action(world, next_id, "order.upsert", row)
summary["salesOrders"] = summary.get("salesOrders", 0) + 1
else:
values = {
"orderNo": order_no,
"productCode": row.get("productCode"), "quantity": row["quantity"],
"dueDate": row["deliveryDate"], "priority": row.get("priority", 5),
"customerName": row.get("customerName") or "现场客户",
"customerLevel": row.get("customerLevel") or "B",
"isRush": bool(row.get("isRush")),
"status": "RELEASED",
}
if existing:
existing.update(values)
else:
mid = _next_table_id("flexOrders")
items.append({"id": mid, **values, "orderNo": order_no or f"FO-{mid:04d}"})
summary["flexOrders"] = summary.get("flexOrders", 0) + 1
elif kind == "materials":
existing_material = next(
(item for item in world.setdefault("materials", []) if item.get("code") == row.get("code")),
None,
)
if existing_material:
master_row = {**row, "id": existing_material["id"]}
else:
# 订单先入库时 sync_flex_orders_to_sales 可能已直接建成品,
# 外部发号器仍停在 0,直接 create 会让原材料与成品 id 冲突。
master_row = {**row, "newId": _next_table_id("materials")}
apply_master_action(world, next_id, "master.material.upsert", master_row)
fm = world.setdefault("flexMaterials", [])
ex = next((m for m in fm if m["code"] == row["code"]), None)
if ex:
ex.update({k: row[k] for k in (
"name", "type", "unit", "stock", "inTransit", "safetyStock",
"procurementLeadTime", "expectedArrivalDate", "sourcingType",
) if k in row})
else:
fm.append({"id": _next_table_id("flexMaterials"),
**{k: row.get(k) for k in
("code", "name", "type", "unit", "stock", "inTransit",
"safetyStock", "procurementLeadTime", "expectedArrivalDate", "sourcingType")}})
if existing_material is None:
summary["materials"] = summary.get("materials", 0) + 1
elif kind == "calendar":
items = world.setdefault("flexCalendar", [])
ex = next((item for item in items if item.get("shiftCode") == row.get("shiftCode")), None)
if ex:
ex.update(row)
else:
items.append(dict(row))
summary["flexCalendar"] = summary.get("flexCalendar", 0) + 1
elif kind == "equipment":
items = world.setdefault("flexEquipment", [])
ex = next((e for e in items if e["code"] == row["code"]), None)
if ex:
ex.update(row)
else:
items.append({"id": _next_table_id("flexEquipment"), **row})
summary["flexEquipment"] = summary.get("flexEquipment", 0) + 1
elif kind == "molds":
items = world.setdefault("flexMolds", [])
ex = next((e for e in items if e["code"] == row["code"]), None)
if ex:
ex.update(row)
else:
items.append({"id": _next_table_id("flexMolds"), **row})
summary["flexMolds"] = summary.get("flexMolds", 0) + 1
elif kind == "operations":
items = world.setdefault("flexOperations", [])
ex = next((e for e in items if e["code"] == row["code"]), None)
if ex:
ex.update(row)
else:
items.append(row)
summary["flexOperations"] = summary.get("flexOperations", 0) + 1
elif kind == "zones":
items = world.setdefault("flexZones", [])
ex = next((e for e in items if e["code"] == row["code"]), None)
if ex:
ex.update(row)
else:
items.append(row)
summary["flexZones"] = summary.get("flexZones", 0) + 1
elif kind == "routing":
items = world.setdefault("flexRoutings", [])
items[:] = [s for s in items
if not (s.get("productCode") == row["productCode"] and s.get("seq") == row["seq"])]
items.append(dict(row))
summary["flexRoutings"] = summary.get("flexRoutings", 0) + 1
elif kind == "bom":
items = world.setdefault("flexBom", [])
items[:] = [s for s in items
if not (s.get("productCode") == row["productCode"]
and s.get("materialCode") == row["materialCode"])]
items.append(dict(row))
summary["flexBom"] = summary.get("flexBom", 0) + 1
# 通用表格导入首先写入 flex* 结构;MRP、订单详情和固定轨排产读取经典主数据。
# 批次全部落地后统一做幂等投影,避免出现“分析卡显示 BOM/工艺,但 MRP 只能看到根成品”的双轨漂移。
if any(summary.get(key, 0) for key in (
"materials", "flexEquipment", "flexOperations", "flexRoutings", "flexBom",
)):
from server.importers.sql_pack import sync_flex_to_classic_master
sync_flex_to_classic_master(world)
if summary.get("flexOrders", 0):
from server.aps_domain.orders import sync_flex_orders_to_sales
sync_flex_orders_to_sales(world)
return {"summary": summary, "total": sum(summary.values())}