File size: 3,071 Bytes
7034da5 | 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 | """Build a full accession snapshot in resumable parallel partitions.
Run list_annotation_objects.py first. This command indexes the captured inventory,
merges all partitions, and leaves publication to the operator.
"""
import argparse
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
ROOT=Path(__file__).resolve().parent
def main():
p=argparse.ArgumentParser(description=__doc__)
p.add_argument('--inventory',type=Path,required=True)
p.add_argument('--cache-dir',type=Path,default=ROOT/'.cache/full-index')
p.add_argument('--output',type=Path,default=ROOT/'data/catalog.sqlite')
p.add_argument('--parts',type=int,default=8)
p.add_argument('--workers-per-part',type=int,default=32)
args=p.parse_args()
if not 1<=args.parts<=32 or not 1<=args.workers_per_part<=128: p.error('Invalid concurrency')
work=args.cache_dir.resolve(); work.mkdir(parents=True,exist_ok=True)
config=work/'partition-config.json'
if config.exists() and json.loads(config.read_text())['parts']!=args.parts:
p.error('Use the same partition count when resuming, or a new cache directory.')
config.write_text(json.dumps({'parts':args.parts}))
items=[json.loads(s) for s in args.inventory.open()]
expected={x['path']:x['hash'] for x in items}
seen=set(); assigned=[[] for _ in range(args.parts)]
for i in range(args.parts):
done=work/f'part-{i}.sqlite'; checkpoint=work/f'part-{i}.building.sqlite'
if done.exists() and not checkpoint.exists(): done.rename(checkpoint)
if checkpoint.exists():
with sqlite3.connect(checkpoint) as c:
retained={r[0] for r in c.execute('SELECT path,hash FROM files') if expected.get(r[0])==r[1]}
if retained & seen: raise ValueError('Duplicate checkpoint objects')
assigned[i]=[x for x in items if x['path'] in retained]
seen.update(retained)
for n,item in enumerate(x for x in items if x['path'] not in seen):
assigned[n%args.parts].append(item)
processes=[]
for i in range(args.parts):
inventory=work/f'inventory-{i}.jsonl'
inventory.write_text(''.join(json.dumps(x)+'\n' for x in assigned[i]))
log=(work/f'part-{i}.log').open('w')
command=[sys.executable,str(ROOT/'build_full_index.py'),'--inventory',str(inventory),
'--output',str(work/f'part-{i}.sqlite'),'--workers',str(args.workers_per_part)]
processes.append((i,subprocess.Popen(command,stdout=log,stderr=subprocess.STDOUT),log))
failed=[]
for i,process,log in processes:
status=process.wait(); log.close()
if status: failed.append(i)
if failed: raise RuntimeError(f'Partitions {failed} failed; inspect logs and rerun to resume.')
subprocess.run([sys.executable,str(ROOT/'merge_full_index.py'),'--inventory',str(args.inventory.resolve()),
'--parts',*[str(work/f'part-{i}.sqlite') for i in range(args.parts)],'--output',str(args.output.resolve())],check=True)
if __name__=='__main__': main()
|