当前位置：首页 > article >正文

wzl-django学习

article 2026/4/1 21:10:01

####################################################总的urls.py

from django.contrib import admin
from django.urls import path,include, re_path
from django.views.static import serve
from django.conf import settings
from drf_yasg import openapi
from drf_yasg.views import get_schema_view
from rest_framework import permissions
from django.views.generic.base import RedirectView
from rest_framework_simplejwt.views import (
TokenRefreshView,
)

from mysystem.views.login import LoginView,CaptchaView
from utils.swagger import CustomOpenAPISchemaGenerator

#前端接口view
from apps.logins.views import UsernamePassWordLoginView,APPMobilePasswordLoginView,SendSmsCodeView,APPMobileSMSLoginView,ForgetPasswdResetView,RegisterView
from apps.lyusers.views import SetUserNicknameView,ChangeAvatarView,uploadImagesView,DestroyUserView
from apps.lymessages.views import UserMessagesView,UserMessagesNoticeView,GetUnreadMessageNumView
from apps.platformsettings.views import *

#媒体文件流式响应
from utils.streamingmedia_response import streamingmedia_serve
#部署vue
from django.views.generic import TemplateView
#是否允许前端接口访问
from utils.middleware import OperateAllowFrontendView

schema_view = get_schema_view(
openapi.Info(
title="django-vue-lyadmin API",
default_version='v1',
# description="Test description",
# terms_of_service="https://www.google.com/policies/terms/",
# contact=openapi.Contact(email="contact@snippets.local"),
# license=openapi.License(name="BSD License"),
),
# public 如果为False，则只包含当前用户可以访问的端点。True返回全部
public=True,
permission_classes=(permissions.AllowAny,),# 可以允许任何人查看该接口
# permission_classes=(permissions.IsAuthenticated) # 只允许通过认证的查看该接口
generator_class=CustomOpenAPISchemaGenerator,
)

urlpatterns = [
path('static/<path:path>', serve, {'document_root': settings.STATIC_ROOT},), # 处理静态文件
# path('media/<path:path>', serve, {'document_root': settings.MEDIA_ROOT},), # 处理媒体文件
path('media/<path:path>', streamingmedia_serve, {'document_root': settings.MEDIA_ROOT}, ), # 处理媒体文件

#管理后台的标准接口
path('api/system/', include('mysystem.urls')),
path('api/monitor/', include('lymonitor.urls')),
path('api/terminal/', include('lywebsocket.urls')),
path('api/token/', LoginView.as_view(), name='token_obtain_pair'),
path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
path('api/captcha/', CaptchaView.as_view()),

#管理后台其他自定义接口
path('api/platformsettings/', include('apps.platformsettings.urls')),
path('api/messages/', include('apps.lymessages.urls')),
path('api/users/', include('apps.lyusers.urls')),
path('api/crontab/', include('apps.lycrontab.urls')),
path('api/wholeMachine/', include('apps.wholeMachine.urls')),

#获取平台信息
path('api/getsysconfig/', GetSystemConfigSettingsView.as_view(), name='前端用户获取平台其他设置'),
path('api/getothersettings/', GetOtherManageDetailView.as_view(), name='前端用户获取平台其他设置'),
path('api/getrotationimgs/', GetLunboManageListView.as_view(), name='前端用户获取平台轮播图设置'),

#是否允许前端接口访问(临时操作，重启后无效)
path('api/super/operate/', OperateAllowFrontendView.as_view(), name='超级管理员动态操作是否允许前端api接口访问'),

#集成部署后端管理页面

path('favicon.ico',RedirectView.as_view(url=r'static/favicon.ico')),
path('', TemplateView.as_view(template_name="index.html"),name='后台管理默认页面'),
]

####################################################models.py

from django.db import models
from utils.models import CoreModel
from application import settings

# ================================================= #
# ************** 服务器分类 model ************** #
# ================================================= #

class ServersUsage(CoreModel):
"""整机用途"""
objects = None
name = models.CharField(max_length=255, unique=True, verbose_name='名称')
desc = models.CharField( max_length=255,null=True, blank=True,verbose_name='描述')
sort = models.PositiveSmallIntegerField(default=0, verbose_name="排序", help_text="显示顺序")
status = models.BooleanField(default=True, verbose_name="整机用途状态", help_text="整机用途状态")
is_delete = models.BooleanField(default=False, verbose_name="是否逻辑删除", help_text="是否逻辑删除")
class Meta:
db_table = 'tb_servers_usage'
verbose_name = '整机用途'
verbose_name_plural = verbose_name
def __str__(self):
return self.name

class ServersModel(CoreModel):
"""整机机型"""
objects = None
name = models.CharField( max_length=255,unique=True, verbose_name='名称')
area= models.CharField(max_length=255, null=True, verbose_name='设备领域')
desc = models.CharField( max_length=255,null=True, blank=True,verbose_name='描述')
sort = models.PositiveSmallIntegerField(default=0, verbose_name="排序", help_text="显示顺序")
status = models.BooleanField(default=True, verbose_name="整机机型状态", help_text="整机机型状态")
is_delete = models.BooleanField(default=False, verbose_name="是否逻辑删除", help_text="是否逻辑删除")

class Meta:
db_table = 'tb_servers_model'
verbose_name = '整机机型'
verbose_name_plural = verbose_name

def __str__(self):
return self.name

# # ================================================= #
# # ************** 跳板机信息 ************** #
# # ================================================= ##
class JumpServers(CoreModel):
"""跳板机信息"""
jumpip = models.CharField(max_length=255, verbose_name="跳板机IP")
jumpuser = models.CharField(max_length=255, verbose_name="跳板机user")
jumppassword= models.CharField(max_length=255,verbose_name="跳板机password")

# ================================================= #
# ************** 整机信息 ************** #
# ================================================= #

class ServersInfo(CoreModel):
"""整机信息"""
objects = None
serverFrom = models.CharField(max_length=255,blank=True, null=True, verbose_name="信息来源") #
serverCode = models.CharField(max_length=255,blank=False, null=False, unique=True,verbose_name="设备编号") #
location = models.CharField(max_length=255, blank=False, null=False, verbose_name="设备位置")
hardware = models.CharField(max_length=255, blank=True, null=True, verbose_name="硬件配置")
serversModel = models.ForeignKey(ServersModel, on_delete=models.PROTECT,blank=False, null=False,verbose_name='设备型号')
serversUsage = models.ForeignKey(ServersUsage, on_delete=models.PROTECT, blank=False, null=False,verbose_name='设备用途')
status = models.BooleanField(default=True, verbose_name="设备状态", help_text="设备状态")
bmcIP = models.CharField(max_length=255, verbose_name="BMC地址")
bmcUser = models.CharField(max_length=255, verbose_name="bmcuser",blank=True)
bmcPassword= models.CharField(max_length=255, verbose_name="bmcpassword",blank=True)
hostIP = models.CharField(max_length=255, verbose_name="host地址",blank=True)
hostUser = models.CharField(max_length=255, verbose_name="hostUser", blank=True)
hostPassword = models.CharField(max_length=255, verbose_name="hostPassword", blank=True)
serialIP = models.CharField(max_length=255, verbose_name="串口地址", blank=True,null=True)
serialPort= models.CharField(max_length=255, verbose_name="串口端口", blank=True,null=True)
serialUser = models.CharField(max_length=255, verbose_name="串口User", blank=True,null=True)
serialPassword = models.CharField(max_length=255, verbose_name="串口Password",blank=True,null=True)
toolingMachine = models.CharField(max_length=255, verbose_name="工装机", blank=True,null=True)
borrowUser = models.CharField(max_length=255, null=True, blank=True,verbose_name="借用人")
borrowFromtime= models.DateField(null=True, blank=True, verbose_name="借用时间")
borrowEndtime= models.DateField(null=True, blank=True, verbose_name="归还时间")
borrowUse= models.CharField(max_length=255, verbose_name="借用用途", blank=True,null=True)
notes = models.TextField( verbose_name="备注", blank=True,null=True)
missingStatus = models.BooleanField( default=True, verbose_name="联通状态")
recordStatus =models.BooleanField( default=False,verbose_name="备案不下电")
jumpserver = models.ForeignKey(JumpServers, on_delete=models.SET_NULL,null=True, verbose_name='跳板机')
class Meta:
db_table = 'tb_servers_info'
verbose_name = '服务器信息'
verbose_name_plural = verbose_name
def __str__(self):
return self.serverscode

class HostOffRecord(CoreModel):
"""下电记录"""
servers = models.ForeignKey(ServersInfo, on_delete=models.PROTECT, verbose_name='整机')
status = models.BooleanField( verbose_name="下电是否成功")
class Meta:
db_table = 'tb_servers_hostoffrecord'
verbose_name = 'host下电记录'
verbose_name_plural = verbose_name

class ServersUseRecord(CoreModel):
"""整机使用记录"""
servers = models.ForeignKey(ServersInfo, on_delete=models.CASCADE, verbose_name='整机')
borrowUser = models.CharField(max_length=255,verbose_name="借用人", blank=True)
borrowFromtime= models.DateField(null=True, blank=True, verbose_name="借用时间")
borrowEndtime= models.DateField(null=True, blank=True, verbose_name="归还时间")
class Meta:
db_table = 'tb_servers_useRecord'
verbose_name = '整机使用记录'
verbose_name_plural = verbose_name

############views.py

import openpyxl
from django.db.models import Count, Q, Case, When, IntegerField, Value
from django.utils.timezone import now
from django_filters.rest_framework import DjangoFilterBackend
from apps.wholeMachine.models import ServersUsage, ServersModel, ServersInfo
from rest_framework import serializers
from apps.wholeMachine.mytools import is_valid_date
from utils.export_excel import export_excel
from utils.jsonResponse import SuccessResponse, ErrorResponse
from utils.serializers import CustomModelSerializer
from utils.viewset import CustomModelViewSet
class ServersUsageManageSerializer(CustomModelSerializer):
"""
服务器用途-序列化器
"""
class Meta:
model = ServersUsage
read_only_fields = ["id"]
fields = '__all__'
class ServersUsageManageViewSet(CustomModelViewSet):
"""
服务器用途接口
"""
queryset = ServersUsage.objects.all().order_by("sort")
serializer_class = ServersUsageManageSerializer
search_fields = ('name','status')
class ServersModelManageSerializer(CustomModelSerializer):
"""
服务器机型-序列化器
"""
class Meta:
model = ServersModel
read_only_fields = ["id"]
fields = '__all__'

class ServersModelManageSerializer_UsageandModel(serializers.Serializer):
serversUsage__name = serializers.CharField()
serversUsage = serializers.CharField()
serversModel__name= serializers.CharField()
serversModel= serializers.CharField()
serversCount = serializers.IntegerField()
serversCount_dis = serializers.IntegerField()
class ServersModelManageViewSet(CustomModelViewSet):
"""
服务器机型接口
"""
queryset = ServersModel.objects.all().order_by("sort")
serializer_class = ServersModelManageSerializer
search_fields = ('name',)
filterset_fields = ('status','area',)
def getArea(self,request,*args, **kwargs):
queryset = ServersModel.objects.values('area')
return SuccessResponse(data=queryset, msg='success')

class ServersInfoManageSerializer(CustomModelSerializer):
"""
整机信息-序列化器
"""
serversmodel_area = serializers.SerializerMethodField(read_only=True)
serversmodel_name = serializers.SerializerMethodField(read_only=True)
serversusage_name = serializers.SerializerMethodField(read_only=True)
def get_serversmodel_area(self,obj):
if obj.serversModel :
return obj.serversModel.area
def get_serversmodel_name(self,obj):
if obj.serversModel :
return obj.serversModel.name
def get_serversusage_name(self,obj):
if obj.serversUsage :
return obj.serversUsage.name
class Meta:
model = ServersInfo
read_only_fields = ["id"]
fields = '__all__'

class ExportServersInfoManageSerializer(CustomModelSerializer):
"""
导出整机信息简单序列化器
"""
serversmodel_area = serializers.SerializerMethodField(read_only=True)
serversusage_name = serializers.SerializerMethodField(read_only=True)
serversmodel_name = serializers.SerializerMethodField(read_only=True)
def get_serversmodel_area(self,obj):
if obj.serversModel :
return obj.serversModel.area
def get_serversusage_name(self,obj):
if obj.serversUsage :
return obj.serversUsage.name
def get_serversmodel_name(self,obj):
if obj.serversModel :
return obj.serversModel.name
class Meta:
model = ServersInfo
fields = ('serverCode', 'location','serversmodel_area', 'serversusage_name','serversmodel_name','bmcip','bmcuser','bmcpassword','serialIP','serialPort')
# class ServersInfoCountSerializer_TypeandModel(serializers.Serializer):
# serversModel__area = serializers.CharField()
# serversModel__name= serializers.CharField()
# serversModel= serializers.CharField()
# serversCount = serializers.IntegerField()
# serversCount_dis = serializers.IntegerField()
# serversCount_nouse = serializers.IntegerField()

class ServersInfoManageViewSet(CustomModelViewSet):
"""
整机信息接口
"""
queryset = ServersInfo.objects.filter(status=1).order_by("-create_datetime")
# queryset = ServersInfo.objects.all().order_by("-create_datetime")
serializer_class = ServersInfoManageSerializer
create_serializer_class = ServersInfoManageSerializer
update_serializer_class =ServersInfoManageSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = {
'serverCode': ['icontains'],
'serverFrom': ['icontains'],
'hardware': ['icontains'],
'borrowEndtime': ['lte'],
'serversUsage': ['exact'],
'serversModel': ['exact'],
'serversModel__area': ['exact'],
}
search_fields = ('serverCode', 'serverFrom', 'hardware')
def get_queryset(self):
queryset = super().get_queryset()
search_param = self.request.query_params.get('search', None)
borrow_user_is_null = self.request.query_params.get('borrowUserIsNull', None)
borrow_endtime_earlier_param = self.request.query_params.get('borrowEndtimeEarlierThan', None)
borrow_endtime_later_param = self.request.query_params.get('borrowEndtimeLaterThan', None)
if borrow_user_is_null is not None:
borrow_user_is_null = borrow_user_is_null.lower()=='true'
queryset = queryset.filter(borrowUser__isnull=borrow_user_is_null)
if search_param:
queryset = queryset.filter(
Q(serverCode__icontains=search_param) |
Q(serverFrom__icontains=search_param) |
Q(hardware__icontains=search_param)|
Q(bmcIP__icontains=search_param)
)
if borrow_endtime_earlier_param:
queryset = queryset.filter(
Q(borrowEndtime__lt=borrow_endtime_earlier_param) | Q(borrowEndtime__isnull=True)
)
if borrow_endtime_later_param:
queryset = queryset.filter(
Q(borrowEndtime__gte=borrow_endtime_later_param) | Q(borrowEndtime__isnull=True)
)
return queryset

def countAll(self,request,*args, **kwargs):
total_books = ServersInfo.objects.count()
serversCount_dis = ServersInfo.objects.filter(missingStatus=0).count()
data = {
'total_books': total_books,
'serversCount_dis':serversCount_dis,
}
return SuccessResponse(data=data, msg='success')

def countbyAreaandModel(self,request,*args, **kwargs):
current_date = now().date()
queryset = ServersInfo.objects.values('serversModel__area', 'serversModel__name','serversModel').annotate(
serversCount=Count('id'),
serversCount_dis=Count('id', filter=Q(missingStatus=0)),
serversCount_nouse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
((Q(borrowEndtime__lt=current_date) & Q(borrowUser__isnull=False)) |
Q(borrowUser__isnull=True)),
then=Value(1)
),
output_field=IntegerField()
)
),
serversCount_inuse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
(Q(borrowEndtime__gte=current_date) | Q(borrowEndtime__isnull=True)) &
Q(borrowUser__isnull=False),
then=Value(1)
),
output_field=IntegerField()
)
)
).order_by('-serversCount')
# serializer = ServersInfoCountSerializer_TypeandModel(queryset, many=True)
return SuccessResponse(data=queryset, msg='success')
def countbyArea(self,request,*args, **kwargs):
current_date = now().date()
queryset = ServersInfo.objects.values('serversModel__area').annotate(
serversCount=Count('id'),
serversCount_dis = Count('id', filter=Q(missingStatus=0)),
serversCount_nouse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
((Q(borrowEndtime__lt=current_date) & Q(borrowUser__isnull=False)) |
Q(borrowUser__isnull=True)),
then=Value(1)
),
output_field=IntegerField()
)
),
serversCount_inuse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
(Q(borrowEndtime__gte=current_date) | Q(borrowEndtime__isnull=True))&
Q(borrowUser__isnull=False),
then=Value(1)
),
output_field=IntegerField()
)
)
).order_by('-serversCount')
return SuccessResponse(data=queryset, msg='success')

def countbyUsageandModel(self,request,*args, **kwargs):
current_date = now().date()
queryset = ServersInfo.objects.values('serversModel__name','serversUsage__name','serversModel','serversUsage').annotate(
serversCount=Count('id'),
serversCount_dis = Count('id', filter=Q(missingStatus=0)),
serversCount_nouse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
((Q(borrowEndtime__lt=current_date) & Q(borrowUser__isnull=False)) |
Q(borrowUser__isnull=True)),
then=Value(1)
),
output_field=IntegerField()
)
),
serversCount_inuse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
(Q(borrowEndtime__gte=current_date) | Q(borrowEndtime__isnull=True)) &
Q(borrowUser__isnull=False),
then=Value(1)
),
output_field=IntegerField()
)
)

).order_by('-serversModel__name','-serversCount')
# serializer = ServersInfoCountSerializer_UsageandModel(queryset, many=True)
return SuccessResponse(data=queryset, msg='success')
def countbyUsage(self,request,*args, **kwargs):
current_date = now().date()
queryset = ServersInfo.objects.values('serversUsage__name','serversUsage').annotate(
serversCount=Count('id'),
serversCount_dis = Count('id', filter=Q(missingStatus=0)),
serversCount_nouse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
((Q(borrowEndtime__lt=current_date) & Q(borrowUser__isnull=False)) |
Q(borrowUser__isnull=True)),
then=Value(1)
),
output_field=IntegerField()
)
),
serversCount_inuse=Count(
Case(
When(
Q(serversUsage__name='公共资源') &
(Q(borrowEndtime__gte=current_date) | Q(borrowEndtime__isnull=True))&
Q(borrowUser__isnull=False),
# queryset=queryset.filter(borrowUser__isnull=borrow_user_is_null)
then=Value(1)
),
output_field=IntegerField()
)
)
).order_by('-serversCount')
return SuccessResponse(data=queryset, msg='success')

def exportexecl(self, request):
field_data = ['服务器编号', '服务器位置', '整机类别', '整机用途', '整机型号', 'BMCIP','bmcuser','bmcpassword','串口地址','串口端口']
queryset = self.filter_queryset(self.get_queryset())
data = ExportServersInfoManageSerializer(queryset, many=True).data
return SuccessResponse(data=export_excel(request, field_data, data, '服务器数据.xls'), msg='success')

def importexcel(self,request, *args, **kwargs):
"""
导入execl中字段如果为数字会自动加.0 ，解决方法：execl设置该列为文本型
"""
f = request.FILES.get('file')
if not f:
return ErrorResponse(msg="未找到上传的文件")
excel_type = f.name.split('.')[1]
if excel_type not in ['xlsx']:
return ErrorResponse(msg="仅允许导入xlsx文件")
workbook = openpyxl.load_workbook(filename=f,read_only=True)
sheet = workbook.active
data_import_list = []
error_import_list = []
rowsnum=1
existing_server_codes = set(ServersInfo.objects.values_list('serverCode', flat=True))
for rowvalue in sheet.iter_rows(min_row=2, values_only=True):
rowsnum = rowsnum+1
errordict={}
errordict['rowsnum']= rowsnum
serverCode = rowvalue[2]
if serverCode is not None:
# serverCode = serverCode.replace(' ', '')
serverCode = serverCode.strip()
else:
errordict['serverCode'] = f'设备编号错误----{serverCode}'
error_import_list.append(errordict)
continue
if serverCode in existing_server_codes:
errordict['serverCode'] = f'serverCode 重复----{serverCode}'
error_import_list.append(errordict)
continue
serversUsage =rowvalue[5]
if serversUsage != None:
serversUsage =rowvalue[5].replace(' ', '')
else:
errordict['serversUsage'] = f'设备用途错误----{serversUsage}'
error_import_list.append(errordict)
continue
servers_Info_obj=False
try:
servers_Info_obj = ServersInfo.objects.get(serverCode=serverCode)
except :
print('')
if servers_Info_obj:
errordict['serverCode'] = f'serverCode重复----{serverCode}'
error_import_list.append(errordict)
continue
serversModel=''
index = serverCode.find('-')
print()
if index != -1:
serversModel=serverCode[:index]
else:
serversModel = serverCode
try:
servers_model_obj = ServersModel.objects.get(name=serversModel)
except ServersModel.DoesNotExist as e:
errordict['ServersModel']=f'设备机型不在机型列表中---{serversModel}'
error_import_list.append(errordict)
continue
try:
servers_Usage_obj =ServersUsage.objects.get(name=rowvalue[5])
except ServersUsage.DoesNotExist as e:
errordict['ServersUsage'] = f'设备用途不在用途列表中---{rowvalue[5]}'
error_import_list.append(errordict)
continue
borrowFromtime = rowvalue[9]
borrowEndtime = rowvalue[10]
if borrowFromtime!=None:
borrowFromtime=borrowFromtime.replace(' ', '').replace('\ufeff','')
if borrowFromtime!=''and is_valid_date(borrowFromtime) == False:
errordict['borrowFromtime'] = f'borrowFromtime日期格式不对---{borrowFromtime}'
error_import_list.append(errordict)
continue
if borrowFromtime == '':
borrowFromtime = None
if borrowEndtime != None:
borrowEndtime= borrowEndtime.replace(' ', '').replace('\ufeff','')
if borrowEndtime!='' and is_valid_date(borrowEndtime) == False:
errordict['borrowEndtime'] = f'borrowEndtime日期格式不对---{borrowEndtime}'
error_import_list.append(errordict)
continue
if borrowEndtime == '':
borrowEndtime = None
borrowUser = rowvalue[8]
if borrowUser != None:
borrowUser=borrowUser.replace(' ', '').replace('\ufeff','')
if len(borrowUser) > 130:
errordict['borrowUser'] = f'borrowUser数据错误---{borrowUser}'
error_import_list.append(errordict)
continue
if borrowUser=='':
borrowUser = None

try:
data_dict = {
'serverFrom':rowvalue[1] ,
'serverCode':serverCode,
'serversModel':servers_model_obj ,
'location':rowvalue[3],
'hardware':rowvalue[4],
'serversUsage':servers_Usage_obj,
'bmcIP':rowvalue[6],
'hostIP':rowvalue[7],
'borrowUser':borrowUser,
'borrowFromtime':borrowFromtime,
'borrowEndtime':borrowEndtime,
'borrowUse':rowvalue[11],
'notes':rowvalue[12]
}
obj = ServersInfo(**data_dict)
data_import_list.append(obj)
existing_server_codes.add(serverCode) # 将当前 serverCode 添加到已存在集合中
except Exception as e:
errordict['datachange'] = e
error_import_list.append(errordict)
continue
lenObjects=len(data_import_list)
lenOerror=len(error_import_list)
ServersInfo.objects.bulk_create(data_import_list)
return SuccessResponse(data= error_import_list, msg=f"导入{lenObjects}行,错误行数{lenOerror}")