# ============================================================ # 现场完整生产路线导入黄金测试 # ============================================================ from __future__ import annotations from pathlib import Path import pytest from server.aps_domain.kangni_intake import ( DEFAULT_ROUTE_XLSX, build_flex_bundle, load_site_into_world, parse_complete_route_xlsx, ) from server.engines import PoolEngine from server.state.seed import seed_world ROUTE = Path(DEFAULT_ROUTE_XLSX) @pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机") def test_parse_complete_route_order_102285668(): pkg = parse_complete_route_xlsx(ROUTE) assert pkg["orderNo"] == "102285668" assert pkg["productCode"] == "28200003654300" assert len(pkg["operations"]) == 10 assert pkg["operations"][0]["operationCode"] == "0503727Z1M" assert len(pkg["bom"]) >= 15 assert all(op["stdTimePerUnit"] > 0 for op in pkg["operations"]) assert any(op["stdTimeInferred"] for op in pkg["operations"]) @pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机") def test_load_site_writes_master_and_orders_and_schedules(): world = seed_world() assert any(m["code"].startswith("CTRL-") for m in world["materials"]) assert any(o["orderNo"].startswith("SO") for o in world["salesOrders"]) meta = load_site_into_world(world, route_path=ROUTE, include_sibling_orders=False, clear_all=True) assert meta["primaryOrder"] == "102285668" assert meta.get("clearedDemo") is True # 演示垃圾已清 assert not any(m["code"].startswith("CTRL-") for m in world["materials"]) assert not any(o["orderNo"].startswith("FO-") for o in world["flexOrders"]) assert not any(o["orderNo"].startswith("SO20") for o in world["salesOrders"]) # 主数据页可读 assert world["factories"][0]["name"] == "城轨机构装配工厂" assert len(world["operations"]) == 10 assert any(m["code"] == "28200003654300" for m in world["materials"]) assert len(world["bomItems"]) >= 15 assert len(world["routingSteps"]) == 10 assert world["salesOrders"][0]["orderNo"] == "102285668" assert world["salesOrders"][0]["status"] == "APPROVED" # 柔性轨 assert world["flexOrders"][0]["orderNo"] == "102285668" assert world["flexEquipment"] assert world["flexParams"].get("fixedMasterSynced") is True counters: dict[str, int] = {} def nid(k: str) -> int: counters[k] = counters.get(k, 0) + 1 return counters[k] result = PoolEngine().solve(world, nid, sort_mode="BOTTLENECK", start_date="2026-06-26") assert result["orderCount"] == 1 assert result["vlCount"] == 1 assert result["woCount"] == 10 @pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机") def test_bundle_meta_has_inferences(): bundle = build_flex_bundle(ROUTE, include_sibling_orders=False) assert bundle["_meta"]["inferences"] assert any("工时" in x or "工位" in x for x in bundle["_meta"]["inferences"]) def _write_std_time_fixtures(tmp_path: Path) -> tuple[Path, Path]: from openpyxl import Workbook route = tmp_path / "工艺路线.xlsx" wb = Workbook() ws = wb.active ws.title = "102285668" ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时", "排序号"]) ws.append(["1", "0503727Z1M", "干线机构部装", "部装", None, "0010"]) ws.append(["2", "0503722Z1M", "干线机构总装1", "装配", None, "0020"]) wb.save(route) std = tmp_path / "工时.xlsx" wb2 = Workbook() ws2 = wb2.active ws2.title = "102285668" ws2.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时/分钟", "排序号"]) ws2.append(["1", "0503727Z1M", "干线机构部装", "部装", 27.0, "0010"]) ws2.append(["2", "0503722Z1M", "干线机构总装1", "装配", 5.0, "0020"]) wb2.save(std) return route, std def test_routing_blank_std_prefers_std_time_workbook(tmp_path): from server.aps_domain.kangni_intake import load_std_time_map, parse_routing_sheet route, _ = _write_std_time_fixtures(tmp_path) times = load_std_time_map(tmp_path) assert times["102285668"]["0503727Z1M"] == 27.0 ops = parse_routing_sheet(route, "102285668", std_times=times["102285668"]) assert ops[0]["stdTimePerUnit"] == 27.0 assert ops[0]["stdTimeInferred"] is False assert ops[0]["stdTimeSource"] == "工时表" assert ops[1]["stdTimePerUnit"] == 5.0 def test_routing_blank_std_falls_back_to_inference(tmp_path): from server.aps_domain.kangni_intake import parse_routing_sheet route, _ = _write_std_time_fixtures(tmp_path) ops = parse_routing_sheet(route, "102285668") assert ops[0]["stdTimeInferred"] is True assert ops[0]["stdTimeSource"] == "推断" assert ops[0]["stdTimePerUnit"] > 0 def _write_mapping_fixtures(tmp_path: Path) -> tuple[Path, Path]: from openpyxl import Workbook equip = tmp_path / "设备能力映射模板.xlsx" wb = Workbook() ws = wb.active ws.title = "设备能力" ws.append(["设备编号", "设备名称", "可执行工序编号", "单件工时(分钟,工序:分钟;多值分号)", "可动率", "是否可移动", "移动耗时(分钟)", "区域编码", "适配模具编号", "状态", "备注"]) ws.append(["EQ-CG-01", "城轨装配台1", "0503727Z1M,0503722Z1M", "0503727Z1M:27;0503722Z1M:5", 0.9, "否", 0, "ZONE-CG", "MOLD-01", "RUNNING", ""]) wb.save(equip) mold = tmp_path / "模具适配映射模板.xlsx" wb2 = Workbook() ws2 = wb2.active ws2.title = "模具适配" ws2.append(["模具编号", "模具名称", "适用工序编号", "适配设备编号(逗号分隔)", "寿命上限", "已用寿命", "区域编码", "换型耗时(分钟)", "状态", "备注"]) ws2.append(["MOLD-01", "压接模具A", "0503722Z1M", "EQ-CG-01", 50000, 12000, "ZONE-CG", 20, "AVAILABLE", ""]) wb2.save(mold) return equip, mold def test_parse_equipment_and_mold_maps(tmp_path): from server.aps_domain.kangni_intake import ( parse_equipment_capability_map, parse_mold_adaptation_map, ) equip, mold = _write_mapping_fixtures(tmp_path) rows = parse_equipment_capability_map(equip) assert rows[0]["code"] == "EQ-CG-01" assert rows[0]["capabilities"] == ["0503727Z1M", "0503722Z1M"] assert rows[0]["opStdTime"]["0503727Z1M"] == 27.0 assert rows[0]["movable"] is False assert rows[0]["availabilityRate"] == 0.9 mrows = parse_mold_adaptation_map(mold) assert mrows[0]["code"] == "MOLD-01" assert mrows[0]["operationCode"] == "0503722Z1M" assert mrows[0]["adaptableEquipment"] == ["EQ-CG-01"] assert mrows[0]["lifeTotal"] == 50000 def test_build_flex_equipment_prefers_mapping_table(tmp_path): from server.aps_domain.kangni_intake import build_flex_equipment_and_molds _write_mapping_fixtures(tmp_path) equipment, molds, mold_ops, infs = build_flex_equipment_and_molds( tmp_path, [], {"0503727Z1M": 27.0, "0503722Z1M": 5.0}, ["0503727Z1M", "0503722Z1M"], {"zone": {"code": "ZONE-CG"}}, ) assert len(equipment) == 1 assert equipment[0]["inferred"] is False assert equipment[0]["availabilityRate"] == 0.9 assert len(molds) == 1 assert "0503722Z1M" in mold_ops assert any("设备能力映射表" in x for x in infs) assert any("模具适配映射表" in x for x in infs) def test_build_flex_equipment_falls_back_to_synthetic(tmp_path): from server.aps_domain.kangni_intake import build_flex_equipment_and_molds equipment, molds, mold_ops, infs = build_flex_equipment_and_molds( tmp_path, [], {"OP-1": 10.0}, ["OP-1"], {"stationCount": 2, "zone": {"code": "ZONE-CG"}, "availabilityRate": 0.95}, ) assert len(equipment) == 2 assert all(e["inferred"] for e in equipment) assert molds == [] assert mold_ops == set() assert any("合成 2 台" in x for x in infs) def test_fixed_master_preserves_measured_table_and_inferred_time_sources(): from server.aps_domain.kangni_intake import _apply_std_time_map, build_fixed_master package = { "orderNo": "TIME-SOURCE-001", "productCode": "TIME-SOURCE-PRODUCT", "productName": "工时来源验证产品", "drawingNo": "DRAWING-001", "quantity": 1.0, "planStart": "2026-08-17", "dueDate": "2026-08-20", "wbs": "WBS-TIME-SOURCE", "project": "康尼工时来源验证", "status": "RELEASED", "kitStatus": "未知", "bom": [], "operations": [ { "seq": 10, "operationCode": "OP-MEASURED", "operationName": "实测工序", "opType": "装配", "stdTimePerUnit": 11.0, "stdTimeInferred": False, "stdTimeSource": "实测", }, { "seq": 20, "operationCode": "OP-TABLE", "operationName": "工时表工序", "opType": "装配", "stdTimePerUnit": 30.0, "stdTimeInferred": True, "stdTimeSource": "推断", }, { "seq": 30, "operationCode": "OP-INFERRED", "operationName": "推断工序", "opType": "装配", "stdTimePerUnit": 33.0, "stdTimeInferred": True, "stdTimeSource": "推断", }, ], } _apply_std_time_map( package, {"TIME-SOURCE-001": {"OP-MEASURED": 99.0, "OP-TABLE": 22.0}}, ) assert package["operations"][0]["stdTimePerUnit"] == 11.0 assert package["operations"][0]["stdTimeSource"] == "实测" assert package["operations"][1]["stdTimePerUnit"] == 22.0 assert package["operations"][1]["stdTimeSource"] == "工时表" assert package["operations"][2]["stdTimePerUnit"] == 33.0 assert package["operations"][2]["stdTimeSource"] == "推断" fixed = build_fixed_master([package], station_count=1) steps = fixed["routingSteps"] assert [row["runTimePerUnit"] for row in steps] == [11.0, 22.0, 33.0] assert [row["stdTimeSource"] for row in steps] == ["实测", "工时表", "推断"] def _write_route_book(path: Path, sheets: dict[str, list[tuple[str, str, str, str]]]) -> None: from openpyxl import Workbook wb = Workbook() wb.remove(wb.active) for sheet_name, operations in sheets.items(): ws = wb.create_sheet(sheet_name) ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时", "排序号"]) for index, (code, seq, name, op_type) in enumerate(operations, start=1): ws.append([index, code, name, op_type, None, seq]) wb.save(path) def _write_std_book(path: Path, orders: dict[str, list[tuple[str, str, str, str, float]]]) -> None: from openpyxl import Workbook wb = Workbook() wb.remove(wb.active) for order_no, operations in orders.items(): ws = wb.create_sheet(order_no) ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时/分钟", "排序号"]) for index, (code, seq, name, op_type, minutes) in enumerate(operations, start=1): ws.append([index, code, name, op_type, minutes, seq]) wb.save(path) def _write_bom_book(path: Path, sheets: dict[str, list[tuple[str, str, str, float, str]]]) -> None: from openpyxl import Workbook wb = Workbook() wb.remove(wb.active) for sheet_name, rows in sheets.items(): ws = wb.create_sheet(sheet_name) ws.append(["工单编号", "MES工序编号", "物料编号", "物料名称", "单位用量"]) for order_no, operation, material, quantity, name in rows: ws.append([order_no, operation, material, name, quantity]) wb.save(path) def _write_orders_book(path: Path, order_numbers: list[str]) -> None: from openpyxl import Workbook wb = Workbook() ws = wb.active ws.title = "订单" ws.append([ "生产订单", "WBS号", "料号", "物料名称", "计划开始时间", "计划结束时间", "订单数量", ]) for index, order_no in enumerate(order_numbers, start=1): ws.append([ order_no, f"WBS-{index:03}", f"PRODUCT-{index:03}", f"产品{index:03}", "2026-08-01", "2026-08-31", 1, ]) wb.save(path) def _write_data_dir_fixture(root: Path, order_count: int = 10) -> list[str]: root.mkdir(parents=True, exist_ok=True) order_numbers = [f"ORDER-{index:03}" for index in range(1, order_count + 1)] _write_orders_book(root / "订单.xlsx", order_numbers) _write_route_book(root / "工艺路线.xlsx", { order_no: [(f"OP-{index:03}", "0010", f"工序{index:03}", "装配")] for index, order_no in enumerate(order_numbers, start=1) }) _write_std_book(root / "工时.xlsx", { order_no: [(f"OP-{index:03}", "0010", f"工序{index:03}", "装配", float(index))] for index, order_no in enumerate(order_numbers, start=1) }) _write_bom_book(root / "BOM.xlsx", { order_no: [(order_no, f"OP-{index:03}", f"MAT-{index:03}", 1.0, f"物料{index:03}")] for index, order_no in enumerate(order_numbers, start=1) }) return order_numbers def test_data_dir_only_payload_builds_10_orders_and_apply_is_pure(tmp_path: Path): import json import shutil from copy import deepcopy from server.aps_domain.kangni_intake import ( apply_site_payload_to_world, build_site_payload_from_data_dir, ) data_dir = tmp_path / "upload" _write_data_dir_fixture(data_dir) assert not list(data_dir.glob("*完整生产路线*.xlsx")) payload = build_site_payload_from_data_dir(data_dir, station_count=2) serialized = json.dumps(payload, ensure_ascii=False, sort_keys=True, allow_nan=False) assert json.loads(serialized)["meta"]["sourceMode"] == "data-dir-only" assert str(data_dir) not in serialized assert payload["meta"]["sourceFiles"] == { "orders": "订单.xlsx", "routing": "工艺路线.xlsx", "stdtime": "工时.xlsx", "bom": "BOM.xlsx", "workbooks": ["BOM.xlsx", "工时.xlsx", "工艺路线.xlsx", "订单.xlsx"], } assert len(payload["fixed"]["salesOrders"]) == 10 assert len(payload["fixed"]["routingSteps"]) == 10 assert len(payload["flex"]["flexOrders"]) == 10 assert len(payload["flex"]["flexRoutings"]) == 10 assert len(payload["flex"]["flexOperations"]) == 10 assert len(payload["flex"]["flexBom"]) == 10 assert payload["meta"]["bomSourceRowCount"] == 10 assert payload["meta"]["stdTimeSourceCounts"] == {"工时表": 10} assert all( row["stdTimeSource"] == "工时表" for row in payload["flex"]["flexRoutings"] ) mirror_dir = tmp_path / "different-upload-root" mirror_dir.mkdir() for source in data_dir.glob("*.xlsx"): shutil.copy2(source, mirror_dir / source.name) mirror_payload = build_site_payload_from_data_dir(mirror_dir, station_count=2) mirror_serialized = json.dumps( mirror_payload, ensure_ascii=False, sort_keys=True, allow_nan=False, ) assert str(mirror_dir) not in mirror_serialized assert mirror_serialized == serialized payload_before = deepcopy(payload) world: dict = {"unrelated": {"keep": True}} first_meta = apply_site_payload_to_world(world, payload) first_world = deepcopy(world) second_meta = apply_site_payload_to_world(world, payload) assert payload == payload_before assert world == first_world assert first_meta == second_meta assert world["unrelated"] == {"keep": True} assert len(world["salesOrders"]) == 10 assert len(world["flexOrders"]) == 10 def test_data_dir_only_payload_rejects_routing_ambiguity_and_aggregates_bom_usage(tmp_path: Path): from server.aps_domain.kangni_intake import ( KangniDataMappingError, build_site_payload_from_data_dir, ) routing_dir = tmp_path / "routing-ambiguous" _write_data_dir_fixture(routing_dir, order_count=1) _write_route_book(routing_dir / "工艺路线.xlsx", { "Sheet2": [("OP-001", "0010", "名称A", "装配")], "Sheet4": [("OP-001", "0010", "名称B", "装配")], }) with pytest.raises(KangniDataMappingError, match="内容不一致"): build_site_payload_from_data_dir(routing_dir) bom_dir = tmp_path / "bom-ambiguous" _write_data_dir_fixture(bom_dir, order_count=1) row = ("ORDER-001", "OP-001", "MAT-001", 1.0, "物料001") _write_bom_book(bom_dir / "BOM.xlsx", { "Sheet2": [row], "Sheet3": [("ORDER-001", "OP-001", "MAT-001", 2.0, "物料001")], }) payload = build_site_payload_from_data_dir(bom_dir) assert payload["flex"]["flexBom"][0]["quantity"] == 3.0 bom_mapping = payload["meta"]["bomMappings"]["ORDER-001"] assert bom_mapping["aggregatedRows"] == 1 assert bom_mapping["aggregationGroups"][0]["quantity"] == 3.0 def test_routing_exact_sheet_has_priority_over_generic(tmp_path: Path): from server.aps_domain.kangni_intake import parse_routing_sheet route = tmp_path / "工艺路线.xlsx" _write_route_book(route, { "ORDER-EXACT": [("OP-EXACT", "0010", "精确工序", "装配")], "Sheet2": [("OP-GENERIC", "0010", "通用工序", "装配")], }) diagnostics = {} operations = parse_routing_sheet( route, "ORDER-EXACT", std_times={"OP-EXACT": 12.0}, canonical_operations=[{"operationCode": "OP-GENERIC", "seq": 10}], diagnostics=diagnostics, ) assert [row["operationCode"] for row in operations] == ["OP-EXACT"] assert operations[0]["stdTimeSource"] == "工时表" assert diagnostics["matchMode"] == "exact-sheet" assert diagnostics["matchedSheetCandidates"] == ["ORDER-EXACT"] def test_generic_routing_accepts_identical_candidates_and_records_evidence(tmp_path: Path): from server.aps_domain.kangni_intake import ( load_std_route_map, load_std_time_map, parse_routing_sheet, ) route = tmp_path / "工艺路线.xlsx" shared = [ ("OP-1", "0070", "部装", "部装"), ("OP-2", "0080", "总装", "装配"), ] _write_route_book(route, { "Sheet2": shared, "Sheet3": [("OP-X", "0010", "其它", "装配")], "Sheet4": shared, }) _write_std_book(tmp_path / "工时.xlsx", { "ORDER-A": [ ("OP-1", "0070", "部装", "部装", 27.0), ("OP-2", "0080", "总装", "装配", 5.0), ], }) diagnostics = {} operations = parse_routing_sheet( route, "ORDER-A", std_times=load_std_time_map(tmp_path)["ORDER-A"], canonical_operations=load_std_route_map(tmp_path)["ORDER-A"], diagnostics=diagnostics, ) assert [row["operationCode"] for row in operations] == ["OP-1", "OP-2"] assert [row["stdTimePerUnit"] for row in operations] == [27.0, 5.0] assert diagnostics["matchedSheet"] == "Sheet2" assert diagnostics["matchedSheetCandidates"] == ["Sheet2", "Sheet4"] def test_generic_routing_rejects_missing_or_ambiguous_content(tmp_path: Path): from server.aps_domain.kangni_intake import KangniDataMappingError, parse_routing_sheet route = tmp_path / "工艺路线.xlsx" _write_route_book(route, { "Sheet2": [("OP-1", "0010", "名称A", "装配")], "Sheet4": [("OP-1", "0010", "名称B", "装配")], }) canonical = [{"operationCode": "OP-1", "seq": 10, "operationName": "名称A", "opType": "装配"}] with pytest.raises(KangniDataMappingError, match="内容不一致"): parse_routing_sheet(route, "ORDER-A", canonical_operations=canonical) with pytest.raises(KangniDataMappingError, match="没有.*一致"): parse_routing_sheet( route, "ORDER-MISSING", canonical_operations=[{"operationCode": "OP-NONE", "seq": 10}], ) with pytest.raises(KangniDataMappingError, match="无规范工序序列"): parse_routing_sheet(route, "ORDER-NO-CANONICAL") def test_bom_exact_and_generic_sheets_filter_orders_without_cross_talk(tmp_path: Path): from server.aps_domain.kangni_intake import parse_bom_sheet bom = tmp_path / "BOM.xlsx" _write_bom_book(bom, { "ORDER-A": [("ORDER-A", "OP-1", "MAT-A1", 1.0, "物料A1")], "Sheet2": [ ("ORDER-B", "OP-B", "MAT-B", 1.0, "物料B"), ("ORDER-A", "OP-2", "MAT-A2", 2.0, "物料A2"), ], }) diagnostics = {} rows = parse_bom_sheet(bom, "ORDER-A", diagnostics=diagnostics) assert [row["materialCode"] for row in rows] == ["MAT-A1", "MAT-A2"] assert diagnostics["matchedSheetCandidates"] == ["ORDER-A", "Sheet2"] missing = {} assert parse_bom_sheet(bom, "ORDER-C", diagnostics=missing) == [] assert missing["status"] == "missing" def test_bom_cross_sheet_duplicate_is_evidenced_and_distinct_usage_is_aggregated(tmp_path: Path): from server.aps_domain.kangni_intake import parse_bom_sheet duplicate = tmp_path / "BOM-duplicate.xlsx" row = ("ORDER-A", "OP-1", "MAT-A", 1.0, "物料A") _write_bom_book(duplicate, {"Sheet2": [row], "Sheet3": [row]}) diagnostics = {} rows = parse_bom_sheet(duplicate, "ORDER-A", diagnostics=diagnostics) assert len(rows) == 1 assert diagnostics["sourceRowCount"] == 2 assert diagnostics["deduplicatedRows"] == 1 conflict = tmp_path / "BOM-conflict.xlsx" _write_bom_book(conflict, { "Sheet2": [row], "Sheet3": [("ORDER-A", "OP-1", "MAT-A", 2.0, "物料A")], }) aggregated = {} rows = parse_bom_sheet(conflict, "ORDER-A", diagnostics=aggregated) assert rows[0]["quantity"] == 3.0 assert aggregated["aggregatedRows"] == 1 assert aggregated["aggregationGroups"][0]["sourceRows"] == 2 def test_bom_same_sheet_exact_signature_is_deduplicated_and_distinct_usage_is_summed(tmp_path: Path): from server.aps_domain.kangni_intake import parse_bom_sheet duplicate = tmp_path / "BOM-same-sheet-duplicate.xlsx" row = ("ORDER-A", "OP-1", "MAT-A", 1.0, "物料A") _write_bom_book(duplicate, {"Sheet2": [row, row]}) diagnostics = {} rows = parse_bom_sheet(duplicate, "ORDER-A", diagnostics=diagnostics) assert len(rows) == 1 assert diagnostics["sourceRowCount"] == 2 assert diagnostics["outputRowCount"] == 1 assert diagnostics["deduplicatedRows"] == 1 quantity_conflict = tmp_path / "BOM-same-sheet-quantity-conflict.xlsx" _write_bom_book(quantity_conflict, { "Sheet2": [ row, ("ORDER-A", "OP-1", "MAT-A", 2.0, "物料A"), ], }) aggregation = {} rows = parse_bom_sheet(quantity_conflict, "ORDER-A", diagnostics=aggregation) assert rows[0]["quantity"] == 3.0 assert rows[0]["isKey"] is True assert aggregation["aggregatedRows"] == 1 def test_flex_bom_projection_preserves_same_material_at_different_consume_ops(tmp_path: Path): from server.aps_domain.kangni_intake import ( build_site_payload_from_data_dir, ) data_dir = tmp_path / "consume-op-conflict" _write_data_dir_fixture(data_dir, order_count=1) _write_bom_book(data_dir / "BOM.xlsx", { "ORDER-001": [ ("ORDER-001", "OP-001", "MAT-SHARED", 1.0, "共享物料"), ("ORDER-001", "OP-ALT", "MAT-SHARED", 1.0, "共享物料"), ], }) payload = build_site_payload_from_data_dir(data_dir) projected = payload["flex"]["flexBom"] assert {(row["materialCode"], row["consumeOp"]) for row in projected} == { ("MAT-SHARED", "OP-001"), ("MAT-SHARED", "OP-ALT"), } def test_flex_bom_projection_cross_order_conflict_and_exact_dedup(tmp_path: Path): from openpyxl import load_workbook from server.aps_domain.kangni_intake import ( KangniDataMappingError, build_site_payload_from_data_dir, ) conflict_dir = tmp_path / "cross-order-conflict" order_numbers = _write_data_dir_fixture(conflict_dir, order_count=2) wb = load_workbook(conflict_dir / "订单.xlsx") ws = wb.active ws.cell(row=2, column=3, value="PRODUCT-SHARED") ws.cell(row=3, column=3, value="PRODUCT-SHARED") ws.cell(row=2, column=4, value="共享产品") ws.cell(row=3, column=4, value="共享产品") wb.save(conflict_dir / "订单.xlsx") shared_route = [("OP-SHARED", "0010", "共享工序", "装配")] _write_route_book(conflict_dir / "工艺路线.xlsx", { order_no: shared_route for order_no in order_numbers }) _write_std_book(conflict_dir / "工时.xlsx", { order_no: [("OP-SHARED", "0010", "共享工序", "装配", 5.0)] for order_no in order_numbers }) _write_bom_book(conflict_dir / "BOM.xlsx", { order_numbers[0]: [(order_numbers[0], "OP-SHARED", "MAT-SHARED", 1.0, "共享物料")], order_numbers[1]: [(order_numbers[1], "OP-SHARED", "MAT-SHARED", 2.0, "共享物料")], }) with pytest.raises(KangniDataMappingError, match="flex BOM 投影冲突"): build_site_payload_from_data_dir(conflict_dir) dedup_dir = tmp_path / "cross-order-exact-dedup" order_numbers = _write_data_dir_fixture(dedup_dir, order_count=2) wb = load_workbook(dedup_dir / "订单.xlsx") ws = wb.active ws.cell(row=2, column=3, value="PRODUCT-SHARED") ws.cell(row=3, column=3, value="PRODUCT-SHARED") ws.cell(row=2, column=4, value="共享产品") ws.cell(row=3, column=4, value="共享产品") wb.save(dedup_dir / "订单.xlsx") shared_route = [("OP-SHARED", "0010", "共享工序", "装配")] _write_route_book(dedup_dir / "工艺路线.xlsx", { order_no: shared_route for order_no in order_numbers }) _write_std_book(dedup_dir / "工时.xlsx", { order_no: [("OP-SHARED", "0010", "共享工序", "装配", 5.0)] for order_no in order_numbers }) _write_bom_book(dedup_dir / "BOM.xlsx", { order_no: [(order_no, "OP-SHARED", "MAT-SHARED", 1.0, "共享物料")] for order_no in order_numbers }) payload = build_site_payload_from_data_dir(dedup_dir) assert payload["meta"]["bomSourceRowCount"] == 2 assert payload["meta"]["projectedBomCount"] == 1 assert payload["flex"]["flexBom"] == [{ "productCode": "PRODUCT-SHARED", "materialCode": "MAT-SHARED", "quantity": 1.0, "consumeOp": "OP-SHARED", "isKey": True, }] def test_flex_material_name_conflict_across_orders_fails_closed(tmp_path: Path): from server.aps_domain.kangni_intake import ( KangniDataMappingError, build_site_payload_from_data_dir, ) data_dir = tmp_path / "material-name-conflict" order_numbers = _write_data_dir_fixture(data_dir, order_count=2) _write_bom_book(data_dir / "BOM.xlsx", { order_numbers[0]: [(order_numbers[0], "OP-001", "MAT-SHARED", 1.0, "共享物料A")], order_numbers[1]: [(order_numbers[1], "OP-002", "MAT-SHARED", 1.0, "共享物料B")], }) with pytest.raises(KangniDataMappingError, match="物料编码 MAT-SHARED 名称冲突"): build_site_payload_from_data_dir(data_dir) def test_bom_and_routing_same_generic_name_are_not_bound(tmp_path: Path): from server.aps_domain.kangni_intake import parse_bom_sheet, parse_routing_sheet route = tmp_path / "工艺路线.xlsx" _write_route_book(route, { "Sheet3": [("OP-B", "0010", "B工序", "装配")], "Sheet4": [("OP-A", "0010", "A工序", "装配")], }) bom = tmp_path / "BOM.xlsx" _write_bom_book(bom, { "Sheet3": [("ORDER-A", "OP-A", "MAT-A", 1.0, "物料A")], "Sheet4": [("ORDER-B", "OP-B", "MAT-B", 1.0, "物料B")], }) bom_diagnostics = {} routing_diagnostics = {} assert parse_bom_sheet(bom, "ORDER-A", diagnostics=bom_diagnostics) operations = parse_routing_sheet( route, "ORDER-A", canonical_operations=[{ "operationCode": "OP-A", "seq": 10, "operationName": "A工序", "opType": "装配", }], diagnostics=routing_diagnostics, ) assert [row["operationCode"] for row in operations] == ["OP-A"] assert bom_diagnostics["matchedSheet"] == "Sheet3" assert routing_diagnostics["matchedSheet"] == "Sheet4"