初始提交：希姆计算硬件资产管理系统
功能： - Django + MySQL + 深色主题 - 资产增删改查（含资产编号、BMC地址、设备位置、备注） - Excel导入导出（分类自动创建） - 设备分类管理 - 资产变更记录追踪 - 质保到期提醒 - 用户认证系统 - Docker部署支持
This commit is contained in:
cnbugs
@@ -0,0 +1,24 @@
|
||||
from django.contrib import admin
|
||||
from .models import Asset, Category, AssetChangeLog
|
||||
|
||||
|
||||
@admin.register(Category)
|
||||
class CategoryAdmin(admin.ModelAdmin):
|
||||
list_display = ['name', 'description', 'created_at']
|
||||
search_fields = ['name']
|
||||
|
||||
|
||||
@admin.register(Asset)
|
||||
class AssetAdmin(admin.ModelAdmin):
|
||||
list_display = ['asset_number', 'name', 'category', 'brand', 'model', 'status', 'location', 'created_at']
|
||||
list_filter = ['status', 'category', 'brand']
|
||||
search_fields = ['asset_number', 'name', 'serial_number', 'ip_address']
|
||||
readonly_fields = ['created_at', 'updated_at']
|
||||
|
||||
|
||||
@admin.register(AssetChangeLog)
|
||||
class AssetChangeLogAdmin(admin.ModelAdmin):
|
||||
list_display = ['asset_number', 'action', 'field_name', 'operator', 'created_at']
|
||||
list_filter = ['action']
|
||||
search_fields = ['asset_number']
|
||||
readonly_fields = ['created_at']
|
||||
@@ -0,0 +1,6 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class AssetappConfig(AppConfig):
|
||||
default_auto_field = 'django.db.models.BigAutoField'
|
||||
name = 'assetapp'
|
||||
@@ -0,0 +1,224 @@
|
||||
"""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 = [
|
||||
('asset_number', '资产编号', 18),
|
||||
('name', '设备名称', 20),
|
||||
('category', '设备分类', 12),
|
||||
('brand', '品牌', 12),
|
||||
('model', '型号', 20),
|
||||
('serial_number', '序列号', 25),
|
||||
('location', '设备位置', 20),
|
||||
('cabinet', '机柜', 10),
|
||||
('cabinet_position', '机柜位置', 10),
|
||||
('bmc_address', 'BMC地址', 16),
|
||||
('ip_address', 'IP地址', 16),
|
||||
('purchase_date', '采购日期', 12),
|
||||
('warranty_expire', '质保到期', 12),
|
||||
('supplier', '供应商', 15),
|
||||
('responsible_person', '负责人', 10),
|
||||
('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 == '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 = [
|
||||
'IT-2024-0001', '测试服务器', '服务器', 'Dell', 'PowerEdge R740',
|
||||
'ABC123456', '3楼机房A区', 'A01', 'U10-U15', '192.168.1.200',
|
||||
'192.168.1.100', '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
|
||||
|
||||
# 检查重复
|
||||
if Asset.objects.filter(asset_number=asset_number).exists():
|
||||
results['skipped'] += 1
|
||||
results['errors'].append(f'第{row_idx}行: 资产编号 {asset_number} 已存在')
|
||||
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
|
||||
|
||||
asset = Asset.objects.create(
|
||||
asset_number=asset_number,
|
||||
name=data.get('name', ''),
|
||||
category=category,
|
||||
brand=data.get('brand', ''),
|
||||
model=data.get('model', ''),
|
||||
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,
|
||||
purchase_date=purchase_date,
|
||||
warranty_expire=warranty_expire,
|
||||
supplier=data.get('supplier', ''),
|
||||
responsible_person=data.get('responsible_person', ''),
|
||||
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
|
||||
@@ -0,0 +1,34 @@
|
||||
from django import forms
|
||||
from .models import Asset, Category
|
||||
|
||||
|
||||
class AssetForm(forms.ModelForm):
|
||||
class Meta:
|
||||
model = Asset
|
||||
fields = [
|
||||
'asset_number', 'name', 'category', 'brand', 'model', 'serial_number',
|
||||
'location', 'cabinet', 'cabinet_position', 'bmc_address', 'ip_address',
|
||||
'purchase_date', 'warranty_expire', 'supplier',
|
||||
'responsible_person', 'status', 'remark',
|
||||
]
|
||||
widgets = {
|
||||
'purchase_date': forms.DateInput(attrs={'type': 'date'}),
|
||||
'warranty_expire': forms.DateInput(attrs={'type': 'date'}),
|
||||
'remark': forms.Textarea(attrs={'rows': 3}),
|
||||
'status': forms.Select(attrs={'class': 'form-select'}),
|
||||
'category': forms.Select(attrs={'class': 'form-select'}),
|
||||
}
|
||||
|
||||
|
||||
class AssetImportForm(forms.Form):
|
||||
excel_file = forms.FileField(
|
||||
label='Excel文件',
|
||||
help_text='支持 .xlsx 格式',
|
||||
widget=forms.FileInput(attrs={'accept': '.xlsx,.xls'})
|
||||
)
|
||||
|
||||
|
||||
class CategoryForm(forms.ModelForm):
|
||||
class Meta:
|
||||
model = Category
|
||||
fields = ['name', 'description']
|
||||
@@ -0,0 +1,38 @@
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.contrib.auth.models import User
|
||||
from assetapp.models import Category
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = '初始化系统：创建管理员用户和默认分类'
|
||||
|
||||
def handle(self, *args, **options):
|
||||
# 创建管理员
|
||||
if not User.objects.filter(username='admin').exists():
|
||||
User.objects.create_superuser('admin', 'admin@example.com', 'admin123')
|
||||
self.stdout.write(self.style.SUCCESS('✓ 管理员用户已创建 (admin/admin123)'))
|
||||
else:
|
||||
self.stdout.write(' 管理员用户已存在，跳过')
|
||||
|
||||
# 创建默认分类
|
||||
default_categories = [
|
||||
('服务器', '各类物理服务器、虚拟化主机'),
|
||||
('交换机', '网络交换设备'),
|
||||
('路由器', '网络路由设备'),
|
||||
('存储设备', 'SAN/NAS等存储设备'),
|
||||
('防火墙', '网络安全设备'),
|
||||
('UPS', '不间断电源设备'),
|
||||
('其他', '其他类型设备'),
|
||||
]
|
||||
|
||||
created_count = 0
|
||||
for name, desc in default_categories:
|
||||
cat, created = Category.objects.get_or_create(
|
||||
name=name,
|
||||
defaults={'description': desc}
|
||||
)
|
||||
if created:
|
||||
created_count += 1
|
||||
|
||||
self.stdout.write(self.style.SUCCESS(f'✓ 分类初始化完成 (新增 {created_count} 个)'))
|
||||
self.stdout.write(self.style.SUCCESS('\n初始化完成！请使用 admin/admin123 登录'))
|
||||
@@ -0,0 +1,81 @@
|
||||
# Generated by Django 5.2.13 on 2026-04-24 10:33
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Category',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=50, unique=True, verbose_name='分类名称')),
|
||||
('description', models.CharField(blank=True, default='', max_length=200, verbose_name='描述')),
|
||||
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '设备分类',
|
||||
'verbose_name_plural': '设备分类',
|
||||
'ordering': ['name'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='Asset',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('asset_number', models.CharField(db_index=True, max_length=50, unique=True, verbose_name='资产编号')),
|
||||
('name', models.CharField(max_length=100, verbose_name='设备名称')),
|
||||
('brand', models.CharField(blank=True, default='', max_length=50, verbose_name='品牌')),
|
||||
('model', models.CharField(blank=True, default='', max_length=100, verbose_name='型号')),
|
||||
('serial_number', models.CharField(blank=True, db_index=True, default='', max_length=100, verbose_name='序列号')),
|
||||
('location', models.CharField(blank=True, default='', help_text='楼层/房间/区域', max_length=200, verbose_name='设备位置')),
|
||||
('cabinet', models.CharField(blank=True, default='', max_length=50, verbose_name='机柜')),
|
||||
('cabinet_position', models.CharField(blank=True, default='', help_text='U位', max_length=50, verbose_name='机柜位置')),
|
||||
('ip_address', models.GenericIPAddressField(blank=True, null=True, verbose_name='IP地址')),
|
||||
('purchase_date', models.DateField(blank=True, null=True, verbose_name='采购日期')),
|
||||
('warranty_expire', models.DateField(blank=True, null=True, verbose_name='质保到期')),
|
||||
('supplier', models.CharField(blank=True, default='', max_length=100, verbose_name='供应商')),
|
||||
('responsible_person', models.CharField(blank=True, default='', max_length=50, verbose_name='负责人')),
|
||||
('status', models.CharField(choices=[('in_use', '在用'), ('idle', '闲置'), ('maintenance', '维修中'), ('scrapped', '已报废')], default='in_use', max_length=20, verbose_name='状态')),
|
||||
('remark', models.TextField(blank=True, default='', verbose_name='备注')),
|
||||
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
|
||||
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
|
||||
('created_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL, verbose_name='创建人')),
|
||||
('category', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='assets', to='assetapp.category', verbose_name='设备分类')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '硬件资产',
|
||||
'verbose_name_plural': '硬件资产',
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='AssetChangeLog',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('asset_number', models.CharField(db_index=True, max_length=50, verbose_name='资产编号')),
|
||||
('action', models.CharField(choices=[('create', '创建'), ('update', '更新'), ('delete', '删除'), ('import', '导入'), ('export', '导出'), ('status_change', '状态变更')], max_length=20, verbose_name='操作类型')),
|
||||
('field_name', models.CharField(blank=True, default='', max_length=50, verbose_name='变更字段')),
|
||||
('old_value', models.TextField(blank=True, default='', verbose_name='旧值')),
|
||||
('new_value', models.TextField(blank=True, default='', verbose_name='新值')),
|
||||
('description', models.TextField(blank=True, default='', verbose_name='描述')),
|
||||
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='操作时间')),
|
||||
('asset', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='change_logs', to='assetapp.asset', verbose_name='资产')),
|
||||
('operator', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL, verbose_name='操作人')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '变更记录',
|
||||
'verbose_name_plural': '变更记录',
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
# Generated by Django 5.2.13 on 2026-04-24 23:24
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('assetapp', '0001_initial'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='asset',
|
||||
name='bmc_address',
|
||||
field=models.GenericIPAddressField(blank=True, null=True, verbose_name='BMC地址'),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,116 @@
|
||||
from django.db import models
|
||||
from django.contrib.auth.models import User
|
||||
from datetime import date
|
||||
|
||||
|
||||
class Category(models.Model):
|
||||
"""设备分类"""
|
||||
name = models.CharField('分类名称', max_length=50, unique=True)
|
||||
description = models.CharField('描述', max_length=200, blank=True, default='')
|
||||
created_at = models.DateTimeField('创建时间', auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = '设备分类'
|
||||
verbose_name_plural = verbose_name
|
||||
ordering = ['name']
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
class Asset(models.Model):
|
||||
"""硬件资产"""
|
||||
|
||||
STATUS_CHOICES = [
|
||||
('in_use', '在用'),
|
||||
('idle', '闲置'),
|
||||
('maintenance', '维修中'),
|
||||
('scrapped', '已报废'),
|
||||
]
|
||||
|
||||
# 基本信息
|
||||
asset_number = models.CharField('资产编号', max_length=50, unique=True, db_index=True)
|
||||
name = models.CharField('设备名称', max_length=100)
|
||||
category = models.ForeignKey(Category, on_delete=models.PROTECT, verbose_name='设备分类', related_name='assets')
|
||||
|
||||
# 硬件信息
|
||||
brand = models.CharField('品牌', max_length=50, blank=True, default='')
|
||||
model = models.CharField('型号', max_length=100, blank=True, default='')
|
||||
serial_number = models.CharField('序列号', max_length=100, blank=True, default='', db_index=True)
|
||||
|
||||
# 位置信息
|
||||
location = models.CharField('设备位置', max_length=200, blank=True, default='', help_text='楼层/房间/区域')
|
||||
cabinet = models.CharField('机柜', max_length=50, blank=True, default='')
|
||||
cabinet_position = models.CharField('机柜位置', max_length=50, blank=True, default='', help_text='U位')
|
||||
|
||||
# 网络信息
|
||||
bmc_address = models.GenericIPAddressField('BMC地址', blank=True, null=True)
|
||||
ip_address = models.GenericIPAddressField('IP地址', blank=True, null=True)
|
||||
|
||||
# 采购与质保
|
||||
purchase_date = models.DateField('采购日期', blank=True, null=True)
|
||||
warranty_expire = models.DateField('质保到期', blank=True, null=True)
|
||||
supplier = models.CharField('供应商', max_length=100, blank=True, default='')
|
||||
|
||||
# 管理信息
|
||||
responsible_person = models.CharField('负责人', max_length=50, blank=True, default='')
|
||||
status = models.CharField('状态', max_length=20, choices=STATUS_CHOICES, default='in_use')
|
||||
remark = models.TextField('备注', blank=True, default='')
|
||||
|
||||
# 系统字段
|
||||
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, verbose_name='创建人')
|
||||
created_at = models.DateTimeField('创建时间', auto_now_add=True)
|
||||
updated_at = models.DateTimeField('更新时间', auto_now=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = '硬件资产'
|
||||
verbose_name_plural = verbose_name
|
||||
ordering = ['-created_at']
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.asset_number} - {self.name}'
|
||||
|
||||
@property
|
||||
def is_expired(self):
|
||||
if self.warranty_expire:
|
||||
return self.warranty_expire < date.today()
|
||||
return False
|
||||
|
||||
@property
|
||||
def is_expiring_soon(self):
|
||||
if self.warranty_expire:
|
||||
from datetime import timedelta
|
||||
return date.today() <= self.warranty_expire <= date.today() + timedelta(days=30)
|
||||
return False
|
||||
|
||||
|
||||
class AssetChangeLog(models.Model):
|
||||
"""资产变更记录"""
|
||||
|
||||
ACTION_CHOICES = [
|
||||
('create', '创建'),
|
||||
('update', '更新'),
|
||||
('delete', '删除'),
|
||||
('import', '导入'),
|
||||
('export', '导出'),
|
||||
('status_change', '状态变更'),
|
||||
]
|
||||
|
||||
asset = models.ForeignKey(Asset, on_delete=models.SET_NULL, null=True, blank=True,
|
||||
verbose_name='资产', related_name='change_logs')
|
||||
asset_number = models.CharField('资产编号', max_length=50, db_index=True)
|
||||
action = models.CharField('操作类型', max_length=20, choices=ACTION_CHOICES)
|
||||
field_name = models.CharField('变更字段', max_length=50, blank=True, default='')
|
||||
old_value = models.TextField('旧值', blank=True, default='')
|
||||
new_value = models.TextField('新值', blank=True, default='')
|
||||
description = models.TextField('描述', blank=True, default='')
|
||||
operator = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, verbose_name='操作人')
|
||||
created_at = models.DateTimeField('操作时间', auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = '变更记录'
|
||||
verbose_name_plural = verbose_name
|
||||
ordering = ['-created_at']
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.asset_number} - {self.get_action_display()}'
|
||||
@@ -0,0 +1,9 @@
|
||||
from django import template
|
||||
|
||||
register = template.Library()
|
||||
|
||||
@register.filter
|
||||
def get_item(dictionary, key):
|
||||
if isinstance(dictionary, dict):
|
||||
return dictionary.get(key, 0)
|
||||
return 0
|
||||
@@ -0,0 +1,3 @@
|
||||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
||||
@@ -0,0 +1,32 @@
|
||||
from django.urls import path
|
||||
from . import views
|
||||
|
||||
urlpatterns = [
|
||||
# 认证
|
||||
path('login/', views.login_view, name='login'),
|
||||
path('logout/', views.logout_view, name='logout'),
|
||||
|
||||
# 仪表盘
|
||||
path('', views.dashboard, name='dashboard'),
|
||||
|
||||
# 资产CRUD
|
||||
path('assets/', views.asset_list, name='asset_list'),
|
||||
path('assets/create/', views.asset_create, name='asset_create'),
|
||||
path('assets/<int:pk>/', views.asset_detail, name='asset_detail'),
|
||||
path('assets/<int:pk>/edit/', views.asset_update, name='asset_update'),
|
||||
path('assets/<int:pk>/delete/', views.asset_delete, name='asset_delete'),
|
||||
|
||||
# Excel导入导出
|
||||
path('assets/export/', views.asset_export, name='asset_export'),
|
||||
path('assets/import/', views.asset_import, name='asset_import'),
|
||||
path('assets/import/template/', views.download_template, name='download_template'),
|
||||
|
||||
# 变更记录
|
||||
path('changelog/', views.change_log_list, name='change_log_list'),
|
||||
|
||||
# 分类管理
|
||||
path('categories/', views.category_list, name='category_list'),
|
||||
path('categories/create/', views.category_create, name='category_create'),
|
||||
path('categories/<int:pk>/edit/', views.category_update, name='category_update'),
|
||||
path('categories/<int:pk>/delete/', views.category_delete, name='category_delete'),
|
||||
]
|
||||
@@ -0,0 +1,416 @@
|
||||
from django.shortcuts import render, redirect, get_object_or_404
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.contrib.auth import authenticate, login, logout
|
||||
from django.contrib import messages
|
||||
from django.http import HttpResponse, JsonResponse
|
||||
from django.db.models import Count, Q
|
||||
from django.core.paginator import Paginator
|
||||
from django.conf import settings
|
||||
from openpyxl import load_workbook
|
||||
from io import BytesIO
|
||||
from datetime import date, timedelta
|
||||
|
||||
from .models import Asset, Category, AssetChangeLog
|
||||
from .forms import AssetForm, AssetImportForm, CategoryForm
|
||||
from .excel_utils import (
|
||||
export_assets_to_excel, generate_import_template,
|
||||
import_assets_from_excel, STATUS_MAP,
|
||||
)
|
||||
|
||||
|
||||
# ─── 认证 ────────────────────────────────────────
|
||||
|
||||
def login_view(request):
|
||||
if request.user.is_authenticated:
|
||||
return redirect('dashboard')
|
||||
|
||||
if request.method == 'POST':
|
||||
username = request.POST.get('username', '')
|
||||
password = request.POST.get('password', '')
|
||||
user = authenticate(request, username=username, password=password)
|
||||
if user is not None:
|
||||
login(request, user)
|
||||
next_url = request.GET.get('next', '/')
|
||||
return redirect(next_url)
|
||||
else:
|
||||
messages.error(request, '用户名或密码错误')
|
||||
|
||||
return render(request, 'assetapp/login.html')
|
||||
|
||||
|
||||
def logout_view(request):
|
||||
logout(request)
|
||||
return redirect('login')
|
||||
|
||||
|
||||
# ─── 仪表盘 ──────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def dashboard(request):
|
||||
total_assets = Asset.objects.count()
|
||||
status_stats = Asset.objects.values('status').annotate(count=Count('id'))
|
||||
status_data = {item['status']: item['count'] for item in status_stats}
|
||||
|
||||
category_stats = Asset.objects.values('category__name').annotate(
|
||||
count=Count('id')
|
||||
).order_by('-count')
|
||||
|
||||
# 即将过保（30天内）
|
||||
today = date.today()
|
||||
expiring_soon = Asset.objects.filter(
|
||||
warranty_expire__lte=today + timedelta(days=30),
|
||||
warranty_expire__gte=today,
|
||||
status='in_use',
|
||||
).count()
|
||||
expired = Asset.objects.filter(
|
||||
warranty_expire__lt=today,
|
||||
status='in_use',
|
||||
).count()
|
||||
|
||||
# 最近变更
|
||||
recent_changes = AssetChangeLog.objects.select_related('operator')[:10]
|
||||
|
||||
context = {
|
||||
'total_assets': total_assets,
|
||||
'status_data': status_data,
|
||||
'status_map': STATUS_MAP,
|
||||
'category_stats': category_stats,
|
||||
'expiring_soon': expiring_soon,
|
||||
'expired': expired,
|
||||
'recent_changes': recent_changes,
|
||||
}
|
||||
return render(request, 'assetapp/dashboard.html', context)
|
||||
|
||||
|
||||
# ─── 资产列表 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_list(request):
|
||||
queryset = Asset.objects.select_related('category')
|
||||
|
||||
# 搜索
|
||||
search = request.GET.get('search', '').strip()
|
||||
if search:
|
||||
queryset = queryset.filter(
|
||||
Q(asset_number__icontains=search) |
|
||||
Q(name__icontains=search) |
|
||||
Q(serial_number__icontains=search) |
|
||||
Q(ip_address__icontains=search) |
|
||||
Q(brand__icontains=search) |
|
||||
Q(model__icontains=search) |
|
||||
Q(location__icontains=search) |
|
||||
Q(responsible_person__icontains=search)
|
||||
)
|
||||
|
||||
# 筛选
|
||||
category_id = request.GET.get('category')
|
||||
if category_id:
|
||||
queryset = queryset.filter(category_id=category_id)
|
||||
|
||||
status = request.GET.get('status')
|
||||
if status:
|
||||
queryset = queryset.filter(status=status)
|
||||
|
||||
location = request.GET.get('location')
|
||||
if location:
|
||||
queryset = queryset.filter(location__icontains=location)
|
||||
|
||||
# 排序
|
||||
sort = request.GET.get('sort', '-created_at')
|
||||
valid_sorts = ['asset_number', '-asset_number', 'name', '-name',
|
||||
'created_at', '-created_at', 'updated_at', '-updated_at',
|
||||
'purchase_date', '-purchase_date']
|
||||
if sort not in valid_sorts:
|
||||
sort = '-created_at'
|
||||
queryset = queryset.order_by(sort)
|
||||
|
||||
# 分页
|
||||
paginator = Paginator(queryset, settings.ASSETS_PER_PAGE)
|
||||
page_number = request.GET.get('page', 1)
|
||||
page_obj = paginator.get_page(page_number)
|
||||
|
||||
categories = Category.objects.all()
|
||||
locations = Asset.objects.values_list('location', flat=True).exclude(location='').distinct()
|
||||
|
||||
context = {
|
||||
'page_obj': page_obj,
|
||||
'search': search,
|
||||
'categories': categories,
|
||||
'current_category': category_id,
|
||||
'current_status': status,
|
||||
'current_location': location,
|
||||
'locations': locations,
|
||||
'status_map': STATUS_MAP,
|
||||
'sort': sort,
|
||||
}
|
||||
return render(request, 'assetapp/asset_list.html', context)
|
||||
|
||||
|
||||
# ─── 资产详情 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_detail(request, pk):
|
||||
asset = get_object_or_404(Asset.objects.select_related('category', 'created_by'), pk=pk)
|
||||
change_logs = asset.change_logs.select_related('operator')[:20]
|
||||
context = {
|
||||
'asset': asset,
|
||||
'change_logs': change_logs,
|
||||
'status_map': STATUS_MAP,
|
||||
}
|
||||
return render(request, 'assetapp/asset_detail.html', context)
|
||||
|
||||
|
||||
# ─── 资产创建 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_create(request):
|
||||
if request.method == 'POST':
|
||||
form = AssetForm(request.POST)
|
||||
if form.is_valid():
|
||||
asset = form.save(commit=False)
|
||||
asset.created_by = request.user
|
||||
asset.save()
|
||||
|
||||
AssetChangeLog.objects.create(
|
||||
asset=asset,
|
||||
asset_number=asset.asset_number,
|
||||
action='create',
|
||||
description='创建资产',
|
||||
operator=request.user,
|
||||
)
|
||||
|
||||
messages.success(request, f'资产 {asset.asset_number} 创建成功！')
|
||||
return redirect('asset_detail', pk=asset.pk)
|
||||
else:
|
||||
form = AssetForm()
|
||||
|
||||
context = {'form': form, 'action': 'create'}
|
||||
return render(request, 'assetapp/asset_form.html', context)
|
||||
|
||||
|
||||
# ─── 资产编辑 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_update(request, pk):
|
||||
asset = get_object_or_404(Asset, pk=pk)
|
||||
|
||||
if request.method == 'POST':
|
||||
form = AssetForm(request.POST, instance=asset)
|
||||
if form.is_valid():
|
||||
# 记录变更
|
||||
changes = []
|
||||
for field in form.changed_data:
|
||||
old_val = str(form.initial.get(field, ''))
|
||||
new_val = str(form.cleaned_data.get(field, ''))
|
||||
changes.append(f'{field}: {old_val} → {new_val}')
|
||||
|
||||
AssetChangeLog.objects.create(
|
||||
asset=asset,
|
||||
asset_number=asset.asset_number,
|
||||
action='update',
|
||||
field_name=field,
|
||||
old_value=str(form.initial.get(field, '')),
|
||||
new_value=str(form.cleaned_data.get(field, '')),
|
||||
operator=request.user,
|
||||
)
|
||||
|
||||
asset = form.save()
|
||||
|
||||
if changes:
|
||||
messages.success(request, f'已更新 {len(changes)} 个字段')
|
||||
return redirect('asset_detail', pk=asset.pk)
|
||||
else:
|
||||
form = AssetForm(instance=asset)
|
||||
|
||||
context = {'form': form, 'action': 'update', 'asset': asset}
|
||||
return render(request, 'assetapp/asset_form.html', context)
|
||||
|
||||
|
||||
# ─── 资产删除 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_delete(request, pk):
|
||||
asset = get_object_or_404(Asset, pk=pk)
|
||||
|
||||
if request.method == 'POST':
|
||||
AssetChangeLog.objects.create(
|
||||
asset_number=asset.asset_number,
|
||||
action='delete',
|
||||
description=f'删除资产: {asset.name}',
|
||||
operator=request.user,
|
||||
)
|
||||
asset.delete()
|
||||
messages.success(request, f'资产 {asset.asset_number} 已删除')
|
||||
return redirect('asset_list')
|
||||
|
||||
context = {'asset': asset}
|
||||
return render(request, 'assetapp/asset_confirm_delete.html', context)
|
||||
|
||||
|
||||
# ─── Excel导出 ────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_export(request):
|
||||
queryset = Asset.objects.select_related('category')
|
||||
|
||||
# 支持筛选导出
|
||||
category_id = request.GET.get('category')
|
||||
if category_id:
|
||||
queryset = queryset.filter(category_id=category_id)
|
||||
|
||||
status = request.GET.get('status')
|
||||
if status:
|
||||
queryset = queryset.filter(status=status)
|
||||
|
||||
wb = export_assets_to_excel(queryset)
|
||||
|
||||
output = BytesIO()
|
||||
wb.save(output)
|
||||
output.seek(0)
|
||||
|
||||
AssetChangeLog.objects.create(
|
||||
asset_number='-',
|
||||
action='export',
|
||||
description=f'导出 {queryset.count()} 条资产记录',
|
||||
operator=request.user,
|
||||
)
|
||||
|
||||
today_str = date.today().strftime('%Y%m%d')
|
||||
response = HttpResponse(
|
||||
output.read(),
|
||||
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
||||
)
|
||||
response['Content-Disposition'] = f'attachment; filename=硬件资产_{today_str}.xlsx'
|
||||
return response
|
||||
|
||||
|
||||
# ─── Excel导入 ────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def asset_import(request):
|
||||
if request.method == 'POST':
|
||||
form = AssetImportForm(request.POST, request.FILES)
|
||||
if form.is_valid():
|
||||
try:
|
||||
wb = load_workbook(request.FILES['excel_file'])
|
||||
ws = wb.active
|
||||
|
||||
category_map = {c.name: c for c in Category.objects.all()}
|
||||
results = import_assets_from_excel(ws, category_map, operator=request.user)
|
||||
|
||||
if results['success'] > 0:
|
||||
messages.success(request, f"成功导入 {results['success']} 条资产")
|
||||
|
||||
if results['skipped'] > 0:
|
||||
messages.warning(request, f"跳过 {results['skipped']} 条（资产编号重复）")
|
||||
|
||||
if results['errors']:
|
||||
for error in results['errors'][:10]:
|
||||
messages.error(request, error)
|
||||
if len(results['errors']) > 10:
|
||||
messages.error(request, f"...还有 {len(results['errors']) - 10} 条错误")
|
||||
|
||||
except Exception as e:
|
||||
messages.error(request, f'导入失败: {str(e)}')
|
||||
|
||||
return redirect('asset_list')
|
||||
else:
|
||||
form = AssetImportForm()
|
||||
|
||||
return render(request, 'assetapp/asset_import.html', {'form': form})
|
||||
|
||||
|
||||
# ─── 下载导入模板 ──────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def download_template(request):
|
||||
wb = generate_import_template()
|
||||
output = BytesIO()
|
||||
wb.save(output)
|
||||
output.seek(0)
|
||||
|
||||
response = HttpResponse(
|
||||
output.read(),
|
||||
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
||||
)
|
||||
response['Content-Disposition'] = 'attachment; filename=资产导入模板.xlsx'
|
||||
return response
|
||||
|
||||
|
||||
# ─── 变更记录 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def change_log_list(request):
|
||||
logs = AssetChangeLog.objects.select_related('operator', 'asset')
|
||||
|
||||
asset_number = request.GET.get('asset_number', '').strip()
|
||||
if asset_number:
|
||||
logs = logs.filter(asset_number__icontains=asset_number)
|
||||
|
||||
action = request.GET.get('action')
|
||||
if action:
|
||||
logs = logs.filter(action=action)
|
||||
|
||||
paginator = Paginator(logs, 30)
|
||||
page_number = request.GET.get('page', 1)
|
||||
page_obj = paginator.get_page(page_number)
|
||||
|
||||
context = {
|
||||
'page_obj': page_obj,
|
||||
'asset_number': asset_number,
|
||||
'current_action': action,
|
||||
}
|
||||
return render(request, 'assetapp/changelog.html', context)
|
||||
|
||||
|
||||
# ─── 分类管理 ─────────────────────────────────────
|
||||
|
||||
@login_required
|
||||
def category_list(request):
|
||||
categories = Category.objects.annotate(asset_count=Count('assets')).order_by('name')
|
||||
return render(request, 'assetapp/category_list.html', {'categories': categories})
|
||||
|
||||
|
||||
@login_required
|
||||
def category_create(request):
|
||||
if request.method == 'POST':
|
||||
form = CategoryForm(request.POST)
|
||||
if form.is_valid():
|
||||
form.save()
|
||||
messages.success(request, f'分类 "{form.instance.name}" 创建成功')
|
||||
return redirect('category_list')
|
||||
else:
|
||||
form = CategoryForm()
|
||||
|
||||
return render(request, 'assetapp/category_form.html', {'form': form, 'action': 'create'})
|
||||
|
||||
|
||||
@login_required
|
||||
def category_update(request, pk):
|
||||
category = get_object_or_404(Category, pk=pk)
|
||||
if request.method == 'POST':
|
||||
form = CategoryForm(request.POST, instance=category)
|
||||
if form.is_valid():
|
||||
form.save()
|
||||
messages.success(request, f'分类 "{category.name}" 已更新')
|
||||
return redirect('category_list')
|
||||
else:
|
||||
form = CategoryForm(instance=category)
|
||||
|
||||
return render(request, 'assetapp/category_form.html', {'form': form, 'action': 'update', 'category': category})
|
||||
|
||||
|
||||
@login_required
|
||||
def category_delete(request, pk):
|
||||
category = get_object_or_404(Category, pk=pk)
|
||||
asset_count = category.assets.count()
|
||||
if asset_count > 0:
|
||||
messages.error(request, f'分类 "{category.name}" 下还有 {asset_count} 个资产，无法删除')
|
||||
return redirect('category_list')
|
||||
|
||||
if request.method == 'POST':
|
||||
category.delete()
|
||||
messages.success(request, f'分类 "{category.name}" 已删除')
|
||||
return redirect('category_list')
|
||||
|
||||
return render(request, 'assetapp/category_confirm_delete.html', {'category': category})
|
||||
Reference in New Issue
Block a user