Skip to content

Commit

Permalink
add sourecry guard close
Browse files Browse the repository at this point in the history
  • Loading branch information
MartinBelthle committed Oct 18, 2023
1 parent 2b5d277 commit ad9bfff
Showing 1 changed file with 46 additions and 40 deletions.
86 changes: 46 additions & 40 deletions antareslauncher/use_cases/retrieve/final_zip_extractor.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,43 +20,49 @@ def extract_final_zip(self, study: StudyDTO) -> None:
Args:
study: The current study
"""
if study.finished and not study.with_error and study.local_final_zipfile_path and not study.final_zip_extracted:
zip_path = Path(study.local_final_zipfile_path)
try:
with zipfile.ZipFile(zip_path) as zf:
names = zf.namelist()
if len(names) > 1 and os.path.commonpath(names):
# If all files are in the same directory, we can extract the ZIP
# file directly in the target directory.
target_dir = zip_path.parent
progress_bar = self._display.generate_progress_bar(
names, desc="Extracting archive:", total=len(names)
)
for file in progress_bar:
zf.extract(member=file, path=target_dir)
else:
# The directory is already an output and does not need to be unzipped.
# All we have to do is rename it by removing the prefix "_finished"
# and the suffix "job_id" that lies before the ".zip".
# If these prefix/suffix prefix change, this code needs to be adapted.
zip_path.rename((zip_path.parent / (zip_path.name[9:zip_path.name.rfind("_")] + ".zip")))

except (OSError, zipfile.BadZipFile) as exc:
# If we cannot extract the final ZIP file, either because the file
# doesn't exist or the ZIP file is corrupted, we find ourselves
# in a situation where the results are unusable.
# In such cases, it's best to consider the simulation as failed,
# enabling the user to restart its simulation.
study.final_zip_extracted = False
study.with_error = True
self._display.show_error(
f'"{study.name}": Final zip not extracted: {exc}',
LOG_NAME,
)

else:
study.final_zip_extracted = True
self._display.show_message(
f'"{study.name}": Final zip extracted',
LOG_NAME,
)
if (
not study.finished
or study.with_error
or not study.local_final_zipfile_path
or study.final_zip_extracted
):
return
zip_path = Path(study.local_final_zipfile_path)
try:
with zipfile.ZipFile(zip_path) as zf:
names = zf.namelist()
if len(names) > 1 and os.path.commonpath(names):
# If all files are in the same directory, we can extract the ZIP
# file directly in the target directory.
target_dir = zip_path.parent
progress_bar = self._display.generate_progress_bar(
names, desc="Extracting archive:", total=len(names)
)
for file in progress_bar:
zf.extract(member=file, path=target_dir)
else:
# The directory is already an output and does not need to be unzipped.
# All we have to do is rename it by removing the prefix "_finished"
# and the suffix "job_id" that lies before the ".zip".
# If these prefix/suffix prefix change, this code needs to be adapted.
zip_path.rename((zip_path.parent / (zip_path.name[9:zip_path.name.rfind("_")] + ".zip")))

except (OSError, zipfile.BadZipFile) as exc:
# If we cannot extract the final ZIP file, either because the file
# doesn't exist or the ZIP file is corrupted, we find ourselves
# in a situation where the results are unusable.
# In such cases, it's best to consider the simulation as failed,
# enabling the user to restart its simulation.
study.final_zip_extracted = False
study.with_error = True
self._display.show_error(
f'"{study.name}": Final zip not extracted: {exc}',
LOG_NAME,
)

else:
study.final_zip_extracted = True
self._display.show_message(
f'"{study.name}": Final zip extracted',
LOG_NAME,
)

0 comments on commit ad9bfff

Please sign in to comment.