Functions

def compress_7z(
    path_in: Path,
    path_out: Optional[Path] = None,
    word_size: WordSize = WordSize.WS16,
    dict_size: DictionarySize = DictionarySize.DS1536
) -> Path:
    """Compress a target file and save it as a 7z archive to the output directory.

    Args:
        path_in: File to compress.
        path_out: Path to the archive to be saved. Use 'compressed' subdirectory if not provided.
        word_size: Word size value to use for the compression.
        dict_size: Dictionary size value to use for the compression.

    Returns:
        Path to the resulting 7z archive.
    """
    # Define the output file path
    path_out = path_out or Path(path_in.parent, '.compressed', path_in.name)
    path_out = path_out.with_suffix('.7z')
    null_device = open(os.devnull, 'w')

    # Compress the file
    with suppress(Exception):
        subprocess.run([
            "7z", "a", "-t7z", "-m0=LZMA", "-mx=9",
            f"-md={dict_size}M",
            f"-mfb={word_size}",
            str(path_out),
            str(path_in)
        ], stdout=null_device, stderr=null_device)
    return path_out

def compress_7z_all(
    path_in: Path,
    path_out: Path = None,
    word_size: WordSize = WordSize.WS16,
    dict_size: DictionarySize = DictionarySize.DS1536
) -> None:
    """Compress every file inside `path_in` directory as 7z archives, then output
    those archives in the `path_out`.

    Args:
        path_in: Directory containing files to compress.
        path_out: Directory to place the archives. Use a subdirectory 'compressed' if not provided.
        word_size: Word size value to use for the compression.
        dict_size: Dictionary size value to use for the compression.
    """
    # Use "compressed" subdirectory if not provided, ensure output directory exists
    path_out = path_out or Path(path_in, '.compressed')
    path_out.mkdir(mode=777, parents=True, exist_ok=True)

    # Get a list of all .psd files in the directory
    files = [
        Path(path_in, n) for n in os.listdir(path_in)
        if Path(path_in, n).is_file()]

    # Compress each file
    with tqdm(total=len(files), desc="Compressing files", unit="file") as pbar:
        for f in files:
            p = (path_out / f.name).with_suffix('.7z')
            pbar.set_description(f.name)
            compress_7z(
                path_in=f,
                path_out=p,
                word_size=word_size,
                dict_size=dict_size)
            pbar.update()

def unpack_zip(path: Path) -> None:
    """Unpack target 'zip' archive.

    Args:
        path: Path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    with zipfile.ZipFile(path, 'r') as zip_ref:
        zip_ref.extractall(path=path.parent)

def unpack_gz(path: Path) -> None:
    """Unpack target 'gz' archive.

    Args:
        path: Path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    output = path.parent / path.name.replace('.gz', '')
    with gzip.open(path, 'rb') as arch:
        with open(output, 'wb') as f:
            shutil.copyfileobj(arch, f)

def unpack_xz(path: Path) -> None:
    """Unpack target 'xz' archive.

    Args:
        path: Path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    output = path.parent / path.name.replace('.xz', '')
    with lzma.open(path, mode='r') as arch:
        with open(output, 'wb') as f:
            shutil.copyfileobj(arch, f)

def unpack_7z(path: Path) -> None:
    """Unpack target '7z' archive.

    Args:
        path: path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    with py7zr.SevenZipFile(path, 'r') as arch:
        arch.extractall(path=path.parent)

def unpack_tar_gz(path: Path) -> None:
    """Unpack target `tar.gz` archive.

    Args:
        path: Path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    with tarfile.open(path, 'r:gz') as tar:
        tar.extractall(path=path.parent)

def unpack_tar_xz(path: Path) -> None:
    """Unpack target `tar.xz` archive.

    Args:
        path: Path to the archive.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    if not path.is_file():
        raise FileNotFoundError(f'Archive not found: {str(path)}')
    with tarfile.open(path, 'r:xz') as tar:
        tar.extractall(path=path.parent)

def unpack_archive(path: Path, remove: bool = True) -> None:
    """Unpack an archive using the correct methodology based on its extension.

    Args:
        path: Path to the archive.
        remove: Whether to remove the archive after unpacking.

    Raises:
        FileNotFoundError: If archive couldn't be located.
    """
    action_map: dict[str, Callable] = {
        ArchiveExt.Zip: unpack_zip,
        ArchiveExt.GZip: unpack_gz,
        ArchiveExt.XZip: unpack_xz,
        ArchiveExt.TarGZip: unpack_tar_gz,
        ArchiveExt.TarXZip: unpack_tar_xz,
        ArchiveExt.SevenZip: unpack_7z,
        ArchiveExt.TarSevenZip: unpack_7z
    }
    if path.suffix not in action_map:
        return
    action = action_map[path.suffix]
    with ARCHIVE_LOCK:
        _ = action(path)
    if remove:
        os.remove(path)
    gc.collect()