reconfig zipseq
This commit is contained in:
@@ -11,10 +11,11 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
from typing import Iterator, Sequence
|
||||
@@ -67,7 +68,6 @@ if isinstance(zipper_val, bool):
|
||||
ZIPPER_TYPE = "7z" if zipper_val else "zip"
|
||||
else:
|
||||
ZIPPER_TYPE = str(zipper_val).lower()
|
||||
USE_7Z = ZIPPER_TYPE == "7z"
|
||||
COMPRESSION_LEVEL = CONFIG.get("compression", 9)
|
||||
if isinstance(COMPRESSION_LEVEL, str):
|
||||
try:
|
||||
@@ -79,11 +79,10 @@ if not isinstance(COMPRESSION_LEVEL, int):
|
||||
COMPRESSION_LEVEL = max(0, min(9, COMPRESSION_LEVEL))
|
||||
|
||||
SEVEN_Z_EXE: str | None = None
|
||||
if USE_7Z:
|
||||
if ZIPPER_TYPE == "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
|
||||
print("[zip] Warning: 7z compression requested but no 7z executable was found in PATH.", file=sys.stderr)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
@@ -192,20 +191,47 @@ def state_path_for(zip_path: Path) -> Path:
|
||||
|
||||
|
||||
def zip_sequence(seq_dir: Path, zip_path: Path) -> None:
|
||||
if USE_7Z and SEVEN_Z_EXE:
|
||||
if ZIPPER_TYPE == "7z":
|
||||
if SEVEN_Z_EXE is None:
|
||||
raise RuntimeError("7z compression requested but 7z executable not found in PATH")
|
||||
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)
|
||||
# Build list of files to archive with relative paths
|
||||
file_list = []
|
||||
for file_path in iter_sequence_files(seq_dir):
|
||||
rel_path = file_path.relative_to(seq_dir).as_posix()
|
||||
file_list.append(rel_path)
|
||||
|
||||
# Use a temporary list file to avoid Windows command line length limits
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".lst", delete=False, encoding="utf-8") as list_file:
|
||||
list_file_path = Path(list_file.name)
|
||||
for rel_path in file_list:
|
||||
list_file.write(rel_path + "\n")
|
||||
|
||||
try:
|
||||
cmd = [
|
||||
SEVEN_Z_EXE,
|
||||
"a",
|
||||
"-y",
|
||||
f"-mx={COMPRESSION_LEVEL}",
|
||||
"-tzip",
|
||||
str(zip_path),
|
||||
f"@{list_file_path}",
|
||||
]
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
cwd=seq_dir,
|
||||
check=False,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"7z compression failed: {result.stderr}")
|
||||
finally:
|
||||
list_file_path.unlink(missing_ok=True)
|
||||
return
|
||||
|
||||
# Use zipfile (ZIPPER_TYPE == "zip" or fallback)
|
||||
from zipfile import ZIP_DEFLATED, ZIP_STORED, ZipFile
|
||||
|
||||
zip_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
@@ -227,7 +253,9 @@ def expand_sequence(zip_path: Path, seq_state: dict) -> None:
|
||||
shutil.rmtree(target_dir)
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if USE_7Z and SEVEN_Z_EXE:
|
||||
if ZIPPER_TYPE == "7z":
|
||||
if SEVEN_Z_EXE is None:
|
||||
raise RuntimeError("7z extraction requested but 7z executable not found in PATH")
|
||||
cmd = [
|
||||
SEVEN_Z_EXE,
|
||||
"x",
|
||||
|
||||
Reference in New Issue
Block a user