aps-agent/server/importers/mom_pack.py

# ============================================================
# 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}