Skip to content
Snippets Groups Projects

UI for creating follow on imaging requests from completed calibration requests

Merged Charlotte Hausman requested to merge Restore_Imaging_Capability into main
Files
37
@@ -30,8 +30,8 @@ def get_fields_for(product_type: str, filename: str) -> list:
"creationTime",
"productLocator",
"destinationDirectory",
"cms_path",
"sdmId",
"calibrationSourceDirectory",
"cmsName",
]
restore_metadata_list = [
@@ -74,8 +74,9 @@ class AuditFiles(AuditorIF):
def check_required_fields(self, file: AbstractTextFile, fields: list) -> bool:
missing = []
content = file.content
if ".xml" in file.filename:
ppr_content = BeautifulSoup(file.content, "xml")
ppr_content = BeautifulSoup(content, "xml")
for tag in fields:
try:
ppr_content.find(tag).string
@@ -84,8 +85,7 @@ class AuditFiles(AuditorIF):
if ".json" in file.filename:
for tag in fields:
metadata = json.loads(file.content)
if tag not in metadata or len(metadata[tag]) == 0:
if tag not in content:
missing.append(tag)
if len(missing) > 0:
print(f"Missing fields: {missing}")
@@ -110,7 +110,7 @@ class AuditFiles(AuditorIF):
def audit(self) -> bool:
invalid_files = []
for file in self.files:
self.logger.info(f"Auditing file {file}...")
self.logger.info(f"Auditing file {file.filename}...")
if file.filename == "PPR.xml":
self.logger.info("Correcting PPR.xml for condor processing...")
@@ -123,7 +123,7 @@ class AuditFiles(AuditorIF):
invalid_files.append(file.filename)
if len(invalid_files) != 0:
self.logger.info(f"INVALID FILE FOUND: {invalid_files}")
self.logger.error(f"INVALID FILE FOUND: {invalid_files}")
return False
else:
return True
@@ -142,7 +142,7 @@ class AuditDirectories(AuditorIF):
current = os.getcwd()
needed = self.rootDirectory + "/" + self.relative_path
if needed != current:
self.logger.info("DIRECTORY ERROR: not in correct directory for processing.")
self.logger.error("DIRECTORY ERROR: not in correct directory for processing.")
return False
else:
working = Path(current + "/working").is_dir()
@@ -155,14 +155,14 @@ class AuditDirectories(AuditorIF):
if len(data) > 0:
self.logger.info("Data is available. Proceeding...")
if self.parameters["product_type"] is "restore":
if self.parameters["product_type"] == "restore":
self.logger.info("Checking products/ for calibration tables...")
cal_data = os.listdir(Path(current + "/products/"))
if len(cal_data) > 0:
self.logger.info("Calibration data is available. Proceeding...")
else:
self.logger.error("FAILURE: calibration data not found in products/")
return False
else:
self.logger.error("FAILURE: calibration data not found in products/")
return False
return True
else:
Loading