asset-management/assetapp/excel_utils.py

"""Excel导入导出工具"""
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter
from django.utils import timezone
from .models import Category


# Excel列定义
EXPORT_COLUMNS = [
    ('id', 'ID', 8),
    ('asset_number', '资产编号', 18),
    ('name', '设备名称', 20),
    ('category', '设备分类', 12),
    ('brand', '品牌', 12),
    ('model', '型号', 20),
    ('asset_value', '资产面值', 12),
    ('serial_number', '序列号', 25),
    ('location', '机房', 20),
    ('cabinet', '机柜', 10),
    ('cabinet_position', '机柜位置', 10),
    ('bmc_address', 'BMC地址', 16),
    ('ip_address', 'IP地址', 16),
    ('gpu_type', '显卡类型', 15),
    ('gpu_count', '卡数', 6),
    ('purchase_date', '采购日期', 12),
    ('warranty_expire', '质保到期', 12),
    ('supplier', '供应商', 15),
    ('responsible_person', '负责人', 10),
    ('department', '使用部门', 15),
    ('user', '使用人', 10),
    ('business_type', '业务类型', 15),
    ('status', '状态', 8),
    ('remark', '备注', 30),
]

STATUS_MAP = {
    'in_use': '在用', 'idle': '闲置', 'maintenance': '维修中', 'scrapped': '已报废',
}
STATUS_MAP_REVERSE = {v: k for k, v in STATUS_MAP.items()}

# 深色主题样式
HEADER_FILL = PatternFill(start_color='1B2838', end_color='1B2838', fill_type='solid')
HEADER_FONT = Font(name='微软雅黑', bold=True, color='FFFFFF', size=11)
CELL_FONT = Font(name='微软雅黑', size=10)
THIN_BORDER = Border(
    left=Side(style='thin', color='3A4A5C'),
    right=Side(style='thin', color='3A4A5C'),
    top=Side(style='thin', color='3A4A5C'),
    bottom=Side(style='thin', color='3A4A5C'),
)


def export_assets_to_excel(queryset):
    """导出资产到Excel"""
    wb = Workbook()
    ws = wb.active
    ws.title = '硬件资产'

    # 写表头
    for col_idx, (field, header, width) in enumerate(EXPORT_COLUMNS, 1):
        cell = ws.cell(row=1, column=col_idx, value=header)
        cell.fill = HEADER_FILL
        cell.font = HEADER_FONT
        cell.alignment = Alignment(horizontal='center', vertical='center')
        cell.border = THIN_BORDER
        ws.column_dimensions[get_column_letter(col_idx)].width = width

    # 写数据
    for row_idx, asset in enumerate(queryset, 2):
        for col_idx, (field, _, _) in enumerate(EXPORT_COLUMNS, 1):
            if field == 'id':
                value = asset.id
            elif field == 'category':
                value = str(asset.category) if asset.category else ''
            elif field == 'status':
                value = STATUS_MAP.get(asset.status, asset.status)
            elif field in ('purchase_date', 'warranty_expire'):
                value = str(getattr(asset, field, '')) or ''
            else:
                value = getattr(asset, field, '') or ''

            cell = ws.cell(row=row_idx, column=col_idx, value=value)
            cell.font = CELL_FONT
            cell.border = THIN_BORDER
            cell.alignment = Alignment(vertical='center')

    # 冻结首行
    ws.freeze_panes = 'A2'

    return wb


def generate_import_template():
    """生成导入模板"""
    wb = Workbook()
    ws = wb.active
    ws.title = '资产导入模板'

    headers = [header for _, header, width in EXPORT_COLUMNS]
    for col_idx, (field, header, width) in enumerate(EXPORT_COLUMNS, 1):
        cell = ws.cell(row=1, column=col_idx, value=header)
        cell.fill = HEADER_FILL
        cell.font = HEADER_FONT
        cell.alignment = Alignment(horizontal='center', vertical='center')
        cell.border = THIN_BORDER
        ws.column_dimensions[get_column_letter(col_idx)].width = width

    # 示例数据行
    example_data = [
        '1', 'IT-2024-0001', '测试服务器', '服务器', 'Dell', 'PowerEdge R740',
        '50000.00', 'ABC123456', '3楼机房A区', 'A01', 'U10-U15', '192.168.1.200',
        '192.168.1.100', 'NVIDIA A100', '8', '2024-01-15', '2027-01-15', '戴尔科技',
        '张三', '研发部', '李四', 'AI训练', '在用', '测试备注'
    ]
    for col_idx, value in enumerate(example_data, 1):
        cell = ws.cell(row=2, column=col_idx, value=value)
        cell.font = Font(name='微软雅黑', size=10, color='666666')
        cell.border = THIN_BORDER

    ws.freeze_panes = 'A2'

    return wb


def import_assets_from_excel(ws, category_map, operator=None):
    """从Excel导入资产，返回结果统计"""
    from .models import Asset, AssetChangeLog

    header_row = [cell.value for cell in ws[1]]
    field_map = {header: field for field, header, _ in EXPORT_COLUMNS}

    results = {
        'success': 0,
        'skipped': 0,
        'errors': [],
        'total': 0,
    }

    for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), 2):
        if not row or not row[0]:
            continue

        results['total'] += 1

        try:
            data = {}
            for col_idx, value in enumerate(row):
                if col_idx < len(header_row):
                    field = field_map.get(header_row[col_idx])
                    if field:
                        data[field] = str(value).strip() if value else ''

            # 资产编号必填
            asset_number = data.get('asset_number', '').strip()
            if not asset_number:
                results['errors'].append(f'第{row_idx}行: 缺少资产编号')
                continue

            # 处理分类 - 不存在则自动创建
            category_name = data.get('category', '').strip()
            category = category_map.get(category_name)
            if not category and category_name:
                category = Category.objects.create(name=category_name)
                category_map[category_name] = category
            elif not category_name:
                category = category_map.get('未分类')
                if not category:
                    category = Category.objects.create(name='未分类')
                    category_map['未分类'] = category

            # 处理状态
            status = STATUS_MAP_REVERSE.get(data.get('status', '在用'), 'in_use')

            # 处理日期
            from datetime import datetime as dt
            purchase_date = None
            warranty_expire = None
            if data.get('purchase_date'):
                try:
                    purchase_date = dt.strptime(data['purchase_date'], '%Y-%m-%d').date()
                except ValueError:
                    pass
            if data.get('warranty_expire'):
                try:
                    warranty_expire = dt.strptime(data['warranty_expire'], '%Y-%m-%d').date()
                except ValueError:
                    pass

            # 处理IP
            bmc_address = data.get('bmc_address') or None
            ip_address = data.get('ip_address') or None

            # 处理资产面值
            from decimal import Decimal, InvalidOperation
            asset_value = None
            if data.get('asset_value'):
                try:
                    asset_value = Decimal(str(data['asset_value']).replace(',', ''))
                except (InvalidOperation, ValueError):
                    pass

            # 处理ID - 如果提供了ID且已存在，则更新该记录
            import_id = data.get('id', '').strip()
            asset = None
            is_update = False
            if import_id:
                try:
                    asset = Asset.objects.get(id=int(import_id))
                    is_update = True
                except (Asset.DoesNotExist, ValueError):
                    asset = None

            if asset:
                # 更新已有记录
                asset.asset_number = asset_number
                asset.name = data.get('name', '')
                asset.category = category
                asset.brand = data.get('brand', '')
                asset.model = data.get('model', '')
                asset.asset_value = asset_value
                asset.serial_number = data.get('serial_number', '')
                asset.location = data.get('location', '')
                asset.cabinet = data.get('cabinet', '')
                asset.cabinet_position = data.get('cabinet_position', '')
                asset.bmc_address = bmc_address
                asset.ip_address = ip_address
                asset.gpu_type = data.get('gpu_type', '')
                gpu_count_str = data.get('gpu_count', '').strip()
                asset.gpu_count = int(gpu_count_str) if gpu_count_str else None
                asset.purchase_date = purchase_date
                asset.warranty_expire = warranty_expire
                asset.supplier = data.get('supplier', '')
                asset.responsible_person = data.get('responsible_person', '')
                asset.department = data.get('department', '')
                asset.user = data.get('user', '')
                asset.business_type = data.get('business_type', '')
                asset.status = status
                asset.remark = data.get('remark', '')
                asset.save()

                AssetChangeLog.objects.create(
                    asset=asset,
                    asset_number=asset.asset_number,
                    action='import',
                    description=f'通过Excel导入更新(ID:{asset.id})',
                    operator=operator,
                )
            else:
                # 创建新记录
                gpu_count_str = data.get('gpu_count', '').strip()
                gpu_count = int(gpu_count_str) if gpu_count_str else None
                asset = Asset.objects.create(
                    asset_number=asset_number,
                    name=data.get('name', ''),
                    category=category,
                    brand=data.get('brand', ''),
                    model=data.get('model', ''),
                    asset_value=asset_value,
                    serial_number=data.get('serial_number', ''),
                    location=data.get('location', ''),
                    cabinet=data.get('cabinet', ''),
                    cabinet_position=data.get('cabinet_position', ''),
                    bmc_address=bmc_address,
                    ip_address=ip_address,
                    gpu_type=data.get('gpu_type', ''),
                    gpu_count=gpu_count,
                    purchase_date=purchase_date,
                    warranty_expire=warranty_expire,
                    supplier=data.get('supplier', ''),
                    responsible_person=data.get('responsible_person', ''),
                    department=data.get('department', ''),
                    user=data.get('user', ''),
                    business_type=data.get('business_type', ''),
                    status=status,
                    remark=data.get('remark', ''),
                    created_by=operator,
                )

                AssetChangeLog.objects.create(
                    asset=asset,
                    asset_number=asset.asset_number,
                    action='import',
                    description=f'通过Excel导入创建',
                    operator=operator,
                )

            results['success'] += 1

        except Exception as e:
            results['errors'].append(f'第{row_idx}行: {str(e)}')

    return results