starandWorkBuddy
93f3ef458b
- 新增中间表 prod_order_bom(order_id, bom_id, qty)，一个订单可挂多个 BOM，每个 BOM 单独设生产数量 - prod_order.qty 改为各 BOM 数量汇总，bom_id 仅作主 BOM 冗余摘要 - 物料需求按 Σ(BOM 单套用量 × 各自生产数量) 跨 BOM 合并聚合，覆盖详情/领料/调入/导出 - 订单表单改为动态多行(BOM 下拉 + 数量输入 + 删除行)，列表与导出展示「BOM清单(型号×数量)」 - 导出 Excel 增至 3 个 sheet：订单信息 / BOM清单 / 领料明细 - 修复 _parse_bom_pairs 表单字段名 bom_id[]/qty[] 解析；order_delete 一并级联删除 prod_order_bom - build_exe.py 增加构建后自动拷贝 dist_pkg→dist（EXE 同目录即数据库所在） Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
658 lines
24 KiB
Python
658 lines
24 KiB
Python
"""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="导入", factory_id=None):
|
||
"""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 == "期初",
|
||
Inbound.factory_id == factory_id))
|
||
if exists:
|
||
exists.qty = op
|
||
else:
|
||
db.session.add(Inbound(material_id=m.id, factory_id=factory_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 export_prep(path, rows, scope=""):
|
||
"""导出备料清单(采购下单用, 不记账; 手工维护, 与生产订单无关)。
|
||
|
||
rows: [{id, m, plan(计划数量), stock(实际库存), diff(缺口=计划-库存), short}]
|
||
scope: 范围说明(工厂/全部)。
|
||
"""
|
||
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 = "需采购" if r["diff"] > 0 else ("刚好够" if r["diff"] == 0 else "已有盈余")
|
||
ws.append([m.part_number, m.manufacturer or "", m.unit or "个",
|
||
r["plan"], r["stock"], r["diff"], status, scope])
|
||
# 缺口着色: >0 需采购标红; <0 盈余标绿
|
||
if r["diff"] > 0:
|
||
ws.cell(row=ws.max_row, column=6).fill = PatternFill("solid", fgColor="FFC7CE")
|
||
elif r["diff"] < 0:
|
||
ws.cell(row=ws.max_row, column=6).fill = PatternFill("solid", fgColor="C6EFCE")
|
||
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 export_ledger(target, rows):
|
||
"""导出流水台账。target 可为路径或文件对象(如 BytesIO)。
|
||
|
||
rows: _build_ledger_rows() 返回的流水行列表, 每项含
|
||
date/type/cat/m/qty/fac/who/ref/op 字段。
|
||
"""
|
||
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"]
|
||
ws.append([
|
||
r["date"].isoformat() if r["date"] else "",
|
||
r["type"],
|
||
r["cat"],
|
||
m.part_number if m else "",
|
||
m.manufacturer if m else "",
|
||
m.unit if m else "",
|
||
r["qty"],
|
||
r["fac"].code if r["fac"] else "",
|
||
r["who"] or "",
|
||
r["ref"] or "",
|
||
r["op"] or "",
|
||
])
|
||
# 出入着色: 入(绿) / 出(红)
|
||
for i in range(2, ws.max_row + 1):
|
||
tval = ws.cell(row=i, column=2).value
|
||
if tval == "入":
|
||
ws.cell(row=i, column=2).fill = PatternFill("solid", fgColor="C6EFCE")
|
||
elif tval == "出":
|
||
ws.cell(row=i, column=2).fill = PatternFill("solid", fgColor="FFC7CE")
|
||
ws.freeze_panes = "A2"
|
||
wb.save(target)
|
||
|
||
|
||
def export_orders(target, data):
|
||
"""导出生产订单列表。
|
||
|
||
data: [{o, lines, factory, total}] (与 orders() 返回一致)。
|
||
lines: [(Bom, qty)] 该订单的 BOM 行(可多个)。
|
||
"""
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "生产订单"
|
||
head = ["订单号", "BOM清单(型号×数量)", "生产工厂", "状态",
|
||
"经手人", "下单时间", "备注", "总接单量"]
|
||
ws.append(head)
|
||
for c in ws[1]:
|
||
c.font = Font(bold=True)
|
||
c.fill = PatternFill("solid", fgColor="DDEBF7")
|
||
for d in data:
|
||
o = d["o"]
|
||
lines = d.get("lines") or []
|
||
fac = d["factory"]
|
||
total = d.get("total") or 0
|
||
bom_info = "\n".join(
|
||
f"{b.code}{('·' + b.name) if b.name else ''}×{q}" for b, q in lines) or "—"
|
||
status = "已领料" if o.status == "picked" else "待领料"
|
||
ws.append([
|
||
o.order_no,
|
||
bom_info,
|
||
(f"{fac.code}·{fac.name}" if fac else "未指定"),
|
||
status,
|
||
o.operator or "",
|
||
o.created_at.strftime("%Y-%m-%d %H:%M") if o.created_at else "",
|
||
o.remark or "",
|
||
total,
|
||
])
|
||
if o.status == "picked":
|
||
ws.cell(row=ws.max_row, column=4).fill = PatternFill("solid", fgColor="C6EFCE")
|
||
else:
|
||
ws.cell(row=ws.max_row, column=4).fill = PatternFill("solid", fgColor="FFE699")
|
||
ws.cell(row=ws.max_row, column=2).alignment = Alignment(wrap_text=True, vertical="top")
|
||
ws.freeze_panes = "A2"
|
||
wb.save(target)
|
||
|
||
|
||
def export_order_detail(target, o, lines, reqs, factory,
|
||
posted_total, planned_total, posted_count):
|
||
"""导出单个生产订单详情(派工单)：订单信息 sheet + BOM清单 sheet + 领料明细 sheet。
|
||
|
||
lines: [(Bom, qty)] 该订单的 BOM 行(可多个); 领料明细已按全部 BOM 汇总。
|
||
"""
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "订单信息"
|
||
ws.append(["生产订单派工单"])
|
||
ws["A1"].font = Font(bold=True, size=14)
|
||
total_qty = sum(q for _, q in (lines or []))
|
||
info = [
|
||
("订单号", o.order_no),
|
||
("BOM清单", "\n".join(
|
||
f"{b.code}{('·' + b.name) if b.name else ''}×{q}" for b, q in (lines or [])) or "-"),
|
||
("总接单量", total_qty),
|
||
("生产工厂", (f"{factory.code}·{factory.name}" if factory else "未指定（按汇总库存）")),
|
||
("经手人", o.operator or "-"),
|
||
("下单时间", o.created_at.strftime("%Y-%m-%d %H:%M") if o.created_at else ""),
|
||
("状态", "已领料" if o.status == "picked" else "待领料"),
|
||
("备注", o.remark or "-"),
|
||
("调入总览", f"已调入 {posted_count} 项 / 合计 {posted_total} 件；计划调入合计 {planned_total} 件"),
|
||
]
|
||
r = 3
|
||
for k, v in info:
|
||
ws.cell(row=r, column=1, value=k).font = Font(bold=True)
|
||
cell = ws.cell(row=r, column=2, value=v)
|
||
if k == "BOM清单":
|
||
cell.alignment = Alignment(wrap_text=True, vertical="top")
|
||
r += 1
|
||
|
||
ws2 = wb.create_sheet("BOM清单")
|
||
ws2.append(["BOM型号", "BOM名称", "生产数量"])
|
||
for c in ws2[1]:
|
||
c.font = Font(bold=True)
|
||
c.fill = PatternFill("solid", fgColor="DDEBF7")
|
||
for b, q in (lines or []):
|
||
ws2.append([b.code, b.name or "", q])
|
||
ws2.freeze_panes = "A2"
|
||
|
||
ws3 = wb.create_sheet("领料明细")
|
||
head = ["物料编号", "型号", "下单前库存", "需求数量", "缺口",
|
||
"调入数量", "状态", "工厂库存", "可替代料"]
|
||
ws3.append(head)
|
||
for c in ws3[1]:
|
||
c.font = Font(bold=True)
|
||
c.fill = PatternFill("solid", fgColor="DDEBF7")
|
||
for x in reqs:
|
||
subs = "、".join(
|
||
f"{s['pn']}·{s['stock']}{'✓' if s['available'] else '✗'}" for s in x["subs"]
|
||
) if x["subs"] else "—"
|
||
ws3.append([
|
||
x["m"].part_number,
|
||
x["m"].manufacturer or "-",
|
||
round(x["have"], 3),
|
||
round(x["need"], 3),
|
||
round(x["gap"], 3),
|
||
round(x["alloc"], 3),
|
||
"缺料" if x["short"] else "充足",
|
||
round(x["gap"] + x["alloc"], 3),
|
||
subs,
|
||
])
|
||
if x["short"]:
|
||
ws3.cell(row=ws3.max_row, column=7).fill = PatternFill("solid", fgColor="FFC7CE")
|
||
else:
|
||
ws3.cell(row=ws3.max_row, column=7).fill = PatternFill("solid", fgColor="C6EFCE")
|
||
ws3.freeze_panes = "A2"
|
||
wb.save(target)
|
||
|
||
|
||
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
|