| import json, sys, os | |
| from pathlib import Path | |
| from huggingface_hub import list_bucket_tree, download_bucket_files, batch_bucket_files | |
| from huggingface_hub.errors import EntryNotFoundError | |
| action,filename=sys.argv[1:]; job=Path(filename); p=json.loads(job.read_text()); result={} | |
| if action in ('list','restore'): | |
| try: entries=[x.path for x in list_bucket_tree(p['bucket'],prefix=p['prefix'],recursive=True) if x.type=='file'] | |
| except EntryNotFoundError: entries=[] | |
| if action=='list': result=entries | |
| else: | |
| root=Path(p['root']).resolve(); prefix=p['prefix'].rstrip('/')+'/' | |
| jobs=[] | |
| for key in entries: | |
| if not key.startswith(prefix) or key.endswith('.tmp'): continue | |
| local=(root/key[len(prefix):]).resolve() | |
| if not local.is_relative_to(root): raise RuntimeError('Unsafe bucket path') | |
| if not local.exists(): local.parent.mkdir(parents=True,exist_ok=True); jobs.append((key,str(local))) | |
| if jobs: download_bucket_files(p['bucket'],files=jobs,raise_on_missing_files=True) | |
| elif action=='download': | |
| jobs=[] | |
| for remote,local in p['files']: | |
| q=Path(local); q.parent.mkdir(parents=True,exist_ok=True) | |
| if not q.exists(): jobs.append((remote,local)) | |
| if jobs: download_bucket_files(p['bucket'],files=jobs,raise_on_missing_files=True) | |
| elif action=='upload': | |
| batch_bucket_files(p['bucket'],add=[tuple(x) for x in p['files']]) | |
| print('BACKUP OK:',len(p['files']),'files',flush=True) | |
| else: raise ValueError(action) | |
| job.with_suffix('.result.json').write_text(json.dumps(result)) | |
Xet Storage Details
- Size:
- 1.59 kB
- Xet hash:
- 4f8bdbbb4b6e7d4038d279266cbb30c2c59927234bff69df9e4aa8e93e8e3f6c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.