File size: 3,375 Bytes
cd3278e | 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 | """Prepare pinned upstream models, code and public corpus for a reproducible run."""
import argparse, gzip, hashlib, json, os, pathlib, shutil, tarfile, urllib.request
from huggingface_hub import snapshot_download
BASE = ('Qwen/Qwen3.5-4B-Base', '1001bb4d826a52d1f399e183466143f4da7b741b')
INIT = ('jaredpalmer/kev-4b', '485ace8703592fcf405488b262449990824cfed1')
KEV = '90990a5fac2995b9faa3190f7d437e84f2067768'
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--corpus', type=pathlib.Path, help='Unpacked SecJev-Corpus v1.0.0 directory')
parser.add_argument('--inference-only', action='store_true', help='Download pinned backbone and Kev code without training data or parent adapter')
args = parser.parse_args()
if not args.inference_only and args.corpus is None: parser.error('--corpus is required for training setup')
root = pathlib.Path(os.environ.get('SECJEV_WORK', './secjev-4b-work')).resolve()
for d in ['data', 'results', 'logs', 'runs', 'source']:
(root/d).mkdir(parents=True, exist_ok=True)
if not args.inference_only:
corpus = json.loads((args.corpus/'manifest.json').read_text())
assert corpus['name'] == 'SecJev-Corpus' and corpus['version'] == '1.0.0'
manifest = {}
for split, info in corpus['subsets'].items():
src = args.corpus/'records'/f'{split}.jsonl.gz'
dest = root/'data'/src.name
digest = hashlib.sha256()
with gzip.open(src, 'rt') as stream:
for line in stream:
r = json.loads(line)
payload = json.dumps({'state': r['state'], 'questions': r['questions']}, ensure_ascii=False, allow_nan=False, separators=(',', ':'))
digest.update((payload+'\n').encode())
assert digest.hexdigest() == info['content_sha256'], split
shutil.copy2(src, dest)
manifest[split] = {'records': info['scenes'], 'questions': info['questions'], 'sha256': hashlib.sha256(dest.read_bytes()).hexdigest()}
(root/'data/manifest.json').write_text(json.dumps(manifest, indent=2)+'\n')
for name, (repo, revision) in [('base', BASE), ('init', INIT)]:
if args.inference_only and name == 'init': continue
snapshot_download(repo, revision=revision, local_dir=root/name,
allow_patterns=['*.json', '*.safetensors', '*.pt', '*.txt', '*.jinja'])
filename = 'download_provenance.json' if name == 'base' else 'inference-provenance.json'
(root/name/filename).write_text(json.dumps({'repo_id': repo, 'revision': revision}, indent=2)+'\n')
archive = root/'source'/f'kev-{KEV}.tar.gz'
urllib.request.urlretrieve(f'https://codeload.github.com/jaredpalmer/kev/tar.gz/{KEV}', archive)
with tarfile.open(archive) as tf:
for member in tf:
if member.isfile():
rel = pathlib.PurePosixPath(member.name)
assert rel.parts[0] == f'kev-{KEV}' and '..' not in rel.parts
target = root/'source'/'kev'/pathlib.Path(*rel.parts[1:])
target.parent.mkdir(parents=True, exist_ok=True)
with tf.extractfile(member) as source, target.open('wb') as destination:
shutil.copyfileobj(source, destination)
print('Prepared', root)
if __name__ == '__main__':
main()
|