File size: 2,967 Bytes
f75155d | 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 | """Download a pinned release pair. Native Windows; never changes app files."""
import argparse
import hashlib
import json
import os
from pathlib import Path
from huggingface_hub import hf_hub_download,snapshot_download
ASR_REPO='humoiww/PhonoCommand-ASR'
def read(path):return json.loads(Path(path).read_text(encoding='utf-8'))
def sha(path):
h=hashlib.sha256()
with Path(path).open('rb') as f:
for block in iter(lambda:f.read(8*1024*1024),b''):h.update(block)
return h.hexdigest()
def verify(directory,files):
for name,info in files.items():
relative=Path(name)
if relative.is_absolute() or '..' in relative.parts:raise ValueError('Unsafe manifest path')
path=Path(directory)/relative
if path.stat().st_size!=info['size'] or sha(path)!=info['sha256']:
raise ValueError('Checksum mismatch: '+name)
def sync(output,release_lock=None,with_lid=False):
lock=read(release_lock or hf_hub_download(ASR_REPO,'release_set.json',revision='main'))
output=Path(output).resolve();output.mkdir(parents=True,exist_ok=True)
release_dir=output/'releases'/lock['release_id'];release_dir.mkdir(parents=True,exist_ok=True)
paths={}
for component,item in lock['components'].items():
if len(item['revision'])!=40:raise ValueError('Expected immutable component commit')
directory=release_dir/component/item['revision']
snapshot_download(item['repo_id'],revision=item['revision'],local_dir=directory,
allow_patterns=item['files'])
manifest_path=directory/'MANIFEST.json'
if sha(manifest_path)!=item['manifest_sha256']:raise ValueError('Release manifest mismatch')
verify(directory,read(manifest_path)['files'])
paths[component]=str(directory)
upstream=read(Path(paths['asr'])/'upstream_lock.json')
for key,item in upstream.items():
if key=='sensevoice' and not with_lid:continue
directory=output/'upstream'/key/item['revision']
snapshot_download(item['repo_id'],revision=item['revision'],local_dir=directory,
allow_patterns=list(item['files']))
verify(directory,item['files']);paths[key]=str(directory)
state={'release_id':lock['release_id'],'lock':lock,'paths':paths,
'includes_sensevoice':with_lid,'app_reload_required':True}
(release_dir/'release_set.json').write_text(json.dumps(lock,indent=2)+'\n',encoding='utf-8')
target=output/'active_release.json';temporary=output/'active_release.json.tmp'
temporary.write_text(json.dumps(state,indent=2)+'\n',encoding='utf-8')
os.replace(temporary,target)
print(target)
return state
if __name__=='__main__':
p=argparse.ArgumentParser();p.add_argument('--output',type=Path,required=True)
p.add_argument('--release-lock',type=Path);p.add_argument('--with-lid',action='store_true')
args=p.parse_args();sync(args.output,args.release_lock,args.with_lid)
|