From 8bfadf9cfbe9877764e1b63d2435712b0264561a Mon Sep 17 00:00:00 2001 From: Yu Liang Date: Mon, 27 Nov 2023 15:25:31 +1100 Subject: [PATCH] change data type checker --- src/content_uploader/ContentUploader.py | 18 ++-- src/data_service/ContentManager.py | 135 +++++++++++++----------- 2 files changed, 81 insertions(+), 72 deletions(-) diff --git a/src/content_uploader/ContentUploader.py b/src/content_uploader/ContentUploader.py index 481b183..4e41a76 100644 --- a/src/content_uploader/ContentUploader.py +++ b/src/content_uploader/ContentUploader.py @@ -1138,16 +1138,14 @@ def _checkForAllowedFileTypes(self, path:str, allowedFileTypes:List[str], if os.path.isfile(path): mimeType:str = magic.from_file(filename=path, mime=True) # print(f"{path} has mime type: {mimeType}") - mimeTypePrefix = mimeType.split("/") - if mimeTypePrefix: - if mimeType in deniedFileTypes: - # print(f"Found {path} with MIME:{mimeType} in the denied list, not queuing file.") - return False - elif mimeTypePrefix[0] in allowedFileTypes: - # print(f"Found {mimeType} in the allowed list.") - return True - elif not allowedFileTypes: - return True + if mimeType in deniedFileTypes: + print(f"Found {path} with MIME:{mimeType} in the denied list, not queuing file.") + return False + elif mimeType in allowedFileTypes: + # print(f"Found {mimeType} in the allowed list.") + return True + elif not allowedFileTypes: + return True return False def _getNumberofFilesPathsAndTotalSize(self, paths:List[str], diff --git a/src/data_service/ContentManager.py b/src/data_service/ContentManager.py index ed684b6..6ce2d8d 100644 --- a/src/data_service/ContentManager.py +++ b/src/data_service/ContentManager.py @@ -16,6 +16,7 @@ import string from ClinicalTrials import ClinicalTrials from werkzeug.datastructures import MultiDict +from AccessManager import getTrialStructure class ContentManager: def __init__(self) -> None: @@ -687,74 +688,84 @@ def acceptAndSaveFile(self, req:request): print(returnMessage) return make_response(returnMessage) - fileTypeToPathMappingPath = f"{config.DATA_FILESYSTEM_ROOT}/" \ - f"{metadata['clinical_trial']}/" \ - f"{metadata['test_centre']}/" \ - f"metadata/paths.json" - if not os.path.isfile(fileTypeToPathMappingPath): - fileTypeToPathMappingPath = "templates/upload_paths_template.json" + # fileTypeToPathMappingPath = f"{config.DATA_FILESYSTEM_ROOT}/" \ + # f"{metadata['clinical_trial']}/" \ + # f"{metadata['test_centre']}/" \ + # f"metadata/paths.json" + # if not os.path.isfile(fileTypeToPathMappingPath): + # fileTypeToPathMappingPath = "templates/upload_paths_template.json" - with open(fileTypeToPathMappingPath, "r") as pathMappingFile: - fileTypeToPathMapping = json.load(pathMappingFile) + # with open(fileTypeToPathMappingPath, "r") as pathMappingFile: + # fileTypeToPathMapping = json.load(pathMappingFile) + try: + trialDetails = getTrialStructure(metadata["clinical_trial"]) + filePath = trialDetails[metadata["level"]][metadata["file_type"]]['path'] + except KeyError as e: + returnMessage = { + "status": "error", + "message": f"Invalid file type {metadata['file_type']} for {metadata['level']}" + } + print(returnMessage) + return make_response(returnMessage) filesSaved = [] for fileFieldName in req.files.keys(): uploadedFile = req.files[fileFieldName] filename = secure_filename(uploadedFile.filename) - formatedPath = os.path.basename(req.form["file_path"]).replace("\\", "/").replace(filename, "") - if metadata["clinical_trial"] == "CHIRP": - saveFolderPath = self._processCHIRP(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) - elif metadata["file_type"] == "fraction_folder": - saveFolderPath = self._processImageFractionFolder(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) - elif metadata["file_type"] == "DICOM_folder" or metadata["file_type"] == "DVH_folder": - saveFolderPath = self._processDoseReconstructionPlan(metadata, filename, uploadMetaData, fileTypeToPathMapping, filesSaved, formatedPath) - elif metadata["file_type"] == "triangulation_folder" or metadata["file_type"] == "kim_logs": - saveFolderPath = self._processTriangulationAndKimLogs(metadata, filename, uploadMetaData, fileTypeToPathMapping, filesSaved, formatedPath) - elif metadata["file_type"] == "image_folder": - saveFolderPath = self._processImagePatientFolder(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) - elif metadata["file_type"] == "trajectory_log_folder": - saveFolderPath = self._processTrajectoryLog(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) - else: - relativeFolderPath = uploadId + \ - fileTypeToPathMapping[metadata["file_type"]].format( - clinical_trial=metadata['clinical_trial'], - test_centre=metadata["test_centre"], - patient_trial_id=metadata["patient_trial_id"], - fraction_name=metadata["fraction"], - sub_fraction_name=metadata["sub_fraction"], - centre_patient_no=int(metadata["centre_patient_no"]) - ) - relativePath = relativeFolderPath + filename - saveFolderPath = config.UPLOAD_FOLDER + '/' + relativeFolderPath - filesSaved.append(relativePath) - - filePathAppended:bool = False - for uploadedFileRecord in uploadMetaData["uploaded_files"]: - if uploadedFileRecord["file_type"] == metadata["file_type"]: - uploadedFileRecord["Files"].append(relativePath) - filePathAppended = True - break - - if not filePathAppended: - uploadMetaData["uploaded_files"].append( - { - "file_type": metadata["file_type"], - "level": metadata["level"], - "fraction": metadata["fraction"], - "sub_fraction": metadata["sub_fraction"], - "Files": [relativePath] - } - ) - print(f"saving {filename} in {saveFolderPath}") - if saveFolderPath is not None: - if not os.path.isdir(saveFolderPath): - Path(saveFolderPath).mkdir(parents=True, exist_ok=True) - uploadedFile.save(os.path.join(saveFolderPath, filename)) - - with open(config.UPLOAD_FOLDER + '/' + uploadId + '/summary.txt', 'a') \ - as uploadSummaryFile: - for savedFilePath in filesSaved: - uploadSummaryFile.write(savedFilePath + "\n") + # formatedPath = os.path.basename(req.form["file_path"]).replace("\\", "/").replace(filename, "") + # if metadata["clinical_trial"] == "CHIRP": + # saveFolderPath = self._processCHIRP(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) + # elif metadata["file_type"] == "fraction_folder": + # saveFolderPath = self._processImageFractionFolder(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) + # elif metadata["file_type"] == "DICOM_folder" or metadata["file_type"] == "DVH_folder": + # saveFolderPath = self._processDoseReconstructionPlan(metadata, filename, uploadMetaData, fileTypeToPathMapping, filesSaved, formatedPath) + # elif metadata["file_type"] == "triangulation_folder" or metadata["file_type"] == "kim_logs": + # saveFolderPath = self._processTriangulationAndKimLogs(metadata, filename, uploadMetaData, fileTypeToPathMapping, filesSaved, formatedPath) + # elif metadata["file_type"] == "image_folder": + # saveFolderPath = self._processImagePatientFolder(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) + # elif metadata["file_type"] == "trajectory_log_folder": + # saveFolderPath = self._processTrajectoryLog(metadata, filename, uploadMetaData, filesSaved, fileTypeToPathMapping, formatedPath) + # else: + relativeFolderPath = uploadId + \ + filePath.format( + clinical_trial=metadata['clinical_trial'], + test_centre=metadata["test_centre"], + patient_trial_id=metadata["patient_trial_id"], + fraction_name=metadata["fraction"], + sub_fraction_name=metadata["sub_fraction"], + centre_patient_no=int(metadata["centre_patient_no"]) + ) + relativePath = relativeFolderPath + filename + saveFolderPath = config.UPLOAD_FOLDER + '/' + relativeFolderPath + filesSaved.append(relativePath) + + filePathAppended:bool = False + for uploadedFileRecord in uploadMetaData["uploaded_files"]: + if uploadedFileRecord["file_type"] == metadata["file_type"]: + uploadedFileRecord["Files"].append(relativePath) + filePathAppended = True + break + + if not filePathAppended: + uploadMetaData["uploaded_files"].append( + { + "file_type": metadata["file_type"], + "level": metadata["level"], + "fraction": metadata["fraction"], + "sub_fraction": metadata["sub_fraction"], + "Files": [relativePath] + } + ) + print(f"saving {filename} in {saveFolderPath}") + if saveFolderPath is not None: + if not os.path.isdir(saveFolderPath): + Path(saveFolderPath).mkdir(parents=True, exist_ok=True) + uploadedFile.save(os.path.join(saveFolderPath, filename)) + + with open(config.UPLOAD_FOLDER + '/' + uploadId + '/summary.txt', 'a') \ + as uploadSummaryFile: + for savedFilePath in filesSaved: + uploadSummaryFile.write(savedFilePath + "\n") else: # if not direct file upload, just metadata if "files" not in req.form.keys(): returnMessage = {