mirror of https://gitee.com/anolis/sysom.git
213 lines
8.1 KiB
Python
213 lines
8.1 KiB
Python
import os
|
|
import tempfile
|
|
import pandas as pd
|
|
from clogger import logger
|
|
from django.utils import timezone
|
|
from django.core.files.uploadedfile import InMemoryUploadedFile
|
|
from rest_framework import serializers
|
|
from apps.hotfix.models import HotfixModel, OSTypeModel, KernelVersionModel, ReleasedHotfixListModule
|
|
|
|
class HotfixSerializer(serializers.ModelSerializer):
|
|
|
|
class Meta:
|
|
model = HotfixModel
|
|
fields = '__all__' # fields 指定从数据库返回的字段
|
|
|
|
|
|
class OSTypeSerializer(serializers.ModelSerializer):
|
|
|
|
class Meta:
|
|
model = OSTypeModel
|
|
fields = '__all__'
|
|
|
|
class KernelSerializer(serializers.ModelSerializer):
|
|
|
|
class Meta:
|
|
model = KernelVersionModel
|
|
fields = '__all__'
|
|
|
|
class ReleasedHotfixSerializer(serializers.ModelSerializer):
|
|
|
|
deprecated = serializers.SerializerMethodField()
|
|
modified_time = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S")
|
|
released_time = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S")
|
|
|
|
class Meta:
|
|
model = ReleasedHotfixListModule
|
|
fields = '__all__'
|
|
|
|
def get_deprecated(self, obj): return bool(obj.deprecated)
|
|
|
|
|
|
class CreateReleasedHotfixSerializer(serializers.ModelSerializer):
|
|
|
|
deprecated = serializers.BooleanField(default=False)
|
|
deprecated_info = serializers.CharField(default="")
|
|
|
|
class Meta:
|
|
model = ReleasedHotfixListModule
|
|
exclude = ['modified_time', 'modified_user']
|
|
|
|
def validate(self, attrs):
|
|
deprecated_info = attrs.get('deprecated_info', "")
|
|
deprecated = attrs.get('deprecated', False)
|
|
if not deprecated and len(deprecated_info) > 0:
|
|
raise serializers.ValidationError('Field deprecated is false, deprecated_info is empty!')
|
|
|
|
if deprecated and len(deprecated_info) == 0:
|
|
raise serializers.ValidationError("Field deprecated is true, field deprecated_info required!")
|
|
|
|
return super().validate(attrs)
|
|
|
|
def validate_hotfix_id(self, attrs):
|
|
instance = ReleasedHotfixListModule.objects.filter(hotfix_id=attrs).first()
|
|
if instance is None:
|
|
return super().validate(attrs)
|
|
else:
|
|
raise serializers.ValidationError(detail='Field hotfix_id does not exist')
|
|
|
|
def validate_released_kernel_version(self, attrs):
|
|
instance = ReleasedHotfixListModule.objects.filter(released_kernel_version=attrs).first()
|
|
if instance is None:
|
|
return super().validate(attrs)
|
|
else:
|
|
raise serializers.ValidationError(detail='Field released_kernel_version does not exist')
|
|
|
|
|
|
class UpdateReleasedHotfixSerializer(serializers.ModelSerializer):
|
|
released_kernel_version = serializers.CharField(required=False)
|
|
hotfix_id = serializers.CharField(required=False)
|
|
released_time = serializers.DateTimeField(required=False)
|
|
|
|
class Meta:
|
|
model = ReleasedHotfixListModule
|
|
exclude = ['modified_time', 'modified_user']
|
|
|
|
def validate_hotfix_id(self, attrs):
|
|
"""Filter the data named "hotfix_id" so that it cannot be repeated in the database"""
|
|
instance = ReleasedHotfixListModule.objects.filter(hotfix_id=attrs).first()
|
|
if instance:
|
|
raise serializers.ValidationError(f"field hotfix_id: {attrs} Exist")
|
|
else:
|
|
return attrs
|
|
|
|
def validate(self, attrs):
|
|
"""
|
|
Filter the data named "released_kernel_version"
|
|
so that it cannot be repeated in the database
|
|
"""
|
|
hotfix_id=attrs.get("hotfix_id",None)
|
|
released_kernel_version=attrs.get("released_kernel_version",None)
|
|
if hotfix_id and released_kernel_version:
|
|
instance_kernel_version=ReleasedHotfixListModule.objects.filter(**attrs)
|
|
if instance_kernel_version:
|
|
raise serializers.ValidationError(f"released_kernel_version:{released_kernel_version} is already existed!")
|
|
return attrs
|
|
|
|
|
|
class UpdatePutReleasedHotfixSerializer(serializers.ModelSerializer):
|
|
deprecated = serializers.BooleanField()
|
|
|
|
class Meta:
|
|
model = ReleasedHotfixListModule
|
|
exclude = ['modified_time', 'modified_user']
|
|
|
|
def validate_deprecated(self, attrs):
|
|
return 1 if attrs else 0
|
|
|
|
def validate(self, attrs):
|
|
deprecated_info = attrs.get('deprecated_info', "")
|
|
deprecated = attrs.get('deprecated', False)
|
|
if not deprecated and len(deprecated_info) > 0:
|
|
raise serializers.ValidationError('Field deprecated is false, deprecated_info is empty!')
|
|
|
|
if deprecated and len(deprecated_info) == 0:
|
|
raise serializers.ValidationError("Field deprecated is true, field deprecated_info required!")
|
|
return super().validate(attrs)
|
|
|
|
|
|
class BulkImportHotfixReleasedSerializer(serializers.Serializer):
|
|
file = serializers.FileField(required=True)
|
|
|
|
def __init__(self, instance=None, data=..., **kwargs):
|
|
super().__init__(instance, data, **kwargs)
|
|
self._file: InMemoryUploadedFile = None
|
|
self._save_temporary_file = None
|
|
self._suffix: str = None
|
|
self._save_file_path = None
|
|
self._default_save_path = '/tmp/'
|
|
self._suffixs = ['xls', 'csv', 'xlsx']
|
|
self._action = {
|
|
'csv': pd.read_csv,
|
|
'xls': pd.read_excel,
|
|
'xlsx': pd.read_excel
|
|
}
|
|
|
|
def validate_file(self, file: InMemoryUploadedFile):
|
|
"""validate param `file`"""
|
|
file_name = file.name
|
|
self._suffix = file_name.split('.')[-1]
|
|
|
|
if self._suffix not in self._suffixs:
|
|
raise serializers.ValidationError('file suffix invaild!')
|
|
self._file = file
|
|
|
|
self._save_upload_file()
|
|
try:
|
|
context = self._file_parse()
|
|
except Exception as e:
|
|
raise serializers.ValidationError(f'parse file faid! Error: {e}')
|
|
finally:
|
|
self.close_file()
|
|
return context
|
|
|
|
def _save_upload_file(self):
|
|
"""Save the file to a temporary file"""
|
|
self._save_temporary_file = tempfile.NamedTemporaryFile(dir=self._default_save_path)
|
|
self._save_file_path = os.path.join(self._default_save_path, self._save_temporary_file.name)
|
|
|
|
with open(self._save_file_path, 'wb') as f:
|
|
for chunk in self._file.chunks(chunk_size=1024):
|
|
f.write(chunk)
|
|
|
|
def _file_parse(self) -> [dict]:
|
|
"""parse the file"""
|
|
if self._suffix is None:
|
|
raise serializers.ValidationError("file suffix is not none!")
|
|
|
|
if self._save_file_path is None:
|
|
raise serializers.ValidationError('file save file is not none!')
|
|
|
|
context = self._action[self._suffix](self._save_file_path)
|
|
return context
|
|
|
|
def create(self, validated_data):
|
|
"""Save the data to the database"""
|
|
file_data=validated_data.get("file")
|
|
file_header_fields=["hotfix_id","modified_time","released_kernel_version"]
|
|
file_headers=set(filter(lambda x: x in file_header_fields, list(file_data)))
|
|
if not file_headers == set(file_header_fields):
|
|
raise serializers.ValidationError(f"upload files must have {file_header_fields} ,your file only have {file_headers},please check your file!")
|
|
data_list=file_data.to_dict(orient='records')
|
|
instance_list = []
|
|
for data in data_list:
|
|
data.pop("status")
|
|
if not data["hotfix_id"] or not data["modified_time"] or not data["released_kernel_version"]:
|
|
continue
|
|
modified_time=data['modified_time']
|
|
if isinstance(modified_time, str):
|
|
modified_time = modified_time.rsplit('.')[0]
|
|
data['released_time'] = modified_time if modified_time else timezone.now()
|
|
|
|
if not ReleasedHotfixListModule.objects.filter(
|
|
hotfix_id=data["hotfix_id"],
|
|
released_kernel_version=data['released_kernel_version']
|
|
):
|
|
instance_list.append(ReleasedHotfixListModule(**data))
|
|
return ReleasedHotfixListModule.objects.bulk_create(instance_list)
|
|
|
|
def close_file(self):
|
|
"""delete the temporary file!"""
|
|
if self._save_temporary_file:
|
|
self._save_temporary_file.close()
|