| """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) |
|
|
