| """BLAKE3 helpers for content-addressed provenance. |
| |
| Files: streamed, returns a hex digest. |
| Directories: hash sorted (relpath, file-hash) pairs to make the dir hash |
| deterministic regardless of filesystem walk order or mtime. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
|
|
| from blake3 import blake3 |
|
|
| _CHUNK = 1 << 20 |
|
|
|
|
| def blake3_bytes(data: bytes) -> str: |
| """Return BLAKE3 hex digest of an in-memory byte string. |
| |
| Used to hash artifacts that never touch disk in a stable form — notably the |
| frozen AutotunePlan JSON — without round-tripping through a temp file. |
| """ |
| return blake3(data).hexdigest() |
|
|
|
|
| def blake3_file(path: Path) -> str: |
| """Return BLAKE3 hex digest of a single file.""" |
| h = blake3() |
| with path.open("rb") as f: |
| while True: |
| chunk = f.read(_CHUNK) |
| if not chunk: |
| break |
| h.update(chunk) |
| return h.hexdigest() |
|
|
|
|
| def blake3_dir(root: Path) -> str: |
| """Return BLAKE3 hex digest of a directory (sorted-relpath + file-hash composition).""" |
| if not root.is_dir(): |
| msg = f"not a directory: {root}" |
| raise NotADirectoryError(msg) |
|
|
| entries: list[tuple[str, str]] = [] |
| for p in sorted(root.rglob("*")): |
| if p.is_file(): |
| rel = p.relative_to(root).as_posix() |
| entries.append((rel, blake3_file(p))) |
|
|
| h = blake3() |
| for rel, file_hash in entries: |
| h.update(rel.encode("utf-8")) |
| h.update(b"\x00") |
| h.update(file_hash.encode("ascii")) |
| h.update(b"\n") |
| return h.hexdigest() |
|
|
