factory/apps/enp/views.py

from django.shortcuts import render
from apps.utils.viewsets import CustomModelViewSet, CustomGenericViewSet
from apps.utils.mixins import BulkCreateModelMixin, BulkDestroyModelMixin, CustomListModelMixin
from .serializers import (DrainSerializer, DrainEquipSerializer, DrainEquipEnvSerializer, 
                          VehicleAccessSerializer, EnvDataSerializer, EnvDataExportSerializer, 
                          CarWashSerializer, DrainWithEquipEnpSerializer, DrainWithEquipBaseSerializer)
from .models import Drain, DrainEquip, VehicleAccess, EnvData, CarWash
from rest_framework.decorators import action
from apps.utils.sql import query_all_dict
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from rest_framework.response import Response
from django.db.models import Count, Case, When, IntegerField
# Create your views here.


class DrainViewSet(CustomModelViewSet):
    """
    list: 排口/污染源

    排口/污染源
    """

    queryset = Drain.objects.all()
    serializer_class = DrainSerializer
    filterset_fields = ["type", "cate", "mgroup", "level"]
    select_related_fields = ["mgroup"]

    def get_serializer_class(self):
        equip_data = self.request.query_params.get("equip_data", "no")
        if self.request.method == "GET":
            if equip_data == "base":
                return DrainWithEquipBaseSerializer
            elif equip_data == "enp":
                return DrainWithEquipEnpSerializer
        return super().get_serializer_class()

    @swagger_auto_schema(
        manual_parameters=[
            openapi.Parameter(name="equip_data", in_=openapi.IN_QUERY, description="Include equip data in the response", type=openapi.TYPE_STRING, enum=["no", "base", "enp"], required=False),
            openapi.Parameter(name="query", in_=openapi.IN_QUERY, description="定制返回数据", type=openapi.TYPE_STRING, required=False),
        ]
    )
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

    @action(methods=["get"], detail=False, perms_map={"get": "*"})
    def count_cate(self, request, *args, **kwargs):
        """排口分类数量

        排口分类数量
        """
        queryset = self.filter_queryset(self.get_queryset())
        result = queryset.aggregate(
            count=Count("id"),
            count_product=Count(Case(When(cate="product", then=1), output_field=IntegerField())),
            count_mtrans=Count(Case(When(cate="mtrans", then=1), output_field=IntegerField())),
            count_mstore=Count(Case(When(cate="mstore", then=1), output_field=IntegerField())),
        )
        json_result = {"count": result["count"], "count_product": result["count_product"], "count_mtrans": result["count_mtrans"], "count_mstore": result["count_mstore"]}

        return Response(json_result)


class DrainEquipViewSet(CustomListModelMixin, BulkCreateModelMixin, BulkDestroyModelMixin, CustomGenericViewSet):
    """
    list: 排口/设备关系

    排口/设备关系
    """

    perms_map = {"get": "*", "post": "drain.update", "delete": "drain.update"}
    queryset = DrainEquip.objects.all()
    serializer_class = DrainEquipSerializer
    select_related_fields = ["equipment", "drain"]
    filterset_fields = ["drain", "equipment", "drain__mgroup", "equipment__mgroup", "drain__type", "equipment__type"]

    def get_serializer_class(self):
        has_envdata = self.request.query_params.get("has_envdata", "no")
        if self.request.method == "GET" and has_envdata == "yes":
            return DrainEquipEnvSerializer
        return super().get_serializer_class()

    @swagger_auto_schema(
        manual_parameters=[
            openapi.Parameter(name="has_envdata", in_=openapi.IN_QUERY, description="Include environmental data in the response", type=openapi.TYPE_STRING, enum=["yes", "no"], required=False),
        ]
    )
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)


class VehicleAccessViewSet(CustomListModelMixin, CustomGenericViewSet):
    """
    list: 车辆出入记录

    车辆出入记录
    """

    perms_map = {"get": "*"}
    queryset = VehicleAccess.objects.all()
    serializer_class = VehicleAccessSerializer
    filterset_fields = {
        "vehicle_number": ["icontains"],
        "emission_standard": ["exact", "in"],
        "type": ["exact", "in"],
        "is_new_energy": ["exact"],
        "access_time": ["gte", "lte", "year", "month", "day", "quarter", "week"],
    }


class EnvDataViewSet(CustomListModelMixin, CustomGenericViewSet):
    """
    list: 环保时序数据

    环保时序数据
    """

    perms_map = {"get": "*"}
    queryset = EnvData.objects.all()
    serializer_class = EnvDataSerializer
    filterset_fields = {
        "timex": ["exact", "gte", "lte", "year", "month", "day"],
        "equipment": ["exact"],
    }
    ordering_fields = ["time"]
    ordering = ["-time"]

    @action(methods=["post"], detail=False, perms_map={"post": "*"}, serializer_class=EnvDataExportSerializer)
    def export_excel(self, request, *args, **kwargs):
        """导出Excel

        导出Excel
        """
        return Response()


class CarWashViewSet(CustomListModelMixin, CustomGenericViewSet):
    """
    list: 洗车记录

    洗车记录
    """

    perms_map = {"get": "*"}
    queryset = CarWash.objects.all()
    serializer_class = CarWashSerializer
    select_related_fields = ["station"]
    filterset_fields = {
        "station": ["exact"],
        "start_time": ["exact", "gte", "lte", "year", "month", "day"],
    }
    ordering = ["-start_time"]