feat: 库存管理系统 MaterialWMS 初始版本
- Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留)
This commit is contained in:
+473
@@ -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
|
||||
Reference in New Issue
Block a user