diff --git a/inventory-backend/app/services/inbound/base_service.py b/inventory-backend/app/services/inbound/base_service.py index d6dcf7b..561cab0 100644 --- a/inventory-backend/app/services/inbound/base_service.py +++ b/inventory-backend/app/services/inbound/base_service.py @@ -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) diff --git a/inventory-backend/app/services/purchase_service.py b/inventory-backend/app/services/purchase_service.py index c583b03..adf9eaf 100644 --- a/inventory-backend/app/services/purchase_service.py +++ b/inventory-backend/app/services/purchase_service.py @@ -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'] = {