_tools/final_heavy_exceptions.py

"""Remove source error bodies masquerading as artwork; add documented case-only audio aliases."""
import os,json,hashlib,tempfile
from pathlib import Path
from huggingface_hub import HfApi
R=Path('/home/user/game-archive/reports');B='smodusermc/offline-games';api=HfApi(token=os.environ['HF_TOKEN']);media=json.loads((R/'heavy-batch-media.json').read_text());removed=[]
for game in media['results']:
 bad=[f for f in game['files'] if f.get('decoded') is False]
 if not bad:continue
 i=game['id'];mp=R/f'game-{i}.json';m=json.loads(mp.read_text());paths={f['path'] for f in bad};records={f['path']:f for f in m['files']}
 for f in bad:
  rec=records.get(f['path']);entry={'id':i,'path':f['path'],'sha256':f['stored_sha256'],'action':'Removed invalid source response from game-asset storage; retained failure evidence','source_problem':'Empty PNG response' if i==140 else 'HTTP 200 image/jpeg body was plain text: PS 404 Not Found!'};removed.append(entry);f['resolution']=entry['action'];f['source_problem']=entry['source_problem']
  if rec:m.setdefault('failures',[]).append({'url':rec.get('url'),'path':rec['path'],'error':entry['source_problem'],'invalid_response_sha256':f['stored_sha256']})
 api.batch_bucket_files(B,delete=sorted(paths));m['files']=[f for f in m['files'] if f['path'] not in paths];mp.write_text(json.dumps(m,indent=2));api.batch_bucket_files(B,add=[(mp,f'_reports/games/{i:04d}.json')])
q=R/'heavy-invalid-artwork.json';q.write_text(json.dumps(removed,indent=2));(R/'heavy-batch-media.json').write_text(json.dumps(media,indent=2));api.batch_bucket_files(B,add=[(q,'_verification/heavy-invalid-artwork.json'),(R/'heavy-batch-media.json','_verification/heavy-batch-media.json')])
mp=R/'game-131.json';m=json.loads(mp.read_text());records={f['path']:f for f in m['files']};gp=R/'gamemaker-audio-audit.json';g=json.loads(gp.read_text());aliases=[]
for missing in g['required_external_unresolved']:
 dst=missing['path'];matches=[k for k in records if k.lower()==dst.lower() and k!=dst]
 if len(matches)!=1:continue
 src=matches[0];f=records[src];info=next(iter(api.get_bucket_paths_info(B,[src])));api.batch_bucket_files(B,copy=[('bucket',B,info.xet_hash,dst)])
 with tempfile.TemporaryDirectory(dir='/home/user/.cache') as tmp:
  p=Path(tmp)/'alias.ogg';api.download_bucket_files(B,[(dst,p)],raise_on_missing_files=True)
  with p.open('rb') as stream:digest=hashlib.file_digest(stream,'sha256').hexdigest()
  assert digest==f['sha256']
 rec={**f,'path':dst,'restored_from_bucket_path':src,'repair_type':'Documented case-only compatibility alias','note':'The SOND metadata uses this spelling; the same folder supplies the lowercase original. Identical bytes copied, original retained. The uppercase URL is not claimed to exist upstream.'};records[dst]=rec;aliases.append({'path':dst,'source_path':src,'sha256':digest,'readback_match':True})
fixed={x['path'] for x in aliases};g['case_only_aliases']=aliases;g['required_external_unresolved']=[x for x in g['required_external_unresolved'] if x['path'] not in fixed]
for c in g['chapters']:
 for f in c.get('streamed_files',[]):
  if f['path'] in fixed:f.update(present=True,case_only_alias=True)
gp.write_text(json.dumps(g,indent=2));m['files']=list(records.values());mp.write_text(json.dumps(m,indent=2));p=R/'heavy-batch-validation.json';s=json.loads(p.read_text());x=next(x for x in s['results'] if x['id']==131);x['unresolved']=[f for f in x['unresolved'] if f.get('path') not in fixed];x['case_only_audio_aliases']=aliases;p.write_text(json.dumps(s,indent=2));api.batch_bucket_files(B,add=[(gp,'_verification/gamemaker-audio-audit.json'),(mp,'_reports/games/0131.json'),(p,'_verification/heavy-batch-validation.json')]);print('INVALID ARTWORK REMOVED',len(removed),'CASE ALIASES',len(aliases),'GMS EXTERNAL UNRESOLVED',len(g['required_external_unresolved']),flush=True)