Inventory-Management-System/tools/check_bom_exe.py

# -*- coding: utf-8 -*-
"""EXE 端到端验证(聚焦真实 BOM 模板支持):
1. 启动 MaterialWMS.exe
2. 登录
3. /boms 页面含 "批量导入窗口" + "Part Number" 真实格式说明
4. /template/bom 下载模板, 表头含 Part Number / Quantity
5. 生成多 sheet 真实格式 xlsx, POST /api/import/boms/preview (multipart) -> valid==2
6. POST /api/import/boms/commit -> created_materials >= 2
7. 校验 dist/app.db 中 Bom XM-A 存在
"""
import http.client, os, time, subprocess, sys, json
from openpyxl import Workbook

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _exe_util import kill_dist_exes, start_exe, stop_exe

HOST, PORT = "127.0.0.1", 8080
EXE = r"D:\2026-09-10-task-20\material_wms\dist\MaterialWMS.exe"
DB  = r"D:\2026-09-10-task-20\material_wms\dist\app.db"


def wait_port():
    for _ in range(60):
        try:
            c = http.client.HTTPConnection(HOST, PORT, timeout=2)
            c.request("GET", "/")
            r = c.getresponse(); r.read(); c.close()
            if r.status in (200, 302):
                return
        except Exception:
            time.sleep(1)
    raise RuntimeError("EXE 端口未起")


def req(method, path, body=None, headers=None):
    c = http.client.HTTPConnection(HOST, PORT, timeout=15)
    c.request(method, path, body=body, headers=headers or {})
    r = c.getresponse()
    raw = r.read()
    setc = r.getheader("Set-Cookie")
    c.close()
    return r.status, raw, setc


def post_file(path, file_path, cookie):
    with open(file_path, "rb") as f:
        data = f.read()
    boundary = "WMSBND"
    body = (
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="file"; filename="{os.path.basename(file_path)}"\r\n'
        f"Content-Type: application/octet-stream\r\n\r\n"
    ).encode() + data + f"\r\n--{boundary}--\r\n".encode()
    hdrs = {"Content-Type": f"multipart/form-data; boundary={boundary}",
            "Content-Length": str(len(body))}
    if cookie:
        hdrs["Cookie"] = cookie
    return req("POST", path, body=body, headers=hdrs)


def login(cookie):
    body = urllib_parse("username=admin&password=admin123")
    return req("POST", "/login", body=body,
               headers={"Content-Type": "application/x-www-form-urlencoded",
                        "Cookie": cookie or ""})


def urllib_parse(s):
    import urllib.parse as up
    return up.urlencode(dict(up.parse_qsl(s))).encode()


def main():
    # 1. 启动
    print("[1] 启动 EXE ...")
    p = start_exe(EXE)
    try:
        wait_port()
        print("    [OK] 端口就绪")

        # 2. 登录
        _, _, setc = req("GET", "/login")
        cookie = setc.split(";")[0] if setc else ""
        _, _, setc2 = login(cookie)
        if setc2:
            cookie = setc2.split(";")[0]
        # 验证登录
        s, html, _ = req("GET", "/dashboard", headers={"Cookie": cookie})
        assert s == 200 and "库存看板" in html.decode("utf-8", "ignore"), "登录失败"
        print("    [OK] 登录")

        # 3. BOM 页面含按钮 + 真实格式说明
        s, html, _ = req("GET", "/boms", headers={"Cookie": cookie})
        html = html.decode("utf-8", "ignore")
        if "批量导入窗口" not in html:
            print("[DEBUG] /boms status:", s)
            print("[DEBUG] HTML head:", html[:800])
            raise AssertionError("BOM 页缺按钮")
        assert "Part Number" in html, "BOM 页未说明真实格式"
        assert "嘉立创" in html or "Altium" in html, "BOM 页未提及 Altium/嘉立创"
        print("    [OK] BOM 页含按钮 + 真实格式说明")

        # 4. 下载 BOM 模板
        s, raw, _ = req("GET", "/template/bom", headers={"Cookie": cookie})
        assert s == 200, f"模板下载 status={s}"
        tmpdir = os.path.join(os.path.dirname(DB), "_check_tmp")
        os.makedirs(tmpdir, exist_ok=True)
        tpath = os.path.join(tmpdir, "_tpl_bom_check.xlsx")
        with open(tpath, "wb") as f: f.write(raw)
        from openpyxl import load_workbook
        tws = load_workbook(tpath).active
        hdr = [c.value for c in tws[1]]
        assert "Part Number" in hdr and "Quantity" in hdr, hdr
        print(f"    [OK] BOM 模板表头: {hdr}")

        # 5. 生成多 sheet 真实格式 xlsx
        wb = Workbook()
        ws1 = wb.active; ws1.title = "XM-A"
        ws1.append(["Item", "Part Number", "Manufacture Part Number", "Quantity", "Reference", "Part", "Package"])
        ws1.append([1, "EXA100", "MFA", 2, "U1", "MFA", "PKGA"])
        ws1.append([2, "EXA101", "MFB", 1, "U2", "MFB", "PKGB"])
        ws2 = wb.create_sheet("组装")
        ws2.append(["序号", "层级", "物料代码", "描述", "数量", "材质"])
        ws2.append([1, "0.1", "EXA200", "外壳", 1, "ABS"])
        up = os.path.join(tmpdir, "_bom_check.xlsx")
        wb.save(up)

        # 6. 上传预览
        s, raw, _ = post_file("/api/import/boms/preview", up, cookie)
        pv = json.loads(raw.decode("utf-8", "ignore"))
        codes = [b["code"] for b in pv["boms"]]
        assert "XM-A" in codes and "组装" in codes, pv
        assert pv["valid"] == 2, pv
        print(f"    [OK] 上传预览: 解析 {len(codes)} 个 BOM, valid={pv['valid']}")

        # 7. 提交
        body = json.dumps({"boms": pv["boms"]}).encode()
        s, raw, _ = req("POST", "/api/import/boms/commit", body=body,
                        headers={"Content-Type": "application/json", "Cookie": cookie})
        rs = json.loads(raw.decode("utf-8", "ignore"))
        assert rs["boms"] == 2 and rs["created_materials"] >= 2, rs
        print(f"    [OK] 提交: boms={rs['boms']} added={rs['added_items']} created_materials={rs['created_materials']}")

        # 8. 校验数据库
        import sqlite3
        con = sqlite3.connect(DB)
        n = con.execute("select count(*) from bom where code in ('XM-A','组装')").fetchone()[0]
        con.close()
        assert n >= 2, f"DB 中 BOM 不足: {n}"
        print(f"    [OK] 数据库确认: BOM 表新增 {n} 条")

        print("\nEXE 真实 BOM 模板 端到端验证通过")
    finally:
        stop_exe(p)


if __name__ == "__main__":
    main()