import sys from bot import aria2, LOGGER, DOWNLOAD_DIR import shutil import os import pathlib import magic import tarfile from .exceptions import NotSupportedExtractionArchive def clean_download(path: str): if os.path.exists(path): LOGGER.info(f"Cleaning download: {path}") shutil.rmtree(path) def start_cleanup(): try: shutil.rmtree(DOWNLOAD_DIR) except FileNotFoundError: pass def clean_all(): aria2.remove_all(True) shutil.rmtree(DOWNLOAD_DIR) def exit_clean_up(signal, frame): try: LOGGER.info("Please wait, while we clean up the downloads and stop running downloads") clean_all() sys.exit(0) except KeyboardInterrupt: LOGGER.warning("Force Exiting before the cleanup finishes!") sys.exit(1) def get_path_size(path): if os.path.isfile(path): return os.path.getsize(path) total_size = 0 for root, dirs, files in os.walk(path): for f in files: abs_path = os.path.join(root, f) total_size += os.path.getsize(abs_path) return total_size def tar(org_path): tar_path = org_path + ".tar" path = pathlib.PurePath(org_path) LOGGER.info(f'Tar: orig_path: {org_path}, tar_path: {tar_path}') tar = tarfile.open(tar_path, "w") tar.add(org_path, arcname=path.name) tar.close() return tar_path def get_base_name(orig_path: str): if orig_path.endswith(".tar.bz2"): return orig_path.replace(".tar.bz2", "") elif orig_path.endswith(".tar.gz"): return orig_path.replace(".tar.gz", "") elif orig_path.endswith(".bz2"): return orig_path.replace(".bz2", "") elif orig_path.endswith(".gz"): return orig_path.replace(".gz", "") elif orig_path.endswith(".tar"): return orig_path.replace(".tar", "") elif orig_path.endswith(".tbz2"): return orig_path.replace("tbz2", "") elif orig_path.endswith(".tgz"): return orig_path.replace(".tgz", "") elif orig_path.endswith(".zip"): return orig_path.replace(".zip", "") elif orig_path.endswith(".7z"): return orig_path.replace(".7z", "") elif orig_path.endswith(".Z"): return orig_path.replace(".Z", "") elif orig_path.endswith(".rar"): return orig_path.replace(".rar", "") else: raise NotSupportedExtractionArchive('File format not supported for extraction') def get_mime_type(file_path): mime = magic.Magic(mime=True) mime_type = mime.from_file(file_path) mime_type = mime_type if mime_type else "text/plain" return mime_type