"""Search original indexed code/activity data blocks, without rescanning headers."""
import concurrent.futures
import hashlib
import json
import re
import runpy
import sys
import zlib
from pathlib import Path
sys.argv=['inventory','artifacts/galaxy-query-research/palette-resources/unused.json']
assets=runpy.run_path('research/galaxy_asset_inventory.py')
out=Path('artifacts/galaxy-query-research/palette-resources')
metadata=json.loads((out/'signed-metadata.json').read_text())
pattern=re.compile(rb'GalaxyMapVisualisation|GalaxyMapView|tradeData|noData|unclaimed|alreadyPopulated|localPlayerClaimed|starclass',re.I)
selected=[row for row in metadata['headers'] if len(row['strings'])>1 and row['strings'][1] in (':activity','STATIC')]
def scan(row):
key=row['key'];path=assets['ROOT']/('0'+key[0])/key
if path.stat().st_size>8*1024*1024:return None
prefix=assets['decode'](path)
blocks=[prefix];descriptions=[]
for match in re.finditer(rb'\x78[\x01\x9c\xda]',prefix):
try:
decoder=zlib.decompressobj();body=decoder.decompress(prefix[match.start():],32*1024*1024)
if not decoder.eof:continue
except zlib.error:continue
if len(body)<16:continue
blocks.append(body);descriptions.append(dict(offset=match.start(),bytes=len(body)))
if len(blocks)>=9:break
hits=[]
for index,body in enumerate(blocks):
matches=[dict(offset=match.start(),term=match.group().decode()) for match in pattern.finditer(body)]
if not matches:continue
(out/(key+f'.indexed{index}')).write_bytes(body)
hits.append(dict(block=index,matches=matches,strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',body)][:1000]))
if hits:return dict(path=str(path),sha256=hashlib.sha256(path.read_bytes()).hexdigest(),header=row['strings'],blocks=descriptions,hits=hits)
rows=[]
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for count,result in enumerate(pool.map(scan,selected),1):
if result:rows.append(result);print('hit',result['path'],[m['term'] for hit in result['hits'] for m in hit['matches']][:12],flush=True)
if count%500==0:print('searched',count,'/',len(selected),flush=True)
(out/'indexed-palette-search.json').write_text(json.dumps(dict(scope='Original indexed activity/static assets<=8MiB; complete zlib blocks<=32MiB, no game access',count=len(selected),resources=rows),indent=2),encoding='utf-8')
print('complete',len(selected),'assets, matches',len(rows),flush=True)
"""Search original indexed code/activity data blocks, without rescanning headers."""
import concurrent.futures
import hashlib
import json
import re
import runpy
import sys
import zlib
from pathlib import Path
sys.argv=['inventory','artifacts/galaxy-query-research/palette-resources/unused.json']
assets=runpy.run_path('research/galaxy_asset_inventory.py')
out=Path('artifacts/galaxy-query-research/palette-resources')
metadata=json.loads((out/'signed-metadata.json').read_text())
pattern=re.compile(rb'GalaxyMapVisualisation|GalaxyMapView|tradeData|noData|unclaimed|alreadyPopulated|localPlayerClaimed|starclass',re.I)
selected=[row for row in metadata['headers'] if len(row['strings'])>1 and row['strings'][1] in (':activity','STATIC')]
def scan(row):
key=row['key'];path=assets['ROOT']/('0'+key[0])/key
if path.stat().st_size>8*1024*1024:return None
prefix=assets['decode'](path)
blocks=[prefix];descriptions=[]
for match in re.finditer(rb'\x78[\x01\x9c\xda]',prefix):
try:
decoder=zlib.decompressobj();body=decoder.decompress(prefix[match.start():],32*1024*1024)
if not decoder.eof:continue
except zlib.error:continue
if len(body)<16:continue
blocks.append(body);descriptions.append(dict(offset=match.start(),bytes=len(body)))
if len(blocks)>=9:break
hits=[]
for index,body in enumerate(blocks):
matches=[dict(offset=match.start(),term=match.group().decode()) for match in pattern.finditer(body)]
if not matches:continue
(out/(key+f'.indexed{index}')).write_bytes(body)
hits.append(dict(block=index,matches=matches,strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',body)][:1000]))
if hits:return dict(path=str(path),sha256=hashlib.sha256(path.read_bytes()).hexdigest(),header=row['strings'],blocks=descriptions,hits=hits)
rows=[]
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for count,result in enumerate(pool.map(scan,selected),1):
if result:rows.append(result);print('hit',result['path'],[m['term'] for hit in result['hits'] for m in hit['matches']][:12],flush=True)
if count%500==0:print('searched',count,'/',len(selected),flush=True)
(out/'indexed-palette-search.json').write_text(json.dumps(dict(scope='Original indexed activity/static assets<=8MiB; complete zlib blocks<=32MiB, no game access',count=len(selected),resources=rows),indent=2),encoding='utf-8')
print('complete',len(selected),'assets, matches',len(rows),flush=True)
