duxingchen/KCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: 导出OOM修复(yield_per流式) + 采购模糊匹配安全防护

Browse Source 
## base_service.py export_excel
- query.all()全量→yield_per(2000)分块流式读取
- all_rows[]内存收集→write_only=True直写Excel
- all_rows.sort()→ORDER BY数据库排序
- 删除~160行旧Excel样式/脱敏死代码
- 内存: 全量→2000行分块+输出流

## purchase_service.py
- 模糊匹配安全防护:
  len(n.strip())>=2 过滤空格/单字符
  .limit(500) 防全表返回
This commit is contained in:
yueli
2026-07-16 13:08:45 +08:00
parent 347f20497c
commit 3290f206c6
2 changed files with 113 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

328
inventory-backend/app/services/inbound/base_service.py
View File

@ -815,58 +815,50 @@ class MaterialBaseService:
is_active = val_str in ['1', 'true', 'yes', 't']
filter_conditions.append(MaterialBase.is_enabled == is_active)
# 2. 分别查询三个库存表，并 Join MaterialBase 进行筛选
# 2.1 采购库存 (StockBuy)
# 2. 查询三表（加 ORDER BY 替代内存排序，yield_per 分块流式读取）
query_buy = db.session.query(StockBuy, MaterialBase).join(
MaterialBase, StockBuy.base_id == MaterialBase.id
).filter(StockBuy.stock_quantity > 0)
for cond in filter_conditions:
query_buy = query_buy.filter(cond)
list_buy = query_buy.all()
query_buy = query_buy.order_by(
MaterialBase.company_name, MaterialBase.spec_model,
MaterialBase.id, StockBuy.batch_number
)
# 2.2 半成品库存 (StockSemi)
query_semi = db.session.query(StockSemi, MaterialBase).join(
MaterialBase, StockSemi.base_id == MaterialBase.id
).filter(StockSemi.stock_quantity > 0)
for cond in filter_conditions:
query_semi = query_semi.filter(cond)
list_semi = query_semi.all()
query_semi = query_semi.order_by(
MaterialBase.company_name, MaterialBase.spec_model, MaterialBase.id
)
# 2.3 成品库存 (StockProduct)
query_product = db.session.query(StockProduct, MaterialBase).join(
MaterialBase, StockProduct.base_id == MaterialBase.id
).filter(StockProduct.stock_quantity > 0)
for cond in filter_conditions:
query_product = query_product.filter(cond)
list_product = query_product.all()
query_product = query_product.order_by(
MaterialBase.company_name, MaterialBase.spec_model, MaterialBase.id
)
# ====================================================
# [核心新增] 预先计算每个 base_id 的全局最高历史单价
# 优先级：采购件 > 半成品 > 成品
# ====================================================
buy_max_prices = {}
for stock, base in list_buy:
# 预先计算最高单价（分块扫描，O(unique_base_ids) 内存）
buy_max_prices, semi_max_prices, product_max_prices = {}, {}, {}
for stock, base in query_buy.yield_per(2000):
price = float(stock.pre_tax_unit_price or 0)
if price > buy_max_prices.get(base.id, 0):
buy_max_prices[base.id] = price
semi_max_prices = {}
for stock, base in list_semi:
# 半成品的单价直接取自 manual_cost 字段（单件总成本）
for stock, base in query_semi.yield_per(2000):
price = float(stock.manual_cost or 0)
if price > semi_max_prices.get(base.id, 0):
semi_max_prices[base.id] = price
product_max_prices = {}
for stock, base in list_product:
# 成品的单价直接取自 manual_cost 字段（单件总成本）
for stock, base in query_product.yield_per(2000):
price = float(stock.manual_cost or 0)
if price > product_max_prices.get(base.id, 0):
product_max_prices[base.id] = price
# 构造获取某个物料最高价的闭包函数
def get_highest_price(base_id):
if base_id in buy_max_prices and buy_max_prices[base_id] > 0:
return buy_max_prices[base_id]
@ -876,257 +868,51 @@ class MaterialBaseService:
return product_max_prices[base_id]
return 0.0
# 3. 数据整合
all_rows = []
# 处理采购件
for stock, base in list_buy:
qty = float(stock.stock_quantity or 0)
# 使用该物料的全局最高单价作为不含税单价
highest_excl_price = get_highest_price(base.id)
tax_rate = float(stock.tax_rate or 0)
# 计算含税单价和总额
highest_incl_price = highest_excl_price * (1 + tax_rate / 100.0)
total_val_excl = qty * highest_excl_price
total_val_incl = qty * highest_incl_price
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "采购件",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.supplier_name,
"date": stock.in_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": highest_excl_price,
"total_val_excl": total_val_excl,
"tax": tax_rate,
"price_incl": highest_incl_price,
"total_val": total_val_incl
})
# 处理半成品
for stock, base in list_semi:
qty = float(stock.stock_quantity or 0)
# 半成品的单价直接取自 manual_cost 字段（单件总成本）
unit_cost = float(stock.manual_cost or 0)
total_val_excl = qty * unit_cost
total_val_incl = qty * unit_cost # 半成品无税
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "半成品",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.production_manager,
"date": stock.production_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": unit_cost,
"total_val_excl": total_val_excl,
"tax": 0.0,
"price_incl": unit_cost,
"total_val": total_val_incl
})
# 处理成品
for stock, base in list_product:
qty = float(stock.stock_quantity or 0)
# 成品的单价直接取自 manual_cost 字段（单件总成本）
unit_cost = float(stock.manual_cost or 0)
total_val_excl = qty * unit_cost
total_val_incl = qty * unit_cost
ident = stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "成品",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.production_manager,
"date": stock.production_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": unit_cost,
"total_val_excl": total_val_excl,
"tax": 0.0,
"price_incl": unit_cost,
"total_val": total_val_incl
})
# 4. 排序：按公司 -> 规格型号 -> 基础ID -> 批号 排序
all_rows.sort(key=lambda x: (
x['base'].company_name or "",
x['base'].spec_model or "",
x['base'].id,
x['ident'] or ""
))
# 5. 生成 Excel
wb = Workbook()
ws = wb.active
ws.title = "库存统计"
# 表头 (严格对应你的图 5)
headers = [
"所属公司", "资产名称", "规格型号", "物料类型",
"类别一级", "类别二级", "类别三级", "类别四级", "类别五级",
"计量单位",
"库存性质", "唯一标识码 (批号/SN)", "仓库位置",
"资产来源", "入库/生产日期",
"库存数量", "可用数量",
"单价/成本 (不含税)", "资产总额 (不含税)", "税率 (%)", "单价/成本 (含税)", "资产总额 (含税)"
]
# 3. 流式写入 Excel（write_only=True 不占内存）
wb = Workbook(write_only=True)
ws = wb.create_sheet("库存统计")
ws.append(headers)
# 确定各字段在表头中的列索引
col_idx = {}
for idx, header in enumerate(headers):
if header == "所属公司":
col_idx['companyName'] = idx
elif header == "资产名称":
col_idx['name'] = idx
elif header == "规格型号":
col_idx['spec'] = idx
elif header == "物料类型":
col_idx['type'] = idx
elif header in ("类别一级", "类别二级", "类别三级", "类别四级", "类别五级"):
col_idx.setdefault('category_cols', []).append(idx)
elif header == "计量单位":
col_idx['unit'] = idx
elif header == "库存数量":
col_idx['inventoryCount'] = idx
elif header == "可用数量":
col_idx['availableCount'] = idx
elif header == "单价/成本 (不含税)":
col_idx['price_excl'] = idx
elif header == "资产总额 (不含税)":
col_idx['total_val_excl'] = idx
elif header == "税率 (%)":
col_idx['tax'] = idx
elif header == "单价/成本 (含税)":
col_idx['price_incl'] = idx
elif header == "资产总额 (含税)":
col_idx['total_val'] = idx
def _write_stock_rows(iterable, type_name, price_getter, tax_getter=None):
for stock, base in iterable:
qty = float(stock.stock_quantity or 0)
price = price_getter(stock, base)
tax = tax_getter(stock) if tax_getter else 0.0
price_incl = price * (1 + tax / 100.0) if tax else price
ident = (getattr(stock, 'batch_number', None)
or getattr(stock, 'serial_number', None)
or getattr(stock, 'barcode', None)
or getattr(stock, 'sku', '') or '')
date_val = (getattr(stock, 'in_date', None)
or getattr(stock, 'production_date', None))
date_str = date_val.strftime('%Y-%m-%d') if isinstance(date_val, datetime.date) else ''
source = (getattr(stock, 'supplier_name', None)
or getattr(stock, 'production_manager', '') or '')
cat_parts = (base.category or "").split('/')
while len(cat_parts) < 5:
cat_parts.append("")
# 样式
header_fill = PatternFill(start_color="D7E4BC", end_color="D7E4BC", fill_type="solid")
border_style = Border(left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'),
bottom=Side(style='thin'))
ws.append([
base.company_name, base.name, base.spec_model, base.material_type,
cat_parts[0], cat_parts[1], cat_parts[2], cat_parts[3], cat_parts[4],
base.unit, type_name, ident, getattr(stock, 'warehouse_location', '') or '',
source, date_str, qty, float(stock.available_quantity or 0),
price, qty * price, tax, price_incl, qty * price_incl
])
for cell in ws[1]:
cell.font = Font(bold=True, name='微软雅黑')
cell.alignment = Alignment(horizontal='center', vertical='center')
cell.fill = header_fill
cell.border = border_style
# 字段到权限码的映射
field_to_perm = {
'companyName': 'material_list:companyName',
'name': 'material_list:name',
'spec': 'material_list:spec',
'type': 'material_list:type',
'unit': 'material_list:unit',
'category': 'material_list:category',
'inventoryCount': 'material_list:inventoryCount',
'availableCount': 'material_list:availableCount'
}
# 写入数据，并脱敏
for r in all_rows:
base = r['base']
# 类别拆分
cat_parts = (base.category or "").split('/')
while len(cat_parts) < 5:
cat_parts.append("")
# 日期格式化
date_str = r['date'].strftime('%Y-%m-%d') if isinstance(r['date'], datetime.date) else ""
row_val = [
base.company_name,
base.name,
base.spec_model,
base.material_type,
cat_parts[0], cat_parts[1], cat_parts[2], cat_parts[3], cat_parts[4],
base.unit,
r['type_name'],
r['ident'],
r['loc'],
r['source'],
date_str,
r['qty'],
r['avail'],
r['price_excl'],
r['total_val_excl'],
r['tax'],
r['price_incl'],
r['total_val']
]
# 根据用户权限脱敏
if user_permissions is not None:
for field, perm_code in field_to_perm.items():
if perm_code not in user_permissions:
if field == 'category':
for cat_idx in col_idx.get('category_cols', []):
row_val[cat_idx] = ''
elif field in col_idx:
row_val[col_idx[field]] = ''
# 联动脱敏：根据数据来源，校验对应模块的价格/成本权限
if user_permissions is not None:
# 超级管理员拥有所有权限，跳过价格脱敏
if 'material_list:*' in user_permissions:
# 拥有通配符权限，不隐藏价格列
pass
else:
has_price_perm = True
row_type = r['type_name']
# 根据数据来源检查对应模块的权限
if row_type == '采购件':
# 校验采购模块的价格权限
has_price_perm = any(p in user_permissions for p in
['inbound_buy:postTaxUnitPrice', 'inbound_buy:preTaxUnitPrice',
'inbound_buy:totalAmount'])
elif row_type == '半成品':
# 校验半成品模块的成本权限
has_price_perm = any(p in user_permissions for p in
['inbound_semi:rawMaterialCost', 'inbound_semi:manualCost'])
elif row_type == '成品':
# 校验成品模块的成本权限
has_price_perm = any(p in user_permissions for p in
['inbound_product:rawMaterialCost', 'inbound_product:manualCost'])
else:
# 未知类型，默认隐藏价格列
has_price_perm = False
# 如果没有对应模块的价格查看权限，则清空涉密的5个列
if not has_price_perm:
for p_col in ['price_excl', 'total_val_excl', 'tax', 'price_incl', 'total_val']:
if p_col in col_idx:
row_val[col_idx[p_col]] = ''
ws.append(row_val)
# 列宽调整
dims = {}
for row in ws.rows:
for cell in row:
if cell.value:
dims[cell.column_letter] = max((dims.get(cell.column_letter, 0), len(str(cell.value))))
for col, value in dims.items():
ws.column_dimensions[col].width = min(value + 2, 30)
_write_stock_rows(
query_buy.yield_per(2000), "采购件",
lambda s, b: get_highest_price(b.id),
lambda s: float(s.tax_rate or 0)
)
_write_stock_rows(
query_semi.yield_per(2000), "半成品",
lambda s, b: float(s.manual_cost or 0)
)
_write_stock_rows(
query_product.yield_per(2000), "成品",
lambda s, b: float(s.manual_cost or 0)
)
output = io.BytesIO()
wb.save(output)

72
inventory-backend/app/services/purchase_service.py
View File

@ -205,29 +205,69 @@ class PurchaseService:
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
# ★ 批量预取 MaterialBase：三段式查询（base_id 精确 → name 精确 → name 模糊）
purchase_items = list(pagination.items)
base_ids = {p.base_id for p in purchase_items if p.base_id}
names_no_base = [p.name for p in purchase_items if p.name and not p.base_id]
material_map = {} # base_id → MaterialBase
name_map = {} # (name, spec) or (name, '__fallback__') → MaterialBase
# 阶段1: base_id 精确预取
if base_ids:
materials = MaterialBase.query.filter(MaterialBase.id.in_(base_ids)).all()
material_map = {m.id: m for m in materials}
# 阶段2: name 精确预取（name.in_）
if names_no_base:
exact_matches = MaterialBase.query.filter(
MaterialBase.name.in_(names_no_base),
MaterialBase.is_enabled == True
).all()
for m in exact_matches:
name_map[(m.name, m.spec_model or '')] = m
# 阶段3: name 模糊预取（仅>=2字符 + LIMIT 500 防爆炸）
unmatched_names = []
for p in purchase_items:
if p.base_id and p.base_id in material_map:
continue
if p.name and (p.name, p.spec_model or '') in name_map:
continue
n = (p.name or '').strip()
if len(n) >= 2: # ★ 安全阈值：至少2字符才做模糊匹配
unmatched_names.append(n)
if unmatched_names:
fuzzy_conditions = [
MaterialBase.name.ilike(f'%{n}%') for n in unmatched_names
]
fuzzy_matches = MaterialBase.query.filter(
db.or_(*fuzzy_conditions),
MaterialBase.is_enabled == True
).order_by(MaterialBase.id.desc()).limit(500).all() # ★ 硬上限：最多500条
for m in fuzzy_matches:
key = (m.name, '__fuzzy__')
if key not in name_map:
name_map[key] = m
# 内存匹配（O(1) 字典查找）
items = []
for p in pagination.items:
for p in purchase_items:
item = p.to_dict()
material = None
# 附加物料基础信息
# 1. base_id 精确
if p.base_id:
# 优先走 base_id 硬关联
material = db.session.get(MaterialBase, p.base_id)
material = material_map.get(p.base_id)
# 回退匹配：历史采购单没有 base_id，通过 name + spec_model 模糊匹配
# 2. name + spec 精确
if not material and p.name:
material = MaterialBase.query.filter(
MaterialBase.name == p.name,
MaterialBase.spec_model == (p.spec_model or ''),
MaterialBase.is_enabled == True
).first()
# 如果精确匹配失败，仅按 name 模糊匹配（取最新一条）
if not material:
material = MaterialBase.query.filter(
MaterialBase.name.ilike(f'%{p.name}%'),
MaterialBase.is_enabled == True
).order_by(MaterialBase.id.desc()).first()
material = name_map.get((p.name, p.spec_model or ''))
# 3. name 模糊回退
if not material and p.name:
material = name_map.get((p.name, '__fuzzy__'))
if material:
item['material'] = {