xiaji/netmeeting-infra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: 视频主设备管理系统 - 完整DRF后端项目

Browse Source
This commit is contained in:
xiaji
2026-05-18 17:27:59 +08:00
commit c1565169cd
18 changed files with 992 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
.gitignore vendored Normal file
View File

@@ -0,0 +1,18 @@
venv/
env/
*.pyc
__pycache__/
*.pyo
*.pyd
.Python
*.so
media/
static/
db.sqlite3
*.log
.env
.idea/
.vscode/
*.swp
*.swo
*~

0
config/__init__.py Normal file
View File

16
config/asgi.py Normal file
View File

@@ -0,0 +1,16 @@
"""
ASGI config for config project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/5.0/howto/deployment/asgi/
"""
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
application = get_asgi_application()

140
config/settings.py Normal file
View File

@@ -0,0 +1,140 @@
"""
Django settings for config project.
Generated by 'django-admin startproject' using Django 5.0.6.
For more information on this file, see
https://docs.djangoproject.com/en/5.0/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/5.0/ref/settings/
"""
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_KEY = 'django-insecure-$6qm&o2(b=hi7snx-dvgcbvbd9p6y%hsk-@=ddv593%r(f)l%%'
DEBUG = True
ALLOWED_HOSTS = ['*']
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'rest_framework',
'rest_framework_simplejwt',
'django_filters',
'drf_yasg',
'device_management',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'config.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'config.wsgi.application'
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': BASE_DIR / 'db.sqlite3',
}
}
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_TZ = True
STATIC_URL = 'static/'
STATIC_ROOT = BASE_DIR / 'static'
MEDIA_URL = 'media/'
MEDIA_ROOT = BASE_DIR / 'media'
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_simplejwt.authentication.JWTAuthentication',
'rest_framework.authentication.SessionAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticatedOrReadOnly',
],
'DEFAULT_FILTER_BACKENDS': [
'django_filters.rest_framework.DjangoFilterBackend',
'rest_framework.filters.SearchFilter',
'rest_framework.filters.OrderingFilter',
],
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 20,
}
SWAGGER_SETTINGS = {
'SECURITY_DEFINITIONS': {
'Bearer': {
'type': 'apiKey',
'name': 'Authorization',
'in': 'header'
}
},
'USE_SESSION_AUTH': True,
}
from datetime import timedelta
SIMPLE_JWT = {
'ACCESS_TOKEN_LIFETIME': timedelta(days=1),
'REFRESH_TOKEN_LIFETIME': timedelta(days=7),
'ROTATE_REFRESH_TOKENS': False,
'BLACKLIST_AFTER_ROTATION': False,
}

40
config/urls.py Normal file
View File

@@ -0,0 +1,40 @@
"""
URL configuration for config project.
"""
from django.contrib import admin
from django.urls import path, include
from django.conf import settings
from django.conf.urls.static import static
from rest_framework import permissions
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
from rest_framework_simplejwt.views import (
TokenObtainPairView,
TokenRefreshView,
)
schema_view = get_schema_view(
openapi.Info(
title="视频主设备管理系统 API",
default_version='v1',
description="视频编码器、解码器、矩阵、NVR等设备管理系统API文档",
terms_of_service="https://www.example.com/terms/",
contact=openapi.Contact(email="contact@example.com"),
license=openapi.License(name="BSD License"),
),
public=True,
permission_classes=(permissions.AllowAny,),
)
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', include('device_management.urls')),
path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
path('api-auth/', include('rest_framework.urls')),
path('swagger<format>/', schema_view.without_ui(cache_timeout=0), name='schema-json'),
path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

16
config/wsgi.py Normal file
View File

@@ -0,0 +1,16 @@
"""
WSGI config for config project.
It exposes the WSGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/5.0/howto/deployment/wsgi/
"""
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
application = get_wsgi_application()

0
device_management/__init__.py Normal file
View File

65
device_management/admin.py Normal file
View File

@@ -0,0 +1,65 @@
from django.contrib import admin
from .models import (
Device, DeviceSerial, DeviceIP, MaintenanceRecord, DeviceAttachment
)
class DeviceSerialInline(admin.TabularInline):
model = DeviceSerial
extra = 1
class DeviceIPInline(admin.TabularInline):
model = DeviceIP
extra = 1
class MaintenanceRecordInline(admin.TabularInline):
model = MaintenanceRecord
extra = 1
class DeviceAttachmentInline(admin.TabularInline):
model = DeviceAttachment
extra = 1
@admin.register(Device)
class DeviceAdmin(admin.ModelAdmin):
list_display = [
'id', 'device_name', 'location', 'building', 'status',
'responsible_person', 'enable_date', 'service_duration_days'
]
list_filter = ['status', 'location', 'building']
search_fields = ['device_name', 'location', 'brand', 'model']
inlines = [DeviceSerialInline, DeviceIPInline, MaintenanceRecordInline, DeviceAttachmentInline]
date_hierarchy = 'enable_date'
@admin.register(DeviceSerial)
class DeviceSerialAdmin(admin.ModelAdmin):
list_display = ['id', 'device', 'serial_number', 'serial_type', 'is_primary']
list_filter = ['serial_type', 'is_primary']
search_fields = ['serial_number']
@admin.register(DeviceIP)
class DeviceIPAdmin(admin.ModelAdmin):
list_display = ['id', 'device', 'ip_address', 'ip_type', 'is_primary']
list_filter = ['ip_type', 'is_primary']
search_fields = ['ip_address']
@admin.register(MaintenanceRecord)
class MaintenanceRecordAdmin(admin.ModelAdmin):
list_display = ['id', 'device', 'maintenance_date', 'maintenance_by', 'cost']
list_filter = ['maintenance_date', 'maintenance_by']
search_fields = ['fault_description', 'repair_content']
date_hierarchy = 'maintenance_date'
@admin.register(DeviceAttachment)
class DeviceAttachmentAdmin(admin.ModelAdmin):
list_display = ['id', 'device', 'file_name', 'file_type', 'uploaded_at']
list_filter = ['file_type', 'uploaded_at']
search_fields = ['file_name']

6
device_management/apps.py Normal file
View File

@@ -0,0 +1,6 @@
from django.apps import AppConfig
class DeviceManagementConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'device_management'

120
device_management/migrations/0001_initial.py Normal file
View File

@@ -0,0 +1,120 @@
# Generated by Django 5.2.14 on 2026-05-18 09:21
import device_management.models
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Device',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('location', models.CharField(max_length=100, verbose_name='地点')),
('building', models.CharField(blank=True, max_length=50, null=True, verbose_name='楼栋')),
('floor', models.CharField(blank=True, max_length=20, null=True, verbose_name='楼层')),
('cabinet', models.CharField(blank=True, max_length=50, null=True, verbose_name='机柜')),
('device_name', models.CharField(max_length=100, verbose_name='设备名称')),
('model', models.CharField(blank=True, max_length=100, null=True, verbose_name='型号')),
('brand', models.CharField(blank=True, max_length=100, null=True, verbose_name='品牌')),
('mac_address', models.CharField(blank=True, max_length=17, null=True, verbose_name='MAC地址')),
('enable_date', models.DateField(blank=True, null=True, verbose_name='启用日期')),
('last_inspection_date', models.DateField(blank=True, null=True, verbose_name='最后巡检日期')),
('thumbnail', models.ImageField(blank=True, null=True, upload_to=device_management.models.device_thumbnail_path, verbose_name='缩略图')),
('status', models.CharField(choices=[('normal', '正常'), ('warning', '告警'), ('offline', '离线'), ('repair', '维修中'), ('scrap', '已报废')], default='normal', max_length=20, verbose_name='状态')),
('responsible_person', models.CharField(blank=True, max_length=50, null=True, verbose_name='负责人')),
('warranty_expire', models.DateField(blank=True, null=True, verbose_name='保修到期日期')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
],
options={
'verbose_name': '设备',
'verbose_name_plural': '设备',
'db_table': 'device',
'indexes': [models.Index(fields=['location', 'building'], name='device_locatio_6a4f0a_idx'), models.Index(fields=['status'], name='device_status_6321a6_idx')],
},
),
migrations.CreateModel(
name='DeviceAttachment',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('file', models.FileField(upload_to=device_management.models.device_attachment_path, verbose_name='文件')),
('file_name', models.CharField(max_length=255, verbose_name='文件名')),
('file_type', models.CharField(blank=True, max_length=50, null=True, verbose_name='文件类型')),
('uploaded_at', models.DateTimeField(auto_now_add=True, verbose_name='上传时间')),
('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='attachments', to='device_management.device', verbose_name='设备')),
],
options={
'verbose_name': '设备附件',
'verbose_name_plural': '设备附件',
'db_table': 'device_attachment',
},
),
migrations.CreateModel(
name='DeviceIP',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('ip_address', models.CharField(max_length=45, unique=True, verbose_name='IP地址')),
('ip_type', models.CharField(default='management', max_length=20, verbose_name='IP类型')),
('is_primary', models.BooleanField(default=False, verbose_name='是否主IP')),
('subnet_mask', models.CharField(blank=True, max_length=15, null=True, verbose_name='子网掩码')),
('gateway', models.CharField(blank=True, max_length=45, null=True, verbose_name='网关')),
('vlan_id', models.CharField(blank=True, max_length=10, null=True, verbose_name='VLAN ID')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ips', to='device_management.device', verbose_name='设备')),
],
options={
'verbose_name': '设备IP地址',
'verbose_name_plural': '设备IP地址',
'db_table': 'device_ip',
},
),
migrations.CreateModel(
name='DeviceSerial',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('serial_number', models.CharField(max_length=100, unique=True, verbose_name='序列号')),
('serial_type', models.CharField(default='main', max_length=50, verbose_name='序列号类型')),
('is_primary', models.BooleanField(default=False, verbose_name='是否主序列号')),
('remark', models.CharField(blank=True, max_length=255, null=True, verbose_name='备注')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='serials', to='device_management.device', verbose_name='设备')),
],
options={
'verbose_name': '设备序列号',
'verbose_name_plural': '设备序列号',
'db_table': 'device_serial',
},
),
migrations.CreateModel(
name='MaintenanceRecord',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('maintenance_date', models.DateField(verbose_name='维修日期')),
('fault_description', models.TextField(blank=True, null=True, verbose_name='故障描述')),
('repair_content', models.TextField(blank=True, null=True, verbose_name='维修内容')),
('replaced_parts', models.CharField(blank=True, max_length=255, null=True, verbose_name='更换配件')),
('maintenance_by', models.CharField(blank=True, max_length=50, null=True, verbose_name='维修人')),
('cost', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True, verbose_name='费用')),
('remark', models.TextField(blank=True, null=True, verbose_name='备注')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='maintenance_records', to='device_management.device', verbose_name='设备')),
],
options={
'verbose_name': '维修记录',
'verbose_name_plural': '维修记录',
'db_table': 'maintenance_record',
'ordering': ['-maintenance_date'],
},
),
]

0
device_management/migrations/__init__.py Normal file
View File

170
device_management/models.py Normal file
View File

@@ -0,0 +1,170 @@
import os
from datetime import date
from django.db import models
from django.utils.translation import gettext_lazy as _
from PIL import Image
from io import BytesIO
from django.core.files.base import ContentFile
from loguru import logger
class DeviceStatus(models.TextChoices):
NORMAL = 'normal', _('正常')
WARNING = 'warning', _('告警')
OFFLINE = 'offline', _('离线')
REPAIR = 'repair', _('维修中')
SCRAP = 'scrap', _('已报废')
def device_attachment_path(instance, filename):
return f'devices/{instance.id}/attachments/{filename}'
def device_thumbnail_path(instance, filename):
return f'devices/{instance.id}/thumbnails/{filename}'
class Device(models.Model):
location = models.CharField(max_length=100, verbose_name='地点')
building = models.CharField(max_length=50, blank=True, null=True, verbose_name='楼栋')
floor = models.CharField(max_length=20, blank=True, null=True, verbose_name='楼层')
cabinet = models.CharField(max_length=50, blank=True, null=True, verbose_name='机柜')
device_name = models.CharField(max_length=100, verbose_name='设备名称')
model = models.CharField(max_length=100, blank=True, null=True, verbose_name='型号')
brand = models.CharField(max_length=100, blank=True, null=True, verbose_name='品牌')
mac_address = models.CharField(max_length=17, blank=True, null=True, verbose_name='MAC地址')
enable_date = models.DateField(blank=True, null=True, verbose_name='启用日期')
last_inspection_date = models.DateField(blank=True, null=True, verbose_name='最后巡检日期')
thumbnail = models.ImageField(upload_to=device_thumbnail_path, blank=True, null=True, verbose_name='缩略图')
status = models.CharField(
max_length=20,
choices=DeviceStatus.choices,
default=DeviceStatus.NORMAL,
verbose_name='状态'
)
responsible_person = models.CharField(max_length=50, blank=True, null=True, verbose_name='负责人')
warranty_expire = models.DateField(blank=True, null=True, verbose_name='保修到期日期')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')
class Meta:
db_table = 'device'
verbose_name = '设备'
verbose_name_plural = '设备'
indexes = [
models.Index(fields=['location', 'building']),
models.Index(fields=['status']),
]
def __str__(self):
return f'{self.device_name} - {self.location}'
@property
def service_duration_days(self):
if self.enable_date:
return (date.today() - self.enable_date).days
return 0
@property
def is_warranty_expired(self):
if self.warranty_expire:
return date.today() > self.warranty_expire
return False
def generate_thumbnail(self, image_field, size=(200, 200)):
if not image_field:
return
try:
img = Image.open(image_field)
if img.mode in ('RGBA', 'P'):
img = img.convert('RGB')
img.thumbnail(size)
thumb_io = BytesIO()
img.save(thumb_io, format='JPEG', quality=85)
thumb_name = f'thumb_{os.path.basename(image_field.name)}'
self.thumbnail.save(thumb_name, ContentFile(thumb_io.getvalue()), save=False)
logger.info(f'Generated thumbnail for device {self.id}')
except Exception as e:
logger.error(f'Failed to generate thumbnail: {e}')
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
class DeviceSerial(models.Model):
device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='serials', verbose_name='设备')
serial_number = models.CharField(max_length=100, unique=True, verbose_name='序列号')
serial_type = models.CharField(max_length=50, default='main', verbose_name='序列号类型')
is_primary = models.BooleanField(default=False, verbose_name='是否主序列号')
remark = models.CharField(max_length=255, blank=True, null=True, verbose_name='备注')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')
class Meta:
db_table = 'device_serial'
verbose_name = '设备序列号'
verbose_name_plural = '设备序列号'
def __str__(self):
return f'{self.serial_number} ({self.serial_type})'
class DeviceIP(models.Model):
device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='ips', verbose_name='设备')
ip_address = models.CharField(max_length=45, unique=True, verbose_name='IP地址')
ip_type = models.CharField(max_length=20, default='management', verbose_name='IP类型')
is_primary = models.BooleanField(default=False, verbose_name='是否主IP')
subnet_mask = models.CharField(max_length=15, blank=True, null=True, verbose_name='子网掩码')
gateway = models.CharField(max_length=45, blank=True, null=True, verbose_name='网关')
vlan_id = models.CharField(max_length=10, blank=True, null=True, verbose_name='VLAN ID')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')
class Meta:
db_table = 'device_ip'
verbose_name = '设备IP地址'
verbose_name_plural = '设备IP地址'
def __str__(self):
return f'{self.ip_address} ({self.ip_type})'
class MaintenanceRecord(models.Model):
device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='maintenance_records', verbose_name='设备')
maintenance_date = models.DateField(verbose_name='维修日期')
fault_description = models.TextField(blank=True, null=True, verbose_name='故障描述')
repair_content = models.TextField(blank=True, null=True, verbose_name='维修内容')
replaced_parts = models.CharField(max_length=255, blank=True, null=True, verbose_name='更换配件')
maintenance_by = models.CharField(max_length=50, blank=True, null=True, verbose_name='维修人')
cost = models.DecimalField(max_digits=10, decimal_places=2, blank=True, null=True, verbose_name='费用')
remark = models.TextField(blank=True, null=True, verbose_name='备注')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')
class Meta:
db_table = 'maintenance_record'
verbose_name = '维修记录'
verbose_name_plural = '维修记录'
ordering = ['-maintenance_date']
def __str__(self):
return f'{self.device.device_name} - {self.maintenance_date}'
class DeviceAttachment(models.Model):
device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='attachments', verbose_name='设备')
file = models.FileField(upload_to=device_attachment_path, verbose_name='文件')
file_name = models.CharField(max_length=255, verbose_name='文件名')
file_type = models.CharField(max_length=50, blank=True, null=True, verbose_name='文件类型')
uploaded_at = models.DateTimeField(auto_now_add=True, verbose_name='上传时间')
class Meta:
db_table = 'device_attachment'
verbose_name = '设备附件'
verbose_name_plural = '设备附件'
def __str__(self):
return self.file_name

138
device_management/serializers.py Normal file
View File

@@ -0,0 +1,138 @@
from rest_framework import serializers
from django.db import transaction
from .models import (
Device, DeviceSerial, DeviceIP, MaintenanceRecord, DeviceAttachment
)
from loguru import logger
class DeviceSerialSerializer(serializers.ModelSerializer):
class Meta:
model = DeviceSerial
fields = ['id', 'serial_number', 'serial_type', 'is_primary', 'remark']
class DeviceIPSerializer(serializers.ModelSerializer):
class Meta:
model = DeviceIP
fields = ['id', 'ip_address', 'ip_type', 'is_primary', 'subnet_mask', 'gateway', 'vlan_id']
class MaintenanceRecordSerializer(serializers.ModelSerializer):
class Meta:
model = MaintenanceRecord
fields = ['id', 'maintenance_date', 'fault_description', 'repair_content',
'replaced_parts', 'maintenance_by', 'cost', 'remark']
class DeviceAttachmentSerializer(serializers.ModelSerializer):
class Meta:
model = DeviceAttachment
fields = ['id', 'file', 'file_name', 'file_type', 'uploaded_at']
read_only_fields = ['uploaded_at']
class DeviceSerializer(serializers.ModelSerializer):
serials = DeviceSerialSerializer(many=True, required=False)
ips = DeviceIPSerializer(many=True, required=False)
maintenance_records = MaintenanceRecordSerializer(many=True, read_only=True)
attachments = DeviceAttachmentSerializer(many=True, read_only=True)
service_duration_days = serializers.IntegerField(read_only=True)
is_warranty_expired = serializers.BooleanField(read_only=True)
primary_serial = serializers.SerializerMethodField()
primary_ip = serializers.SerializerMethodField()
latest_maintenance = serializers.SerializerMethodField()
class Meta:
model = Device
fields = ['id', 'location', 'building', 'floor', 'cabinet', 'device_name',
'model', 'brand', 'mac_address', 'enable_date', 'last_inspection_date',
'thumbnail', 'status', 'responsible_person', 'warranty_expire',
'service_duration_days', 'is_warranty_expired', 'primary_serial',
'primary_ip', 'latest_maintenance', 'serials', 'ips',
'maintenance_records', 'attachments', 'created_at', 'updated_at']
read_only_fields = ['created_at', 'updated_at']
def get_primary_serial(self, obj):
primary = obj.serials.filter(is_primary=True).first()
if primary:
return DeviceSerialSerializer(primary).data
return None
def get_primary_ip(self, obj):
primary = obj.ips.filter(is_primary=True).first()
if primary:
return DeviceIPSerializer(primary).data
return None
def get_latest_maintenance(self, obj):
latest = obj.maintenance_records.first()
if latest:
return MaintenanceRecordSerializer(latest).data
return None
@transaction.atomic
def create(self, validated_data):
serials_data = validated_data.pop('serials', [])
ips_data = validated_data.pop('ips', [])
device = Device.objects.create(**validated_data)
for serial_data in serials_data:
DeviceSerial.objects.create(device=device, **serial_data)
for ip_data in ips_data:
DeviceIP.objects.create(device=device, **ip_data)
logger.info(f'Created device: {device.id} - {device.device_name}')
return device
@transaction.atomic
def update(self, instance, validated_data):
serials_data = validated_data.pop('serials', None)
ips_data = validated_data.pop('ips', None)
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()
if serials_data is not None:
instance.serials.all().delete()
for serial_data in serials_data:
DeviceSerial.objects.create(device=instance, **serial_data)
if ips_data is not None:
instance.ips.all().delete()
for ip_data in ips_data:
DeviceIP.objects.create(device=instance, **ip_data)
logger.info(f'Updated device: {instance.id}')
return instance
class DeviceListSerializer(serializers.ModelSerializer):
primary_serial_number = serializers.SerializerMethodField()
primary_ip_address = serializers.SerializerMethodField()
latest_maintenance_summary = serializers.SerializerMethodField()
service_duration_days = serializers.IntegerField(read_only=True)
class Meta:
model = Device
fields = ['id', 'location', 'building', 'device_name', 'model', 'brand',
'status', 'responsible_person', 'enable_date', 'thumbnail',
'primary_serial_number', 'primary_ip_address',
'latest_maintenance_summary', 'service_duration_days']
def get_primary_serial_number(self, obj):
primary = obj.serials.filter(is_primary=True).first()
return primary.serial_number if primary else None
def get_primary_ip_address(self, obj):
primary = obj.ips.filter(is_primary=True).first()
return primary.ip_address if primary else None
def get_latest_maintenance_summary(self, obj):
latest = obj.maintenance_records.first()
if latest:
return f"{latest.maintenance_date}: {latest.fault_description[:50]}..." if latest.fault_description else str(latest.maintenance_date)
return None

3
device_management/tests.py Normal file
View File

@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.

17
device_management/urls.py Normal file
View File

@@ -0,0 +1,17 @@
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import (
DeviceViewSet, DeviceSerialViewSet, DeviceIPViewSet,
MaintenanceRecordViewSet, DeviceAttachmentViewSet
)
router = DefaultRouter()
router.register(r'devices', DeviceViewSet, basename='device')
router.register(r'serials', DeviceSerialViewSet, basename='deviceserial')
router.register(r'ips', DeviceIPViewSet, basename='deviceip')
router.register(r'maintenance', MaintenanceRecordViewSet, basename='maintenancerecord')
router.register(r'attachments', DeviceAttachmentViewSet, basename='deviceattachment')
urlpatterns = [
path('', include(router.urls)),
]

210
device_management/views.py Normal file
View File

@@ -0,0 +1,210 @@
import os
from datetime import date
from django.http import HttpResponse
from django.shortcuts import get_object_or_404
from django_filters import rest_framework as filters
from rest_framework import viewsets, status, parsers, renderers
from rest_framework.decorators import action, api_view, parser_classes
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly, IsAuthenticated
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font
import pandas as pd
from loguru import logger
from .models import (
Device, DeviceSerial, DeviceIP, MaintenanceRecord, DeviceAttachment
)
from .serializers import (
DeviceSerializer, DeviceListSerializer, DeviceSerialSerializer,
DeviceIPSerializer, MaintenanceRecordSerializer, DeviceAttachmentSerializer
)
class DeviceFilter(filters.FilterSet):
location = filters.CharFilter(lookup_expr='icontains')
building = filters.CharFilter(lookup_expr='icontains')
from .models import DeviceStatus
status = filters.ChoiceFilter(choices=DeviceStatus.choices)
device_name = filters.CharFilter(lookup_expr='icontains')
brand = filters.CharFilter(lookup_expr='icontains')
enable_date_from = filters.DateFilter(field_name='enable_date', lookup_expr='gte')
enable_date_to = filters.DateFilter(field_name='enable_date', lookup_expr='lte')
warranty_expire_soon = filters.BooleanFilter(method='filter_warranty_expire_soon')
class Meta:
model = Device
fields = ['location', 'building', 'status', 'device_name', 'brand']
def filter_warranty_expire_soon(self, queryset, name, value):
if value:
from datetime import timedelta
soon_date = date.today() + timedelta(days=30)
return queryset.filter(warranty_expire__lte=soon_date, warranty_expire__gte=date.today())
return queryset
class DeviceViewSet(viewsets.ModelViewSet):
queryset = Device.objects.all().prefetch_related('serials', 'ips', 'maintenance_records')
serializer_class = DeviceSerializer
filterset_class = DeviceFilter
search_fields = ['device_name', 'location', 'building', 'brand', 'model']
ordering_fields = ['id', 'device_name', 'location', 'enable_date', 'created_at']
permission_classes = [IsAuthenticatedOrReadOnly]
def get_serializer_class(self):
if self.action == 'list':
return DeviceListSerializer
return DeviceSerializer
@action(detail=True, methods=['post'], parser_classes=[parsers.MultiPartParser])
def upload_attachment(self, request, pk=None):
device = self.get_object()
file_obj = request.FILES.get('file')
if not file_obj:
return Response({'error': 'No file provided'}, status=status.HTTP_400_BAD_REQUEST)
file_type = request.data.get('file_type', '')
attachment = DeviceAttachment.objects.create(
device=device,
file=file_obj,
file_name=file_obj.name,
file_type=file_type
)
if file_type.startswith('image/') or file_obj.name.lower().endswith(('.png', '.jpg', '.jpeg', '.gif', '.bmp')):
device.generate_thumbnail(file_obj)
device.save()
logger.info(f'Uploaded attachment {attachment.id} for device {device.id}')
return Response(DeviceAttachmentSerializer(attachment).data, status=status.HTTP_201_CREATED)
@action(detail=False, methods=['get'])
def export_excel(self, request):
queryset = self.filter_queryset(self.get_queryset())
wb = Workbook()
ws = wb.active
ws.title = '设备列表'
headers = ['ID', '地点', '楼栋', '设备名称', '型号', '品牌', '状态',
'负责人', '启用日期', '主序列号', '主IP', '最近维修简述', '服役天数']
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True)
for row, device in enumerate(queryset, 2):
primary_serial = device.serials.filter(is_primary=True).first()
primary_ip = device.ips.filter(is_primary=True).first()
latest_maintenance = device.maintenance_records.first()
ws.cell(row=row, column=1, value=device.id)
ws.cell(row=row, column=2, value=device.location)
ws.cell(row=row, column=3, value=device.building or '')
ws.cell(row=row, column=4, value=device.device_name)
ws.cell(row=row, column=5, value=device.model or '')
ws.cell(row=row, column=6, value=device.brand or '')
ws.cell(row=row, column=7, value=device.get_status_display())
ws.cell(row=row, column=8, value=device.responsible_person or '')
ws.cell(row=row, column=9, value=str(device.enable_date) if device.enable_date else '')
ws.cell(row=row, column=10, value=primary_serial.serial_number if primary_serial else '')
ws.cell(row=row, column=11, value=primary_ip.ip_address if primary_ip else '')
ws.cell(row=row, column=12, value=str(latest_maintenance) if latest_maintenance else '')
ws.cell(row=row, column=13, value=device.service_duration_days)
response = HttpResponse(content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
response['Content-Disposition'] = f'attachment; filename="devices_{date.today()}.xlsx"'
wb.save(response)
logger.info(f'Exported {queryset.count()} devices to Excel')
return response
@action(detail=False, methods=['post'], parser_classes=[parsers.MultiPartParser])
def import_excel(self, request):
file_obj = request.FILES.get('file')
if not file_obj:
return Response({'error': 'No file provided'}, status=status.HTTP_400_BAD_REQUEST)
try:
df = pd.read_excel(file_obj)
created_count = 0
for _, row in df.iterrows():
device_data = {
'location': str(row.get('地点', '')),
'building': str(row.get('楼栋', '')) if pd.notna(row.get('楼栋')) else None,
'device_name': str(row.get('设备名称', '')),
'model': str(row.get('型号', '')) if pd.notna(row.get('型号')) else None,
'brand': str(row.get('品牌', '')) if pd.notna(row.get('品牌')) else None,
'status': str(row.get('状态', 'normal')) if pd.notna(row.get('状态')) else 'normal',
'responsible_person': str(row.get('负责人', '')) if pd.notna(row.get('负责人')) else None,
}
device = Device.objects.create(**device_data)
if pd.notna(row.get('主序列号')):
DeviceSerial.objects.create(
device=device,
serial_number=str(row.get('主序列号')),
serial_type='main',
is_primary=True
)
if pd.notna(row.get('主IP')):
DeviceIP.objects.create(
device=device,
ip_address=str(row.get('主IP')),
ip_type='management',
is_primary=True
)
created_count += 1
logger.info(f'Imported {created_count} devices from Excel')
return Response({'message': f'成功导入 {created_count} 台设备', 'count': created_count})
except Exception as e:
logger.error(f'Excel import failed: {e}')
return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)
class DeviceSerialViewSet(viewsets.ModelViewSet):
queryset = DeviceSerial.objects.all()
serializer_class = DeviceSerialSerializer
search_fields = ['serial_number', 'serial_type']
permission_classes = [IsAuthenticatedOrReadOnly]
class DeviceIPViewSet(viewsets.ModelViewSet):
queryset = DeviceIP.objects.all()
serializer_class = DeviceIPSerializer
search_fields = ['ip_address', 'ip_type']
permission_classes = [IsAuthenticatedOrReadOnly]
class MaintenanceRecordViewSet(viewsets.ModelViewSet):
queryset = MaintenanceRecord.objects.all()
serializer_class = MaintenanceRecordSerializer
filterset_fields = ['device_id', 'maintenance_by']
search_fields = ['fault_description', 'repair_content', 'maintenance_by']
permission_classes = [IsAuthenticatedOrReadOnly]
class DeviceAttachmentViewSet(viewsets.ModelViewSet):
queryset = DeviceAttachment.objects.all()
serializer_class = DeviceAttachmentSerializer
parser_classes = [parsers.MultiPartParser, parsers.FormParser]
permission_classes = [IsAuthenticatedOrReadOnly]
def perform_create(self, serializer):
device_id = self.request.data.get('device')
device = get_object_or_404(Device, id=device_id)
file_obj = self.request.FILES.get('file')
attachment = serializer.save(
device=device,
file_name=file_obj.name if file_obj else ''
)
logger.info(f'Created attachment {attachment.id}')

22
manage.py Normal file
View File

@@ -0,0 +1,22 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == '__main__':
main()

11
requirements.txt Normal file
View File

@@ -0,0 +1,11 @@
Django==4.2.11
djangorestframework==3.15.1
djangorestframework-simplejwt==5.3.1
drf-yasg==1.21.7
Pillow==10.3.0
mysqlclient==2.2.4
django-filter==24.2
openpyxl==3.1.2
pandas==2.2.2
numpy==1.26.4
loguru==0.7.2