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

refactor(import): 导入服务核心逻辑改造

Browse Source 
- 预览阶段增加 Excel 内部重复检测 + 空规格去重 + 字段长度校验
- 执行阶段支持 skip 模式（跳过重复行继续导入）+ update 模式（覆盖更新已有记录）
- update 模式空值保护：Excel 空单元格不覆盖已有数据
- DB commit 异常防御：字段超长等异常优雅返回而非 500
- 去重规则改为仅依据 spec_model（货号唯一键）
- BOM 导入物料查找同步改为 spec-only
This commit is contained in:
yueli
2026-08-03 15:53:10 +08:00
parent bc9cc18142
commit 95057dedf5
1 changed files with 166 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
inventory-backend/app/services/import_service.py
View File

@ -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': '父件或子件在系统中不存在'})