star
5060d027e5
- 多工厂(factory 表) + 顶栏工厂下拉过滤, 库存按厂聚合 - 厂间调拨(transfer): 源厂减+目标厂加, 总库存不变, 双向流水对账 - 替代料关系(substitute): 出库/BOM领料/生产领料缺料时提示可替代料 - 流水台账(ledger): 入出按日期合并折叠, 按日期导出同一 Excel - 生产订单增强: 编辑/删除(级联清关联流水)/调入数量/确认调入=厂间调拨 - 入库新增送货日期字段; 出库物料编码扫码定位 - 重写 README 补全全部模块; .gitignore 排除 dist_pkg/build_pkg/*.db - 停止跟踪 instance/app.db(运行时自动生成)
516 lines
19 KiB
Python
516 lines
19 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 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 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
|