commit 7d34a481b343041124f6b570a1ae3177e83ed93d Author: star <395723461@qq.com> Date: Fri Sep 11 15:31:55 2026 +0800 feat: 库存管理系统 MaterialWMS 初始版本 - Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d0cb41f --- /dev/null +++ b/.gitignore @@ -0,0 +1,25 @@ +# ---- Python ---- +__pycache__/ +*.pyc +*.pyo +.venv/ +venv/ + +# ---- PyInstaller 打包产物(体积大, 不入库; EXE 走 GitLab Release 分发) ---- +build/ +dist/ +*.spec + +# ---- 运行时日志(EXE 冻结时排查用) ---- +*.log +error.log +startup.log +request.log + +# ---- 临时校验文件 ---- +_check_tmp/ + +# ---- 编辑器 ---- +.vscode/ +.idea/ +*.swp diff --git a/BOM导入模板.xlsx b/BOM导入模板.xlsx new file mode 100644 index 0000000..bd88eb9 Binary files /dev/null and b/BOM导入模板.xlsx differ diff --git a/README.md b/README.md new file mode 100644 index 0000000..bb2b80c --- /dev/null +++ b/README.md @@ -0,0 +1,76 @@ +# 物料入库 / 出库系统 (M1 + M2 + M3 完整版) + +本地 Web 系统,对标桌面 `物料明细表.xlsx` 的台账 + BOM 核算能力。Flask + SQLite,免外部数据库,即开即用。 +库存由 SQL 实时聚合(在库 = 入库 − 出库),彻底规避原 Excel 的 `[1]物料入库!D_` 外部链接断链风险。 + +## 运行方式 + +**方式一 · 单文件 EXE(推荐车间用)** +`dist/MaterialWMS.exe` 双击即开,自动弹出浏览器到 http://127.0.0.1:8080。 +库文件 `app.db` 自动生成在 exe 同目录,换机器直接拷贝 exe 即可。 + +**方式二 · 源码(双击 bat)** +双击 `启动.bat` 启动;或命令行 `python app.py`。 + +**方式三 · 指定库位置** +`set WMS_DB=D:/path/app.db` 后再启动,可把数据放到任意位置。 + +## 默认账号 +- 管理员 `admin / admin123`(首次启动自动建;可在「用户」页加 operator / manager) + +## 角色权限 +| 角色 | 权限 | +|---|---| +| admin | 全部,含用户管理 | +| manager | 物料/BOM/订单/报表/日志/Excel 导入导出 | +| operator | 库存看板 / 物料查看 / 入库 / 出库 | + +## 功能清单 +| 模块 | 说明 | +|---|---| +| 库存看板 | 实时核算在库;低于安全库存标黄、库存为负标红,顶部统计负数/预警数 | +| 物料 | 主数据 CRUD(编号/型号/单位/安全库存)+ 搜索 | +| 入库 | 流水登记(批次/数量/经手人/来源),写入即更新库存 | +| 出库 | 流水登记,**库存不足自动拦截** | +| **批量导入窗口** | 入库/出库/物料/BOM 页「批量导入窗口」按钮弹出模态框:粘贴多行清单或上传 Excel/CSV,实时预览解析与校验,一键批量生成。各页格式见下 | +| 入库/出库批量窗口 | 每行 `物料编号 数量 批次(可选) 日期(可选) 备注(可选)`,校验物料存在/数量>0/出库库存充足 | +| 物料批量窗口 | 每行 `物料编号 型号/规格 单位 安全库存(可选) 期初库存(可选)`,校验重复编号;已存在编号显示「更新」,一键批量 upsert(含期初流水) | +| BOM 批量窗口 | ① **上传真实 Altium / 嘉立创 BOM 模板**(`Item/Part Number/Manufacture Part Number/Quantity/Reference/Part/Package`):每个 worksheet 视为一个 BOM,sheet 名即 BOM 型号,Part Number 关联物料、Quantity 列即单套用量(若存在两个 Quantity 列,取首列,首列空时回退次列);也支持中文「组装」sheet(物料代码 + 数量)。② 粘贴文本:以 `#BOM <型号> <名称>` 分段、其下 `物料编号 单套用量`。物料不存在的项预览时标「新建」,**导入时自动新建物料**(并写入型号/规格),一键多 BOM 落库 | +| **BOM(M2)** | BOM 型号 + 组成物料单套用量;支持 Excel 导入 | +| **生产订单(M2)** | 选 BOM + 接单量 → 自动展开「需求 = 单套用量 × 接单量」,一键「确认领料」生成出库,缺料在看板预警 | +| **Excel 双向(M2)** | 物料导入/导出、BOM 导入、库存报表导出、导入模板下载 | +| **报表(M3)** | 库存报表页,支持打印(浏览器打印)与导出 Excel | +| **操作日志(M3)** | 每次收发/BOM/订单/导入/用户操作全记录(谁/何时/做了什么) | +| **用户管理(M3)** | admin 增删改角色、启停账号 | + +## 数据来源 +`init_db.py` 已从桌面 `物料明细表.xlsx` 的「物料入库」表导入 **439 条真实物料**, +把 Excel 库存数量作为"期初入库"写入,看板开箱即有数。 + +## 目录结构 +``` +material_wms/ +├─ app.py # 路由 + 鉴权 + 库存核算 + 冻结路径处理 +├─ models.py # SQLAlchemy 2.0 模型(user/material/inbound/outbound/bom/bom_item/prod_order/operation_log) +├─ excel_io.py # Excel 双向导入导出 + 模板生成 +├─ init_db.py # 建表 + 管理员 + Excel 导入真实物料 +├─ build_exe.py # PyInstaller 单文件打包脚本 +├─ requirements.txt +├─ 启动.bat +├─ templates/ # 登录/看板/物料/收发/BOM/订单/报表/日志/用户/导入 共 15 个页面 +├─ static/style.css +├─ instance/app.db # SQLite 库(源码模式自动生成) +├─ dist/MaterialWMS.exe # 打包产物(单文件) +└─ tools/ # e2e_test.py(全量冒烟) / check_server.py +``` + +## 打包 EXE +```bat +python build_exe.py +``` +产物在 `dist/MaterialWMS.exe`。冻结后模板/静态走 `sys._MEIPASS`,库文件落 exe 同目录。 + +## 测试 +```bat +python tools/e2e_test.py # 覆盖 登录/收发核算/BOM/订单展开/领料/日志/报表/导入导出/入库·物料·BOM 批量导入窗口/角色拦截 +``` diff --git a/app.py b/app.py new file mode 100644 index 0000000..3fb8b8c --- /dev/null +++ b/app.py @@ -0,0 +1,987 @@ +"""物料入库/出库系统 - Flask 应用 (M1+M2+M3) + +M1 基础: 物料主数据 + 入库 + 出库 + 实时库存看板(预警标红) +M2 核算: BOM 清单 + 生产订单(接单量)按 BOM 展开领料需求 + Excel 双向导入导出 +M3 治理: 多角色(admin/manager/operator) + 操作日志 + 报表打印导出 + +技术: Flask + SQLAlchemy + SQLite (单文件库, 免外部数据库) +打包: PyInstaller --onefile, 冻结后模板/静态走 sys._MEIPASS, 库文件落 exe 同目录。 +""" +import os +import sys +import hashlib +import webbrowser +from datetime import date, datetime +from functools import wraps + +from flask import (Flask, render_template, request, redirect, url_for, + session, flash, g, send_file, jsonify) + +from models import (db, User, Material, Inbound, Outbound, + Bom, BomItem, ProdOrder, OperationLog) + +# ---------------- 路径(兼容 PyInstaller 冻结) ---------------- +if getattr(sys, "frozen", False): + BASE = sys._MEIPASS # 模板/静态解压目录(只读) + EXE_DIR = os.path.dirname(sys.executable) # exe 同目录(可写) +else: + BASE = os.path.dirname(os.path.abspath(__file__)) + EXE_DIR = BASE + +template_folder = os.path.join(BASE, "templates") +static_folder = os.path.join(BASE, "static") +db_path = os.environ.get("WMS_DB") or os.path.join(EXE_DIR, "app.db") + +app = Flask(__name__, template_folder=template_folder, static_folder=static_folder) +app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///" + db_path +app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False +app.config["SECRET_KEY"] = "material-wms-change-me" +db.init_app(app) + + +# ---------------- 鉴权 ---------------- +def sha256(s: str) -> str: + return hashlib.sha256(s.encode("utf-8")).hexdigest() + + +def login_required(f): + @wraps(f) + def wrapper(*a, **k): + if "user_id" not in session: + return redirect(url_for("login")) + return f(*a, **k) + return wrapper + + +def role_required(*roles): + def deco(f): + @wraps(f) + def wrapper(*a, **k): + if "user_id" not in session: + return redirect(url_for("login")) + u = g.user + if u is None or u.role not in roles: + flash("权限不足", "error") + return redirect(url_for("dashboard")) + return f(*a, **k) + return wrapper + return deco + + +@app.before_request +def load_user(): + uid = session.get("user_id") + g.user = db.session.get(User, uid) if uid else None + + +def log_action(action, target="", detail=""): + try: + db.session.add(OperationLog( + operator=(g.user.display_name or g.user.username) if g.user else "?", + action=action, target=str(target), detail=str(detail))) + db.session.commit() + except Exception: + db.session.rollback() + + +# ---------------- 库存核算 ---------------- +def stock_map(): + """返回 {material_id: 在库数量}""" + rows = db.session.execute( + db.text("SELECT material_id, COALESCE(SUM(qty),0) FROM inbound GROUP BY material_id") + ).fetchall() + inb = {r[0]: float(r[1]) for r in rows} + rows = db.session.execute( + db.text("SELECT material_id, COALESCE(SUM(qty),0) FROM outbound GROUP BY material_id") + ).fetchall() + out = {r[0]: float(r[1]) for r in rows} + ids = set(inb) | set(out) + return {i: inb.get(i, 0) - out.get(i, 0) for i in ids} + + +def inbound_total_map(): + """返回 {material_id: 入库总数量}""" + rows = db.session.execute( + db.text("SELECT material_id, COALESCE(SUM(qty),0) FROM inbound GROUP BY material_id") + ).fetchall() + return {r[0]: float(r[1]) for r in rows} + + +def outbound_total_map(): + """返回 {material_id: 消耗(出库)总数量}""" + rows = db.session.execute( + db.text("SELECT material_id, COALESCE(SUM(qty),0) FROM outbound GROUP BY material_id") + ).fetchall() + return {r[0]: float(r[1]) for r in rows} + + +def _date(s): + if s: + try: + return date.fromisoformat(s) + except ValueError: + return date.today() + return date.today() + + +def _validate_txn_rows(kind, rows): + """校验批量交易行, 返回 [{part_number, material_id, found, qty, batch_no, date, remark, ok, error}]""" + sm = stock_map() + out = [] + for r in rows: + pn = str(r.get("part_number", "") or "").strip() + qty = 0 + try: + qty = float(r.get("qty") or 0) + except (TypeError, ValueError): + qty = 0 + err = "" + mid = None + if not pn: + err = "缺少物料编号" + else: + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + if not m: + err = "物料不存在" + else: + mid = m.id + if not err and qty <= 0: + err = "数量必须大于 0" + if not err and kind == "outbound": + stock = sm.get(mid, 0) + if qty > stock: + err = f"库存不足(在库 {stock})" + out.append({ + "part_number": pn, "material_id": mid, "found": bool(mid), + "qty": qty, "batch_no": str(r.get("batch_no", "") or ""), + "date": str(r.get("date", "") or ""), "remark": str(r.get("remark", "") or ""), + "ok": not err, "error": err, + }) + return out + + +def _validate_material_rows(rows): + """校验物料批量行 -> [{part_number, manufacturer, unit, safety_stock, opening_stock, exists, ok, error}]""" + seen = set() + out = [] + for r in rows: + pn = str(r.get("part_number", "") or "").strip() + mf = str(r.get("manufacturer", "") or "").strip() + unit = str(r.get("unit", "") or "").strip() or "个" + err = "" + try: + ss = float(r.get("safety_stock") or 0) + except (TypeError, ValueError): + ss = 0 + try: + op = float(r.get("opening_stock") or 0) + except (TypeError, ValueError): + op = 0 + if not pn: + err = "缺少物料编号" + exists = False + if not err: + if pn in seen: + err = "本批重复编号" + else: + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + exists = bool(m) + seen.add(pn) + out.append({ + "part_number": pn, "manufacturer": mf, "unit": unit, + "safety_stock": ss, "opening_stock": op, + "exists": exists, "ok": not err, "error": err, + }) + return out + + +def _validate_bom_rows(boms): + """校验 BOM 批量 -> [{code, name, items:[{part_number, qty_per_unit, manufacturer, ok, error, auto_create}], ok, error, skip, valid_items}] + + 物料不存在时记为 auto_create=True(提交时会自动新建物料), 视为可导入。 + """ + out = [] + for b in boms: + code = str(b.get("code", "") or "").strip() + name = str(b.get("name", "") or "") + items = b.get("items", []) or [] + err = "" + if not code: + err = "缺少BOM型号" + item_rows = [] + skip = 0 + for it in items: + pn = str(it.get("part_number", "") or "").strip() + mf = str(it.get("manufacture", "") or "").strip() + try: + qty = float(it.get("qty_per_unit") or 0) + except (TypeError, ValueError): + qty = 0 + ierr = "" + auto = False + if not pn: + ierr = "缺物料编号" + else: + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + if not m: + auto = True + ierr = "物料不存在, 导入时自动新建" + elif qty <= 0: + ierr = "用量需>0" + ok = auto or (ierr == "") + if ierr and not auto: + skip += 1 + item_rows.append({"part_number": pn, "qty_per_unit": qty, "manufacturer": mf, + "ok": ok, "error": ierr, "auto_create": auto}) + bom_ok = (not err) and any(x["ok"] for x in item_rows) + out.append({ + "code": code, "name": name, "items": item_rows, + "ok": bom_ok, "error": err, "skip": skip, + "valid_items": sum(1 for x in item_rows if x["ok"]), + }) + return out + + +# ---------------- 页面 ---------------- +@app.route("/") +@login_required +def index(): + return redirect(url_for("dashboard")) + + +@app.route("/login", methods=["GET", "POST"]) +def login(): + if request.method == "POST": + u = db.session.scalar(db.select(User).where(User.username == request.form["username"])) + if u and u.active and u.password == sha256(request.form["password"]): + session["user_id"] = u.id + return redirect(url_for("dashboard")) + flash("用户名或密码错误", "error") + return render_template("login.html") + + +@app.route("/logout") +def logout(): + session.clear() + return redirect(url_for("login")) + + +@app.route("/dashboard") +@login_required +def dashboard(): + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + sm = stock_map() + rows = [] + negative = 0 + low = 0 + for m in mats: + stock = sm.get(m.id, 0) + if stock < 0: + negative += 1 + status = "danger" + elif stock < (m.safety_stock or 0): + low += 1 + status = "warn" + else: + status = "ok" + rows.append({"m": m, "stock": stock, "status": status}) + return render_template("dashboard.html", rows=rows, negative=negative, low=low) + + +# ---------------- 物料 ---------------- +@app.route("/materials") +@login_required +def materials(): + q = request.args.get("q", "").strip() + stmt = db.select(Material) + if q: + like = f"%{q}%" + stmt = stmt.where((Material.part_number.like(like)) | (Material.manufacturer.like(like))) + mats = db.session.scalars(stmt.order_by(Material.part_number)).all() + sm = stock_map() + inb_total = inbound_total_map() + out_total = outbound_total_map() + return render_template("materials.html", mats=mats, sm=sm, + inb_total=inb_total, out_total=out_total, q=q) + + +@app.route("/material/add", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def material_add(): + if request.method == "POST": + pn = request.form["part_number"].strip() + if not pn: + flash("物料编号必填", "error") + elif db.session.scalar(db.select(Material).where(Material.part_number == pn)): + flash("物料编号已存在", "error") + else: + m = Material( + part_number=pn, + manufacturer=request.form.get("manufacturer", ""), + unit=request.form.get("unit", "个") or "个", + safety_stock=float(request.form.get("safety_stock") or 0), + note=request.form.get("note", ""), + ) + db.session.add(m) + db.session.commit() + log_action("新增物料", pn, request.form.get("manufacturer", "")) + flash("物料已添加", "ok") + return redirect(url_for("materials")) + return render_template("material_form.html", m=None) + + +@app.route("/material/edit/", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def material_edit(mid): + m = db.session.get(Material, mid) + if not m: + return redirect(url_for("materials")) + if request.method == "POST": + m.manufacturer = request.form.get("manufacturer", "") + m.unit = request.form.get("unit", "个") or "个" + m.safety_stock = float(request.form.get("safety_stock") or 0) + m.note = request.form.get("note", "") + db.session.commit() + log_action("编辑物料", m.part_number) + flash("已保存", "ok") + return redirect(url_for("materials")) + return render_template("material_form.html", m=m) + + +@app.route("/material/delete/", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def material_delete(mid): + m = db.session.get(Material, mid) + if m: + db.session.delete(m) + db.session.commit() + log_action("删除物料", m.part_number) + flash("已删除", "ok") + return redirect(url_for("materials")) + + +@app.route("/material/") +@login_required +def material_detail(mid): + m = db.session.get(Material, mid) + if not m: + return redirect(url_for("materials")) + sm = stock_map() + inb_total = inbound_total_map() + out_total = outbound_total_map() + inbounds = db.session.scalars( + db.select(Inbound).where(Inbound.material_id == mid) + .order_by(Inbound.occur_date.desc(), Inbound.id.desc()) + ).all() + outbounds = db.session.scalars( + db.select(Outbound).where(Outbound.material_id == mid) + .order_by(Outbound.occur_date.desc(), Outbound.id.desc()) + ).all() + return render_template("material_detail.html", m=m, + stock=sm.get(m.id, 0), + inb_total=inb_total.get(m.id, 0), + out_total=out_total.get(m.id, 0), + inbounds=inbounds, outbounds=outbounds) + + +# ---------------- 入库 / 出库 ---------------- +@app.route("/inbound", methods=["GET", "POST"]) +@login_required +def inbound(): + if request.method == "POST": + mid = int(request.form["material_id"]) + qty = float(request.form["qty"]) + if qty <= 0: + flash("入库数量必须大于 0", "error") + else: + rec = Inbound( + material_id=mid, batch_no=request.form.get("batch_no", ""), + qty=qty, operator=request.form.get("operator", ""), + source=request.form.get("source", ""), + occur_date=_date(request.form.get("occur_date")), + remark=request.form.get("remark", "")) + db.session.add(rec) + db.session.commit() + log_action("入库", db.session.get(Material, mid).part_number, f"{qty}") + flash("入库成功", "ok") + return redirect(url_for("inbound")) + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + recs = db.session.scalars(db.select(Inbound).order_by(Inbound.id.desc()).limit(50)).all() + return render_template("txn.html", kind="inbound", mats=mats, recs=recs, today=date.today().isoformat()) + + +@app.route("/outbound", methods=["GET", "POST"]) +@login_required +def outbound(): + if request.method == "POST": + mid = int(request.form["material_id"]) + qty = float(request.form["qty"]) + if qty <= 0: + flash("出库数量必须大于 0", "error") + else: + stock = stock_map().get(mid, 0) + if qty > stock: + flash(f"库存不足! 当前在库 {stock}", "error") + else: + rec = Outbound( + material_id=mid, batch_no=request.form.get("batch_no", ""), + qty=qty, operator=request.form.get("operator", ""), + purpose=request.form.get("purpose", ""), + order_ref=request.form.get("order_ref", ""), + occur_date=_date(request.form.get("occur_date")), + remark=request.form.get("remark", "")) + db.session.add(rec) + db.session.commit() + log_action("出库", db.session.get(Material, mid).part_number, f"{qty}") + flash("出库成功", "ok") + return redirect(url_for("outbound")) + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + recs = db.session.scalars(db.select(Outbound).order_by(Outbound.id.desc()).limit(50)).all() + return render_template("txn.html", kind="outbound", mats=mats, recs=recs, today=date.today().isoformat()) + + +# ---------------- M2: BOM ---------------- +@app.route("/boms") +@login_required +@role_required("admin", "manager") +def boms(): + bs = db.session.scalars(db.select(Bom).order_by(Bom.code)).all() + data = [] + for b in bs: + items = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == b.id)).all() + data.append({"b": b, "count": len(items)}) + return render_template("boms.html", data=data) + + +@app.route("/bom/add", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def bom_add(): + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + if request.method == "POST": + code = request.form.get("code", "").strip() + if not code: + flash("BOM 型号必填", "error") + elif db.session.scalar(db.select(Bom).where(Bom.code == code)): + flash("BOM 型号已存在", "error") + else: + b = Bom(code=code, name=request.form.get("name", ""), note=request.form.get("note", "")) + db.session.add(b) + db.session.flush() + pns = request.form.getlist("pn") + qts = request.form.getlist("qty") + for pn, qt in zip(pns, qts): + pn = pn.strip() + if not pn: + continue + try: + q = float(qt) + except ValueError: + continue + if q <= 0: + continue + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + if m: + db.session.add(BomItem(bom_id=b.id, material_id=m.id, qty_per_unit=q)) + db.session.commit() + log_action("新增BOM", code) + flash("BOM 已保存", "ok") + return redirect(url_for("boms")) + return render_template("bom_form.html", b=None, mats=mats, items=[]) + + +@app.route("/bom/edit/", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def bom_edit(bid): + b = db.session.get(Bom, bid) + if not b: + return redirect(url_for("boms")) + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + if request.method == "POST": + b.name = request.form.get("name", "") + b.note = request.form.get("note", "") + db.session.execute(db.delete(BomItem).where(BomItem.bom_id == b.id)) + pns = request.form.getlist("pn") + qts = request.form.getlist("qty") + for pn, qt in zip(pns, qts): + pn = pn.strip() + if not pn: + continue + try: + q = float(qt) + except ValueError: + continue + if q <= 0: + continue + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + if m: + db.session.add(BomItem(bom_id=b.id, material_id=m.id, qty_per_unit=q)) + db.session.commit() + log_action("编辑BOM", b.code) + flash("已保存", "ok") + return redirect(url_for("boms")) + raw = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == b.id)).all() + items = [{"pn": db.session.get(Material, it.material_id).part_number, + "qty": it.qty_per_unit} for it in raw] + return render_template("bom_form.html", b=b, mats=mats, items=items) + + +@app.route("/bom/delete/", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def bom_delete(bid): + b = db.session.get(Bom, bid) + if b: + db.session.execute(db.delete(BomItem).where(BomItem.bom_id == b.id)) + db.session.delete(b) + db.session.commit() + log_action("删除BOM", b.code) + flash("已删除", "ok") + return redirect(url_for("boms")) + + +@app.route("/bom/") +@login_required +@role_required("admin", "manager") +def bom_detail(bid): + b = db.session.get(Bom, bid) + if not b: + return redirect(url_for("boms")) + items = db.session.scalars( + db.select(BomItem).where(BomItem.bom_id == b.id) + .order_by(BomItem.id) + ).all() + rows = [] + for it in items: + m = db.session.get(Material, it.material_id) + rows.append({"m": m, "qty": it.qty_per_unit}) + return render_template("bom_detail.html", b=b, rows=rows) + + +# ---------------- M2: 生产订单 ---------------- +@app.route("/orders") +@login_required +@role_required("admin", "manager") +def orders(): + os_ = db.session.scalars(db.select(ProdOrder).order_by(ProdOrder.id.desc())).all() + data = [] + for o in os_: + bom = db.session.get(Bom, o.bom_id) + data.append({"o": o, "bom": bom}) + return render_template("orders.html", data=data) + + +@app.route("/order/new", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def order_new(): + boms_all = db.session.scalars(db.select(Bom).order_by(Bom.code)).all() + if request.method == "POST": + bom_id = int(request.form["bom_id"]) + qty = float(request.form["qty"]) + if qty <= 0: + flash("接单量必须大于 0", "error") + else: + order_no = request.form.get("order_no", "").strip() or _gen_order_no() + if db.session.scalar(db.select(ProdOrder).where(ProdOrder.order_no == order_no)): + order_no = _gen_order_no() + o = ProdOrder(order_no=order_no, bom_id=bom_id, qty=qty, + status="planned", operator=(g.user.display_name or g.user.username)) + db.session.add(o) + db.session.commit() + log_action("新建生产订单", order_no, f"接单量 {qty}") + return redirect(url_for("order_detail", oid=o.id)) + return render_template("order_form.html", boms=boms_all, o=None) + + +@app.route("/order/") +@login_required +@role_required("admin", "manager") +def order_detail(oid): + o = db.session.get(ProdOrder, oid) + if not o: + return redirect(url_for("orders")) + bom = db.session.get(Bom, o.bom_id) + items = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == o.bom_id)).all() + sm = stock_map() + reqs = [] + for it in items: + m = db.session.get(Material, it.material_id) + need = it.qty_per_unit * o.qty + have = sm.get(m.id, 0) + reqs.append({"m": m, "need": need, "have": have, + "gap": need - have, "short": need > have}) + return render_template("order_detail.html", o=o, bom=bom, reqs=reqs) + + +@app.route("/order//pick", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def order_pick(oid): + o = db.session.get(ProdOrder, oid) + if not o or o.status == "picked": + return redirect(url_for("orders")) + items = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == o.bom_id)).all() + sm = stock_map() + picks = 0 + shorts = [] + for it in items: + m = db.session.get(Material, it.material_id) + need = it.qty_per_unit * o.qty + if need <= 0: + continue + db.session.add(Outbound(material_id=m.id, qty=need, + operator=(g.user.display_name or g.user.username), + purpose="生产领用", order_ref=o.order_no, + occur_date=date.today())) + picks += 1 + if need > sm.get(m.id, 0): + shorts.append(m.part_number) + o.status = "picked" + db.session.commit() + log_action("生产领料", o.order_no, f"生成出库 {picks} 项" + (f", 缺料: {','.join(shorts)}" if shorts else "")) + if shorts: + flash(f"已领料, 但缺料: {', '.join(shorts)}", "warn") + else: + flash("领料出库完成", "ok") + return redirect(url_for("order_detail", oid=oid)) + + +def _gen_order_no(): + d = date.today().strftime("%Y%m%d") + n = db.session.scalar(db.select(db.func.count(ProdOrder.id))) + 1 + return f"WO{d}{n:03d}" + + +# ---------------- M3: 报表 / 日志 / 用户 ---------------- +@app.route("/report") +@login_required +@role_required("admin", "manager") +def report(): + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + sm = stock_map() + rows = [] + for m in mats: + stock = sm.get(m.id, 0) + if stock < 0: + status = "danger" + elif stock < (m.safety_stock or 0): + status = "warn" + else: + status = "ok" + rows.append({"m": m, "stock": stock, "status": status}) + return render_template("report.html", rows=rows, today=date.today().isoformat()) + + +@app.route("/export/inventory") +@login_required +@role_required("admin", "manager") +def export_inventory(): + from excel_io import export_inventory + mats = db.session.scalars(db.select(Material).order_by(Material.part_number)).all() + sm = stock_map() + rows = [] + for m in mats: + stock = sm.get(m.id, 0) + status = "danger" if stock < 0 else ("warn" if stock < (m.safety_stock or 0) else "ok") + rows.append({"m": m, "stock": stock, "status": status}) + path = os.path.join(EXE_DIR, f"库存导出_{date.today().isoformat()}.xlsx") + export_inventory(path, rows) + return send_file(path, as_attachment=True, download_name=os.path.basename(path)) + + +@app.route("/logs") +@login_required +@role_required("admin", "manager") +def logs(): + recs = db.session.scalars(db.select(OperationLog).order_by(OperationLog.id.desc()).limit(200)).all() + return render_template("logs.html", recs=recs) + + +@app.route("/users") +@login_required +@role_required("admin") +def users(): + us = db.session.scalars(db.select(User).order_by(User.username)).all() + return render_template("users.html", us=us) + + +@app.route("/user/add", methods=["POST"]) +@login_required +@role_required("admin") +def user_add(): + un = request.form.get("username", "").strip() + pw = request.form.get("password", "") + role = request.form.get("role", "operator") + if not un or not pw: + flash("用户名和密码必填", "error") + elif db.session.scalar(db.select(User).where(User.username == un)): + flash("用户名已存在", "error") + else: + db.session.add(User(username=un, password=sha256(pw), role=role, + display_name=request.form.get("display_name", ""))) + db.session.commit() + log_action("新增用户", un, role) + flash("用户已添加", "ok") + return redirect(url_for("users")) + + +@app.route("/user/toggle/", methods=["POST"]) +@login_required +@role_required("admin") +def user_toggle(uid): + u = db.session.get(User, uid) + if u and u.id != g.user.id: + u.active = not u.active + db.session.commit() + log_action("启用/停用用户", u.username) + return redirect(url_for("users")) + + +# ---------------- Excel 导入 ---------------- +@app.route("/import/materials", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def import_materials(): + if request.method == "POST": + f = request.files.get("file") + if not f or not f.filename: + flash("请选择文件", "error") + else: + tmp = os.path.join(EXE_DIR, "_tmp_materials.xlsx") + f.save(tmp) + from excel_io import read_materials, import_materials as imp + rows = read_materials(tmp) + added, updated = imp(rows, operator=(g.user.display_name or g.user.username)) + os.remove(tmp) + log_action("导入物料", "Excel", f"新增 {added} / 更新 {updated}") + flash(f"导入完成: 新增 {added} 条, 更新 {updated} 条", "ok") + return redirect(url_for("materials")) + return render_template("import_form.html", kind="materials") + + +@app.route("/import/bom", methods=["GET", "POST"]) +@login_required +@role_required("admin", "manager") +def import_bom(): + if request.method == "POST": + f = request.files.get("file") + if not f or not f.filename: + flash("请选择文件", "error") + else: + tmp = os.path.join(EXE_DIR, "_tmp_bom.xlsx") + f.save(tmp) + from excel_io import read_bom, import_bom as impb + data = read_bom(tmp) + os.remove(tmp) + if not data: + flash("Excel 格式不符(需含 BOM型号/物料编号/单套用量 列)", "error") + else: + bid, ok, skip = impb(data, operator=(g.user.display_name or g.user.username)) + log_action("导入BOM", data["code"], f"成功 {ok} / 跳过 {skip}") + flash(f"BOM {data['code']} 导入: 成功 {ok} 项, 跳过(物料不存在) {skip} 项", "ok") + return redirect(url_for("boms")) + return render_template("import_form.html", kind="bom") + + +@app.route("/template/") +@login_required +@role_required("admin", "manager") +def download_template(kind): + from excel_io import build_material_template, build_bom_template, build_transaction_template + if kind == "materials": + p = os.path.join(EXE_DIR, "物料导入模板.xlsx") + build_material_template(p) + return send_file(p, as_attachment=True, download_name="物料导入模板.xlsx") + elif kind == "bom": + p = os.path.join(EXE_DIR, "BOM导入模板.xlsx") + build_bom_template(p) + return send_file(p, as_attachment=True, download_name="BOM导入模板.xlsx") + elif kind == "transactions": + p = os.path.join(EXE_DIR, "批量导入模板.xlsx") + build_transaction_template(p) + return send_file(p, as_attachment=True, download_name="批量导入模板.xlsx") + return redirect(url_for("dashboard")) + + +# ---------------- 交易批量导入窗口(入库/出库) ---------------- +@app.route("/api/import/transactions/preview", methods=["POST"]) +@login_required +def api_import_txn_preview(): + """接收粘贴的 JSON 行 或 上传文件, 返回解析+校验结果(供前端预览)""" + kind = (request.values.get("kind") or (request.get_json(silent=True) or {}).get("kind") or "inbound") + if request.files.get("file"): + tmp = os.path.join(EXE_DIR, "_tmp_txn.xlsx") + request.files["file"].save(tmp) + try: + from excel_io import read_transactions + rows = read_transactions(tmp) + finally: + if os.path.exists(tmp): + os.remove(tmp) + else: + payload = request.get_json(silent=True) or {} + rows = payload.get("rows", []) + result = _validate_txn_rows(kind, rows) + return jsonify({"kind": kind, "rows": result, + "valid": sum(1 for r in result if r["ok"]), + "invalid": sum(1 for r in result if not r["ok"])}) + + +@app.route("/api/import/transactions/commit", methods=["POST"]) +@login_required +def api_import_txn_commit(): + """提交校验通过的行, 批量写入入库/出库流水""" + payload = request.get_json(silent=True) or {} + kind = payload.get("kind", "inbound") + rows = payload.get("rows", []) + validated = _validate_txn_rows(kind, rows) + ok_rows = [v for v in validated if v["ok"]] + op_name = (g.user.display_name or g.user.username) if g.user else "?" + added = 0 + for v in ok_rows: + occur = _date(v["date"]) + if kind == "inbound": + rec = Inbound(material_id=v["material_id"], batch_no=v["batch_no"], qty=v["qty"], + operator=op_name, source="批量导入", occur_date=occur, remark=v["remark"]) + else: + rec = Outbound(material_id=v["material_id"], batch_no=v["batch_no"], qty=v["qty"], + operator=op_name, purpose="批量导入", occur_date=occur, remark=v["remark"]) + db.session.add(rec) + added += 1 + db.session.commit() + if added: + log_action("批量导入" + ("入库" if kind == "inbound" else "出库"), "窗口", + f"成功 {added} 条") + return jsonify({"added": added, + "skipped": len(validated) - added, + "errors": [r["error"] for r in validated if not r["ok"]]}) + + +# ---------------- 物料批量导入窗口 ---------------- +@app.route("/api/import/materials/preview", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def api_import_materials_preview(): + """接收粘贴文本 JSON 或上传文件, 返回物料解析+校验结果""" + if request.files.get("file"): + tmp = os.path.join(EXE_DIR, "_tmp_materials.xlsx") + request.files["file"].save(tmp) + try: + from excel_io import read_materials + rows = read_materials(tmp) + finally: + if os.path.exists(tmp): + os.remove(tmp) + else: + payload = request.get_json(silent=True) or {} + from excel_io import parse_materials_text + rows = parse_materials_text(payload.get("text", "")) + result = _validate_material_rows(rows) + return jsonify({"rows": result, + "valid": sum(1 for r in result if r["ok"]), + "invalid": sum(1 for r in result if not r["ok"])}) + + +@app.route("/api/import/materials/commit", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def api_import_materials_commit(): + """提交校验通过的物料行, 批量 upsert (含期初流水)""" + payload = request.get_json(silent=True) or {} + rows = payload.get("rows", []) + rows = [r for r in rows if r.get("ok", True)] # 只提交校验通过的行 + from excel_io import import_materials as imp_materials + op_name = (g.user.display_name or g.user.username) if g.user else "?" + added, updated = imp_materials(rows, operator=op_name) + if added or updated: + log_action("批量导入物料", "窗口", f"新增 {added} / 更新 {updated}") + return jsonify({"added": added, "updated": updated, + "skipped": len(rows) - added - updated}) + + +# ---------------- BOM 批量导入窗口 ---------------- +@app.route("/api/import/boms/preview", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def api_import_boms_preview(): + """接收粘贴文本 JSON 或上传文件, 返回多 BOM 解析+校验结果""" + if request.files.get("file"): + tmp = os.path.join(EXE_DIR, "_tmp_bom.xlsx") + request.files["file"].save(tmp) + try: + from excel_io import read_boms_multi + boms = read_boms_multi(tmp) + finally: + if os.path.exists(tmp): + os.remove(tmp) + else: + payload = request.get_json(silent=True) or {} + from excel_io import parse_boms_text + boms = parse_boms_text(payload.get("text", "")) + result = _validate_bom_rows(boms) + return jsonify({"boms": result, + "valid": sum(1 for b in result if b["ok"]), + "invalid": sum(1 for b in result if not b["ok"])}) + + +@app.route("/api/import/boms/commit", methods=["POST"]) +@login_required +@role_required("admin", "manager") +def api_import_boms_commit(): + """提交校验通过的 BOM(多个), 批量写入 BOM + BomItem""" + payload = request.get_json(silent=True) or {} + boms = payload.get("boms", []) + from excel_io import import_bom + op_name = (g.user.display_name or g.user.username) if g.user else "?" + created = 0 + total_ok = 0 + total_skip = 0 + total_created_mat = 0 + for b in boms: + code = str(b.get("code", "") or "").strip() + if not code: + continue + items = [{"part_number": it["part_number"], "qty_per_unit": it["qty_per_unit"], + "manufacture": it.get("manufacture", "")} + for it in b.get("items", []) if it.get("ok")] + if not items: + continue + _, ok, skip, created_mat = import_bom( + {"code": code, "name": b.get("name", ""), "items": items}, + operator=op_name, auto_create=True) + created += 1 + total_ok += ok + total_skip += skip + total_created_mat += created_mat + if created: + msg = f"{created}个BOM, 成功项 {total_ok}, 跳过 {total_skip}" + if total_created_mat: + msg += f", 自动新建物料 {total_created_mat}" + log_action("批量导入BOM", "窗口", msg) + return jsonify({"boms": created, "added_items": total_ok, + "skipped_items": total_skip, "created_materials": total_created_mat}) + + +# ---------------- 启动 ---------------- +def seed_admin(): + with app.app_context(): + db.create_all() + if not db.session.scalar(db.select(User).where(User.username == "admin")): + db.session.add(User(username="admin", password=sha256("admin123"), + role="admin", display_name="管理员")) + db.session.commit() + + +if __name__ == "__main__": + port = 8080 + if len(sys.argv) > 1 and sys.argv[1].isdigit(): + port = int(sys.argv[1]) + seed_admin() + if getattr(sys, "frozen", False): + webbrowser.open(f"http://127.0.0.1:{port}/") + app.run(host="127.0.0.1", port=port, debug=False) diff --git a/build_exe.py b/build_exe.py new file mode 100644 index 0000000..dcc33b6 --- /dev/null +++ b/build_exe.py @@ -0,0 +1,34 @@ +"""打包单文件 EXE (PyInstaller --onefile --windowed) + +冻结后: + - 模板/静态走 sys._MEIPASS (用 --add-data 打入) + - 库文件 app.db 落在 exe 同目录 (app.py 中 EXE_DIR 处理) + - 启动时自动打开浏览器 (app.py __main__ 中 webbrowser.open) + +用法: venv 的 python build_exe.py +""" +import os, sys, platform +from PyInstaller.__main__ import run + +HERE = os.path.dirname(os.path.abspath(__file__)) +sep = ";" if platform.system() == "Windows" else ":" + +args = [ + os.path.join(HERE, "app.py"), + "--onefile", + "--windowed", + "--name", "MaterialWMS", + "--distpath", os.path.join(HERE, "dist"), + "--workpath", os.path.join(HERE, "build"), + "--specpath", os.path.join(HERE, "build"), + "--add-data", os.path.join(HERE, "templates") + sep + "templates", + "--add-data", os.path.join(HERE, "static") + sep + "static", + "--hidden-import", "flask", + "--hidden-import", "flask_sqlalchemy", + "--hidden-import", "sqlalchemy", + "--hidden-import", "openpyxl", + "--noconfirm", +] +print("PyInstaller args:", args) +run(args) +print("BUILD DONE ->", os.path.join(HERE, "dist", "MaterialWMS.exe")) diff --git a/excel_io.py b/excel_io.py new file mode 100644 index 0000000..c3b6060 --- /dev/null +++ b/excel_io.py @@ -0,0 +1,473 @@ +"""Excel 双向导入/导出 (openpyxl) + +- 物料导入 / 导出 +- BOM 导入(含组成行) +- 交易(入库/出库)批量导入 +- 模板下载 +""" +import os +import re +import csv +from datetime import date +from openpyxl import Workbook, load_workbook +from openpyxl.styles import Font, PatternFill, Alignment + +from models import db, Material, Inbound, Bom, BomItem, OperationLog + + +# ---------------- 交易(入库/出库)批量导入 ---------------- +def _cell(r, i): + """安全取单元格, 越界返回 None""" + try: + return r[i] if i is not None else None + except (IndexError, TypeError): + return None + + +def read_transactions(path): + """Excel / CSV -> [{part_number, qty, batch_no, date, remark}] + + 表头识别: 物料编号 / 数量 / 批次 / 日期 / 备注 (名称大小写/中英均可)。 + """ + if str(path).lower().endswith(".csv"): + with open(path, newline="", encoding="utf-8-sig") as f: + rows = [r for r in csv.reader(f)] + else: + wb = load_workbook(path, data_only=True) + ws = wb.active + rows = [list(r) for r in ws.iter_rows(values_only=True)] + if not rows: + return [] + header = [str(c).strip() if c is not None else "" for c in rows[0]] + idx = {h: i for i, h in enumerate(header)} + + def col(*names): + for n in names: + if n in idx: + return idx[n] + return None + + c_pn = col("物料编号", "Part Number", "part_number", "编号") + c_qty = col("数量", "qty", "qty_per_unit", "用量") + c_batch = col("批次", "batch_no", "batch") + c_date = col("日期", "date", "occur_date", "业务日期") + c_remark = col("备注", "remark", "note") + out = [] + for r in rows[1:]: + if not any(c not in (None, "") for c in r): + continue + pn = str(_cell(r, c_pn) or "").strip() + if not pn: + continue + q = _f(_cell(r, c_qty)) + batch = str(_cell(r, c_batch) or "").strip() + d = str(_cell(r, c_date) or "").strip() + remark = str(_cell(r, c_remark) or "").strip() + out.append({"part_number": pn, "qty": q, "batch_no": batch, + "date": d, "remark": remark}) + return out + + +def build_transaction_template(path, kind="inbound"): + wb = Workbook() + ws = wb.active + ws.append(["物料编号", "数量", "批次", "日期", "备注"]) + for c in ws[1]: + c.font = Font(bold=True) + ws.append(["P0001", "100", "B20260911", "2026-09-11", "示例到货"]) + ws.append(["P0002", "50", "B20260911", "2026-09-11", "示例到货"]) + wb.save(path) + return path + + +# ---------------- 物料 ---------------- +def read_materials(path): + """返回 [{part_number, manufacturer, unit, safety_stock, opening_stock}]""" + wb = load_workbook(path, data_only=True) + ws = wb.active + rows = list(ws.iter_rows(values_only=True)) + if not rows: + return [] + header = [str(c).strip() if c is not None else "" for c in rows[0]] + idx = {h: i for i, h in enumerate(header)} + def col(*names): + for n in names: + if n in idx: + return idx[n] + return None + c_pn = col("物料编号", "Part Number", "part_number", "编号") + c_mf = col("型号", "Manufacture", "manufacturer", "规格") + c_unit = col("单位", "unit") + c_ss = col("安全库存", "safety_stock", "安全量") + c_op = col("期初库存", "opening_stock", "期初", "库存") + out = [] + for r in rows[1:]: + if c_pn is None or r[c_pn] in (None, ""): + continue + pn = str(r[c_pn]).strip() + if not pn: + continue + mf = str(r[c_mf]).strip() if (c_mf is not None and r[c_mf] is not None) else "" + unit = str(r[c_unit]).strip() if (c_unit is not None and r[c_unit] is not None) else "个" + ss = _f(r[c_ss]) if c_ss is not None else 0 + op = _f(r[c_op]) if c_op is not None else 0 + out.append({ + "part_number": pn, "manufacturer": mf, + "unit": unit or "个", "safety_stock": ss, "opening_stock": op, + }) + return out + + +def import_materials(rows, operator="导入"): + """upsert 物料; opening_stock>0 写入一条"期初"入库流水。返回 (新增, 更新)""" + added = 0 + updated = 0 + for d in rows: + m = db.session.scalar(db.select(Material).where(Material.part_number == d["part_number"])) + if m is None: + m = Material(part_number=d["part_number"], manufacturer=d.get("manufacturer", ""), + unit=d.get("unit", "个"), safety_stock=d.get("safety_stock", 0)) + db.session.add(m) + db.session.flush() + added += 1 + else: + m.manufacturer = d.get("manufacturer", m.manufacturer) + m.unit = d.get("unit", m.unit) or "个" + m.safety_stock = d.get("safety_stock", m.safety_stock) + updated += 1 + op = d.get("opening_stock", 0) or 0 + if op > 0: + exists = db.session.scalar( + db.select(Inbound).where(Inbound.material_id == m.id, Inbound.source == "期初")) + if exists: + exists.qty = op + else: + db.session.add(Inbound(material_id=m.id, qty=op, source="期初", + occur_date=date.today(), operator=operator)) + db.session.commit() + return added, updated + + +# ---------------- BOM ---------------- +def read_bom(path): + """返回 {code, name, items:[{part_number, qty_per_unit}]} 或 None(格式不符)""" + wb = load_workbook(path, data_only=True) + ws = wb.active + rows = list(ws.iter_rows(values_only=True)) + if not rows: + return None + header = [str(c).strip() if c is not None else "" for c in rows[0]] + idx = {h: i for i, h in enumerate(header)} + def col(*names): + for n in names: + if n in idx: + return idx[n] + return None + c_code = col("BOM型号", "code", "BOM") + c_name = col("名称", "name", "产品") + c_pn = col("物料编号", "Part Number", "part_number", "编号") + c_qty = col("单套用量", "qty_per_unit", "用量") + if c_code is None or c_pn is None or c_qty is None: + return None + code = str(rows[1][c_code]).strip() if rows[1][c_code] else "" + name = str(rows[1][c_name]).strip() if (c_name is not None and rows[1][c_name]) else "" + items = [] + for r in rows[1:]: + if r[c_pn] in (None, ""): + continue + pn = str(r[c_pn]).strip() + q = _f(r[c_qty]) + if q <= 0: + continue + items.append({"part_number": pn, "qty_per_unit": q}) + return {"code": code, "name": name, "items": items} + + +def import_bom(data, operator="导入", auto_create=False): + """写入 BOM + BomItem。 + + auto_create=True 时, 物料主数据不存在则按 Part Number + Manufacture 自动新建(便于真实 BOM 一步导入); + 否则物料不存在则该项被跳过。返回 (bom_id, 成功项, 跳过项, 自动新建物料数)。 + """ + if not data or not data["code"]: + return None, 0, 0, 0 + bom = db.session.scalar(db.select(Bom).where(Bom.code == data["code"])) + if bom is None: + bom = Bom(code=data["code"], name=data.get("name", "")) + db.session.add(bom) + db.session.flush() + else: + db.session.execute(db.delete(BomItem).where(BomItem.bom_id == bom.id)) + ok = 0 + skip = 0 + created_mat = 0 + for it in data["items"]: + pn = str(it["part_number"]).strip() + m = db.session.scalar(db.select(Material).where(Material.part_number == pn)) + if m is None: + if not auto_create: + skip += 1 + continue + m = Material(part_number=pn, + manufacturer=str(it.get("manufacture", "") or "").strip(), + unit="个") + db.session.add(m) + db.session.flush() + created_mat += 1 + db.session.add(BomItem(bom_id=bom.id, material_id=m.id, qty_per_unit=it["qty_per_unit"])) + ok += 1 + db.session.commit() + return bom.id, ok, skip, created_mat + + +# ---------------- 导出 ---------------- +def export_inventory(path, rows): + """rows: [{m, stock, status}]""" + wb = Workbook() + ws = wb.active + ws.title = "库存" + head = ["物料编号", "型号/规格", "单位", "在库数量", "安全库存", "状态"] + ws.append(head) + for c in ws[1]: + c.font = Font(bold=True) + c.fill = PatternFill("solid", fgColor="DDEBF7") + for r in rows: + m = r["m"] + status = {"ok": "正常", "warn": "低于安全库存", "danger": "库存为负"}[r["status"]] + ws.append([m.part_number, m.manufacturer, m.unit, r["stock"], + m.safety_stock or 0, status]) + if r["status"] == "danger": + ws.cell(row=ws.max_row, column=4).fill = PatternFill("solid", fgColor="F8CBAD") + elif r["status"] == "warn": + ws.cell(row=ws.max_row, column=4).fill = PatternFill("solid", fgColor="FFE699") + ws.freeze_panes = "A2" + wb.save(path) + + +def build_material_template(path): + wb = Workbook() + ws = wb.active + ws.append(["物料编号", "型号/规格", "单位", "安全库存", "期初库存"]) + for c in ws[1]: + c.font = Font(bold=True) + ws.append(["P0001", "示例型号A", "个", "100", "500"]) + ws.append(["P0002", "示例型号B", "卷", "50", "200"]) + wb.save(path) + return path + + +def build_bom_template(path): + """生成与真实 BOM 模板一致的格式(Altium/嘉立创导出): + 每个 worksheet 一个 BOM, sheet 名即 BOM 型号; 列头 Item/Part Number/ + Manufacture Part Number/Quantity/Reference/Part/Package。""" + wb = Workbook() + ws = wb.active + ws.title = "示例型号A" + ws.append(["Item", "Part Number", "Manufacture Part Number", + "Quantity", "Reference", "Part", "Package"]) + for c in ws[1]: + c.font = Font(bold=True) + c.fill = PatternFill("solid", fgColor="DDEBF7") + ws.append([1, "1001003", "CFW32C7UL_BUA", 1, "U1", "CFW32C7UL_BUA", "CFW32C7UL_IU_SOCKET"]) + ws.append([2, "1003002", "BY25Q32BSSIG", 1, "U2", "GD25Q127CSIG", "SOIC-8-S"]) + ws.append([3, "1007004", "SGM2036-2.5YUDH4G/TR", 1, "U11", "SGM2036-2.5YUDH4G/TR", ""]) + ws.freeze_panes = "A2" + wb.save(path) + return path + + +def _norm(h): + return str(h or "").strip().lower() + + +def _find_col(header, names): + """返回第一个精确匹配或包含匹配的列索引; 精确优先, 避免误命中(如 'part' 命中 'Part Number')""" + low = [_norm(h) for h in header] + # 1) 精确匹配 + for n in names: + nn = _norm(n) + for i, h in enumerate(low): + if h == nn: + return i + # 2) 包含匹配(子串) + for n in names: + nn = _norm(n) + if not nn: + continue + for i, h in enumerate(low): + if nn in h: + return i + return None + + +def _find_cols(header, names): + """返回所有匹配的列索引(用于重复表头, 如两个 'Quantity')""" + low = [_norm(h) for h in header] + out = [] + for n in names: + nn = _norm(n) + if not nn: + continue + for i, h in enumerate(low): + if h == nn or nn in h: + if i not in out: + out.append(i) + return out + + +def _f(v): + try: + return float(v) + except (TypeError, ValueError): + return 0 + + +# ---------------- 批量导入窗口: 文本解析 ---------------- +_ITEM_HEADER_RE = re.compile( + r"物料|编号|part\s?number|型号|manufacture|单位|unit|安全|safety|期初|库存|名称|bom|单套|用量|数量|qty", + re.I, +) + + +def parse_materials_text(text): + """粘贴多行 -> [{part_number, manufacturer, unit, safety_stock, opening_stock}] + + 每行格式: 物料编号 型号/规格 单位 安全库存 期初库存 + 列之间可用 空格/制表符/逗号/中文逗号 分隔; 首列命中表头关键词的行会被跳过。 + """ + out = [] + for raw in (text or "").split("\n"): + s = re.sub(r"[\t,，;；]+", " ", raw).strip() + if not s: + continue + parts = s.split() + if not parts: + continue + if _ITEM_HEADER_RE.search(parts[0]): + continue # 跳过表头 + pn = parts[0] + if not pn: + continue + mf = parts[1] if len(parts) > 1 else "" + unit = parts[2] if len(parts) > 2 else "个" + ss = _f(parts[3]) if len(parts) > 3 else 0 + op = _f(parts[4]) if len(parts) > 4 else 0 + out.append({ + "part_number": pn, + "manufacturer": mf, + "unit": unit or "个", + "safety_stock": ss, + "opening_stock": op, + }) + return out + + +def parse_boms_text(text): + """粘贴多行(以 #BOM 标记分段) -> [{code, name, items:[{part_number, qty_per_unit}]}] + + 格式: + #BOM M4021 示例产品A + P0001 2 + P0002 0.5 + + #BOM M4022 示例产品B + P0003 4 + """ + boms = [] + cur = None + for raw in (text or "").split("\n"): + s = re.sub(r"[\t,，;；]+", " ", raw).strip() + if not s: + continue + parts = s.split() + if not parts: + continue + is_head = s.startswith("#") or re.match(r"^(BOM|型号|MODEL)\b", parts[0], re.I) is not None + if is_head: + body = re.sub(r"^#+\s*", "", s, flags=re.I) + body = re.sub(r"^(BOM|型号|MODEL)\s*", "", body, flags=re.I).strip() + bp = body.split() + if not bp: + continue + if len(bp) > 1 and re.search(r"名称|物料|编号|单套|用量|数量", bp[1]): + continue # 表头行(如 "BOM型号 名称 物料编号 单套用量") + cur = {"code": bp[0], "name": " ".join(bp[1:]) if len(bp) > 1 else "", "items": []} + boms.append(cur) + continue + if _ITEM_HEADER_RE.search(parts[0]): + continue # 表头行 + if len(parts) < 2: + continue + if cur is None: + continue + cur["items"].append({"part_number": parts[0], "qty_per_unit": _f(parts[1])}) + return boms + + +def read_boms_multi(path): + """Excel/CSV -> [{code, name, items:[{part_number, qty_per_unit, manufacture}]}] + + 适配真实 BOM 模板(Altium / 嘉立创 EDA 导出): + - 每个 worksheet 视为一个 BOM, BOM 型号 = sheet 名 + - 电子料表: 表头含 'Part Number'(物料编号) + 'Quantity'(单套用量, 取第一个 Quantity 列) + - 组装表: 表头含 '物料代码'(物料编号) + '数量'(单套用量) + - 兼容旧模板: 若表头含 'BOM型号' 列, 则按该列分组(单 sheet 多 BOM) + 返回的每个 item 带 manufacture(型号/规格), 便于物料不存在时自动建物料。 + """ + if str(path).lower().endswith(".csv"): + with open(path, newline="", encoding="utf-8-sig") as f: + rows = [r for r in csv.reader(f)] + sheets = [("Sheet1", rows)] + else: + wb = load_workbook(path, data_only=True) + sheets = [(ws.title, list(ws.iter_rows(values_only=True))) for ws in wb.worksheets] + + result = [] + for sheet_name, rows in sheets: + if not rows or len(rows) < 2: + continue + header = [str(c).strip() if c is not None else "" for c in rows[0]] + if not any(header): + continue + pn_col = _find_col(header, ["part number", "partnumber", "物料编号", "物料代码"]) + qty_cols = _find_cols(header, ["quantity", "数量", "单套用量", "qty", "qty_per_unit"]) + mf_col = _find_col(header, ["manufacture part number", "manufacture", "型号", "part"]) + code_col = _find_col(header, ["bom型号", "code", "bom"]) + name_col = _find_col(header, ["名称", "name", "产品"]) + if pn_col is None or not qty_cols: + continue # 该 sheet 不是物料清单, 跳过(如说明页) + qty_col = qty_cols[0] # 取第一个 Quantity 列(嘉立创常出现两个 Quantity) + + items = [] + for r in rows[1:]: + if not any(c not in (None, "") for c in r): + continue + pn = str(r[pn_col]).strip() if pn_col < len(r) and r[pn_col] not in (None, "") else "" + if not pn: + continue + qty = _f(r[qty_col]) if qty_col < len(r) else 0 + # 嘉立创常出现两个 Quantity 列, 若首列空/0 则回退到次列, 避免丢料 + if qty <= 0 and len(qty_cols) > 1 and qty_cols[1] < len(r): + q2 = _f(r[qty_cols[1]]) + if q2 > 0: + qty = q2 + if qty <= 0: + continue + mf = "" + if mf_col is not None and mf_col < len(r) and r[mf_col] not in (None, ""): + mf = str(r[mf_col]).strip() + items.append({"part_number": pn, "qty_per_unit": qty, "manufacture": mf}) + if not items: + continue + + code = sheet_name.strip() + name = "" + if code_col is not None: # 兼容旧模板: 取首行 BOM型号/名称 + for r in rows[1:]: + if code_col < len(r) and r[code_col] not in (None, ""): + code = str(r[code_col]).strip() + if name_col is not None and name_col < len(r) and r[name_col] not in (None, ""): + name = str(r[name_col]).strip() + break + result.append({"code": code, "name": name, "items": items}) + return result diff --git a/init_db.py b/init_db.py new file mode 100644 index 0000000..69c3ca0 --- /dev/null +++ b/init_db.py @@ -0,0 +1,65 @@ +"""首次初始化: 建表 + 管理员 + 从桌面 Excel 导入真实物料(含期初库存) + +用法: 用 venv 的 python 运行 + python init_db.py +""" +import hashlib +from datetime import date +from app import app, db +from models import User, Material, Inbound + +EXCEL = r"C:/Users/wps18/Desktop/物料明细表.xlsx" + + +def main(): + with app.app_context(): + db.create_all() + + # 管理员 + if not db.session.scalar(db.select(User).where(User.username == "admin")): + db.session.add(User( + username="admin", + password=hashlib.sha256("admin123".encode("utf-8")).hexdigest(), + role="admin", + display_name="管理员", + )) + print("已创建管理员 admin / admin123") + + # 导入物料 + import openpyxl + wb = openpyxl.load_workbook(EXCEL, data_only=True) + ws = wb["物料入库"] + imported = 0 + skipped = 0 + for row in ws.iter_rows(min_row=2, values_only=True): + pn = row[1] # B 列 Part Number + if pn is None: + continue + pn = str(pn).strip() + if not pn: + continue + if db.session.scalar(db.select(Material).where(Material.part_number == pn)): + skipped += 1 + continue + man = str(row[2]) if row[2] else "" + m = Material(part_number=pn, manufacturer=man, unit="个", safety_stock=0) + db.session.add(m) + db.session.flush() + # 期初库存: 优先 E 列(库存数量), 其次 D 列(总数量) + init_qty = row[4] if row[4] not in (None, "") else row[3] + try: + init_qty = float(init_qty) + except (TypeError, ValueError): + init_qty = 0 + if init_qty > 0: + db.session.add(Inbound( + material_id=m.id, qty=init_qty, + source="期初", occur_date=date.today(), operator="系统", + )) + imported += 1 + db.session.commit() + print(f"导入物料 {imported} 条, 跳过已存在 {skipped} 条") + + +if __name__ == "__main__": + main() diff --git a/instance/app.db b/instance/app.db new file mode 100644 index 0000000..7dc18d7 Binary files /dev/null and b/instance/app.db differ diff --git a/models.py b/models.py new file mode 100644 index 0000000..ea5fe99 --- /dev/null +++ b/models.py @@ -0,0 +1,108 @@ +"""物料入库/出库系统 - 数据模型 (SQLAlchemy 2.0) + +实体: + User 账号(admin / manager / operator) + Material 物料主数据 + Inbound 入库流水(含"期初"来源的首笔) + Outbound 出库流水(含生产领用) + Bom BOM 清单(产品型号) + BomItem BOM 组成(某物料单套用量) + ProdOrder 生产订单(接单量) -> 按 BOM 展开领料需求 + OperationLog 操作日志(谁/何时/做了什么) + +库存口径: 在库 = SUM(入库) - SUM(出库), 由 app 实时聚合, 不冗余存储。 +""" +import os +from datetime import datetime, date +from flask_sqlalchemy import SQLAlchemy + +db = SQLAlchemy() + + +class User(db.Model): + __tablename__ = "user" + id = db.Column(db.Integer, primary_key=True) + username = db.Column(db.String(64), unique=True, nullable=False, index=True) + password = db.Column(db.String(128), nullable=False) # sha256 hex + role = db.Column(db.String(32), default="operator") # admin / manager / operator + display_name = db.Column(db.String(64), default="") + active = db.Column(db.Boolean, default=True) + + +class Material(db.Model): + __tablename__ = "material" + id = db.Column(db.Integer, primary_key=True) + part_number = db.Column(db.String(64), unique=True, nullable=False, index=True) + manufacturer = db.Column(db.String(255), default="") # 型号 / 规格 + unit = db.Column(db.String(16), default="个") + safety_stock = db.Column(db.Float, default=0) # 安全库存 + note = db.Column(db.String(255), default="") + created_at = db.Column(db.DateTime, default=datetime.now) + + +class Inbound(db.Model): + __tablename__ = "inbound" + id = db.Column(db.Integer, primary_key=True) + material_id = db.Column(db.Integer, db.ForeignKey("material.id"), nullable=False, index=True) + batch_no = db.Column(db.String(64), default="") # 批次号 + qty = db.Column(db.Float, nullable=False) # 入库数量(正) + operator = db.Column(db.String(64), default="") # 经手人 + source = db.Column(db.String(255), default="") # 来源 / 供应商 + occur_date = db.Column(db.Date, default=date.today) + remark = db.Column(db.String(255), default="") + created_at = db.Column(db.DateTime, default=datetime.now) + + +class Outbound(db.Model): + __tablename__ = "outbound" + id = db.Column(db.Integer, primary_key=True) + material_id = db.Column(db.Integer, db.ForeignKey("material.id"), nullable=False, index=True) + batch_no = db.Column(db.String(64), default="") # 批次号 + qty = db.Column(db.Float, nullable=False) # 出库数量(正) + operator = db.Column(db.String(64), default="") # 经手人 + purpose = db.Column(db.String(255), default="") # 用途 + order_ref = db.Column(db.String(64), default="") # 关联生产单号 + occur_date = db.Column(db.Date, default=date.today) + remark = db.Column(db.String(255), default="") + created_at = db.Column(db.DateTime, default=datetime.now) + + +# ---------------- M2: BOM 与生产订单 ---------------- +class Bom(db.Model): + __tablename__ = "bom" + id = db.Column(db.Integer, primary_key=True) + code = db.Column(db.String(64), unique=True, nullable=False, index=True) # BOM 型号简码 + name = db.Column(db.String(255), default="") + note = db.Column(db.String(255), default="") + created_at = db.Column(db.DateTime, default=datetime.now) + + +class BomItem(db.Model): + __tablename__ = "bom_item" + id = db.Column(db.Integer, primary_key=True) + bom_id = db.Column(db.Integer, db.ForeignKey("bom.id"), nullable=False, index=True) + material_id = db.Column(db.Integer, db.ForeignKey("material.id"), nullable=False, index=True) + qty_per_unit = db.Column(db.Float, nullable=False, default=0) # 单套用量 + + +class ProdOrder(db.Model): + __tablename__ = "prod_order" + id = db.Column(db.Integer, primary_key=True) + order_no = db.Column(db.String(64), unique=True, nullable=False, index=True) + bom_id = db.Column(db.Integer, db.ForeignKey("bom.id"), nullable=False, index=True) + qty = db.Column(db.Float, nullable=False, default=0) # 接单量(台/套) + status = db.Column(db.String(32), default="planned") # planned / picked / closed + operator = db.Column(db.String(64), default="") + remark = db.Column(db.String(255), default="") + created_at = db.Column(db.DateTime, default=datetime.now) + + +# ---------------- M3: 操作日志 ---------------- +class OperationLog(db.Model): + __tablename__ = "operation_log" + id = db.Column(db.Integer, primary_key=True) + operator = db.Column(db.String(64), default="") + action = db.Column(db.String(64), default="") # 动作分类 + target = db.Column(db.String(255), default="") # 对象 + detail = db.Column(db.String(512), default="") # 详情 + created_at = db.Column(db.DateTime, default=datetime.now) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..215eda6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +flask +flask-sqlalchemy +sqlalchemy +openpyxl +pyinstaller diff --git a/static/style.css b/static/style.css new file mode 100644 index 0000000..451bfcd --- /dev/null +++ b/static/style.css @@ -0,0 +1,90 @@ +* { box-sizing: border-box; } +body { + margin: 0; font-family: "Microsoft YaHei", "PingFang SC", system-ui, sans-serif; + background: #f4f6f8; color: #1f2733; font-size: 15px; +} +.topbar { + display: flex; align-items: center; justify-content: space-between; + background: #1f3a5f; color: #fff; padding: 0 20px; height: 52px; +} +.topbar .brand { font-weight: 700; font-size: 17px; } +.topbar .ver { background: #2e8b57; border-radius: 4px; padding: 1px 6px; font-size: 12px; margin-left: 6px; } +.topbar nav a { color: #dce6f2; text-decoration: none; margin-left: 18px; font-size: 14px; } +.topbar nav a:hover { color: #fff; } +.topbar .who { margin-left: 18px; color: #9fb3cc; font-size: 13px; } +main { max-width: 1100px; margin: 18px auto; padding: 0 16px; } + +.stats { display: flex; gap: 14px; margin-bottom: 16px; } +.stat { flex: 1; background: #fff; border-radius: 8px; padding: 16px; text-align: center; box-shadow: 0 1px 3px rgba(0,0,0,.08); } +.stat .num { font-size: 30px; font-weight: 700; } +.stat .lbl { color: #6b7785; font-size: 13px; margin-top: 4px; } +.stat.warn .num { color: #d48806; } +.stat.danger .num { color: #d4380d; } + +h3 { margin: 18px 0 10px; } +.card { background: #fff; border-radius: 8px; padding: 18px; box-shadow: 0 1px 3px rgba(0,0,0,.08); } +.login-wrap { display: flex; justify-content: center; margin-top: 60px; } +.login-card { width: 320px; } +.login-card h2 { margin-top: 0; } + +label { display: block; margin: 10px 0 4px; font-size: 13px; color: #44525f; } +input, select, textarea { + width: 100%; padding: 8px 10px; border: 1px solid #cdd5df; border-radius: 6px; font-size: 14px; +} +button, .btn { + display: inline-block; cursor: pointer; border: none; border-radius: 6px; + padding: 9px 16px; font-size: 14px; text-decoration: none; background: #eef1f5; color: #1f2733; +} +button.primary, .btn.primary { background: #1f3a5f; color: #fff; } +button:hover, .btn:hover { filter: brightness(.96); } +.row-actions { margin-top: 14px; display: flex; gap: 10px; } + +.toolbar { display: flex; justify-content: space-between; align-items: center; margin-bottom: 12px; } +.search input { width: 280px; } +.search button { margin-left: 8px; } + +.tbl { width: 100%; border-collapse: collapse; background: #fff; border-radius: 8px; overflow: hidden; box-shadow: 0 1px 3px rgba(0,0,0,.08); } +.tbl th, .tbl td { padding: 9px 12px; text-align: left; border-bottom: 1px solid #eef1f5; } +.tbl th { background: #f0f3f7; font-size: 13px; color: #44525f; } +.tbl td.num, .stat .num { text-align: right; font-variant-numeric: tabular-nums; } +.tbl a { color: #1f3a5f; text-decoration: none; } +.row-danger { background: #fff1f0; } +.row-warn { background: #fffbe6; } + +.badge { padding: 2px 8px; border-radius: 10px; font-size: 12px; } +.badge.ok { background: #e6f7ed; color: #1a7f47; } +.badge.warn { background: #fff7e6; color: #ad6800; } +.badge.danger { background: #fff1f0; color: #cf1322; } + +.flash { padding: 10px 14px; border-radius: 6px; margin-bottom: 12px; font-size: 14px; } +.flash.ok { background: #e6f7ed; color: #1a7f47; } +.flash.error { background: #fff1f0; color: #cf1322; } +.muted { color: #8a96a3; } +.small { font-size: 12px; } +.txn-form { max-width: 520px; margin-bottom: 22px; } + +/* 批量导入窗口(模态) */ +.modal-mask { + position: fixed; inset: 0; background: rgba(15,23,35,.5); + display: flex; align-items: center; justify-content: center; z-index: 50; +} +.modal { + background: #fff; border-radius: 10px; width: 720px; max-width: 94vw; + max-height: 90vh; display: flex; flex-direction: column; overflow: hidden; + box-shadow: 0 10px 40px rgba(0,0,0,.25); +} +.modal-head { + background: #1f3a5f; color: #fff; padding: 12px 16px; + display: flex; justify-content: space-between; align-items: center; font-weight: 600; +} +.modal-head .x { background: transparent; color: #fff; font-size: 18px; padding: 0 6px; } +.modal-body { padding: 16px; overflow: auto; } +.modal-body textarea { font-family: Consolas, "Courier New", monospace; font-size: 13px; } +.modal-body label.btn { margin: 0; } + +/* 批量导入窗口: 多 BOM 卡片 */ +.bom-card { border: 1px solid #e3e8ef; border-radius: 8px; margin-bottom: 10px; overflow: hidden; } +.bom-card.card-danger { border-color: #ffccc7; } +.bom-card .bom-head { background: #f0f3f7; padding: 8px 12px; font-size: 14px; display: flex; align-items: center; gap: 8px; } +.bom-card.card-danger .bom-head { background: #fff1f0; } +.bom-card .tbl { box-shadow: none; border-radius: 0; margin: 0; } diff --git a/templates/base.html b/templates/base.html new file mode 100644 index 0000000..7a1d44c --- /dev/null +++ b/templates/base.html @@ -0,0 +1,39 @@ + + + + + +{% block title %}物料入库系统{% endblock %} + + + +
+
📦 物料入库系统 M1+M2+M3
+ +
+
+ {% with msgs = get_flashed_messages(with_categories=true) %} + {% for cat, msg in msgs %} +
{{ msg }}
+ {% endfor %} + {% endwith %} + {% block content %}{% endblock %} +
+ + diff --git a/templates/bom_detail.html b/templates/bom_detail.html new file mode 100644 index 0000000..cdf8e88 --- /dev/null +++ b/templates/bom_detail.html @@ -0,0 +1,39 @@ +{% extends "base.html" %} +{% block title %}BOM 明细 {{ b.code }}{% endblock %} +{% block content %} +
+

BOM 明细 · {{ b.code }}

+ +
+ +
+ + + + + + +
BOM 型号{{ b.code }}名称{{ b.name or '-' }}
备注{{ b.note or '-' }}
+
+ +

组成物料（物料编号 / 规格型号 / 单套用量）

+ + + + {% for r in rows %} + + + + + + + + {% else %} + + {% endfor %} + +
#物料编号型号 / 规格单位单套用量
{{ loop.index }}{{ r.m.part_number }}{{ r.m.manufacturer or '-' }}{{ r.m.unit }}{{ '%.4g'|format(r.qty) }}
该 BOM 还没有组成物料
+{% endblock %} diff --git a/templates/bom_form.html b/templates/bom_form.html new file mode 100644 index 0000000..cdcd1e5 --- /dev/null +++ b/templates/bom_form.html @@ -0,0 +1,44 @@ +{% extends "base.html" %} +{% block title %}{{ '编辑 BOM' if b else '新增 BOM' }}{% endblock %} +{% block content %} +
+

{{ '编辑 BOM' if b else '新增 BOM' }}

+ + + + + + + +

组成物料 (单套用量)

+
+ {% for it in items %} +
+ + + +
+ {% else %} +
+ + + +
+ {% endfor %} +
+ +
+
+ + +{% endblock %} diff --git a/templates/boms.html b/templates/boms.html new file mode 100644 index 0000000..fb79748 --- /dev/null +++ b/templates/boms.html @@ -0,0 +1,122 @@ +{% extends "base.html" %} +{% block title %}BOM 清单{% endblock %} +{% block content %} +
+

BOM 清单 ({{ data|length }})

+
+ ⬇ 下载模板 + +
+
+ + + + {% for d in data %} + + + + + + + {% else %} + + {% endfor %} + +
BOM 型号名称组成项数操作
{{ d.b.code }}{{ d.b.name or '-' }}{{ d.count }} + 查看 + 编辑 +
+ +
+
还没有 BOM, 点右上角新增
+ + + + + +{% endblock %} diff --git a/templates/dashboard.html b/templates/dashboard.html new file mode 100644 index 0000000..aa14b44 --- /dev/null +++ b/templates/dashboard.html @@ -0,0 +1,34 @@ +{% extends "base.html" %} +{% block title %}库存看板{% endblock %} +{% block content %} +
+
{{ rows|length }}
物料种类
+
{{ low }}
低于安全库存
+
{{ negative }}
库存为负(缺料)
+
+ +

库存预警看板

+ + + + + + {% for r in rows %} + + + + + + + + + {% else %} + + {% endfor %} + +
物料编号型号 / 规格单位在库安全库存状态
{{ r.m.part_number }}{{ r.m.manufacturer or '-' }}{{ r.m.unit }}{{ '%.2f'|format(r.stock) }}{{ '%.2f'|format(r.m.safety_stock or 0) }} + {% if r.status=='danger' %}缺料 + {% elif r.status=='warn' %}偏低 + {% else %}正常{% endif %} +
暂无物料, 请先到"物料"页面添加
+{% endblock %} diff --git a/templates/import_form.html b/templates/import_form.html new file mode 100644 index 0000000..cb8598a --- /dev/null +++ b/templates/import_form.html @@ -0,0 +1,20 @@ +{% extends "base.html" %} +{% block title %}Excel 导入{% endblock %} +{% block content %} +
+

导入 {{ '物料' if kind=='materials' else 'BOM' }}

+
+ + +
+
+

+ {% if kind=='materials' %} + 表头需含: 物料编号 / 型号规格 / 单位 / 安全库存 / 期初库存 (后三列可选)。 + {% else %} + 表头需含: BOM型号 / 名称 / 物料编号 / 单套用量。 + {% endif %} +

+

⬇ 下载导入模板

+
+{% endblock %} diff --git a/templates/login.html b/templates/login.html new file mode 100644 index 0000000..e6403db --- /dev/null +++ b/templates/login.html @@ -0,0 +1,16 @@ +{% extends "base.html" %} +{% block title %}登录{% endblock %} +{% block content %} + +{% endblock %} diff --git a/templates/logs.html b/templates/logs.html new file mode 100644 index 0000000..6beb5a4 --- /dev/null +++ b/templates/logs.html @@ -0,0 +1,21 @@ +{% extends "base.html" %} +{% block title %}操作日志{% endblock %} +{% block content %} +

操作日志 (最近 200 条)

+ + + + {% for r in recs %} + + + + + + + + {% else %} + + {% endfor %} + +
时间操作人动作对象详情
{{ r.created_at.strftime('%Y-%m-%d %H:%M') }}{{ r.operator or '-' }}{{ r.action }}{{ r.target or '-' }}{{ r.detail or '-' }}
暂无日志
+{% endblock %} diff --git a/templates/material_detail.html b/templates/material_detail.html new file mode 100644 index 0000000..5f04668 --- /dev/null +++ b/templates/material_detail.html @@ -0,0 +1,69 @@ +{% extends "base.html" %} +{% block title %}物料明细 {{ m.part_number }}{% endblock %} +{% block content %} +
+

物料明细 · {{ m.part_number }}

+ +
+ +
+ + + + + + + + +
物料编号{{ m.part_number }}型号 / 规格{{ m.manufacturer or '-' }}
单位{{ m.unit }}安全库存{{ '%.2f'|format(m.safety_stock or 0) }}
备注{{ m.note or '-' }}
+
+ +
+
{{ '%.2f'|format(stock) }}
当前在库
+
{{ '%.2f'|format(inb_total) }}
入库总数量
+
{{ '%.2f'|format(out_total) }}
消耗总数量
+
+ +

入库流水（每次入库时间 / 数量）

+ + + + {% for r in inbounds %} + + + + + + + + + + {% else %} + + {% endfor %} + +
入库日期录入时间数量批次号来源 / 供应商经手人备注
{{ r.occur_date }}{{ r.created_at.strftime('%Y-%m-%d %H:%M') if r.created_at else '-' }}{{ '%.2f'|format(r.qty) }}{{ r.batch_no or '-' }}{{ r.source or '-' }}{{ r.operator or '-' }}{{ r.remark or '-' }}
暂无入库记录
+ +

出库 / 消耗流水

+ + + + {% for r in outbounds %} + + + + + + + + + + {% else %} + + {% endfor %} + +
出库日期录入时间数量用途关联单号经手人备注
{{ r.occur_date }}{{ r.created_at.strftime('%Y-%m-%d %H:%M') if r.created_at else '-' }}{{ '%.2f'|format(r.qty) }}{{ r.purpose or '-' }}{{ r.order_ref or '-' }}{{ r.operator or '-' }}{{ r.remark or '-' }}
暂无出库记录
+{% endblock %} diff --git a/templates/material_form.html b/templates/material_form.html new file mode 100644 index 0000000..c3e6ba3 --- /dev/null +++ b/templates/material_form.html @@ -0,0 +1,21 @@ +{% extends "base.html" %} +{% block title %}{{ '编辑' if m else '新增' }}物料{% endblock %} +{% block content %} +
+

{{ '编辑物料' if m else '新增物料' }}

+ + + + + + + + + + +
+ + 返回 +
+
+{% endblock %} diff --git a/templates/materials.html b/templates/materials.html new file mode 100644 index 0000000..fe471fa --- /dev/null +++ b/templates/materials.html @@ -0,0 +1,115 @@ +{% extends "base.html" %} +{% block title %}物料{% endblock %} +{% block content %} +
+
+ + +
+
+ ⬇ 下载模板 + +
+
+ + + + + + + {% for m in mats %} + + + + + + + + + + + {% else %} + + {% endfor %} + +
物料编号型号 / 规格单位安全库存在库入库总数量消耗总数量操作
{{ m.part_number }}{{ m.manufacturer or '-' }}{{ m.unit }}{{ '%.2f'|format(m.safety_stock or 0) }}{{ '%.2f'|format(sm.get(m.id, 0)) }}{{ '%.2f'|format(inb_total.get(m.id, 0)) }}{{ '%.2f'|format(out_total.get(m.id, 0)) }} + 明细 + 编辑 +
没有匹配的物料
+ + + + + +{% endblock %} diff --git a/templates/order_detail.html b/templates/order_detail.html new file mode 100644 index 0000000..2c39fd8 --- /dev/null +++ b/templates/order_detail.html @@ -0,0 +1,38 @@ +{% extends "base.html" %} +{% block title %}订单 {{ o.order_no }}{% endblock %} +{% block content %} +
+

生产订单 {{ o.order_no }}

+ {{ '已领料' if o.status=='picked' else '待领料' }} +
+
+

BOM: {{ bom.code if bom else '-' }} ({{ bom.name if bom else '' }}) ｜ 接单量: {{ '%.2f'|format(o.qty) }} ｜ 经手人: {{ o.operator or '-' }}

+
+ +

领料需求 (单套用量 × 接单量)

+ + + + {% for r in reqs %} + + + + + + + + + {% else %} + + {% endfor %} + +
物料编号型号需求数量当前库存缺口状态
{{ r.m.part_number }}{{ r.m.manufacturer or '-' }}{{ '%.2f'|format(r.need) }}{{ '%.2f'|format(r.have) }}{{ '%.2f'|format(r.gap) }}{{ '缺料' if r.short else '充足' }}
该 BOM 没有组成物料
+ +{% if o.status != 'picked' %} +
+
+
+{% else %} +

该订单已领料出库。

+{% endif %} +{% endblock %} diff --git a/templates/order_form.html b/templates/order_form.html new file mode 100644 index 0000000..b86be33 --- /dev/null +++ b/templates/order_form.html @@ -0,0 +1,20 @@ +{% extends "base.html" %} +{% block title %}新建生产订单{% endblock %} +{% block content %} +
+

新建生产订单

+ + + + + + +
+

生成后可在订单详情页按 BOM 展开领料需求, 并一键"确认领料"出库。

+
+{% endblock %} diff --git a/templates/orders.html b/templates/orders.html new file mode 100644 index 0000000..ac789a7 --- /dev/null +++ b/templates/orders.html @@ -0,0 +1,25 @@ +{% extends "base.html" %} +{% block title %}生产订单{% endblock %} +{% block content %} +
+

生产订单 ({{ data|length }})

+ + 新建订单 +
+ + + + {% for d in data %} + + + + + + + + + {% else %} + + {% endfor %} + +
订单号BOM接单量状态经手人操作
{{ d.o.order_no }}{{ d.bom.code if d.bom else '-' }} ({{ d.bom.name if d.bom else '' }}){{ '%.2f'|format(d.o.qty) }}{{ '已领料' if d.o.status=='picked' else '待领料' }}{{ d.o.operator or '-' }}查看
还没有生产订单
+{% endblock %} diff --git a/templates/report.html b/templates/report.html new file mode 100644 index 0000000..662e51b --- /dev/null +++ b/templates/report.html @@ -0,0 +1,31 @@ +{% extends "base.html" %} +{% block title %}库存报表{% endblock %} +{% block content %} +
+

库存报表 ({{ today }})

+
+ + ⬇ 导出 Excel +
+
+ + + + {% for r in rows %} + + + + + + + + + {% endfor %} + +
物料编号型号/规格单位在库数量安全库存状态
{{ r.m.part_number }}{{ r.m.manufacturer or '-' }}{{ r.m.unit }}{{ '%.2f'|format(r.stock) }}{{ '%.2f'|format(r.m.safety_stock or 0) }}{{ '库存为负' if r.status=='danger' else ('低于安全库存' if r.status=='warn' else '正常') }}
+ +{% endblock %} diff --git a/templates/txn.html b/templates/txn.html new file mode 100644 index 0000000..ee7eca5 --- /dev/null +++ b/templates/txn.html @@ -0,0 +1,167 @@ +{% extends "base.html" %} +{% block title %}{{ '入库登记' if kind=='inbound' else '出库登记' }}{% endblock %} +{% block content %} + +
+

{{ '入库登记' if kind=='inbound' else '出库登记' }}

+
+ ⬇ 下载批量模板 + +
+
+ +
+ + + + + + + + + {% if kind=='inbound' %} + + + {% else %} + + + + + {% endif %} + + + + +
+ +
+
+ +

最近 {{ recs|length }} 条{{ '入库' if kind=='inbound' else '出库' }}记录

+ + + + + {% if kind=='inbound' %}{% else %}{% endif %} + + + + + {% for r in recs %} + {% set m = mats|selectattr('id','equalto',r.material_id)|list|first %} + + + + + + {% if kind=='inbound' %} + {% else %}{% endif %} + + + {% else %} + + {% endfor %} + +
物料批次数量经手人来源用途单号日期
{{ m.part_number if m else r.material_id }}{{ r.batch_no or '-' }}{{ '%.2f'|format(r.qty) }}{{ r.operator or '-' }}{{ r.source or '-' }}{{ r.purpose or '-' }}{{ r.order_ref or '-' }}{{ r.occur_date }}
暂无记录
+ + + + + +{% endblock %} diff --git a/templates/users.html b/templates/users.html new file mode 100644 index 0000000..4f4c9e3 --- /dev/null +++ b/templates/users.html @@ -0,0 +1,42 @@ +{% extends "base.html" %} +{% block title %}用户管理{% endblock %} +{% block content %} +

用户列表

+ + + + {% for u in us %} + + + + + + + + {% endfor %} + +
用户名姓名角色状态操作
{{ u.username }}{{ u.display_name or '-' }}{{ u.role }}{{ '启用' if u.active else '停用' }} + {% if u.id != g.user.id %} +
+ +
+ {% else %}当前账号{% endif %} +
+ +

新增用户

+
+ + + + + + + + +
+
+{% endblock %} diff --git a/tools/_exe_util.py b/tools/_exe_util.py new file mode 100644 index 0000000..bc35b71 --- /dev/null +++ b/tools/_exe_util.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +"""EXE 启动 / 清理公共工具。 + +背景(踩坑记录): + PyInstaller `--onefile --windowed` 的启动器(runw.exe)会**派生子进程**运行真正的 + Flask 服务器。若只 `p.terminate()` 父进程,子进程会残留并继续霸占 8080； + Windows 允许同一端口被多个 socket 绑定,于是后续请求会被**随机路由到旧实例**, + 表现为: 随机 500、改了模板/代码"不生效"、时好时坏的返回内容。 + +正确做法: + 1) 启动前按可执行文件路径清掉 dist 下所有 EXE(兼容 *.old.exe / _prev_*.exe 等改名备份); + 2) 收尾用 `taskkill /F /T /PID` 杀掉**整棵进程树**,再兜底按路径清一次。 +""" +import os +import subprocess +import time + +EXE = r"D:\2026-09-10-task-20\material_wms\dist\MaterialWMS.exe" + + +def kill_dist_exes(exe_path=EXE): + """杀掉 exe_path 所在目录下的所有 EXE 进程(按路径匹配任意文件名)。""" + dist = os.path.dirname(exe_path) + ps = ( + "$d = '%s'; " + "Get-CimInstance Win32_Process | " + "Where-Object { $_.ExecutablePath -like ($d + '\\*.exe') } | " + "ForEach-Object { Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue }" + ) % dist + try: + subprocess.run(["powershell", "-NoProfile", "-Command", ps], + capture_output=True, timeout=30) + except Exception as e: + print(" [warn] kill_dist_exes:", e) + + +def start_exe(exe_path=EXE): + """清残留 -> 启动 EXE -> 返回 Popen(用完务必 stop_exe)。""" + kill_dist_exes(exe_path) + time.sleep(1) + return subprocess.Popen([exe_path]) + + +def stop_exe(p): + """杀掉整棵进程树并兜底清理(幂等, 传入 None 安全)。""" + if p is not None: + try: + subprocess.run(["taskkill", "/F", "/T", "/PID", str(p.pid)], + capture_output=True, timeout=15) + except Exception: + pass + try: + p.wait(timeout=5) + except Exception: + try: + p.kill() + except Exception: + pass + kill_dist_exes() diff --git a/tools/check_batch_exe.py b/tools/check_batch_exe.py new file mode 100644 index 0000000..e403ad3 --- /dev/null +++ b/tools/check_batch_exe.py @@ -0,0 +1,89 @@ +"""验证打包后的 EXE 包含批量导入窗口功能并端到端可用。""" +import sqlite3, time, subprocess, sys, urllib.request, urllib.parse, json, os +import atexit + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _exe_util import start_exe, stop_exe + +EXE = r"D:\2026-09-10-task-20\material_wms\dist\MaterialWMS.exe" +BASE = "http://127.0.0.1:8080" + +# 取一条真实物料编号 +conn = sqlite3.connect(r"D:\2026-09-10-task-20\material_wms\dist\app.db") +pn = conn.execute("select part_number from material limit 1").fetchone()[0] +conn.close() +print("使用物料编号:", pn) + +print("启动 EXE ...") +p = start_exe(EXE) +atexit.register(stop_exe, p) +for _ in range(40): + try: + urllib.request.urlopen(BASE + "/login", timeout=2) + break + except Exception: + time.sleep(0.5) +else: + print("ERROR: 服务未启动"); sys.exit(1) +print("服务已就绪") + +def post(path, data=None, raw=None, headers=None): + if raw is not None: + req = urllib.request.Request(BASE + path, data=raw, headers=headers or {}, method="POST") + else: + req = urllib.request.Request(BASE + path, data=urllib.parse.urlencode(data).encode(), method="POST") + try: + return urllib.request.urlopen(req, timeout=5) + except urllib.error.HTTPError as e: + return e + +# 登录(保持 cookie) +jar = {} +opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor()) +def get(path): + return opener.open(BASE + path, timeout=5) +def postj(path, obj): + req = urllib.request.Request(BASE + path, data=json.dumps(obj).encode(), + headers={"Content-Type": "application/json"}, method="POST") + return opener.open(req, timeout=5) + +# 登录 +login = urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode() +opener.open(urllib.request.Request(BASE + "/login", data=login, method="POST")) + +# 1. 入库页含批量导入窗口按钮 +html = get("/inbound").read().decode("utf-8", "ignore") +assert "批量导入窗口" in html, "入库页缺少批量导入按钮" +print("[OK] 入库页含「批量导入窗口」按钮") + +# 2. 下载批量模板 +t = get("/template/transactions") +assert t.status == 200, "模板下载失败" +print("[OK] 批量导入模板下载 (HTTP %d)" % t.status) + +# 3. 预览: 1 条有效(真实物料) + 1 条异常(物料不存在) +prev = json.loads(postj("/api/import/transactions/preview", { + "kind": "inbound", + "rows": [ + {"part_number": pn, "qty": 25, "batch_no": "EXEBATCH", "date": "2026-09-11", "remark": "exe验证"}, + {"part_number": "NO_SUCH", "qty": 1}, + ], +}).read()) +assert prev["valid"] == 1 and prev["invalid"] == 1, prev +print("[OK] 预览接口: 有效 %d / 异常 %d" % (prev["valid"], prev["invalid"])) + +# 4. 提交: 应写入 1 条入库(另一条跳过) +res = json.loads(postj("/api/import/transactions/commit", { + "kind": "inbound", "rows": prev["rows"]}).read()) +assert res["added"] == 1 and res["skipped"] == 1, res +print("[OK] 提交接口: 入库 %d 条 / 跳过 %d 条" % (res["added"], res["skipped"])) + +# 5. 确认库里确有这条 EXEBATCH 入库 +conn = sqlite3.connect(r"D:\2026-09-10-task-20\material_wms\dist\app.db") +cnt = conn.execute("select count(*) from inbound where batch_no='EXEBATCH'").fetchone()[0] +conn.close() +assert cnt == 1, "入库流水未写入" +print("[OK] 数据库确认: EXEBATCH 入库流水已生成") + +stop_exe(p) +print("\nEXE 批量导入窗口 端到端验证通过 ✅") diff --git a/tools/check_bom_exe.py b/tools/check_bom_exe.py new file mode 100644 index 0000000..20f8354 --- /dev/null +++ b/tools/check_bom_exe.py @@ -0,0 +1,158 @@ +# -*- 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() diff --git a/tools/check_exe.py b/tools/check_exe.py new file mode 100644 index 0000000..51f3ca9 --- /dev/null +++ b/tools/check_exe.py @@ -0,0 +1,65 @@ +"""验证打包后的 EXE: 等端口 -> 登录 -> 访问关键页面 -> Excel 导出 +用法: 先手动启动 dist/MaterialWMS.exe, 再 python tools/check_exe.py +""" +import sys, time, urllib.request, urllib.parse + +BASE = "http://127.0.0.1:8080" + +def wait_port(t=40): + for i in range(t): + try: + urllib.request.urlopen(BASE + "/login", timeout=2) + return True + except Exception: + time.sleep(1) + return False + +def get(path, cookie=None): + req = urllib.request.Request(BASE + path) + if cookie: + req.add_header("Cookie", cookie) + return urllib.request.urlopen(req, timeout=5) + +def post(path, data, cookie=None): + req = urllib.request.Request(BASE + path, data=urllib.parse.urlencode(data).encode(), method="POST") + if cookie: + req.add_header("Cookie", cookie) + return urllib.request.urlopen(req, timeout=5) + +def main(): + if not wait_port(): + print("[FAIL] 端口 8080 未就绪 (EXE 可能未启动或解压慢)") + sys.exit(1) + print("[OK] EXE 服务已就绪") + + # 登录页 + r = get("/login") + assert "登录" in r.read().decode("utf-8", "ignore") + print("[OK] 登录页(冻结模板加载正常)") + + # 登录 + opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor()) + req = urllib.request.Request(BASE + "/login", + data=urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode(), + method="POST") + resp = opener.open(req) + body = resp.read().decode("utf-8", "ignore") + assert "库存看板" in body, "登录后未进入看板" + print("[OK] 管理员登录成功, 看板渲染正常") + + # 关键页面(冻结模板 + 路由) + for p in ["/boms", "/orders", "/report", "/logs", "/users", "/import/materials", "/template/bom"]: + rr = opener.open(BASE + p) + assert rr.status == 200 + print("[OK] BOM/订单/报表/日志/用户/导入/模板 页面均可访问") + + # 导出 Excel + rr = opener.open(BASE + "/export/inventory") + data = rr.read() + assert rr.status == 200 and len(data) > 1000 + print(f"[OK] 库存 Excel 导出成功 ({len(data)} 字节)") + + print("\nEXE 端到端验证通过 ✅") + +if __name__ == "__main__": + main() diff --git a/tools/check_materials_bom_exe.py b/tools/check_materials_bom_exe.py new file mode 100644 index 0000000..f0ce5ce --- /dev/null +++ b/tools/check_materials_bom_exe.py @@ -0,0 +1,104 @@ +"""EXE 端到端验证: 物料 / BOM 批量导入窗口 + +用 urllib + cookiejar 维持会话, 对 http://127.0.0.1:8080 实测: + 1) /materials 含"批量导入窗口"按钮 + 2) 物料预览(粘贴) -> 校验 -> 提交 + 3) /boms 含"批量导入窗口"按钮 + 4) BOM 多 BOM 预览 -> 校验 -> 提交 + 5) 数据库确认新物料 / 新 BOM 已落库 +""" +import sys, time, json, urllib.request, urllib.parse, http.cookiejar + +BASE = "http://127.0.0.1:8080" + +def wait_port(timeout=30): + end = time.time() + timeout + while time.time() < end: + try: + urllib.request.urlopen(BASE + "/login", timeout=2) + return True + except Exception: + time.sleep(0.5) + return False + +def post(cj, url, data=None, is_json=False, headers=None): + req = urllib.request.Request(url) + if is_json: + req.add_header("Content-Type", "application/json") + body = json.dumps(data).encode("utf-8") + elif data is not None: + req.add_header("Content-Type", "application/x-www-form-urlencoded") + body = urllib.parse.urlencode(data).encode("utf-8") + else: + body = None + if headers: + for k, v in headers.items(): + req.add_header(k, v) + opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) + return opener.open(req, data=body, timeout=10) + +def main(): + cj = http.cookiejar.CookieJar() + assert wait_port(), "服务端口未就绪" + print("[OK] 服务已启动可访问") + + # 登录 + r = post(cj, BASE + "/login", {"username": "admin", "password": "admin123"}) + assert "库存看板" in r.read().decode("utf-8", "ignore") or r.status == 200 + print("[OK] 登录成功") + + # 物料页含批量按钮 + r = post(cj, BASE + "/materials") + assert "批量导入窗口" in r.read().decode("utf-8", "ignore") + print("[OK] 物料页含批量导入窗口按钮") + + # 物料预览(粘贴 2 条新 + 1 条重复) + text = "MTEST1 批量料A 个 10 200\nMTEST2 批量料B 卷 5 50\nMTEST1 重复料 个 1 1" + r = post(cj, BASE + "/api/import/materials/preview", + {"text": text}, is_json=True) + mp = json.loads(r.read()) + assert mp["valid"] == 2 and mp["invalid"] == 1, mp + print(f"[OK] 物料预览: 可导入 {mp['valid']} / 异常 {mp['invalid']}") + + r = post(cj, BASE + "/api/import/materials/commit", + {"rows": mp["rows"]}, is_json=True) + mc = json.loads(r.read()) + assert mc["added"] == 2, mc + print(f"[OK] 物料提交: 新增 {mc['added']} 条") + + # BOM 页含批量按钮 + r = post(cj, BASE + "/boms") + assert "批量导入窗口" in r.read().decode("utf-8", "ignore") + print("[OK] BOM 页含批量导入窗口按钮") + + # BOM 预览(2 个 BOM, 其中一个含不存在物料) + btext = "#BOM PROMOTE1 推广品A\nMTEST1 2\nMTEST2 0.5\nGHOST 3\n#BOM PROMOTE2 推广品B\nMTEST1 4" + r = post(cj, BASE + "/api/import/boms/preview", {"text": btext}, is_json=True) + bp = json.loads(r.read()) + assert bp["valid"] == 2 and bp["invalid"] == 0, bp + print(f"[OK] BOM 预览: 可导入 {bp['valid']} 个 BOM") + + r = post(cj, BASE + "/api/import/boms/commit", {"boms": bp["boms"]}, is_json=True) + bc = json.loads(r.read()) + assert bc["boms"] == 2 and bc["added_items"] == 3, bc + print(f"[OK] BOM 提交: {bc['boms']} 个 BOM / 成功项 {bc['added_items']}") + + # 数据库确认落库 + import sqlite3 + dbp = r"D:\2026-09-10-task-20\material_wms\dist\app.db" + c = sqlite3.connect(dbp) + mats = c.execute("select count(*) from material where part_number in ('MTEST1','MTEST2')").fetchone()[0] + boms = c.execute("select count(*) from bom where code in ('PROMOTE1','PROMOTE2')").fetchone()[0] + c.close() + assert mats == 2, f"物料落库数={mats}" + assert boms == 2, f"BOM落库数={boms}" + print(f"[OK] 数据库确认: 新物料 {mats} 条 / 新 BOM {boms} 个 已落库") + + print("\nEXE 物料/BOM 批量导入窗口 端到端验证通过 ✅") + +if __name__ == "__main__": + try: + main() + except Exception as e: + print("验证失败:", e) + sys.exit(1) diff --git a/tools/check_server.py b/tools/check_server.py new file mode 100644 index 0000000..e9b8d72 --- /dev/null +++ b/tools/check_server.py @@ -0,0 +1,43 @@ +"""真实 HTTP 启动检查: 等待端口 -> 登录 -> 看板渲染""" +import urllib.request, urllib.parse, http.cookiejar, time, sys + +BASE = "http://127.0.0.1:8080" +cj = http.cookiejar.CookieJar() +op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) + + +def get(path): + return op.open(BASE + path, timeout=5).read().decode("utf-8", "ignore") + + +# 等待服务起来 +for i in range(20): + try: + html = get("/login") + if "物料入库系统" in html: + print("[OK] 服务已启动, 登录页正常") + break + except Exception: + time.sleep(0.5) +else: + print("[FAIL] 服务未在 10s 内就绪") + sys.exit(1) + +# 登录 +data = urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode() +op.open(urllib.request.Request(BASE + "/login", data=data, method="POST")) +html = get("/dashboard") +assert "库存看板" in html, "看板未渲染" +print("[OK] 登录后看板渲染正常") + +# 物料页 +html = get("/materials") +assert "物料" in html +print("[OK] 物料页渲染正常") + +# 入库页 +html = get("/inbound") +assert "入库登记" in html +print("[OK] 入库页渲染正常") + +print("\n真实服务冒烟通过 ✅") diff --git a/tools/e2e_test.py b/tools/e2e_test.py new file mode 100644 index 0000000..a00b927 --- /dev/null +++ b/tools/e2e_test.py @@ -0,0 +1,228 @@ +"""M1+M2+M3 冒烟测试: 登录 -> 看板 -> 入库出库 -> BOM -> 订单展开 -> 领料 -> 日志 -> 导出 -> 导入 -> 用户 + +用 Flask test_client; 通过 WMS_DB 指向临时库, 不污染真实数据。 +""" +import sys, os, tempfile +tmp = os.path.join(tempfile.gettempdir(), "wms_smoke.db") +if os.path.exists(tmp): + os.remove(tmp) +os.environ["WMS_DB"] = tmp + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from app import app, db, stock_map, seed_admin +from models import (Material, Inbound, Outbound, Bom, BomItem, ProdOrder, OperationLog, User) + +with app.app_context(): + db.create_all() + seed_admin() + +c = app.test_client() + +# 1. 未登录跳登录 +assert c.get("/dashboard").status_code == 302 +print("[OK] 未登录跳登录") + +# 2. 登录 +r = c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True) +assert r.status_code == 200 and "库存看板" in r.text +print("[OK] 登录成功") + +# 3. 物料收发核算 +with app.app_context(): + m = db.session.scalars(db.select(Material).limit(1)).first() + if not m: # 临时库无数据, 造一条 + m = Material(part_number="T001", manufacturer="测试料", unit="个") + db.session.add(m); db.session.flush() + mid = m.id + before = stock_map().get(mid, 0) + c.post("/inbound", data={"material_id": mid, "qty": 100}, follow_redirects=True) + assert abs(stock_map().get(mid, 0) - (before + 100)) < 1e-6 + c.post("/outbound", data={"material_id": mid, "qty": 30}, follow_redirects=True) + assert abs(stock_map().get(mid, 0) - (before + 70)) < 1e-6 + rr = c.post("/outbound", data={"material_id": mid, "qty": 999999}, follow_redirects=True) + assert "库存不足" in rr.text + print(f"[OK] 库存核算: 期初 {before} -> +100 -> -30 -> {stock_map().get(mid,0)}; 超额拦截") + +# 4. BOM 新增 +r = c.post("/bom/add", data={"code": "BOMTEST", "name": "测试产品", + "pn": ["T001"], "qty": ["2"]}, follow_redirects=True) +assert r.status_code == 200 +with app.app_context(): + b = db.session.scalar(db.select(Bom).where(Bom.code == "BOMTEST")) + assert b and db.session.scalar(db.select(BomItem).where(BomItem.bom_id == b.id)) +print("[OK] BOM 新增 + 组成项") + +# 5. 生产订单 + 展开 +r = c.post("/order/new", data={"bom_id": b.id, "qty": 10}, follow_redirects=True) +assert r.status_code == 200 and "领料需求" in r.text +with app.app_context(): + o = db.session.scalar(db.select(ProdOrder).where(ProdOrder.bom_id == b.id)) + assert o +print("[OK] 生产订单生成 + 需求展开 (单套2 × 10 = 20)") + +# 6. 确认领料 +c.post(f"/order/{o.id}/pick", data={}, follow_redirects=True) +with app.app_context(): + assert db.session.scalar(db.select(ProdOrder).where(ProdOrder.id == o.id)).status == "picked" + pick = db.session.scalar(db.select(Outbound).where(Outbound.order_ref == o.order_no)) + assert pick and abs(pick.qty - 20) < 1e-6 +print("[OK] 领料出库生成 (20)") + +# 7. 操作日志 +r = c.get("/logs") +assert "操作日志" in r.text and "领料" in r.text +print("[OK] 操作日志已记录") + +# 8. 报表 + 导出 Excel +assert "库存报表" in c.get("/report").text +r = c.get("/export/inventory") +assert r.status_code == 200 and r.mimetype in ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/octet-stream") +print("[OK] 报表页 + Excel 导出") + +# 9. 模板下载 +assert c.get("/template/materials").status_code == 200 +assert c.get("/template/bom").status_code == 200 +print("[OK] 导入模板下载") + +# 10. Excel 导入物料 +import openpyxl +from excel_io import read_materials, import_materials +wb = openpyxl.Workbook(); ws = wb.active +ws.append(["物料编号", "型号/规格", "单位", "安全库存", "期初库存"]) +ws.append(["IMP001", "导入料A", "个", "10", "100"]) +tp = os.path.join(tempfile.gettempdir(), "imp_m.xlsx"); wb.save(tp) +rows = read_materials(tp) +assert rows and rows[0]["part_number"] == "IMP001" +with app.app_context(): + a, u = import_materials(rows, operator="tester") + assert a == 1 +print("[OK] Excel 物料导入 (新增 IMP001)") + +# 10.5 批量导入窗口(入库) 预览 + 提交 +assert "批量导入窗口" in c.get("/inbound").text +preview = c.post("/api/import/transactions/preview", + json={"kind": "inbound", "rows": [ + {"part_number": "IMP001", "qty": 50, "batch_no": "BA", "date": "2026-09-11", "remark": "批导"}, + {"part_number": "NOPE", "qty": 5}, + ]}) +pv = preview.get_json() +assert pv["valid"] == 1 and pv["invalid"] == 1, pv +commit = c.post("/api/import/transactions/commit", json={"kind": "inbound", "rows": pv["rows"]}) +cj = commit.get_json() +assert cj["added"] == 1 and cj["skipped"] == 1, cj +with app.app_context(): + rec = db.session.scalars(db.select(Inbound).where(Inbound.batch_no == "BA")).first() + assert rec and abs(rec.qty - 50) < 1e-6 +# 出库预览: 物料不存在应标记异常 +opv = c.post("/api/import/transactions/preview", + json={"kind": "outbound", "rows": [{"part_number": "GHOST", "qty": 1}]}).get_json() +assert opv["rows"][0]["ok"] is False +print("[OK] 批量导入窗口: 预览校验 + 提交入库 1 条 / 跳过 1 条(物料不存在)") + +# 11. 用户管理 (admin) +assert "用户列表" in c.get("/users").text +c.post("/user/add", data={"username": "op1", "password": "pw", "role": "operator"}, follow_redirects=True) +with app.app_context(): + assert db.session.scalar(db.select(User).where(User.username == "op1")) +print("[OK] 用户新增 (operator)") + +# 12. 角色拦截: operator 不能访问 /users +c.post("/logout") +c.post("/login", data={"username": "op1", "password": "pw"}, follow_redirects=True) +assert c.get("/users").status_code == 302 +assert c.get("/boms").status_code == 302 +print("[OK] 角色权限拦截 (operator 无权访问用户/BOM)") + +# 13. 物料批量导入窗口(粘贴预览 + 提交) +c.post("/logout") +c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True) +assert "批量导入窗口" in c.get("/materials").text +mpv = c.post("/api/import/materials/preview", json={"text": "PK001 料一 个 10 200\nPK002 料二 卷 5 50\nPK001 重复料 个 1 1"}) +mp = mpv.get_json() +assert mp["valid"] == 2 and mp["invalid"] == 1, mp # 第三行本批重复编号 +# 提交有效行 +mc = c.post("/api/import/materials/commit", json={"rows": mp["rows"]}).get_json() +assert mc["added"] == 2 and mc["updated"] == 0, mc +with app.app_context(): + assert db.session.scalar(db.select(Material).where(Material.part_number == "PK001")) +# operator 越权应被拦截 +c.post("/logout") +c.post("/login", data={"username": "op1", "password": "pw"}, follow_redirects=True) +assert c.post("/api/import/materials/preview", json={"text": "X1 a 个 1 1"}).status_code == 302 +c.post("/logout") +c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True) +print("[OK] 物料批量导入窗口: 预览校验 + 提交 2 条 / 重复拦截 + 角色拦截") + +# 14. BOM 批量导入窗口(多 BOM 粘贴预览 + 提交 + 自动建物料) +assert "批量导入窗口" in c.get("/boms").text +bpv = c.post("/api/import/boms/preview", json={"text": + "#BOM BM1 产品一\nPK001 2\nPK002 0.5\nNOPE 3\n#BOM BM2 产品二\nPK001 4"}) +bp = bpv.get_json() +assert bp["valid"] == 2 and bp["invalid"] == 0, bp +bm1 = next(b for b in bp["boms"] if b["code"] == "BM1") +# NOPE 物料不存在 -> 预览标 auto_create, 仍算可导入项(导入时自动新建) +assert bm1["valid_items"] == 3 and bm1["skip"] == 0, bm1 +nope_item = next(it for it in bm1["items"] if it["part_number"] == "NOPE") +assert nope_item["auto_create"] is True, nope_item +bc = c.post("/api/import/boms/commit", json={"boms": bp["boms"]}).get_json() +# NOPE 被自动新建物料; BM2 的 PK001 已存在 -> 共新建 1 个物料, 成功项 4 +assert bc["boms"] == 2 and bc["added_items"] == 4 and bc["created_materials"] == 1, bc +with app.app_context(): + assert db.session.scalar(db.select(Bom).where(Bom.code == "BM1")) + assert db.session.scalar(db.select(Bom).where(Bom.code == "BM2")) + assert db.session.scalar(db.select(Material).where(Material.part_number == "NOPE")) +print("[OK] BOM 批量导入窗口: 多 BOM 预览 + 物料不存在自动新建 + 提交 2 BOM / 成功项 4 / 新建物料 1") + +# 15. 真实 Altium/嘉立创 格式: 多 sheet, Part Number + Quantity 列 +wb = openpyxl.Workbook() +ws1 = wb.active +ws1.title = "XM-A" +ws1.append(["Item", "Part Number", "Manufacture Part Number", "Quantity", "Reference", "Part", "Package"]) +ws1.append([1, "P100", "MF100", 2, "U1", "MF100", "PKG1"]) +ws1.append([2, "P101", "MF101", 1, "U2", "MF101", "PKG2"]) +ws2 = wb.create_sheet("组装") +ws2.append(["序号", "层级", "物料代码", "描述", "数量", "材质"]) +ws2.append([1, "0.1", "P200", "外壳", 3, "ABS"]) +ws2.append([2, "0.1", "P201", "螺丝", 4, "铁"]) +tp2 = os.path.join(tempfile.gettempdir(), "bom_real.xlsx") +wb.save(tp2) +with open(tp2, "rb") as f: + r = c.post("/api/import/boms/preview", data={"file": (f, "bom_real.xlsx")}, + content_type="multipart/form-data") +rpv = r.get_json() +codes = [b["code"] for b in rpv["boms"]] +assert "XM-A" in codes and "组装" in codes, rpv +assert rpv["valid"] == 2, rpv +# 提交(自动建 4 个物料) +rbc = c.post("/api/import/boms/commit", json={"boms": rpv["boms"]}).get_json() +assert rbc["boms"] == 2 and rbc["created_materials"] == 4, rbc +with app.app_context(): + from models import Material as _M + xm = db.session.scalar(db.select(Bom).where(Bom.code == "XM-A")) + assert xm + its = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == xm.id)).all() + assert len(its) == 2 + p100 = db.session.scalar(db.select(_M).where(_M.part_number == "P100")) + assert p100 + it = next(i for i in its if i.material_id == p100.id) + assert abs(it.qty_per_unit - 2) < 1e-6 +print("[OK] 真实格式 BOM: 多 sheet 解析 + 自动建 4 物料 + 提交 2 个 BOM") + +# 15b. 用户提供的真实 BOM 模板文件直接解析验证(双 Quantity 取首列) +REAL = r"D:\办公\Q其他\20190316\2301045(码灵枪模组)\原理图\2301045T03\2301045T03\Output\2301045T03(生产)\2301045T03BOM(032A)-20260811.xlsx" +if os.path.exists(REAL): + from excel_io import read_boms_multi + real = read_boms_multi(REAL) + assert len(real) >= 2, real + assert "2301045T03" in [b["code"] for b in real], [b["code"] for b in real] + pcb = next(b for b in real if b["code"] == "2301045T03") + it_1501001 = next((it for it in pcb["items"] if it["part_number"] == "1501001"), None) + # 双 Quantity 列: 取第一个(D 列)=2, 而非 E 列=1 + assert it_1501001 and it_1501001["qty_per_unit"] == 2, it_1501001 + it_1001003 = next(it for it in pcb["items"] if it["part_number"] == "1001003") + assert it_1001003["manufacture"] == "CFW32C7UL_BUA", it_1001003 + print(f"[OK] 真实模板文件解析: {len(real)} 个 BOM, 主 BOM {len(pcb['items'])} 项, 双Quantity取首列正确") +else: + print("[SKIP] 真实模板文件不存在, 跳过 15b") + +print("\n全部冒烟测试通过 ✅") diff --git a/tools/verify_detail_features.py b/tools/verify_detail_features.py new file mode 100644 index 0000000..2e487f1 --- /dev/null +++ b/tools/verify_detail_features.py @@ -0,0 +1,122 @@ +# -*- coding: utf-8 -*- +"""验证新增功能: 物料明细(入库/消耗总量 + 入库流水时间) 与 BOM 明细(规格型号)。 + +正确收尾: 用 _exe_util.stop_exe 杀整棵进程树, 避免残留进程霸占 8080。 +""" +import http.client +import os +import shutil +import sqlite3 +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _exe_util import EXE, start_exe, stop_exe + +HOST, PORT = "127.0.0.1", 8080 +DB = r"D:\2026-09-10-task-20\material_wms\dist\app.db" +SEED = r"D:\2026-09-10-task-20\material_wms\instance\app.db" + + +def seed_test_bom(): + """往 dist 库插入一个测试 BOM(幂等), 引用已存在的物料 1/2/3。""" + c = sqlite3.connect(DB) + c.execute("DELETE FROM bom_item WHERE bom_id IN (SELECT id FROM bom WHERE code='TESTBOM01')") + c.execute("DELETE FROM bom WHERE code='TESTBOM01'") + c.execute("INSERT INTO bom(code,name,note) VALUES('TESTBOM01','测试产品','verify')") + bid = c.execute("SELECT id FROM bom WHERE code='TESTBOM01'").fetchone()[0] + for mid, q in [(1, 2.0), (2, 3.0), (3, 1.0)]: + c.execute("INSERT INTO bom_item(bom_id,material_id,qty_per_unit) VALUES(?,?,?)", (bid, mid, q)) + c.commit() + c.close() + return bid + + +def req(c, method, path, cookie="", body=None, headers=None): + h = {"Cookie": cookie} + if headers: + h.update(headers) + if body is not None and "Content-Type" not in h: + h["Content-Type"] = "application/x-www-form-urlencoded" + c.request(method, path, body=body, headers=h) + r = c.getresponse() + data = r.read().decode("utf-8", "ignore") + return r.status, data, r.getheader("Set-Cookie") or "" + + +def wait_port(): + for _ in range(60): + try: + http.client.HTTPConnection(HOST, PORT, timeout=2).connect() + return True + except Exception: + time.sleep(1) + return False + + +def main(): + if os.path.exists(SEED): + shutil.copyfile(SEED, DB) + bid = seed_test_bom() + + p = start_exe() + try: + assert wait_port(), "端口未就绪" + c = http.client.HTTPConnection(HOST, PORT, timeout=15) + _, _, ck = req(c, "GET", "/login") + _, _, ck2 = req(c, "POST", "/login", + body=b"username=admin&password=admin123", + cookie=ck.split(";")[0]) + cookie = (ck2.split(";")[0]) if ck2 else ck.split(";")[0] + + ok = True + + # 1) 物料列表含 入库总数量 / 消耗总数量 列 + s, html, _ = req(c, "GET", "/materials", cookie=cookie) + print(f"[materials] status={s}") + assert s == 200, f"materials status={s}" + for kw in ["入库总数量", "消耗总数量"]: + has = kw in html + print(f" [{'OK' if has else 'FAIL'}] 列表含列: {kw}") + ok = ok and has + + # 2) 物料明细 material/1: 入库=100, 消耗=30, 含入库流水日期 + s, html, _ = req(c, "GET", "/material/1", cookie=cookie) + print(f"[material/1] status={s} len={len(html)}") + assert s == 200, f"material_detail status={s}" + for kw in ["入库总数量", "消耗总数量", "入库流水", "当前在库"]: + has = kw in html + print(f" [{'OK' if has else 'FAIL'}] 明细含: {kw}") + ok = ok and has + has_date = "2026-09-11" in html + print(f" [{'OK' if has_date else 'FAIL'}] 入库流水含入库日期 2026-09-11") + ok = ok and has_date + has_in = "100.00" in html + print(f" [{'OK' if has_in else 'FAIL'}] 入库总数量显示 100.00") + ok = ok and has_in + has_out = "30.00" in html + print(f" [{'OK' if has_out else 'FAIL'}] 消耗总数量显示 30.00") + ok = ok and has_out + + # 3) BOM 明细含规格型号 + s, html, _ = req(c, "GET", f"/bom/{bid}", cookie=cookie) + print(f"[bom/{bid}] status={s} len={len(html)}") + assert s == 200, f"bom_detail status={s}" + for kw in ["型号 / 规格", "单套用量", "组成物料"]: + has = kw in html + print(f" [{'OK' if has else 'FAIL'}] BOM明细含: {kw}") + ok = ok and has + for spec in ["X1500", "FH8626V100", "CFW32C7UL_BUA"]: + has = spec in html + print(f" [{'OK' if has else 'FAIL'}] BOM明细含物料规格型号 {spec}") + ok = ok and has + + c.close() + print("\n" + ("✅ 新功能端到端验证通过" if ok else "❌ 存在未通过项")) + return 0 if ok else 1 + finally: + stop_exe(p) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/启动.bat b/启动.bat new file mode 100644 index 0000000..fb7bb6c --- /dev/null +++ b/启动.bat @@ -0,0 +1,7 @@ +@echo off +chcp 65001 >nul +cd /d "%~dp0" +set "PY=%USERPROFILE%\.workbuddy\binaries\python\envs\default\Scripts\python.exe" +if not exist "%PY%" set "PY=python" +start "" http://127.0.0.1:8080 +"%PY%" app.py 8080 diff --git a/物料导入模板.xlsx b/物料导入模板.xlsx new file mode 100644 index 0000000..8c460d2 Binary files /dev/null and b/物料导入模板.xlsx differ