Emulate unzip in Python

Author: sjpfenninger

tests/integration/Snakefile

@@ -16,13 +16,15 @@ rule download_netherlands_shapes:
 rule download_netherlands_protected_areas:
     message:
         "Download and unzip a dummy drop-in dataset for Netherlands protected areas (not based on WDPA)."
+    input:
+        script=workflow.source_path("../../workflow/scripts/unzip_like.py"),
     output:
         zipfile="results/module_area_potentials/resources/user/wdpa.gdb.zip",
         wdpa=directory("results/module_area_potentials/resources/user/wdpa.gdb"),
     shell:
         """
         curl -sSLo {output.zipfile} https://surfdrive.surf.nl/files/index.php/s/msuXQETNbCWawDh/download
-        unzip {output.zipfile} -d results/module_area_potentials/resources/user/
+        python {input.script} {output.zipfile} -t results/module_area_potentials/resources/user/
         """
 
 

workflow/envs/shell.yaml

@@ -3,4 +3,6 @@ channels:
   - conda-forge
   - nodefaults
 dependencies:
+  - python=3.13
+  - click=8.2.1
   - curl=8.9.1

workflow/rules/automatic.smk

@@ -46,7 +46,8 @@ rule unzip_globcover:
     params:
         target_file=internal["resources"]["automatic"]["globcover_landcover_tif"],
     input:
-        rules.download_globcover.output,
+        script=workflow.source_path("../scripts/unzip_like.py"),
+        zipfile=rules.download_globcover.output,
     output:
         "resources/automatic/global/globcover-landcover.tif",
     log:
@@ -56,7 +57,7 @@ rule unzip_globcover:
     shell:
         """
         temp_dir=$(mktemp -d)
-        unzip -j {input} {params.target_file} -d $temp_dir
+        python {input.script} {input.zipfile} -f {params.target_file} -t $temp_dir
         mv $temp_dir/{params.target_file} {output}
         rm -R $temp_dir
         """
@@ -81,15 +82,16 @@ rule unzip_ghsl:
     params:
         target_file=internal["resources"]["automatic"]["ghsl_tif"],
     input:
-        rules.download_ghsl.output,
+        script=workflow.source_path("../scripts/unzip_like.py"),
+        zipfile=rules.download_ghsl.output,
     output:
         "resources/automatic/global/ghsl_built_s.tif",
     conda:
         "../envs/shell.yaml"
     shell:
         """
         temp_dir=$(mktemp -d)
-        unzip -j {input} {params.target_file} -d $temp_dir
+        python {input.script} {input.zipfile} -f {params.target_file} -t $temp_dir
         mv $temp_dir/{params.target_file} {output}
         rm -R $temp_dir
         """

workflow/scripts/unzip_like.py

@@ -0,0 +1,38 @@
+import os
+import zipfile
+
+import click
+
+
+@click.command()
+@click.argument("zip_path", type=click.Path(exists=True))
+@click.option(
+    "--target",
+    "-t",
+    type=click.Path(),
+    default=".",
+    help="Target directory to extract to.",
+)
+@click.option("--file", "-f", help="Specific file inside the zip to extract.")
+def unzip(zip_path, target, file):
+    """Emulates the `unzip` command across platforms.
+
+    ZIP_PATH: Path to the .zip file
+    """
+    os.makedirs(target, exist_ok=True)
+
+    with zipfile.ZipFile(zip_path, "r") as zip_ref:
+        if file:
+            # Check if file exists in zip
+            if file not in zip_ref.namelist():
+                click.echo(f"Error: '{file}' not found in archive.")
+                return
+            zip_ref.extract(file, target)
+            click.echo(f"Extracted '{file}' to '{target}'.")
+        else:
+            zip_ref.extractall(target)
+            click.echo(f"Extracted all files to '{target}'.")
+
+
+if __name__ == "__main__":
+    unzip()