star
68b4831d8b
- 库存看板新增「BOM 物料齐套」标签页(仅 admin/manager), 逐棵 BOM 列出组成物料的 单套用量/当前在库/状态, 并给出整体齐套结论 - BOM 详情页组成物料表新增「当前在库」「状态」两列 - 抽出统一 _status_of() 判定函数, 红/黄/绿逻辑与看板一致 - 新增 tools/verify_bom_board.py 端到端验证脚本(已通过) - 新增使用教程 物料入库出库系统_使用教程.md - .gitignore 忽略 _backup/(旧 EXE 备份)
1026 lines
38 KiB
Python
1026 lines
38 KiB
Python
"""物料入库/出库系统 - 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 _status_of(stock, safety):
|
|
"""库存状态: danger=负库存, warn=低于安全库存, ok=正常"""
|
|
if stock < 0:
|
|
return "danger"
|
|
if stock < (safety or 0):
|
|
return "warn"
|
|
return "ok"
|
|
|
|
|
|
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)
|
|
status = _status_of(stock, m.safety_stock)
|
|
if status == "danger":
|
|
negative += 1
|
|
elif status == "warn":
|
|
low += 1
|
|
rows.append({"m": m, "stock": stock, "status": status})
|
|
|
|
# BOM 物料齐套看板(仅管理员/主管可见)
|
|
can_bom = bool(g.user and g.user.role in ("admin", "manager"))
|
|
bom_board = []
|
|
if can_bom:
|
|
boms = db.session.scalars(db.select(Bom).order_by(Bom.code)).all()
|
|
for b in boms:
|
|
items = db.session.scalars(
|
|
db.select(BomItem).where(BomItem.bom_id == b.id).order_by(BomItem.id)
|
|
).all()
|
|
item_rows = []
|
|
for it in items:
|
|
m = db.session.get(Material, it.material_id)
|
|
if not m:
|
|
continue
|
|
stock = sm.get(m.id, 0)
|
|
item_rows.append({"m": m, "qty": it.qty_per_unit,
|
|
"stock": stock, "status": _status_of(stock, m.safety_stock)})
|
|
if any(x["status"] == "danger" for x in item_rows):
|
|
bstatus = "danger"
|
|
elif any(x["status"] == "warn" for x in item_rows):
|
|
bstatus = "warn"
|
|
else:
|
|
bstatus = "ok"
|
|
bom_board.append({"b": b, "parts": item_rows, "status": bstatus})
|
|
|
|
return render_template("dashboard.html", rows=rows, negative=negative, low=low,
|
|
bom_board=bom_board, can_bom=can_bom)
|
|
|
|
|
|
# ---------------- 物料 ----------------
|
|
@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/<int:mid>", 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/<int:mid>", 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/<int:mid>")
|
|
@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/<int:bid>", 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/<int:bid>", 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/<int:bid>")
|
|
@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()
|
|
sm = stock_map()
|
|
rows = []
|
|
for it in items:
|
|
m = db.session.get(Material, it.material_id)
|
|
if not m:
|
|
continue
|
|
stock = sm.get(m.id, 0)
|
|
rows.append({"m": m, "qty": it.qty_per_unit,
|
|
"stock": stock, "status": _status_of(stock, m.safety_stock)})
|
|
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/<int:oid>")
|
|
@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/<int:oid>/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/<int:uid>", 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/<kind>")
|
|
@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)
|