layout: restructure into src/ tests/ android/ scripts/ tools/

- extension/src/{port,platform,codepage} -> src/; native_render -> src/host
  (+ dxt, shaders/); libgr2 -> src/gr2; extension/third_party -> third_party
- extension/tests -> tests/port, libgr2/tests -> tests/gr2
- android-native -> android (build.sh, push-client.sh moved in)
- script -> scripts; tools/40250 -> tools/server; oracle -> tools/granny-oracle;
  perf tools -> tools/perf
- all build trees under build/ (native, release, android, port-gate)
- xrender:: CMake aliases -> mt::; port-map ledger impl paths rewritten

No code changes. ctest 15/15, port_gate macos+android PASS, port_map check 0 errors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
shenlei
2026-09-29 19:08:19 +09:00
co-authored by Claude Opus 5.5
parent 70710477cf
commit a46093104c
2817 changed files with 13728 additions and 13744 deletions
+75
View File
@@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""Compare a read-only 40250 database name export with the older Chinese tables."""
import argparse
import json
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
DEFAULT_EXPORT = REPO / "build/zh-locale/server-audit"
DEFAULT_TABLES = REPO.parent / "server_extracted/40250_deploy"
def live_names(path: Path):
names = {}
for number, line in enumerate(path.read_text(encoding="ascii").splitlines(), 1):
try:
vnum, encoded = line.split("\t", 1)
name = bytes.fromhex(encoded).decode("gbk")
if int(vnum) in names:
raise ValueError("duplicate VNUM")
names[int(vnum)] = name
except (ValueError, UnicodeDecodeError) as error:
raise ValueError(f"{path}:{number}: {error}") from error
return names
def older_names(path: Path):
names = {}
for line in path.read_text(encoding="utf-8").splitlines()[1:]:
vnum, name = line.split("\t", 1)
names[int(vnum)] = name
return names
def audit(export_dir: Path, tables_dir: Path):
result = {}
for kind in ("item", "mob"):
live = live_names(export_dir / f"{kind}_names_hex.tsv")
older = older_names(tables_dir / f"{kind}_names_zh.txt")
shared = live.keys() & older.keys()
conflicts = [
{"vnum": vnum, "server": live[vnum], "older_table": older[vnum]}
for vnum in sorted(shared) if live[vnum] != older[vnum]
]
result[kind] = {
"server_rows": len(live),
"older_table_rows": len(older),
"shared_vnums": len(shared),
"server_only_vnums": len(live.keys() - older.keys()),
"older_only_vnums": len(older.keys() - live.keys()),
"server_names_without_hanzi": sum(
not any("\u4e00" <= char <= "\u9fff" for char in name)
for name in live.values()
),
"conflicts": conflicts,
}
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--export-dir", type=Path, default=DEFAULT_EXPORT)
parser.add_argument("--tables-dir", type=Path, default=DEFAULT_TABLES)
parser.add_argument("--output", type=Path, default=DEFAULT_EXPORT / "name-audit.json")
args = parser.parse_args()
report = audit(args.export_dir, args.tables_dir)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
for kind, data in report.items():
print(f"{kind}: rows={data['server_rows']} no_hanzi={data['server_names_without_hanzi']} "
f"conflicts={len(data['conflicts'])}")
if __name__ == "__main__":
main()
+230
View File
@@ -0,0 +1,230 @@
#!/usr/bin/env python3
"""Build a CP936 Chinese locale on top of the 40250 English client resources.
The Longju locale is an older client version. Only keyed UI text with compatible
format arguments is overlaid; 40250 proto, scripts, and UI layouts stay intact.
The files are staged in build/zh-locale/locale-src/zh and packed like the original
locale packs (eterpack_pack: COMPRESS entries, encrypted index) into locale_zh.eix/.epk.
The output Client has a pack/ of its own: links to the original packs, the zh pack and
an Index registering locale/zh/ -> locale_zh. No loose locale/zh is left next to it,
since the client searches loose files first.
"""
import argparse
import json
import re
import shutil
import subprocess
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
DEFAULT_EN = REPO.parent / "40250/Server Client TMP4/Client/Eternexus/locale_en/locale/en"
DEFAULT_ZH = REPO.parent / "server_extracted/client_locale/longju/locale/taiwan"
DEFAULT_OUTPUT = REPO / "build/zh-locale/Client"
BUILD_DIRS = ("build/release", "build/native")
LOCALE_CFG = "10002 936 zh\n"
DEFAULT_NAMES = REPO / "data/zh_40250"
FORMAT = re.compile(r"%(?:\([^)]*\))?[-+#0 ]*\d*(?:\.\d+)?[diouxXeEfFgGcrs%]")
KEYED_FILES = ("locale_game.txt", "locale_interface.txt")
def keyed_lines(path: Path, encoding: str):
lines = path.read_bytes().decode(encoding).splitlines()
values = {}
for line in lines:
parts = line.split("\t")
if len(parts) >= 2 and parts[0]:
values[parts[0]] = parts[1]
return lines, values
def name_rows(path: Path):
rows = {}
if path.is_file():
for line in path.read_text(encoding="utf-8").splitlines():
parts = line.split("\t", 1)
if len(parts) == 2 and parts[0].isdigit():
rows[parts[0]] = parts[1]
return rows
def convert_names(source: Path, destination: Path, overlay: Path = None):
"""The server name snapshot, with the reviewed translations of its English names laid over it."""
names = name_rows(source)
overlaid = {vnum: name for vnum, name in name_rows(overlay).items() if vnum in names} if overlay else {}
names.update(overlaid)
converted = 0
unrepresentable = []
rows = []
for vnum, name in names.items():
try:
rows.append(f"{vnum}\t{name}\n".encode("gbk"))
converted += 1
except UnicodeEncodeError:
unrepresentable.append(vnum)
destination.write_bytes(b"".join(rows))
return {"converted": converted, "overlaid": len(overlaid), "unrepresentable_vnums": unrepresentable}
def tool(name: str) -> Path:
for build_dir in BUILD_DIRS:
path = REPO / build_dir / "src/port" / name
if path.is_file():
return path
return REPO / BUILD_DIRS[0] / "src/port" / name
def register(index: bytes) -> bytes:
"""pack/Index with locale/zh/ -> locale_zh added before the locale/en/ pair (CRLF, as the original)."""
lines = index.split(b"\r\n")
if b"locale/zh/" in lines:
return index
at = lines.index(b"locale/en/")
if at % 2 != 1:
raise ValueError("pack/Index: locale/en/ is not a folder line")
lines[at:at] = [b"locale/zh/", b"locale_zh"]
return b"\r\n".join(lines)
def make_pack_dir(client_base: Path, pack_dir: Path, locale_dir: Path, pack_tool: Path):
source = client_base / "pack"
if pack_dir.is_symlink():
pack_dir.unlink() # earlier builds linked the whole directory
pack_dir.mkdir(parents=True, exist_ok=True)
for entry in source.iterdir():
if entry.name == "Index" or entry.name.lower().startswith("locale_zh."):
continue
target = pack_dir / entry.name
if target.is_symlink() or target.exists():
target.unlink()
target.symlink_to(entry.resolve())
(pack_dir / "Index").write_bytes(register((source / "Index").read_bytes()))
result = subprocess.run([str(pack_tool), str(pack_dir / "locale_zh"), str(locale_dir), "locale/zh/"],
check=True, capture_output=True, text=True)
return result.stdout.strip()
def build(en: Path, zh: Path, output: Path, client_base: Path,
proto_tool: Path = None, names_dir: Path = None, pack_tool: Path = None):
"""names_dir holds the server name snapshot and the reviewed *_zh_overlay.txt translations."""
if not (en / "locale_game.txt").is_file() or not (zh / "locale_game.txt").is_file():
raise ValueError("English 40250 and Chinese Longju locale directories are required")
if pack_tool is None or not pack_tool.is_file():
raise ValueError("build eterpack_pack (cmake --build <build dir> --target eterpack_pack)")
locale_dir = output.parent / "locale-src/zh"
if locale_dir.exists():
shutil.rmtree(locale_dir)
shutil.copytree(en, locale_dir)
loose = output / "locale/zh"
if loose.exists():
shutil.rmtree(loose) # earlier builds ran the locale from loose files
if not any((output / "locale").iterdir()):
(output / "locale").rmdir()
report = {"english_source": str(en), "chinese_source": str(zh), "files": {}}
for filename in KEYED_FILES:
english_lines, english = keyed_lines(en / filename, "cp1252")
_, chinese = keyed_lines(zh / filename, "gbk")
# Reviewed translations of the keys Longju lacks or formats differently; they win over Longju.
overlay = names_dir / (Path(filename).stem + "_zh_overlay.txt") if names_dir else None
if overlay and overlay.is_file():
_, reviewed = keyed_lines(overlay, "utf-8")
chinese.update(reviewed)
translated = []
incompatible = []
merged = []
for line in english_lines:
parts = line.split("\t")
if len(parts) >= 2 and parts[0] in chinese:
key = parts[0]
candidate = chinese[key]
if FORMAT.findall(parts[1]) == FORMAT.findall(candidate):
parts[1] = candidate
translated.append(key)
else:
incompatible.append(key)
line = "\t".join(parts)
merged.append(line)
(locale_dir / filename).write_bytes(("\n".join(merged) + "\n").encode("gbk"))
report["files"][filename] = {
"reference_keys": len(english),
"translated_keys": len(translated),
"english_fallback_keys": sorted(english.keys() - set(translated)),
"incompatible_format_keys": sorted(incompatible),
"chinese_only_keys": sorted(chinese.keys() - english.keys()),
}
# These files are prose, not versioned game data. Keep the 40250 UI layouts
# and prototype binaries copied above; only replace their explanatory text.
for filename in (
"jobdesc_warrior.txt", "jobdesc_assassin.txt", "jobdesc_sura.txt",
"jobdesc_shaman.txt", "empiredesc_a.txt", "empiredesc_b.txt", "empiredesc_c.txt",
):
if (zh / filename).is_file():
shutil.copyfile(zh / filename, locale_dir / filename)
if proto_tool is not None:
if not proto_tool.is_file() or names_dir is None:
raise ValueError("build locale_proto_patch and provide the Chinese name tables")
item_names = output.parent / "item_names_gbk.txt"
mob_names = output.parent / "mob_names_gbk.txt"
item_conversion = convert_names(names_dir / "item_names_zh.txt", item_names,
names_dir / "item_names_zh_overlay.txt")
mob_conversion = convert_names(names_dir / "mob_names_zh.txt", mob_names,
names_dir / "mob_names_zh_overlay.txt")
result = subprocess.run(
[str(proto_tool), str(en / "item_proto"), str(en / "mob_proto"),
str(item_names), str(mob_names), str(locale_dir / "item_proto"),
str(locale_dir / "mob_proto")],
check=True, capture_output=True, text=True,
)
report["proto"] = {
"item_name_source": item_conversion,
"mob_name_source": mob_conversion,
"patch_result": result.stdout.strip().splitlines(),
}
report["pack"] = make_pack_dir(client_base, output / "pack", locale_dir, pack_tool)
(output / "locale.cfg").write_text(LOCALE_CFG, encoding="ascii")
(output / "locale_zh.cfg").write_text(LOCALE_CFG, encoding="ascii")
# A local test checkout shares the original 40250 assets through symbolic links
# (pack/ holds one per original pack); android/push-client.sh copies their targets.
for name in ("sound", "BGM", "miles", "lib", "channel.inf"):
source = client_base / name
target = output / name
if not source.exists():
continue
if target.is_symlink():
if target.resolve() == source.resolve():
continue
target.unlink()
elif target.exists():
raise ValueError(f"refusing to replace existing client resource: {target}")
target.symlink_to(source.resolve())
report_path = output.parent / "zh-locale-report.json"
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return report_path
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--english", type=Path, default=DEFAULT_EN)
parser.add_argument("--chinese", type=Path, default=DEFAULT_ZH)
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
parser.add_argument("--client-base", type=Path, default=REPO / "Client")
parser.add_argument("--proto-tool", type=Path, default=tool("locale_proto_patch"))
parser.add_argument("--pack-tool", type=Path, default=tool("eterpack_pack"))
parser.add_argument("--name-tables", type=Path, default=DEFAULT_NAMES)
args = parser.parse_args()
report_path = build(args.english, args.chinese, args.output, args.client_base,
args.proto_tool, args.name_tables, args.pack_tool)
print(f"Built {args.output / 'pack/locale_zh.eix'} (.epk), registered in pack/Index")
print(f"Report: {report_path}")
if __name__ == "__main__":
main()
+133
View File
@@ -0,0 +1,133 @@
#!/usr/bin/env python3
"""Build the 40250 newcibn locale_string.txt from its Korean keys and GBK text."""
import argparse
import hashlib
import json
import re
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
DEFAULT_ENGLISH = REPO.parent / "40250/Server Client TMP4/Server/metin2/server/share/locale/english/locale_string.txt"
DEFAULT_CHINESE = REPO.parent / "server_extracted/server_locale/locale_string.txt"
DEFAULT_OUTPUT = REPO / "build/zh-locale/server/newcibn/locale_string.txt"
QUOTED = re.compile(rb'^"(.*)"\s*;$')
FORMAT = re.compile(rb'%(?:\d+\$)?[-+#0 ]*\d*(?:\.\d+)?(?:hh|ll|[hlLzjt])?[diouxXeEfFgGaAcspD%]')
# These keys exist in 40250 but not in the older newcibn dictionary.
ADDED = {
"등급 개량에 성공했습니다.": "品级提升成功。",
"%s이 필요합니다.": "需要%s。",
"용심 추출에 성공하였습니다.": "龙心提取成功。",
"용심 추출에 실패하였습니다.": "龙心提取失败。",
"단계 개량에 필요한 재료가 아닙니다.": "这不是提升阶段所需的材料。",
"이미 장비를 착용하고 있습니다.": "已经装备了同类物品。",
"등급 개량할 수 없는 용혼석입니다.": "该龙魂石无法提升品级。",
"강화에 성공했습니다.": "强化成功。",
"<길드> 잠시후에 이용해주십시오.": "<公会>请稍后再试。",
"%d 분 %d 초 동안 채팅금지 상태입니다": "聊天禁言还剩%d分%d秒。",
"이미 같은 종류의 용혼석을 착용하고 있습니다.": "已经装备了同类龙魂石。",
"인벤이 활성화 되지 않음.": "龙魂石背包尚未开启,请先完成龙魂石任务。",
"단계 개량할 수 없는 용혼석입니다.": "该龙魂石无法提升阶段。",
"벨트 인벤토리에 아이템이 존재하면 해제할 수 없습니다.": "腰带背包中还有物品,无法卸下腰带。",
"강화창을 연 상태에서는 아이템을 옮길 수 없습니다.": "强化窗口打开时无法移动物品。",
"먼저 용혼석 퀘스트를 완료하셔야 합니다.": "请先完成龙魂石任务。",
"단계 개량에 실패했습니다.": "阶段提升失败。",
"강화에 실패했습니다.": "强化失败。",
"용혼석 추출에 실패하여 %s를 얻었습니다.": "龙魂石提取失败,但获得了%s。",
"다른 거래중(창고,교환,상점)에는 사용할 수 없습니다.": "使用仓库、交易或商店时无法进行此操作。",
"강화할 수 없는 용혼석입니다.": "该龙魂石无法强化。",
"용혼석 추출에 실패하였습니다.": "龙魂石提取失败。",
"%d 시간 %d 분 %d 초 동안 채팅금지 상태입니다": "聊天禁言还剩%d小时%d分%d秒。",
"이미 탈것을 이용중입니다.": "你已经骑乘坐骑,请先下坐骑。",
"<길드> %u의 경험치를 투자하였습니다.": "<公会>投入了%u点经验值。",
"이 아이템은 벨트 인벤토리로 옮길 수 없습니다.": "此物品无法放入腰带背包。",
"%d 초 동안 채팅금지 상태입니다": "聊天禁言还剩%d秒。",
"등급 개량에 필요한 재료가 아닙니다.": "这不是提升品级所需的材料。",
"용혼석 추출에 성공하였습니다.": "龙魂石提取成功。",
"%d 시간 %d 초 동안 채팅금지 상태입니다": "聊天禁言还剩%d小时%d秒。",
"Pregunta.": "问题。",
"착용 중인 용혼석은 추출할 수 없습니다.": "无法提取已装备的龙魂石。",
"등급 개량에 실패했습니다.": "品级提升失败。",
"외치기는 유료회원만 사용할수 있습니다": "当前无法使用世界喊话。",
"강화에 필요한 재료가 아닙니다.": "这不是强化所需的材料。",
"단계 개량에 성공했습니다.": "阶段提升成功。",
}
# These two live entries lose a printf argument. Keep their keys and restore both fields.
CORRECTED = {
"OX 상품을 지급하였습니다. ID: %d 수량: %d": "已发放OX奖品,编号:%d,数量:%d。",
"곡괭이의 수련도가 증가하였습니다! (%d/%d)": "矿镐熟练度提高了!(%d/%d)",
}
def read_pairs(path: Path):
quoted = []
for number, line in enumerate(path.read_bytes().splitlines(), 1):
line = line.strip()
if not line:
continue
match = QUOTED.fullmatch(line)
if match is None:
raise ValueError(f"{path}:{number}: malformed locale entry")
quoted.append(match.group(1))
if len(quoted) % 2:
raise ValueError(f"{path}: unmatched locale key")
pairs = dict(zip(quoted[::2], quoted[1::2]))
if len(pairs) != len(quoted) // 2:
raise ValueError(f"{path}: duplicate locale key")
return pairs
def format_args(value: bytes):
return [token for token in FORMAT.findall(value) if token != b"%%"]
def build(english: Path, chinese: Path, output: Path):
reference = read_pairs(english)
existing = read_pairs(chinese)
missing = reference.keys() - existing.keys()
added_keys = {key.encode("cp949") for key in ADDED}
if missing != added_keys:
raise ValueError(f"40250 key coverage changed: missing={len(missing)}, translations={len(added_keys)}")
result = []
for key in list(reference) + [key for key in existing if key not in reference]:
source_key = key.decode("cp949")
if source_key in CORRECTED:
value = CORRECTED[source_key].encode("gbk")
elif key in existing:
value = existing[key]
else:
value = ADDED[source_key].encode("gbk")
if format_args(key) != format_args(value):
raise ValueError(f"format arguments differ for {source_key!r}")
if b'"' in value or b'\n' in value or b'\r' in value:
raise ValueError(f"unsafe locale value for {source_key!r}")
result.extend((b'"' + key + b'";', b'"' + value + b'";', b''))
payload = b"\n".join(result) + b"\n"
output.parent.mkdir(parents=True, exist_ok=True)
output.write_bytes(payload)
report = {
"reference_keys": len(reference),
"existing_chinese_keys": len(existing),
"new_chinese_keys": len(missing),
"corrected_format_keys": len(CORRECTED),
"legacy_extra_keys": len(existing.keys() - reference.keys()),
"output_sha256": hashlib.sha256(payload).hexdigest(),
}
output.with_suffix(".report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
return report
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--english", type=Path, default=DEFAULT_ENGLISH)
parser.add_argument("--chinese", type=Path, default=DEFAULT_CHINESE)
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
args = parser.parse_args()
print(json.dumps(build(args.english, args.chinese, args.output), ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
+227
View File
@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""Generate src/codepage/codepage_tables.inc from Microsoft's WindowsBestFit tables.
The port emulates MultiByteToWideChar / WideCharToMultiByte off Windows. Instead of the host iconv
(Android's bionic has no GBK, and iconv tables differ from Windows in the corners), the conversions
use Microsoft's own code page definitions, published by unicode.org:
https://www.unicode.org/Public/MAPPINGS/VENDORS/MICSFT/WindowsBestFit/bestfit<cp>.txt
Each file has an MBTABLE (single bytes), DBCSRANGE/DBCSTABLE (lead byte + trail byte) and a WCTABLE
(Unicode -> bytes, including the many-to-one "best fit" entries WideCharToMultiByte uses without
WC_NO_BEST_FIT_CHARS). The generated data keeps the byte -> Unicode side and only the WCTABLE entries
that differ from its first-match inverse, so the runtime can rebuild the encoder.
Usage: python3 scripts/gen_codepage_tables.py [--cache DIR] [--output FILE]
Downloads missing files into the cache (build/bestfit by default) and checks their SHA-256.
"""
import argparse
import hashlib
import re
import sys
import urllib.request
from pathlib import Path
REPO = Path(__file__).resolve().parent.parent
URL = "https://www.unicode.org/Public/MAPPINGS/VENDORS/MICSFT/WindowsBestFit/bestfit{}.txt"
# Every code page 40250 EterLib/Util.cpp names (1252 included; CP_ACP maps to it).
SHA256 = {
874: "663f43ca662e037c4534cb16298b560f29ce29c27b49b3589601ec3d97dd89fd",
932: "2614cfea35c3c86c41d33198793a84ca44edee3cf0ee0013a61a43fba4ece331",
936: "e5070a2d6ad26619f5872ddbe64d3381c11620af5adbb04cda0f0abb1a91fdae",
949: "50e13b60ea8fda66a8223ecc85270e0f182303222244e2345d3d57f3e839d20a",
950: "cf8c23389a42a226ea707f7ec32c665556d1fc3364db25bd765ce64d54eaee2a",
1250: "cef9f171e67b09445bcb3f9ffccdc89418250ff825f1bd2d29a92d2074d7a53b",
1251: "59ec85612ff908d9da0e877893c935941e56b13a2882b4fb9c9599be3d1ce4e7",
1252: "72ea23c939c5b26fae7aded0207b327e2f3902d7d3c168d7087f5cfc38ee76a9",
1253: "ea80c442aff7f09b36da6335f85f8e527f51c146beeb9825ec00d1b6ca99a99e",
1254: "3d02512087634dc493b720992b590277736ffb2d5b0b665d69b6b9727e2c361a",
1255: "fdd4bdda74f6571d89171b0070ac052cd3714c395dc3d1799bcd5e4a4da6f83a",
1256: "745c447ada04a838da8bea406c13f446c7453b6371e8c6c7863a632443d56007",
1257: "b8c5d7f3b8c25c3d5625d44dd3d6ee7a06e652ddf77373d050282c1cb7517366",
1258: "5d52a9357b7d6b5b5014ed5a51be0ff9809b0c33625793d2a4feaf502e0682f1",
}
NONE = 0xFFFF
PAIR = re.compile(rb"^\s*0x([0-9a-fA-F]+)\s+0x([0-9a-fA-F]+)")
def fetch(cp: int, cache: Path) -> bytes:
path = cache / f"bestfit{cp}.txt"
if not path.is_file():
cache.mkdir(parents=True, exist_ok=True)
with urllib.request.urlopen(URL.format(cp)) as response:
path.write_bytes(response.read())
data = path.read_bytes()
digest = hashlib.sha256(data).hexdigest()
if digest != SHA256[cp]:
raise SystemExit(f"{path}: SHA-256 {digest} != {SHA256[cp]}")
return data
def parse(cp: int, data: bytes):
single = [NONE] * 256
leads = [] # [(lo, hi)]
dbcs = {} # lead -> {trail: wc}
wctable = {} # wc -> mb
mb_default = None
section = None
lead = None
for raw in data.splitlines():
line = raw.split(b";", 1)[0].strip()
if not line:
continue
head = line.split()[0]
if head == b"CODEPAGE":
if int(line.split()[1]) != cp:
raise SystemExit(f"bestfit{cp}.txt declares {line!r}")
continue
if head == b"CPINFO":
fields = line.split()
if int(fields[2], 16) != 0x3F:
raise SystemExit(f"cp{cp}: byte default char {fields[2]!r} is not '?'")
mb_default = int(fields[3], 16)
continue
if head in (b"MBTABLE", b"WCTABLE", b"DBCSRANGE"):
section = head
continue
if head == b"DBCSTABLE":
section = head
match = re.search(rb"LeadByte\s*=\s*0x([0-9a-fA-F]+)", raw)
lead = int(match.group(1), 16)
dbcs.setdefault(lead, {})
continue
if head == b"ENDCODEPAGE":
break
match = PAIR.match(line)
if not match:
raise SystemExit(f"cp{cp}: unexpected line {raw!r}")
a, b = int(match.group(1), 16), int(match.group(2), 16)
# cp932 lists its second lead range (0xe0-0xfc) after the first range's DBCSTABLEs.
if b"Lead Byte Range" in raw and section in (b"DBCSRANGE", b"DBCSTABLE"):
leads.append((a, b))
continue
if section == b"MBTABLE":
single[a] = b
elif section == b"DBCSRANGE":
leads.append((a, b))
elif section == b"DBCSTABLE":
dbcs[lead][a] = b
elif section == b"WCTABLE":
wctable[a] = b
if mb_default is None:
raise SystemExit(f"cp{cp}: no CPINFO")
for lead in dbcs:
if not any(lo <= lead <= hi for lo, hi in leads):
raise SystemExit(f"cp{cp}: DBCSTABLE 0x{lead:02x} outside the lead ranges {leads}")
return single, leads, dbcs, wctable, mb_default
def decode_map(single, dbcs):
"""bytes -> wc, bytes as int (single byte < 0x100, pair = lead << 8 | trail), ascending."""
out = {}
for b, wc in enumerate(single):
if wc != NONE:
out[b] = wc
for lead in sorted(dbcs):
for trail in sorted(dbcs[lead]):
out[lead << 8 | trail] = dbcs[lead][trail]
return out
def default_inverse(decoded):
"""The encoder the runtime builds: the first byte sequence (ascending) that decodes to wc."""
inverse = {}
for mb in sorted(decoded):
inverse.setdefault(decoded[mb], mb)
return inverse
def encode_overrides(decoded, wctable):
"""WCTABLE entries the default inverse gets wrong: 0 round-trip, 1 best fit, 2 not encodable."""
inverse = default_inverse(decoded)
out = []
for wc in sorted(set(inverse) | set(wctable)):
want = wctable.get(wc)
if want is None:
out.append((wc, 0, 2))
elif decoded.get(want) != wc:
out.append((wc, want, 1))
elif inverse.get(wc) != want:
out.append((wc, want, 0))
return out
def rows(values, per_line=16):
return "\n".join("\t" + ", ".join(values[i:i + per_line]) + "," for i in range(0, len(values), per_line))
def emit(cp, single, leads, dbcs, wctable, mb_default):
decoded = decode_map(single, dbcs)
overrides = encode_overrides(decoded, wctable)
# Sanity: the rebuilt encoder must reproduce WCTABLE exactly.
inverse = default_inverse(decoded)
for wc, mb, kind in overrides:
if kind == 2:
inverse.pop(wc, None)
else:
inverse[wc] = mb
if inverse != wctable:
raise SystemExit(f"cp{cp}: encoder rebuild does not match WCTABLE")
name = f"kCp{cp}"
text = [f"// Code page {cp}: {len(decoded)} byte sequences, {len(wctable)} WCTABLE entries."]
text.append(f"const uint16_t {name}Single[256] = {{\n{rows([f'0x{v:04X}' for v in single])}\n}};")
lead_ranges = "nullptr"
dbcs_rows = "nullptr"
dbcs_data = "nullptr"
if leads:
lead_ranges = f"{name}Leads"
text.append(f"const uint8_t {name}Leads[] = {{" + ", ".join(f"0x{lo:02X}, 0x{hi:02X}" for lo, hi in leads) + "};")
row_defs = []
data = []
for lead in sorted(dbcs):
trails = dbcs[lead]
lo, hi = min(trails), max(trails)
row_defs.append(f"\t{{0x{lead:02X}, 0x{lo:02X}, 0x{hi:02X}, {len(data)}}},")
data.extend(trails.get(t, NONE) for t in range(lo, hi + 1))
dbcs_rows = f"{name}Rows"
dbcs_data = f"{name}Dbcs"
text.append(f"const DbcsRow {name}Rows[] = {{\n" + "\n".join(row_defs) + "\n};")
text.append(f"const uint16_t {name}Dbcs[] = {{\n{rows([f'0x{v:04X}' for v in data])}\n}};")
text.append(f"const EncodeOverride {name}Overrides[] = {{\n"
+ rows([f"{{0x{wc:04X}, 0x{mb:04X}, {kind}}}" for wc, mb, kind in overrides], 6)
+ "\n};" if overrides else f"const EncodeOverride {name}Overrides[] = {{{{0, 0, 2}}}};")
count = f"sizeof({name}Overrides) / sizeof({name}Overrides[0])" if overrides else "0"
entry = (f"\t{{{cp}, 0x{mb_default:04X}, {name}Single, {lead_ranges}, "
f"{len(leads)}, {dbcs_rows}, {len(dbcs) if leads else 0}, {dbcs_data}, "
f"{name}Overrides, {count}}},")
return "\n".join(text) + "\n", entry, len(overrides)
def main():
parser = argparse.ArgumentParser(description=__doc__.split("\n")[0])
parser.add_argument("--cache", type=Path, default=REPO / "build/bestfit")
parser.add_argument("--output", type=Path, default=REPO / "src/codepage/codepage_tables.inc")
args = parser.parse_args()
body = []
entries = []
for cp in sorted(SHA256):
parsed = parse(cp, fetch(cp, args.cache))
text, entry, n_over = emit(cp, *parsed)
body.append(text)
entries.append(entry)
print(f"cp{cp}: {n_over} encode overrides", file=sys.stderr)
header = ("// Generated by scripts/gen_codepage_tables.py from Microsoft's WindowsBestFit tables\n"
"// (unicode.org MAPPINGS/VENDORS/MICSFT/WindowsBestFit). Do not edit.\n"
"// 0xFFFF marks a byte / pair with no mapping. Included by codepage.cpp only.\n")
table = "const CodePageTable kTables[] = {\n" + "\n".join(entries) + "\n};\n"
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(header + "\n" + "\n".join(body) + "\n" + table, encoding="ascii")
print(f"wrote {args.output}", file=sys.stderr)
if __name__ == "__main__":
main()
+205
View File
@@ -0,0 +1,205 @@
#!/usr/bin/env node
// Run the macOS native client with a real server or the local fake server.
// A real-server run remains unverified until a person checks the in-game flows.
import fs from 'node:fs';
import path from 'node:path';
import net from 'node:net';
import { spawn, execFile } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import { redact, secretLiterals } from './redact_stream.mjs';
const repo = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
const defaults = {
binary: path.join(repo, 'build/release/src/host/mt_native_render'),
client: path.join(repo, 'Client'),
output: path.join(repo, 'build/native-acceptance', new Date().toISOString().replaceAll(':', '-')),
};
function usage() {
console.log(`Usage: node scripts/native_mac_acceptance.mjs (--server HOST:AUTH_PORT:GAME_PORT | --fake) [options]
--binary FILE Native live-client binary (default: build/release/src/host/mt_native_render)
--client DIR 40250 Client directory (default: Client)
--output DIR New evidence directory (default: build/native-acceptance/<timestamp>)
--min-seconds N Minimum real-server observation time (default: 1800)
--frames N Fake-server smoke frames (default: 180; fake mode only)
--help Show this help
Real-server mode opens the login screen. Enter credentials in the app, exercise
the checklist in the report, then close the window. Credentials are never
passed on the command line. The report requires manual review before PASS.`);
}
function optionsOf(argv) {
const options = { ...defaults, minSeconds: 1800, frames: 180 };
for (let i = 0; i < argv.length; i += 1) {
const key = argv[i];
if (key === '--help') { options.help = true; continue; }
if (!['--server', '--fake', '--binary', '--client', '--output', '--min-seconds', '--frames'].includes(key))
throw new Error(`Unknown option: ${key}`);
if (key === '--fake') { options.fake = true; continue; }
if (++i >= argv.length) throw new Error(`Missing value for ${key}`);
const field = { '--server': 'server', '--binary': 'binary', '--client': 'client',
'--output': 'output', '--min-seconds': 'minSeconds', '--frames': 'frames' }[key];
options[field] = argv[i];
}
if (!options.help) {
if (Boolean(options.server) === Boolean(options.fake)) throw new Error('Choose exactly one of --server or --fake');
for (const field of ['minSeconds', 'frames']) {
options[field] = Number(options[field]);
if (!Number.isInteger(options[field]) || options[field] < (field === 'frames' ? 1 : 0))
throw new Error(`Invalid ${field}`);
}
if (options.server && argv.includes('--frames')) throw new Error('--frames is only for fake-server smoke');
}
return options;
}
function parseServer(spec) {
const match = /^([^:\s]+):([0-9]+):([0-9]+)$/.exec(spec);
if (!match || [match[2], match[3]].some((value) => +value < 1 || +value > 65535))
throw new Error('Server must be HOST:AUTH_PORT:GAME_PORT (one shared host)');
return { host: match[1], authPort: +match[2], gamePort: +match[3] };
}
function reachable(host, port) {
return new Promise((resolve) => {
const socket = net.createConnection({ host, port });
socket.setTimeout(3000);
socket.once('connect', () => { socket.destroy(); resolve(true); });
socket.once('error', () => { socket.destroy(); resolve(false); });
socket.once('timeout', () => { socket.destroy(); resolve(false); });
});
}
function readRss(pid) {
return new Promise((resolve) => execFile('ps', ['-o', 'stat=,rss=', '-p', String(pid)],
{ timeout: 3000 }, (error, stdout) => {
if (error) return resolve(null);
const [state, value] = stdout.trim().split(/\s+/);
const rss = Number(value);
resolve(state?.startsWith('Z') || !Number.isInteger(rss) || rss <= 0 ? null : rss);
}));
}
function parseSummary(log) {
const line = log.trim().split('\n').reverse().find((row) => row.startsWith('device=') && row.includes(' frames='));
if (!line) return null;
const values = {};
for (const match of line.matchAll(/(?:^| )([a-z0-9_]+)=([^ ]+)/g)) {
const number = Number(match[2]);
values[match[1]] = Number.isFinite(number) ? number : match[2];
}
return values;
}
async function main() {
let options;
try { options = optionsOf(process.argv.slice(2)); }
catch (error) { console.error(error.message); usage(); return 2; }
if (options.help) { usage(); return 0; }
if (process.platform !== 'darwin') { console.error('This acceptance runner targets macOS'); return 2; }
if (!fs.existsSync(options.binary) || !fs.existsSync(path.join(path.dirname(options.binary), 'python27.zip'))
|| !fs.existsSync(path.join(options.client, 'pack/Index'))) {
console.error('Missing native live-client binary, adjacent python27.zip, or Client/pack/Index');
return 2;
}
if (fs.existsSync(options.output)) { console.error(`Output already exists: ${options.output}`); return 2; }
let server = null;
if (options.server) {
try { server = parseServer(options.server); }
catch (error) { console.error(error.message); return 2; }
for (const port of [server.authPort, server.gamePort]) {
if (!await reachable(server.host, port)) {
console.error(`Server endpoint unreachable: ${server.host}:${port}; client not started`);
return 2;
}
}
}
fs.mkdirSync(options.output, { recursive: true, mode: 0o700 });
const logPath = path.join(options.output, 'client.log');
const rssPath = path.join(options.output, 'rss.jsonl');
const logFile = fs.openSync(logPath, 'w', 0o600);
const rssFile = fs.openSync(rssPath, 'w', 0o600);
const args = ['--live-client', options.client];
if (server) args.push('--live-server', options.server, '--login-screen');
else args.push('--fake-mobs', '24', '--frames', String(options.frames));
const started = Date.now();
const child = spawn(options.binary, args, { cwd: repo, stdio: ['inherit', 'pipe', 'pipe'] });
const literals = secretLiterals();
let log = '';
let pending = '';
const append = (chunk) => {
pending += chunk.toString('utf8');
const lines = pending.split('\n');
pending = lines.pop();
for (const line of lines) {
const clean = `${redact(line, literals)}\n`;
fs.writeSync(logFile, clean);
process.stdout.write(clean);
log += clean;
}
};
child.stdout.on('data', append);
child.stderr.on('data', append);
let sampleBusy = false;
let pendingSample = Promise.resolve();
const samples = [];
const sample = async () => {
if (sampleBusy) return;
sampleBusy = true;
const rss = await readRss(child.pid);
if (rss !== null) {
const row = { wall_ms: Date.now(), rss_kib: rss };
samples.push(row);
fs.writeSync(rssFile, `${JSON.stringify(row)}\n`);
}
sampleBusy = false;
};
await sample();
const interval = setInterval(() => { pendingSample = sample(); }, 1000);
const onSignal = (signal) => child.kill(signal);
process.on('SIGINT', onSignal);
process.on('SIGTERM', onSignal);
const result = await new Promise((resolve) => {
child.once('error', (error) => resolve({ code: null, signal: null, error: error.message }));
child.once('close', (code, signal) => resolve({ code, signal }));
});
clearInterval(interval);
await pendingSample;
process.off('SIGINT', onSignal);
process.off('SIGTERM', onSignal);
if (pending) append('\n');
fs.closeSync(logFile);
fs.closeSync(rssFile);
const durationSeconds = Math.round((Date.now() - started) / 1000);
const summary = parseSummary(log);
const completed = result.code === 0 && summary?.frames > 0;
const status = !completed ? 'FAIL' : server ? 'NEEDS_MANUAL_REVIEW' : 'SMOKE_PASS';
const report = {
status,
mode: server ? 'real_server' : 'fake_server',
started_at: new Date(started).toISOString(), duration_seconds: durationSeconds,
minimum_seconds: server ? options.minSeconds : null,
minimum_met: server ? durationSeconds >= options.minSeconds : null,
exit: result, summary,
rss: { samples: samples.length, first_kib: samples[0]?.rss_kib ?? null,
last_kib: samples.at(-1)?.rss_kib ?? null,
max_kib: samples.length ? Math.max(...samples.map((row) => row.rss_kib)) : null },
manual_checks: server ? {
login: null, character_selection: null, movement: null, combat: null,
map_change: null, inventory_and_chat: null, resize_and_focus: null,
visual_comparison: null,
} : null,
note: server ? 'Review manual checks and frame/RSS trends; this runner does not certify gameplay or memory leaks.'
: 'Fake-server smoke checks startup and rendering only.',
};
const reportPath = path.join(options.output, 'report.json');
fs.writeFileSync(reportPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 });
console.log(`Native acceptance: ${status}; report=${reportPath}`);
return completed ? 0 : 1;
}
main().then((code) => { process.exitCode = code; }).catch((error) => {
console.error(error.message);
process.exitCode = 1;
});
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env bash
# Batch 2A gate (docs/PORT-PLAN.md): build port_logic, port_platform + the header self-containment gate per
# platform. A platform whose toolchain is missing prints BLOCKED and never counts as a pass.
#
# scripts/port_gate.sh [macos|android|ios|windows|linux ...] (default: all)
#
# Build trees go to $MT_PORT_GATE_DIR (default build/port-gate/). Toolchains:
# android ANDROID_NDK_HOME / ANDROID_NDK_ROOT (NDK 27.2)
# ios Xcode with the iphoneos SDK
# windows MT_WIN_CXX (default x86_64-w64-mingw32-g++); MSVC is covered on a Windows host
# linux MT_LINUX_TOOLCHAIN (a CMake toolchain file), or a Linux host
set -uo pipefail
cd "$(dirname "$0")/.."
OUT="${MT_PORT_GATE_DIR:-build/port-gate}"
PLATFORMS=("$@")
[ ${#PLATFORMS[@]} -eq 0 ] && PLATFORMS=(macos android ios windows linux)
PORT_ONLY=(-DMT_BUILD_NATIVE_RENDER=OFF)
declare -a RESULTS=()
run() { # name, build dir, configure args...
local name="$1" dir="$2"; shift 2
if cmake -S . -B "$dir" "$@" >"$dir.log" 2>&1 \
&& cmake --build "$dir" --target port_platform ${BUILD_EXTRA[@]+"${BUILD_EXTRA[@]}"} >>"$dir.log" 2>&1; then
RESULTS+=("$name PASS")
else
RESULTS+=("$name FAIL (see $dir.log)")
fi
}
mkdir -p "$OUT"
for p in "${PLATFORMS[@]}"; do
BUILD_EXTRA=()
case "$p" in
macos)
# Every port test target, so the ctest run below never meets a test the gate did not build.
BUILD_EXTRA=(--target all)
run macos "$OUT/macos" "${PORT_ONLY[@]}" -DBUILD_TESTING=ON
last=$((${#RESULTS[@]} - 1))
if [[ "${RESULTS[$last]}" == *PASS ]] && ! ctest --test-dir "$OUT/macos" -R '^port\.' --output-on-failure >>"$OUT/macos.log" 2>&1; then
RESULTS[$last]="macos FAIL (port tests, see $OUT/macos.log)"
fi
;;
android)
NDK="${ANDROID_NDK_ROOT:-${ANDROID_NDK_HOME:-}}"
SDK="${ANDROID_HOME:-${ANDROID_SDK_ROOT:-$HOME/Library/Android/sdk}}"
if [ -z "$NDK" ] && [ -d "$SDK/ndk" ]; then
NDK="$SDK/ndk/$(ls -1 "$SDK/ndk" | sort -V | tail -1)"
fi
if [ -z "$NDK" ] || [ ! -f "$NDK/build/cmake/android.toolchain.cmake" ]; then
RESULTS+=("android BLOCKED (no NDK)"); continue
fi
run android "$OUT/android" "${PORT_ONLY[@]}" -DBUILD_TESTING=OFF \
-DCMAKE_TOOLCHAIN_FILE="$NDK/build/cmake/android.toolchain.cmake" \
-DANDROID_ABI=arm64-v8a -DANDROID_PLATFORM=android-24
;;
ios)
if ! xcrun --sdk iphoneos --show-sdk-path >/dev/null 2>&1; then
RESULTS+=("ios BLOCKED (no iphoneos SDK)"); continue
fi
BUILD_EXTRA=(-- -sdk iphoneos CODE_SIGNING_ALLOWED=NO)
run ios "$OUT/ios" "${PORT_ONLY[@]}" -DBUILD_TESTING=OFF -G Xcode -DCMAKE_SYSTEM_NAME=iOS \
-DCMAKE_OSX_ARCHITECTURES=arm64 -DCMAKE_OSX_DEPLOYMENT_TARGET=14.0
;;
windows)
CXX_W="${MT_WIN_CXX:-x86_64-w64-mingw32-g++}"
if [[ "$(uname -s)" != MINGW* ]] && ! command -v "$CXX_W" >/dev/null; then
RESULTS+=("windows BLOCKED (no $CXX_W)"); continue
fi
run windows "$OUT/windows" "${PORT_ONLY[@]}" -DBUILD_TESTING=OFF -DCMAKE_SYSTEM_NAME=Windows \
-DCMAKE_C_COMPILER="${CXX_W%g++}gcc" -DCMAKE_CXX_COMPILER="$CXX_W"
;;
linux)
if [ "$(uname -s)" = Linux ]; then
BUILD_EXTRA=(--target port_common_test --target port_eterpack_test)
run linux "$OUT/linux" "${PORT_ONLY[@]}" -DBUILD_TESTING=ON
elif [ -n "${MT_LINUX_TOOLCHAIN:-}" ]; then
run linux "$OUT/linux" "${PORT_ONLY[@]}" -DBUILD_TESTING=OFF -DCMAKE_TOOLCHAIN_FILE="$MT_LINUX_TOOLCHAIN"
else
RESULTS+=("linux BLOCKED (not a Linux host and no MT_LINUX_TOOLCHAIN)")
fi
;;
*) RESULTS+=("$p UNKNOWN") ;;
esac
done
printf '%s\n' "${RESULTS[@]}"
! printf '%s\n' "${RESULTS[@]}" | grep -q FAIL
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env node
/* Copy stdin to stdout line by line, replacing the test credentials before any
* byte reaches disk. Secrets come only from MT_ACCOUNT / MT_PASSWORD in the
* environment and are never printed, logged or passed as arguments.
*
* node scripts/redact_stream.mjs <fifo >client.log
*
* Uses blocking reads on fd 0: async stdin on a macOS FIFO does not reliably
* report EOF, which would leave the runner waiting for the redactor. */
import fs from 'node:fs';
import { fileURLToPath } from 'node:url';
const MIN_LENGTH = 2;
export function secretLiterals(env = process.env) {
return ['MT_PASSWORD', 'MT_ACCOUNT']
.map((name) => env[name] || '')
.filter((value) => value.length >= MIN_LENGTH)
// Longest first so an account contained in a password is not half-replaced.
.sort((a, b) => b.length - a.length);
}
export function redact(line, literals = secretLiterals()) {
let out = line;
for (const literal of literals) out = out.split(literal).join('[redacted]');
return out;
}
function pump() {
const literals = secretLiterals();
const buffer = Buffer.alloc(64 * 1024);
let pending = '';
const write = (text) => {
const bytes = Buffer.from(text);
let offset = 0;
while (offset < bytes.length) offset += fs.writeSync(1, bytes, offset);
};
for (;;) {
let count;
try {
count = fs.readSync(0, buffer, 0, buffer.length, null);
} catch (error) {
if (error.code === 'EAGAIN') continue;
if (error.code === 'EOF') break;
throw error;
}
if (count === 0) break;
pending += buffer.toString('utf8', 0, count);
const lines = pending.split('\n');
pending = lines.pop();
if (lines.length > 0) write(lines.map((line) => `${redact(line, literals)}\n`).join(''));
}
// A trailing partial line is still redacted as a whole.
if (pending !== '') write(`${redact(pending, literals)}\n`);
}
if (process.argv[1] && fileURLToPath(import.meta.url) === fs.realpathSync(process.argv[1])) {
if (process.argv.includes('--help')) {
console.log('usage: node scripts/redact_stream.mjs < INPUT > OUTPUT (reads MT_ACCOUNT/MT_PASSWORD from env)');
} else {
pump();
}
}
+81
View File
@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""Focused safeguards for the 40250/Longju locale merge."""
import importlib.util
import json
import tempfile
import unittest
from pathlib import Path
SPEC = importlib.util.spec_from_file_location("build_zh_locale", Path(__file__).with_name("build_zh_locale.py"))
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
class LocaleBuildTest(unittest.TestCase):
def test_preserves_40250_keys_formats_and_proto(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
en, zh, client, output = (root / name for name in ("en", "zh", "client", "out/Client"))
for path in (en, zh, client):
path.mkdir()
(en / "locale_game.txt").write_bytes(b"GOOD\tValue %d\tSA\nBAD\t%s at %d\tSA\nONLY_EN\tFallback\n")
(zh / "locale_game.txt").write_bytes("GOOD\t中文 %d\nBAD\t中文 %d 和 %s\n".encode("gbk"))
(en / "locale_interface.txt").write_bytes(b"ACCEPT\tAccept\n")
(zh / "locale_interface.txt").write_bytes("ACCEPT\t确定\n".encode("gbk"))
(en / "item_proto").write_bytes(b"40250-proto")
(zh / "item_proto").write_bytes(b"old-proto")
(client / "pack").mkdir()
(client / "pack/Index").write_bytes(b"PACK\r\nlocale/de/\r\nlocale_de\r\nlocale/en/\r\nlocale_en\r\n")
(client / "pack/locale_en.eix").write_bytes(b"en")
# Stands in for eterpack_pack: records its arguments as the .eix.
pack_tool = root / "eterpack_pack"
pack_tool.write_text("#!/bin/sh\necho \"$2 $3\" > \"$1.eix\"\n")
pack_tool.chmod(0o755)
report_path = MODULE.build(en, zh, output, client, pack_tool=pack_tool)
staged = output.parent / "locale-src/zh"
merged = (staged / "locale_game.txt").read_bytes().decode("gbk")
self.assertIn("GOOD\t中文 %d\tSA", merged)
self.assertIn("BAD\t%s at %d\tSA", merged)
self.assertIn("ONLY_EN\tFallback", merged)
self.assertEqual((staged / "item_proto").read_bytes(), b"40250-proto")
self.assertEqual((output / "locale.cfg").read_text(), "10002 936 zh\n")
self.assertEqual((output / "locale_zh.cfg").read_text(), "10002 936 zh\n")
# A pack/ of its own: original packs linked, locale_zh packed from the staging dir and registered.
self.assertFalse((output / "pack").is_symlink())
self.assertFalse((output / "locale/zh").exists())
self.assertTrue((output / "pack/locale_en.eix").is_symlink())
self.assertEqual((output / "pack/locale_zh.eix").read_text(), f"{staged} locale/zh/\n")
self.assertEqual((output / "pack/Index").read_bytes(),
b"PACK\r\nlocale/de/\r\nlocale_de\r\nlocale/zh/\r\nlocale_zh\r\n"
b"locale/en/\r\nlocale_en\r\n")
self.assertEqual((client / "pack/Index").read_bytes()[:4], b"PACK")
self.assertFalse((client / "pack/locale_zh.eix").exists())
report = json.loads(report_path.read_text())
self.assertEqual(report["files"]["locale_game.txt"]["incompatible_format_keys"], ["BAD"])
# Reviewed overlay: fills an English-only key and replaces a format-incompatible Longju value.
names = root / "names"
names.mkdir()
(names / "locale_game_zh_overlay.txt").write_text("KEY\tVALUE\nBAD\t%s 于 %d\nONLY_EN\t备用\n")
report_path = MODULE.build(en, zh, output, client, names_dir=names, pack_tool=pack_tool)
merged = (staged / "locale_game.txt").read_bytes().decode("gbk")
self.assertIn("BAD\t%s 于 %d\tSA", merged)
self.assertIn("ONLY_EN\t备用", merged)
game = json.loads(report_path.read_text())["files"]["locale_game.txt"]
self.assertEqual((game["english_fallback_keys"], game["incompatible_format_keys"]), ([], []))
def test_name_overlay_replaces_snapshot_rows(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / "names.txt").write_text("VNUM\tLOCALE_NAME\n1\tYang\n2\t剑\n")
(root / "overlay.txt").write_text("VNUM\tLOCALE_NAME\n1\t金币\n99\t不存在\n")
result = MODULE.convert_names(root / "names.txt", root / "out.txt", root / "overlay.txt")
self.assertEqual((root / "out.txt").read_bytes().decode("gbk"), "1\t金币\n2\t剑\n")
self.assertEqual(result["overlaid"], 1)
if __name__ == "__main__":
unittest.main()
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env python3
"""Checks for the 40250 server's GBK locale dictionary overlay."""
import tempfile
import unittest
from pathlib import Path
import build_zh_server_locale as locale
class ServerLocaleTest(unittest.TestCase):
def test_build_preserves_keys_and_existing_translations(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "locale_string.txt"
report = locale.build(locale.DEFAULT_ENGLISH, locale.DEFAULT_CHINESE, output)
reference = locale.read_pairs(locale.DEFAULT_ENGLISH)
existing = locale.read_pairs(locale.DEFAULT_CHINESE)
generated = locale.read_pairs(output)
self.assertEqual(report["new_chinese_keys"], 36)
self.assertEqual(report["corrected_format_keys"], 2)
self.assertEqual(set(generated), set(reference) | set(existing))
corrected = {key.encode("cp949") for key in locale.CORRECTED}
for key, value in existing.items():
if key not in corrected:
self.assertEqual(generated[key], value)
for key, value in generated.items():
value.decode("gbk")
self.assertEqual(locale.format_args(key), locale.format_args(value))
self.assertEqual(
generated["OX 상품을 지급하였습니다. ID: %d 수량: %d".encode("cp949")].decode("gbk"),
"已发放OX奖品,编号:%d,数量:%d。",
)
if __name__ == "__main__":
unittest.main()
+78
View File
@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""Verify every RUN_AS_IS root script against the actual 40250 root.epk bytes."""
from __future__ import annotations
import argparse
import hashlib
import json
import sys
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / "tools" / "epk_scan"))
import epk_scan # noqa: E402
def digest(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--client", type=Path, default=REPO / "../40250/Server Client TMP4/Client")
args = parser.parse_args()
client = args.client.resolve()
reference = epk_scan.refroot.reference_root(REPO)
_, keys, _, packs, _ = epk_scan.load(client, reference)
root_pack = packs["root"]
if root_pack["missing"]:
print("FAIL: root.eix/root.epk missing", file=sys.stderr)
return 1
entries = {entry["name"].lower(): entry for entry in root_pack["entries"]}
source_dir = client / "Eternexus" / "root"
ledger_dir = REPO / "audit" / "port-map" / "Client" / "root"
failures: list[str] = []
scripts = sorted(source_dir.glob("*.py"))
if {p.stem for p in scripts} != {p.name.removesuffix(".py.json") for p in ledger_dir.glob("*.py.json")}:
failures.append("ledger files do not match the reference root script list")
with root_pack["epk"].open("rb") as archive:
for source in scripts:
name = source.name
entry = entries.get(name.lower())
ledger_path = ledger_dir / f"{name}.json"
if entry is None or not ledger_path.is_file():
failures.append(f"{name}: missing pack entry or ledger")
continue
archive.seek(entry["pos"])
data = archive.read(entry["size"])
if entry["type"] in (1, 2):
archive.seek(entry["pos"])
data = archive.read(epk_scan.lz_object_len(data[:16]))
data = epk_scan.lz_object(
data, keys["s_adwEterPackSecurityKey"] if entry["type"] == 2 else None
)
elif entry["type"] != 0:
failures.append(f"{name}: unsupported pack compression type {entry['type']}")
continue
ledger = json.loads(ledger_path.read_text(encoding="utf-8"))
expected = digest(source.read_bytes())
if digest(data) != expected or ledger["reference_sha256"] != expected:
failures.append(f"{name}: root.epk, reference source and ledger hashes differ")
for function, info in ledger["functions"].items():
if info.get("status") != "RUN_AS_IS" or info.get("impl") != [f"pack://{name}:{function}"]:
failures.append(f"{name}:{function}: incorrect RUN_AS_IS mapping")
if not isinstance(info.get("evidence"), list) or not info["evidence"]:
failures.append(f"{name}:{function}: missing evidence list")
for failure in failures:
print(f"FAIL: {failure}", file=sys.stderr)
if failures:
return 1
print(f"PASS: {len(scripts)} root scripts match 40250 source, root.epk and port-map")
return 0
if __name__ == "__main__":
raise SystemExit(main())