""" 出入库模块视图 - 管理端 使用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)