from app.extensions import db from app.models.bom import BomTable from app.models.base import MaterialBase from app.models.inbound.buy import StockBuy from app.models.inbound.semi import StockSemi from app.models.inbound.product import StockProduct from app.utils.decorators import get_current_company_filter from sqlalchemy import func, distinct, or_, case from collections import defaultdict import uuid import json import logging from datetime import datetime logger = logging.getLogger(__name__) # Redis 缓存键前缀 + TTL（秒） BOM_CACHE_PREFIX = 'bom:tree' BOM_CACHE_TTL = 43200 # 12小时 def _get_redis(): """ 获取 Redis 客户端实例，带容错保护。 若 extensions 中没有 redis_client 或连接失败，返回 None。 绝不抛出异常，确保业务不因此中断。 """ try: from app.extensions import redis_client return redis_client except Exception: return None def _cache_get(bom_no, version=None): """ 从 Redis 读取 BOM 缓存。 键 = bom:tree:{bom_no} 或 bom:tree:{bom_no}:{version} 返回：反序列化后的 dict 或 None """ client = _get_redis() if not client: return None key = f"{BOM_CACHE_PREFIX}:{bom_no}" + (f":{version}" if version else "") try: raw = client.get(key) if raw: logger.debug(f"[BOM Cache] HIT key={key}") return json.loads(raw) logger.debug(f"[BOM Cache] MISS key={key}") return None except Exception as e: logger.warning(f"[BOM Cache] GET 失败，降级查库. key={key}, err={e}") return None def _cache_set(bom_no, version, data): """ 将 BOM 数据写入 Redis，设置 12 小时 TTL。 即使写入失败也只是打日志，不阻断业务流程。 """ client = _get_redis() if not client: return key = f"{BOM_CACHE_PREFIX}:{bom_no}" + (f":{version}" if version else "") try: client.setex(key, BOM_CACHE_TTL, json.dumps(data, ensure_ascii=False)) logger.debug(f"[BOM Cache] SET key={key} ttl={BOM_CACHE_TTL}s") except Exception as e: logger.warning(f"[BOM Cache] SET 失败，已忽略. key={key}, err={e}") def _cache_delete(bom_no, version=None): """ 删除 Redis 中指定 BOM 的缓存条目。 在写操作（增/改/删）成功后调用，确保后续读请求拿到最新数据。 """ client = _get_redis() if not client: return # 删除版本级缓存 if version: key = f"{BOM_CACHE_PREFIX}:{bom_no}:{version}" try: client.delete(key) logger.debug(f"[BOM Cache] DEL key={key}") except Exception as e: logger.warning(f"[BOM Cache] DEL 失败，已忽略. key={key}, err={e}") # 同时删除"最新版"缓存（不带版本后缀），避免缓存不一致 key_latest = f"{BOM_CACHE_PREFIX}:{bom_no}" try: client.delete(key_latest) logger.debug(f"[BOM Cache] DEL key={key_latest}") except Exception as e: logger.warning(f"[BOM Cache] DEL 失败，已忽略. key={key_latest}, err={e}") class BomService: # ====================== 新版 BOM 逻辑（基于 bom_no） ====================== @staticmethod def generate_bom_no(): """生成唯一的 BOM 编号 (作为默认备选)""" timestamp = datetime.now().strftime('%Y%m%d%H%M%S') unique = str(uuid.uuid4())[:8] return f'BOM-{timestamp}-{unique}' @staticmethod def get_bom_list(keyword=None, active_only=False, category=None, page=1, limit=15, status=None): """ 获取所有 BOM 配方（按 bom_no + version 分组，单条 SQL 聚合 + 分页） 性能优化（v2）： - 消除 N+1：单条 GROUP BY + string_agg 查询替代循环内逐条查询 - 消除全量加载：DB 层 .paginate() 替代 .all() + Python 内存分页 - 消除 Python 二次过滤：关键词过滤完全下沉到 SQL（子查询 + EXISTS 语义） Args: category: 可选，按 parent_category 精确过滤（用于懒加载分组展开） """ child_alias = db.aliased(MaterialBase) # ===== 主聚合查询（单条 SQL，GROUP BY + string_agg） ===== query = db.session.query( BomTable.bom_no, BomTable.version, BomTable.parent_id, MaterialBase.name.label('parent_name'), MaterialBase.spec_model.label('parent_spec'), MaterialBase.category.label('parent_category'), BomTable.is_enabled, BomTable.is_archived, func.count(BomTable.child_id).label('child_count'), func.string_agg(child_alias.name, ', ').label('child_names'), func.string_agg(child_alias.spec_model, ', ').label('child_specs') ).join( MaterialBase, BomTable.parent_id == MaterialBase.id ).outerjoin( child_alias, BomTable.child_id == child_alias.id ).group_by( BomTable.bom_no, BomTable.version, BomTable.parent_id, MaterialBase.name, MaterialBase.spec_model, MaterialBase.category, BomTable.is_enabled, BomTable.is_archived ) # 状态过滤：status 优先于旧参数 active_only # enabled=启用且未归档(在用)；disabled=停用；archived=归档(保留可见/编辑，但不作子件引用候选) if status == 'enabled': query = query.filter(BomTable.is_enabled == True, BomTable.is_archived == False) elif status == 'disabled': query = query.filter(BomTable.is_enabled == False, BomTable.is_archived == False) elif status == 'archived': query = query.filter(BomTable.is_archived == True) elif active_only: query = query.filter(BomTable.is_enabled == True, BomTable.is_archived == False) # 【行级数据隔离】基于 JWT 多租户公司过滤 company_limit = get_current_company_filter() if company_limit is not None: query = query.filter(MaterialBase.company_name == company_limit) # 按类别过滤（用于懒加载分组展开） if category: query = query.filter(MaterialBase.category == category) # ===== 关键词过滤（完全下沉到 SQL，消除 Python 二次过滤） ===== if keyword: kw = f'%{keyword.strip()}%' # 子查询：找到所有匹配的 (bom_no, version) 对 kw_child_alias = db.aliased(MaterialBase) match_subq = db.session.query( BomTable.bom_no, BomTable.version ).join( MaterialBase, BomTable.parent_id == MaterialBase.id ).outerjoin( kw_child_alias, BomTable.child_id == kw_child_alias.id ).filter( or_( BomTable.bom_no.ilike(kw), MaterialBase.name.ilike(kw), MaterialBase.spec_model.ilike(kw), kw_child_alias.name.ilike(kw), kw_child_alias.spec_model.ilike(kw) ) ).distinct().subquery() # 用子查询结果过滤主查询 query = query.join( match_subq, db.and_( BomTable.bom_no == match_subq.c.bom_no, BomTable.version == match_subq.c.version ) ) # 排序（最新在前） query = query.order_by(BomTable.bom_no.desc(), BomTable.version.desc()) # ===== 数据库层分页（不再 .all() 到内存） ===== pagination = query.paginate(page=page, per_page=limit, error_out=False) # 组装结果 items = [] for row in pagination.items: items.append({ 'bom_no': row.bom_no, 'version': row.version, 'parent_id': row.parent_id, 'parent_name': row.parent_name, 'parent_spec': row.parent_spec or '', 'parent_category': row.parent_category or '', 'is_enabled': row.is_enabled, 'is_archived': row.is_archived, 'child_count': row.child_count, 'child_names': row.child_names or '', 'child_specs': row.child_specs or '' }) return { 'items': items, 'total': pagination.total, 'pages': pagination.pages, 'current_page': page } @staticmethod def get_bom_summary(keyword=None, status=None): """ BOM 分组摘要 API（极轻量，单条 GROUP BY + COUNT） SQL: SELECT m.category, COUNT(DISTINCT (b.bom_no, b.version)) AS count FROM bom_table b JOIN material_base m ON b.parent_id = m.id WHERE ... GROUP BY m.category ORDER BY m.category Returns: [{"category": "IRIS/半成品/无人机U", "count": 15}, ...] """ query = db.session.query( MaterialBase.category, func.count(func.distinct( func.concat(BomTable.bom_no, '|', BomTable.version) )).label('count') ).join( MaterialBase, BomTable.parent_id == MaterialBase.id ) # 行级数据隔离 company_limit = get_current_company_filter() if company_limit is not None: query = query.filter(MaterialBase.company_name == company_limit) # 状态过滤（按 (bom_no, version) 维度计数，故过滤粒度到行即可，行状态同版本一致） # enabled=启用且未归档；disabled=停用；archived=归档 if status == 'enabled': query = query.filter(BomTable.is_enabled == True, BomTable.is_archived == False) elif status == 'disabled': query = query.filter(BomTable.is_enabled == False, BomTable.is_archived == False) elif status == 'archived': query = query.filter(BomTable.is_archived == True) # 关键词搜索：与 get_bom_list 保持同一口径——子查询匹配 (bom_no, version)， # 同时覆盖 bom_no、父件名/规格、子件名/规格，保证任意关键字下 summary 计数与 list 明细绝对一致 if keyword: kw = f'%{keyword.strip()}%' kw_child_alias = db.aliased(MaterialBase) match_subq = db.session.query( BomTable.bom_no, BomTable.version ).join( MaterialBase, BomTable.parent_id == MaterialBase.id ).outerjoin( kw_child_alias, BomTable.child_id == kw_child_alias.id ).filter( or_( BomTable.bom_no.ilike(kw), MaterialBase.name.ilike(kw), MaterialBase.spec_model.ilike(kw), kw_child_alias.name.ilike(kw), kw_child_alias.spec_model.ilike(kw) ) ).distinct().subquery() query = query.join( match_subq, db.and_( BomTable.bom_no == match_subq.c.bom_no, BomTable.version == match_subq.c.version ) ) query = query.group_by(MaterialBase.category) \ .order_by(MaterialBase.category) rows = query.all() return [ {"category": r.category or "未分类", "count": r.count} for r in rows ] @staticmethod def get_bom_detail(bom_no, version=None): """ 根据 bom_no (和 version) 获取配方详情。 Cache-Aside 模式（三步走）: 1. 先查 Redis，有值直接返回（Cache Hit） 2. 无值或 Redis 报错，查数据库（Cache Miss → Fallback） 3. 数据库查好后写入 Redis，TTL=12h，供下次命中 注意：查询"最新版"时（version=None），缓存键不带版本后缀； 写入时也写入不带版本的键，这样无需指定 version 就能命中。 """ # ===== 第一步：尝试从 Redis 读取缓存 ===== cached = _cache_get(bom_no, version if version else None) if cached is not None: # Cache Hit：直接返回缓存数据，不再查库 logger.debug(f"[BOM] get_bom_detail bom_no={bom_no} version={version} → 命中缓存") return cached # ===== 第二步：Cache Miss → 查数据库 ===== logger.debug(f"[BOM] get_bom_detail bom_no={bom_no} version={version} → 查询数据库") query = db.session.query( BomTable, MaterialBase.name.label('child_name'), MaterialBase.spec_model.label('child_spec') ).outerjoin( # ★ LEFT JOIN：子件物料被删除后仍返回父件 BOM 结构，避免误报 "BOM 不存在" MaterialBase, BomTable.child_id == MaterialBase.id ).filter( BomTable.bom_no == bom_no ) if version: query = query.filter(BomTable.version == version) else: # ★ 自动"取最新"只认"启用且未归档"，避免默认命中已停用/已归档版本 latest_ver = db.session.query(BomTable.version).filter( BomTable.bom_no == bom_no, BomTable.is_enabled == True, BomTable.is_archived == False ).order_by(BomTable.version.desc()).limit(1).scalar() if not latest_ver: return None query = query.filter(BomTable.version == latest_ver) # 记录本次实际查的版本，用于缓存键 version = latest_ver rows = query.all() if not rows: return None first = rows[0] parent_id = first.BomTable.parent_id parent_material = MaterialBase.query.get(parent_id) children = [] # ★ 子件引用版本：优先读 bom_table 存储列（新建 BOM 时选定并落库）。 # 仅对"存储列为空但子件实为自制件"的旧数据/Excel 导入行，回退取其"最新启用配方"作展示兼容。 child_self_bom = {} self_ids = [ bom.child_id for bom, _, _ in rows if bom.child_id and not (bom.child_bom_no and bom.child_bom_version) ] if self_ids: _self_q = (db.session.query(BomTable.parent_id, BomTable.bom_no, BomTable.version) .filter(BomTable.parent_id.in_(self_ids), BomTable.is_enabled == True, BomTable.is_archived == False) .order_by(BomTable.parent_id, BomTable.id.desc()).all()) for _pid, _no, _ver in _self_q: if _pid not in child_self_bom: child_self_bom[_pid] = {'bom_no': _no, 'version': _ver} for bom, child_name, child_spec in rows: if bom.child_bom_no and bom.child_bom_version: cb_no, cb_ver = bom.child_bom_no, bom.child_bom_version else: _ref = child_self_bom.get(bom.child_id) or {} cb_no, cb_ver = _ref.get('bom_no', ''), _ref.get('version', '') children.append({ 'child_id': bom.child_id, 'child_name': child_name or '[已删除物料]', 'child_spec': child_spec or '', 'dosage': float(bom.dosage) if bom.dosage else 0.0, 'remark': bom.remark or '', 'child_bom_no': cb_no, 'child_bom_version': cb_ver }) result = { 'bom_no': bom_no, 'version': first.BomTable.version, 'parent_id': parent_id, 'parent_name': parent_material.name if parent_material else '', 'parent_spec': parent_material.spec_model if parent_material else '', 'is_enabled': first.BomTable.is_enabled, # ★ BOM 级备注：改取 bom_remark。此前误取 first.BomTable.remark # （第一个子件行的备注）冒充主表备注，是本次一并修掉的连带 bug 'remark': first.BomTable.bom_remark or '', 'children': children } # ===== 第三步：写入 Redis 缓存（TTL=12h），失败只打日志不阻断 ===== _cache_set(bom_no, version, result) return result @staticmethod def save_bom(data): """保存 BOM (支持多版本)，新增跨版本内容查重""" bom_no = data.get('bom_no') version = data.get('version', 'V1.0') parent_id = data['parent_id'] children = data['children'] is_enabled = data.get('is_enabled', True) # ★ BOM 级备注：写入本版本每一行（与 is_enabled 等父件级字段同口径） bom_remark = data.get('remark', '') or '' if not bom_no: raise ValueError('BOM编号不能为空') for child in children: if child['child_id'] == parent_id: raise ValueError('父件与子件不能是同一物料') # ===== 跨版本内容查重 ===== # 将当前提交的 children 转换为可比较集合 (child_id, dosage, 引用BOM号, 引用版本)。 # 引用版本纳入比较，避免"仅升级了自制件子件的引用版本"被误判为与旧版本完全重复。 def _cmp_key(child): return ( child['child_id'], int(child.get('dosage', 0) or 0), child.get('child_bom_no') or '', child.get('child_bom_version') or '', ) current_children_set = {_cmp_key(c) for c in children} # 查询该 bom_no 下所有其他版本的子件配置 existing_versions = db.session.query( BomTable.version, BomTable.child_id, BomTable.dosage, BomTable.child_bom_no, BomTable.child_bom_version ).filter( BomTable.bom_no == bom_no, BomTable.version != version # 排除当前正在保存的版本 ).all() # 按版本分组，构建每个版本的子件集合 version_children = {} for ver, child_id, dosage, child_bom_no, child_bom_version in existing_versions: version_children.setdefault(ver, set()).add( (child_id, int(dosage or 0), child_bom_no or '', child_bom_version or '') ) # 比对每个版本 for ver, existing_set in version_children.items(): if current_children_set == existing_set: raise ValueError(f'保存失败！当前子件配置与已有版本 {ver} 完全一致，请勿重复保存') # ===== 自制件子件"选版本"校验 ===== # 判定口径: 子件若存在"启用的下级 BOM"（即它作为父件有 is_enabled 配方）→ 视为自制件, 必须选定引用版本; # 否则视为外购件/无现行配方, 引用版本字段不保存（置 NULL）。 # 与读取侧(get_bom_detail)、停用占位配方后的启用集合保持一致，避免录入不在启用集的版本。 all_child_ids = [c.get('child_id') for c in children if c.get('child_id')] self_recipe_map = {} if all_child_ids: _rows = db.session.query( BomTable.parent_id, BomTable.bom_no, BomTable.version ).filter( BomTable.parent_id.in_(all_child_ids), BomTable.is_enabled == True, BomTable.is_archived == False ).distinct().all() for _pid, _bn, _ver in _rows: self_recipe_map.setdefault(_pid, set()).add((_bn, _ver)) self_names = {} if self_recipe_map: self_names = dict( db.session.query(MaterialBase.id, MaterialBase.name) .filter(MaterialBase.id.in_(list(self_recipe_map.keys()))).all() ) for child in children: cid = child.get('child_id') if cid is None: continue ref_no = (child.get('child_bom_no') or '').strip() ref_ver = (child.get('child_bom_version') or '').strip() recipes = self_recipe_map.get(cid) if recipes: # 自制件: 版本为必选, 且必须落在其启用配方集内 if not ref_no or not ref_ver: raise ValueError( f"子件「{self_names.get(cid, cid)}」为自制件，请选择其 BOM 版本后才可保存") if (ref_no, ref_ver) not in recipes: raise ValueError( f"子件「{self_names.get(cid, cid)}」所选版本 {ref_no} {ref_ver} 不存在或已停用，请重新选择") else: # 外购件/无现行 BOM: 清除版本字段 child['child_bom_no'] = None child['child_bom_version'] = None # ===== 执行保存 ===== # 仅删除当前版本的旧记录（改为对象级删除以触发审计事件） old_records = BomTable.query.filter_by(bom_no=bom_no, version=version).all() for rec in old_records: db.session.delete(rec) # 【核心修复】：强制立即执行 DELETE 语句，为后续的 INSERT 腾出唯一键空间 db.session.flush() for child in children: bom = BomTable( bom_no=bom_no, version=version, parent_id=parent_id, child_id=child['child_id'], dosage=child.get('dosage', 0), remark=child.get('remark', ''), bom_remark=bom_remark, child_bom_no=child.get('child_bom_no') or None, child_bom_version=child.get('child_bom_version') or None, is_enabled=is_enabled ) db.session.add(bom) db.session.commit() # ===== 写入后立刻清除缓存（Cache Invalidation） ===== # 确保后续 get_bom_detail 读取到最新数据，而不是 stale cache _cache_delete(bom_no, version) logger.info(f"[BOM Cache] save_bom → 缓存已失效 bom_no={bom_no} version={version}") return bom_no @staticmethod def update_enabled(bom_no, version, is_enabled): """ 三态互斥：切“启用/停用”时同步清除“归档”。 启用=在用；停用=有问题/暂停；两者都与归档互斥。 不触碰子件结构；成功后清除该 bom_no 的缓存。 """ bom_no = (bom_no or '').strip() version = (version or '').strip() if not bom_no or not version: raise ValueError('BOM编号/版本不能为空') rows = BomTable.query.filter_by(bom_no=bom_no, version=version).all() if not rows: raise ValueError(f'BOM不存在（bom_no={bom_no}, version={version}）') for rec in rows: rec.is_enabled = bool(is_enabled) rec.is_archived = False # 切启/停即退出归档（三者互斥） db.session.commit() _cache_delete(bom_no, version) logger.info(f"[BOM Cache] update_enabled → 缓存已失效 bom_no={bom_no} version={version} enabled={is_enabled}") return len(rows) @staticmethod def update_archived(bom_no, version, archived): """ 三态互斥归档：归档=彻底废弃（is_archived=True 且 is_enabled=False）； 取消归档=恢复“启用(在用)”（is_archived=False 且 is_enabled=True）。 归档后不出现在启用/停用视图、不作为子件引用候选。 """ bom_no = (bom_no or '').strip() version = (version or '').strip() if not bom_no or not version: raise ValueError('BOM编号/版本不能为空') rows = BomTable.query.filter_by(bom_no=bom_no, version=version).all() if not rows: raise ValueError(f'BOM不存在（bom_no={bom_no}, version={version}）') target_archived = bool(archived) for rec in rows: rec.is_archived = target_archived rec.is_enabled = not target_archived # 归档→停用；取消归档→启用（互斥） db.session.commit() _cache_delete(bom_no, version) logger.info(f"[BOM Cache] update_archived → 缓存已失效 bom_no={bom_no} version={version} archived={archived}") return len(rows) @staticmethod def get_bom_with_stock_by_bom_no(bom_no, version=None): """ 根据 bom_no (和可选 version) 获取配方详情，并计算（已修复 N+1 性能问题） """ detail = BomService.get_bom_detail(bom_no, version=version) if not detail or not detail.get('children'): return detail # 1. 提取所有子件的 ID 列表 child_ids = [child['child_id'] for child in detail['children']] # 2. 分别查三张库存表，Python 侧合并聚合（避免 UNION ALL 子查询列名问题） from collections import defaultdict stock_agg = defaultdict(lambda: {'qty': 0.0, 'locs': set()}) for model in (StockBuy, StockSemi, StockProduct): rows = db.session.query( model.base_id, model.available_quantity, model.warehouse_location ).filter( model.base_id.in_(child_ids), model.available_quantity > 0 ).all() for base_id, qty, loc in rows: stock_agg[base_id]['qty'] += float(qty or 0) if loc: stock_agg[base_id]['locs'].add(loc) # 3. 将聚合结果转换为字典 (Map)，方便后续 O(1) 极速匹配 stock_map = { base_id: { 'qty': agg['qty'], 'loc': ', '.join(sorted(agg['locs'])) if agg['locs'] else '' } for base_id, agg in stock_agg.items() } # 4. 遍历组装数据（纯内存操作，极快） for child in detail['children']: base_id = child['child_id'] stat = stock_map.get(base_id, {'qty': 0, 'loc': ''}) stock_qty = float(stat['qty']) dosage = float(child['dosage']) if child.get('dosage') else 0 child['current_stock'] = stock_qty child['warehouse_location'] = stat['loc'] child['max_producible'] = int(stock_qty // dosage) if dosage > 0 else 0 return detail # ====================== 兼容旧接口 ====================== @staticmethod def get_bom_no_by_parent(parent_id): # ★ 仅取"启用且未归档"的最新配方，避免默认命中已停用/已归档版本 row = BomTable.query.filter( BomTable.parent_id == parent_id, BomTable.is_enabled == True, BomTable.is_archived == False ).order_by(BomTable.version.desc()).first() return row.bom_no if row else None @staticmethod def get_child_bom_versions(child_id): """ 返回某物料作为父件时的"启用配方版本清单"(bom_no, version)， 供前端在选中自制件子件后生成必选版本下拉。空列表 = 非自制件/无现行 BOM，无需选版本。 排序: 版本倒序在前（最新优先，作为下拉默认项）。注意 version 为字符串，跨 V10 需另行规范。 """ if not child_id: return [] rows = (db.session.query(BomTable.bom_no, BomTable.version) .filter(BomTable.parent_id == child_id, BomTable.is_enabled == True, BomTable.is_archived == False) .distinct() .order_by(BomTable.version.desc(), BomTable.bom_no.desc()) .all()) return [{'bom_no': bn, 'version': ver} for bn, ver in rows] @staticmethod def create_or_update_bom(parent_id, child_list, bom_no=None, version='V1.0'): try: if not bom_no: existing = BomTable.query.filter_by(parent_id=parent_id).first() bom_no = existing.bom_no if existing else BomService.generate_bom_no() # 改为对象级删除以触发审计事件 old_records = BomTable.query.filter_by(bom_no=bom_no, version=version).all() for rec in old_records: db.session.delete(rec) for item in child_list: bom = BomTable( bom_no=bom_no, version=version, parent_id=parent_id, child_id=item['child_id'], dosage=item.get('dosage', 0), remark=item.get('remark', ''), child_bom_no=item.get('child_bom_no') or None, child_bom_version=item.get('child_bom_version') or None ) db.session.add(bom) db.session.commit() # ===== 写入后立刻清除缓存（Cache Invalidation） ===== _cache_delete(bom_no, version) logger.info(f"[BOM Cache] create_or_update_bom → 缓存已失效 bom_no={bom_no} version={version}") except Exception as e: db.session.rollback() logger.error(f"[BOM] create_or_update_bom 失败 bom_no={bom_no}: {e}") raise return True @staticmethod def get_bom_with_stock(parent_id): bom_no = BomService.get_bom_no_by_parent(parent_id) if not bom_no: return [] detail = BomService.get_bom_with_stock_by_bom_no(bom_no) return detail['children'] if detail else [] @staticmethod def calculate_cascade_inventory(bom_no, order_qty): """ 根据 bom_no 和订单数量，计算所有子件的级联库存缺口。 返回结构供 AI 消费，每个子件包含：parent_name / spec / name / level_type / need_qty / available_stock / suggested_qty / gap 若 BOM 不存在返回 None。 """ # 1. 获取 BOM 明细 detail = BomService.get_bom_detail(bom_no) if not detail or not detail.get('children'): return None parent_name = detail.get('parent_name', '') # 2. 提取所有子件 ID，分别查三张库存表，Python 侧合并聚合 child_ids = [child['child_id'] for child in detail['children']] from collections import defaultdict stock_agg = defaultdict(float) for model in (StockBuy, StockSemi, StockProduct): rows = db.session.query( model.base_id, model.available_quantity ).filter( model.base_id.in_(child_ids) ).all() for base_id, qty in rows: stock_agg[base_id] += float(qty or 0) stock_map = dict(stock_agg) # 3. 提取所有子件的基础物料信息（名称/规格/类型） materials = db.session.query( MaterialBase.id, MaterialBase.name, MaterialBase.spec_model ).filter(MaterialBase.id.in_(child_ids)).all() mat_map = { m.id: {'name': m.name or '', 'spec': m.spec_model or ''} for m in materials } # 4. 遍历子件，计算每个子件的缺口数据 results = [] for child in detail['children']: child_id = child['child_id'] dosage = float(child.get('dosage') or 0) need_qty = dosage * order_qty available_stock = stock_map.get(child_id, 0) suggested_qty = max(0.0, min(need_qty, available_stock)) gap = available_stock - need_qty mat_info = mat_map.get(child_id, {'name': '', 'spec': ''}) results.append({ 'parent_name': parent_name, 'spec': mat_info['spec'], 'name': mat_info['name'], 'level_type': 'child', 'need_qty': round(need_qty, 4), 'available_stock': round(available_stock, 4), 'suggested_qty': round(suggested_qty, 4), 'gap': round(gap, 4), }) return results