| |
| """Byte + sha256 gate for a pinned HF snapshot. Compares against the tree API's lfs.oid |
| (the real content sha256) -- never the plain etag, which is a Xet/CAS id. |
| usage: verify_download.py <repo> <revision> <local_dir>""" |
| import hashlib, json, os, sys, urllib.request |
| from concurrent.futures import ThreadPoolExecutor |
| repo, rev, d = sys.argv[1:4] |
| tree = json.load(urllib.request.urlopen( |
| f"https://huggingface.co/api/models/{repo}/tree/{rev}?recursive=true", timeout=60)) |
| files = [t for t in tree if t.get("type") == "file"] |
| def sha(p): |
| h = hashlib.sha256() |
| with open(p, "rb") as f: |
| for b in iter(lambda: f.read(64 << 20), b""): h.update(b) |
| return h.hexdigest() |
| def one(t): |
| p = os.path.join(d, t["path"]) |
| if not os.path.exists(p): return (t["path"], "MISSING") |
| if os.path.getsize(p) != t["size"]: return (t["path"], f"SIZE {os.path.getsize(p)} != {t['size']}") |
| lfs = t.get("lfs") |
| if lfs: |
| got = sha(p) |
| return (t["path"], "ok-sha" if got == lfs["oid"] else f"SHA {got[:12]} != {lfs['oid'][:12]}") |
| return (t["path"], "ok-size") |
| with ThreadPoolExecutor(6) as ex: res = list(ex.map(one, files)) |
| bad = [r for r in res if not r[1].startswith("ok")] |
| print(f"files={len(res)} sha-verified={sum(r[1]=='ok-sha' for r in res)} size-only={sum(r[1]=='ok-size' for r in res)} bad={len(bad)}") |
| for r in bad: print(" BAD", r) |
| print("RESULT:", "PASS" if not bad else "FAIL") |
| sys.exit(0 if not bad else 1) |
|
|
