_tools/check_gamemaker_audio.py

"""Distinguish embedded SOND audio from actual external streams/audio-group packages.
A raw .ogg string inside game.unx is NOT by itself an external dependency.
"""
import os,json,struct,mmap,re,hashlib,shutil,requests
from pathlib import Path
from urllib.parse import urljoin,unquote,urlparse
from huggingface_hub import HfApi
R=Path('/home/user/game-archive/reports');B='smodusermc/offline-games';api=HfApi(token=os.environ['HF_TOKEN']);stage=Path('/home/user/.cache/gamemaker-audio');stage.mkdir(parents=True,exist_ok=True);mp=R/'game-131.json';m=json.loads(mp.read_text());records={x['path']:x for x in m['files']};report={'id':131,'method':'Parse FORM/SOND flags, audio group IDs and audio indices. Embedded audio names are not treated as missing external files.','chapters':[],'repairs':[],'required_external_unresolved':[]};expected=set()
def sha(p):
 with p.open('rb') as f:return hashlib.file_digest(f,'sha256').hexdigest()
def chunks(b):
 assert b[:4]==b'FORM','Not a GameMaker FORM container'
 assert struct.unpack_from('<I',b,4)[0]+8==len(b),'FORM byte count does not match file'
 pos=8;out={}
 while pos+8<=len(b):
  n=struct.unpack_from('<I',b,pos+4)[0];assert pos+8+n<=len(b),'Chunk outside container';out[b[pos:pos+4].decode('ascii')]=(pos+8,n);pos+=8+n
 assert pos==len(b),'Incomplete chunk table'
 return out
def ensure(key,reason):
 assert '..' not in Path(key).parts and '\x00' not in key,'Unsafe resource path'
 expected.add(key);p=stage/key;p.parent.mkdir(parents=True,exist_ok=True)
 if key in records:
  if not p.exists():api.download_bucket_files(B,[(key,p)],raise_on_missing_files=True)
  assert p.stat().st_size==records[key]['bytes']
  if records[key].get('sha256'):assert sha(p)==records[key]['sha256']
  return p
 u='https://garbsoftball.com/'+key.removeprefix('site/');h=hashlib.sha256();n=0
 try:
  with requests.get(u,stream=True,timeout=(10,20)) as r:
   r.raise_for_status()
   with p.open('wb') as f:
    for block in r.iter_content(1024*1024):
     if n==0 and re.match(br'\s*(?:<!doctype|<html)',block,re.I):raise ValueError('HTML fallback, not a game asset')
     n+=len(block);assert n<=2*1024**3,'Per-file safety limit';h.update(block);f.write(block)
  source_sha=h.hexdigest();api.batch_bucket_files(B,add=[(p,key)]);p.unlink();api.download_bucket_files(B,[(key,p)],raise_on_missing_files=True);assert sha(p)==source_sha
  rec={'path':key,'url':u,'bytes':n,'sha256':source_sha,'discovery':reason};records[key]=rec;report['repairs'].append(rec);return p
 except Exception as e:
  p.unlink(missing_ok=True);report['required_external_unresolved'].append({'path':key,'url':u,'reason':reason,'error':str(e)[:220]});return None
for chapter in range(1,6):
 base=f'site/games/deltarune/chapter{chapter}/';parts=sorted((k for k in records if re.fullmatch(re.escape(base)+r'game\.unx\.part\d+',k)),key=lambda k:int(k.rsplit('part',1)[1]));sources=parts or ([base+'game.unx'] if base+'game.unx' in records else []);row={'chapter':chapter,'package_files':sources,'sounds':0,'embedded_sounds':0,'streamed_sounds':0,'group_ids':[],'container_errors':[]}
 if not sources:row['container_errors'].append('No captured game.unx or split package');report['chapters'].append(row);continue
 package=stage/'assembled.unx'
 with package.open('wb') as dst:
  for key in sources:
   p=ensure(key,'GameMaker package')
   if p is None:raise RuntimeError('Package unavailable: '+key)
   with p.open('rb') as src:shutil.copyfileobj(src,dst,1024*1024)
 sounds=[]
 try:
  with package.open('rb') as stream:
   b=mmap.mmap(stream.fileno(),0,access=mmap.ACCESS_READ);cs=chunks(b);off,size=cs['SOND'];n=struct.unpack_from('<I',b,off)[0];assert 4+n*4<=size;row['sounds']=n;group_ids=set()
   for i in range(n):
    ptr=struct.unpack_from('<I',b,off+4+i*4)[0];assert ptr+36<=len(b);v=struct.unpack_from('<9I',b,ptr);end=b.find(b'\0',v[3]);assert 0<=v[3]<=end<len(b);filename=b[v[3]:end].decode('utf8').replace('\\','/');embedded=bool(v[1]&1);group,idx=v[7:9]
    sounds.append({'filename':filename,'embedded':embedded,'group_id':group,'audio_index':idx})
    if embedded:row['embedded_sounds']+=1;group_ids.add(group)
    else:row['streamed_sounds']+=1
   row['group_ids']=sorted(group_ids);row['main_audio_entries']=struct.unpack_from('<I',b,cs['AUDO'][0])[0] if 'AUDO' in cs else 0;b.close()
  # Emscripten often embeds audiogroupN.dat in runner.data: do not mistake it for a missing HTTP file.
  loader=ensure(base+'runner.js','GameMaker loader metadata');packed={}
  if loader:
   for obj in re.findall(r'\{[^{}]*"filename"[^{}]*\}',loader.read_text(errors='replace')):
    try:
     ent=json.loads(obj)
     if all(k in ent for k in ['filename','start','end']):packed[ent['filename'].removeprefix('/assets/')]=ent
    except ValueError:pass
  runner_parts=sorted((k for k in records if re.fullmatch(re.escape(base)+r'runner\.data\.part\d+',k)),key=lambda k:int(k.rsplit('part',1)[1]));runner_sources=runner_parts or ([base+'runner.data'] if base+'runner.data' in records else [])
  runner=stage/'assembled-runner.data'
  if packed and runner_sources:
   with runner.open('wb') as dst:
    for key in runner_sources:
     source=ensure(key,'Emscripten preload package')
     with source.open('rb') as src:shutil.copyfileobj(src,dst,1024*1024)
  def embedded_file(name):
   ent=packed.get(name)
   if ent is None or not runner.exists():return None
   start,end=ent['start'],ent['end'];assert 0<=start<=end<=runner.stat().st_size,'Preloaded file outside runner.data'
   dst=stage/'extracted-audio-group.dat'
   with runner.open('rb') as src,dst.open('wb') as out:
    src.seek(start);left=end-start
    while left:
     block=src.read(min(left,1024*1024));assert block;out.write(block);left-=len(block)
   return dst
  groups={0:row['main_audio_entries']};row['group_packages']=[]
  for group in sorted(group_ids-{0}):
   key=base+f'audiogroup{group}.dat';p=embedded_file(f'audiogroup{group}.dat');bundled=p is not None
   if not bundled:p=ensure(key,'SOND external audio group')
   if p:
    with p.open('rb') as f:
     b=mmap.mmap(f.fileno(),0,access=mmap.ACCESS_READ);cs=chunks(b);groups[group]=struct.unpack_from('<I',b,cs['AUDO'][0])[0];b.close()
    row['group_packages'].append({'path':key,'audio_entries':groups[group],'source_bucket_hash_match':True,'bundled_in_runner_data':bundled,'package_files':runner_sources if bundled else [key]})
  row['embedded_index_errors']=[];row['streamed_files']=[]
  for sound in sounds:
   if sound['embedded']:
    if sound['group_id'] in groups and sound['audio_index']>=groups[sound['group_id']]:row['embedded_index_errors'].append(sound)
   else:
    key='site/'+unquote(urlparse(urljoin('https://garbsoftball.com/'+base.removeprefix('site/'),sound['filename'])).path).lstrip('/');p=embedded_file(sound['filename']);bundled=p is not None
    if not bundled:p=ensure(key,'SOND non-embedded streaming sound')
    row['streamed_files'].append({'path':key,'present':p is not None,'bundled_in_runner_data':bundled})
 except Exception as e:row['container_errors'].append(str(e))
 report['chapters'].append(row);package.unlink(missing_ok=True);(stage/'assembled-runner.data').unlink(missing_ok=True);print('GAMEMAKER',chapter,row['sounds'],row['embedded_sounds'],row['streamed_sounds'],'group packages',len(row.get('group_packages',[])),flush=True)
report['expected_paths']=sorted(expected);m['files']=list(records.values());m['gamemaker_audio_note']=report['method'];mp.write_text(json.dumps(m,indent=2));q=R/'gamemaker-audio-audit.json';q.write_text(json.dumps(report,indent=2));api.batch_bucket_files(B,add=[(mp,'_reports/games/0131.json'),(q,'_verification/gamemaker-audio-audit.json')])
s=R/'heavy-batch-validation.json';state=json.loads(s.read_text());x=next(x for x in state['results'] if x['id']==131);x['binary_string_probes_not_proven_dependencies']=[f for f in x['unresolved'] if f.get('kind')=='GameMaker binary audio filename'];x['unresolved']=[f for f in x['unresolved'] if f.get('kind')!='GameMaker binary audio filename']+report['required_external_unresolved'];x['gamemaker_audio_audit']='_verification/gamemaker-audio-audit.json';x['gamemaker_embedded_count']=sum(c['embedded_sounds'] for c in report['chapters']);x['gamemaker_streamed_count']=sum(c['streamed_sounds'] for c in report['chapters']);x['gamemaker_group_repairs']=report['repairs'];x['declared_packages_passed']=not x['unresolved'] and all(c['matches'] for c in x['package_size_checks']) and not any(c['container_errors'] or c.get('embedded_index_errors') for c in report['chapters']);s.write_text(json.dumps(state,indent=2));api.batch_bucket_files(B,add=[(s,'_verification/heavy-batch-validation.json')]);shutil.rmtree(stage,ignore_errors=True)