# ============================================================ # 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 # 整表替换（replace=True）只认「生产模型主干」行：物料 / BOM / 订单 / 工艺路线。 # 刻意不含另外三类： # - flexOperations：解析器对任何工作簿都会生成 5 条钣金默认模板工序，空表也有； # - flexZones / flexEquipment：车间与设备属辅助台账，单独出现不足以证明 # 这是一份可整体替换的主数据收集表（例：只有一张 01-生产模型、一行车间记录）。 _MOM_CORE_KEYS = ("flexMaterials", "flexBom", "flexOrders", "flexRoutings") # 整表替换还要求结构证据：工作簿至少含 2 张模型表（与 is_mom_workbook 不靠文件名时的 # 判定同口径）。只有文件名匹配、或单表 + 解析出的标签行，都不足以背书清空既有主数据。 _MOM_MIN_MODEL_SHEETS = 2 # 重复/畸形表头行：不得造出「伪物料」主干行（02-工艺模型里 ERP品号/品号 已有同类先例）。 _HEADER_ROW_LABELS = frozenset({ "ERP品号", "品号", "物料编码", "物料名称", "产品编码", "产品名称", "编码", "名称", "品名", }) # 模板填写指引行 / 占位行：MOM 模板在数据区留了「填写说明」「请填写真实物料信息」之类的说明文字， # 客户未填写时这些行很容易被当成真实物料（并派生出假订单）。既不能入库，也不能作为整表替换证据。 # 一律「前缀 / 整值」匹配，不做任意位置子串匹配——「说明书」「测试夹具」这类真实行不能被误杀。 _PLACEHOLDER_CODE_PREFIXES = ( "填写", "填寫", "填表", "请填", "請填", "待填", "未填", "必填", "选填", "選填", "示例", "样例", "模板", "占位", "测试", "演示", "假数据", "无效", "test", "demo", "sample", "dummy", "placeholder", "tmp", "temp", "xxx", "xxxx", ) # 名称侧更窄：客户可能真有「测试夹具」这类物料名，所以名称不认 测试/演示 前缀。 _PLACEHOLDER_NAME_PREFIXES = ( "填写", "填寫", "填表", "请填", "請填", "待填", "未填", "示例", "样例", "模板", "占位", ) _PLACEHOLDER_VALUES = frozenset({ "说明", "备注", "提示", "注意", "模板", "无", "-", "--", "占位", "测试", "test", "demo", "sample", "dummy", "placeholder", "xx", "xxx", "xxxx", "n/a", "na", "tbd", "待定", "待补", "待补充", "null", "none", }) # 主干证据还要「像真实业务编码」：模板说明文字通常不含字母/数字，而真实主数据编码几乎总是 # 字母数字混排（A0050101-00280、FG-001、SO-001）。 _ALNUM_RE = re.compile(r"[0-9A-Za-z]") # 每类主干行的证据字段：(标识字段, 展示字段)。标识字段必须非占位且至少一个含字母/数字。 _MOM_CORE_EVIDENCE_FIELDS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = { "flexMaterials": (("code",), ("name",)), "flexBom": (("productCode", "materialCode"), ()), "flexOrders": (("productCode",), ("productName",)), "flexRoutings": (("productCode",), ("productName",)), } def mom_pack_row_count(pack: dict[str, Any]) -> int: """MOM 解析结果的生产模型主干行数（物料/BOM/订单/工艺路线）。 空表、仅表头、只有辅助台账（车间/设备）、只剩模板填写指引行，或只是文件名带 MOM 的普通工作簿 → 0； 与 mom_pack_is_adoptable() 一起构成 replace=True 的 fail-closed 准入。 只认「能追到真实物料行」的主干行：BOM / 订单 / 工艺路线的产品编码必须落在被判定为 真实物料的编码集合里，模板占位物料派生出的假订单、假路线因此不计入。 """ materials = [ row for row in (pack.get("flexMaterials") or []) if _row_is_business_evidence(row, keys=_MOM_CORE_EVIDENCE_FIELDS["flexMaterials"][0]) ] material_codes = {_cell(row.get("code")) for row in materials} total = len(materials) for key in ("flexBom", "flexOrders", "flexRoutings"): id_keys, _display_keys = _MOM_CORE_EVIDENCE_FIELDS[key] for row in pack.get(key) or []: if not _row_is_business_evidence(row, keys=id_keys): continue product_code = _cell(row.get("productCode")) if key == "flexBom": if (product_code in material_codes and _cell(row.get("materialCode")) in material_codes): total += 1 elif product_code in material_codes: total += 1 return total def mom_pack_is_adoptable(pack: dict[str, Any]) -> bool: """整表替换（replace=True）准入判定：结构证据 + 主干行证据，两者缺一不可。""" if int(pack.get("modelSheetHits") or 0) < _MOM_MIN_MODEL_SHEETS: return False return mom_pack_row_count(pack) > 0 def _cell(v: Any) -> str: if v is None: return "" return str(v).replace("\u200c", "").replace("\ufeff", "").strip() def _is_placeholder_text(value: Any, *, side: str = "code") -> bool: """模板填写指引 / 占位文本：既不是主数据，也不能充当整表替换的证据。 side="code" 用编码侧前缀（含测试/演示类前缀），side="name" 用更窄的名称侧前缀； 匹配是前缀或整值，避免把「说明书」「测试夹具」这类真实行误杀。 """ text = _cell(value) if not text: return False lowered = text.lower() if lowered in _PLACEHOLDER_VALUES: return True prefixes = _PLACEHOLDER_NAME_PREFIXES if side == "name" else _PLACEHOLDER_CODE_PREFIXES return any(lowered.startswith(prefix) for prefix in prefixes) def _row_is_business_evidence(row: Any, *, keys: tuple[str, ...]) -> bool: """主干行证据：标识字段非占位，且至少一个标识字段含字母/数字（真实业务编码形态）。""" if not isinstance(row, dict): return False values = [_cell(row.get(key)) for key in keys] values = [value for value in values if value] if not values or any(_is_placeholder_text(value) for value in values): return False return any(_ALNUM_RE.search(value) for value in values) 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 _HEADER_ROW_LABELS or name in _HEADER_ROW_LABELS or _is_placeholder_text(code) or _is_placeholder_text(name, side="name")): 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 _HEADER_ROW_LABELS or name in _HEADER_ROW_LABELS or _is_placeholder_text(code) or _is_placeholder_text(name, side="name")): 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), # 按「工作表张数」计，不按命中的 hint 数量：一个叫「生产模型-工艺模型」的单表 # 不能算作两张模型表的结构证据。 "modelSheetHits": sum( 1 for title in sheets if any(hint in title for hint in _SHEET_HINTS) ), "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}