#!/usr/bin/env python3 """Assemble extracted dialogue text into wait-delimited screen prompt units.""" from __future__ import annotations import argparse import html import json import re import unicodedata from collections import Counter from pathlib import Path from probe_exe_scene_tables import read_sections, va_to_offset ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" WAIT_OPCODE = "0x06" CONTEXT_OPCODE = "0x09" RENDER_OPCODE = "0x0b" SOURCE_OPCODE = "0x0d" MAX_RENDER_SCAN_SPAN = 0x700 MAX_CONTEXT_SCAN_SPAN = 0x180 HANGUL_RANGE = r"\u3131-\u318e\uac00-\ud7a3" JAPANESE_RANGE = r"\u3040-\u30ff" CJK_RANGE = r"\u4e00-\u9fff" SCRIPT_CORE_RANGE = HANGUL_RANGE + r"A-Za-z0-9" + JAPANESE_RANGE + CJK_RANGE SCRIPT_BODY_RANGE = ( SCRIPT_CORE_RANGE + r" !?.,:_+\-,~()/&·「」『』♥♡☆★○◎×←→↑↓" ) SCRIPT_TEXT_RE = re.compile( rf"[.!?「『♥♡☆★○◎←→↑↓]*[{SCRIPT_CORE_RANGE}]" rf"[{SCRIPT_BODY_RANGE}]*[.!?」』♥♡☆★○◎]*" ) PUNCT_TRANSLATION = str.maketrans({ "\u3000": " ", "，": ",", "．": ".", "！": "!", "？": "?", "…": "...", "∼": "~", "～": "~", "“": '"', "”": '"', "‘": "'", "’": "'", }) EXTRA_VISIBLE_CHARS = set("「」『』♥♡☆★○◎×←→↑↓·") COMMON_SPEAKER_NAMES = { "아타호", "스마슈", "린샹", "다리오스", "페톰", "해골", "치호", "론", "주작", "창룡", "현무", "백호", "사범", "주작권사", "창룡권사", "현무권사", "백호권 사범", "맹호권 사범", "암각권 총통", "술집 주인", "술집주인", "여주인", "간호사", "의사", "상인", "상점주인", "시험관", "대회주최자", } COMMON_CHOICE_TEXTS = { "거럼", "그래", "기록", "결과를 지켜본다", "간단한 코스", "갖고 간다", "그냥간다", "그냥둬", "그만둘래", "그만둔다", "내비둬", "넣는다", "노우!", "누른다", "느리게", "다시 도전한다", "다시 설명을 듣는다", "당연히 받는다", "대회주최자를 막는다", "도전한다", "때려 치운다", "말이라고 해", "문을 통과한다", "물론", "물론 받는다", "물론이지", "물론이죠", "물을 길어 간다", "받는다", "받아간다", "밑으로 내려간다", "산다", "상급", "소모품을 산다", "쉰다", "수면비약을 준다", "싫다니까", "싫어", "싫어", "싫은데", "쾌속", "아까워서 안준다", "아니, 그만둔다", "아니, 잠깐만요", "아니", "아니야", "아니요", "아니오", "아, 잠깐만요", "아무 것도 안한다", "아무것도 안한다", "알았어", "어허, 잠깐만", "연다", "열어본다", "올라 타본다", "올라탄다", "움직인다", "음", "장비를 산다", "장비품을 산다", "저장", "적당한 코스", "준비", "준비 OK", "좋아요", "중급", "진귀한 물품은?", "초급", "초난관 코스", "초특급", "최상급", "한 명씩 밟는다", "한사람씩 밟는다", "휴식", "확인해볼까", "에이, 그만두자", "네, 좋아요", "타지 않는다", "봐줘~잉", } CHOICE_HINTS = ( "그만", "기록", "도전", "결과", "설명", "준비", "잠깐", "통과", "누른다", "넣는다", "연다", "열어", "산다", "쉰", "휴식", "물론", "아니오", "싫", "길어", "내려간다", "올라", "초급", "중급", "상급", "쾌속", "느리게", ) def hex32(value: int) -> str: return f"0x{value:08x}" def int_hex(value: str | None, fallback: int = 0) -> int: if not value: return fallback return int(value, 16) def load_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def is_hangul(char: str) -> bool: return "\u3131" <= char <= "\u318e" or "\uac00" <= char <= "\ud7a3" def has_hangul(text: str) -> bool: return any(is_hangul(char) for char in text) def is_japanese(char: str) -> bool: return "\u3040" <= char <= "\u30ff" def is_cjk(char: str) -> bool: return "\u4e00" <= char <= "\u9fff" def is_meaningful_punctuation_line(text: str) -> bool: return bool(re.fullmatch(r"[.?!]{2,}", text.strip())) def looks_like_binary_noise(text: str) -> bool: stripped = text.strip() if not stripped or is_meaningful_punctuation_line(stripped): return False tokens = [token for token in re.split(r"\s+", stripped) if token] if not tokens: return False hangul_total = sum(1 for char in stripped if is_hangul(char)) ascii_alnum_total = sum(1 for char in stripped if char.isascii() and char.isalnum()) if hangul_total <= 1 and ascii_alnum_total >= 1: return True simple_noise_tokens = [] for token in tokens: core = token.strip(".!?\"'「」『』()[]{}<>") if not core: simple_noise_tokens.append(False) continue hangul_count = sum(1 for char in core if is_hangul(char)) has_other_script = any(is_japanese(char) or is_cjk(char) for char in core) ascii_or_digit = any(char.isascii() and char.isalnum() for char in core) simple_noise_tokens.append( hangul_count == 1 and not has_other_script and bool(re.fullmatch(rf"[A-Za-z0-9]*[{HANGUL_RANGE}][A-Za-z0-9]*", core)) and (ascii_or_digit or len(tokens) > 1) ) if len(tokens) == 1 and simple_noise_tokens[0]: return True if len(tokens) >= 2 and all(simple_noise_tokens): return True return False def is_plausible_script_text(text: str) -> bool: if looks_like_binary_noise(text): return False return has_hangul(text) or is_meaningful_punctuation_line(text) def section_for_va(sections: list[dict], va: int) -> dict | None: for section in sections: start = section["va"] end = start + section["raw_size"] if start <= va < end: return section return None def interval_offsets(sections: list[dict], start_va: int, end_va: int) -> tuple[int, int, int] | None: section = section_for_va(sections, start_va) or section_for_va(sections, max(start_va, end_va - 1)) if section is None: return None clamped_start = max(start_va, section["va"]) clamped_end = min(end_va, section["va"] + section["raw_size"]) if clamped_end <= clamped_start: return None start_off = va_to_offset(sections, clamped_start) end_off = va_to_offset(sections, clamped_end - 1) if start_off is None or end_off is None: return None return start_off, end_off + 1, clamped_start def normal_script_text(decoded: str) -> str: decoded = unicodedata.normalize("NFKC", decoded) decoded = decoded.translate(PUNCT_TRANSLATION) chars = [] for char in decoded: code = ord(char) if char.isspace(): chars.append(" ") elif ( 0x20 <= code <= 0x7E or is_hangul(char) or is_japanese(char) or is_cjk(char) or char in EXTRA_VISIBLE_CHARS ): chars.append(char) else: chars.append(" ") return "".join(chars) def clean_script_text(text: str) -> str: text = " ".join(text.split()) text = re.sub(r"^[0-9]{1,2} (?=[\u3131-\u318e\uac00-\ud7a3])", "", text) text = re.sub(r"^[0-9]{1,2}(?=[\u3131-\u318e\uac00-\ud7a3])", "", text) text = re.sub(r"\s+([,.?!])", r"\1", text) text = re.sub(r"([,.?!])(?=[\u3131-\u318e\uac00-\ud7a3A-Za-z0-9])", r"\1 ", text) text = text.strip(" -_") return text def line_indent_level(decoded: str, match_start: int = 0) -> int: count = 0 if match_start > 0: index = min(match_start, len(decoded)) - 1 while index >= 0 and decoded[index] in {"\u3000", " "}: count += 1 index -= 1 return min(3, count) for char in decoded: if char in {"\u3000", " "}: count += 1 continue break return min(3, count) def extract_script_text_lines(data: bytes, base_va: int) -> list[dict]: lines = [] current = bytearray() current_start = 0 for index, byte in enumerate(data + b"\0"): if byte in {0x00, 0x40}: if len(current) >= 2: raw_decoded = current.decode("cp949", "ignore") decoded = normal_script_text(raw_decoded) whole_text = clean_script_text(decoded) if is_meaningful_punctuation_line(whole_text): lines.append({ "vaHex": hex32(base_va + current_start), "text": whole_text, "hangulCount": 0, "indentLevel": line_indent_level(decoded), "leadingQuote": False, }) current.clear() current_start = index + 1 continue for match in SCRIPT_TEXT_RE.finditer(decoded): text = clean_script_text(match.group(0)) if len(text) < 2 or not is_plausible_script_text(text): continue indent_level = line_indent_level(decoded, match.start()) lines.append({ "vaHex": hex32(base_va + current_start), "text": text, "hangulCount": sum(1 for char in text if is_hangul(char)), "indentLevel": indent_level, "leadingQuote": text.startswith(("「", "『")), }) current.clear() current_start = index + 1 else: if not current: current_start = index current.append(byte) return lines def wait_events(data: bytes, base_va: int) -> list[int]: waits = [] for index in range(max(0, len(data) - 1)): if data[index] == 0x40 and data[index + 1] == 0x06: waits.append(base_va + index) return waits def choice_marker_events(data: bytes, base_va: int) -> list[int]: markers = [] for index in range(max(0, len(data) - 1)): if data[index] == 0x40 and data[index + 1] == 0x18: markers.append(base_va + index) return markers def render_rows(event_text_source_flow: dict) -> list[dict]: rows = [] seen = set() for row in event_text_source_flow.get("rows") or []: if row.get("opcodeHex") != RENDER_OPCODE or row.get("confidence") != "medium": continue snippets = row.get("nearbyTextSnippets") or [] snippet_text = " ".join(str(item.get("text") or "") for item in snippets) if not has_hangul(snippet_text): continue va = int_hex(row.get("vaHex")) if va in seen: continue seen.add(va) rows.append({ "va": va, "vaHex": row.get("vaHex"), "classification": row.get("classification") or "", "nearbyText": snippet_text, }) return sorted(rows, key=lambda row: row["va"]) def prompt_text(lines: list[str]) -> str: return "\n".join(line for line in lines if line) def display_line(row: dict) -> str: return ("\u3000" * display_indent_level(row)) + str(row.get("text") or "") def display_indent_level(row: dict) -> int: if row.get("kind") == "speaker": return 0 return int(row.get("indentLevel") or 0) def compact_text(text: str) -> str: return " ".join(str(text or "").replace("\u3000", " ").split()).strip() def is_name_like(text: str) -> bool: stripped = compact_text(text) if stripped in COMMON_SPEAKER_NAMES: return True if not stripped or len(stripped) > 14: return False if any(mark in stripped for mark in ".,!?()[]{}「」『』"): return False return has_hangul(stripped) and len(stripped.split()) <= 3 def is_item_quote_line(text: str) -> bool: stripped = compact_text(text) return bool(re.match(r"^「[^」]{1,24}」", stripped) or re.match(r"^『[^』]{1,24}』", stripped)) def is_variable_fragment_line(text: str) -> bool: stripped = compact_text(text) if not stripped: return False variable_suffixes = ( "를 발견했다", "를 손에 넣었다", "를 줏어버렸다", "를 얼른 줏었다", "를 입수했다", "를 챙겼다", "가 들어있다", "가 사라져버렸다", "가 안됐다", "가 안됐다!", "은 흔적도 없이 사라졌다", ) if any(stripped.startswith(suffix) for suffix in variable_suffixes): return True return bool(re.match(r"^(를|을|가|은|는)\s+(발견|손에|줏|입수|챙겼|들어|사라|안됐)", stripped)) def is_exact_choice_text(text: str) -> bool: stripped = compact_text(text).strip("()") return stripped in COMMON_CHOICE_TEXTS def is_choice_candidate(row: dict) -> bool: text = compact_text(row.get("text") or "") if not text or row.get("leadingQuote") or is_item_quote_line(text): return False if is_exact_choice_text(text): return True if compact_text(text) in COMMON_SPEAKER_NAMES: return False if int(row.get("indentLevel") or 0) <= 0: return False if len(text) > 18: return False if text.endswith(("!", "...", ",", ".")): return False return any(hint in text for hint in CHOICE_HINTS) def classify_line_rows(rows: list[dict]) -> list[dict]: classified = [dict(row) for row in rows] for row in classified: text = compact_text(row.get("text") or "") if row.get("choiceBlock"): row["kind"] = "choice" elif is_variable_fragment_line(text): row["kind"] = "system-fragment" elif is_item_quote_line(text): row["kind"] = "system" else: row["kind"] = "text" for index, row in enumerate(classified): text = compact_text(row.get("text") or "") next_row = classified[index + 1] if index + 1 < len(classified) else None next_text = str(next_row.get("text") or "") if next_row else "" if ( next_row and is_name_like(text) and ( (next_text.startswith(("「", "『")) and not is_item_quote_line(next_text)) or next_text.startswith("...") ) ): row["kind"] = "speaker" for index, row in enumerate(classified): if row.get("kind") not in {"text", "speaker"}: continue text = str(row.get("text") or "") previous = classified[index - 1] if index > 0 else None if text.startswith(("「", "『")) and not is_item_quote_line(text): row["kind"] = "dialogue" elif previous and previous.get("kind") in {"speaker", "dialogue"}: if ( (int(row.get("indentLevel") or 0) > 0 or compact_text(text).startswith("...")) and not is_exact_choice_text(text) ): row["kind"] = "dialogue" index = 0 while index < len(classified): if classified[index].get("kind") not in {"text", "system"}: index += 1 continue if not is_choice_candidate(classified[index]): index += 1 continue start = index while ( index < len(classified) and classified[index].get("kind") in {"text", "system"} and is_choice_candidate(classified[index]) ): index += 1 run = classified[start:index] followed_by_speech = index < len(classified) and classified[index].get("kind") in { "speaker", "dialogue", } previous = classified[start - 1] if start > 0 else None preceded_by_prompt = ( start == 0 or ( previous is not None and previous.get("kind") in {"text", "system"} and ( compact_text(previous.get("text") or "").endswith("?") or "선택" in compact_text(previous.get("text") or "") ) ) ) has_exact_choice = any(is_exact_choice_text(row.get("text") or "") for row in run) should_mark = has_exact_choice or (preceded_by_prompt and (len(run) >= 2 or followed_by_speech)) if should_mark: for row in run: row["kind"] = "choice" first_speech_index = next( ( index for index, row in enumerate(classified) if row.get("kind") in {"speaker", "dialogue"} ), None, ) if first_speech_index and first_speech_index >= 2: leading_rows = classified[:first_speech_index] if all( row.get("kind") == "text" and int(row.get("indentLevel") or 0) > 0 and not row.get("leadingQuote") for row in leading_rows ): for row in leading_rows: row["kind"] = "choice" for index, row in enumerate(classified): if row.get("kind") == "speaker": continue previous = classified[index - 1] if index > 0 else None if previous and previous.get("kind") == "speaker": text = str(row.get("text") or "") if text.startswith(("「", "『")) and not is_item_quote_line(text): row["kind"] = "dialogue" return classified def plain_line_rows(lines: list[str], line_vas: list[int] | None = None) -> list[dict]: line_vas = line_vas or [] return classify_line_rows([ { "text": line, "indentLevel": 0, "leadingQuote": str(line).startswith(("「", "『")), "vaHex": hex32(line_vas[index]) if index < len(line_vas) else "", } for index, line in enumerate(lines) ]) def prompt_status(current: dict, wait_va: int | None, block: dict) -> str: if wait_va is None: return "trailing-open-review" first_render = min( (int_hex(item) for item in block.get("renderCommandVas") or []), default=None, ) if first_render is not None and current["startVa"] < first_render: return "leading-window-fragment-review" return "wait-delimited-review" def classify_prompt_lines(lines: list[str], block: dict | None = None) -> str: if block and block.get("classification"): return str(block["classification"]) joined = "\n".join(lines) system_markers = { "처음부터", "이어서", "이어서하기", "시나리오", "시나리오 선택", "데이터 저장", "데이터 로드", "도구", "무기", "방어구", "환경설정", } if sum(1 for marker in system_markers if marker in joined) >= 2: return "system-text-like" return "render-dialogue-like" def block_for_va(blocks: list[dict], va: int) -> dict | None: for block in blocks: if int(block.get("startVa") or 0) <= va < int(block.get("endVa") or 0): return block return None def block_events(block: dict) -> list[dict]: events: list[dict] = [] for command in (block.get("vmTrace") or {}).get("commands") or []: opcode = command.get("opcodeHex") if opcode not in {RENDER_OPCODE, SOURCE_OPCODE}: continue events.append({ "va": int_hex(command.get("vaHex")), "kind": "render" if opcode == RENDER_OPCODE else "source", "row": command, }) for event in (block.get("vmTrace") or {}).get("groundedControlEvents") or []: if event.get("opcodeHex") != WAIT_OPCODE: continue events.append({ "va": int_hex(event.get("vaHex")), "kind": "wait", "row": event, }) for row in block.get("textLines") or []: text = str(row.get("text") or "").strip() if not text: continue events.append({ "va": int_hex(row.get("vaHex")), "kind": "text", "row": row, }) order = {"source": 0, "render": 1, "text": 2, "wait": 3} return sorted(events, key=lambda event: (event["va"], order.get(event["kind"], 9))) def finalize_prompt( summary: dict, block: dict, current: dict | None, wait_va: int | None, status_reason: str = "", ) -> dict | None: if not current or not current.get("lines"): return None global_index = len(summary["prompts"]) + 1 prompt_index = current["promptIndex"] lines = current["lines"] start_va = current["startVa"] end_va = wait_va if wait_va is not None else current["lineVas"][-1] status = prompt_status(current, wait_va, block) if status_reason: status = status_reason prompt = { "id": f"story-prompt-{global_index:04d}", "globalIndex": global_index, "blockId": block.get("blockId") or "", "blockIndex": block.get("index"), "promptIndexInBlock": prompt_index, "classification": block.get("classification") or "", "status": status, "startVa": start_va, "startVaHex": hex32(start_va), "endVa": end_va, "endVaHex": hex32(end_va), "waitVaHex": hex32(wait_va) if wait_va is not None else "", "renderVaHex": hex32(current["renderVa"]) if current.get("renderVa") is not None else "", "sourceVaHex": hex32(current["sourceVa"]) if current.get("sourceVa") is not None else "", "textSourceValueHex": current.get("textSourceValueHex") or "", "lineCount": len(lines), "lines": lines, "lineRows": plain_line_rows(lines, current["lineVas"]), "lineVas": [hex32(value) for value in current["lineVas"]], "text": prompt_text(lines), "displayText": prompt_text([ display_line(row) for row in plain_line_rows(lines, current["lineVas"]) ]), "resourceNames": block.get("resourceNames") or [], "fieldMaps": block.get("fieldMaps") or [], "tilesets": block.get("tilesets") or [], "routeContexts": block.get("routeContexts") or [], "source": "out/event_dialogue_blocks.json", "screenPromptBoundary": "wait-for-input-release" if wait_va is not None else "open-ended-window", "originalRouteLinkedEventExecution": bool(block.get("routeContexts")), "originalStoryOrderBound": False, } summary["prompts"].append(prompt) return prompt def assemble_block_prompts(summary: dict, block: dict) -> None: current: dict | None = None prompt_index = 0 last_render_va: int | None = None last_source_va: int | None = None last_source_value = "" empty_waits = 0 for event in block_events(block): kind = event["kind"] va = event["va"] row = event["row"] if kind == "source": if current and current.get("lines"): finalize_prompt(summary, block, current, va, "source-delimited-review") current = None last_source_va = va last_source_value = str(row.get("textSourceValueHex") or "") continue if kind == "render": if current and current.get("lines"): finalize_prompt(summary, block, current, va, "render-delimited-review") current = None last_render_va = va continue if kind == "text": if current is None: prompt_index += 1 current = { "promptIndex": prompt_index, "startVa": va, "renderVa": last_render_va, "sourceVa": last_source_va, "textSourceValueHex": last_source_value, "lineVas": [], "lines": [], } current["lineVas"].append(va) current["lines"].append(str(row.get("text") or "").strip()) continue if kind == "wait": if current and current.get("lines"): finalize_prompt(summary, block, current, va) current = None else: empty_waits += 1 if current and current.get("lines"): finalize_prompt(summary, block, current, None) if empty_waits: summary["emptyWaitEventCount"] += empty_waits def finalize_render_prompt( summary: dict, blocks: list[dict], render_va: int, current: dict | None, wait_va: int | None, scan_interval: tuple[int, int], status: str = "render-scan-wait-delimited", boundary: str = "wait-for-input-release", ) -> None: if not current or not current.get("lines"): return if wait_va is None: summary["openRenderScanSkippedCount"] += 1 return key = (tuple(current["lineVas"]), tuple(current["lines"])) if key in summary["_seenPromptKeys"]: return summary["_seenPromptKeys"].add(key) block = block_for_va(blocks, current["startVa"]) or block_for_va(blocks, render_va) global_index = len(summary["prompts"]) + 1 block_id = block.get("blockId") if block else f"render-stream-{render_va:08x}" end_va = wait_va if wait_va is not None else current["lineVas"][-1] lines = current["lines"] line_rows = classify_line_rows(current.get("lineRows") or plain_line_rows(lines, current["lineVas"])) prompt = { "id": f"story-prompt-{global_index:04d}", "globalIndex": global_index, "blockId": block_id, "blockIndex": block.get("index") if block else None, "promptIndexInBlock": None, "classification": classify_prompt_lines(lines, block), "status": status, "startVa": current["startVa"], "startVaHex": hex32(current["startVa"]), "endVa": end_va, "endVaHex": hex32(end_va), "waitVaHex": hex32(wait_va), "renderVaHex": hex32(render_va), "sourceVaHex": "", "textSourceValueHex": "", "lineCount": len(lines), "lines": lines, "lineRows": line_rows, "lineVas": [hex32(value) for value in current["lineVas"]], "text": prompt_text(lines), "displayText": prompt_text([display_line(row) for row in line_rows]), "resourceNames": block.get("resourceNames") if block else [], "fieldMaps": block.get("fieldMaps") if block else [], "tilesets": block.get("tilesets") if block else [], "routeContexts": block.get("routeContexts") if block else [], "source": "event_text_source_flow-render-scan", "screenPromptBoundary": boundary, "scanStartVaHex": hex32(scan_interval[0]), "scanEndVaHex": hex32(scan_interval[1]), "originalRouteLinkedEventExecution": bool(block and block.get("routeContexts")), "originalStoryOrderBound": False, } summary["prompts"].append(prompt) def finalize_context_prompt( summary: dict, blocks: list[dict], context_va: int, current: dict | None, wait_va: int | None, scan_interval: tuple[int, int], ) -> None: if not current or not current.get("lines") or wait_va is None: return key = (tuple(current["lineVas"]), tuple(current["lines"])) if key in summary["_seenPromptKeys"]: return summary["_seenPromptKeys"].add(key) block = block_for_va(blocks, current["startVa"]) or block_for_va(blocks, context_va) global_index = len(summary["prompts"]) + 1 block_id = block.get("blockId") if block else f"context-stream-{context_va:08x}" lines = current["lines"] line_rows = classify_line_rows(current.get("lineRows") or plain_line_rows(lines, current["lineVas"])) prompt = { "id": f"story-prompt-{global_index:04d}", "globalIndex": global_index, "blockId": block_id, "blockIndex": block.get("index") if block else None, "promptIndexInBlock": None, "classification": classify_prompt_lines(lines, block), "status": "context-scan-wait-delimited", "startVa": current["startVa"], "startVaHex": hex32(current["startVa"]), "endVa": wait_va, "endVaHex": hex32(wait_va), "waitVaHex": hex32(wait_va), "renderVaHex": "", "sourceVaHex": hex32(context_va), "textSourceValueHex": "", "lineCount": len(lines), "lines": lines, "lineRows": line_rows, "lineVas": [hex32(value) for value in current["lineVas"]], "text": prompt_text(lines), "displayText": prompt_text([display_line(row) for row in line_rows]), "resourceNames": block.get("resourceNames") if block else [], "fieldMaps": block.get("fieldMaps") if block else [], "tilesets": block.get("tilesets") if block else [], "routeContexts": block.get("routeContexts") if block else [], "source": "event_text_source_flow-context-scan", "screenPromptBoundary": "wait-for-input-release", "scanStartVaHex": hex32(scan_interval[0]), "scanEndVaHex": hex32(scan_interval[1]), "originalRouteLinkedEventExecution": bool(block and block.get("routeContexts")), "originalStoryOrderBound": False, } summary["prompts"].append(prompt) summary["contextScanPromptCount"] = summary.get("contextScanPromptCount", 0) + 1 def assemble_render_scan_prompts( summary: dict, blocks: list[dict], exe: bytes, sections: list[dict], renders: list[dict], ) -> None: for index, row in enumerate(renders): render_va = row["va"] next_render_va = renders[index + 1]["va"] if index + 1 < len(renders) else render_va + MAX_RENDER_SCAN_SPAN end_va = min(next_render_va, render_va + MAX_RENDER_SCAN_SPAN) offsets = interval_offsets(sections, render_va, end_va) if offsets is None: continue start_off, end_off, base_va = offsets data = exe[start_off:end_off] text_rows = extract_script_text_lines(data, base_va) if not text_rows: continue events = [] for text_row in text_rows: events.append((int_hex(text_row.get("vaHex")), "text", text_row)) for wait_va in wait_events(data, base_va): events.append((wait_va, "wait", "")) for marker_va in choice_marker_events(data, base_va): events.append((marker_va, "choice-marker", "")) events.sort(key=lambda item: (item[0], 0 if item[1] == "text" else 1)) current = None for event_va, kind, value in events: if kind == "text": if current is None: current = { "startVa": event_va, "lineVas": [], "lines": [], "lineRows": [], } current["lineVas"].append(event_va) text = str(value.get("text") or "") current["lines"].append(text) current["lineRows"].append({ "text": text, "indentLevel": int(value.get("indentLevel") or 0), "leadingQuote": bool(value.get("leadingQuote")), "vaHex": value.get("vaHex") or hex32(event_va), }) elif kind == "choice-marker": if current and current.get("lineRows"): for line_row in current["lineRows"]: if ( int(line_row.get("indentLevel") or 0) > 0 and not line_row.get("leadingQuote") ): line_row["choiceBlock"] = True finalize_render_prompt( summary, blocks, render_va, current, event_va, (base_va, base_va + len(data)), "choice-marker-delimited", "choice-selection-marker", ) current = None elif kind == "wait": finalize_render_prompt(summary, blocks, render_va, current, event_va, (base_va, base_va + len(data))) current = None finalize_render_prompt(summary, blocks, render_va, current, None, (base_va, base_va + len(data))) def context_rows(exe: bytes, sections: list[dict]) -> list[dict]: rows: list[dict] = [] seen = set() for section in sections: if section["name"] not in {".data", ".rdata"}: continue start = section["raw"] end = section["raw"] + section["raw_size"] data = exe[start:end] search = 0 while True: hit = data.find(b"\x40\x09\x00\x00", search) if hit < 0: break search = hit + 1 va = section["va"] + hit if va in seen: continue seen.add(va) rows.append({"va": va}) return sorted(rows, key=lambda row: row["va"]) def assemble_context_scan_prompts( summary: dict, blocks: list[dict], exe: bytes, sections: list[dict], ) -> None: rows = context_rows(exe, sections) summary["contextScanRowCount"] = len(rows) for row in rows: context_va = row["va"] section = section_for_va(sections, context_va) if section is None: continue scan_start_va = context_va + 8 scan_end_va = min(context_va + MAX_CONTEXT_SCAN_SPAN, section["va"] + section["raw_size"]) offsets = interval_offsets(sections, scan_start_va, scan_end_va) if offsets is None: continue start_off, end_off, base_va = offsets data = exe[start_off:end_off] waits = wait_events(data, base_va) if not waits: continue wait_va = waits[0] wait_offset = wait_va - base_va if b"\x40\x09\x00\x00" in data[:wait_offset]: continue text_rows = extract_script_text_lines(data[:wait_offset], base_va) if not text_rows: continue current = { "startVa": int_hex(text_rows[0].get("vaHex")), "lineVas": [], "lines": [], "lineRows": [], } for text_row in text_rows: event_va = int_hex(text_row.get("vaHex")) text = str(text_row.get("text") or "") current["lineVas"].append(event_va) current["lines"].append(text) current["lineRows"].append({ "text": text, "indentLevel": int(text_row.get("indentLevel") or 0), "leadingQuote": bool(text_row.get("leadingQuote")), "vaHex": text_row.get("vaHex") or hex32(event_va), }) finalize_context_prompt(summary, blocks, context_va, current, wait_va, (base_va, base_va + len(data))) def new_summary_base(blocks_summary: dict, scope: str, boundary_rule: str) -> dict: return { "scope": scope, "source": "out/event_dialogue_blocks.json", "boundaryRule": boundary_rule, "blockCount": blocks_summary.get("blockCount", 0), "sourceTextLineCount": blocks_summary.get("textLineCount", 0), "sourceUniqueTextLineCount": blocks_summary.get("uniqueTextLineCount", 0), "routeLinkedBlockCount": blocks_summary.get("routeLinkedBlockCount", 0), "prompts": [], "emptyWaitEventCount": 0, "openRenderScanSkippedCount": 0, "contextScanRowCount": 0, "contextScanPromptCount": 0, "_seenPromptKeys": set(), } def finish_summary(summary: dict) -> dict: summary.pop("_seenPromptKeys", None) status_counts = Counter(prompt["status"] for prompt in summary["prompts"]) class_counts = Counter(prompt["classification"] for prompt in summary["prompts"]) line_counts = Counter(str(prompt["lineCount"]) for prompt in summary["prompts"]) summary.update({ "promptCount": len(summary["prompts"]), "promptLineCount": sum(prompt["lineCount"] for prompt in summary["prompts"]), "uniquePromptLineCount": len({line for prompt in summary["prompts"] for line in prompt["lines"]}), "statusCounts": dict(sorted(status_counts.items())), "classificationCounts": dict(sorted(class_counts.items())), "lineCountDistribution": dict(sorted(line_counts.items(), key=lambda item: int(item[0]))), "dialogueLikePromptCount": sum( count for key, count in class_counts.items() if key in { "story-dialogue-like", "map-dialogue-like", "battle-dialogue-like", "render-dialogue-like", } ), "systemPromptCount": class_counts.get("system-text-like", 0), "choiceLineCount": sum( 1 for prompt in summary["prompts"] for row in prompt.get("lineRows", []) if row.get("kind") == "choice" ), "systemFragmentLineCount": sum( 1 for prompt in summary["prompts"] for row in prompt.get("lineRows", []) if row.get("kind") == "system-fragment" ), "trailingOpenPromptCount": sum( count for key, count in status_counts.items() if "open" in key ), "openRenderScanSkippedCount": summary.get("openRenderScanSkippedCount", 0), "contextScanRowCount": summary.get("contextScanRowCount", 0), "contextScanPromptCount": summary.get("contextScanPromptCount", 0), "leadingFragmentPromptCount": status_counts.get("leading-window-fragment-review", 0), "originalStoryOrderBound": False, "originalRouteLinkedEventExecution": summary.get("routeLinkedBlockCount", 0) > 0, }) return summary def build_block_summary(blocks_summary: dict) -> dict: summary = { "scope": "Wait-delimited screen prompt units assembled from extracted event dialogue blocks.", "source": "out/event_dialogue_blocks.json", "boundaryRule": ( "Text payloads are grouped until opcode 0x06 wait-for-input-release events. " "This reconstructs reviewable screen prompt units, but does not prove story route order." ), "blockCount": blocks_summary.get("blockCount", 0), "sourceTextLineCount": blocks_summary.get("textLineCount", 0), "sourceUniqueTextLineCount": blocks_summary.get("uniqueTextLineCount", 0), "routeLinkedBlockCount": blocks_summary.get("routeLinkedBlockCount", 0), "prompts": [], "emptyWaitEventCount": 0, "_seenPromptKeys": set(), } for block in blocks_summary.get("blocks") or []: assemble_block_prompts(summary, block) return finish_summary(summary) def build_render_scan_summary( blocks_summary: dict, event_text_source_flow: dict, exe: bytes, ) -> dict: summary = new_summary_base( blocks_summary, "Screen prompt units assembled from every medium-confidence opcode 0x0b text render row.", ( "Text payloads after each opcode 0x0b render row are grouped until opcode 0x06 " "wait-for-input-release events. This is broader than the old 0x0d-window dialogue block index " "and is intended as a searchable extraction workbook, not proven story route order." ), ) summary["source"] = "out/event_text_source_flow.json + Hwanse2.exe render scan" summary["renderScan"] = { "maxSpan": MAX_RENDER_SCAN_SPAN, "maxSpanHex": hex(MAX_RENDER_SCAN_SPAN), } renders = render_rows(event_text_source_flow) summary["renderScan"]["renderRowCount"] = len(renders) assemble_render_scan_prompts( summary, blocks_summary.get("blocks") or [], exe, read_sections(exe), renders, ) assemble_context_scan_prompts( summary, blocks_summary.get("blocks") or [], exe, read_sections(exe), ) return finish_summary(summary) def build_summary( blocks_summary: dict, event_text_source_flow: dict | None = None, exe: bytes | None = None, ) -> dict: if event_text_source_flow is not None and exe is not None: return build_render_scan_summary(blocks_summary, event_text_source_flow, exe) return build_block_summary(blocks_summary) def escape_md(text: str) -> str: return text.replace("|", "\\|") def markdown(summary: dict) -> str: lines = [ "# Story Prompts", "", summary["scope"], "", summary["boundaryRule"], "", f"- prompts: {summary['promptCount']}", f"- prompt lines: {summary['promptLineCount']}", f"- dialogue-like prompts: {summary['dialogueLikePromptCount']}", f"- system prompts: {summary['systemPromptCount']}", f"- choice lines: {summary.get('choiceLineCount', 0)}", f"- variable/system fragments: {summary.get('systemFragmentLineCount', 0)}", f"- skipped open render fragments: {summary.get('openRenderScanSkippedCount', 0)}", f"- trailing/open prompts: {summary['trailingOpenPromptCount']}", f"- leading fragments: {summary['leadingFragmentPromptCount']}", f"- status: {json.dumps(summary['statusCounts'], ensure_ascii=False)}", f"- classes: {json.dumps(summary['classificationCounts'], ensure_ascii=False)}", "", "| prompt | block | class | status | render | wait | lines | text |", "| --- | --- | --- | --- | --- | --- | ---: | --- |", ] for prompt in summary["prompts"]: compact_parts = [] for row in prompt.get("lineRows") or plain_line_rows(prompt["lines"]): prefix = "" if row.get("kind") == "choice": prefix = "[선택] " elif row.get("kind") == "system-fragment": prefix = "[변수조각] " compact_parts.append(prefix + display_line(row)) compact = " / ".join(compact_parts) lines.append( f"| `{prompt['id']}` | `{prompt['blockId']}` | {prompt['classification']} | " f"{prompt['status']} | `{prompt['renderVaHex'] or '-'}` | `{prompt['waitVaHex'] or '-'}` | " f"{prompt['lineCount']} | {escape_md(compact)} |" ) return "\n".join(lines) + "\n" def prompt_card(prompt: dict) -> str: rows = prompt.get("lineRows") or plain_line_rows(prompt["lines"]) chunks: list[str] = [] choice_buffer: list[tuple[int, dict]] = [] def line_div(index: int, row: dict) -> str: indent = display_indent_level(row) kind = str(row.get("kind") or "text") classes = ["line", kind] if indent: classes.append("indented") return ( f"
" f"{html.escape(str(row.get('text') or ''))}
" ) def flush_choices() -> None: if not choice_buffer: return choices = "\n".join(line_div(index, row) for index, row in choice_buffer) chunks.append(f"
{choices}
") choice_buffer.clear() for index, row in enumerate(rows, start=1): if row.get("kind") == "choice": choice_buffer.append((index, row)) continue flush_choices() chunks.append(line_div(index, row)) flush_choices() lines = "\n".join(chunks) resources = ", ".join(prompt.get("fieldMaps") or prompt.get("tilesets") or prompt.get("resourceNames") or []) or "-" return ( f"
" "
" f"

{html.escape(prompt['id'])}

" f"{html.escape(prompt['blockId'])}" "
" f"
{html.escape(prompt['classification'])} · {html.escape(prompt['status'])} · " f"render {html.escape(prompt['renderVaHex'] or '-')} · wait {html.escape(prompt['waitVaHex'] or '-')}
" f"
{lines}
" f"
resources: {html.escape(resources)}
" "
" ) def html_page(summary: dict) -> str: prompts = "\n".join(prompt_card(prompt) for prompt in summary["prompts"]) return "\n".join([ "", '', "", ' ', ' ', " 환세취호전 대사 프롬프트", " ", "", "", "
", "

환세취호전 대사 프롬프트

", f"

{html.escape(summary['boundaryRule'])}

", "
", f"
{summary['promptCount']}prompts
", f"
{summary['promptLineCount']}lines
", f"
{summary['dialogueLikePromptCount']}dialogue-like
", f"
{summary['systemPromptCount']}system
", f"
{summary.get('openRenderScanSkippedCount', 0)}open skipped
", "
", "
", ' ', ' ', ' ', "
", f"
{prompts}
", "
", "", "", "", "", ]) def runtime_js(summary: dict) -> str: payload = { key: value for key, value in summary.items() if key != "prompts" } payload["prompts"] = [ { "id": prompt["id"], "blockId": prompt["blockId"], "classification": prompt["classification"], "status": prompt["status"], "renderVaHex": prompt["renderVaHex"], "waitVaHex": prompt["waitVaHex"], "lineCount": prompt["lineCount"], "lines": prompt["lines"], "lineRows": prompt.get("lineRows") or plain_line_rows(prompt["lines"]), "text": prompt["text"], "fieldMaps": prompt["fieldMaps"], "resourceNames": prompt["resourceNames"], } for prompt in summary["prompts"] ] return ( "window.HWANSE_STORY_PROMPTS = " + json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + ";\n" ) def write_outputs(summary: dict, out_dir: Path, runtime_js_out: Path | None = None) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "story_prompts.json").write_text( json.dumps(summary, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8", ) (out_dir / "story_prompts.html").write_text(html_page(summary), encoding="utf-8") if runtime_js_out is not None: runtime_js_out.parent.mkdir(parents=True, exist_ok=True) runtime_js_out.write_text(runtime_js(summary), encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--dialogue-blocks", type=Path, default=OUT / "event_dialogue_blocks.json") parser.add_argument("--event-text-source-flow", type=Path, default=OUT / "event_text_source_flow.json") parser.add_argument("--exe", type=Path, default=ROOT / "Hwanse2.exe") parser.add_argument( "--block-window-only", action="store_true", help="Use only out/event_dialogue_blocks.json instead of the broader render scan.", ) parser.add_argument("--out-dir", type=Path, default=OUT) parser.add_argument( "--runtime-js-out", type=Path, default=None, help=( "Optional legacy JS wrapper output. The active contract is " "story_prompts.json/html; do not write this large duplicate by default." ), ) args = parser.parse_args() blocks_summary = load_json(args.dialogue_blocks) if args.block_window_only: summary = build_summary(blocks_summary) else: summary = build_summary( blocks_summary, load_json(args.event_text_source_flow), args.exe.read_bytes(), ) write_outputs(summary, args.out_dir, args.runtime_js_out) print( "wrote story prompts -> " f"{args.out_dir / 'story_prompts.html'} " f"({summary['promptCount']} prompts, {summary['promptLineCount']} lines)" ) if __name__ == "__main__": main()