_tools/audit_heavy_batch.py

"""Sequential read-back integrity and explicit package/chunk checks; no gameplay pass implied."""
import os,json,ast,re,hashlib,shutil,subprocess,sys,time
from pathlib import Path
from urllib.parse import urljoin,urlparse
from huggingface_hub import HfApi
H=Path('/home/user/game-archive');R=H/'reports';B='smodusermc/offline-games';api=HfApi(token=os.environ['HF_TOKEN']);ids=json.loads((R/'heavy-batch-plan.json').read_text())['ids'];stage=Path('/home/user/.cache/heavy-audit');results=[]
nodes=[]
for n in ast.parse((H/'tools/collect_next.py').read_text()).body:
 if isinstance(n,(ast.Import,ast.ImportFrom)):nodes.append(n)
 elif isinstance(n,ast.FunctionDef) and n.name in ['normalized','dest','extract','fetch']:nodes.append(n)
 elif isinstance(n,ast.Assign) and any(isinstance(x,ast.Name) and x.id in ['EXT','PATH_RE','LITERAL','IGNORE_HOSTS','sessions'] for x in n.targets):nodes.append(n)
ns={'BASE':'https://garbsoftball.com','ROOT':stage,'FILE_CAP':2*1024**3};exec(compile(ast.Module(body=nodes,type_ignores=[]),'discovery','exec'),ns)
for i in ids:
 shutil.rmtree(stage,ignore_errors=True);stage.mkdir(parents=True)
 mp=R/f'game-{i}.json';api.download_bucket_files(B,[(f'_reports/games/{i:04d}.json',mp)],raise_on_missing_files=True);m=json.loads(mp.read_text());records={f['path']:f for f in m['files']};invalid=[k for k in records if re.search(r'https?:',k)]
 if invalid:
  api.batch_bucket_files(B,delete=invalid)
  for k in invalid:records.pop(k)
 j={'id':i,'name':m['name'],'sha_comparisons':0,'sha_mismatches':[],'size_mismatches':[],'unbaselined':[],'repairs':[],'declarations':[],'unresolved':[],'full_playthrough_verified':False}
 # Download bounded batches; only one game's payload exists on disk.
 keys=list(records)
 for start in range(0,len(keys),16):api.download_bucket_files(B,[(k,stage/k) for k in keys[start:start+16]],raise_on_missing_files=True)
 texts={};seen=set();queue=list(records);attempted={f['url'] for f in m.get('failures',[])};j['additional_candidate_failures']=[]
 for key in queue:
  if key in seen:continue
  seen.add(key);p=stage/key
  if not p.exists() or p.stat().st_size>24*1024**2 or not re.search(r'\.(html?|js|mjs|json|count|css)$',key,re.I):continue
  text=p.read_text(errors='replace');texts[key]=text;url=records[key].get('url') or 'https://garbsoftball.com/'+key.removeprefix('site/')
  refs,external=ns['extract'](text,url,m['entry'])
  # Enumerated split/chapter/framework declarations are independent of the collector run.
  strict=refs
  for u,kind in strict:
   dep=ns['dest'](u);j['declarations'].append({'path':dep,'kind':kind,'input':key})
   if dep in records:continue
   if u in attempted:
    if kind in ['declared-split-part','ags-split-file','ags-support-file','chapter-entry','emulator-image','counted-tar-chunk']:j['unresolved'].append({'url':u,'kind':kind,'error':'Previously failed source request; see capture manifest'})
    continue
   attempted.add(u)
   f=ns['fetch']((u,kind))
   if 'error' in f:
    j['additional_candidate_failures'].append(f)
    if kind in ['declared-split-part','ags-split-file','ags-support-file','chapter-entry','emulator-image','counted-tar-chunk']:j['unresolved'].append(f)
    continue
   api.batch_bucket_files(B,add=[(f['local'],dep)]);target=stage/dep;target.parent.mkdir(parents=True,exist_ok=True);Path(f['local']).unlink(missing_ok=True)
   api.download_bucket_files(B,[(dep,target)],raise_on_missing_files=True)
   rec={a:b for a,b in f.items() if a not in ['local','kind']};records[dep]=rec;queue.append(dep);j['repairs'].append(rec)
 # Raw audio filename strings can denote embedded samples, so use the separate
 # check_gamemaker_audio.py SOND/group metadata pass instead of treating every string as a URL.
 j['gamemaker_audio_candidates']=[]
 # Source and bucket hashes, including all chunks and large binaries.
 for key,f in records.items():
  p=stage/key
  with p.open('rb') as stream:sha=hashlib.file_digest(stream,'sha256').hexdigest()
  if p.stat().st_size!=f['bytes']:j['size_mismatches'].append(key)
  if f.get('sha256'):
   j['sha_comparisons']+=1
   if sha!=f['sha256']:j['sha_mismatches'].append(key)
  else:j['unbaselined'].append({'path':key,'stored_sha256':sha})
 j['package_size_checks']=[]
 for key,text in texts.items():
  if not key.endswith('.html'):continue
  base='https://garbsoftball.com/'+key.removeprefix('site/')
  for name,n in re.findall(r'["\']([^"\']+\.(?:pck|wasm))["\']\s*:\s*(\d+)',text):
   dep=ns['dest'](urljoin(base,name));n=int(n);parts=sorted((k for k in records if re.fullmatch(re.escape(dep)+r'\.part\d+',k)),key=lambda x:int(x.rsplit('part',1)[1]))
   actual=records[dep]['bytes'] if dep in records else sum(records[k]['bytes'] for k in parts)
   j['package_size_checks'].append({'path':dep,'declared_bytes':n,'captured_bytes':actual,'split_parts':len(parts),'matches':actual==n})
 j['files']=len(records);j['bytes']=sum(f['bytes'] for f in records.values());j['capture_limits']=m.get('limits',[]);j['source_candidate_failures']=m.get('failures',[]);j['external_dependencies']=m.get('external_dependencies',[])
 j['stored_integrity_passed']=not(j['sha_mismatches'] or j['size_mismatches'] or j['unbaselined']);j['declared_packages_passed']=not j['unresolved'] and all(x['matches'] for x in j['package_size_checks'])
 m['files']=list(records.values());m['capture_limit_history']=m.get('capture_limit_history',[])+m.get('limits',[]);m['limits']=[];m['closure_audit_note']='Re-traversed stored scripts with improved filename, chapter, split-file and GameMaker-audio rules; unresolved source requests remain separately reported. Not an exhaustive runtime asset proof.';mp.write_text(json.dumps(m,indent=2));api.batch_bucket_files(B,add=[(mp,f'_reports/games/{i:04d}.json')]);shutil.rmtree(stage,ignore_errors=True)
 # Runtime checks are sequential, guarded, and separately classified; large builds may be deferred.
 subprocess.run([sys.executable,str(H/'tools/run_isolated.py'),str(i)],check=True,env={**os.environ,'GAME_VERIFY_STAGE':'/home/user/.cache/game-verify-stage','VERIFY_SECONDS':'70'})
 vp=R/f'verification-{i:04d}.json';runtime=json.loads(vp.read_text());j['runtime_status']=runtime.get('status');j['runtime_report']=f'_verification/games/{i:04d}.json';j['runtime_missing']=runtime.get('offline_missing_requests',{})
 results.append(j);report={'phase':'checking' if len(results)<len(ids) else 'checks_complete','results':results,'scope':'Read-back hashes and enumerated package/chunk declarations. Runtime separate; no full-playthrough verification.'};out=R/'heavy-batch-validation.json';out.write_text(json.dumps(report,indent=2))
 api.batch_bucket_files(B,add=[(out,'_verification/heavy-batch-validation.json'),(vp,j['runtime_report'])]);print('AUDIT',len(results),i,j['files'],j['bytes'],j['stored_integrity_passed'],j['declared_packages_passed'],j['runtime_status'],flush=True)
print('HEAVY AUDIT FINISHED',flush=True)