Verstack Local
841a95c56a
Save the tested release workflows, RAM media processing, program discovery, FLIRT library, and shared-object maintenance. Add Git exclusions, file attributes, and instructions for a later push using a forwarded SSH agent.
107 lines
8.2 KiB
Python
107 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Sequential corpus validation in disposable directories; preserves receipts, not an archive import."""
|
|
import argparse,hashlib,json,pathlib,shutil,subprocess,sys,tempfile,time,os
|
|
sys.path.insert(0,str(pathlib.Path(__file__).resolve().parents[1]/'plugins'))
|
|
import spike_package,media_extract
|
|
from inventory_corpus import atomic
|
|
from ram_workspace import ram_workspace, scratch_environment
|
|
|
|
def invoke(script,request,path):
|
|
path.write_text(json.dumps(request))
|
|
# Each decoder gets a fresh process and a hard wall timeout. Core archive
|
|
# execution adds its own process-group cleanup and workspace monitoring.
|
|
child=subprocess.Popen([sys.executable,str(pathlib.Path(__file__).resolve().parents[1]/'plugins'/script),str(path)],start_new_session=True,env=scratch_environment(path.parent),cwd=path.parent)
|
|
try:
|
|
if child.wait(timeout=7200):raise ValueError(f'{script} failed; inspect validation log')
|
|
except BaseException:
|
|
import signal
|
|
try:os.killpg(child.pid,signal.SIGKILL)
|
|
except ProcessLookupError:pass
|
|
child.wait();raise
|
|
result=json.loads(pathlib.Path(request['result_file']).read_text())
|
|
if result.get('error'):raise ValueError(result['error'])
|
|
return result
|
|
|
|
def validate_families(root, output, media, request, report, package, row, config):
|
|
"""Exercise node/audio adapters and representative browser preview readback.
|
|
|
|
Keep stage failures independent, so a codec failure cannot erase node evidence.
|
|
Preview samples are explicit; they do not certify every decoded asset.
|
|
"""
|
|
from configure_pipeline import configure
|
|
pipeline=configure(json.loads(json.dumps(config)),pathlib.Path(__file__).resolve().parents[1])['plugins']
|
|
stages={};row['family_stages']=stages
|
|
decoded_roots=[('media',media)]
|
|
for name,script,evidence_name in [('node-extract','node_extract.py','node-evidence.json'),('spike2-audio','spike2_audio.py','spike2-audio-evidence.json')]:
|
|
destination=root/name;destination.mkdir()
|
|
try:
|
|
result=invoke(script,{**request,'input_dir':str(output),'output_dir':str(destination),'settings':pipeline[name]['settings'],'result_file':str(root/(name+'-result.json'))},root/(name+'-request.json'))
|
|
evidence=json.loads((destination/evidence_name).read_text())
|
|
atomic(report/(package+'.'+evidence_name),evidence)
|
|
stages[name]={'result':{k:v for k,v in result.items() if k!='files'},'evidence':package+'.'+evidence_name}
|
|
if name=='spike2-audio':decoded_roots.append((name,destination))
|
|
except Exception as error:stages[name]={'error':str(error)}
|
|
samples=root/'preview-samples';samples.mkdir();selected=[];seen=set()
|
|
# At most one source of each detected type in each decoder output tree.
|
|
# Record exact identities and source paths so sampling is reproducible.
|
|
for family,directory in decoded_roots:
|
|
for path in sorted(directory.rglob('*')):
|
|
if not path.is_file() or path.is_symlink():continue
|
|
with path.open('rb') as f:kind=media_extract.sniff(f.read(32))
|
|
key=(family,kind)
|
|
if kind is None or key in seen:continue
|
|
seen.add(key)
|
|
target=samples/(str(len(selected))+kind);shutil.copyfile(path,target)
|
|
with path.open('rb') as f:identity=hashlib.file_digest(f,'sha256').hexdigest()
|
|
selected.append({'sample':target.name,'family':family,'source':path.relative_to(directory).as_posix(),'sha256':identity})
|
|
destination=root/'previews';destination.mkdir()
|
|
try:
|
|
result=invoke('media_preview.py',{**request,'input_dir':str(samples),'output_dir':str(destination),'settings':{},'result_file':str(root/'preview-result.json')},root/'preview-request.json')
|
|
evidence=json.loads((destination/'preview-evidence.json').read_text());evidence['sampling']={'scope':'one detected type per decoder output; not exhaustive playback','selected':selected}
|
|
atomic(report/(package+'.preview-evidence.json'),evidence)
|
|
stages['media-preview']={'result':{k:v for k,v in result.items() if k!='files'},'sample_count':len(selected),'evidence':package+'.preview-evidence.json'}
|
|
except Exception as error:stages['media-preview']={'error':str(error)}
|
|
|
|
def validate(source,report,config,refresh_stale=False,all_families=False):
|
|
settings={**config['plugins']['spike3-extract']['settings'],**{k:v for k,v in config['plugins']['spike3-unpack']['settings'].items() if k in ('key_file','cryptsetup','debugfs')}}
|
|
packages=sorted(p for p in source.iterdir() if p.name.lower().endswith(('.spk','.spk.zip')))
|
|
report.mkdir(parents=True,exist_ok=True)
|
|
scratch,budget=ram_workspace(config)
|
|
print(f'RAM workspace: {scratch}; allowance {budget//1024**3} GiB; only receipts are persisted',flush=True)
|
|
for i,path in enumerate(packages):
|
|
receipt=report/(path.name+'.json')
|
|
processing={'package':2,'media':media_extract.REVISION}
|
|
if all_families:processing.update(node=2,spike2_audio=1,preview=3,sampling=1)
|
|
if receipt.exists():
|
|
previous=json.loads(receipt.read_text())
|
|
if (not all_families or previous.get('processing')==processing) and (not refresh_stale or previous.get('processing')==processing):continue
|
|
row={'schema':2,'processing':processing,'package':path.name,'bytes':path.stat().st_size,'scope':'disposable validation; not archived','started':time.time()}
|
|
print(f'BEGIN {i+1}/{len(packages)} {path.name}',flush=True)
|
|
try:
|
|
if min(shutil.disk_usage(scratch).free,budget)<path.stat().st_size*7+1024**3:raise ValueError('insufficient RAM headroom for bounded validation')
|
|
with tempfile.TemporaryDirectory(prefix='corpus-',dir=scratch) as tmp:
|
|
root=pathlib.Path(tmp);inputs=root/'input';output=root/'output';media=root/'media';inputs.mkdir();output.mkdir();media.mkdir()
|
|
shutil.copyfile(path,inputs/path.name)
|
|
request={'protocol':1,'input_dir':str(inputs),'output_dir':str(output),'result_file':str(root/'result.json'),'settings':settings,'workspace_bytes':budget}
|
|
result=invoke('spike_package.py',request,root/'extract-request.json');row['extraction']=result
|
|
entries=[]
|
|
for p in sorted(output.rglob('*')):
|
|
if not p.is_file():continue
|
|
with p.open('rb') as f:digest=hashlib.file_digest(f,'sha256').hexdigest()
|
|
entries.append({'path':p.relative_to(output).as_posix(),'bytes':p.stat().st_size,'sha256':digest})
|
|
row['inventory']=entries
|
|
row['wrapper']=json.loads((output/'wrapper-evidence.json').read_text())
|
|
decoded=invoke('media_extract.py',{**request,'input_dir':str(output),'output_dir':str(media),'settings':{},'result_file':str(root/'media-result.json')},root/'media-request.json')
|
|
row['media_result']={k:v for k,v in decoded.items() if k!='files'}
|
|
evidence=json.loads((media/'media-evidence.json').read_text());row['processing']['media']=evidence['processing_revision'];atomic(report/(path.name+'.media.json'),evidence)
|
|
row['media_counts']={}
|
|
for asset in evidence['assets']:
|
|
status=asset['status'];row['media_counts'][status]=row['media_counts'].get(status,0)+1
|
|
if all_families:validate_families(root,output,media,request,report,path.name,row,config)
|
|
row['status']='complete' if decoded['coverage']=='complete' else 'coverage_blocked'
|
|
if all_families and any(s.get('error') or s.get('result',{}).get('coverage')!='complete' for s in row['family_stages'].values()):row['status']='coverage_blocked'
|
|
except Exception as error:row.update(status='failed',error=str(error))
|
|
row['finished']=time.time();atomic(receipt,row);print(f"END {path.name} {row['status']}",flush=True)
|
|
if __name__=='__main__':
|
|
p=argparse.ArgumentParser();p.add_argument('source',type=pathlib.Path);p.add_argument('report',type=pathlib.Path);p.add_argument('--config',type=pathlib.Path,default=pathlib.Path('config.json'));p.add_argument('--refresh-stale',action='store_true');p.add_argument('--all-families',action='store_true');a=p.parse_args();validate(a.source,a.report,json.loads(a.config.read_text()),a.refresh_stale,a.all_families)
|