From 95057dedf569ba19d2ac16986fd5f22eeeeaafc2 Mon Sep 17 00:00:00 2001 From: yueli Date: Mon, 3 Aug 2026 15:53:10 +0800 Subject: [PATCH] =?UTF-8?q?refactor(import):=20=E5=AF=BC=E5=85=A5=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E6=A0=B8=E5=BF=83=E9=80=BB=E8=BE=91=E6=94=B9=E9=80=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 预览阶段增加 Excel 内部重复检测 + 空规格去重 + 字段长度校验 - 执行阶段支持 skip 模式（跳过重复行继续导入）+ update 模式（覆盖更新已有记录） - update 模式空值保护：Excel 空单元格不覆盖已有数据 - DB commit 异常防御：字段超长等异常优雅返回而非 500 - 去重规则改为仅依据 spec_model（货号唯一键） - BOM 导入物料查找同步改为 spec-only --- .../app/services/import_service.py | 204 ++++++++++++++---- 1 file changed, 166 insertions(+), 38 deletions(-) diff --git a/inventory-backend/app/services/import_service.py b/inventory-backend/app/services/import_service.py index 377fa3e..91eb7fe 100644 --- a/inventory-backend/app/services/import_service.py +++ b/inventory-backend/app/services/import_service.py @@ -116,8 +116,14 @@ def generate_template(import_type: str) -> io.BytesIO: return output -def preview_material(file_stream) -> list: - """Dry-Run 预览基础信息导入""" +def preview_material(file_stream, mode: str = 'skip') -> list: + """Dry-Run 预览基础信息导入（含 Excel 内部重复检测 + 空规格去重） + + Args: + file_stream: Excel 文件流 + mode: 'skip' — DB 重复视为错误（默认） + 'update' — DB 重复标记为 'update' 状态（将被覆盖更新，不视为错误） + """ import pandas as pd df = pd.read_excel(file_stream, header=0) @@ -138,15 +144,19 @@ def preview_material(file_stream) -> list: } df.rename(columns=col_map, inplace=True) - # 批量预查已存在的 name+spec 组合（用于重复检测） - existing = db.session.query(MaterialBase.name, MaterialBase.spec_model).all() - existing_set = {(r.name, r.spec_model or '') for r in existing} + # 批量预查已存在的 spec_model（作为唯一货号，用于重复检测） + existing = db.session.query(MaterialBase.spec_model).all() + existing_set = {r.spec_model for r in existing if r.spec_model} + + # Excel 文件内部去重映射（spec → 首次出现行号） + excel_seen_map = {} results = [] for idx, row in df.iterrows(): row_num = idx + 2 # Excel 行号 errors = [] name = str(row.get('name', '')).strip() if pd.notna(row.get('name')) else '' + # spec_model 作为唯一货号，空值严格拦截 spec = str(row.get('spec', '')).strip() if pd.notna(row.get('spec')) else '' category = str(row.get('category', '')).strip() if pd.notna(row.get('category')) else '' mat_type = str(row.get('type', '')).strip() if pd.notna(row.get('type')) else '' @@ -155,7 +165,7 @@ def preview_material(file_stream) -> list: if not name: errors.append("名称不能为空") if not spec: - errors.append("规格型号不能为空") + errors.append("缺少必填项: 规格型号 (作为唯一货号不能为空)") if not category: errors.append("类别不能为空") if not mat_type: @@ -163,8 +173,40 @@ def preview_material(file_stream) -> list: if not unit: errors.append("计量单位不能为空") - if name and spec and (name, spec) in existing_set: - errors.append(f"已存在相同名称和规格的数据: {name} / {spec}") + # ★ 新增：字段长度校验（预防 DB commit 时因字段超长回滚） + MAX_LENS = { + '名称': (name, 255), '规格型号': (spec, 255), + '所属公司': (str(row.get('company_name', '')) if pd.notna(row.get('company_name')) else '', 255), + '专业名称': (str(row.get('common_name', '')) if pd.notna(row.get('common_name')) else '', 255), + '类别': (category, 100), '类型': (mat_type, 100), + '计量单位': (unit, 50), + } + for field_label, (val, max_len) in MAX_LENS.items(): + if len(val) > max_len: + errors.append(f"{field_label}超出数据库限制({len(val)}>{max_len}字符)") + + # ★ 去重检测（仅依据 spec_model 作为唯一货号） + if spec: + is_db_dup = spec in existing_set + + if is_db_dup: + if mode == 'update': + # 覆盖更新模式：DB 重复不视为错误 + if spec in excel_seen_map: + first_row = excel_seen_map[spec] + errors.append(f"与表格内第 {first_row} 行的规格型号重复: {spec}") + else: + excel_seen_map[spec] = row_num + else: + # 跳过模式：DB 重复是错误 + errors.append(f"数据库已存在相同规格型号 (货号) 的数据: {spec}") + else: + # 不在 DB 中 → 检查是否与 Excel 内前面行重复 + if spec in excel_seen_map: + first_row = excel_seen_map[spec] + errors.append(f"与表格内第 {first_row} 行的规格型号重复: {spec}") + else: + excel_seen_map[spec] = row_num company = str(row.get('company_name', '')).strip() if pd.notna(row.get('company_name')) else '' common = str(row.get('common_name', '')).strip() if pd.notna(row.get('common_name')) else '' @@ -173,10 +215,18 @@ def preview_material(file_stream) -> list: is_enabled = parse_boolean(row.get('is_enabled', True)) is_inspection = parse_boolean(row.get('is_inspection_required', False)) + # 决定行状态：update 模式下 DB 重复但不含其他错误的行 → 'update' + if mode == 'update' and not errors and spec and spec in existing_set: + row_status = 'update' + elif errors: + row_status = 'error' + else: + row_status = 'success' + results.append({ 'row': row_num, - 'status': 'error' if errors else 'success', - 'error_msg': '; '.join(errors) if errors else '', + 'status': row_status, + 'error_msg': '; '.join(errors) if errors else ('将被覆盖更新' if row_status == 'update' else ''), 'data': { 'company_name': company, 'name': name, @@ -221,14 +271,14 @@ def preview_bom(file_stream) -> list: } df.rename(columns=col_map, inplace=True) - # 批量预查所有 MaterialBase + # 批量预查所有 MaterialBase（仅依据 spec_model 查找） all_materials = db.session.query( MaterialBase.id, MaterialBase.name, MaterialBase.spec_model ).all() mat_lookup = {} for m in all_materials: - key = (m.name, m.spec_model or '') - mat_lookup[key] = m.id + if m.spec_model: + mat_lookup[m.spec_model] = m.id # 预查所有已存在的 (bom_no, version, parent_id, child_id) 组合 all_bom_pairs = db.session.query( @@ -259,15 +309,13 @@ def preview_bom(file_stream) -> list: if pd.isna(dosage) or float(dosage) <= 0: errors.append("用量必须大于0") - parent_key = (parent_name, parent_spec) - parent_id = mat_lookup.get(parent_key) if parent_name else None + parent_id = mat_lookup.get(parent_spec) if parent_spec else None if parent_name and parent_id is None: - errors.append(f"父件不存在: {parent_name} / {parent_spec}") + errors.append(f"父件不存在 (规格型号): {parent_spec}") - child_key = (child_name, child_spec) - child_id = mat_lookup.get(child_key) if child_name else None + child_id = mat_lookup.get(child_spec) if child_spec else None if child_name and child_id is None: - errors.append(f"子件不存在: {child_name} / {child_spec}") + errors.append(f"子件不存在 (规格型号): {child_spec}") if parent_id and child_id and parent_id == child_id: errors.append("父件与子件不能是同一物料") @@ -298,21 +346,75 @@ def preview_bom(file_stream) -> list: return results -def execute_material_import(rows: list) -> dict: - """执行基础信息批量导入（事务保护）""" +def execute_material_import(rows: list, mode: str = 'skip') -> dict: + """执行基础信息批量导入 + + Args: + rows: 待导入行数据 + mode: 'skip' - 跳过重复行，继续导入其余数据（默认） + 'update' - 匹配到重复时覆盖更新已有记录 + """ + import pandas as pd + + def _has_value(val) -> bool: + """判断值是否有效（Excel 单元格非空、非 NaN）""" + if val is None: + return False + try: + if pd.isna(val): + return False + except Exception: + pass + s = str(val).strip() + return s != '' and s != 'nan' and s != 'None' + inserted = 0 + updated = 0 + skipped = 0 errors = [] try: for i, item in enumerate(rows): d = item.get('data', item) + spec = d.get('spec_model', '') try: - existing = MaterialBase.query.filter_by( - name=d['name'], spec_model=d['spec_model'] - ).first() + # ★ 仅依据 spec_model（唯一货号）查找已有记录 + existing = MaterialBase.query.filter_by(spec_model=spec).first() if spec else None + if existing: - errors.append({'row': i + 2, 'msg': f"已存在: {d['name']} / {d['spec_model']}"}) - continue + if mode == 'update': + # ★ 覆盖更新模式：用 Excel 数据更新已有记录（含名称） + # 仅当 Excel 单元格有实际值时覆盖，空值不覆盖已有数据 + if _has_value(d.get('name')): + existing.name = str(d['name']).strip() + if _has_value(d.get('company_name')): + existing.company_name = str(d['company_name']).strip() + if _has_value(d.get('common_name')): + existing.common_name = str(d['common_name']).strip() + if _has_value(d.get('category')): + existing.category = str(d['category']).strip() + if _has_value(d.get('material_type')): + existing.material_type = str(d['material_type']).strip() + if _has_value(d.get('spec_model')): + existing.spec_model = str(d['spec_model']).strip() + if _has_value(d.get('unit')): + existing.unit = str(d['unit']).strip() + if d.get('reference_price') is not None and _has_value(d.get('reference_price')): + try: + existing.reference_price = float(d['reference_price']) + except Exception: + pass + if _has_value(d.get('is_inspection_required')): + existing.is_inspection_required = parse_boolean(d.get('is_inspection_required')) + if _has_value(d.get('is_enabled')): + existing.is_enabled = parse_boolean(d.get('is_enabled')) + updated += 1 + continue + else: + # ★ 跳过模式（默认）：记录跳过，继续导入其余行 + skipped += 1 + errors.append({'row': i + 2, 'msg': f"重复数据已跳过 (货号): {spec}"}) + continue mat = MaterialBase( company_name=d.get('company_name', ''), @@ -331,18 +433,43 @@ def execute_material_import(rows: list) -> dict: except Exception as e: errors.append({'row': i + 2, 'msg': str(e)}) - if errors: + # ★ 改造：不再"一刀切"回滚，只提交成功的行 + # 防御：捕获 DB 级异常（字段超长、约束冲突等），优雅返回而非 500 + try: + db.session.commit() + except Exception as commit_err: db.session.rollback() - return {'success': False, 'inserted': 0, 'errors': errors, - 'msg': f'导入失败，{len(errors)} 条数据校验未通过'} + return { + 'success': False, + 'inserted': 0, + 'updated': 0, + 'skipped': 0, + 'errors': [{'row': 0, 'msg': f'数据库写入失败: {str(commit_err)}'}], + 'msg': f'导入失败，数据库写入异常（可能字段超长或约束冲突）: {str(commit_err)}', + } - db.session.commit() - return {'success': True, 'inserted': inserted, 'errors': [], - 'msg': f'成功导入 {inserted} 条基础信息'} + # 构建结果消息 + parts = [] + if inserted > 0: + parts.append(f"新增 {inserted} 条") + if updated > 0: + parts.append(f"更新 {updated} 条") + if skipped > 0: + parts.append(f"跳过 {skipped} 条重复") + msg = "成功: " + "，".join(parts) if parts else "没有数据被导入" + + return { + 'success': True, + 'inserted': inserted, + 'updated': updated, + 'skipped': skipped, + 'errors': errors, + 'msg': msg, + } except Exception as e: db.session.rollback() - return {'success': False, 'inserted': 0, + return {'success': False, 'inserted': 0, 'updated': 0, 'skipped': 0, 'errors': [{'row': 0, 'msg': str(e)}], 'msg': f'导入异常: {str(e)}'} @@ -354,7 +481,8 @@ def execute_bom_import(rows: list) -> dict: ).all() mat_lookup = {} for m in all_materials: - mat_lookup[(m.name, m.spec_model or '')] = m.id + if m.spec_model: + mat_lookup[m.spec_model] = m.id inserted = 0 errors = [] @@ -369,11 +497,11 @@ def execute_bom_import(rows: list) -> dict: version = d.get('version', 'V1.0') if not parent_id: - pk = (d.get('parent_name', ''), d.get('parent_spec', '')) - parent_id = mat_lookup.get(pk) + parent_spec = d.get('parent_spec', '') + parent_id = mat_lookup.get(parent_spec) if parent_spec else None if not child_id: - ck = (d.get('child_name', ''), d.get('child_spec', '')) - child_id = mat_lookup.get(ck) + child_spec = d.get('child_spec', '') + child_id = mat_lookup.get(child_spec) if child_spec else None if not parent_id or not child_id: errors.append({'row': i + 2, 'msg': '父件或子件在系统中不存在'})