from enum import Enum import pathlib from helix.constants.file_validation_error import FileValidationError from helix.constants.file_validation_error import FileValidationMessage from helix.validators.csv_input_validator import CsvInputValidator from helix.validators.dxf_input_validator import DxfInputValidator class UnknownFileValidator(object): def __init__(self, _): pass def validate(self, _): return FileValidationError(FileValidationMessage.UnknownFileUploaded, None) class FileType(Enum): Csv = 0 AuroraDxf = 1 Unknown = 2 @property def validator(self): """Provide the proper implementation of the validation for the selected FileType """ return { FileType.Csv: CsvInputValidator, FileType.AuroraDxf: DxfInputValidator, FileType.Unknown: UnknownFileValidator }.get(self) def valid_mapping(self, extension): """Validates the extension obtained file Arguments; extension: string Returns: boolean """ try: mapping = {FileType.Csv: ["txt", "csv"], FileType.AuroraDxf: ["dxf"], FileType.Unknown: [None]}.get(self) assert extension in mapping except AssertionError: raise InvalidMappingException def get_invalid_mapping_error(self): """Get the validation message when choosing the wrong file for the selected FileType Returns: dict """ mapping = {FileType.Csv: FileValidationMessage.ExpectedTxtFile, FileType.AuroraDxf: FileValidationMessage.ExpectedDxfFile, FileType.Unknown: FileValidationMessage.UnknownFileUploaded}.get(self) return mapping class FileValidator(object): """Validates that the received stream belongs to a valid FileType """ def __init__(self, user_values): self.values = user_values def obtain_stream(self, path): """Obtain the content of the file Argument: path (str): A path that will be opened Returns: str The content of the file, otherwise an empty string """ content = "" try: content = path.read().decode("utf-8") except UnicodeDecodeError: pass finally: return content def validate(self, stream, expected, file=None, extension=None): """Validates the uploaded file by extension and content Arguments; stream (string): File content file (FileObject): A file object provided from the ui. Used only to get the file extension. """ try: file_type = self.identify_file_type(stream) assert file_type == expected validator = file_type.validator(self.values) if extension is None: extension = self.obtain_extension(file) file_type.valid_mapping(extension) return validator.validate(stream) except AssertionError: return FileValidationError(expected.get_invalid_mapping_error(), None) except InvalidMappingException: error_type = file_type.get_invalid_mapping_error() return FileValidationError(error_type, None) except IndexError: error_type = FileValidationMessage.UnknownFileUploaded return FileValidationError(error_type, None) except InvalidExtensionException: error_type = FileValidationMessage.UnknownFileUploaded return FileValidationError(error_type, None) def obtain_extension(self, file_object): """Obtain the extension from the user uploaded file Arguments: file_object (file): A file like object """ extension = pathlib.Path(file_object.filename).suffix file_pieces = extension.split(".") return file_pieces[1] def identify_file_type(self, file): """Determines the FileType object based on the beginning of the provided string Arguments; file (str): The raw stream Returns: enum """ if file.startswith("999\r\nDesign created by Aurora"): return FileType.AuroraDxf if file.startswith("HANDLE\t"): return FileType.Csv return FileType.Unknown class InvalidExtensionException(Exception): pass class InvalidMappingException(Exception): pass