Files
mrleo1nidandClaude Sonnet 5.5 f49ee90477 feat: browser access (allowed_origins, CORS with Private Network Access), job cancel, Tomie/Unity adapters, benchmark
Includes earlier uncommitted work: Tomie and Unity adapters, benchmark, structured output, call metrics.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-25 18:59:28 +03:00

564 lines
28 KiB
Python

"""Prepare and reversibly install a small, offline Tomie UI translation."""
import argparse
from contextlib import contextmanager
import base64
import csv
from datetime import datetime, timezone
import hashlib
import io
import json
import os
from pathlib import Path
import re
import shutil
import struct
import subprocess
import sys
import time
from uuid import uuid4
import zlib
import httpx
from .benchmark import write_json
from .models import Project
from .naninovel import parse_dictionary
from .protection import protect
DATA = Path("TomieWGM_Data")
STATE = Path(".local-translator/state.json")
MENU_LABELS = {"new game", "start", "continue", "load game", "exit", "quit", "credit", "credits", "gallery", "back", "return", "close", "new",
"content\noption", "game guide\nwalkthrough", "achievements", "achievement", "mod folder", "translation", "savefile folder", "next series"}
def digest(path):
with Path(path).open("rb") as stream:
return hashlib.file_digest(stream, "sha256").hexdigest()
def inside(root, relative):
relative = Path(relative)
if relative.is_absolute() or ".." in relative.parts:
raise ValueError("Unsafe relative path in patch manifest")
target = (Path(root).resolve() / relative).resolve()
if not target.is_relative_to(Path(root).resolve()):
raise ValueError("Patch path escapes its root")
return target
def rewrite_dictionary(script, replacements):
seen, lines = set(), []
for line in script.splitlines(keepends=True):
body = line.rstrip("\r\n")
ending = line[len(body):]
key, separator, value = body.partition(":")
if separator and key.strip() in replacements:
key_name = key.strip()
if key_name in seen:
raise ValueError("Duplicate key in managed text: " + key_name)
seen.add(key_name)
translated = replacements[key_name]
if not isinstance(translated, str) or not translated.strip() or "\n" in translated or "\r" in translated:
raise ValueError("Managed text must contain a nonempty single-line value: " + key_name)
spacing = value[:len(value) - len(value.lstrip())]
body = key + separator + spacing + translated
lines.append(body + ending)
if set(replacements) - seen:
raise ValueError("Translations refer to unknown managed text keys")
return "".join(lines)
def append_managed_records(script, records):
existing = {row["id"] for row in parse_dictionary(script)}
if existing.intersection(records):
raise ValueError("Extra managed text records duplicate existing keys")
newline = "\r\n" if "\r\n" in script else "\n"
if records and script and not script.endswith(("\r", "\n")):
script += newline
for key, value in records.items():
if not value.strip() or any(char in value for char in "\r\n"):
raise ValueError("Extra managed text values must be single-line")
script += key + ": " + value + newline
return script
def bundle_crc(data):
import UnityPy
from UnityPy.streams import EndianBinaryReader
bundle = next(iter(UnityPy.load(data).files.values()))
reader = EndianBinaryReader(data)
if reader.read_string_to_null() != "UnityFS":
raise ValueError("Only UnityFS bundles are supported")
reader.read_int()
reader.read_string_to_null()
reader.read_string_to_null()
_, blocks = bundle.read_fs(reader)
return zlib.crc32(blocks.bytes) & 0xffffffff
def patch_catalog(document, old_hash, crc, size, new_hash):
"""Preserve binary record lengths and all offsets in Addressables metadata."""
raw = base64.b64decode(document["m_ExtraDataString"], validate=True)
needle = ('"m_Hash":"' + old_hash + '"').encode("utf-16-le")
if raw.count(needle) != 1:
raise ValueError("Addressables bundle metadata is missing or ambiguous")
position = raw.index(needle)
start = raw.rfind(b"{\x00", 0, position)
end = raw.find(b"}\x00", position) + 2
if start < 4 or end < 2 or struct.unpack("<I", raw[start-4:start])[0] != end - start:
raise ValueError("Unsupported Addressables metadata record layout")
original = raw[start:end].decode("utf-16-le")
text = original
for name, value in (("m_Crc", crc), ("m_BundleSize", size)):
def replace(match):
value_text = str(value)
if len(value_text) > len(match[2]):
raise ValueError("New bundle metadata exceeds original numeric field width")
return match[1] + value_text.ljust(len(match[2]))
text, count = re.subn(r'("' + name + r'":)(\d+ *)', replace, text)
if count != 1:
raise ValueError("Missing Addressables field: " + name)
text = text.replace(old_hash, new_hash)
encoded = text.encode("utf-16-le")
if len(encoded) != end - start:
raise ValueError("Addressables record length changed")
updated = dict(document)
updated["m_ExtraDataString"] = base64.b64encode(raw[:start] + encoded + raw[end:]).decode()
return updated, json.loads(original), json.loads(text)
def game_running():
if os.name != "nt":
return False
result = subprocess.run(["tasklist", "/FI", "IMAGENAME eq TomieWGM.exe", "/FO", "CSV", "/NH"],
capture_output=True, text=True, errors="replace", timeout=15, creationflags=0x08000000)
if result.returncode:
raise RuntimeError("Could not check whether the game is closed")
return any(row and row[0].lower() == "tomiewgm.exe" for row in csv.reader(result.stdout.splitlines()))
def atomic_copy(source, target, expected):
target.parent.mkdir(parents=True, exist_ok=True)
temporary = target.with_name(target.name + ".lt-" + uuid4().hex + ".tmp")
try:
shutil.copy2(source, temporary)
if digest(temporary) != expected:
raise ValueError("Copied file failed its checksum: " + str(target))
os.replace(temporary, target)
finally:
if temporary.exists():
temporary.unlink()
def load_package(package):
manifest = json.loads((Path(package) / "manifest.json").read_text(encoding="utf-8"))
paths = [record["path"] for record in manifest["files"]]
if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", manifest["id"]):
raise ValueError("Invalid patch ID")
if len(set(paths)) != len(paths) or not paths:
raise ValueError("Patch manifest has duplicate paths or no files")
for record in manifest["files"]:
inside(Path(package) / "payload", record["path"])
if not record["path"].replace("\\", "/").startswith("TomieWGM_Data/"):
raise ValueError("Patch may only modify TomieWGM_Data resources")
return manifest
@contextmanager
def patch_lock(game):
path = inside(game, ".local-translator/operation.lock")
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a+b") as stream:
acquired = False
try:
if os.name == "nt":
import msvcrt
if stream.tell() == 0:
stream.write(b"0")
stream.flush()
stream.seek(0)
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
else:
import fcntl
fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
acquired = True
yield
except OSError as exc:
if not acquired:
raise RuntimeError("Другой установщик уже работает с этой игрой.") from exc
raise
finally:
if acquired:
if os.name == "nt":
stream.seek(0)
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
def install(package, game, running_check=game_running):
if running_check():
raise RuntimeError("Закрой игру перед установкой перевода.")
with patch_lock(game):
return _install(package, game, running_check)
def _install(package, game, running_check):
package, game = Path(package).resolve(), Path(game).resolve()
if running_check():
raise RuntimeError("Закрой игру перед установкой перевода.")
manifest = load_package(package)
if not (game / "TomieWGM.exe").is_file():
raise ValueError("Game executable not found in selected directory")
state_path = inside(game, STATE)
if state_path.exists():
old = json.loads(state_path.read_text(encoding="utf-8"))
if old["phase"] == "installed" and old["patch_id"] == manifest["id"]:
if all(digest(inside(game, record["path"])) == record["after"] for record in old["files"]):
print("Этот перевод уже установлен.")
return
if old["phase"] != "restored":
raise RuntimeError("Сначала запусти откат предыдущего перевода.")
# Validate the entire set before writing any game file.
for record in manifest["files"]:
if digest(inside(package / "payload", record["path"])) != record["after"]:
raise ValueError("Patch payload checksum mismatch")
if digest(inside(game, record["path"])) != record["before"]:
raise ValueError("Game file differs from the supported original: " + record["path"])
records = []
for record in manifest["files"]:
backup_relative = Path(".local-translator/backups") / manifest["id"] / record["path"]
backup = inside(game, backup_relative)
if backup.exists() and digest(backup) != record["before"]:
raise ValueError("An existing backup has a different checksum")
if not backup.exists():
atomic_copy(inside(game, record["path"]), backup, record["before"])
records.append({**record, "backup": backup_relative.as_posix()})
state = {"phase": "installing", "patch_id": manifest["id"], "files": records}
write_json(state_path, state)
try:
for record in records:
# Recheck each original immediately before replacing it.
if digest(inside(game, record["path"])) != record["before"]:
raise ValueError("Game file changed during installation")
atomic_copy(inside(package / "payload", record["path"]), inside(game, record["path"]), record["after"])
except BaseException:
_restore(game, running_check=lambda: False)
raise
state["phase"] = "installed"
write_json(state_path, state)
print("Перевод установлен. Запусти игру обычным способом. Оригиналы сохранены для отката.")
def restore(game, running_check=game_running):
if running_check():
raise RuntimeError("Закрой игру перед откатом.")
with patch_lock(game):
return _restore(game, running_check)
def _restore(game, running_check):
game = Path(game).resolve()
if running_check():
raise RuntimeError("Закрой игру перед откатом.")
state_path = inside(game, STATE)
if not state_path.exists():
print("Установленного патча нет; файлы игры не менялись.")
return
state = json.loads(state_path.read_text(encoding="utf-8"))
for record in state["files"]:
if not record["path"].replace("\\", "/").startswith("TomieWGM_Data/") or not record["backup"].replace("\\", "/").startswith(".local-translator/backups/"):
raise ValueError("Unsafe resource or backup path in installation journal")
if digest(inside(game, record["backup"])) != record["before"]:
raise ValueError("Backup checksum mismatch; refusing restore")
current = digest(inside(game, record["path"]))
if current not in {record["before"], record["after"]}:
raise ValueError("Game file was modified by another tool; refusing to overwrite it")
state["phase"] = "restoring"
write_json(state_path, state)
for record in state["files"]:
atomic_copy(inside(game, record["backup"]), inside(game, record["path"]), record["before"])
state["phase"] = "restored"
write_json(state_path, state)
print("Оригинальные файлы восстановлены. Резервные копии сохранены.")
def snapshot(environment):
return {(obj.assets_file.name, obj.path_id): hashlib.sha256(obj.get_raw_data()).hexdigest() for obj in environment.objects}
def ui_text_template(environment):
for obj in environment.objects:
if obj.type.name != "MonoBehaviour":
continue
data = obj.read(check_read=False)
try:
script = data.m_Script.read()
except (ValueError, AttributeError, FileNotFoundError):
continue
if script.m_Namespace == "UnityEngine.UI" and script.m_ClassName == "Text":
return obj.serialized_type.node
raise ValueError("Unity UI Text schema not found in bundle")
def ui_text_components(environment, node):
for obj in environment.objects:
if obj.type.name != "MonoBehaviour":
continue
try:
script = obj.read(check_read=False).m_Script.read()
except (ValueError, AttributeError, FileNotFoundError):
continue
if script.m_Namespace == "UnityEngine.UI" and script.m_ClassName == "Text":
# A known Text with an incompatible layout must fail the build,
# rather than silently disappear from the translation coverage.
yield obj, obj.read_typetree(node)
def original_sources(game, inputs):
sources = {path: path for path in inputs}
live_hashes = {path: digest(path) for path in inputs}
state_path = game / STATE
if state_path.exists():
state = json.loads(state_path.read_text(encoding="utf-8"))
if state["phase"] == "installed":
records = {record["path"]: record for record in state["files"]}
for path in inputs:
record = records[path.relative_to(game).as_posix()]
backup = inside(game, record["backup"])
if live_hashes[path] != record["after"] or digest(backup) != record["before"]:
raise ValueError("Installed files or original backups changed; refusing rebuild")
sources[path] = backup
print("Для новой сборки использую проверенные резервные копии оригиналов.", flush=True)
elif state["phase"] != "restored":
raise RuntimeError("Сначала заверши откат через scripts/windows/tomie/restore_tomie_original.bat")
return sources, live_hashes
def font_fixes(environment, required):
from fontTools.ttLib import TTFont
fonts = [obj for obj in environment.objects if obj.type.name == "Font"]
donor = next(obj.read_typetree() for obj in fonts if obj.read().m_Name == "Candara")
changed = set()
for obj in fonts:
data = obj.read_typetree()
font = TTFont(io.BytesIO(bytes(data["m_FontData"])))
cmap = font.getBestCmap()
missing = [char for char in required if ord(char) not in cmap and char.isalpha()]
font.close()
if not missing:
continue
for key in ("m_FontData", "m_FontNames", "m_LineSpacing", "m_Ascent", "m_Descent", "m_PixelScale"):
data[key] = donor[key]
obj.save_typetree(data)
changed.add((obj.assets_file.name, obj.path_id))
return changed
def verify_bundle(path, before, allowed):
import UnityPy
environment = UnityPy.load(str(path))
after = snapshot(environment)
if before.keys() != after.keys():
raise ValueError("Unity object IDs changed while packing")
changed = {key for key in before if before[key] != after[key]}
if changed - allowed:
raise ValueError("Unrelated Unity objects changed while packing")
return {"objects_checked": len(before), "changed_objects": [list(key) for key in sorted(changed)],
"untouched_objects": len(before) - len(changed)}
def ensure_api(root, base):
from urllib.parse import urlsplit
url = urlsplit(base)
if url.scheme != "http" or url.hostname not in {"127.0.0.1", "localhost"} or url.username or url.password:
raise ValueError("This game adapter only uses a local translator API")
def ready():
try:
value = httpx.get(base + "/health", timeout=2, trust_env=False).json()
return value.get("worker_running") is True
except (httpx.HTTPError, ValueError):
return False
if ready():
return
print("Запускаю локальный API; ожидаю готовности...", flush=True)
log_dir = root / "data/tomie"
log_dir.mkdir(parents=True, exist_ok=True)
with (log_dir / "adapter-api.log").open("ab") as log:
subprocess.Popen([sys.executable, "-m", "local_translator", "--config", str(root / "config.toml")], cwd=root,
stdout=log, stderr=log, stdin=subprocess.DEVNULL,
creationflags=0x08000000 if os.name == "nt" else 0)
deadline = time.monotonic() + 120
while time.monotonic() < deadline:
if ready():
return
time.sleep(0.5)
raise RuntimeError("API не готов. Проверь scripts/windows/start.bat и data/tomie/adapter-api.log")
def prepare(config, root, result_path=None):
import UnityPy
from .benchmark import request
game = (root / config["game"]).resolve()
output_root = (root / config["output"]).resolve()
if output_root.is_relative_to(game):
raise ValueError("Build output must be outside the game directory")
package = output_root / (datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + uuid4().hex[:8])
package.mkdir(parents=True, exist_ok=False)
main_path = game / DATA / "data.unity3d"
bundle_path = next((game / DATA / "StreamingAssets/aa").rglob("naninovel_assets_all*.bundle"))
catalog_path = game / DATA / "StreamingAssets/aa/catalog.json"
inputs = [main_path, bundle_path, catalog_path]
sources, live_hashes = original_sources(game, inputs)
original_hashes = {str(path): digest(sources[path]) for path in inputs}
print("Читаю словарь и кнопки интерфейса...", flush=True)
main = UnityPy.load(str(sources[main_path]))
main_before = snapshot(main)
asset = next(obj for obj in main.objects if obj.type.name == "TextAsset" and obj.read().m_Name == "DefaultUI")
script = asset.read().m_Script
entries = parse_dictionary(script)
bundle = UnityPy.load(str(sources[bundle_path]))
bundle_before = snapshot(bundle)
text_node = ui_text_template(bundle)
source_values = {row["text"].casefold() for row in entries}
literals = set()
for obj, tree in list(ui_text_components(bundle, text_node)) + list(ui_text_components(main, text_node)):
text = tree.get("m_Text")
if isinstance(text, str) and text.strip().casefold() in source_values | MENU_LABELS:
literals.add(text.strip())
items = [{"id": row["id"], "text": row["text"]} for row in entries]
items += [{"id": f"literal.{index}", "text": text} for index, text in enumerate(sorted(literals))]
quality_objects = [obj for obj in main.objects if obj.type.name == "QualitySettings"]
quality_names = sorted({setting["name"] for obj in quality_objects for setting in obj.read_typetree()["m_QualitySettings"]})
items += [{"id": f"quality.{index}", "text": name} for index, name in enumerate(quality_names)]
write_json(package / "input.json", items)
project = Project.model_validate(json.loads((root / config["project_file"]).read_text(encoding="utf-8-sig")))
if result_path:
job = json.loads(Path(result_path).read_text(encoding="utf-8-sig"))
expected = {item["id"]: item["text"] for item in items}
rows = job.get("items", [])
if len(rows) != len(items) or {row["id"]: row["source"] for row in rows} != expected or any(row.get("status") != "completed" or not isinstance(row.get("translation"), str) for row in rows):
raise ValueError("Saved result does not match the current game's complete input dataset")
print("Использую указанный файл переводов без повторного запроса модели.", flush=True)
else:
base = config["api_base"].rstrip("/")
ensure_api(root, base)
with httpx.Client(base_url=base, timeout=40, trust_env=False) as client:
project_data = request(client, "POST", "/projects", project.model_dump())
job = request(client, "POST", f"/projects/{project_data['id']}/jobs", {"items": items, "batch_size": config.get("batch_size", 10)})
write_json(package / "job-reference.json", {"job_id": job["id"], "project_id": project_data["id"], "api_base": base})
deadline = time.monotonic() + 900
last_progress = None
while job["status"] in {"queued", "running"}:
if time.monotonic() > deadline:
raise RuntimeError("Перевод ещё не завершён; ID задания сохранён в job-reference.json")
if job["progress"] != last_progress:
print(f"Перевод: {job['progress']}", flush=True)
last_progress = job["progress"]
time.sleep(0.5)
job = request(client, "GET", "/jobs/" + job["id"])
write_json(package / "translations.json", job)
if job["status"] != "completed":
raise RuntimeError("Есть ошибки перевода. Патч не собран; подробности в " + str(package / "translations.json"))
translations = {row["id"]: row["translation"] for row in job["items"]}
for item in items:
source = protect(item["text"], project)
target = protect(translations[item["id"]], project)
if list(source.tokens.values()) != list(target.tokens.values()):
raise ValueError("Translation changed protected syntax: " + item["id"])
replacements = {row["id"]: translations[row["id"]] for row in entries}
updated_script = rewrite_dictionary(script, replacements)
quality_translations = {name: translations[f"quality.{index}"] for index, name in enumerate(quality_names)}
extra_records = {}
for obj in quality_objects:
for index, setting in enumerate(obj.read_typetree()["m_QualitySettings"]):
key = f"Naninovel.UI.GameSettingsGraphicsDropdown.GraphicOption{index + 1}"
if key not in replacements:
extra_records[key] = quality_translations[setting["name"]]
updated_script = append_managed_records(updated_script, extra_records)
data = asset.read()
data.m_Script = updated_script
data.save()
write_json(package / "ui-preview.json", {**replacements, **extra_records})
(package / "DefaultUI.ru.txt").write_text(updated_script, encoding="utf-8", newline="")
main_allowed = {(asset.assets_file.name, asset.path_id)}
required = set("".join(translations.values()))
main_allowed |= font_fixes(main, required)
literal_translations = {text: translations[f"literal.{index}"] for index, text in enumerate(sorted(literals))}
bundle_allowed = font_fixes(bundle, required)
button_count = 0
for environment, allowed in ((bundle, bundle_allowed), (main, main_allowed)):
for obj, tree in ui_text_components(environment, text_node):
text = tree.get("m_Text")
if isinstance(text, str) and text.strip() in literal_translations:
translated = literal_translations[text.strip()]
if text.strip() == translated:
continue
tree["m_Text"] = text.replace(text.strip(), translated, 1)
font_data = tree.get("m_FontData", {})
if "m_BestFit" in font_data:
font_data["m_BestFit"] = True
font_data["m_MinSize"] = min(10, font_data["m_FontSize"])
font_data["m_MaxSize"] = font_data["m_FontSize"]
obj.save_typetree(tree, nodes=text_node)
allowed.add((obj.assets_file.name, obj.path_id))
button_count += 1
print("Собираю ресурсы и проверяю неизменённые объекты...", flush=True)
verification = {}
for original, environment, before, allowed in ((main_path, main, main_before, main_allowed), (bundle_path, bundle, bundle_before, bundle_allowed)):
relative = original.relative_to(game)
target = inside(package / "payload", relative)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(next(iter(environment.files.values())).save(packer="lz4"))
verification[relative.as_posix()] = verify_bundle(target, before, allowed)
bundle_payload = inside(package / "payload", bundle_path.relative_to(game))
catalog = json.loads(sources[catalog_path].read_text(encoding="utf-8-sig"))
old_hash = re.search(r"_([a-f0-9]{32})\.bundle$", bundle_path.name)[1]
new_bundle_data = bundle_payload.read_bytes()
updated_catalog, old_options, new_options = patch_catalog(catalog, old_hash, bundle_crc(new_bundle_data), len(new_bundle_data), hashlib.md5(new_bundle_data).hexdigest())
if bundle_crc(sources[bundle_path].read_bytes()) != old_options["m_Crc"]:
raise ValueError("Original bundle CRC does not match the catalog")
write_json(inside(package / "payload", catalog_path.relative_to(game)), updated_catalog)
records = []
for original in inputs:
if digest(sources[original]) != original_hashes[str(original)] or digest(original) != live_hashes[original]:
raise ValueError("Original game changed while building the patch")
relative = original.relative_to(game).as_posix()
records.append({"path": relative, "before": original_hashes[str(original)], "after": digest(inside(package / "payload", relative))})
manifest = {"id": package.name, "game": str(game), "created_at": datetime.now(timezone.utc).isoformat(), "files": records,
"managed_text_entries": len(entries) + len(extra_records), "literal_buttons": button_count, "quality_names": len(extra_records), "verification": verification,
"addressables_before": old_options, "addressables_after": new_options,
"scope": "DefaultUI and selected static UI labels; no scenario/dialogue translation; no in-game verification"}
write_json(package / "manifest.json", manifest)
write_json(output_root / "latest.json", {"package": str(package)})
print(f"Готово: {package}\nЗаписей словаря: {len(entries)}; статических надписей: {button_count}.\nЗакрой игру и запусти scripts/windows/tomie/install_tomie_translation.bat.")
return package
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--config", type=Path, default=Path("tomie.json"))
parser.add_argument("action", choices=("prepare", "install", "restore", "status"))
parser.add_argument("--result", type=Path, help="Explicitly reuse a complete saved translations.json when rebuilding the same dataset")
args = parser.parse_args()
root = args.config.resolve().parent
config = json.loads(args.config.read_text(encoding="utf-8-sig"))
game = (root / config["game"]).resolve()
if args.action == "prepare":
prepare(config, root, args.result.resolve() if args.result else None)
elif args.action == "restore":
restore(game)
elif args.action == "status":
state = game / STATE
print(state.read_text(encoding="utf-8") if state.exists() else "Патч не установлен.")
else:
latest = json.loads((root / config["output"] / "latest.json").read_text(encoding="utf-8"))
install(Path(latest["package"]), game)
if __name__ == "__main__":
try:
main()
except (OSError, RuntimeError, ValueError, httpx.HTTPError, StopIteration, subprocess.SubprocessError) as exc:
raise SystemExit(str(exc)) from exc