88b8f023f4
Some checks failed
Build and Deploy Backend / build-and-deploy (push) Failing after 1m36s
263 lines
9.2 KiB
Python
263 lines
9.2 KiB
Python
"""
|
||
出入库模块视图 - 管理端
|
||
使用AdminJWTAuthentication进行认证
|
||
"""
|
||
from django.http import HttpResponse
|
||
from rest_framework import viewsets, status
|
||
from rest_framework.decorators import action
|
||
from rest_framework.parsers import MultiPartParser
|
||
from drf_spectacular.utils import extend_schema
|
||
|
||
from utils.response import success, error
|
||
from utils.exceptions import ErrorCode
|
||
from apps.admins.authentication import AdminJWTAuthentication
|
||
from apps.admins.permissions import IsAdminUser
|
||
from apps.devices.models import DeviceType, DeviceBatch, Device
|
||
from apps.devices.serializers import (
|
||
DeviceTypeSerializer,
|
||
DeviceBatchSerializer,
|
||
DeviceBatchCreateSerializer,
|
||
DeviceSimpleSerializer
|
||
)
|
||
from .services import generate_batch_devices, export_batch_excel, import_mac_addresses
|
||
|
||
|
||
@extend_schema(tags=['管理员-库存'])
|
||
class DeviceTypeViewSet(viewsets.ModelViewSet):
|
||
"""设备类型管理视图集 - 管理端"""
|
||
|
||
queryset = DeviceType.objects.all().order_by('-created_at')
|
||
serializer_class = DeviceTypeSerializer
|
||
authentication_classes = [AdminJWTAuthentication]
|
||
permission_classes = [IsAdminUser]
|
||
|
||
def list(self, request, *args, **kwargs):
|
||
"""
|
||
设备类型列表
|
||
GET /api/admin/device-types
|
||
"""
|
||
queryset = self.filter_queryset(self.get_queryset())
|
||
|
||
# 过滤
|
||
brand = request.query_params.get('brand')
|
||
is_active = request.query_params.get('is_active')
|
||
|
||
if brand:
|
||
queryset = queryset.filter(brand__contains=brand)
|
||
if is_active is not None:
|
||
queryset = queryset.filter(is_active=is_active.lower() == 'true')
|
||
|
||
page = self.paginate_queryset(queryset)
|
||
if page is not None:
|
||
serializer = self.get_serializer(page, many=True)
|
||
return success(data={
|
||
'total': self.paginator.page.paginator.count,
|
||
'items': serializer.data
|
||
})
|
||
|
||
serializer = self.get_serializer(queryset, many=True)
|
||
return success(data={'items': serializer.data})
|
||
|
||
def create(self, request, *args, **kwargs):
|
||
"""
|
||
创建设备类型
|
||
POST /api/admin/device-types
|
||
"""
|
||
serializer = self.get_serializer(data=request.data)
|
||
if serializer.is_valid():
|
||
serializer.save()
|
||
return success(data=serializer.data, message='创建成功')
|
||
return error(message=str(serializer.errors))
|
||
|
||
def retrieve(self, request, *args, **kwargs):
|
||
"""
|
||
设备类型详情
|
||
GET /api/admin/device-types/{id}
|
||
"""
|
||
instance = self.get_object()
|
||
serializer = self.get_serializer(instance)
|
||
return success(data=serializer.data)
|
||
|
||
def update(self, request, *args, **kwargs):
|
||
"""
|
||
更新设备类型
|
||
PUT /api/admin/device-types/{id}
|
||
"""
|
||
instance = self.get_object()
|
||
serializer = self.get_serializer(instance, data=request.data, partial=True)
|
||
if serializer.is_valid():
|
||
serializer.save()
|
||
return success(data=serializer.data, message='更新成功')
|
||
return error(message=str(serializer.errors))
|
||
|
||
|
||
@extend_schema(tags=['管理员-库存'])
|
||
class DeviceBatchViewSet(viewsets.ModelViewSet):
|
||
"""设备批次管理视图集 - 管理端"""
|
||
|
||
queryset = DeviceBatch.objects.all().select_related('device_type').order_by('-created_at')
|
||
authentication_classes = [AdminJWTAuthentication]
|
||
permission_classes = [IsAdminUser]
|
||
|
||
def get_serializer_class(self):
|
||
if self.action == 'create':
|
||
return DeviceBatchCreateSerializer
|
||
return DeviceBatchSerializer
|
||
|
||
def list(self, request, *args, **kwargs):
|
||
"""
|
||
批次列表
|
||
GET /api/admin/device-batches
|
||
"""
|
||
queryset = self.filter_queryset(self.get_queryset())
|
||
|
||
# 过滤
|
||
device_type_id = request.query_params.get('device_type_id')
|
||
brand = request.query_params.get('brand')
|
||
batch_no = request.query_params.get('batch_no')
|
||
status_filter = request.query_params.get('status')
|
||
|
||
if device_type_id:
|
||
queryset = queryset.filter(device_type_id=device_type_id)
|
||
if brand:
|
||
queryset = queryset.filter(device_type__brand__contains=brand)
|
||
if batch_no:
|
||
queryset = queryset.filter(batch_no__contains=batch_no)
|
||
if status_filter:
|
||
queryset = queryset.filter(status=status_filter)
|
||
|
||
page = self.paginate_queryset(queryset)
|
||
if page is not None:
|
||
serializer = DeviceBatchSerializer(page, many=True)
|
||
return success(data={
|
||
'total': self.paginator.page.paginator.count,
|
||
'items': serializer.data
|
||
})
|
||
|
||
serializer = DeviceBatchSerializer(queryset, many=True)
|
||
return success(data={'items': serializer.data})
|
||
|
||
def create(self, request, *args, **kwargs):
|
||
"""
|
||
创建批次并生成SN码
|
||
POST /api/admin/device-batches
|
||
"""
|
||
serializer = DeviceBatchCreateSerializer(data=request.data)
|
||
if not serializer.is_valid():
|
||
return error(message=str(serializer.errors))
|
||
|
||
# 验证设备类型
|
||
device_type_id = serializer.validated_data['device_type'].id
|
||
try:
|
||
device_type = DeviceType.objects.get(id=device_type_id, is_active=True)
|
||
except DeviceType.DoesNotExist:
|
||
return error(code=ErrorCode.DEVICE_TYPE_NOT_FOUND, message='设备类型不存在或已禁用')
|
||
|
||
# 创建批次（注意：这里created_by需要存AdminUser的引用，暂时设为None）
|
||
batch = serializer.save(created_by=None)
|
||
|
||
# 生成SN码和设备
|
||
try:
|
||
devices = generate_batch_devices(batch)
|
||
except Exception as e:
|
||
batch.delete()
|
||
return error(message=f'生成SN码失败: {str(e)}')
|
||
|
||
# 返回结果
|
||
batch.refresh_from_db()
|
||
return success(data={
|
||
**DeviceBatchSerializer(batch).data,
|
||
'sn_range': {
|
||
'start': devices[0].sn if devices else '',
|
||
'end': devices[-1].sn if devices else ''
|
||
}
|
||
}, message='批次创建成功')
|
||
|
||
def retrieve(self, request, *args, **kwargs):
|
||
"""
|
||
批次详情
|
||
GET /api/admin/device-batches/{id}
|
||
"""
|
||
instance = self.get_object()
|
||
|
||
# 获取设备统计
|
||
devices = Device.objects.filter(batch=instance)
|
||
stats = {
|
||
'total': devices.count(),
|
||
'in_stock': devices.filter(status='in_stock').count(),
|
||
'out_stock': devices.filter(status='out_stock').count(),
|
||
'bound': devices.filter(status='bound').count(),
|
||
'with_mac': devices.exclude(mac_address__isnull=True).exclude(mac_address='').count()
|
||
}
|
||
|
||
return success(data={
|
||
**DeviceBatchSerializer(instance).data,
|
||
'statistics': stats
|
||
})
|
||
|
||
@action(detail=True, methods=['get'])
|
||
def devices(self, request, pk=None):
|
||
"""
|
||
批次设备列表
|
||
GET /api/admin/device-batches/{id}/devices
|
||
"""
|
||
batch = self.get_object()
|
||
devices = Device.objects.filter(batch=batch).order_by('sn')
|
||
|
||
page = self.paginate_queryset(devices)
|
||
if page is not None:
|
||
serializer = DeviceSimpleSerializer(page, many=True)
|
||
return success(data={
|
||
'total': self.paginator.page.paginator.count,
|
||
'items': serializer.data
|
||
})
|
||
|
||
serializer = DeviceSimpleSerializer(devices, many=True)
|
||
return success(data={'items': serializer.data})
|
||
|
||
@action(detail=True, methods=['get'])
|
||
def export(self, request, pk=None):
|
||
"""
|
||
导出批次SN码Excel
|
||
GET /api/admin/device-batches/{id}/export
|
||
"""
|
||
batch = self.get_object()
|
||
|
||
# 生成Excel
|
||
excel_file = export_batch_excel(batch)
|
||
|
||
# 返回文件
|
||
response = HttpResponse(
|
||
excel_file.getvalue(),
|
||
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
||
)
|
||
filename = f"batch_{batch.batch_no}_{batch.production_week}.xlsx"
|
||
response['Content-Disposition'] = f'attachment; filename="{filename}"'
|
||
|
||
return response
|
||
|
||
@action(detail=True, methods=['post'], parser_classes=[MultiPartParser])
|
||
def import_mac(self, request, pk=None):
|
||
"""
|
||
导入MAC地址
|
||
POST /api/admin/device-batches/{id}/import_mac
|
||
"""
|
||
batch = self.get_object()
|
||
|
||
file_obj = request.FILES.get('file')
|
||
if not file_obj:
|
||
return error(message='请上传Excel文件')
|
||
|
||
# 验证文件类型
|
||
if not file_obj.name.endswith(('.xlsx', '.xls')):
|
||
return error(message='仅支持Excel文件格式(.xlsx, .xls)')
|
||
|
||
# 导入MAC地址
|
||
result = import_mac_addresses(batch, file_obj)
|
||
|
||
message = f"导入完成，成功{result['success']}条"
|
||
if result['failed'] > 0:
|
||
message += f"，失败{result['failed']}条"
|
||
|
||
return success(data=result, message=message)
|