53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
|
"""Explicit provisioning only; inference never imports this downloader."""
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
HERE = Path(__file__).resolve().parent
|
||
|
|
|
||
|
|
|
||
|
|
def digest(path):
|
||
|
|
with path.open("rb") as source:
|
||
|
|
return hashlib.file_digest(source, "sha256").hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def verified(path, spec):
|
||
|
|
return path.is_file() and path.stat().st_size == spec["bytes"] and digest(path) == spec["sha256"]
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description="Explicitly download ~136 MB of pinned, verified model artifacts")
|
||
|
|
parser.add_argument("--model-dir", type=Path, default=HERE / "model")
|
||
|
|
args = parser.parse_args()
|
||
|
|
manifest = json.loads((HERE / "model-manifest.json").read_text())
|
||
|
|
for spec in manifest["files"]:
|
||
|
|
target = args.model_dir / spec["path"]
|
||
|
|
if verified(target, spec):
|
||
|
|
print(f"Verified existing {spec['path']}")
|
||
|
|
continue
|
||
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
temporary = target.with_suffix(target.suffix + ".part")
|
||
|
|
url = f"https://huggingface.co/{manifest['model_id']}/resolve/{manifest['revision']}/{spec['path']}"
|
||
|
|
print(f"Downloading {spec['path']} ({spec['bytes']} bytes)", flush=True)
|
||
|
|
try:
|
||
|
|
request = urllib.request.Request(url, headers={"User-Agent": "Searchi-semantic-proof/0.1"})
|
||
|
|
with urllib.request.urlopen(request, timeout=60) as response, temporary.open("wb") as output:
|
||
|
|
total = 0
|
||
|
|
while chunk := response.read(1024 * 1024):
|
||
|
|
total += len(chunk)
|
||
|
|
if total > spec["bytes"]:
|
||
|
|
raise ValueError("Download exceeds manifest size")
|
||
|
|
output.write(chunk)
|
||
|
|
if not verified(temporary, spec):
|
||
|
|
raise ValueError(f"Size or SHA256 mismatch for {spec['path']}")
|
||
|
|
os.replace(temporary, target)
|
||
|
|
print(f"Verified SHA256 {spec['sha256']}")
|
||
|
|
finally:
|
||
|
|
temporary.unlink(missing_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|