fixes
This commit is contained in:
parent
b9ab810c26
commit
7a1109d8be
9 changed files with 760 additions and 100 deletions
|
|
@ -87,7 +87,7 @@ from gui.workflow_tab import (
|
|||
_make_text_btn,
|
||||
_make_section_label,
|
||||
)
|
||||
from util import wolf_names
|
||||
from util.wolfdawn import names as wolf_names
|
||||
from util.paths import PROJECT_ROOT
|
||||
from util.project_scanner import (
|
||||
detect_wolf_layout,
|
||||
|
|
@ -1696,8 +1696,10 @@ class WolfWorkflowTab(QWidget):
|
|||
layout.addWidget(self._desc(
|
||||
"Sends the extracted files in files/ to the Translation tab using the Wolf RPG "
|
||||
"(WolfDawn) module. Only the 'text' fields are filled in; 'source' is preserved so "
|
||||
"injection can verify each line. Run Step 3 (names) first so vocab.txt is seeded, "
|
||||
"then Phase 1 (database text) and Phase 2 (maps / events) in order."
|
||||
"injection can verify each line. Lines whose text is already translated (no Japanese) "
|
||||
"are skipped, so re-running a phase only sends unfinished lines. Run Step 3 (names) "
|
||||
"first so vocab.txt is seeded, then Phase 1 (database text) and Phase 2 "
|
||||
"(maps / events) in order."
|
||||
))
|
||||
|
||||
self._add_phase_buttons(layout)
|
||||
|
|
@ -2166,7 +2168,9 @@ class WolfWorkflowTab(QWidget):
|
|||
"byte-exact. Translated safe/refs name values from names.json are applied across Data/ "
|
||||
"(only the entries you translated change). Lines whose inline codes changed are skipped and "
|
||||
"reported (unless you allow drift). Full inject and any inject that includes "
|
||||
f"{NAMES_JSON} reset live Data/ from {WORK_DIR_NAME}/originals/ first."
|
||||
f"{NAMES_JSON} reset live Data/ from {WORK_DIR_NAME}/originals/ first. "
|
||||
f"Before each inject, inline \\\\codes are auto-repaired from source; on a full inject "
|
||||
f"every translated/ JSON is copied into {WORK_DIR_NAME}/ for the game-repo diff."
|
||||
))
|
||||
|
||||
self.en_punct_cb = QCheckBox("Convert Japanese punctuation to ASCII (--en-punct)")
|
||||
|
|
@ -2346,14 +2350,73 @@ class WolfWorkflowTab(QWidget):
|
|||
now_btn.clicked.connect(lambda: self._run_relayout(manual=True))
|
||||
layout.addWidget(now_btn)
|
||||
|
||||
def _tool_root(self) -> Path:
|
||||
"""DazedMTLTool project root (``files/``, ``translated/``, etc.)."""
|
||||
return Path(__file__).resolve().parent.parent
|
||||
|
||||
def _translated_path(self, json_name: str) -> Path | None:
|
||||
"""Absolute path to ``translated/<json_name>`` when it exists."""
|
||||
p = self._tool_root() / "translated" / json_name
|
||||
return p if p.is_file() else None
|
||||
|
||||
def _translated_or_source(self, json_name: str) -> Path | None:
|
||||
"""Prefer translated/<name>; fall back to files/<name>."""
|
||||
for base in ("translated", "files"):
|
||||
p = Path(base) / json_name
|
||||
root = self._tool_root()
|
||||
for sub in ("translated", "files"):
|
||||
p = root / sub / json_name
|
||||
if p.is_file():
|
||||
return p
|
||||
return None
|
||||
|
||||
def _repair_inject_json(self, src: Path, log) -> Path:
|
||||
"""Auto-repair WOLF inline codes in a translated JSON before inject."""
|
||||
from util.wolfdawn import codes as wolf_codes
|
||||
|
||||
data = json.loads(src.read_text(encoding="utf-8-sig"))
|
||||
data, repairs = wolf_codes.repair_document(data)
|
||||
if repairs:
|
||||
log(
|
||||
f" Auto-repaired {len(repairs)} line(s) with control-code drift in {src.name}"
|
||||
)
|
||||
src.write_text(
|
||||
json.dumps(data, ensure_ascii=False, indent=4) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return src
|
||||
|
||||
def _sync_translated_json_to_wolf_json(
|
||||
self, json_name: str, log, game_json_dir: Path
|
||||
) -> bool:
|
||||
"""Copy one translated/ JSON into the game's wolf_json/ folder."""
|
||||
src = self._translated_path(json_name)
|
||||
if src is None:
|
||||
return False
|
||||
dest = game_json_dir / json_name
|
||||
try:
|
||||
if src.resolve() == dest.resolve():
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(src, dest)
|
||||
return True
|
||||
except Exception as exc:
|
||||
log(f" ⚠ could not update {json_name} in {WORK_DIR_NAME}/: {exc}")
|
||||
return False
|
||||
|
||||
def _log_inject_guard_lines(self, mismatches: list[str], untranslated: int | None, log):
|
||||
"""Surface WolfDawn guard failures instead of burying them in the scrollback."""
|
||||
if untranslated:
|
||||
log(f" ⚠ {untranslated} line(s) left untranslated by WolfDawn safety guards")
|
||||
if not mismatches:
|
||||
return
|
||||
log(f" ⚠ {len(mismatches)} control-code mismatch(es) (first 10):")
|
||||
for line in mismatches[:10]:
|
||||
log(f" {line}")
|
||||
if len(mismatches) > 10:
|
||||
log(f" … and {len(mismatches) - 10} more (see log above)")
|
||||
|
||||
def _on_step_changed(self, idx: int):
|
||||
previous = self._current_step_index
|
||||
self._current_step_index = idx
|
||||
|
|
@ -2382,7 +2445,9 @@ class WolfWorkflowTab(QWidget):
|
|||
json_name = entry.get("json")
|
||||
if not json_name:
|
||||
continue
|
||||
if not (Path("translated") / json_name).is_file():
|
||||
if entry.get("kind") == "names":
|
||||
continue
|
||||
if not self._translated_path(json_name):
|
||||
continue
|
||||
item = QListWidgetItem(json_name)
|
||||
item.setData(Qt.UserRole, json_name)
|
||||
|
|
@ -2465,21 +2530,6 @@ class WolfWorkflowTab(QWidget):
|
|||
quick = only_json is not None
|
||||
game_json_dir = self._work_dir()
|
||||
|
||||
def _sync_json(json_name: str, src: Path, log):
|
||||
"""Copy the translated JSON into the game's git-tracked wolf_json/ so the
|
||||
diff shows both the updated JSON source and the injected binary."""
|
||||
dest = game_json_dir / json_name
|
||||
try:
|
||||
if src.resolve() == dest.resolve():
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(src, dest)
|
||||
except Exception as e:
|
||||
log(f" ⚠ could not update {json_name} in {WORK_DIR_NAME}/: {e}")
|
||||
|
||||
def task(log, progress=None):
|
||||
from util import wolfdawn
|
||||
|
||||
|
|
@ -2492,6 +2542,7 @@ class WolfWorkflowTab(QWidget):
|
|||
data_dir_path = Path(data_dir)
|
||||
applied = 0
|
||||
failed = 0
|
||||
guard_failures: list[str] = []
|
||||
|
||||
names_entry = next((e for e in entries if e["kind"] == "names"), None)
|
||||
names_src = (
|
||||
|
|
@ -2520,7 +2571,7 @@ class WolfWorkflowTab(QWidget):
|
|||
e["json"]
|
||||
for e in entries
|
||||
if e.get("kind") != "names"
|
||||
and (Path("translated") / e["json"]).is_file()
|
||||
and self._translated_path(e["json"]) is not None
|
||||
}
|
||||
strings_targets |= {j for j in only_json if j != NAMES_JSON}
|
||||
else:
|
||||
|
|
@ -2532,10 +2583,13 @@ class WolfWorkflowTab(QWidget):
|
|||
continue
|
||||
if entry["json"] not in strings_targets:
|
||||
continue
|
||||
src = self._translated_or_source(entry["json"])
|
||||
if src is None:
|
||||
inject_src = self._translated_path(entry["json"])
|
||||
if inject_src is None:
|
||||
inject_src = self._translated_or_source(entry["json"])
|
||||
if inject_src is None:
|
||||
log(f" ⚠ no JSON for {entry['json']} — skipped")
|
||||
continue
|
||||
inject_src = self._repair_inject_json(inject_src, log)
|
||||
out = entry["base"] # live game binary (or txt-dir) we write into
|
||||
orig = self._orig_base_for(entry, data_dir_path)
|
||||
base = str(orig) if orig.exists() else out
|
||||
|
|
@ -2546,24 +2600,32 @@ class WolfWorkflowTab(QWidget):
|
|||
)
|
||||
log(f"Injecting {entry['json']} → {Path(out).name} …")
|
||||
res = wolfdawn.strings_inject(
|
||||
str(src), base, out,
|
||||
str(inject_src), base, out,
|
||||
allow_code_drift=allow_drift, en_punct=en_punct, log_fn=log,
|
||||
)
|
||||
a, d = wolfdawn.parse_strings_inject_counts(res.stdout)
|
||||
a, d = wolfdawn.parse_strings_inject_counts(res.stdout, res.stderr)
|
||||
untranslated = wolfdawn.parse_strings_inject_untranslated(
|
||||
res.stdout, res.stderr
|
||||
)
|
||||
mismatches = wolfdawn.parse_strings_inject_mismatches(
|
||||
res.stdout, res.stderr
|
||||
)
|
||||
strings_ok = wolfdawn.inject_had_applied(a) or (
|
||||
res.ok and not (a == 0 and (d or 0) > 0)
|
||||
)
|
||||
if strings_ok:
|
||||
applied += 1
|
||||
_sync_json(entry["json"], src, log)
|
||||
if mismatches or (untranslated or 0) > 0:
|
||||
guard_failures.append(entry["json"])
|
||||
self._log_inject_guard_lines(mismatches, untranslated, log)
|
||||
if not res.ok and wolfdawn.inject_had_applied(a):
|
||||
log(
|
||||
f" ⚠ {entry['json']}: wolf exited {res.returncode} "
|
||||
f"after applying {a} change(s) — see warnings above"
|
||||
)
|
||||
elif res.ok:
|
||||
# Exit 0 but nothing applied and lines drifted = stale/injected base.
|
||||
failed += 1
|
||||
guard_failures.append(entry["json"])
|
||||
log(
|
||||
f" ⚠ {entry['json']}: 0 applied, {d} skipped as drift — the base "
|
||||
"looks already translated. Restore the pristine original (re-run "
|
||||
|
|
@ -2571,6 +2633,8 @@ class WolfWorkflowTab(QWidget):
|
|||
)
|
||||
else:
|
||||
failed += 1
|
||||
guard_failures.append(entry["json"])
|
||||
self._log_inject_guard_lines(mismatches, untranslated, log)
|
||||
log(f" ⚠ inject exit {res.returncode} for {entry['json']}")
|
||||
|
||||
# Name values: apply across the whole data dir when translated/names.json
|
||||
|
|
@ -2589,10 +2653,7 @@ class WolfWorkflowTab(QWidget):
|
|||
allow_code_drift=allow_drift, en_punct=en_punct, log_fn=log,
|
||||
)
|
||||
out = (res.stdout or "") + (res.stderr or "")
|
||||
a, d = wolfdawn.parse_names_inject_counts(out)
|
||||
# Keep wolf_json/names.json aligned with translated/ even when the
|
||||
# binary pass is a no-op (stale base) or exits 2 after partial guards.
|
||||
_sync_json(names_entry["json"], names_src, log)
|
||||
a, d = wolfdawn.parse_names_inject_counts(res.stdout, res.stderr)
|
||||
if wolfdawn.inject_had_applied(a):
|
||||
applied += 1
|
||||
if not res.ok:
|
||||
|
|
@ -2620,9 +2681,30 @@ class WolfWorkflowTab(QWidget):
|
|||
failed += 1
|
||||
log(f" ⚠ names-inject exit {res.returncode}")
|
||||
|
||||
# Refresh wolf_json/ from translated/ for git (full inject: all docs;
|
||||
# quick inject: only the files we touched).
|
||||
if only_json is None:
|
||||
sync_targets = [
|
||||
e["json"] for e in entries if e.get("kind") != "names"
|
||||
]
|
||||
else:
|
||||
sync_targets = sorted(strings_targets)
|
||||
if will_names and names_entry:
|
||||
sync_targets.append(names_entry["json"])
|
||||
synced = 0
|
||||
for json_name in sync_targets:
|
||||
if self._sync_translated_json_to_wolf_json(
|
||||
json_name, log, game_json_dir
|
||||
):
|
||||
synced += 1
|
||||
if synced:
|
||||
log(f"Synced {synced} translated JSON file(s) into {WORK_DIR_NAME}/")
|
||||
|
||||
msg = f"Injected {applied} document(s)"
|
||||
if guard_failures:
|
||||
msg += f"; {len(guard_failures)} had guard warnings (see log)"
|
||||
if failed:
|
||||
msg += f", {failed} had guard/errors (see log)"
|
||||
msg += f"; {failed} failed"
|
||||
if quick:
|
||||
msg += ". Review the git diff in the game project and test in-game."
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -15,7 +15,10 @@ Supported document ``kind``s (all share the ``{source, text}`` leaf pattern):
|
|||
|
||||
Only entries whose ``source`` contains target-language (Japanese by default)
|
||||
text are sent to the model; everything else keeps ``text == source`` so inject
|
||||
is a no-op for it.
|
||||
is a no-op for it. When ``IGNORETLTEXT`` is on (default), entries whose ``text``
|
||||
is already translated are skipped so Phase 2 can resume a partial run without
|
||||
retranslating completed lines. Skip looks past Japanese nameplates on line 1
|
||||
and Japanese embedded only in WOLF control codes (e.g. ``\\r[...]`` ruby).
|
||||
|
||||
names.json safety: WolfDawn tags every value name with a static ``safety`` badge
|
||||
(``safe``, ``refs``, or ``verify``). For ``kind == "names"`` documents, only
|
||||
|
|
@ -65,7 +68,8 @@ from util.translation import (
|
|||
)
|
||||
from util import speakers as wolf_speakers
|
||||
from util import vocab as wolf_vocab
|
||||
from util import wolf_names
|
||||
from util.wolfdawn import codes as wolf_codes
|
||||
from util.wolfdawn import names as wolf_names
|
||||
|
||||
# Globals (mirror the other engine modules; populated from .env at import time)
|
||||
MODEL = os.getenv("model")
|
||||
|
|
@ -84,13 +88,24 @@ FILENAME = None
|
|||
# Regex - default matches Japanese (kanji, kana, full-width forms).
|
||||
LANGREGEX = r"[一-龠ぁ-ゔァ-ヴーａ-ｚＡ-Ｚ０-９\uFF61-\uFF9F]+"
|
||||
|
||||
# Skip lines whose ``text`` is already translated. WolfDawn keeps Japanese in
|
||||
# ``source`` forever, so skip-translated must look at ``text``, not ``source``.
|
||||
# Set False for a forced full retranslate.
|
||||
IGNORETLTEXT = True
|
||||
|
||||
# Control codes that can embed Japanese inside an otherwise-translated line
|
||||
# (ruby ``\r[kanji,kana]``, colour / font indexes, ``\cself[n]``, etc.).
|
||||
_WOLF_CODE_RE = re.compile(
|
||||
r"\\(?:r\[[^\]]*\]|c(?:self)?\[[^\]]*\]|[A-Za-z]+\[[^\]]*\]|[A-Za-z])"
|
||||
)
|
||||
|
||||
# Speaker handling: for first-line-speaker formats, reshape the line into the
|
||||
# shared "[Speaker]: line" transport before translating and restore WOLF's
|
||||
# native "Speaker\nline" layout on write-back. Which formats are reshaped is
|
||||
# configurable from the workflow (data/wolf_speakers.json).
|
||||
SPEAKER_CONFIG = wolf_speakers.load_config()
|
||||
|
||||
# names.json: translate per-entry safe/refs badges only (see util.wolf_names).
|
||||
# names.json: translate per-entry safe/refs badges only (see util.wolfdawn.names).
|
||||
|
||||
# Text wrapping: optional advanced rewrap via dazedwrap (wolfWrap/wolfWidth). Defaults
|
||||
# off - the workflow uses WolfDawn relayout after inject instead.
|
||||
|
|
@ -134,6 +149,11 @@ TRANSLATION_CONFIG = TranslationConfig(
|
|||
)
|
||||
|
||||
|
||||
def _batch_phase() -> str:
|
||||
"""Current Anthropic batch phase from the GUI subprocess env (``collect`` / ``consume``)."""
|
||||
return (os.getenv("BATCH_PHASE") or "").strip().lower()
|
||||
|
||||
|
||||
def handleWolfDawn(filename, estimate):
|
||||
"""Entry point used by the CLI/GUI dispatchers. Returns a summary string or 'Fail'."""
|
||||
global ESTIMATE, TOKENS, FILENAME, SPEAKER_CONFIG, VOCAB
|
||||
|
|
@ -151,7 +171,11 @@ def handleWolfDawn(filename, estimate):
|
|||
start = time.time()
|
||||
translatedData = openFiles(filename)
|
||||
|
||||
if not estimate:
|
||||
# Batch collect only queues API requests; text is still Japanese. Writing
|
||||
# translated/ here would overwrite prior work with rewrapped source (and with
|
||||
# wolfWrap on, the git diff looks like "translations" that never changed).
|
||||
# Real English is written on the consume pass (or a live non-batch run).
|
||||
if not estimate and _batch_phase() != "collect":
|
||||
try:
|
||||
with open("translated/" + filename, "w", encoding="utf-8", newline="\n") as outFile:
|
||||
json.dump(translatedData[0], outFile, ensure_ascii=False, indent=4)
|
||||
|
|
@ -252,6 +276,44 @@ def _wrap_plain(text, is_firstline):
|
|||
return prefix + _wrap_body(rest)
|
||||
|
||||
|
||||
def _text_check_body(text: str, speaker_src: str = "") -> str:
|
||||
"""Body text used for skip-translated checks (nameplates / codes ignored).
|
||||
|
||||
Already-translated WOLF dialogue often keeps a Japanese speaker name on
|
||||
line 1 (``司祭\\nSorry to keep you...``) and may embed Japanese only inside
|
||||
control codes like ``\\r[我,わ]``. Those lines are finished translations.
|
||||
"""
|
||||
if not isinstance(text, str):
|
||||
return ""
|
||||
_prefix, rest = wolf_speakers.split_window_prefix(text)
|
||||
if "\n" in rest:
|
||||
first, body = rest.split("\n", 1)
|
||||
# Known first-line speaker formats, or a short nameplate-like first line.
|
||||
if (
|
||||
speaker_src in wolf_speakers.FIRSTLINE_SRCS
|
||||
or speaker_src in ("ui", "narration")
|
||||
or (0 < len(first.strip()) <= 20 and re.search(LANGREGEX, first))
|
||||
):
|
||||
rest = body
|
||||
return _WOLF_CODE_RE.sub("", rest)
|
||||
|
||||
|
||||
def _text_still_needs_translation(entry) -> bool:
|
||||
"""True when ``text`` still needs the model (empty, Japanese, or identical)."""
|
||||
txt = entry.get("text")
|
||||
if not isinstance(txt, str) or not txt.strip():
|
||||
return True
|
||||
src = entry.get("source")
|
||||
# Fresh / unfinished extract: text still mirrors Japanese source.
|
||||
if isinstance(src, str) and txt == src:
|
||||
return True
|
||||
# Fully English (no Japanese chars at all).
|
||||
if not re.search(LANGREGEX, txt):
|
||||
return False
|
||||
# Japanese only in the nameplate / control codes while the body is done.
|
||||
return bool(re.search(LANGREGEX, _text_check_body(txt, entry.get("speaker_src", ""))))
|
||||
|
||||
|
||||
def parseDocument(data, filename):
|
||||
"""Translate every translatable leaf entry and return [data, tokens, error]."""
|
||||
global PBAR
|
||||
|
|
@ -264,71 +326,92 @@ def parseDocument(data, filename):
|
|||
src = e.get("source")
|
||||
if not (isinstance(src, str) and re.search(LANGREGEX, src)):
|
||||
return False
|
||||
# Already translated: look at ``text`` (source stays Japanese forever).
|
||||
# Fresh extracts keep ``text == source``, so they still queue.
|
||||
if IGNORETLTEXT and not _text_still_needs_translation(e):
|
||||
return False
|
||||
# names.json: only translate WolfDawn safe/refs entries (per-name badge).
|
||||
if is_names and not wolf_names.is_name_translatable(e):
|
||||
return False
|
||||
return True
|
||||
|
||||
# Only translate entries that actually contain target-language text (and, for
|
||||
# names.json, carry a translatable safety badge); the rest keep text == source so
|
||||
# WolfDawn treats them as untouched on inject.
|
||||
# Only translate entries that still need work; names.json also requires a
|
||||
# safe/refs badge. Untouched leaves keep ``text == source`` so WolfDawn
|
||||
# treats them as no-ops on inject.
|
||||
translatable = [e for e in entries if _translatable(e)]
|
||||
|
||||
with tqdm(bar_format=BAR_FORMAT, position=POSITION, total=len(translatable), leave=LEAVE) as pbar:
|
||||
pbar.desc = filename
|
||||
PBAR = pbar
|
||||
if not translatable:
|
||||
return [data, totalTokens, None]
|
||||
|
||||
# Reshape first-line-speaker lines into the shared "[Speaker]: line"
|
||||
# transport format. plans[i] carries what is needed to restore each
|
||||
# entry after translation.
|
||||
sources = []
|
||||
plans = [] # (entry, prefix, has_speaker, is_firstline)
|
||||
for entry in translatable:
|
||||
src = entry["source"]
|
||||
is_firstline = entry.get("speaker_src", "") in wolf_speakers.FIRSTLINE_SRCS
|
||||
split = wolf_speakers.split_source(src, entry.get("speaker_src", ""), SPEAKER_CONFIG)
|
||||
if split is not None:
|
||||
prefix, speaker, body = split
|
||||
sources.append(wolf_speakers.to_prefixed(speaker, body))
|
||||
plans.append((entry, prefix, True, is_firstline))
|
||||
else:
|
||||
sources.append(src)
|
||||
plans.append((entry, "", False, is_firstline))
|
||||
|
||||
try:
|
||||
response = translateAI(sources, [])
|
||||
except Exception as e:
|
||||
return [data, totalTokens, e]
|
||||
|
||||
translated, tokens = response[0], response[1]
|
||||
totalTokens[0] += tokens[0]
|
||||
totalTokens[1] += tokens[1]
|
||||
|
||||
# Write translations back (skip in estimate mode: translated == sources).
|
||||
if not ESTIMATE and isinstance(translated, list) and len(translated) == len(plans):
|
||||
for (entry, prefix, has_speaker, is_firstline), text in zip(plans, translated):
|
||||
if not isinstance(text, str):
|
||||
continue
|
||||
if is_names:
|
||||
entry["text"] = _wrap_names_entry(text, entry)
|
||||
elif has_speaker:
|
||||
speaker_en, body_en = wolf_speakers.parse_prefixed(text)
|
||||
if speaker_en is not None:
|
||||
# Wrap only the body; the name stays on its own line.
|
||||
entry["text"] = wolf_speakers.restore_source(
|
||||
prefix, speaker_en, _wrap_body(body_en)
|
||||
)
|
||||
else:
|
||||
# Model dropped the [Speaker]: prefix; keep its output as-is.
|
||||
entry["text"] = prefix + _wrap_body(text)
|
||||
if translatable:
|
||||
# Reshape first-line-speaker lines into the shared "[Speaker]: line"
|
||||
# transport format. plans[i] carries what is needed to restore each
|
||||
# entry after translation.
|
||||
sources = []
|
||||
plans = [] # (entry, prefix, has_speaker, is_firstline, code_map)
|
||||
for entry in translatable:
|
||||
src = entry["source"]
|
||||
protected_src, code_map = wolf_codes.protect_wolf_codes(src)
|
||||
is_firstline = entry.get("speaker_src", "") in wolf_speakers.FIRSTLINE_SRCS
|
||||
split = wolf_speakers.split_source(
|
||||
protected_src, entry.get("speaker_src", ""), SPEAKER_CONFIG
|
||||
)
|
||||
if split is not None:
|
||||
prefix, speaker, body = split
|
||||
sources.append(wolf_speakers.to_prefixed(speaker, body))
|
||||
plans.append((entry, prefix, True, is_firstline, code_map))
|
||||
else:
|
||||
entry["text"] = _wrap_plain(text, is_firstline)
|
||||
sources.append(protected_src)
|
||||
plans.append((entry, "", False, is_firstline, code_map))
|
||||
|
||||
# Phase 0 feeds the glossary: harvest the just-translated safe name values
|
||||
# into vocab.txt (grouped by note) so the DB-text and dialogue phases keep
|
||||
# item/skill/term names consistent. Mirrors the RPGMaker DB-first strategy.
|
||||
try:
|
||||
response = translateAI(sources, [])
|
||||
except Exception as e:
|
||||
return [data, totalTokens, e]
|
||||
|
||||
translated, tokens = response[0], response[1]
|
||||
totalTokens[0] += tokens[0]
|
||||
totalTokens[1] += tokens[1]
|
||||
|
||||
# Write translations back. Skip estimate mode, and skip the batch
|
||||
# collect pass (response is still the Japanese source list — wrapping
|
||||
# it into ``text`` only reflows JP and is what poisoned translated/).
|
||||
collecting = _batch_phase() == "collect"
|
||||
if (
|
||||
not ESTIMATE
|
||||
and not collecting
|
||||
and isinstance(translated, list)
|
||||
and len(translated) == len(plans)
|
||||
):
|
||||
for (entry, prefix, has_speaker, is_firstline, code_map), text, src in zip(
|
||||
plans, translated, sources
|
||||
):
|
||||
if not isinstance(text, str):
|
||||
continue
|
||||
# Model / collect echoed the sent payload — leave the entry alone.
|
||||
if text == src:
|
||||
continue
|
||||
text = wolf_codes.restore_wolf_code_placeholders(text, code_map)
|
||||
if is_names:
|
||||
entry["text"] = _wrap_names_entry(text, entry)
|
||||
elif has_speaker:
|
||||
speaker_en, body_en = wolf_speakers.parse_prefixed(text)
|
||||
if speaker_en is not None:
|
||||
# Wrap only the body; the name stays on its own line.
|
||||
entry["text"] = wolf_speakers.restore_source(
|
||||
prefix, speaker_en, _wrap_body(body_en)
|
||||
)
|
||||
else:
|
||||
# Model dropped the [Speaker]: prefix; keep its output as-is.
|
||||
entry["text"] = prefix + _wrap_body(text)
|
||||
else:
|
||||
entry["text"] = _wrap_plain(text, is_firstline)
|
||||
wolf_codes.repair_entry(entry)
|
||||
|
||||
# Phase 0 feeds the glossary: harvest safe name values into vocab.txt
|
||||
# (grouped by note) so the DB-text and dialogue phases keep item/skill/term
|
||||
# names consistent. Runs even when every name was already translated (skip),
|
||||
# so a resume still seeds vocab. Mirrors the RPGMaker DB-first strategy.
|
||||
if is_names and not ESTIMATE:
|
||||
_harvest_names_to_vocab(data)
|
||||
|
||||
|
|
|
|||
64
tests/test_wolf_codes.py
Normal file
64
tests/test_wolf_codes.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Tests for WOLF inline control-code repair."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
os.chdir(ROOT)
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from util.wolfdawn import codes as wolf_codes # noqa: E402
|
||||
|
||||
|
||||
class WolfCodesRepairTests(unittest.TestCase):
|
||||
def test_fixes_spurious_space_before_caret(self):
|
||||
source = "占い師\nはぁ！\\^"
|
||||
text = "Fortune-teller\nHa!\\ ^"
|
||||
fixed = wolf_codes.rebuild_text_preserving_source_codes(source, text)
|
||||
self.assertEqual(fixed, "Fortune-teller\nHa!\\^")
|
||||
|
||||
def test_rebuild_preserves_multiple_codes(self):
|
||||
source = "A\\c[1]B\\f[2]C"
|
||||
text = "A\\c[ 1]B\\f[ 2]C"
|
||||
fixed = wolf_codes.rebuild_text_preserving_source_codes(source, text)
|
||||
self.assertEqual(fixed, "A\\c[1]B\\f[2]C")
|
||||
|
||||
def test_protect_and_restore_roundtrip(self):
|
||||
src = "Line with \\^ and \\cself[8]"
|
||||
protected, mapping = wolf_codes.protect_wolf_codes(src)
|
||||
self.assertNotIn("\\^", protected)
|
||||
restored = wolf_codes.restore_wolf_code_placeholders(protected, mapping)
|
||||
self.assertEqual(restored, src)
|
||||
|
||||
def test_repair_document_updates_leaf(self):
|
||||
doc = {
|
||||
"kind": "map",
|
||||
"scenes": [
|
||||
{
|
||||
"event": 374,
|
||||
"lines": [
|
||||
{
|
||||
"cmd": 233,
|
||||
"str": 0,
|
||||
"source": "占い師\nはぁ！\\^",
|
||||
"text": "Fortune-teller\nHa!\\ ^",
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
_doc, notes = wolf_codes.repair_document(doc)
|
||||
self.assertEqual(len(notes), 1)
|
||||
self.assertEqual(
|
||||
doc["scenes"][0]["lines"][0]["text"],
|
||||
"Fortune-teller\nHa!\\^",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -33,6 +33,20 @@ class WolfInjectCountsTests(unittest.TestCase):
|
|||
self.assertEqual(applied, 91)
|
||||
self.assertEqual(drifted, 3)
|
||||
|
||||
def test_parse_strings_inject_counts_with_untranslated_on_stderr(self):
|
||||
stderr = (
|
||||
"event 374 cmd 233 str 0: control-code mismatch - source has [\"\\\\^\"], "
|
||||
"translation has [\"\\\\ \"]\n"
|
||||
"applied 24547 translation(s) (37 untranslated, 0 drifted); wrote CommonEvent.dat\n"
|
||||
)
|
||||
applied, drifted = wolfdawn.parse_strings_inject_counts("", stderr)
|
||||
self.assertEqual(applied, 24547)
|
||||
self.assertEqual(drifted, 0)
|
||||
self.assertEqual(wolfdawn.parse_strings_inject_untranslated("", stderr), 37)
|
||||
mismatches = wolfdawn.parse_strings_inject_mismatches("", stderr)
|
||||
self.assertEqual(len(mismatches), 1)
|
||||
self.assertIn("control-code mismatch", mismatches[0])
|
||||
|
||||
def test_inject_had_applied_rejects_zero(self):
|
||||
self.assertFalse(wolfdawn.inject_had_applied(0))
|
||||
self.assertFalse(wolfdawn.inject_had_applied(None))
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Unit tests for WolfDawn names.json helpers in util/wolf_names.py."""
|
||||
"""Unit tests for WolfDawn names.json helpers in util/wolfdawn/names.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -14,7 +14,7 @@ ROOT = Path(__file__).resolve().parents[1]
|
|||
os.chdir(ROOT)
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from util import wolf_names as wn # noqa: E402
|
||||
from util.wolfdawn import names as wn # noqa: E402
|
||||
|
||||
|
||||
class TestNameTranslatable(unittest.TestCase):
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ class _WolfTranslateHarness:
|
|||
self.vocab_writes = []
|
||||
self.vocab_remove_writes = []
|
||||
|
||||
def run(self, data, filename="doc.json", estimate=False):
|
||||
def run(self, data, filename="doc.json", estimate=False, ignore_tl_text=True):
|
||||
def translate(text, history, history_ctx=None):
|
||||
self.captured.append(copy.deepcopy(text))
|
||||
return _mock_translate(text, history, history_ctx)
|
||||
|
|
@ -52,11 +52,13 @@ class _WolfTranslateHarness:
|
|||
orig_t = wd.translateAI
|
||||
orig_estimate = wd.ESTIMATE
|
||||
orig_wrap = wd.WRAP
|
||||
orig_ignore = wd.IGNORETLTEXT
|
||||
orig_update = wd.wolf_vocab.update_vocab_section
|
||||
orig_labels = wd.wolf_names.derive_db_labels
|
||||
wd.translateAI = translate
|
||||
wd.ESTIMATE = estimate
|
||||
wd.WRAP = False # keep write-back byte-faithful; wrapping tested separately
|
||||
wd.IGNORETLTEXT = ignore_tl_text
|
||||
# Never touch the real glossary / DB files during tests.
|
||||
wd.wolf_vocab.update_vocab_section = capture_vocab
|
||||
wd.wolf_names.derive_db_labels = lambda _p: {}
|
||||
|
|
@ -68,6 +70,7 @@ class _WolfTranslateHarness:
|
|||
wd.translateAI = orig_t
|
||||
wd.ESTIMATE = orig_estimate
|
||||
wd.WRAP = orig_wrap
|
||||
wd.IGNORETLTEXT = orig_ignore
|
||||
wd.wolf_vocab.update_vocab_section = orig_update
|
||||
wd.wolf_names.derive_db_labels = orig_labels
|
||||
|
||||
|
|
@ -244,6 +247,213 @@ class TestTranslationWriteback(unittest.TestCase):
|
|||
# In estimate mode text stays equal to source.
|
||||
self.assertEqual(data["scenes"][0]["lines"][0]["text"], "こんにちは")
|
||||
|
||||
def test_collect_pass_does_not_mutate_or_write(self):
|
||||
"""Batch collect must not reflow JP into text or overwrite translated/."""
|
||||
orig_phase = os.environ.get("BATCH_PHASE")
|
||||
os.environ["BATCH_PHASE"] = "collect"
|
||||
orig_wrap = wd.WRAP
|
||||
orig_width = wd.WRAPWIDTH
|
||||
wd.WRAP = True
|
||||
wd.WRAPWIDTH = 20
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
(root / "files").mkdir()
|
||||
(root / "translated").mkdir()
|
||||
doc = copy.deepcopy(MAP_DOC)
|
||||
# A long JP line that wrap would reflow if write-back ran.
|
||||
doc["scenes"][0]["lines"][0]["source"] = "あ" * 40
|
||||
doc["scenes"][0]["lines"][0]["text"] = "あ" * 40
|
||||
src_path = root / "files" / "Map001.mps.json"
|
||||
src_path.write_text(json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
||||
out_path = root / "translated" / "Map001.mps.json"
|
||||
out_path.write_text('{"kind":"map","marker":"keep-me"}', encoding="utf-8")
|
||||
|
||||
old_cwd = os.getcwd()
|
||||
os.chdir(root)
|
||||
try:
|
||||
# Echo sources (what collect's queue path effectively returns).
|
||||
def echo(text, history, history_ctx=None):
|
||||
return [text if isinstance(text, list) else text, [0, 0]]
|
||||
|
||||
orig_t = wd.translateAI
|
||||
wd.translateAI = echo
|
||||
try:
|
||||
result = wd.handleWolfDawn("Map001.mps.json", estimate=False)
|
||||
finally:
|
||||
wd.translateAI = orig_t
|
||||
finally:
|
||||
os.chdir(old_cwd)
|
||||
|
||||
self.assertNotEqual(result, "Fail")
|
||||
# Prior translated/ content must be left alone.
|
||||
self.assertEqual(
|
||||
out_path.read_text(encoding="utf-8"),
|
||||
'{"kind":"map","marker":"keep-me"}',
|
||||
)
|
||||
finally:
|
||||
wd.WRAP = orig_wrap
|
||||
wd.WRAPWIDTH = orig_width
|
||||
if orig_phase is None:
|
||||
os.environ.pop("BATCH_PHASE", None)
|
||||
else:
|
||||
os.environ["BATCH_PHASE"] = orig_phase
|
||||
|
||||
def test_echoed_source_does_not_get_wrapped_into_text(self):
|
||||
"""If the model/collect echoes JP source, leave text alone even with wrap on."""
|
||||
orig_wrap = wd.WRAP
|
||||
orig_width = wd.WRAPWIDTH
|
||||
wd.WRAP = True
|
||||
wd.WRAPWIDTH = 10
|
||||
try:
|
||||
doc = {
|
||||
"kind": "map",
|
||||
"scenes": [
|
||||
{
|
||||
"event": 1,
|
||||
"name": "ev",
|
||||
"lines": [
|
||||
{
|
||||
"cmd": 0,
|
||||
"str": 0,
|
||||
"source": "あいうえおかきくけこさしすせそ",
|
||||
"text": "あいうえおかきくけこさしすせそ",
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
def echo(text, history, history_ctx=None):
|
||||
return [text if isinstance(text, list) else text, [1, 1]]
|
||||
|
||||
orig_t = wd.translateAI
|
||||
wd.translateAI = echo
|
||||
try:
|
||||
data, _tok, err = wd.parseDocument(copy.deepcopy(doc), "echo.mps.json")
|
||||
finally:
|
||||
wd.translateAI = orig_t
|
||||
self.assertIsNone(err)
|
||||
# Must still be the unbroken source, not a wrapped JP reflow.
|
||||
self.assertEqual(
|
||||
data["scenes"][0]["lines"][0]["text"],
|
||||
"あいうえおかきくけこさしすせそ",
|
||||
)
|
||||
finally:
|
||||
wd.WRAP = orig_wrap
|
||||
wd.WRAPWIDTH = orig_width
|
||||
|
||||
def test_skips_already_translated_text(self):
|
||||
doc = {
|
||||
"kind": "map",
|
||||
"scenes": [
|
||||
{
|
||||
"event": 1,
|
||||
"name": "ev",
|
||||
"lines": [
|
||||
{
|
||||
"cmd": 0,
|
||||
"str": 0,
|
||||
"source": "こんにちは",
|
||||
"text": "Hello",
|
||||
},
|
||||
{
|
||||
"cmd": 1,
|
||||
"str": 0,
|
||||
"source": "さようなら",
|
||||
"text": "さようなら",
|
||||
},
|
||||
{
|
||||
"cmd": 2,
|
||||
"str": 0,
|
||||
"source": "まだ日本語",
|
||||
"text": "Partial 日本語 left",
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
(data, _t, err), captured = _WolfTranslateHarness().run(doc, "partial.mps.json")
|
||||
self.assertIsNone(err)
|
||||
lines = data["scenes"][0]["lines"]
|
||||
self.assertEqual(lines[0]["text"], "Hello")
|
||||
self.assertEqual(lines[1]["text"], "EN_さようなら")
|
||||
# Still-Japanese body is not skipped; the model still gets ``source``.
|
||||
self.assertEqual(lines[2]["text"], "EN_まだ日本語")
|
||||
self.assertEqual(captured, [["さようなら", "まだ日本語"]])
|
||||
|
||||
def test_skips_english_body_with_japanese_nameplate(self):
|
||||
doc = {
|
||||
"kind": "map",
|
||||
"scenes": [
|
||||
{
|
||||
"event": 1,
|
||||
"name": "ev",
|
||||
"lines": [
|
||||
{
|
||||
"cmd": 0,
|
||||
"str": 0,
|
||||
"speaker": "司祭",
|
||||
"speaker_src": "literal_line1_lowconf",
|
||||
"source": "司祭\n皆さんお待たせしました……。",
|
||||
"text": "司祭\nSorry to keep you all waiting......",
|
||||
},
|
||||
{
|
||||
"cmd": 1,
|
||||
"str": 0,
|
||||
"speaker": "UI",
|
||||
"speaker_src": "ui",
|
||||
"source": "まだだ",
|
||||
"text": "まだだ",
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
(data, _t, err), captured = _WolfTranslateHarness().run(doc, "nameplate.mps.json")
|
||||
self.assertIsNone(err)
|
||||
lines = data["scenes"][0]["lines"]
|
||||
self.assertEqual(lines[0]["text"], "司祭\nSorry to keep you all waiting......")
|
||||
self.assertEqual(lines[1]["text"], "EN_まだだ")
|
||||
self.assertEqual(captured, [["まだだ"]])
|
||||
|
||||
def test_ignore_tl_text_false_retranslates(self):
|
||||
doc = {
|
||||
"kind": "map",
|
||||
"scenes": [
|
||||
{
|
||||
"event": 1,
|
||||
"name": "ev",
|
||||
"lines": [
|
||||
{"cmd": 0, "str": 0, "source": "こんにちは", "text": "Hello"},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
(data, _t, err), captured = _WolfTranslateHarness().run(
|
||||
doc, "force.mps.json", ignore_tl_text=False
|
||||
)
|
||||
self.assertIsNone(err)
|
||||
self.assertEqual(data["scenes"][0]["lines"][0]["text"], "EN_こんにちは")
|
||||
self.assertEqual(captured, [["こんにちは"]])
|
||||
|
||||
def test_names_already_translated_still_harvest(self):
|
||||
doc = {
|
||||
"kind": "names",
|
||||
"names": [
|
||||
{"source": "剣", "text": "Sword", "note": "武器", "safety": "safe"},
|
||||
{"source": "槍", "text": "Spear", "note": "武器", "safety": "refs"},
|
||||
],
|
||||
}
|
||||
harness = _WolfTranslateHarness()
|
||||
(_data, _t, err), captured = harness.run(doc, "names.json")
|
||||
self.assertIsNone(err)
|
||||
self.assertEqual(captured, [])
|
||||
self.assertEqual(
|
||||
harness.vocab_writes,
|
||||
[("Weapon · 武器", [("剣", "Sword"), ("槍", "Spear")])],
|
||||
)
|
||||
|
||||
|
||||
import re # noqa: E402
|
||||
|
||||
|
|
|
|||
|
|
@ -42,6 +42,8 @@ __all__ = [
|
|||
"strings_inject",
|
||||
"names_inject",
|
||||
"parse_strings_inject_counts",
|
||||
"parse_strings_inject_untranslated",
|
||||
"parse_strings_inject_mismatches",
|
||||
"parse_names_inject_counts",
|
||||
"inject_had_applied",
|
||||
"pack",
|
||||
|
|
@ -55,6 +57,17 @@ __all__ = [
|
|||
_INJECT_COUNTS_RE = re.compile(
|
||||
r"applied\s+(\d+)\s+translation.*?(\d+)\s+drifted", re.IGNORECASE | re.DOTALL
|
||||
)
|
||||
_INJECT_UNTRANSLATED_RE = re.compile(
|
||||
r"applied\s+\d+\s+translation.*?\((\d+)\s+untranslated", re.IGNORECASE | re.DOTALL
|
||||
)
|
||||
_INJECT_MISMATCH_RE = re.compile(
|
||||
r"^(?:event\s+\d+\s+)?cmd\s+(\d+)\s+str\s+(\d+):\s+control-code mismatch",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_INJECT_MISMATCH_EVENT_RE = re.compile(
|
||||
r"^event\s+(\d+)\s+cmd\s+(\d+)\s+str\s+(\d+):\s+control-code mismatch",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_NAMES_INJECT_COUNTS_RE = re.compile(
|
||||
r"applied\s+(\d+)\s+name change.*?(\d+)\s+drifted", re.IGNORECASE | re.DOTALL
|
||||
)
|
||||
|
|
@ -69,21 +82,46 @@ _UNPACK_SUMMARY_RE = re.compile(
|
|||
)
|
||||
|
||||
|
||||
def parse_strings_inject_counts(stdout: str) -> tuple[int | None, int | None]:
|
||||
def _inject_output_text(stdout: str, stderr: str) -> str:
|
||||
"""WolfDawn prints per-line guards to stderr and the summary to either stream."""
|
||||
return "\n".join(part for part in (stdout or "", stderr or "") if part)
|
||||
|
||||
|
||||
def parse_strings_inject_counts(stdout: str, stderr: str = "") -> tuple[int | None, int | None]:
|
||||
"""Return (applied, drifted) from wolf strings-inject output, or (None, None)."""
|
||||
if not stdout:
|
||||
text = _inject_output_text(stdout, stderr)
|
||||
if not text:
|
||||
return None, None
|
||||
m = _INJECT_COUNTS_RE.search(stdout)
|
||||
m = _INJECT_COUNTS_RE.search(text)
|
||||
if not m:
|
||||
return None, None
|
||||
return int(m.group(1)), int(m.group(2))
|
||||
|
||||
|
||||
def parse_names_inject_counts(stdout: str) -> tuple[int | None, int | None]:
|
||||
def parse_strings_inject_untranslated(stdout: str, stderr: str = "") -> int | None:
|
||||
"""Return the untranslated line count from strings-inject summary, if present."""
|
||||
text = _inject_output_text(stdout, stderr)
|
||||
m = _INJECT_UNTRANSLATED_RE.search(text)
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
|
||||
def parse_strings_inject_mismatches(stdout: str, stderr: str = "") -> list[str]:
|
||||
"""Return human-readable control-code mismatch locations from inject output."""
|
||||
text = _inject_output_text(stdout, stderr)
|
||||
out: list[str] = []
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if "control-code mismatch" in line:
|
||||
out.append(line)
|
||||
return out
|
||||
|
||||
|
||||
def parse_names_inject_counts(stdout: str, stderr: str = "") -> tuple[int | None, int | None]:
|
||||
"""Return (applied, drifted) from wolf names-inject output, or (None, None)."""
|
||||
if not stdout:
|
||||
text = _inject_output_text(stdout, stderr)
|
||||
if not text:
|
||||
return None, None
|
||||
m = _NAMES_INJECT_COUNTS_RE.search(stdout)
|
||||
m = _NAMES_INJECT_COUNTS_RE.search(text)
|
||||
if not m:
|
||||
return None, None
|
||||
return int(m.group(1)), int(m.group(2))
|
||||
|
|
|
|||
169
util/wolfdawn/codes.py
Normal file
169
util/wolfdawn/codes.py
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
"""WOLF RPG Editor inline control-code helpers for WolfDawn JSON.
|
||||
|
||||
WolfDawn ``strings-inject`` requires each ``text`` line to carry the same
|
||||
backslash control codes as its ``source`` (only the human-readable words may
|
||||
change). Models often break codes (e.g. ``\\^`` becomes ``\\ ^``). These
|
||||
helpers protect codes during translation and repair them before inject.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from util import speakers as wolf_speakers
|
||||
|
||||
# Inline codes WolfDawn compares during inject (see strings-inject guard output).
|
||||
# Order matters: longer bracketed forms before single-letter fallbacks.
|
||||
WOLF_INLINE_CODE_RE = re.compile(
|
||||
r"\\(?:"
|
||||
r"r\[[^\]]*\]|" # ruby \r[kanji,kana]
|
||||
r"c(?:self)?\[[^\]]*\]|" # \c[...] / \cself[...]
|
||||
r"[A-Za-z]+\[[^\]]*\]|" # \font[0], \cdb[21:78:0], \wE[1], ...
|
||||
r"\^|" # \^ (wait / beat — often mangled to "\ ^")
|
||||
r"[ @<>\-]|" # other single-char escapes Wolf uses
|
||||
r"[A-Za-z]" # \n, \c, \f, ...
|
||||
r")"
|
||||
)
|
||||
|
||||
# Placeholder transport (same convention as util.translation.protect_script_codes).
|
||||
_PLACEHOLDER_PREFIX = "__WOLF_CODE_"
|
||||
|
||||
|
||||
def _tokenize(rest: str) -> list[tuple[str, str]]:
|
||||
"""Split *rest* into ``('lit', ...)`` and ``('code', ...)`` tokens."""
|
||||
if not rest:
|
||||
return []
|
||||
out: list[tuple[str, str]] = []
|
||||
last = 0
|
||||
for m in WOLF_INLINE_CODE_RE.finditer(rest):
|
||||
if m.start() > last:
|
||||
out.append(("lit", rest[last : m.start()]))
|
||||
out.append(("code", m.group(0)))
|
||||
last = m.end()
|
||||
if last < len(rest):
|
||||
out.append(("lit", rest[last:]))
|
||||
return out
|
||||
|
||||
|
||||
def _fix_spurious_spaces_in_codes(source: str, text: str) -> str:
|
||||
"""Undo common model errors like ``\\ ^`` when *source* has ``\\^``."""
|
||||
for m in WOLF_INLINE_CODE_RE.finditer(source):
|
||||
code = m.group(0)
|
||||
# Single-char escapes: \^ \c etc. Models often insert a space after \\.
|
||||
if len(code) == 2 and code[1] not in (" ", "\t"):
|
||||
spaced = f"{code[0]} {code[1]}"
|
||||
if spaced in text:
|
||||
text = text.replace(spaced, code, 1)
|
||||
return text
|
||||
|
||||
|
||||
def rebuild_text_preserving_source_codes(source: str, text: str) -> str:
|
||||
"""Return *text* with *source*'s inline code tokens and translated literals.
|
||||
|
||||
Keeps any leading ``@<option>\\n`` window prefix from the translated line when
|
||||
present; otherwise preserves the source prefix byte-for-byte.
|
||||
"""
|
||||
if not isinstance(source, str) or not isinstance(text, str):
|
||||
return text
|
||||
if source == text:
|
||||
return text
|
||||
|
||||
src_prefix, src_rest = wolf_speakers.split_window_prefix(source)
|
||||
txt_prefix, txt_rest = wolf_speakers.split_window_prefix(text)
|
||||
out_prefix = txt_prefix if txt_prefix else src_prefix
|
||||
|
||||
txt_rest = _fix_spurious_spaces_in_codes(src_rest, txt_rest)
|
||||
|
||||
src_parts = _tokenize(src_rest)
|
||||
txt_parts = _tokenize(txt_rest)
|
||||
src_lit = [p[1] for p in src_parts if p[0] == "lit"]
|
||||
txt_lit = [p[1] for p in txt_parts if p[0] == "lit"]
|
||||
|
||||
if not src_parts:
|
||||
return out_prefix + txt_rest
|
||||
|
||||
# Rebuild using source code skeleton + translated literal slots.
|
||||
rebuilt: list[str] = []
|
||||
lit_i = 0
|
||||
for kind, val in src_parts:
|
||||
if kind == "code":
|
||||
rebuilt.append(val)
|
||||
else:
|
||||
if lit_i < len(txt_lit):
|
||||
rebuilt.append(txt_lit[lit_i])
|
||||
lit_i += 1
|
||||
else:
|
||||
rebuilt.append(val)
|
||||
return out_prefix + "".join(rebuilt)
|
||||
|
||||
|
||||
def protect_wolf_codes(text: str) -> tuple[str, dict[str, str]]:
|
||||
"""Replace inline WOLF codes with placeholders before sending text to the model."""
|
||||
if not text or not isinstance(text, str):
|
||||
return text, {}
|
||||
replacements: dict[str, str] = {}
|
||||
counter = 0
|
||||
|
||||
def _sub(m: re.Match) -> str:
|
||||
nonlocal counter
|
||||
key = f"{_PLACEHOLDER_PREFIX}{counter}__"
|
||||
replacements[key] = m.group(0)
|
||||
counter += 1
|
||||
return key
|
||||
|
||||
protected = WOLF_INLINE_CODE_RE.sub(_sub, text)
|
||||
return protected, replacements
|
||||
|
||||
|
||||
def restore_wolf_code_placeholders(text: str, replacements: dict[str, str]) -> str:
|
||||
if not text or not replacements:
|
||||
return text
|
||||
out = text
|
||||
for key, original in replacements.items():
|
||||
out = out.replace(key, original)
|
||||
return out
|
||||
|
||||
|
||||
def repair_entry(entry: dict[str, Any]) -> tuple[bool, str]:
|
||||
"""Repair one ``{source, text}`` leaf. Returns (changed, reason)."""
|
||||
src, txt = entry.get("source"), entry.get("text")
|
||||
if not isinstance(src, str) or not isinstance(txt, str) or not src or src == txt:
|
||||
return False, ""
|
||||
fixed = rebuild_text_preserving_source_codes(src, txt)
|
||||
if fixed == txt:
|
||||
return False, ""
|
||||
entry["text"] = fixed
|
||||
return True, "codes"
|
||||
|
||||
|
||||
def _walk_entries(data: dict) -> list[dict]:
|
||||
kind = data.get("kind")
|
||||
entries: list[dict] = []
|
||||
if kind in ("map", "common"):
|
||||
for scene in data.get("scenes") or []:
|
||||
entries.extend(scene.get("lines") or [])
|
||||
elif kind == "db":
|
||||
for group in data.get("groups") or []:
|
||||
entries.extend(group.get("lines") or [])
|
||||
elif kind in ("gamedat", "txt"):
|
||||
entries.extend(data.get("lines") or [])
|
||||
elif kind == "txt-dir":
|
||||
for file_doc in data.get("files") or []:
|
||||
entries.extend(file_doc.get("lines") or [])
|
||||
return entries
|
||||
|
||||
|
||||
def repair_document(data: dict) -> tuple[dict, list[str]]:
|
||||
"""Repair every ``{source, text}`` leaf in a WolfDawn document in place."""
|
||||
notes: list[str] = []
|
||||
for entry in _walk_entries(data):
|
||||
changed, _reason = repair_entry(entry)
|
||||
if changed:
|
||||
row = entry.get("rowName") or entry.get("speaker") or ""
|
||||
notes.append(
|
||||
f"event {entry.get('event', entry.get('row', '?'))} "
|
||||
f"cmd {entry.get('cmd', entry.get('field', '?'))} "
|
||||
f"str {entry.get('str', '')} {row}".strip()
|
||||
)
|
||||
return data, notes
|
||||
Loading…
Reference in a new issue