KCGL/inventory-backend/app/services/import_service.py

"""
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, 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)
    # 跳过说明行（第一列包含"请勿"的行）
    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)

    # 批量预查已存在的 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 ''
        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("计量单位不能为空")

        # ★ 新增：字段长度校验（预防 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 ''
        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))

        # 决定行状态：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': row_status,
            'error_msg': '; '.join(errors) if errors else ('将被覆盖更新' if row_status == 'update' 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（仅依据 spec_model 查找）
    all_materials = db.session.query(
        MaterialBase.id, MaterialBase.name, MaterialBase.spec_model
    ).all()
    mat_lookup = {}
    for m in all_materials:
        if m.spec_model:
            mat_lookup[m.spec_model] = 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_id = mat_lookup.get(parent_spec) if parent_spec else None
        if parent_name and parent_id is None:
            errors.append(f"父件不存在 (规格型号): {parent_spec}")

        child_id = mat_lookup.get(child_spec) if child_spec else None
        if child_name and child_id is None:
            errors.append(f"子件不存在 (规格型号): {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, 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:
                # ★ 仅依据 spec_model（唯一货号）查找已有记录
                existing = MaterialBase.query.filter_by(spec_model=spec).first() if spec else None

                if existing:
                    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', ''),
                    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)})

        # ★ 改造：不再"一刀切"回滚，只提交成功的行
        # 防御：捕获 DB 级异常（字段超长、约束冲突等），优雅返回而非 500
        try:
            db.session.commit()
        except Exception as commit_err:
            db.session.rollback()
            return {
                'success': False,
                'inserted': 0,
                'updated': 0,
                'skipped': 0,
                'errors': [{'row': 0, 'msg': f'数据库写入失败: {str(commit_err)}'}],
                'msg': f'导入失败，数据库写入异常（可能字段超长或约束冲突）: {str(commit_err)}',
            }

        # 构建结果消息
        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, 'updated': 0, 'skipped': 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:
        if m.spec_model:
            mat_lookup[m.spec_model] = 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:
                    parent_spec = d.get('parent_spec', '')
                    parent_id = mat_lookup.get(parent_spec) if parent_spec else None
                if not child_id:
                    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': '父件或子件在系统中不存在'})
                    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)}'}