"""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), ('location', '机房', 20), ('cabinet', '机柜', 10), ('cabinet_position', '机柜位置', 10), ('asset_number', '资产编号', 18), ('name', '设备名称', 20), ('category', '设备分类', 12), ('brand', '品牌', 12), ('model', '型号', 20), ('asset_value', '资产面值', 12), ('serial_number', '序列号', 25), ('bmc_address', 'BMC地址', 16), ('ip_address', 'IP地址', 16), ('gpu_type', '显卡类型', 15), ('gpu_count', '卡数', 6), ('responsible_person', '负责人', 10), ('department', '使用部门', 15), ('user', '维护人', 10), ('business_type', '业务类型', 15), ('status', '状态', 8), ('purchase_date', '采购日期', 12), ('warranty_expire', '质保到期', 12), ('supplier', '供应商', 15), ('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 # 示例数据行（顺序与EXPORT_COLUMNS一致） example_data = [ '1', '3楼机房A区', 'A01', 'U10-U15', 'IT-2024-0001', '测试服务器', '服务器', 'Dell', 'PowerEdge R740', '50000.00', 'ABC123456', '192.168.1.200', '192.168.1.100', 'NVIDIA A100', '8', '张三', '研发部', '李四', 'AI训练', '在用', '2024-01-15', '2027-01-15', '戴尔科技', '测试备注' ] 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 # 创建参数 create_kwargs = { '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, } # 如果Excel提供了ID，使用该ID创建 if import_id: try: create_kwargs['id'] = int(import_id) except ValueError: pass asset = Asset.objects.create(**create_kwargs) 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