436 lines
18 KiB
Python
436 lines
18 KiB
Python
# ============================================================
|
||
# MOM 主数据收集表导入（moduleId: importers-mom-pack, 可重生 ✅）
|
||
# 识别「01-生产模型 / 02-工艺模型 / 04-物料模型 / 07-设备模型」多页工作簿，
|
||
# 写入 flexMaterials / flexBom / flexEquipment / flexOrders（缺工艺时生成钣金默认路线占位）。
|
||
# 「02-工艺模型」在锐扬表里实际是多层 BOM（物料清单），不是工序步骤表。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from typing import Any
|
||
|
||
World = dict[str, Any]
|
||
|
||
_SHEET_HINTS = ("生产模型", "工艺模型", "物料模型", "设备模型")
|
||
|
||
|
||
def is_mom_workbook(path: str) -> bool:
|
||
"""根据文件名或 sheet 名判断是否为 MOM 主数据收集表。"""
|
||
name = os.path.basename(path)
|
||
if re.search(r"MOM|主数据收集", name, re.I):
|
||
return True
|
||
try:
|
||
from openpyxl import load_workbook
|
||
wb = load_workbook(path, read_only=True, data_only=True)
|
||
titles = " ".join(wb.sheetnames)
|
||
wb.close()
|
||
return sum(1 for h in _SHEET_HINTS if h in titles) >= 2
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _cell(v: Any) -> str:
|
||
if v is None:
|
||
return ""
|
||
return str(v).replace("\u200c", "").replace("\ufeff", "").strip()
|
||
|
||
|
||
def _sheet_matrix(path: str) -> dict[str, list[list[str]]]:
|
||
from openpyxl import load_workbook
|
||
wb = load_workbook(path, read_only=True, data_only=True)
|
||
out: dict[str, list[list[str]]] = {}
|
||
for ws in wb.worksheets:
|
||
rows: list[list[str]] = []
|
||
for row in ws.iter_rows(values_only=True):
|
||
rows.append([_cell(c) for c in (row or ())])
|
||
out[ws.title] = rows
|
||
wb.close()
|
||
return out
|
||
|
||
|
||
def _find_sheet(sheets: dict[str, list], *keywords: str) -> tuple[str, list[list[str]]] | None:
|
||
for title, rows in sheets.items():
|
||
if all(k in title for k in keywords):
|
||
return title, rows
|
||
if any(k in title for k in keywords) and len(keywords) == 1:
|
||
return title, rows
|
||
for title, rows in sheets.items():
|
||
if any(k in title for k in keywords):
|
||
return title, rows
|
||
return None
|
||
|
||
|
||
def _find_header(rows: list[list[str]], required: list[str]) -> tuple[int, dict[str, int]] | None:
|
||
"""找包含必备列名的表头行，返回 (row_index, {规范名: col_index})。"""
|
||
alias = {
|
||
"code": ("物料编码", "ERP品号", "品号", "财务编号", "厂内编号", "固定资产编号"),
|
||
"name": ("物料名称", "产品名称", "设备名称", "物料简称"),
|
||
"level": ("层次", "层级", "BOM层次"),
|
||
"qty": ("单套用量", "用量", "数量", "单位用量"),
|
||
"attr": ("品号属性", "物料来源", "来源"),
|
||
"unit": ("单位",),
|
||
"group": ("物料组名称", "物料组"),
|
||
"drawing": ("图号", "客户图号"),
|
||
"spec": ("设备型号", "型号"),
|
||
"equip_code": ("厂内编号", "财务编号", "固定资产编号"),
|
||
"asset_code": ("固定资产编号",),
|
||
"factory_code": ("出厂编号",),
|
||
"status": ("设备状态", "状态"),
|
||
}
|
||
for i, row in enumerate(rows[:40]):
|
||
joined: dict[str, int] = {}
|
||
for idx, cell in enumerate(row):
|
||
normalized = cell.replace(" ", "").strip()
|
||
if normalized:
|
||
joined.setdefault(normalized, idx)
|
||
if not joined:
|
||
continue
|
||
mapped: dict[str, int] = {}
|
||
for key, names in alias.items():
|
||
for n in names:
|
||
n2 = n.replace(" ", "")
|
||
if n2 in joined:
|
||
mapped[key] = joined[n2]
|
||
break
|
||
if all(r in mapped for r in required):
|
||
return i, mapped
|
||
return None
|
||
|
||
|
||
def _mat_type(attr: str, group: str = "") -> str:
|
||
text = f"{attr} {group}"
|
||
if any(k in text for k in ("外购", "采购", "原材料", "YCL", "P外")):
|
||
return "RAW_MATERIAL"
|
||
if any(k in text for k in ("半成品", "BCP", "焊接", "组件")):
|
||
return "SEMI_FINISHED"
|
||
if any(k in text for k in ("成品", "CP", "自制")):
|
||
return "FINISHED_PRODUCT"
|
||
if "M自制" in text:
|
||
return "SEMI_FINISHED"
|
||
return "RAW_MATERIAL"
|
||
|
||
|
||
def _parent_level(level: str) -> str | None:
|
||
level = level.strip()
|
||
if not level or "." not in level:
|
||
return None
|
||
return level.rsplit(".", 1)[0]
|
||
|
||
|
||
def parse_mom_workbook(path: str) -> dict[str, Any]:
|
||
"""解析 MOM xlsx → flex 结构（不写盘）。"""
|
||
sheets = _sheet_matrix(path)
|
||
materials: dict[str, dict] = {}
|
||
bom: list[dict] = []
|
||
equipment: list[dict] = []
|
||
zones: list[dict] = []
|
||
orders: list[dict] = []
|
||
root_code = ""
|
||
root_name = ""
|
||
customer = ""
|
||
|
||
# ---- 02 工艺模型 = BOM 树 ----
|
||
hit = _find_sheet(sheets, "工艺模型") or _find_sheet(sheets, "物料清单")
|
||
level_to_code: dict[str, str] = {}
|
||
if hit:
|
||
_, rows = hit
|
||
# 从元信息抓产品/客户
|
||
for row in rows[:8]:
|
||
for i, c in enumerate(row):
|
||
if c == "产品名称" and i + 1 < len(row) and row[i + 1]:
|
||
root_name = row[i + 1]
|
||
if c == "客户代码" and i + 1 < len(row) and row[i + 1]:
|
||
customer = row[i + 1]
|
||
if "客户料号" in c and i + 1 < len(row) and row[i + 1]:
|
||
pass
|
||
hdr = _find_header(rows, ["code", "name"])
|
||
if hdr:
|
||
hi, cols = hdr
|
||
for row in rows[hi + 1:]:
|
||
code = row[cols["code"]] if cols["code"] < len(row) else ""
|
||
name = row[cols["name"]] if cols["name"] < len(row) else ""
|
||
if not code or not name:
|
||
continue
|
||
if code in ("ERP品号", "品号"):
|
||
continue
|
||
level = row[cols["level"]] if "level" in cols and cols["level"] < len(row) else ""
|
||
qty_s = row[cols["qty"]] if "qty" in cols and cols["qty"] < len(row) else "1"
|
||
attr = row[cols["attr"]] if "attr" in cols and cols["attr"] < len(row) else ""
|
||
unit = row[cols["unit"]] if "unit" in cols and cols["unit"] < len(row) else "件"
|
||
try:
|
||
qty = float(qty_s) if qty_s else 1.0
|
||
except ValueError:
|
||
qty = 1.0
|
||
mtype = _mat_type(attr)
|
||
materials[code] = {
|
||
"code": code, "name": name, "type": mtype, "unit": unit or "件",
|
||
"stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0,
|
||
"spec": row[cols["drawing"]] if "drawing" in cols and cols["drawing"] < len(row) else "",
|
||
"sourcingType": "BUY" if mtype == "RAW_MATERIAL" else "MAKE",
|
||
}
|
||
if level:
|
||
level_to_code[level] = code
|
||
parent = _parent_level(level)
|
||
if parent and parent in level_to_code:
|
||
bom.append({
|
||
"productCode": level_to_code[parent],
|
||
"materialCode": code,
|
||
"quantity": qty or 1.0,
|
||
"isKey": qty >= 1,
|
||
})
|
||
if level == "1" or (not root_code and level in ("1", "1.0")):
|
||
root_code = code
|
||
root_name = root_name or name
|
||
|
||
# ---- 04 物料模型 ----
|
||
hit = _find_sheet(sheets, "物料模型")
|
||
if hit:
|
||
_, rows = hit
|
||
hdr = _find_header(rows, ["code", "name"])
|
||
if hdr:
|
||
hi, cols = hdr
|
||
for row in rows[hi + 1:]:
|
||
code = row[cols["code"]] if cols["code"] < len(row) else ""
|
||
name = row[cols["name"]] if cols["name"] < len(row) else ""
|
||
if not code or not name or code in ("物料编码", "示例"):
|
||
continue
|
||
group = row[cols["group"]] if "group" in cols and cols["group"] < len(row) else ""
|
||
attr = row[cols["attr"]] if "attr" in cols and cols["attr"] < len(row) else ""
|
||
unit = row[cols["unit"]] if "unit" in cols and cols["unit"] < len(row) else "件"
|
||
mtype = _mat_type(attr, group)
|
||
prev = materials.get(code) or {}
|
||
materials[code] = {
|
||
"code": code,
|
||
"name": name or prev.get("name") or code,
|
||
"type": mtype or prev.get("type") or "RAW_MATERIAL",
|
||
"unit": unit or prev.get("unit") or "件",
|
||
"stock": float(prev.get("stock") or 0),
|
||
"inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0,
|
||
"spec": (row[cols["drawing"]] if "drawing" in cols and cols["drawing"] < len(row) else "")
|
||
or prev.get("spec") or "",
|
||
"sourcingType": "BUY" if mtype == "RAW_MATERIAL" else "MAKE",
|
||
}
|
||
|
||
# ---- 07 设备 ----
|
||
hit = _find_sheet(sheets, "设备模型")
|
||
used_equipment_codes: set[str] = set()
|
||
if hit:
|
||
_, rows = hit
|
||
hdr = _find_header(rows, ["name"])
|
||
if hdr:
|
||
hi, cols = hdr
|
||
for row in rows[hi + 1:]:
|
||
name = row[cols["name"]] if cols["name"] < len(row) else ""
|
||
if not name or name in ("设备名称",):
|
||
continue
|
||
finance_code = row[cols["code"]] if "code" in cols and cols["code"] < len(row) else ""
|
||
internal_code = (
|
||
row[cols["equip_code"]]
|
||
if "equip_code" in cols and cols["equip_code"] < len(row) else ""
|
||
)
|
||
asset_code = (
|
||
row[cols["asset_code"]]
|
||
if "asset_code" in cols and cols["asset_code"] < len(row) else ""
|
||
)
|
||
factory_code = (
|
||
row[cols["factory_code"]]
|
||
if "factory_code" in cols and cols["factory_code"] < len(row) else ""
|
||
)
|
||
code = finance_code or internal_code or asset_code or factory_code
|
||
if not code:
|
||
# 财务编号列常在 name 前
|
||
for c in row:
|
||
if re.match(r"^[A-Z]\d", c):
|
||
code = c
|
||
break
|
||
if not code:
|
||
continue
|
||
if code.endswith(("编码", "编号")) or name in ("单位", "设备名称", "备件名称", "部件名称"):
|
||
continue
|
||
if code in used_equipment_codes:
|
||
for suffix in (internal_code, asset_code, factory_code):
|
||
candidate = f"{code}/{suffix}" if suffix else ""
|
||
if candidate and candidate not in used_equipment_codes:
|
||
code = candidate
|
||
break
|
||
else:
|
||
base = code
|
||
serial = 2
|
||
while f"{base}/{serial}" in used_equipment_codes:
|
||
serial += 1
|
||
code = f"{base}/{serial}"
|
||
used_equipment_codes.add(code)
|
||
status_raw = row[cols["status"]] if "status" in cols and cols["status"] < len(row) else "使用中"
|
||
running = status_raw in ("", "使用中", "RUNNING", "在用", "正常")
|
||
# 按设备名粗分能力
|
||
caps = ["GENERAL"]
|
||
if any(k in name for k in ("激光", "切割", "下料")):
|
||
caps = ["CUT", "GENERAL"]
|
||
elif any(k in name for k in ("折弯", "折板")):
|
||
caps = ["BEND", "GENERAL"]
|
||
elif any(k in name for k in ("焊",)):
|
||
caps = ["WELD", "GENERAL"]
|
||
elif any(k in name for k in ("喷涂", "喷粉")):
|
||
caps = ["PAINT", "GENERAL"]
|
||
equipment.append({
|
||
"code": code, "name": name,
|
||
"capabilities": caps,
|
||
"opStdTime": {c: 1.0 for c in caps},
|
||
"movable": False, "moveTimeMin": 0,
|
||
"zone": "ZONE-A", "adaptableMolds": [],
|
||
"availabilityRate": 0.9,
|
||
"status": "RUNNING" if running else "DOWN",
|
||
"spec": row[cols["spec"]] if "spec" in cols and cols["spec"] < len(row) else "",
|
||
"financeCode": finance_code,
|
||
"internalCode": internal_code,
|
||
"assetCode": asset_code,
|
||
"factoryCode": factory_code,
|
||
})
|
||
|
||
# ---- 01 生产模型 → 车间作区域 ----
|
||
hit = _find_sheet(sheets, "生产模型")
|
||
if hit:
|
||
_, rows = hit
|
||
mode = ""
|
||
for row in rows:
|
||
cells = [c for c in row if c]
|
||
head = row[0] if row else ""
|
||
if "车间管理" in head or (len(row) > 2 and row[2] == "车间名称"):
|
||
mode = "workshop"
|
||
continue
|
||
if "工段管理" in head or "工厂信息" in head:
|
||
mode = ""
|
||
continue
|
||
if mode == "workshop" and len(row) >= 5:
|
||
wname, wcode = row[2], row[4]
|
||
if wname and wcode and wname != "车间名称" and not wname.startswith("示例"):
|
||
zones.append({"code": wcode, "name": wname})
|
||
|
||
if root_code:
|
||
orders.append({
|
||
"orderNo": f"MOM-{root_code[-6:]}",
|
||
"productCode": root_code,
|
||
"productName": root_name or root_code,
|
||
"quantity": 1,
|
||
"dueDate": "",
|
||
"priority": 5,
|
||
"status": "RELEASED",
|
||
"customerName": customer or "MOM客户",
|
||
"customerLevel": "B",
|
||
})
|
||
# 确保根物料存在
|
||
materials.setdefault(root_code, {
|
||
"code": root_code, "name": root_name or root_code,
|
||
"type": "FINISHED_PRODUCT", "unit": "件",
|
||
"stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0,
|
||
"sourcingType": "MAKE",
|
||
})
|
||
|
||
# 自制件默认钣金工艺占位（表内无工序步骤列）
|
||
routings: list[dict] = []
|
||
ops = [
|
||
("CUT", "下料/激光切割", 5.0),
|
||
("BEND", "折弯", 3.0),
|
||
("WELD", "焊接", 8.0),
|
||
("PAINT", "喷涂", 6.0),
|
||
("ASM", "组装", 4.0),
|
||
]
|
||
make_codes = [
|
||
m["code"] for m in materials.values()
|
||
if m.get("type") in ("FINISHED_PRODUCT", "SEMI_FINISHED") and m.get("sourcingType") == "MAKE"
|
||
]
|
||
for pc in make_codes[:80]:
|
||
for seq, (ocode, oname, std) in enumerate(ops, 1):
|
||
routings.append({
|
||
"productCode": pc, "productName": materials[pc]["name"],
|
||
"seq": seq, "operationCode": ocode, "operationName": oname,
|
||
"requireMold": False, "stdTimePerUnit": std, "stdTimeSource": "模板",
|
||
"sourcingType": "MAKE", "isExternal": False,
|
||
})
|
||
|
||
return {
|
||
"filename": os.path.basename(path),
|
||
"flexMaterials": list(materials.values()),
|
||
"flexBom": bom,
|
||
"flexEquipment": equipment,
|
||
"flexZones": zones,
|
||
"flexOrders": orders,
|
||
"flexRoutings": routings,
|
||
"flexOperations": [
|
||
{"code": o, "name": n, "type": "INTERNAL", "sourcingType": "MAKE", "changeoverMin": 10}
|
||
for o, n, _ in ops
|
||
],
|
||
"stats": {
|
||
"materials": len(materials),
|
||
"bom": len(bom),
|
||
"equipment": len(equipment),
|
||
"zones": len(zones),
|
||
"orders": len(orders),
|
||
"routings": len(routings),
|
||
"rootProduct": root_code,
|
||
"note": "02-工艺模型按BOM树解析；工序为钣金默认模板占位，可在主数据中改",
|
||
},
|
||
}
|
||
|
||
|
||
def apply_mom_pack_to_world(world: World, pack: dict[str, Any], *, replace: bool = True) -> dict[str, int]:
|
||
"""把 MOM 解析结果写入世界。"""
|
||
summary: dict[str, int] = {}
|
||
mapping = [
|
||
("flexMaterials", "materials"),
|
||
("flexBom", "bom"),
|
||
("flexEquipment", "equipment"),
|
||
("flexZones", "zones"),
|
||
("flexOrders", "orders"),
|
||
("flexRoutings", "routings"),
|
||
("flexOperations", "operations"),
|
||
]
|
||
for key, label in mapping:
|
||
rows = pack.get(key) or []
|
||
if replace:
|
||
world[key] = []
|
||
bucket = world.setdefault(key, [])
|
||
if key in ("flexMaterials", "flexEquipment", "flexZones", "flexOperations"):
|
||
by_code = {str(x.get("code")): x for x in bucket if x.get("code")}
|
||
for i, row in enumerate(rows, 1):
|
||
code = str(row.get("code") or "")
|
||
payload = dict(row)
|
||
payload.setdefault("id", len(by_code) + i)
|
||
if code and code in by_code:
|
||
by_code[code].update(payload)
|
||
else:
|
||
bucket.append(payload)
|
||
if code:
|
||
by_code[code] = payload
|
||
else:
|
||
for i, row in enumerate(rows, 1):
|
||
payload = dict(row)
|
||
payload.setdefault("id", len(bucket) + i)
|
||
bucket.append(payload)
|
||
summary[label] = len(rows)
|
||
|
||
# 投影经典主数据，便于工艺模型页展示
|
||
try:
|
||
from server.importers.sql_pack import sync_flex_to_classic_master
|
||
sync_flex_to_classic_master(world)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from server.aps_domain.orders import sync_flex_orders_to_sales
|
||
sync_flex_orders_to_sales(world)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from server.aps_domain.sourcing import annotate_world_sourcing
|
||
annotate_world_sourcing(world)
|
||
except Exception:
|
||
pass
|
||
return summary
|
||
|
||
|
||
def import_mom_excel(world: World, path: str, *, replace: bool = True) -> dict[str, Any]:
|
||
pack = parse_mom_workbook(path)
|
||
applied = apply_mom_pack_to_world(world, pack, replace=replace)
|
||
return {"filename": pack.get("filename"), "stats": pack.get("stats"), "applied": applied}
|