93 lines
4.8 KiB
Python
93 lines
4.8 KiB
Python
"""Extract a bounded native BattlePlan and join it to a captured Client StageRow.
|
|
|
|
MemoryPack record fields follow the public GridFightBattlePlan/WavePlan/MonsterPlan
|
|
schemas. This deliberately rejects nonempty drop payloads and ambiguous matches.
|
|
It does not deserialize or publish the reference account's full saved run.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
|
|
|
|
def extract(data, penalty):
|
|
matches = []
|
|
for start in range(max(0, len(data) - 21)):
|
|
if data[start] != 5: # BattlePlan: camp, elite, penalty, difficulty, waves.
|
|
continue
|
|
camp, elite, rule, difficulty, count = struct.unpack_from('<IIIII', data, start + 1)
|
|
if rule != penalty or not camp or not elite or not 0 < count <= 128:
|
|
continue
|
|
offset = start + 21
|
|
waves = []
|
|
try:
|
|
for index in range(count):
|
|
if data[offset] != 3:
|
|
raise ValueError('wave schema differs')
|
|
offset += 1
|
|
wave_id, formation, total = struct.unpack_from('<III', data, offset)
|
|
offset += 12
|
|
if wave_id != index + 1 or not 0 < formation <= 4096 or not 0 < total <= 128:
|
|
raise ValueError('invalid wave fields')
|
|
for _ in range(total):
|
|
if data[offset] != 3:
|
|
raise ValueError('monster schema differs')
|
|
offset += 1
|
|
monster, star, drops = struct.unpack_from('<III', data, offset)
|
|
offset += 12
|
|
if not monster or not 0 < star <= 255 or drops != 0:
|
|
raise ValueError('unsupported monster/drop payload')
|
|
waves.append({'wave_id': wave_id, 'formation_wave_id': formation,
|
|
'monster_count': total})
|
|
matches.append((start, offset, elite, difficulty, waves))
|
|
except (ValueError, IndexError, struct.error):
|
|
continue
|
|
if len(matches) != 1:
|
|
raise ValueError(f'expected one typed native plan, found {len(matches)}')
|
|
return matches[0]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
for name in ('snapshot', 'source-binary', 'stage-log', 'catalog', 'output'):
|
|
parser.add_argument('--' + name, required=True, type=Path)
|
|
for name in ('route', 'chapter', 'section', 'division', 'penalty'):
|
|
parser.add_argument('--' + name, required=True, type=int)
|
|
args = parser.parse_args()
|
|
data = args.snapshot.read_bytes()
|
|
if len(data) > 4 * 1024 * 1024:
|
|
raise ValueError('reference run exceeds extraction bound')
|
|
start, end, elite, difficulty, waves = extract(data, args.penalty)
|
|
rows = [json.loads(line.split(' ROW ', 1)[1].split(' ', 1)[1])
|
|
for line in args.stage_log.read_text(encoding='utf-8-sig').splitlines()
|
|
if ' ROW ' in line]
|
|
if len(rows) != 1 or not rows[0]['Level'] or not rows[0]['HardLevelGroup']:
|
|
raise ValueError('one complete current StageRow required')
|
|
row = rows[0]
|
|
catalog = json.loads(args.catalog.read_text(encoding='utf-8-sig'))
|
|
if not any(n['ID'] == args.route and n['ChapterID'] == args.chapter and
|
|
n['SectionID'] == args.section for n in catalog['stage_routes']):
|
|
raise ValueError('route coordinates absent from target catalog')
|
|
recipe = dict(route_id=args.route, chapter_id=args.chapter, section_id=args.section,
|
|
division_id=args.division, is_overclock=False, penalty_id=args.penalty,
|
|
stage_id=row['StageID'], level=row['Level'], hard_level_group=row['HardLevelGroup'],
|
|
elite_group=elite, extra_difficulty=difficulty, waves=waves)
|
|
result = dict(format_version=1, target_client_version=catalog['client_version'],
|
|
target_assembly_sha256=catalog['provenance']['game_assembly_sha256'],
|
|
source_binary_sha256=hashlib.sha256(args.source_binary.read_bytes()).hexdigest(),
|
|
source_snapshot_sha256=hashlib.sha256(data).hexdigest(),
|
|
source_plan_range=[start, end],
|
|
source_plan_sha256=hashlib.sha256(data[start:end]).hexdigest(),
|
|
target_stage_capture_sha256=hashlib.sha256(args.stage_log.read_bytes()).hexdigest(),
|
|
scope='Captured native wave parameters joined to current Client stage scalars. '
|
|
'Monster identities/stars come from the live saved layer; no reference party copied. '
|
|
'Second-node entry only; complete combat modifiers and later progression remain incomplete.',
|
|
entries=[recipe])
|
|
args.output.write_text(json.dumps(result, indent=2) + '\n', encoding='utf-8')
|
|
print(f'Exported {len(waves)} wave definitions from native bytes {start}:{end}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|