begin develop configloader

zip_sequences.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import subprocess
 import os
 import shutil
 import sys
@@ -30,6 +31,52 @@ SEQUENCE_EXTENSIONS = {
     ".exr",
 }
 STATE_SUFFIX = ".meta.json"
+CONFIG_PATH = Path(__file__).resolve().with_name("config.json")
+DEFAULT_CONFIG = {
+    "zipper": True,
+    "compression": 9,
+}
+
+
+def load_config() -> dict:
+    try:
+        text = CONFIG_PATH.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return DEFAULT_CONFIG.copy()
+    except OSError:
+        return DEFAULT_CONFIG.copy()
+
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError:
+        return DEFAULT_CONFIG.copy()
+
+    if not isinstance(data, dict):
+        return DEFAULT_CONFIG.copy()
+
+    merged = DEFAULT_CONFIG.copy()
+    merged.update(data)
+    return merged
+
+
+CONFIG = load_config()
+USE_7Z = bool(CONFIG.get("zipper", True))
+COMPRESSION_LEVEL = CONFIG.get("compression", 9)
+if isinstance(COMPRESSION_LEVEL, str):
+    try:
+        COMPRESSION_LEVEL = int(COMPRESSION_LEVEL)
+    except ValueError:
+        COMPRESSION_LEVEL = 9
+if not isinstance(COMPRESSION_LEVEL, int):
+    COMPRESSION_LEVEL = 9
+COMPRESSION_LEVEL = max(0, min(9, COMPRESSION_LEVEL))
+
+SEVEN_Z_EXE: str | None = None
+if USE_7Z:
+    SEVEN_Z_EXE = shutil.which("7z") or shutil.which("7za")
+    if SEVEN_Z_EXE is None:
+        print("[zip] Requested 7z compression but no 7z executable was found; falling back to zipfile.", file=sys.stderr)
+        USE_7Z = False
 
 
 def parse_args() -> argparse.Namespace:
@@ -138,24 +185,55 @@ def state_path_for(zip_path: Path) -> Path:
 
 
 def zip_sequence(seq_dir: Path, zip_path: Path) -> None:
-    from zipfile import ZIP_STORED, ZipFile
+    if USE_7Z and SEVEN_Z_EXE:
+        zip_path.parent.mkdir(parents=True, exist_ok=True)
+        cmd = [
+            SEVEN_Z_EXE,
+            "a",
+            "-y",
+            f"-mx={COMPRESSION_LEVEL}",
+            "-tzip",
+            str(zip_path),
+            ".\\*",
+        ]
+        subprocess.run(cmd, cwd=seq_dir, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        return
+
+    from zipfile import ZIP_DEFLATED, ZIP_STORED, ZipFile
 
     zip_path.parent.mkdir(parents=True, exist_ok=True)
-    with ZipFile(zip_path, "w", compression=ZIP_STORED) as archive:
+    if COMPRESSION_LEVEL <= 0:
+        compression = ZIP_STORED
+        zip_kwargs = {}
+    else:
+        compression = ZIP_DEFLATED
+        zip_kwargs = {"compresslevel": COMPRESSION_LEVEL}
+
+    with ZipFile(zip_path, "w", compression=compression, **zip_kwargs) as archive:
         for file_path in iter_sequence_files(seq_dir):
             archive.write(file_path, arcname=file_path.relative_to(seq_dir).as_posix())
 
 
 def expand_sequence(zip_path: Path, seq_state: dict) -> None:
-    from zipfile import ZipFile
-
     target_dir = sequence_dir_for(zip_path)
     if target_dir.exists():
         shutil.rmtree(target_dir)
     target_dir.mkdir(parents=True, exist_ok=True)
 
-    with ZipFile(zip_path, "r") as archive:
-        archive.extractall(target_dir)
+    if USE_7Z and SEVEN_Z_EXE:
+        cmd = [
+            SEVEN_Z_EXE,
+            "x",
+            "-y",
+            str(zip_path),
+            f"-o{target_dir}",
+        ]
+        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    else:
+        from zipfile import ZipFile
+
+        with ZipFile(zip_path, "r") as archive:
+            archive.extractall(target_dir)
 
     for entry in seq_state.get("files", []):
         file_path = target_dir / entry["path"]