nut/nut/utils.py

104 lines
2.8 KiB
Python

import os
import zipfile
import json
import fnmatch
import yaml
def _should_ignore(rel_path: str, patterns: list) -> bool:
"""
Check if a file should be ignored based on patterns.
"""
rel_path = rel_path.replace("\\", "/")
for pattern in patterns:
pattern = pattern.strip()
if not pattern:
continue
if pattern.endswith("/"):
if rel_path.startswith(pattern):
return True
if fnmatch.fnmatch(rel_path, pattern):
return True
return False
def zip_dir(source_dir: str, output_zip_path: str, metadata: dict = None, ignore_patterns: list = None) -> None:
"""
Compress a directory into a zip file with ignore support.
:param source_dir: path to folder
:param output_zip_path: where to save zip
:param metadata: dict -> stored as _meta.json inside zip
:param ignore_patterns: list like gitignore ["*.log", "__pycache__/", "build/*"]
"""
ignore_patterns = ignore_patterns or []
with zipfile.ZipFile(output_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(source_dir):
rel_root = os.path.relpath(root, source_dir).replace("\\", "/")
# filter dirs in-place (important for performance)
dirs[:] = [
d for d in dirs
if not _should_ignore(
os.path.join(rel_root, d).replace("\\", "/"),
ignore_patterns
)
]
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, source_dir).replace("\\", "/")
if _should_ignore(rel_path, ignore_patterns):
continue
zipf.write(full_path, rel_path)
if metadata:
zipf.writestr("_meta.json", json.dumps(metadata, indent=2))
def unzip_to_dir(zip_path: str, output_dir: str) -> None:
"""
Extract zip file to a directory.
"""
with zipfile.ZipFile(zip_path, 'r') as zipf:
zipf.extractall(output_dir)
def unzip_in_memory(zip_path: str) -> dict:
"""
Read zip contents into memory only.
:return: dict {filename: bytes}, metadata separately if exists
"""
data = {}
metadata = None
with zipfile.ZipFile(zip_path, 'r') as zipf:
for name in zipf.namelist():
content = zipf.read(name)
if name == "_meta.json":
metadata = json.loads(content.decode())
else:
data[name] = content
return {
"files": data,
"metadata": metadata
}
def read_yaml(path: str) -> dict:
"""
:param path: path to yaml file
:return: parsed dict
"""
with open(path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
return data