983 lines
40 KiB
Python
983 lines
40 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ysbin_tool.py — Extract / re-pack YU-RIS YSTB (.ybn) dialogue
|
||
|
||
Usage (run from the game root folder):
|
||
python dev/ysbin_tool.py extract # data/ysbin/*.ybn → json/
|
||
python dev/ysbin_tool.py pack # json/*.json → data/ysbin/ (patched in-place)
|
||
|
||
Optional flags:
|
||
--ysbin-dir PATH (default: data/ysbin)
|
||
--json-dir PATH (default: json)
|
||
|
||
Workflow:
|
||
1. Run "extract" to produce one JSON file per story script.
|
||
2. Fill in the "tl" field for each entry (leave blank to keep original).
|
||
3. Run "pack" to write the translations back into the .ybn files.
|
||
"""
|
||
|
||
import argparse, binascii, json, os, re, struct, sys, zlib as _zlib
|
||
|
||
# ── constants ────────────────────────────────────────────────────────────────
|
||
|
||
YSTB_MAGIC = b'YSTB'
|
||
PUSH_STRING = 0x4D # expression bytecode: push string literal
|
||
WORD_OP = 106 # WORD command — dialogue / narration
|
||
SELECT_OP = 45 # SELECT command — choice menu
|
||
|
||
# Known story-script names keyed by yst index
|
||
STORY_NAMES = {
|
||
124: '10エンディング',
|
||
125: '11回想モード用',
|
||
126: '11使えそうなテキスト',
|
||
127: '11抜いたけどパーツ取りに使えそう',
|
||
128: '1日常編改',
|
||
129: '2恋人後',
|
||
130: '3寝取り男登場',
|
||
131: '4寝取られ編',
|
||
132: '5寝取られh',
|
||
133: '6寝取られh',
|
||
134: '7寝取られh',
|
||
135: '8完落ち編',
|
||
136: '9完落ち編２',
|
||
143: 'scenario_start',
|
||
144: 'start',
|
||
145: 'ＵＩ機能',
|
||
146: 'エピローグ',
|
||
147: 'サウンド機能',
|
||
148: 'ノベルサンプル',
|
||
149: 'フラグセット',
|
||
150: '演出機能',
|
||
151: '加筆',
|
||
152: '画像表示関連機能',
|
||
153: '回想シーン',
|
||
154: '開発補助機能',
|
||
155: '基本的な命令のサンプル',
|
||
156: '使える地の文',
|
||
157: '消す項目',
|
||
158: '汎用性高い指示',
|
||
159: '頻繁に使う',
|
||
}
|
||
|
||
_JP_RE = re.compile(r'[\u3040-\u30ff\u4e00-\u9fff\uff00-\uffef]')
|
||
# Matches "SpeakerName「speech text..." — speaker is 1–6 chars, no brackets/whitespace
|
||
_SPEAKER_RE = re.compile(r'^([^「」『』\s]{1,6})「(.+)$', re.DOTALL)
|
||
|
||
def _sanitize(text):
|
||
"""
|
||
If text contains a null byte, strip everything up to and including it
|
||
(likely a corrupt control prefix), then also strip any immediately following
|
||
half-width characters (e.g. half-width katakana like ﾁ) that are also
|
||
artifacts. Returns the cleaned string, or None if nothing meaningful remains.
|
||
"""
|
||
if '\x00' in text:
|
||
text = text[text.index('\x00') + 1:]
|
||
# Strip leading half-width katakana/latin artifacts (U+FF65–U+FF9F and plain ASCII)
|
||
text = text.lstrip(''.join(chr(c) for c in range(0xFF65, 0xFF9F + 1)))
|
||
text = text.lstrip()
|
||
return text if text else None
|
||
|
||
# ── speaker post-processing ───────────────────────────────────────────────────
|
||
|
||
def _post_process_speakers(entries):
|
||
"""
|
||
Add 'speaker' key to dialogue entries and remove redundant standalone-name entries.
|
||
|
||
Two patterns handled:
|
||
A) Combined "Name「speech」" → speaker="Name", src="「speech」", _raw="Name「speech」"
|
||
(_raw is stored so pack can find & replace the original binary string)
|
||
|
||
B) Standalone entry_n.src="Name" + entry_{n+1}.src="「speech」"
|
||
→ entry_{n+1} gains speaker="Name", entry_n is dropped and IDs are re-numbered.
|
||
|
||
Narration entries (『…』, plain prose) are left untouched.
|
||
"""
|
||
# Phase 1 — split combined Name+speech entries (dialogue only)
|
||
for e in entries.values():
|
||
if e['type'] != 'dialogue':
|
||
continue
|
||
m = _SPEAKER_RE.match(e['src'])
|
||
if m:
|
||
e['speaker'] = m.group(1)
|
||
e['_raw'] = e['src'] # keep original for pack lookup
|
||
e['src'] = '\u300c' + m.group(2) # '「' + rest
|
||
|
||
# Phase 2 — merge standalone-name entries into the following speech entry
|
||
ids = sorted(entries.keys(), key=lambda x: int(x))
|
||
to_remove = []
|
||
for i, eid in enumerate(ids[:-1]):
|
||
e = entries[eid]
|
||
if e['type'] != 'dialogue':
|
||
continue
|
||
src = e['src']
|
||
# Skip entries that already have a speaker or contain bracket characters
|
||
if 'speaker' in e or '「' in src or '」' in src or '『' in src or '』' in src:
|
||
continue
|
||
# Only short Japanese-only text qualifies as a speaker name
|
||
if not _JP_RE.search(src) or len(src) > 20:
|
||
continue
|
||
next_e = entries[ids[i + 1]]
|
||
if next_e['type'] == 'dialogue' and next_e['src'].startswith('「') and 'speaker' not in next_e:
|
||
next_e['speaker'] = src
|
||
to_remove.append(eid)
|
||
|
||
for eid in to_remove:
|
||
del entries[eid]
|
||
|
||
# Re-number sequentially, enforcing key order: type, speaker, src, _raw, tl
|
||
new_entries = {}
|
||
for new_i, eid in enumerate(sorted(entries.keys(), key=lambda x: int(x))):
|
||
e = entries[eid]
|
||
ordered = {'type': e['type']}
|
||
if 'speaker' in e:
|
||
ordered['speaker'] = e['speaker']
|
||
ordered['src'] = e['src']
|
||
if '_raw' in e:
|
||
ordered['_raw'] = e['_raw']
|
||
ordered['tl'] = e['tl']
|
||
new_entries[str(new_i)] = ordered
|
||
return new_entries
|
||
|
||
|
||
# ── YSTB binary helpers ───────────────────────────────────────────────────────
|
||
|
||
def _xor(data, key):
|
||
"""XOR data bytes with the 4-byte repeating key."""
|
||
result = bytearray(data)
|
||
n = len(key)
|
||
for i in range(len(result)):
|
||
result[i] ^= key[i % n]
|
||
return bytes(result)
|
||
|
||
|
||
def _read_ystb(raw):
|
||
"""
|
||
Decrypt and parse a YSTB binary.
|
||
|
||
YSTB layout (all sizes in bytes):
|
||
[0:4] magic 'YSTB'
|
||
[4:8] unknown
|
||
[8:12] inst_cnt – number of instructions
|
||
[12:16] code_size – size of instruction table
|
||
[16:20] arg_size – size of attribute-descriptor table (12 bytes each)
|
||
[20:24] res_size – size of values blob
|
||
[24:32] unknown
|
||
[32 ..] instruction table (XOR-encrypted)
|
||
attribute table (XOR-encrypted)
|
||
values blob (XOR-encrypted)
|
||
|
||
The 4-byte XOR key is stored at attrs_off+8 in the *encrypted* data.
|
||
That position is the 'offset' field of the first attr descriptor which,
|
||
after decryption, is always 0 — so encrypted_byte XOR 0 == key_byte.
|
||
|
||
Returns (key, inst_list, attr_descs, vals_raw, header_bytes) or None.
|
||
inst_list : list of (opcode, argc)
|
||
attr_descs : list of [id, type, len, offset] (mutable)
|
||
vals_raw : decrypted values blob (bytes)
|
||
header_bytes: first 32 raw bytes (not encrypted)
|
||
"""
|
||
if raw[:4] != YSTB_MAGIC:
|
||
return None
|
||
|
||
inst_cnt = struct.unpack_from('<I', raw, 8)[0]
|
||
code_size = struct.unpack_from('<I', raw, 12)[0]
|
||
arg_size = struct.unpack_from('<I', raw, 16)[0]
|
||
res_size = struct.unpack_from('<I', raw, 20)[0]
|
||
|
||
instr_off = 32
|
||
attrs_off = instr_off + code_size
|
||
vals_off = attrs_off + arg_size
|
||
|
||
if arg_size < 12:
|
||
return None
|
||
|
||
key = list(raw[attrs_off + 8 : attrs_off + 12])
|
||
instr_raw = _xor(raw[instr_off : instr_off + code_size], key)
|
||
attrs_raw = _xor(raw[attrs_off : attrs_off + arg_size], key)
|
||
vals_raw = _xor(raw[vals_off : vals_off + res_size], key)
|
||
|
||
inst_list = [(instr_raw[i * 4], instr_raw[i * 4 + 1]) for i in range(inst_cnt)]
|
||
|
||
total_attrs = arg_size // 12
|
||
attr_descs = []
|
||
for i in range(total_attrs):
|
||
o = i * 12
|
||
attr_descs.append([
|
||
struct.unpack_from('<H', attrs_raw, o)[0], # id
|
||
struct.unpack_from('<H', attrs_raw, o + 2)[0], # type
|
||
struct.unpack_from('<I', attrs_raw, o + 4)[0], # len
|
||
struct.unpack_from('<I', attrs_raw, o + 8)[0], # offset into vals
|
||
])
|
||
|
||
return key, inst_list, attr_descs, vals_raw, raw[:32]
|
||
|
||
|
||
def _strings_from_expr(bytecodes):
|
||
"""Decode all PUSH_STRING values from expression bytecodes (cp932)."""
|
||
out, i = [], 0
|
||
while i < len(bytecodes):
|
||
if bytecodes[i] == PUSH_STRING and i + 3 <= len(bytecodes):
|
||
slen = struct.unpack_from('<H', bytecodes, i + 1)[0]
|
||
sbytes = bytecodes[i + 3 : i + 3 + slen]
|
||
if len(sbytes) == slen:
|
||
try:
|
||
out.append(sbytes.decode('cp932'))
|
||
except Exception:
|
||
pass
|
||
i += 3 + slen
|
||
else:
|
||
i += 1
|
||
return out
|
||
|
||
|
||
def _replace_in_expr(bytecodes, old_cp932, new_cp932):
|
||
"""Replace the first matching PUSH_STRING value inside expression bytecodes."""
|
||
buf = bytearray(bytecodes)
|
||
i = 0
|
||
while i < len(buf):
|
||
if buf[i] == PUSH_STRING and i + 3 <= len(buf):
|
||
slen = struct.unpack_from('<H', bytes(buf), i + 1)[0]
|
||
sbytes = bytes(buf[i + 3 : i + 3 + slen])
|
||
if sbytes == old_cp932:
|
||
new_len = len(new_cp932)
|
||
return (bytes(buf[:i + 1])
|
||
+ struct.pack('<H', new_len)
|
||
+ new_cp932
|
||
+ bytes(buf[i + 3 + slen:]))
|
||
i += 3 + slen
|
||
else:
|
||
i += 1
|
||
return bytes(bytecodes)
|
||
|
||
|
||
# ── extract ──────────────────────────────────────────────────────────────────
|
||
|
||
def _extract_choices(strings):
|
||
"""
|
||
From the flat string list inside a SELECT expression, return only the
|
||
actual choice labels (sit between 'ES.SEL.SET' and the first empty slot).
|
||
"""
|
||
try:
|
||
set_idx = next(i for i, s in enumerate(strings) if 'ES.SEL.SET' in s)
|
||
except StopIteration:
|
||
return []
|
||
choices = []
|
||
for s in strings[set_idx + 1:]:
|
||
if s in ("''", '""', ''):
|
||
break
|
||
clean = s.strip('"\'')
|
||
if clean and _JP_RE.search(clean):
|
||
choices.append(clean)
|
||
return choices
|
||
|
||
|
||
def extract_dialogue(raw):
|
||
"""
|
||
Extract all WORD_OP dialogue and SELECT_OP choice lines that contain
|
||
Japanese text.
|
||
Returns dict {str(id): {'type': 'dialogue'|'choice', 'src': ..., 'tl': ''}}
|
||
"""
|
||
parsed = _read_ystb(raw)
|
||
if not parsed:
|
||
return {}
|
||
|
||
key, inst_list, attr_descs, vals_raw, _ = parsed
|
||
entries = {}
|
||
eid = 0
|
||
attr_idx = 0
|
||
|
||
for op, argc in inst_list:
|
||
args_data = []
|
||
for ai in range(argc):
|
||
if attr_idx >= len(attr_descs):
|
||
args_data.append(None)
|
||
break
|
||
_, a_type, a_len, a_off = attr_descs[attr_idx]
|
||
if a_len > 0 and a_off + a_len <= len(vals_raw):
|
||
args_data.append((a_type, vals_raw[a_off : a_off + a_len]))
|
||
else:
|
||
args_data.append(None)
|
||
attr_idx += 1
|
||
|
||
if op == WORD_OP and args_data:
|
||
item = args_data[0]
|
||
if item:
|
||
a_type, val = item
|
||
texts = []
|
||
if a_type == 0:
|
||
try:
|
||
texts = [val.decode('cp932')]
|
||
except Exception:
|
||
pass
|
||
elif a_type == 3:
|
||
texts = _strings_from_expr(val)
|
||
for text in texts:
|
||
text = _sanitize(text)
|
||
if text and _JP_RE.search(text):
|
||
entries[str(eid)] = {'type': 'dialogue', 'src': text, 'tl': ''}
|
||
eid += 1
|
||
|
||
elif op == SELECT_OP:
|
||
for item in args_data:
|
||
if not item:
|
||
continue
|
||
a_type, val = item
|
||
strings = _strings_from_expr(val)
|
||
if not any('ES.SEL.SET' in s for s in strings):
|
||
continue
|
||
for choice in _extract_choices(strings):
|
||
entries[str(eid)] = {'type': 'choice', 'src': choice, 'tl': ''}
|
||
eid += 1
|
||
|
||
return _post_process_speakers(entries)
|
||
|
||
|
||
# ── pack ─────────────────────────────────────────────────────────────────────
|
||
|
||
def patch_ystb(raw, translations):
|
||
"""
|
||
Patch translated text into a YSTB binary.
|
||
|
||
translations : {src_text: tl_text}
|
||
Returns patched bytes, or None if nothing changed (no matches found).
|
||
|
||
Strategy:
|
||
- Decrypt instr / attrs / vals sections.
|
||
- For each WORD_OP first-argument that has a translation, record the
|
||
new bytes in attr_replacements.
|
||
- Rebuild the vals blob sequentially (original order), substituting
|
||
replaced values; update attr_descs offsets/lengths accordingly.
|
||
- Re-encrypt and reassemble the file. The XOR key is preserved because
|
||
attr_descs[0].offset is still 0 after the rebuild.
|
||
"""
|
||
parsed = _read_ystb(raw)
|
||
if not parsed:
|
||
return None
|
||
|
||
key, inst_list, attr_descs, vals_raw, header = parsed
|
||
code_size = struct.unpack_from('<I', raw, 12)[0]
|
||
arg_size = struct.unpack_from('<I', raw, 16)[0]
|
||
res_size = struct.unpack_from('<I', raw, 20)[0]
|
||
|
||
instr_off = 32
|
||
instr_enc = raw[instr_off : instr_off + code_size] # instructions unchanged
|
||
|
||
# ── find which attr indices need new content ──────────────────────────────
|
||
attr_replacements = {} # attr_idx -> new raw bytes
|
||
attr_idx = 0
|
||
|
||
for op, argc in inst_list:
|
||
for ai in range(argc):
|
||
if attr_idx >= len(attr_descs):
|
||
break
|
||
_, a_type, a_len, a_off = attr_descs[attr_idx]
|
||
|
||
if a_len > 0 and a_off + a_len <= len(vals_raw):
|
||
val = bytes(vals_raw[a_off : a_off + a_len])
|
||
|
||
if op == WORD_OP and ai == 0:
|
||
if a_type == 0:
|
||
try:
|
||
text = val.decode('cp932')
|
||
except Exception:
|
||
text = None
|
||
if text and text in translations:
|
||
tl = translations[text]
|
||
try:
|
||
new_bytes = tl.encode('cp932')
|
||
except Exception:
|
||
new_bytes = tl.encode('utf-8')
|
||
attr_replacements[attr_idx] = new_bytes
|
||
|
||
elif a_type == 3:
|
||
new_val = val
|
||
for s in _strings_from_expr(val):
|
||
if s in translations:
|
||
tl = translations[s]
|
||
try:
|
||
new_enc = tl.encode('cp932')
|
||
except Exception:
|
||
new_enc = tl.encode('utf-8')
|
||
new_val = _replace_in_expr(new_val, s.encode('cp932'), new_enc)
|
||
if new_val != val:
|
||
attr_replacements[attr_idx] = new_val
|
||
|
||
elif op == SELECT_OP:
|
||
# Choice text lives in expression bytecodes for any arg
|
||
strings = _strings_from_expr(val)
|
||
if any('ES.SEL.SET' in s for s in strings):
|
||
new_val = val
|
||
for choice in _extract_choices(strings):
|
||
if choice in translations:
|
||
tl = translations[choice]
|
||
try:
|
||
new_enc = tl.encode('cp932')
|
||
except Exception:
|
||
new_enc = tl.encode('utf-8')
|
||
new_val = _replace_in_expr(
|
||
new_val,
|
||
choice.encode('cp932'),
|
||
new_enc,
|
||
)
|
||
if new_val != val:
|
||
attr_replacements[attr_idx] = new_val
|
||
|
||
attr_idx += 1
|
||
|
||
if not attr_replacements:
|
||
return None
|
||
|
||
# ── rebuild vals blob ─────────────────────────────────────────────────────
|
||
# Process attr_descs in index order; shared regions are deduplicated via
|
||
# seen_regions so that two descriptors pointing at the same original bytes
|
||
# still map to the same new offset.
|
||
new_vals = bytearray()
|
||
seen_regions = {} # (orig_off, orig_len) -> new_off
|
||
|
||
for i, (aid, a_type, a_len, a_off) in enumerate(attr_descs):
|
||
if a_len == 0:
|
||
attr_descs[i][3] = 0
|
||
continue
|
||
|
||
if i in attr_replacements:
|
||
new_off = len(new_vals)
|
||
new_vals += attr_replacements[i]
|
||
attr_descs[i][2] = len(attr_replacements[i])
|
||
attr_descs[i][3] = new_off
|
||
else:
|
||
region = (a_off, a_len)
|
||
if region in seen_regions:
|
||
attr_descs[i][3] = seen_regions[region]
|
||
else:
|
||
if a_off + a_len <= len(vals_raw):
|
||
new_off = len(new_vals)
|
||
new_vals += vals_raw[a_off : a_off + a_len]
|
||
attr_descs[i][3] = new_off
|
||
seen_regions[region] = new_off
|
||
else:
|
||
attr_descs[i][3] = 0
|
||
|
||
# attr_descs[0][3] is always 0 after this loop, so re-encrypting the attrs
|
||
# section preserves the key at attrs_off+8 of the output (key XOR 0 = key).
|
||
|
||
# ── re-encrypt ────────────────────────────────────────────────────────────
|
||
new_attrs_raw = bytearray()
|
||
for aid, a_type, a_len, a_off in attr_descs:
|
||
new_attrs_raw += struct.pack('<HHII', aid, a_type, a_len, a_off)
|
||
|
||
new_attrs_enc = _xor(bytes(new_attrs_raw), key)
|
||
new_vals_enc = _xor(bytes(new_vals), key)
|
||
|
||
# Update res_size in header
|
||
new_header = bytearray(header)
|
||
struct.pack_into('<I', new_header, 20, len(new_vals))
|
||
|
||
# Preserve any trailing bytes after the vals section (usually none)
|
||
attrs_off = instr_off + code_size
|
||
tail_off = attrs_off + arg_size + res_size
|
||
tail = raw[tail_off:]
|
||
|
||
return bytes(new_header) + instr_enc + new_attrs_enc + new_vals_enc + tail
|
||
|
||
|
||
# ── CLI commands ──────────────────────────────────────────────────────────────
|
||
|
||
def cmd_extract(args):
|
||
ysbin_dir = args.ysbin_dir
|
||
out_dir = args.json_dir
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
|
||
total_files = 0
|
||
total_lines = 0
|
||
|
||
for fname in sorted(os.listdir(ysbin_dir)):
|
||
if not fname.startswith('yst') or not fname.endswith('.ybn'):
|
||
continue
|
||
num_str = fname[3:-4]
|
||
if not num_str.isdigit():
|
||
continue
|
||
idx = int(num_str)
|
||
|
||
fpath = os.path.join(ysbin_dir, fname)
|
||
with open(fpath, 'rb') as f:
|
||
raw = f.read()
|
||
|
||
entries = extract_dialogue(raw)
|
||
if not entries:
|
||
continue
|
||
|
||
name = STORY_NAMES.get(idx, '')
|
||
out_name = f'{idx:03d}_{name}.json' if name else f'{idx:03d}.json'
|
||
out_path = os.path.join(out_dir, out_name)
|
||
|
||
with open(out_path, 'w', encoding='utf-8') as f:
|
||
json.dump(entries, f, ensure_ascii=False, indent=2)
|
||
|
||
label = name or fname
|
||
print(f' [{idx:03d}] {label}: {len(entries)} lines → {out_path}')
|
||
total_files += 1
|
||
total_lines += len(entries)
|
||
|
||
print(f'\nExtracted {total_lines} dialogue lines across {total_files} file(s).')
|
||
|
||
|
||
def cmd_pack(args):
|
||
json_dir = args.json_dir
|
||
ysbin_dir = args.ysbin_dir
|
||
patched = 0
|
||
skipped = 0
|
||
|
||
for fname in sorted(os.listdir(json_dir)):
|
||
if not fname.endswith('.json'):
|
||
continue
|
||
|
||
# Parse yst index from filename prefix (NNN_name.json or NNN.json)
|
||
m = re.match(r'^(\d+)', fname)
|
||
if not m:
|
||
print(f' SKIP {fname}: cannot parse index from filename')
|
||
skipped += 1
|
||
continue
|
||
idx = int(m.group(1))
|
||
|
||
json_path = os.path.join(json_dir, fname)
|
||
with open(json_path, 'r', encoding='utf-8') as f:
|
||
entries = json.load(f)
|
||
|
||
# Build {lookup_key: replacement_text} for every translated entry.
|
||
# - dialogue with _raw: lookup by original binary string, re-attach speaker prefix
|
||
# - dialogue without _raw: lookup by src directly
|
||
# - choice: lookup by src directly (bare choice text in expression bytecodes)
|
||
translations = {}
|
||
for e in entries.values():
|
||
if not e.get('src') or not e.get('tl') or e['tl'] == e['src']:
|
||
continue
|
||
tl = e['tl']
|
||
key = e.get('_raw', e['src'])
|
||
# Re-attach speaker for entries that were originally combined (Name「…」)
|
||
if '_raw' in e and 'speaker' in e:
|
||
tl = e['speaker'] + tl
|
||
translations[key] = tl
|
||
|
||
if not translations:
|
||
print(f' SKIP {fname}: no translations filled in')
|
||
skipped += 1
|
||
continue
|
||
|
||
ybn_path = os.path.join(ysbin_dir, f'yst{idx:05d}.ybn')
|
||
if not os.path.exists(ybn_path):
|
||
print(f' SKIP {fname}: {ybn_path} not found')
|
||
skipped += 1
|
||
continue
|
||
|
||
with open(ybn_path, 'rb') as f:
|
||
raw = f.read()
|
||
|
||
result = patch_ystb(raw, translations)
|
||
if result is None:
|
||
print(f' SKIP {fname}: no matching strings found in binary')
|
||
skipped += 1
|
||
continue
|
||
|
||
with open(ybn_path, 'wb') as f:
|
||
f.write(result)
|
||
|
||
print(f' [{idx:03d}] {fname}: {len(translations)} string(s) patched → {ybn_path}')
|
||
patched += 1
|
||
|
||
print(f'\nPatched {patched} file(s), skipped {skipped}.')
|
||
|
||
|
||
# ── YPF repack ──────────────────────────────────────────────────────────────────────────
|
||
|
||
# Byte-swap tables from GARbro ArcYPF.cs
|
||
_YPF_SWAP_00 = bytes([
|
||
0x03, 0x48, 0x06, 0x35, 0x0C, 0x10, 0x11, 0x19, 0x1C, 0x1E,
|
||
0x09, 0x0B, 0x0D, 0x13, 0x15, 0x1B, 0x20, 0x23, 0x26, 0x29, 0x2C, 0x2F, 0x2E, 0x32,
|
||
])
|
||
_YPF_SWAP_04 = bytes([
|
||
0x0C, 0x10, 0x11, 0x19, 0x1C, 0x1E,
|
||
0x09, 0x0B, 0x0D, 0x13, 0x15, 0x1B, 0x20, 0x23, 0x26, 0x29, 0x2C, 0x2F, 0x2E, 0x32,
|
||
])
|
||
_YPF_SWAP_10 = bytes([
|
||
0x09, 0x0B, 0x0D, 0x13, 0x15, 0x1B, 0x20, 0x23, 0x26, 0x29, 0x2C, 0x2F, 0x2E, 0x32,
|
||
])
|
||
|
||
|
||
def _ypf_swap_table(version):
|
||
if version < 0x100:
|
||
return _YPF_SWAP_04
|
||
if 0x12C <= version < 0x196:
|
||
return _YPF_SWAP_10
|
||
return _YPF_SWAP_00
|
||
|
||
|
||
def _ypf_swap_len(swap, value):
|
||
"""GARbro DecryptLength: swap paired values in the table."""
|
||
for i in range(0, len(swap) - 1, 2):
|
||
if swap[i] == value:
|
||
return swap[i + 1]
|
||
if swap[i + 1] == value:
|
||
return swap[i]
|
||
return value
|
||
|
||
|
||
def _ypf_guess_key(ypf_data):
|
||
"""
|
||
Replicate GARbro's guess-key: decode the true name_len for the first
|
||
directory entry using the swap table, then XOR the 4th-from-last
|
||
encrypted name byte with '.' (assumes every filename ends in .xxx).
|
||
"""
|
||
swap = _ypf_swap_table(struct.unpack_from('<I', ypf_data, 4)[0])
|
||
name_len = _ypf_swap_len(swap, ypf_data[0x24] ^ 0xFF)
|
||
if name_len < 4:
|
||
raise ValueError('First YPF entry name too short to guess key')
|
||
enc_name = ypf_data[0x25 : 0x25 + name_len]
|
||
return enc_name[name_len - 4] ^ ord('.')
|
||
|
||
|
||
def _ypf_patch(ypf_data, substitutions):
|
||
"""
|
||
Rebuild a YPF archive with substituted files.
|
||
|
||
substitutions : {basename_lower: uncompressed_bytes}
|
||
e.g. {'yst00128.ybn': b'...'}
|
||
|
||
Files in substitutions are compared against their original YPF content
|
||
before deciding to recompress; unchanged files keep their original blob
|
||
so the data section stays byte-equal for unmodified entries.
|
||
|
||
Offset strategy: delta-based — only entries after a size-changing
|
||
substitution change their stored offset. Inter-file gaps in the original
|
||
data section are preserved, so the round-trip is byte-identical when no
|
||
file content actually changes.
|
||
|
||
Checksum note: this engine stores 0x00000000 in every checksum field and
|
||
does not validate it, so we never touch that field.
|
||
"""
|
||
assert ypf_data[:4] == b'YPF\x00', 'Not a YPF archive'
|
||
version = struct.unpack_from('<I', ypf_data, 4)[0]
|
||
count = struct.unpack_from('<I', ypf_data, 8)[0]
|
||
data_start = struct.unpack_from('<I', ypf_data, 12)[0] # abs offset of data section
|
||
|
||
swap = _ypf_swap_table(version)
|
||
extra = 4 if version >= 0x1D9 else (8 if version == 0xDE else 0)
|
||
meta_size = 0x12 + extra # metadata bytes per entry after the name
|
||
key = _ypf_guess_key(ypf_data)
|
||
|
||
# ── parse directory ───────────────────────────────────────────────────────────────
|
||
entries = []
|
||
pos = 0x20
|
||
for _ in range(count):
|
||
entry_start = pos
|
||
name_len_enc = ypf_data[pos + 4]
|
||
name_len = _ypf_swap_len(swap, name_len_enc ^ 0xFF)
|
||
pos += 5
|
||
|
||
enc_name = ypf_data[pos : pos + name_len]
|
||
name = bytes(b ^ key for b in enc_name).decode('cp932', errors='replace')
|
||
pos += name_len
|
||
|
||
meta_rel = pos - entry_start # offset of metadata within dir_raw
|
||
file_type = ypf_data[pos]
|
||
is_packed = bool(ypf_data[pos + 1])
|
||
unpacked_size = struct.unpack_from('<I', ypf_data, pos + 2)[0]
|
||
packed_size = struct.unpack_from('<I', ypf_data, pos + 6)[0]
|
||
data_off = struct.unpack_from('<I', ypf_data, pos + 10)[0]
|
||
pos += meta_size
|
||
|
||
basename = name.replace('\\', '/').split('/')[-1].lower()
|
||
entries.append({
|
||
'basename': basename,
|
||
'file_type': file_type,
|
||
'is_packed': is_packed,
|
||
'unpacked_size': unpacked_size,
|
||
'packed_size': packed_size,
|
||
'data_off': data_off,
|
||
'dir_raw': bytearray(ypf_data[entry_start:pos]),
|
||
'meta_rel': meta_rel,
|
||
})
|
||
|
||
dir_end = pos
|
||
dir_gap = bytes(ypf_data[dir_end : data_start]) # gap between dir end and data (usually empty)
|
||
|
||
# ── determine which substitutions actually change content ────────────────────────
|
||
for e in entries:
|
||
e['new_blob'] = None # default: keep original blob
|
||
if e['basename'] not in substitutions:
|
||
continue
|
||
raw = substitutions[e['basename']]
|
||
# Skip re-compression if disk content is identical to what's in the YPF
|
||
try:
|
||
orig_blob = ypf_data[e['data_off'] : e['data_off'] + e['packed_size']]
|
||
orig_raw = _zlib.decompress(orig_blob) if e['is_packed'] else orig_blob
|
||
if orig_raw == raw:
|
||
continue # unchanged — leave new_blob = None
|
||
except Exception:
|
||
pass
|
||
# Content changed: recompress
|
||
blob = _zlib.compress(raw, 9) if e['is_packed'] else raw
|
||
e['new_blob'] = blob
|
||
e['new_unpacked'] = len(raw)
|
||
e['new_packed'] = len(blob)
|
||
|
||
# ── collapse duplicate data_off values into unique regions ────────────────────────
|
||
# YPF archives sometimes deduplicate file data — multiple directory entries share
|
||
# the same data_off. Writing such a region more than once would shift all
|
||
# subsequent offsets and corrupt the archive. We build a unique region map and
|
||
# write each data block exactly once.
|
||
unique_regions = {} # orig_data_off -> dict with orig_size / new_blob etc.
|
||
for e in entries:
|
||
if e['data_off'] not in unique_regions:
|
||
unique_regions[e['data_off']] = {
|
||
'orig_size': e['packed_size'],
|
||
'new_blob': e['new_blob'],
|
||
'new_unpacked': e.get('new_unpacked'),
|
||
'new_packed': e.get('new_packed'),
|
||
}
|
||
elif e['new_blob'] is not None and unique_regions[e['data_off']]['new_blob'] is None:
|
||
# A substitution targets this shared region — use the new blob
|
||
unique_regions[e['data_off']].update({
|
||
'new_blob': e['new_blob'],
|
||
'new_unpacked': e['new_unpacked'],
|
||
'new_packed': e['new_packed'],
|
||
})
|
||
|
||
sorted_regions = sorted(unique_regions.items(), key=lambda x: x[0])
|
||
|
||
# ── assign new absolute offsets (delta-based, gap-preserving) ────────────────────
|
||
orig_to_new = {}
|
||
cur = data_start
|
||
prev_orig_end = data_start
|
||
|
||
for orig_off, r in sorted_regions:
|
||
gap = orig_off - prev_orig_end # gap in original between prev region and this
|
||
cur += gap # reproduce that gap in the output
|
||
orig_to_new[orig_off] = cur
|
||
cur += r['new_packed'] if r['new_blob'] is not None else r['orig_size']
|
||
prev_orig_end = orig_off + r['orig_size']
|
||
|
||
# ── update every directory entry ──────────────────────────────────────────────────
|
||
for e in entries:
|
||
e['new_data_off'] = orig_to_new[e['data_off']]
|
||
# Propagate new sizes only to entries whose region was substituted
|
||
if unique_regions[e['data_off']]['new_blob'] is not None:
|
||
r = unique_regions[e['data_off']]
|
||
e['new_unpacked'] = r['new_unpacked']
|
||
e['new_packed'] = r['new_packed']
|
||
e['new_blob'] = r['new_blob']
|
||
else:
|
||
e['new_blob'] = None
|
||
|
||
# ── rebuild directory ─────────────────────────────────────────────────────────────
|
||
new_dir = bytearray()
|
||
for e in entries:
|
||
raw = e['dir_raw']
|
||
mo = e['meta_rel']
|
||
struct.pack_into('<I', raw, mo + 10, e['new_data_off'])
|
||
if e['new_blob'] is not None:
|
||
struct.pack_into('<I', raw, mo + 2, e['new_unpacked'])
|
||
struct.pack_into('<I', raw, mo + 6, e['new_packed'])
|
||
# checksum field (mo+14) left as-is — engine stores 0
|
||
new_dir += raw
|
||
|
||
# ── rebuild data section from unique regions, preserving original gaps ────────────
|
||
new_data = bytearray()
|
||
prev_orig_end = data_start
|
||
|
||
for orig_off, r in sorted_regions:
|
||
gap = orig_off - prev_orig_end
|
||
if gap > 0:
|
||
new_data += ypf_data[prev_orig_end : orig_off]
|
||
if r['new_blob'] is not None:
|
||
new_data += r['new_blob']
|
||
else:
|
||
new_data += ypf_data[orig_off : orig_off + r['orig_size']]
|
||
prev_orig_end = orig_off + r['orig_size']
|
||
|
||
# Trailing bytes beyond the last known region
|
||
if prev_orig_end < len(ypf_data):
|
||
new_data += ypf_data[prev_orig_end:]
|
||
|
||
# ── assemble ──────────────────────────────────────────────────────────────────────
|
||
hdr = bytearray(ypf_data[:0x20]) # version/count/data_start/padding unchanged
|
||
return bytes(hdr) + bytes(new_dir) + dir_gap + bytes(new_data)
|
||
|
||
|
||
def cmd_verify_ypf(args):
|
||
"""
|
||
Diagnostic: parse a YPF, show key structural info, and verify a round-trip
|
||
reconstruction produces an identical directory section.
|
||
"""
|
||
ypf_path = args.ypf
|
||
if not os.path.exists(ypf_path):
|
||
print(f'ERROR: YPF not found: {ypf_path}')
|
||
sys.exit(1)
|
||
|
||
print(f'Reading {ypf_path} ...')
|
||
with open(ypf_path, 'rb') as f:
|
||
ypf_data = f.read()
|
||
|
||
assert ypf_data[:4] == b'YPF\x00', 'Not a YPF archive'
|
||
version = struct.unpack_from('<I', ypf_data, 4)[0]
|
||
count = struct.unpack_from('<I', ypf_data, 8)[0]
|
||
data_start = struct.unpack_from('<I', ypf_data, 12)[0]
|
||
print(f'version={version:#x} count={count} data_start={data_start:#x} ({data_start})')
|
||
|
||
swap = _ypf_swap_table(version)
|
||
extra = 4 if version >= 0x1D9 else (8 if version == 0xDE else 0)
|
||
meta_size = 0x12 + extra
|
||
key = _ypf_guess_key(ypf_data)
|
||
print(f'key={key:#04x} extra_bytes={extra} meta_size={meta_size}')
|
||
|
||
# Parse directory
|
||
pos = 0x20
|
||
entries = []
|
||
for _ in range(count):
|
||
name_len_enc = ypf_data[pos + 4]
|
||
name_len = _ypf_swap_len(swap, name_len_enc ^ 0xFF)
|
||
pos += 5
|
||
enc_name = ypf_data[pos : pos + name_len]
|
||
name = bytes(b ^ key for b in enc_name).decode('cp932', errors='replace')
|
||
pos += name_len
|
||
is_packed = bool(ypf_data[pos + 1])
|
||
packed_size = struct.unpack_from('<I', ypf_data, pos + 6)[0]
|
||
data_off = struct.unpack_from('<I', ypf_data, pos + 10)[0]
|
||
checksum = struct.unpack_from('<I', ypf_data, pos + 14)[0]
|
||
pos += meta_size
|
||
entries.append({'name': name, 'data_off': data_off,
|
||
'packed_size': packed_size, 'checksum': checksum})
|
||
|
||
dir_end = pos
|
||
gap = data_start - dir_end
|
||
gap_status = 'OK (no gap)' if gap == 0 else f'WARNING: {gap} byte gap between dir end and data!'
|
||
print(f'dir_end={dir_end:#x} gap={gap} [{gap_status}]')
|
||
|
||
# First entry by data order — tells us if offsets are absolute
|
||
first = min(entries, key=lambda e: e['data_off'])
|
||
print(f'\nFirst entry by data offset: {first["name"]!r}')
|
||
print(f' data_off={first["data_off"]:#010x} data_start={data_start:#010x}')
|
||
diff = first['data_off'] - data_start
|
||
if diff == 0:
|
||
print(' → offsets are ABSOLUTE (data_off == data_start for first entry) ✓')
|
||
elif diff < 0:
|
||
print(f' → offsets appear RELATIVE (first entry starts {-diff} bytes before data_start?) ✗')
|
||
else:
|
||
print(f' → first entry is {diff} bytes into the data section')
|
||
|
||
# Checksum spot-checks
|
||
print('\nCheksum spot-checks (first 5 entries by offset):')
|
||
ok_count = err_count = 0
|
||
for e in sorted(entries, key=lambda x: x['data_off'])[:5]:
|
||
blob = ypf_data[e['data_off'] : e['data_off'] + e['packed_size']]
|
||
calc = _zlib.adler32(blob) & 0xFFFFFFFF
|
||
if calc == e['checksum']:
|
||
status = 'OK'
|
||
ok_count += 1
|
||
else:
|
||
status = f'FAIL (calc={calc:#010x} stored={e["checksum"]:#010x})'
|
||
err_count += 1
|
||
print(f' {e["name"]!r:44s} off={e["data_off"]:#010x} adler={status}')
|
||
|
||
# Round-trip directory check (empty substitutions)
|
||
rebuilt = _ypf_patch(ypf_data, {})
|
||
orig_dir = ypf_data[0x20 : data_start]
|
||
rt_dir = rebuilt[0x20 : data_start]
|
||
if orig_dir == rt_dir:
|
||
print('\nRound-trip directory bytes: IDENTICAL ✓')
|
||
else:
|
||
print(f'\nRound-trip directory bytes: DIFFER orig_len={len(orig_dir)} rt_len={len(rt_dir)}')
|
||
for i in range(min(len(orig_dir), len(rt_dir))):
|
||
if orig_dir[i] != rt_dir[i]:
|
||
print(f' First diff at dir+{i:#08x}: orig={orig_dir[i]:#04x} rebuilt={rt_dir[i]:#04x}')
|
||
break
|
||
if len(orig_dir) != len(rt_dir):
|
||
print(f' Length mismatch: orig={len(orig_dir)} rt={len(rt_dir)}')
|
||
|
||
# Check round-trip file size
|
||
if len(rebuilt) == len(ypf_data):
|
||
print(f'Round-trip file size: IDENTICAL ({len(ypf_data)} bytes) ✓')
|
||
else:
|
||
print(f'Round-trip file size: orig={len(ypf_data)} rebuilt={len(rebuilt)} delta={len(rebuilt)-len(ypf_data)}')
|
||
|
||
print(f'\nTotal entries: {count} | YPF size: {len(ypf_data)/1024/1024:.2f} MB')
|
||
|
||
|
||
def cmd_repack_ypf(args):
|
||
ypf_path = args.ypf
|
||
out_path = args.output
|
||
ysbin_dir = args.ysbin_dir
|
||
|
||
if not os.path.exists(ypf_path):
|
||
print(f'ERROR: YPF not found: {ypf_path}')
|
||
sys.exit(1)
|
||
|
||
print(f'Reading {ypf_path} ...')
|
||
with open(ypf_path, 'rb') as f:
|
||
ypf_data = f.read()
|
||
|
||
# Build substitutions from every yst*.ybn in ysbin_dir
|
||
substitutions = {}
|
||
for fname in os.listdir(ysbin_dir):
|
||
if fname.lower().startswith('yst') and fname.lower().endswith('.ybn'):
|
||
with open(os.path.join(ysbin_dir, fname), 'rb') as f:
|
||
substitutions[fname.lower()] = f.read()
|
||
print(f'Substituting {len(substitutions)} .ybn script file(s)...')
|
||
|
||
patched = _ypf_patch(ypf_data, substitutions)
|
||
|
||
with open(out_path, 'wb') as f:
|
||
f.write(patched)
|
||
orig_mb = len(ypf_data) / 1024 / 1024
|
||
new_mb = len(patched) / 1024 / 1024
|
||
print(f'Done. {orig_mb:.1f} MB → {new_mb:.1f} MB')
|
||
print(f'Written to {out_path}')
|
||
|
||
|
||
# ── entry point ───────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description='Extract / re-pack YU-RIS YSTB (.ybn) dialogue, and repack into .ypf.',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=__doc__,
|
||
)
|
||
sub = parser.add_subparsers(dest='command', required=True)
|
||
|
||
# extract
|
||
p_ext = sub.add_parser('extract', help='.ybn → JSON')
|
||
p_ext.add_argument('--ysbin-dir', default='data/ysbin')
|
||
p_ext.add_argument('--json-dir', default='json')
|
||
|
||
# pack
|
||
p_pack = sub.add_parser('pack', help='JSON → .ybn (patch in-place)')
|
||
p_pack.add_argument('--ysbin-dir', default='data/ysbin')
|
||
p_pack.add_argument('--json-dir', default='json')
|
||
|
||
# repack-ypf
|
||
p_ypf = sub.add_parser('repack-ypf', help='data/ysbin/*.ybn → patched .ypf')
|
||
p_ypf.add_argument('--ysbin-dir', default='data/ysbin',
|
||
help='Directory of (patched) yst*.ybn files')
|
||
p_ypf.add_argument('--ypf', default='ypf/data.ypf',
|
||
help='Path to original .ypf archive')
|
||
p_ypf.add_argument('--output', default='data.ypf',
|
||
help='Output path for patched .ypf')
|
||
|
||
# verify-ypf
|
||
p_ver = sub.add_parser('verify-ypf', help='Diagnostic: inspect YPF structure and verify round-trip')
|
||
p_ver.add_argument('--ypf', default='ypf/data.ypf',
|
||
help='Path to .ypf archive to verify')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == 'extract':
|
||
cmd_extract(args)
|
||
elif args.command == 'pack':
|
||
cmd_pack(args)
|
||
elif args.command == 'verify-ypf':
|
||
cmd_verify_ypf(args)
|
||
else:
|
||
cmd_repack_ypf(args)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|