"""Search bounded decompressed prefixes of original unsigned local assets."""
import concurrent.futures
import json
import re
import zlib
from pathlib import Path
root=Path(r'D:/SteamLibrary/steamapps/common/Elite Dangerous/Products/elite-dangerous-odyssey-64/Win64')
pattern=re.compile(rb'galaxy.?map|visualisation|visualization|map.?colou?r|star.?class.?colou?r',re.I)
out=Path('artifacts/galaxy-query-research/palette-resources')
out.mkdir(exist_ok=True)
def scan(directory):
count=0;found=[]
for path in directory.iterdir():
if not path.is_file():continue
with path.open('rb') as stream:
header=stream.read(2)
if header[0:1]!=b'\x78' or len(header)!=2 or ((header[0]<<8)|header[1])%31:continue
data=header+stream.read(65534)
count+=1
try:prefix=zlib.decompressobj().decompress(data,32768)
except zlib.error:continue
if not pattern.search(prefix.replace(b'\0',b'')):continue
strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',prefix)]
matched=[value for value in strings if pattern.search(value.encode())]
found.append(dict(path=str(path),bytes=path.stat().st_size,prefixBytes=len(prefix),matched=matched,strings=strings[:24]))
(out/(path.name+'.prefix')).write_bytes(prefix)
return count,found
rows=[];count=0
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for observed,found in pool.map(scan,sorted(path for path in root.iterdir() if path.is_dir())):
count+=observed;rows.extend(found)
(out/'zlib-inventory.json').write_text(json.dumps(dict(scope='Read-only first32KiB original zlib asset prefixes',count=count,resources=rows),indent=2),encoding='utf-8')
print(f'zlib={count} matched={len(rows)}',flush=True)
"""Search bounded decompressed prefixes of original unsigned local assets."""
import concurrent.futures
import json
import re
import zlib
from pathlib import Path
root=Path(r'D:/SteamLibrary/steamapps/common/Elite Dangerous/Products/elite-dangerous-odyssey-64/Win64')
pattern=re.compile(rb'galaxy.?map|visualisation|visualization|map.?colou?r|star.?class.?colou?r',re.I)
out=Path('artifacts/galaxy-query-research/palette-resources')
out.mkdir(exist_ok=True)
def scan(directory):
count=0;found=[]
for path in directory.iterdir():
if not path.is_file():continue
with path.open('rb') as stream:
header=stream.read(2)
if header[0:1]!=b'\x78' or len(header)!=2 or ((header[0]<<8)|header[1])%31:continue
data=header+stream.read(65534)
count+=1
try:prefix=zlib.decompressobj().decompress(data,32768)
except zlib.error:continue
if not pattern.search(prefix.replace(b'\0',b'')):continue
strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',prefix)]
matched=[value for value in strings if pattern.search(value.encode())]
found.append(dict(path=str(path),bytes=path.stat().st_size,prefixBytes=len(prefix),matched=matched,strings=strings[:24]))
(out/(path.name+'.prefix')).write_bytes(prefix)
return count,found
rows=[];count=0
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for observed,found in pool.map(scan,sorted(path for path in root.iterdir() if path.is_dir())):
count+=observed;rows.extend(found)
(out/'zlib-inventory.json').write_text(json.dumps(dict(scope='Read-only first32KiB original zlib asset prefixes',count=count,resources=rows),indent=2),encoding='utf-8')
print(f'zlib={count} matched={len(rows)}',flush=True)
