Unzip files from src into dst with a progress bar. ``path_filter`` is used to both edit and filter out the files from the zip. If an empty string is returned then that file is not extracted otherwise the string is used as the destination file path.
(
src,
dst,
path_filter=lambda x: "" if x.endswith("/") else x,
buffer_size=16 * 1024 * 1024,
quiet=False,
)
| 112 | |
| 113 | |
| 114 | def unzip( |
| 115 | src, |
| 116 | dst, |
| 117 | path_filter=lambda x: "" if x.endswith("/") else x, |
| 118 | buffer_size=16 * 1024 * 1024, |
| 119 | quiet=False, |
| 120 | ): |
| 121 | """Unzip files from src into dst with a progress bar. |
| 122 | |
| 123 | ``path_filter`` is used to both edit and filter out the files from the zip. |
| 124 | If an empty string is returned then that file is not extracted otherwise |
| 125 | the string is used as the destination file path. |
| 126 | """ |
| 127 | buff = bytearray(buffer_size) |
| 128 | dst = Path(dst) |
| 129 | progress = FileProgressBar(0, f"Extracting {src}", quiet) |
| 130 | with zipfile.ZipFile(src) as f_zip: |
| 131 | for file in f_zip.namelist(): |
| 132 | target = path_filter(file) |
| 133 | if not target: |
| 134 | continue |
| 135 | |
| 136 | target = dst / target |
| 137 | if not target.parent.exists(): |
| 138 | target.parent.mkdir(parents=True, exist_ok=True) |
| 139 | |
| 140 | with f_zip.open(file, "r") as f_in: |
| 141 | with open(target, "wb") as f_out: |
| 142 | while True: |
| 143 | n = f_in.readinto(buff) |
| 144 | if n == 0: |
| 145 | break |
| 146 | f_out.write(buff[:n]) |
| 147 | progress.add(n) |
| 148 | progress.finalize() |
no test coverage detected