ZIP Archive Manager (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: ZIP Archive Manager

Write a program that creates, extracts, lists, and modifies ZIP archives with support for compression levels, password protection, file filtering, and split archives.

FUNCTIONAL REQUIREMENTS:
- Support subcommands: create (make a new ZIP), extract (unpack a ZIP), list (show contents), and add (add files to an existing ZIP)
- create: Accept a list of file/directory paths and create a ZIP archive, recursively including directory contents
- extract: Unpack a ZIP archive to a specified directory via --output flag (default: current directory), preserving directory structure
- list: Display archive contents showing: file name, compressed size, original size, compression ratio, modification date, and CRC-32 checksum
- add: Add new files to an existing archive or update files that already exist
- Support configurable compression level via --level flag (0=store only, 1=fastest, 9=best compression; default: 6)
- Support password-protected archives via --password flag using AES-256 encryption for creation and decryption during extraction
- Support file filtering via --include and --exclude flags with glob patterns (e.g., --include "*.txt" --exclude "*.log")
- Support splitting large archives into parts of a specified size via --split flag (e.g., --split 100m for 100MB parts)
- Verify archive integrity via --verify flag: check CRC-32 checksums for all files without extracting
- Print operation summary to console: total files, total original size, total compressed size, overall compression ratio, and time taken
- Save operation log as JSON with --log flag
- If no arguments are given, create a sample directory structure with various file types and sizes, create a ZIP archive from it, list the contents, extract to a new location, and verify the extraction matches the originals by comparing checksums
- Handle errors: corrupted archives, wrong passwords, disk full during extraction, duplicate entries, and path traversal prevention

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# ZIP Archive Manager (Python)

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Install
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

No external dependencies are required.

## Commands
```bash
python src/main.py create out.zip input1 input2 --include "*.txt" --exclude "*.log"
python src/main.py list out.zip
python src/main.py extract out.zip --output extracted --verify
python src/main.py add out.zip newfile.txt newdir
```

If no command is given, sample data is generated and a full demo run is executed.

# No external dependencies required

#!/usr/bin/env python3
import argparse
import fnmatch
import hashlib
import json
import os
import shutil
import sys
import time
import zipfile
from pathlib import Path


def parse_size(spec: str | None) -> int | None:
    if not spec:
        return None
    spec = spec.strip().lower()
    mult = 1
    if spec.endswith("k"):
        mult, spec = 1024, spec[:-1]
    elif spec.endswith("m"):
        mult, spec = 1024 * 1024, spec[:-1]
    elif spec.endswith("g"):
        mult, spec = 1024 * 1024 * 1024, spec[:-1]
    return int(spec) * mult


def collect_files(path: Path) -> list[Path]:
    if path.is_file():
        return [path.resolve()]
    out: list[Path] = []
    for p in path.rglob("*"):
        if p.is_file():
            out.append(p.resolve())
    return out


def matches(rel: str, includes: list[str], excludes: list[str]) -> bool:
    inc_ok = not includes or any(fnmatch.fnmatch(rel, pat) for pat in includes)
    exc = any(fnmatch.fnmatch(rel, pat) for pat in excludes)
    return inc_ok and not exc


def sha_prefix(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()[:8]


def list_archive(archive: Path) -> dict:
    with zipfile.ZipFile(archive, "r") as zf:
        entries = []
        for info in zf.infolist():
            if info.is_dir():
                continue
            ratio = (1 - info.compress_size / info.file_size) * 100 if info.file_size else 0
            entries.append(
                {
                    "name": info.filename,
                    "compressedSize": info.compress_size,
                    "originalSize": info.file_size,
                    "compressionRatioPct": round(ratio, 2),
                    "modified": "%04d-%02d-%02dT%02d:%02d:%02d" % info.date_time,
                    "crc32": f"0x{info.CRC & 0xFFFFFFFF:08x}",
                }
            )
    return {"entries": entries, "summary": summarize(entries)}


def summarize(entries: list[dict]) -> dict:
    original = sum(e["originalSize"] for e in entries)
    compressed = sum(e["compressedSize"] for e in entries)
    ratio = (1 - compressed / original) * 100 if original else 0
    return {
        "totalFiles": len(entries),
        "totalOriginalSize": original,
        "totalCompressedSize": compressed,
        "compressionRatioPct": round(ratio, 2),
    }


def split_archive(archive: Path, chunk_size: int | None) -> list[str]:
    if not chunk_size:
        return []
    data = archive.read_bytes()
    parts: list[str] = []
    for idx, start in enumerate(range(0, len(data), chunk_size), start=1):
        part = archive.with_suffix(archive.suffix + f".part{idx:03d}")
        part.write_bytes(data[start : start + chunk_size])
        parts.append(str(part))
    return parts


def create_archive(args) -> dict:
    archive = Path(args.archive).resolve()
    if args.password:
        print("Warning: Python zipfile cannot write AES-encrypted ZIPs; archive will be unencrypted.", file=sys.stderr)

    start = time.perf_counter()
    compression = zipfile.ZIP_STORED if args.level == 0 else zipfile.ZIP_DEFLATED

    with zipfile.ZipFile(archive, "w", compression=compression, compresslevel=max(0, min(9, args.level))) as zf:
        for input_path in args.inputs:
            p = Path(input_path)
            if not p.exists():
                print(f"Warning: missing path {p}", file=sys.stderr)
                continue
            root_base = p.parent if p.is_file() else p
            for f in collect_files(p):
                rel = f.relative_to(root_base).as_posix()
                if not matches(rel, args.include, args.exclude):
                    continue
                zf.write(f, rel)

    listed = list_archive(archive)
    parts = split_archive(archive, parse_size(args.split))
    elapsed = (time.perf_counter() - start) * 1000
    return {
        "operation": "create",
        "archive": str(archive),
        "compressionLevel": args.level,
        "elapsedMs": round(elapsed, 2),
        "summary": listed["summary"],
        "entries": listed["entries"],
        "parts": parts,
    }


def safe_extract(zf: zipfile.ZipFile, info: zipfile.ZipInfo, output_dir: Path, password: bytes | None) -> None:
    target = (output_dir / info.filename).resolve()
    if not str(target).startswith(str(output_dir.resolve())):
        raise RuntimeError(f"Path traversal blocked: {info.filename}")
    target.parent.mkdir(parents=True, exist_ok=True)
    with zf.open(info, pwd=password) as src, target.open("wb") as dst:
        shutil.copyfileobj(src, dst)


def verify_extraction(archive: Path, output_dir: Path, password: bytes | None) -> dict:
    mismatches = []
    with zipfile.ZipFile(archive, "r") as zf:
        for info in zf.infolist():
            if info.is_dir():
                continue
            out_path = output_dir / info.filename
            if not out_path.exists():
                mismatches.append({"entry": info.filename, "reason": "missing file"})
                continue
            with zf.open(info, pwd=password) as src:
                in_hash = hashlib.sha256(src.read()).hexdigest()
            file_hash = hashlib.sha256(out_path.read_bytes()).hexdigest()
            if in_hash != file_hash:
                mismatches.append({"entry": info.filename, "reason": "checksum mismatch"})
    return {"ok": not mismatches, "mismatches": mismatches}


def extract_archive(args) -> dict:
    archive = Path(args.archive).resolve()
    output_dir = Path(args.output or ".").resolve()
    output_dir.mkdir(parents=True, exist_ok=True)
    password = args.password.encode() if args.password else None

    start = time.perf_counter()
    entries = []
    with zipfile.ZipFile(archive, "r") as zf:
        for info in zf.infolist():
            if info.is_dir():
                continue
            safe_extract(zf, info, output_dir, password)
            entries.append(
                {
                    "name": info.filename,
                    "compressedSize": info.compress_size,
                    "originalSize": info.file_size,
                    "crc32": f"0x{info.CRC & 0xFFFFFFFF:08x}",
                }
            )

    verify = verify_extraction(archive, output_dir, password) if args.verify else None
    elapsed = (time.perf_counter() - start) * 1000
    return {
        "operation": "extract",
        "archive": str(archive),
        "outputDirectory": str(output_dir),
        "elapsedMs": round(elapsed, 2),
        "summary": summarize(entries),
        "verify": verify,
        "entries": entries,
    }


def add_archive(args) -> dict:
    archive = Path(args.archive).resolve()
    if not archive.exists():
        raise RuntimeError(f"Archive not found: {archive}")

    temp = archive.with_suffix(".tmp.zip")
    with zipfile.ZipFile(archive, "r") as src, zipfile.ZipFile(
        temp, "w", compression=zipfile.ZIP_STORED if args.level == 0 else zipfile.ZIP_DEFLATED, compresslevel=max(0, min(9, args.level))
    ) as dst:
        # copy existing entries
        for info in src.infolist():
            if info.is_dir():
                continue
            dst.writestr(info.filename, src.read(info.filename))

        # add/update new files
        for input_path in args.inputs:
            p = Path(input_path)
            if not p.exists():
                print(f"Warning: missing path {p}", file=sys.stderr)
                continue
            root_base = p.parent if p.is_file() else p
            for f in collect_files(p):
                rel = f.relative_to(root_base).as_posix()
                if not matches(rel, args.include, args.exclude):
                    continue
                dst.writestr(rel, f.read_bytes())

    temp.replace(archive)
    listed = list_archive(archive)
    return {"operation": "add", "archive": str(archive), **listed}


def print_table(entries: list[dict]) -> None:
    headers = ["Name", "Compressed", "Original", "Ratio", "Modified", "CRC-32"]
    rows = [
        [
            e.get("name", ""),
            str(e.get("compressedSize", "")),
            str(e.get("originalSize", "")),
            f"{e.get('compressionRatioPct', '')}%",
            str(e.get("modified", "")),
            e.get("crc32", ""),
        ]
        for e in entries
    ]
    widths = [max(len(h), *(len(r[i]) for r in rows)) for i, h in enumerate(headers)]
    def fmt(row):
        return "  ".join(val.ljust(widths[i]) for i, val in enumerate(row))
    print(fmt(headers))
    print(fmt(["-" * w for w in widths]))
    for r in rows:
        print(fmt(r))


def make_sample() -> Path:
    root = Path("sample_zip_data").resolve()
    if root.exists():
        shutil.rmtree(root)
    (root / "docs").mkdir(parents=True)
    (root / "logs").mkdir(parents=True)
    (root / "docs" / "readme.txt").write_text("Sample README\n", encoding="utf-8")
    (root / "docs" / "data.csv").write_text("id,name\n1,Alice\n2,Bob\n", encoding="utf-8")
    (root / "logs" / "app.log").write_text("INFO boot\nERROR x\n", encoding="utf-8")
    (root / "blob.bin").write_bytes(os.urandom(4096))
    return root


def run_demo() -> None:
    sample_dir = make_sample()
    create_args = argparse.Namespace(archive="sample_archive.zip", inputs=[str(sample_dir)], level=6, password=None, include=[], exclude=[], split=None)
    create_result = create_archive(create_args)

    list_result = {"operation": "list", "archive": str(Path("sample_archive.zip").resolve()), **list_archive(Path("sample_archive.zip"))}

    extract_args = argparse.Namespace(archive="sample_archive.zip", output="sample_extracted", verify=True, password=None)
    extract_result = extract_archive(extract_args)

    print("Sample archive created and verified.")
    print_table(list_result["entries"])

    Path("zip_sample_report.json").write_text(
        json.dumps({"create": create_result, "list": list_result, "extract": extract_result}, indent=2),
        encoding="utf-8",
    )


def main() -> int:
    parser = argparse.ArgumentParser(description="ZIP archive manager")
    sub = parser.add_subparsers(dest="cmd")

    def add_common(p):
        p.add_argument("--level", type=int, default=6)
        p.add_argument("--password")
        p.add_argument("--include", action="append", default=[])
        p.add_argument("--exclude", action="append", default=[])
        p.add_argument("--split")
        p.add_argument("--verify", action="store_true")
        p.add_argument("--log")

    c = sub.add_parser("create")
    c.add_argument("archive")
    c.add_argument("inputs", nargs="+")
    add_common(c)

    e = sub.add_parser("extract")
    e.add_argument("archive")
    e.add_argument("--output")
    add_common(e)

    l = sub.add_parser("list")
    l.add_argument("archive")
    add_common(l)

    a = sub.add_parser("add")
    a.add_argument("archive")
    a.add_argument("inputs", nargs="+")
    add_common(a)

    args = parser.parse_args()
    if not args.cmd:
        run_demo()
        return 0

    try:
        start = time.perf_counter()
        if args.cmd == "create":
            result = create_archive(args)
        elif args.cmd == "extract":
            result = extract_archive(args)
        elif args.cmd == "list":
            result = {"operation": "list", "archive": str(Path(args.archive).resolve()), **list_archive(Path(args.archive))}
        elif args.cmd == "add":
            result = add_archive(args)
        else:
            raise RuntimeError(f"Unknown command: {args.cmd}")

        if "entries" in result:
            print_table(result["entries"])
        if "summary" in result:
            s = result["summary"]
            print(
                f"\nSummary: files={s['totalFiles']}, original={s['totalOriginalSize']}, "
                f"compressed={s['totalCompressedSize']}, ratio={s['compressionRatioPct']}%"
            )
        print(f"Time: {(time.perf_counter() - start) * 1000:.2f} ms")

        log_path = args.log or (None if args.cmd == "list" else "zip_operation_log.json")
        if log_path:
            Path(log_path).write_text(json.dumps(result, indent=2), encoding="utf-8")
        return 0
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())