import concurrent.futures
import json
import os
import re
import subprocess
import sys
import tempfile
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PLACEHOLDER = re.compile(r"<(?:ro|RO)(\d+)_(\d+)>")
def fn_src(name: str) -> str:
if name.isidentifier():
return f"function p(o) {{ return o.{name}; }}\n"
return f"function p(o) {{ return o[{json.dumps(name)}]; }}\n"
def one(ver: str, name: str, node: str) -> tuple:
with tempfile.TemporaryDirectory() as td:
src = os.path.join(td, "p.js")
jsc = os.path.join(td, "p.jsc")
open(src, "w", encoding="utf-8").write(fn_src(name))
r = subprocess.run([node, os.path.join(ROOT, "scripts", "mkcorpus.js"), src, jsc],
capture_output=True, text=True, cwd=ROOT, timeout=180)
if r.returncode != 0:
return (name, None, "compile")
d = subprocess.run([os.path.join(ROOT, "target", "release", "jscd"), "decompile", jsc],
capture_output=True, text=True, cwd=ROOT, timeout=180)
if d.returncode != 0:
return (name, None, (d.stderr or "").strip()[:60])
m = PLACEHOLDER.search(d.stdout)
if m:
return (name, f"{int(m.group(1))}/{int(m.group(2))}", None)
return (name, None, None)
def main() -> None:
argv = [a for a in sys.argv[1:]]
jobs = 1
if "--jobs" in argv:
i = argv.index("--jobs")
jobs = int(argv[i + 1])
argv = argv[:i] + argv[i + 2:]
ver, list_path, out_path = argv[0], argv[1], argv[2]
names, seen = [], set()
for line in open(list_path, encoding="utf-8", errors="ignore"):
n = line.rstrip("\n")
if n and n not in seen:
seen.add(n)
names.append(n)
node = os.path.expanduser(f"~/.local/share/mise/installs/node/{ver}/bin/node")
entries = {}
fails = []
if jobs > 1:
done = 0
with concurrent.futures.ProcessPoolExecutor(max_workers=jobs) as ex:
futs = [ex.submit(one, ver, n, node) for n in names if n not in ("", "\n")]
for fut in concurrent.futures.as_completed(futs):
name, key, err = fut.result()
done += 1
if key:
entries[key] = name
elif err:
fails.append((name, err))
if done % 500 == 0:
print(f" …{done}/{len(names)}(已解出 {len(entries)})", file=sys.stderr)
else:
with tempfile.TemporaryDirectory() as td:
for i, name in enumerate(names):
if name in ("", "\n"):
continue
src = os.path.join(td, "p.js")
jsc = os.path.join(td, "p.jsc")
open(src, "w", encoding="utf-8").write(fn_src(name))
r = subprocess.run(
[node, os.path.join(ROOT, "scripts", "mkcorpus.js"), src, jsc],
capture_output=True, text=True, cwd=ROOT, timeout=120,
)
if r.returncode != 0:
continue
d = subprocess.run(
[os.path.join(ROOT, "target", "release", "jscd"), "decompile", jsc],
capture_output=True, text=True, cwd=ROOT, timeout=120,
)
if d.returncode != 0:
fails.append((name, (d.stderr or "").strip()[:60]))
continue
m = PLACEHOLDER.search(d.stdout)
if m:
entries[f"{int(m.group(1))}/{int(m.group(2))}"] = name
if (i + 1) % 200 == 0:
print(f" …{i + 1}/{len(names)}(已解出 {len(entries)})", file=sys.stderr)
out = {
"schema": "jscd.ro-map/v1",
"v8": ver,
"note": "由 scripts/build_ro_map_small.py 逐名编译生成",
"entries": entries,
}
with open(out_path, "w", encoding="utf-8") as f:
json.dump(out, f, ensure_ascii=False, indent=1, sort_keys=True)
print(f"ro-map: {len(entries)} 条({len(names)} 个候选,{len(fails)} 份解析失败)→ {out_path}")
if fails:
kinds = {}
for name, err in fails:
kinds.setdefault(err, []).append(name)
for k, v in sorted(kinds.items(), key=lambda kv: -len(kv[1]))[:5]:
print(f" 失败 {len(v)} 例:{k[:70]} 例名: {v[:4]}")
if __name__ == "__main__":
main()