| |
| """Hydrate and verify Xet-backed assets in the single Nanobrush Git checkout.""" |
| import argparse |
| import hashlib |
| from pathlib import Path |
| import subprocess |
| import sys |
|
|
| import storage_policy as policy |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def run(*args, capture=False): |
| return subprocess.run(args, cwd=ROOT, check=True, stdout=subprocess.PIPE if capture else None).stdout |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("command", choices=("check", "fetch")) |
| args = parser.parse_args() |
| policy.check_setup() |
| policy.check_index() |
| if args.command == "fetch": |
| run("git", "lfs", "pull", "hf") |
| |
| count = 0 |
| paths = policy.git("ls-files", "-z").split(b"\0") |
| for raw in paths: |
| if not raw: |
| continue |
| path = raw.decode("utf-8") |
| if not policy.is_asset(path): |
| continue |
| pointer = policy.git("show", ":" + path) |
| policy.check_asset(path, pointer) |
| if not pointer: |
| expected_hash, size = hashlib.sha256(b"").hexdigest(), 0 |
| else: |
| lines = pointer.decode().splitlines() |
| expected_hash, size = lines[1].split(":", 1)[1], int(lines[2].split()[1]) |
| payload = ROOT / path |
| if payload.is_symlink() or not payload.is_file() or payload.stat().st_size != size: |
| raise ValueError(f"Missing/unhydrated asset: {path}; run assets-fetch") |
| digest = hashlib.sha256() |
| with payload.open("rb") as stream: |
| for chunk in iter(lambda: stream.read(1024 * 1024), b""): |
| digest.update(chunk) |
| if digest.hexdigest() != expected_hash: |
| raise ValueError(f"Modified/unhydrated asset: {path}; preserve edits before fetching") |
| count += 1 |
| print(f"Verified {count} hydrated Xet/LFS assets against staged pointer identities.") |
|
|
|
|
| if __name__ == "__main__": |
| try: |
| main() |
| except (ValueError, OSError, subprocess.CalledProcessError) as error: |
| print(f"Assets: {error}", file=sys.stderr) |
| sys.exit(1) |
|
|
