General-Platform-Backend/app/serializers.py

import copy
import datetime
import json
import random
import re
from decimal import Decimal

from django.db.models import F
from rest_framework import serializers
from django_filters.rest_framework import FilterSet
import django_filters
from app import consts, shortuuid
from app.models import AuthUser, ServerModel, AlgModelModel, ModelClassificationModel, EnterpriseModel, \
    EnterpriseDeviceModel, DeviceClassificationModel, WarningStatisticsModel, EnterpriseNodeModel, ComputeConfigModel, \
    ModelVersionModel, EnterpriseAlertDeviceModel, ModelHubModel, EnterpriseBusiModelModel, \
    EnterpriseBusiModelNodeModel, EnterpriseBusiModelNodeDeviceModel, ServerLogModel, FilesModel, MessageModel
import logging
logger = logging.getLogger('mylogger')


def md5_validate(v: str) -> bool:
    md5_pattern = re.compile(r'^[a-fA-F0-9]{32}$')
    # 匹配字符串
    if md5_pattern.match(v):
        return True

    return False


class AuthUserSerializer(serializers.ModelSerializer):
    class Meta:
        model = AuthUser
        fields = ["id", "username", "avatar", "is_superuser", "date_joined", "gender", "phone_number", "status", "note"]

    def to_representation(self, instance):

        ret = super().to_representation(instance)
        ret["gender_ch"] = instance.get_gender_display()
        return ret


class ServerSerializer(serializers.ModelSerializer):
    class Meta:
        model = ServerModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        ret["state"] = consts.device_status_map.get(ret["status"], None)

        # TODO 暂时模拟服务器使用情况
        ret["cpu"] = random.randint(20, 30)
        ret["mem"] = random.randint(20, 30)
        ret["storage"] = random.randint(20, 30)
        ret["gpu"] = random.randint(20, 30)

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret


class ServerLogSerializer(serializers.ModelSerializer):
    class Meta:
        model = ServerLogModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret


class EnterpriseSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseModel
        # fields = "__all__"
        exclude = ["pwd"]

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        ret["industry"] = consts.industry_map.get(ret["industry"], None)
        # ret["valid_data"] = f"{ret.pop('start_valid_date')}至{ret.pop('end_valid_date')}"

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        # 有效期无需更新
        else:
            data.pop("start_valid_date", None)
            data.pop("end_valid_date", None)

        ret = super().to_internal_value(data)
        return ret


class EnterpriseBusiModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseBusiModelModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)
        ret["basemodel_list"] = json.loads(ret.pop("base_models", "{}"))

        # node_id_list
        all_node_id_list = EnterpriseBusiModelNodeModel.objects.filter(busi_model_id=instance.id).all().values("node_id")
        node_ids = list()
        for node_id in all_node_id_list:
            node_ids.append(node_id.get("node_id"))

        all_node_objs = EnterpriseNodeModel.objects.filter(id__in=node_ids).all().annotate(node_id=F('id'), node_name=F('name')).values("node_id", "node_name")
        ret["link_node_list"] = all_node_objs

        return ret

    def to_internal_value(self, data):

        data = self.insert_busi_model(data)

        all_fields = self.get_fields()
        filtered_data = {key: value for key, value in data.items() if key in all_fields}

        ret = super().to_internal_value(filtered_data)
        return ret

    def insert_busi_model(self, data):
        if not data.get("entity_suid"):
            entity_obj = EnterpriseModel.objects.filter(id=data.get("entity_id")).last()
            entity_suid = entity_obj.suid if entity_obj else ""
            data['entity_suid'] = entity_suid

        if not (data.get("suid") or data.get("id")):
            data['suid'] = shortuuid.ShortUUID().random(10)

        base_model_ids = [int(model_id) for model_id in data['basemodel_ids'].split(',')]
        logging.info("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@")
        logging.info(base_model_ids)
        base_model = []
        for base_model_id in base_model_ids:
            base_model_info = AlgModelModel.objects.filter(id=base_model_id).get()
            logging.info("#####################")
            logging.info(base_model_info)
            base_model.append({
                'id': base_model_id,
                'suid': base_model_info.suid,
                'name': base_model_info.name,
            })
        data['base_models'] = json.dumps(base_model, ensure_ascii=False)
        new_data = copy.copy(data)

        return new_data


class EnterpriseBusiModelNodeSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseBusiModelNodeModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):

        ret = super().to_internal_value(data)
        return ret


class EnterpriseBusiModelNodeDeviceSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseBusiModelNodeDeviceModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret


class AlgModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = AlgModelModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        try:
            model_class = ModelClassificationModel.objects.filter(id=instance.classification).get()
        except:
            model_class = None

        ret["classification_name"] = model_class.name if model_class else None
        ret["model_type"] = consts.model_type_map[instance.model_type]
        ret["update_time"] = str(instance.update_time)

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret


class ModelClassificationSerializer(serializers.ModelSerializer):
    class Meta:
        model = ModelClassificationModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret


class ModelVersionSerializer(serializers.ModelSerializer):
    class Meta:
        model = ModelVersionModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):

        if not md5_validate(data.get("model_file")):
            raise Exception("模型文件格式错误")
        config_file = data.get("config_file")
        if config_file and not md5_validate(config_file):
            raise Exception("模型配置文件格式错误")

        status = data.get("status")
        # 删除时更新模型的版本
        if status:
            AlgModelModel.objects.filter(id=data.get("model_id")).update(default_version="")

        ret = super().to_internal_value(data)
        return ret

class ModelHubSerializer(serializers.ModelSerializer):
    class Meta:
        model = ModelHubModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret


class EnterpriseDeviceSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseDeviceModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        ##TODO 设备状态
        # ret["state"] = None
        try:
            classification_obj = DeviceClassificationModel.objects.filter(id=ret["classification"]).get()
        except:
            classification_obj = None

        ret["classification"] = classification_obj.name if classification_obj else None
        ret["is_cam"] = "摄像" in classification_obj.name if classification_obj else False

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret


class EnterpriseAlertDeviceSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseAlertDeviceModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret


class DeviceClassificationSerializer(serializers.ModelSerializer):
    class Meta:
        model = DeviceClassificationModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret


class WarningStatisticsSerializer(serializers.ModelSerializer):
    class Meta:
        model = WarningStatisticsModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        try:
            node_obj = EnterpriseNodeModel.objects.filter(id=instance.node_id).get()
        except:
            node_obj = None

        ret["node_name"] = node_obj.name if node_obj else ""

        return ret


class EnterpriseNodeListSerializer(serializers.ListSerializer):
    @property
    def data_with_children(self):
        data = super().data

        for single in data:
            children = EnterpriseNodeModel.objects.filter(parent=single.get("id")).all()

            if children:
                single["children"] = EnterpriseNodeSerializer(children, many=True).data_with_children
            else:
                single["children"] = []

        return data


class EnterpriseNodeSerializer(serializers.ModelSerializer):
    class Meta:
        model = EnterpriseNodeModel
        fields = "__all__"
        list_serializer_class = EnterpriseNodeListSerializer

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        # children = EnterpriseNodeModel.objects.filter(parent=instance.id).all()
        #
        # if children:
        #     ret["children"] = EnterpriseNodeSerializer(children, many=True).data

        return ret

    def to_internal_value(self, data):
        if not (data.get("suid") or data.get("id")):
            suid = shortuuid.ShortUUID().random(10)

            data["suid"] = suid

        ret = super().to_internal_value(data)
        return ret

    @property
    def data_with_children(self):
        data = super().data

        children = EnterpriseNodeModel.objects.filter(parent=data.get("id")).all()

        if children:
            data["children"] = EnterpriseNodeSerializer(children, many=True).data_with_children
        else:
            data["children"] = []

        return data


class ComputeConfigSerializer(serializers.ModelSerializer):
    class Meta:
        model = ComputeConfigModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        ret["percentage"] = f"{100 * instance.percentage}%"

        try:
            model_obj = AlgModelModel.objects.filter(id=instance.model_id).get()
        except:
            model_obj = None

        ret["name"] = model_obj.name if model_obj else None

        return ret

    def to_internal_value(self, data):
        percentage = data.get("percentage")
        if isinstance(percentage, str) and percentage.endswith("%"):
            data["percentage"] = (Decimal(percentage.strip("%")) / Decimal('100.0')).quantize(Decimal('0.00'))

        ret = super().to_internal_value(data)
        return ret


class FilesSerializer(serializers.ModelSerializer):
    class Meta:
        model = FilesModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):

        ret = super().to_internal_value(data)
        return ret


class MessageSerializer(serializers.ModelSerializer):
    class Meta:
        model = MessageModel
        fields = "__all__"

    def to_representation(self, instance):

        ret = super().to_representation(instance)

        return ret

    def to_internal_value(self, data):

        ret = super().to_internal_value(data)
        return ret