45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
|
import importlib
|
||
|
|
import shutil
|
||
|
|
|
||
|
|
|
||
|
|
def compress(filename: str):
|
||
|
|
supp_compr_algo = {"zip": "zipfile", "gzip": "gzip", "bzip2": "bz2", "lzma": "lzma"}
|
||
|
|
extensions = {"zip": ".zip", "gzip": ".gz", "bzip2": ".bz2", "lzma": ".xz"}
|
||
|
|
assert set(supp_compr_algo.keys()) == set(extensions.keys())
|
||
|
|
|
||
|
|
compression = "bzip2"
|
||
|
|
|
||
|
|
if compression not in supp_compr_algo:
|
||
|
|
raise ValueError(
|
||
|
|
f"Unknown compression algorithm '{compression}'; must be one of {list(supp_compr_algo.keys())}")
|
||
|
|
|
||
|
|
try_compression(
|
||
|
|
file=filename,
|
||
|
|
name=compression,
|
||
|
|
module_name=supp_compr_algo[compression],
|
||
|
|
extension=extensions[compression],
|
||
|
|
function=zip_compression if compression == "zip" else compression_open_context_manager
|
||
|
|
)
|
||
|
|
print(f"Successfully compressed '{filename}' to '{filename + extensions[compression]}' "
|
||
|
|
f"using {compression} as compression algorithm")
|
||
|
|
|
||
|
|
|
||
|
|
def try_compression(file: str, name: str, module_name: str, extension: str, function: callable):
|
||
|
|
try:
|
||
|
|
compression_module = importlib.import_module(module_name)
|
||
|
|
|
||
|
|
function(file, compression_module, extension)
|
||
|
|
except ImportError as ex:
|
||
|
|
raise ImportError(f"compression='{name}' failed: required module could not be loaded ({ex})")
|
||
|
|
|
||
|
|
|
||
|
|
def compression_open_context_manager(file: str, module, extension: str):
|
||
|
|
with open(file, "rb") as f_in:
|
||
|
|
with module.open(file + extension, "wb") as f_out:
|
||
|
|
shutil.copyfileobj(f_in, f_out)
|
||
|
|
|
||
|
|
|
||
|
|
def zip_compression(file: str, module, extension: str):
|
||
|
|
with module.ZipFile(file + extension, "w", compression=module.ZIP_DEFLATED) as z:
|
||
|
|
z.write(file)
|