scripts/xet_experiment2.py

# Round 2: HEAD-check round-1 results, then large-file (LFS/xet path) test.
# Linear code only.
import os, time, urllib.request

tok = open("/content/qwenjev_work_v3/.hf_token").read().strip()
from huggingface_hub import HfApi
api = HfApi(token=tok)

# ---- step 1: verify round-1 files -------------------------------------
for p in ["t4_assets_v3/_xet_test/default.safetensors",
          "t4_assets_v3/_xet_test/noxet.safetensors"]:
    url = "https://huggingface.co/tchbcb/qwenjev/resolve/main/" + p
    try:
        r = urllib.request.urlopen(urllib.request.Request(url, method="HEAD"), timeout=30)
        print("R1", p.split("/")[-1], "OK len=" + str(r.headers.get("Content-Length")), flush=True)
    except Exception as e:
        print("R1", p.split("/")[-1], "FAIL " + str(e)[:40], flush=True)

# ---- step 2: 30MB file => LFS/xet path --------------------------------
big = os.urandom(30 * 1024 * 1024)
open("/tmp/big.safetensors", "wb").write(big)
print("big file:", len(big), "bytes", flush=True)

pA = "t4_assets_v3/_xet_test/big_default.safetensors"
t0 = time.time()
try:
    api.upload_file(path_or_fileobj="/tmp/big.safetensors", path_in_repo=pA,
                    repo_id="tchbcb/qwenjev", repo_type="model",
                    commit_message="xet test C big default")
    print("C upload ok in", round(time.time() - t0, 1), "s", flush=True)
except Exception as e:
    print("C upload error:", str(e)[:150], flush=True)
time.sleep(2)
urlA = "https://huggingface.co/tchbcb/qwenjev/resolve/main/" + pA
try:
    r = urllib.request.urlopen(urllib.request.Request(urlA, method="HEAD"), timeout=30)
    print("C big-default HEAD: OK len=" + str(r.headers.get("Content-Length")), flush=True)
except Exception as e:
    print("C big-default HEAD: FAIL " + str(e)[:60], flush=True)

os.environ["HF_HUB_DISABLE_XET"] = "1"
pB = "t4_assets_v3/_xet_test/big_noxet.safetensors"
t0 = time.time()
try:
    api.upload_file(path_or_fileobj="/tmp/big.safetensors", path_in_repo=pB,
                    repo_id="tchbcb/qwenjev", repo_type="model",
                    commit_message="xet test D big no-xet")
    print("D upload ok in", round(time.time() - t0, 1), "s", flush=True)
except Exception as e:
    print("D upload error:", str(e)[:150], flush=True)
time.sleep(2)
urlB = "https://huggingface.co/tchbcb/qwenjev/resolve/main/" + pB
try:
    r = urllib.request.urlopen(urllib.request.Request(urlB, method="HEAD"), timeout=30)
    print("D big-noxet HEAD: OK len=" + str(r.headers.get("Content-Length")), flush=True)
except Exception as e:
    print("D big-noxet HEAD: FAIL " + str(e)[:60], flush=True)

print("EXPERIMENT2_DONE", flush=True)