feat: Excel批量导入后端 — 模板下载+预览验证+执行导入

## import_service.py (新增)
- generate_template(): 基础信息/BOM Excel模板生成(带样例数据)
- parse_boolean(): 是/否/True/False 稳健布尔解析
- preview_material/bom(): Dry-Run解析+验证(重复/存在性/必填)
- BOM ffill: 父件列自动向下填充(支持合并单元格UX)
- execute_material/bom_import(): 事务包裹批量写入

## import_api.py (新增)
- GET /api/v1/import/template?type=material|bom
- POST /api/v1/import/preview (文件→预览验证)
- POST /api/v1/import/execute (确认导入)

## __init__.py + requirements.txt
- 注册 import 蓝图; 新增 pandas>=1.5.0
This commit is contained in:
yueli
2026-07-16 14:00:28 +08:00
parent 5b1cd4d8d0
commit 5411cabd0a
4 changed files with 541 additions and 0 deletions

View File

@ -0,0 +1,114 @@
"""
批量导入 API
端点:
GET /api/v1/import/template?type=material|bom → 下载Excel模板
POST /api/v1/import/preview → Dry-Run预览+验证
POST /api/v1/import/execute → 确认导入执行
"""
from flask import Blueprint, request, jsonify, send_file, current_app
from flask_jwt_extended import jwt_required
from app.utils.decorators import permission_required
from app.services.import_service import (
generate_template,
preview_material, preview_bom,
execute_material_import, execute_bom_import,
)
import traceback
import_bp = Blueprint('import_bp', __name__, url_prefix='/api/v1/import')
@import_bp.route('/template', methods=['GET'])
@jwt_required()
@permission_required('material_list')
def download_template():
"""下载导入模板"""
try:
import_type = request.args.get('type', 'material').strip()
if import_type not in ('material', 'bom'):
return jsonify({'code': 400, 'msg': 'type 必须为 material 或 bom'}), 400
file_stream = generate_template(import_type)
filename = f"{'基础信息' if import_type == 'material' else 'BOM表'}_导入模板.xlsx"
return send_file(
file_stream,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
current_app.logger.error(f'下载模板失败: {str(e)}')
return jsonify({'code': 500, 'msg': str(e)}), 500
@import_bp.route('/preview', methods=['POST'])
@jwt_required()
@permission_required('material_list:operation')
def preview_import():
"""Dry-Run 预览 + 验证(不写入数据库)"""
try:
import_type = request.form.get('type', 'material').strip()
file = request.files.get('file')
if not file:
return jsonify({'code': 400, 'msg': '请上传文件'}), 400
if import_type not in ('material', 'bom'):
return jsonify({'code': 400, 'msg': 'type 必须为 material 或 bom'}), 400
file_stream = file.read()
from io import BytesIO
if import_type == 'material':
results = preview_material(BytesIO(file_stream))
else:
results = preview_bom(BytesIO(file_stream))
success_count = sum(1 for r in results if r['status'] == 'success')
error_count = sum(1 for r in results if r['status'] == 'error')
return jsonify({
'code': 200, 'msg': '预览完成',
'data': {
'rows': results,
'total': len(results),
'success_count': success_count,
'error_count': error_count,
}
})
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'预览失败: {str(e)}'}), 500
@import_bp.route('/execute', methods=['POST'])
@jwt_required()
@permission_required('material_list:operation')
def execute_import():
"""确认导入执行(事务保护)"""
try:
data = request.get_json() or {}
import_type = data.get('type', 'material').strip()
rows = data.get('rows', [])
if not rows:
return jsonify({'code': 400, 'msg': '导入数据不能为空'}), 400
if import_type not in ('material', 'bom'):
return jsonify({'code': 400, 'msg': 'type 必须为 material 或 bom'}), 400
# 仅导入状态为 success 的行
valid_rows = [r for r in rows if r.get('status') == 'success']
if not valid_rows:
return jsonify({'code': 400, 'msg': '没有可导入的有效数据全部为error状态'}), 400
if import_type == 'material':
result = execute_material_import(valid_rows)
else:
result = execute_bom_import(valid_rows)
code = 200 if result['success'] else 400
return jsonify({'code': code, 'msg': result['msg'], 'data': result})
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'导入失败: {str(e)}'}), 500