diff --git a/inventory-backend/app/__init__.py b/inventory-backend/app/__init__.py index d2c41de..780feff 100644 --- a/inventory-backend/app/__init__.py +++ b/inventory-backend/app/__init__.py @@ -258,6 +258,16 @@ def create_app(): except Exception as e: print(f"❌ 错误: Export 模块注册失败: {e}") + # ----------------------------------------------------- + # 2.14 注册批量导入模块 (Import) + # ----------------------------------------------------- + try: + from app.api.v1.import_api import import_bp + app.register_blueprint(import_bp, url_prefix='/api/v1/import') + print("✅ Import 模块注册成功") + except ImportError as e: + print(f"❌ 错误: Import 模块导入失败: {e}") + # ========================================================= # 3. 预加载数据模型 # ========================================================= diff --git a/inventory-backend/app/api/v1/import_api.py b/inventory-backend/app/api/v1/import_api.py new file mode 100644 index 0000000..842e5ad --- /dev/null +++ b/inventory-backend/app/api/v1/import_api.py @@ -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 diff --git a/inventory-backend/app/services/import_service.py b/inventory-backend/app/services/import_service.py new file mode 100644 index 0000000..377fa3e --- /dev/null +++ b/inventory-backend/app/services/import_service.py @@ -0,0 +1,415 @@ +""" +Excel/CSV 批量导入服务 +支持: 基础信息(MaterialBase) / BOM表(BomTable) +架构: 模板下载 → 预览验证(Dry-Run) → 确认执行 +""" +import io +import json +from datetime import datetime +from flask import current_app + +from app.extensions import db +from app.models.base import MaterialBase +from app.models.bom import BomTable + +# ── 模板定义 ── + +MATERIAL_TEMPLATE_HEADERS = [ + "所属公司(company_name)", "名称(name)*", "专业名称(common_name)", + "类别(category)*", "类型(material_type)*", + "规格型号(spec_model)*", "计量单位(unit)*", + "参考价格(reference_price)", "强制质检(is_inspection_required)", + "是否启用(is_enabled)" +] + +MATERIAL_TEMPLATE_INSTRUCTIONS = [ + "请勿修改表头行", "标记*的列为必填", + "类别格式: IRIS/半成品/无人机U (用/分隔层级)", + "强制质检/是否启用: 是/Y/True/1 或 否/N/False/0, 默认否/是", +] + +BOM_TEMPLATE_HEADERS = [ + "BOM编号(bom_no)*", "版本(version)", "父件名称(parent_name)*", + "父件规格(parent_spec)", "子件名称(child_name)*", + "子件规格(child_spec)", "用量(dosage)*", "损耗率%(loss_rate)", + "备注(remark)", "是否启用(is_enabled)" +] + +MATERIAL_EXAMPLE_ROWS = [ + ["IRIS", "LICA-3000", "激光雷达", "IRIS/成品/LICA", "传感器", "LICA-3000/A", "个", 15000, "否", "是"], + ["IRIS", "碳纤维板材", "Carbon Plate", "IRIS/半成品/材料", "原材料", "CF-200/3K", "张", 800, "是", "是"], + ["IRIS", "M3内六角螺丝", "", "IRIS/采购件/紧固件", "标准件", "M3x12", "颗", 0.5, "否", "是"], +] + +BOM_EXAMPLE_ROWS = [ + ["LICA-3000", "V1.0", "LICA-3000", "LICA-3000/A", "M3内六角螺丝", "M3x12", 8, 0, "固定外壳", "是"], + ["", "", "", "", "碳纤维板材", "CF-200/3K", 2, 0, "机身框架", "是"], + ["", "", "", "", "光学镜片", "OG-50", 1, 0, "激光发射", "是"], + ["UAV-X1", "V1.0", "无人机X1", "UAV-X1/Pro", "LICA-3000", "LICA-3000/A", 1, 0, "", "是"], + ["", "", "", "", "GPS模块", "GPS-M9N", 1, 0, "导航", "是"], +] + +BOM_TEMPLATE_INSTRUCTIONS = [ + "请勿修改表头行", "标记*的列为必填", + "同一个父件的多个子件,除第一行外,父件相关信息(BOM编号/版本/名称/规格)可留空,系统自动向下填充", + "父件/子件名称必须与系统中已存在的基础信息完全一致", + "版本默认 V1.0; 用量为正整数; 损耗率默认为0", + "同一BOM编号+版本下,父子件组合必须唯一", +] + + +def parse_boolean(val) -> bool: + """稳健布尔解析：是/Y/True/1 → True, 否/N/False/0/空 → False""" + if val is None: + return False + s = str(val).strip().lower() + return s in ('是', 'y', 'yes', 'true', '1', 't') + + +def generate_template(import_type: str) -> io.BytesIO: + """生成 Excel 模板文件""" + from openpyxl import Workbook + from openpyxl.styles import Font, PatternFill + + if import_type == 'material': + headers = MATERIAL_TEMPLATE_HEADERS + instructions = MATERIAL_TEMPLATE_INSTRUCTIONS + elif import_type == 'bom': + headers = BOM_TEMPLATE_HEADERS + instructions = BOM_TEMPLATE_INSTRUCTIONS + else: + raise ValueError(f"不支持的导入类型: {import_type}") + + wb = Workbook() + ws = wb.active + ws.title = f"{import_type}导入模板" + + # 表头样式 + header_font = Font(bold=True, size=11) + header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") + header_font_white = Font(bold=True, size=11, color="FFFFFF") + + ws.append(headers) + for cell in ws[1]: + cell.font = header_font_white + cell.fill = header_fill + + # 说明行 + for instr in instructions: + ws.append([instr] + [''] * (len(headers) - 1)) + + # 样例数据行（浅绿色背景区分） + example_fill = PatternFill(start_color="E2EFDA", end_color="E2EFDA", fill_type="solid") + example_rows = MATERIAL_EXAMPLE_ROWS if import_type == 'material' else BOM_EXAMPLE_ROWS + for row_data in example_rows: + ws.append(row_data) + for cell in ws[ws.max_row]: + cell.fill = example_fill + + # 例宽 + for col in ws.columns: + ws.column_dimensions[col[0].column_letter].width = 22 + + output = io.BytesIO() + wb.save(output) + output.seek(0) + return output + + +def preview_material(file_stream) -> list: + """Dry-Run 预览基础信息导入""" + import pandas as pd + + df = pd.read_excel(file_stream, header=0) + # 跳过说明行（第一列包含"请勿"的行） + df = df[~df.iloc[:, 0].astype(str).str.contains('请勿|标记|格式|默认')] + + col_map = { + "所属公司(company_name)": "company_name", + "名称(name)*": "name", + "专业名称(common_name)": "common_name", + "类别(category)*": "category", + "类型(material_type)*": "type", + "规格型号(spec_model)*": "spec", + "计量单位(unit)*": "unit", + "参考价格(reference_price)": "reference_price", + "强制质检(is_inspection_required)": "is_inspection_required", + "是否启用(is_enabled)": "is_enabled", + } + 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} + + 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 = 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 '' + unit = str(row.get('unit', '')).strip() if pd.notna(row.get('unit')) else '' + + if not name: + errors.append("名称不能为空") + if not spec: + errors.append("规格型号不能为空") + if not category: + errors.append("类别不能为空") + if not mat_type: + errors.append("类型不能为空") + if not unit: + errors.append("计量单位不能为空") + + if name and spec and (name, spec) in existing_set: + errors.append(f"已存在相同名称和规格的数据: {name} / {spec}") + + 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 '' + ref_price = row.get('reference_price') + ref_price = float(ref_price) if pd.notna(ref_price) and str(ref_price).strip() else None + is_enabled = parse_boolean(row.get('is_enabled', True)) + is_inspection = parse_boolean(row.get('is_inspection_required', False)) + + results.append({ + 'row': row_num, + 'status': 'error' if errors else 'success', + 'error_msg': '; '.join(errors) if errors else '', + 'data': { + 'company_name': company, + 'name': name, + 'common_name': common, + 'category': category, + 'material_type': mat_type, + 'spec_model': spec, + 'unit': unit, + 'reference_price': ref_price, + 'is_inspection_required': is_inspection, + 'is_enabled': is_enabled, + } + }) + + return results + + +def preview_bom(file_stream) -> list: + """Dry-Run 预览 BOM 表导入""" + import pandas as pd + + df = pd.read_excel(file_stream, header=0) + df = df[~df.iloc[:, 0].astype(str).str.contains('请勿|标记|格式|默认|同一|留空')] + + # ★ Forward Fill: 父件上下文列自动向下填充（支持合并单元格 UX） + ffill_cols = ["BOM编号(bom_no)*", "版本(version)", "父件名称(parent_name)*", "父件规格(parent_spec)"] + for c in ffill_cols: + if c in df.columns: + df[c] = df[c].ffill() + + col_map = { + "BOM编号(bom_no)*": "bom_no", + "版本(version)": "version", + "父件名称(parent_name)*": "parent_name", + "父件规格(parent_spec)": "parent_spec", + "子件名称(child_name)*": "child_name", + "子件规格(child_spec)": "child_spec", + "用量(dosage)*": "dosage", + "损耗率%(loss_rate)": "loss_rate", + "备注(remark)": "remark", + "是否启用(is_enabled)": "is_enabled", + } + df.rename(columns=col_map, inplace=True) + + # 批量预查所有 MaterialBase + 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 + + # 预查所有已存在的 (bom_no, version, parent_id, child_id) 组合 + all_bom_pairs = db.session.query( + BomTable.bom_no, BomTable.version, + BomTable.parent_id, BomTable.child_id + ).all() + existing_pairs = {(r.bom_no, r.version, r.parent_id, r.child_id) for r in all_bom_pairs} + + results = [] + for idx, row in df.iterrows(): + row_num = idx + 2 + errors = [] + + bom_no = str(row.get('bom_no', '')).strip() if pd.notna(row.get('bom_no')) else '' + version = str(row.get('version', 'V1.0')).strip() if pd.notna(row.get('version')) else 'V1.0' + parent_name = str(row.get('parent_name', '')).strip() if pd.notna(row.get('parent_name')) else '' + parent_spec = str(row.get('parent_spec', '')).strip() if pd.notna(row.get('parent_spec')) else '' + child_name = str(row.get('child_name', '')).strip() if pd.notna(row.get('child_name')) else '' + child_spec = str(row.get('child_spec', '')).strip() if pd.notna(row.get('child_spec')) else '' + dosage = row.get('dosage', 0) + + if not bom_no: + errors.append("BOM编号不能为空") + if not parent_name: + errors.append("父件名称不能为空") + if not child_name: + errors.append("子件名称不能为空") + 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 + if parent_name and parent_id is None: + errors.append(f"父件不存在: {parent_name} / {parent_spec}") + + child_key = (child_name, child_spec) + child_id = mat_lookup.get(child_key) if child_name else None + if child_name and child_id is None: + errors.append(f"子件不存在: {child_name} / {child_spec}") + + if parent_id and child_id and parent_id == child_id: + errors.append("父件与子件不能是同一物料") + + if bom_no and parent_id and child_id: + pair = (bom_no, version, parent_id, child_id) + if pair in existing_pairs: + errors.append(f"该BOM配方已存在: {bom_no} v{version}") + + dosage_val = float(dosage) if pd.notna(dosage) else 0 + loss_rate = float(row.get('loss_rate', 0)) if pd.notna(row.get('loss_rate')) else 0.0 + remark = str(row.get('remark', '')).strip() if pd.notna(row.get('remark')) else '' + is_enabled = parse_boolean(row.get('is_enabled', True)) + + results.append({ + 'row': row_num, + 'status': 'error' if errors else 'success', + 'error_msg': '; '.join(errors) if errors else '', + 'data': { + 'bom_no': bom_no, 'version': version, + 'parent_id': parent_id, 'parent_name': parent_name, 'parent_spec': parent_spec, + 'child_id': child_id, 'child_name': child_name, 'child_spec': child_spec, + 'dosage': dosage_val, 'loss_rate': loss_rate, + 'remark': remark, 'is_enabled': is_enabled, + } + }) + + return results + + +def execute_material_import(rows: list) -> dict: + """执行基础信息批量导入（事务保护）""" + inserted = 0 + errors = [] + + try: + for i, item in enumerate(rows): + d = item.get('data', item) + try: + existing = MaterialBase.query.filter_by( + name=d['name'], spec_model=d['spec_model'] + ).first() + if existing: + errors.append({'row': i + 2, 'msg': f"已存在: {d['name']} / {d['spec_model']}"}) + continue + + mat = MaterialBase( + company_name=d.get('company_name', ''), + name=d['name'], + common_name=d.get('common_name', ''), + category=d.get('category', ''), + material_type=d.get('material_type', ''), + spec_model=d.get('spec_model', ''), + unit=d.get('unit', ''), + reference_price=d.get('reference_price'), + is_inspection_required=d.get('is_inspection_required', False), + is_enabled=d.get('is_enabled', True), + ) + db.session.add(mat) + inserted += 1 + except Exception as e: + errors.append({'row': i + 2, 'msg': str(e)}) + + if errors: + db.session.rollback() + return {'success': False, 'inserted': 0, 'errors': errors, + 'msg': f'导入失败，{len(errors)} 条数据校验未通过'} + + db.session.commit() + return {'success': True, 'inserted': inserted, 'errors': [], + 'msg': f'成功导入 {inserted} 条基础信息'} + + except Exception as e: + db.session.rollback() + return {'success': False, 'inserted': 0, + 'errors': [{'row': 0, 'msg': str(e)}], 'msg': f'导入异常: {str(e)}'} + + +def execute_bom_import(rows: list) -> dict: + """执行 BOM 批量导入（事务保护）""" + # 预加载所有物料映射 + all_materials = db.session.query( + MaterialBase.id, MaterialBase.name, MaterialBase.spec_model + ).all() + mat_lookup = {} + for m in all_materials: + mat_lookup[(m.name, m.spec_model or '')] = m.id + + inserted = 0 + errors = [] + + try: + for i, item in enumerate(rows): + d = item.get('data', item) + try: + parent_id = d.get('parent_id') + child_id = d.get('child_id') + bom_no = d.get('bom_no', '') + 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) + if not child_id: + ck = (d.get('child_name', ''), d.get('child_spec', '')) + child_id = mat_lookup.get(ck) + + if not parent_id or not child_id: + errors.append({'row': i + 2, 'msg': '父件或子件在系统中不存在'}) + continue + + existing = BomTable.query.filter_by( + bom_no=bom_no, version=version, + parent_id=parent_id, child_id=child_id + ).first() + if existing: + errors.append({'row': i + 2, 'msg': f'BOM配方已存在: {bom_no} v{version}'}) + continue + + bom = BomTable( + bom_no=bom_no, version=version, + parent_id=parent_id, child_id=child_id, + dosage=d.get('dosage', 0), + loss_rate=d.get('loss_rate', 0), + remark=d.get('remark', ''), + is_enabled=d.get('is_enabled', True), + ) + db.session.add(bom) + inserted += 1 + except Exception as e: + errors.append({'row': i + 2, 'msg': str(e)}) + + if errors: + db.session.rollback() + return {'success': False, 'inserted': 0, 'errors': errors, + 'msg': f'导入失败，{len(errors)} 条数据校验未通过'} + + db.session.commit() + return {'success': True, 'inserted': inserted, 'errors': [], + 'msg': f'成功导入 {inserted} 条BOM配方'} + + except Exception as e: + db.session.rollback() + return {'success': False, 'inserted': 0, + 'errors': [{'row': 0, 'msg': str(e)}], 'msg': f'导入异常: {str(e)}'} diff --git a/inventory-backend/requirements.txt b/inventory-backend/requirements.txt index 7be60c8..234c87b 100644 --- a/inventory-backend/requirements.txt +++ b/inventory-backend/requirements.txt @@ -24,6 +24,8 @@ qrcode[pil]>=7.4.2 Flask-JWT-Extended==4.6.0 # [新增] Excel 处理库 (解决 No module named 'openpyxl' 报错) openpyxl>=3.1.2 +# [新增] 数据导入解析库 (Excel/CSV 批量导入) +pandas>=1.5.0 # [新增] 定时任务调度器 (库存预警每日邮件) APScheduler==3.10.4 # [新增] 时区处理 (APScheduler 需要)