cnbugs
294 righe
11 KiB
Python
294 righe
11 KiB
Python
"""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
|