Changeset View
Changeset View
Standalone View
Standalone View
swh/core/tarball.py
Show First 20 Lines • Show All 81 Lines • ▼ Show 20 Lines | try: | ||||
jar.extractall(path=extract_dir) | jar.extractall(path=extract_dir) | ||||
return extract_dir | return extract_dir | ||||
except Exception as e: | except Exception as e: | ||||
raise shutil.ReadError( | raise shutil.ReadError( | ||||
f"Unable to uncompress {jarpath} to {extract_dir}. Reason: {e}" | f"Unable to uncompress {jarpath} to {extract_dir}. Reason: {e}" | ||||
) | ) | ||||
def _unpack_zst(zstpath: str, extract_dir: str) -> str: | |||||
"""Unpack zst files unsupported by the standard python library. Example | |||||
include tar.zst | |||||
This expects the `extract_dir` to exist. | |||||
Raises: | |||||
shutil.ReadError in case of issue uncompressing the archive (zstpath | |||||
""" | |||||
try: | |||||
run( | |||||
["tar", "--force-local", "-I 'zstd'", "-xf", zstpath, "-C", extract_dir], | |||||
check=True, | |||||
) | |||||
return extract_dir | |||||
except Exception as e: | |||||
raise shutil.ReadError( | |||||
f"Unable to uncompress {zstpath} to {extract_dir}. Reason: {e}" | |||||
) | |||||
def register_new_archive_formats(): | def register_new_archive_formats(): | ||||
"""Register new archive formats to uncompress""" | """Register new archive formats to uncompress""" | ||||
registered_formats = [f[0] for f in shutil.get_unpack_formats()] | registered_formats = [f[0] for f in shutil.get_unpack_formats()] | ||||
for name, extensions, function in ADDITIONAL_ARCHIVE_FORMATS: | for name, extensions, function in ADDITIONAL_ARCHIVE_FORMATS: | ||||
if name in registered_formats: | if name in registered_formats: | ||||
continue | continue | ||||
shutil.register_unpack_format(name, extensions, function) | shutil.register_unpack_format(name, extensions, function) | ||||
_mime_to_archive_format = { | _mime_to_archive_format = { | ||||
"application/x-compress": "tar.Z|x", | "application/x-compress": "tar.Z|x", | ||||
"application/x-tar": "tar", | "application/x-tar": "tar", | ||||
"application/x-bzip2": "bztar", | "application/x-bzip2": "bztar", | ||||
"application/gzip": "gztar", | "application/gzip": "gztar", | ||||
"application/x-lzip": "tar.lz", | "application/x-lzip": "tar.lz", | ||||
"application/zip": "zip", | "application/zip": "zip", | ||||
"application/java-archive": "jar", | "application/java-archive": "jar", | ||||
"application/zstd": "tar.zst", | |||||
"application/x-zstd": "tar.zst", | |||||
} | } | ||||
def uncompress(tarpath: str, dest: str): | def uncompress(tarpath: str, dest: str): | ||||
"""Uncompress tarpath to dest folder if tarball is supported. | """Uncompress tarpath to dest folder if tarball is supported. | ||||
Note that this fixes permissions after successfully | Note that this fixes permissions after successfully | ||||
uncompressing the archive. | uncompressing the archive. | ||||
▲ Show 20 Lines • Show All 101 Lines • ▼ Show 20 Lines | |||||
ADDITIONAL_ARCHIVE_FORMATS = [ | ADDITIONAL_ARCHIVE_FORMATS = [ | ||||
# name, extensions, function | # name, extensions, function | ||||
("tar.Z|x", [".tar.Z", ".tar.x"], _unpack_tar), | ("tar.Z|x", [".tar.Z", ".tar.x"], _unpack_tar), | ||||
("jar", [".jar"], _unpack_jar), | ("jar", [".jar"], _unpack_jar), | ||||
("tbz2", [".tbz", "tbz2"], _unpack_tar), | ("tbz2", [".tbz", "tbz2"], _unpack_tar), | ||||
# FIXME: make this optional depending on the runtime lzip package install | # FIXME: make this optional depending on the runtime lzip package install | ||||
("tar.lz", [".tar.lz"], _unpack_tar), | ("tar.lz", [".tar.lz"], _unpack_tar), | ||||
("crate", [".crate"], _unpack_tar), | ("crate", [".crate"], _unpack_tar), | ||||
("tar.zst", [".tar.zst", ".tar.zstd"], _unpack_zst), | |||||
] | ] | ||||
register_new_archive_formats() | register_new_archive_formats() |