File size: 3,643 Bytes
f8a1702 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """Fetch fixed official third-party assets; validate hashes; no model execution."""
import argparse
import hashlib
import importlib.metadata
import json
from pathlib import Path
import urllib.request
import zipfile
from huggingface_hub import hf_hub_download
def sha(path):
h=hashlib.sha256()
with Path(path).open('rb') as f:
for chunk in iter(lambda:f.read(1048576),b''): h.update(chunk)
return h.hexdigest()
def setup(bundle, offline=False):
bundle=Path(bundle).resolve()
spec=json.loads((bundle/'assets/runtime_sources.json').read_text())
checks=[]
for row in spec['sources']:
dest=bundle/row['destination']
dest.parent.mkdir(parents=True,exist_ok=True)
if not dest.exists():
if offline: raise FileNotFoundError(f'Offline asset missing: {row["destination"]}')
if row['kind']=='hub':
source=Path(hf_hub_download(row['repo_id'],row['filename'],revision=row['revision'],token=False))
assert sha(source)==row['sha256'], row['destination']
dest.write_bytes(source.read_bytes())
else:
with urllib.request.urlopen(row['url'],timeout=120) as response:
content=response.read()
if hashlib.sha256(content).hexdigest()!=row['sha256']:
raise RuntimeError(f'Asset checksum failed: {row["destination"]}')
dest.write_bytes(content)
if sha(dest)!=row['sha256']:
raise RuntimeError(f'Asset checksum failed, not overwriting: {row["destination"]}')
if row['kind']=='zip':
target=(bundle/row['extract_to']).resolve()
with zipfile.ZipFile(dest) as archive:
for member in archive.infolist():
path=(target/member.filename).resolve()
if target!=path and target not in path.parents:
raise RuntimeError('Unsafe resource archive path')
if not member.is_dir():
content=archive.read(member)
if path.exists():
if path.read_bytes()!=content:
raise RuntimeError(f'Installed resource differs: {path.name}')
else:
path.parent.mkdir(parents=True,exist_ok=True)
path.write_bytes(content)
checks.append(dict(path=row['destination'],sha256=row['sha256'],status='PASS'))
inventory=json.loads((bundle/'assets/prepared_resource_hashes.json').read_text())
dist=importlib.metadata.distribution('g2p-en')
if dist.version!=inventory['g2p_en_version']: raise RuntimeError('G2P package version differs')
for row in inventory['g2p_files']:
path=Path(dist.locate_file(row['package_relative']))
if not path.is_file() or sha(path)!=row['sha256']:
raise RuntimeError(f'G2P resource differs: {row["package_relative"]}')
checks.append(dict(path=row['package_relative'],sha256=row['sha256'],status='PASS'))
for row in inventory['prepared_resources']:
path=bundle/row['path']
if not path.is_file() or sha(path)!=row['sha256']:
raise RuntimeError(f'Prepared resource differs: {row["path"]}')
print(json.dumps(dict(offline=offline,checks=checks),indent=2))
return checks
if __name__=='__main__':
p=argparse.ArgumentParser()
p.add_argument('--bundle-dir',type=Path,default=Path(__file__).resolve().parents[1])
p.add_argument('--offline',action='store_true')
a=p.parse_args()
setup(a.bundle_dir,a.offline)
|