Skip to content

Commit

Permalink
change data type checker
Browse files Browse the repository at this point in the history
  • Loading branch information
@Gahhh
Gahhh committed Nov 27, 2023
1 parent c87cded commit 8bfadf9
Show file tree
Hide file tree
Showing 2 changed files with 81 additions and 72 deletions.
18 changes: 8 additions & 10 deletions src/content_uploader/ContentUploader.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down
135 changes: 73 additions & 62 deletions src/data_service/ContentManager.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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 = {
Expand Down

0 comments on commit 8bfadf9

Please sign in to comment.