| """Build the deterministic SpreadsheetBench runtime workbook archive.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from pathlib import Path |
| import subprocess |
| import tarfile |
| import tempfile |
|
|
|
|
| ARCHIVE_NAME = "spreadsheetbench-runtime-v0.1.tar.zst" |
|
|
|
|
| def referenced_paths(package_root: Path) -> tuple[str, ...]: |
| task_root = package_root / "src/spreadsheetbench_env/assets/tasks" |
| paths: list[str] = [] |
| for filename in ("dev.json", "test.json"): |
| payload = json.loads((task_root / filename).read_text(encoding="utf-8")) |
| for task in payload["tasks"]: |
| for case in task["cases"]: |
| paths.extend((str(case["input"]), str(case["answer"]))) |
| if len(paths) != len(set(paths)): |
| raise ValueError("canonical SpreadsheetBench cases contain duplicate runtime paths") |
| return tuple(sorted(paths)) |
|
|
|
|
| def _normalized_tar(package_root: Path, source_root: Path, destination: Path) -> None: |
| with tarfile.open(destination, "w", format=tarfile.PAX_FORMAT) as archive: |
| for relative in referenced_paths(package_root): |
| source = (source_root / relative).resolve() |
| if not source.is_relative_to(source_root.resolve()) or not source.is_file(): |
| raise FileNotFoundError(f"missing canonical runtime workbook: {relative}") |
| info = archive.gettarinfo(str(source), arcname=relative) |
| info.uid = 0 |
| info.gid = 0 |
| info.uname = "root" |
| info.gname = "root" |
| info.mtime = 0 |
| info.mode = 0o600 |
| with source.open("rb") as handle: |
| archive.addfile(info, handle) |
|
|
|
|
| def build_archive(repository: Path) -> Path: |
| package_root = repository / "environments/spreadsheetbench" |
| source_root = repository / "data/spreadsheetbench" |
| output_root = package_root / "runtime-data" |
| output_root.mkdir(parents=True, exist_ok=True) |
| archive_path = output_root / ARCHIVE_NAME |
| with tempfile.TemporaryDirectory(prefix="spreadsheetbench-runtime-") as temp: |
| tar_path = Path(temp) / "runtime.tar" |
| compressed_path = Path(temp) / ARCHIVE_NAME |
| _normalized_tar(package_root, source_root, tar_path) |
| subprocess.run( |
| [ |
| "zstd", |
| "--quiet", |
| "--threads=0", |
| "-19", |
| "--no-check", |
| "-o", |
| str(compressed_path), |
| str(tar_path), |
| ], |
| check=True, |
| ) |
| compressed_path.replace(archive_path) |
| digest = hashlib.sha256(archive_path.read_bytes()).hexdigest() |
| archive_path.with_suffix(archive_path.suffix + ".sha256").write_text( |
| f"{digest} {archive_path.name}\n", |
| encoding="utf-8", |
| ) |
| return archive_path |
|
|
|
|
| def main() -> None: |
| repository = Path(__file__).resolve().parents[3] |
| archive = build_archive(repository) |
| print(archive) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
