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"]