Extract file to the output directory. Expected file types are: `zip`, `tar.gz` and `tar`. Args: filepath: the file path of compressed file. output_dir: target directory to save extracted files. hash_val: expected hash value to validate the compressed file.
(
filepath: PathLike,
output_dir: PathLike = ".",
hash_val: str | None = None,
hash_type: str = "md5",
file_type: str = "",
has_base: bool = True,
)
| 301 | |
| 302 | |
| 303 | def extractall( |
| 304 | filepath: PathLike, |
| 305 | output_dir: PathLike = ".", |
| 306 | hash_val: str | None = None, |
| 307 | hash_type: str = "md5", |
| 308 | file_type: str = "", |
| 309 | has_base: bool = True, |
| 310 | ) -> None: |
| 311 | """ |
| 312 | Extract file to the output directory. |
| 313 | Expected file types are: `zip`, `tar.gz` and `tar`. |
| 314 | |
| 315 | Args: |
| 316 | filepath: the file path of compressed file. |
| 317 | output_dir: target directory to save extracted files. |
| 318 | hash_val: expected hash value to validate the compressed file. |
| 319 | if None, skip hash validation. |
| 320 | hash_type: 'md5' or 'sha1', defaults to 'md5'. |
| 321 | file_type: string of file type for decompressing. Leave it empty to infer the type from the filepath basename. |
| 322 | has_base: whether the extracted files have a base folder. This flag is used when checking if the existing |
| 323 | folder is a result of `extractall`, if it is, the extraction is skipped. For example, if A.zip is unzipped |
| 324 | to folder structure `A/*.png`, this flag should be True; if B.zip is unzipped to `*.png`, this flag should |
| 325 | be False. |
| 326 | |
| 327 | Raises: |
| 328 | RuntimeError: When the hash validation of the ``filepath`` compressed file fails. |
| 329 | NotImplementedError: When the ``filepath`` file extension is not one of [zip", "tar.gz", "tar"]. |
| 330 | |
| 331 | """ |
| 332 | if has_base: |
| 333 | # the extracted files will be in this folder |
| 334 | cache_dir = Path(output_dir, _basename(filepath).split(".")[0]) |
| 335 | else: |
| 336 | cache_dir = Path(output_dir) |
| 337 | if cache_dir.exists() and next(cache_dir.iterdir(), None) is not None: |
| 338 | logger.info(f"Non-empty folder exists in {cache_dir}, skipped extracting.") |
| 339 | return |
| 340 | filepath = Path(filepath) |
| 341 | if hash_val and not check_hash(filepath, hash_val, hash_type): |
| 342 | raise RuntimeError( |
| 343 | f"{hash_type} check of compressed file failed: " f"filepath={filepath}, expected {hash_type}={hash_val}." |
| 344 | ) |
| 345 | logger.info(f"Writing into directory: {output_dir}.") |
| 346 | _file_type = file_type.lower().strip() |
| 347 | if filepath.name.endswith("zip") or _file_type == "zip": |
| 348 | _extract_zip(filepath, output_dir) |
| 349 | return |
| 350 | if filepath.name.endswith("tar") or filepath.name.endswith("tar.gz") or "tar" in _file_type: |
| 351 | _extract_tar(filepath, output_dir) |
| 352 | return |
| 353 | raise NotImplementedError( |
| 354 | f'Unsupported file type, available options are: ["zip", "tar.gz", "tar"]. name={filepath} type={file_type}.' |
| 355 | ) |
| 356 | |
| 357 | |
| 358 | def get_filename_from_url(data_url: str) -> str: |
searching dependent graphs…