218 lines
9.6 KiB
Python
218 lines
9.6 KiB
Python
"""Unlock the Veteran and Elite difficulties for a Dungeon Siege II party, as if it had finished the earlier ones.
|
|
|
|
Self-contained (Python 3 standard library only): copy this one file anywhere, then drag a .ds2party file onto it,
|
|
or run it from a command prompt.
|
|
|
|
usage: python ds2_unlock.py PARTY.ds2party [--world ds2_world] [--to veteran|elite] [--dry-run]
|
|
PARTY.ds2party e.g. Documents\\My Games\\Dungeon Siege 2\\Save\\<party>\\<party>.ds2party (single player) or
|
|
...\\Save\\MultiPlayer\\<party>\\<party>.ds2party; the save as it was is kept as PARTY.ds2party.unlockbak
|
|
--world the map whose campaign counts (default ds2_world, the Dungeon Siege II campaign; Broken World's
|
|
is ds2x_world)
|
|
--to elite (default: Veteran and Elite) or veteran (Veteran only)
|
|
Close the game first: it rewrites the party file when it exits.
|
|
|
|
How the game decides (Steam v2.3 sub_4171d7, called from the difficulty choice in single player 0x44d9ef and
|
|
multiplayer 0x4e44a9): Veteran needs the party's journal flag "<world>_completed_mercenary" and Elite needs
|
|
"<world>_completed_veteran". Flags live in the party's LocalJournal, one section per difficulty
|
|
(LocalJournal::SetWorld sub_83261f), set with LocalJournal::SetBool, which stores the int 1. The journal is saved
|
|
in party.ds2 inside the .ds2party tank as
|
|
u32 section count, then per section (sorted by name): string name, u32 entry count,
|
|
entries (sorted by key): string key, string value (string = u32 length + bytes)
|
|
Verified by having the game set both flags itself (harness calls of SetWorld/SetBool, then Save and Exit) and
|
|
comparing the saved file; this tool writes the same entries.
|
|
"""
|
|
import os, shutil, struct, sys, zlib
|
|
|
|
DIFFICULTIES = ('elite', 'mercenary', 'veteran')
|
|
|
|
|
|
# ---- party files: DSg2Tank archives with a flat file list (same format as tools/tank.py, inlined here) ----
|
|
def tank_files(d):
|
|
"""[(name, data)] of a flat tank, in storage order."""
|
|
if d[:8] != b'DSg2Tank':
|
|
raise SystemExit('not a Dungeon Siege II party file (no DSg2Tank header)')
|
|
fileset, data_offset = struct.unpack_from('<I', d, 0x10)[0], struct.unpack_from('<I', d, 0x18)[0]
|
|
files = []
|
|
for i in range(struct.unpack_from('<I', d, fileset)[0]):
|
|
o = fileset + struct.unpack_from('<I', d, fileset + 4 + 4 * i)[0]
|
|
parent, size, off, crc = struct.unpack_from('<4I', d, o)
|
|
fmt, flags, nlen = struct.unpack_from('<HHH', d, o + 24)
|
|
name = d[o + 30:o + 30 + nlen].decode('latin1')
|
|
p = o + 30 + nlen + 1
|
|
p += (-(p - o)) % 4
|
|
base = data_offset + off
|
|
if not fmt:
|
|
blob = d[base:base + size]
|
|
else: # zlib chunks: (size, compressed size, extra, offset)
|
|
csize, chunk = struct.unpack_from('<II', d, p)
|
|
blob = b''
|
|
for k in range((size + chunk - 1) // chunk if chunk else 0):
|
|
usize, cs, extra, coff = struct.unpack_from('<4I', d, p + 8 + 16 * k)
|
|
raw = d[base + coff:base + coff + cs + extra]
|
|
blob += raw[:usize] if usize == cs else (zlib.decompressobj().decompress(raw[:cs]) + raw[cs:])[:usize]
|
|
if zlib.crc32(blob) & 0xffffffff != crc:
|
|
raise SystemExit('%s in the party file is damaged (CRC mismatch)' % name)
|
|
files.append((off, name, blob))
|
|
return [(name, blob) for off, name, blob in sorted(files)]
|
|
|
|
|
|
def tank_pack(d, files):
|
|
"""A flat tank with the header of the original d and these files, stored uncompressed (as tools/tank.py pack)."""
|
|
data_offset = struct.unpack_from('<I', d, 0x18)[0]
|
|
data, entries, off = b'', [], 0
|
|
for name, blob in files:
|
|
entries.append((name, len(blob), off, zlib.crc32(blob) & 0xffffffff))
|
|
data += blob
|
|
off += len(blob)
|
|
head = bytearray(d[:data_offset])
|
|
body = data + b'\0' * 16
|
|
dirset_at = data_offset + len(body)
|
|
fe = []
|
|
for name, size, off, crc in sorted(entries, key=lambda e: e[0].lower()):
|
|
n = name.encode('latin1')
|
|
e = struct.pack('<4I', 0, size, off, crc) + b'\0' * 8 + struct.pack('<HHH', 0, 0, len(n)) + n + b'\0'
|
|
fe.append(e + b'\0' * (-len(e) % 4))
|
|
offs, pos = [], 4 + 4 * len(fe)
|
|
for e in fe:
|
|
offs.append(pos)
|
|
pos += len(e)
|
|
fileset = struct.pack('<I', len(fe)) + b''.join(struct.pack('<I', o) for o in offs) + b''.join(fe)
|
|
dirlen = 8 + 20 + 4 * len(fe) + 16
|
|
root = (struct.pack('<2I', 0, len(fe)) + b'\0' * 12 + b''.join(struct.pack('<I', dirlen + o) for o in offs) +
|
|
b'\0' * 16)
|
|
index = struct.pack('<2I', 1, 8) + root + fileset
|
|
struct.pack_into('<4I', head, 0x0c, dirset_at, dirset_at + dirlen, len(index), data_offset)
|
|
struct.pack_into('<2I', head, 0x50, zlib.crc32(index) & 0xffffffff, zlib.crc32(data) & 0xffffffff)
|
|
return bytes(head) + body + index
|
|
|
|
|
|
# ---- the journal in party.ds2 -------------------------------------------------------------------------
|
|
def read_str(d, o):
|
|
n = struct.unpack_from('<I', d, o)[0]
|
|
if n > 4096 or o + 4 + n > len(d):
|
|
raise ValueError('bad string')
|
|
return d[o + 4:o + 4 + n], o + 4 + n
|
|
|
|
|
|
def put_str(s):
|
|
return struct.pack('<I', len(s)) + s
|
|
|
|
|
|
def parse_journal(d, o):
|
|
"""The journal at offset o: ([(name, [(key, value), ...]), ...], end offset). ValueError if it is not one."""
|
|
count = struct.unpack_from('<I', d, o)[0]
|
|
if not 1 <= count <= 3:
|
|
raise ValueError('count')
|
|
o += 4
|
|
sections, names = [], []
|
|
for _ in range(count):
|
|
name, o = read_str(d, o)
|
|
if name.rstrip(b'\0').decode('latin1') not in DIFFICULTIES:
|
|
raise ValueError('name')
|
|
n = struct.unpack_from('<I', d, o)[0]
|
|
if n > 10000:
|
|
raise ValueError('entries')
|
|
o += 4
|
|
entries = []
|
|
for _ in range(n):
|
|
k, o = read_str(d, o)
|
|
v, o = read_str(d, o)
|
|
entries.append((k, v))
|
|
sections.append((name, entries))
|
|
names.append(name.rstrip(b'\0'))
|
|
if names != sorted(names):
|
|
raise ValueError('order')
|
|
return sections, o
|
|
|
|
|
|
def find_journal(d):
|
|
"""Offset of the journal block: the last offset where a complete journal parses."""
|
|
found = None
|
|
for o in range(0, len(d) - 16):
|
|
if struct.unpack_from('<I', d, o)[0] not in (1, 2, 3):
|
|
continue
|
|
try:
|
|
sections, end = parse_journal(d, o)
|
|
except (ValueError, struct.error, UnicodeDecodeError):
|
|
continue
|
|
found = (o, sections, end)
|
|
return found
|
|
|
|
|
|
def unlock(party_ds2, world, to):
|
|
hit = find_journal(party_ds2)
|
|
if not hit:
|
|
raise SystemExit('no journal found in party.ds2 - not a Dungeon Siege II party file?')
|
|
start, sections, end = hit
|
|
want = [('mercenary', '%s_completed_mercenary' % world)]
|
|
if to == 'elite':
|
|
want.append(('veteran', '%s_completed_veteran' % world))
|
|
secs = {name.rstrip(b'\0').decode('latin1'): (name, list(entries)) for name, entries in sections}
|
|
changed = []
|
|
for difficulty, key in want:
|
|
name, entries = secs.setdefault(difficulty, (difficulty.encode(), []))
|
|
k = key.encode()
|
|
entries = [(ek, ev) for ek, ev in entries if ek.rstrip(b'\0') != k]
|
|
entries.append((k, b'1'))
|
|
entries.sort(key=lambda e: e[0].rstrip(b'\0'))
|
|
secs[difficulty] = (name, entries)
|
|
changed.append('%s: %s = 1' % (difficulty, key))
|
|
block = struct.pack('<I', len(secs))
|
|
for difficulty in sorted(secs):
|
|
name, entries = secs[difficulty]
|
|
block += put_str(name) + struct.pack('<I', len(entries))
|
|
for k, v in entries:
|
|
block += put_str(k) + put_str(v)
|
|
return party_ds2[:start] + block + party_ds2[end:], changed
|
|
|
|
|
|
def main():
|
|
a = sys.argv[1:]
|
|
if not a or a[0].startswith('-'):
|
|
raise SystemExit(__doc__)
|
|
path = a[0]
|
|
world = a[a.index('--world') + 1] if '--world' in a else 'ds2_world'
|
|
to = a[a.index('--to') + 1] if '--to' in a else 'elite'
|
|
raw = open(path, 'rb').read()
|
|
out, changed = [], []
|
|
for name, data in tank_files(raw):
|
|
if name.lower() == 'party.ds2':
|
|
data, changed = unlock(data, world, to)
|
|
out.append((name, data))
|
|
if not changed:
|
|
raise SystemExit('party.ds2 not found in %s' % path)
|
|
for c in changed:
|
|
print('set', c)
|
|
if '--dry-run' in a:
|
|
return
|
|
new = tank_pack(raw, out)
|
|
if [(n, b) for n, b in tank_files(new)] != out:
|
|
raise SystemExit('internal check failed: the rewritten file does not read back the same - nothing written')
|
|
# The save as it is now goes to .unlockbak (refreshed every run, so it is never older progress), the new one
|
|
# is written beside it, read back, and only then swapped in: an interrupted or failed write leaves the save as it was.
|
|
backup = path + '.unlockbak'
|
|
shutil.copy2(path, backup)
|
|
print('current save kept as', backup)
|
|
tmp = path + '.unlocktmp'
|
|
with open(tmp, 'wb') as f:
|
|
f.write(new)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
if open(tmp, 'rb').read() != new:
|
|
os.remove(tmp)
|
|
raise SystemExit('write check failed (disk full?) - save left unchanged')
|
|
os.replace(tmp, path)
|
|
print('written', path)
|
|
print('Done: start the game and pick the difficulty (single player: new game; multiplayer: Map Settings).')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
finally:
|
|
# Dragged onto the script (no command prompt): keep the window open to show the result.
|
|
if 'PROMPT' not in os.environ and sys.stdin and sys.stdin.isatty():
|
|
try:
|
|
input('Press Enter to close.')
|
|
except EOFError: # NUL counts as a tty on Windows
|
|
pass
|