recipe/pipeline/verify_download.py

#!/usr/bin/env python3
"""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)