04dbd0608b
并发的真正约束=九门 play/smoke 固定端口+Chrome 实例(gameId 前缀已隔离产物)。 查清整条链端口可参数化:gen.mjs --port/--cdp(done 门 smoke 用)、play.mjs 每次独立 spawn server+Chrome+userDataDir。据此: - _port_pool:每并发槽一对独立 (server_port,cdp_port),段不重叠。 - run_pair_sync:同步整对跑(线程内 run_studio 自带 loop)。 - run_multi 加 conc:端口池(size=conc 天然限并发)+ asyncio.to_thread 每对一线程 + gather。conc=1 串行;conc>1 有界并发。前台进程内有界并发,非后台子代理 task。 - _gen_node 传 --port/--cdp 给 gen.mjs(并发防 smoke 撞端口)。 CLI 加 --conc。test_compare_multi 14/14(+端口池不撞)。
417 lines
20 KiB
Python
417 lines
20 KiB
Python
"""
|
||
compare_node.py — 便宜档 Python 路 vs Node 旧路对照验证。
|
||
|
||
两层用法:
|
||
· 单品类对照(spike U6 起点):compare_logic / build_report / main_async —— 同 brief 各跑一次,纯逻辑见下。
|
||
· 三品类批量对照(WU-A 后续 U2):run_multi —— 三个 tap-targets 代表品类(点击得分/打地鼠/经营点客)各 n=3–5,
|
||
用每品类金标 play-spec 喂两路 held constant、把驱动器从对照变量里摘掉,按两层判据 + 逐品类绝对地板聚合,
|
||
产"框架等价证据(金标 spec 下)"。
|
||
|
||
对照公平的关键(KTD3 + Codex C1/C4 + Opus O1):
|
||
· 金标 spec 强制覆写到 staged 路径 games/_wg1-gen/<id>/play-spec.json(play.cdp.cjs 真实读取处,非 evidence 目录),
|
||
在「生成」与「九门 play」之间注入 —— 两路同用一份驱动器,过门率差才只反映生成质量。
|
||
· Python 路用 run_studio(run_gates=False) 做 generation-only(内部不 play),Node 路 gen.mjs 本就不 play;
|
||
两路都「生成 → 注入金标 → 单独 play」。
|
||
· 成功定义 = 坐实「Python ≈ Node、范式未回退」(金标 spec 下),≠「过九门」、≠「生产退役许可」(后者另 gate
|
||
在 WU-C 自动 spec 达金标同等过门率)。
|
||
|
||
CLI:cheap-worker/.venv/bin/python cheap-worker/compare_node.py [--genres click-score,whack-mole,shop-serve] [--n 3]
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import statistics
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent)) # → cheap-worker/
|
||
import _bootstrap # noqa: E402,F401 跨包 sys.path 兜底
|
||
import cheap_run # noqa: E402
|
||
import cheap_studio # noqa: E402
|
||
|
||
_GOLDEN_DIR = Path(__file__).resolve().parent / "fixtures" / "golden-specs"
|
||
_RESULTS_DIR = Path(__file__).resolve().parent / "results"
|
||
|
||
# 三个 tap-targets 代表品类(plan KTD2):brief 取自 base1/base2/base4 run-summary,金标 spec 见 fixtures/golden-specs/。
|
||
GENRES = [
|
||
{"key": "click-score", "base": "base1", "golden": "click-score.play-spec.json"},
|
||
{"key": "whack-mole", "base": "base2", "golden": "whack-mole.play-spec.json"},
|
||
{"key": "shop-serve", "base": "base4", "golden": "shop-serve.play-spec.json"},
|
||
]
|
||
_GENRE_BY_KEY = {g["key"]: g for g in GENRES}
|
||
|
||
# 判据阈值(R3 两层 + 逐品类绝对地板 + 双低标红)。
|
||
_EPS = 1e-9
|
||
_PLAY_GATE = "H_progress" # 「真玩起来了」的可玩门 = 有真实游戏进展(driven 后 H 过)。
|
||
_DOUBLE_LOW_PLAY = 0.5 # 两路真玩比例都 < 此 → 双低、对照不充分、标红。
|
||
_FLOOR_PLAY = 0.6 # 坐实绝对地板:Python 须在 ≥60% 局真玩起来(带余量,不只 ≥Node)。
|
||
_MAX_N = 5 # n>5 小批口径(plan R5),上限夹 5、不滑向 n≥30 统计批跑。
|
||
|
||
|
||
# ───────────────────────── 纯逻辑(可单测,不依赖真跑)─────────────────────────
|
||
|
||
def gates_from_verdict(verdict) -> dict:
|
||
"""从九门 verdict.json 抽 {gate: pass_bool}。"""
|
||
if not isinstance(verdict, dict):
|
||
return {}
|
||
guards = verdict.get("guards") or {}
|
||
return {k: bool(g.get("pass")) for k, g in guards.items() if isinstance(g, dict)}
|
||
|
||
|
||
def src_shape(game_dir: Path) -> dict:
|
||
"""产物形态:是否真 src/ 多文件工程(game-logic.js 在 + ≥2 个 .js)。"""
|
||
src = game_dir / "src"
|
||
js = list(src.glob("*.js")) if src.exists() else []
|
||
return {"realSrcMultifile": (src / "game-logic.js").exists() and len(js) >= 2, "jsCount": len(js)}
|
||
|
||
|
||
def _pass_rate(gates: dict) -> float:
|
||
return round(sum(1 for v in gates.values() if v) / len(gates), 3) if gates else 0.0
|
||
|
||
|
||
def compare_logic(py_gates: dict, node_gates: dict, py_real_src: bool) -> dict:
|
||
"""单品类对照判定(spike U6 起点):Python 产真 src/ 多文件 且 九门过门率不低于 Node → 坐实。"""
|
||
py_rate = _pass_rate(py_gates)
|
||
node_rate = _pass_rate(node_gates)
|
||
spike_pass = bool(py_real_src and py_rate >= node_rate)
|
||
return {
|
||
"pyPassRate": py_rate,
|
||
"nodePassRate": node_rate,
|
||
"rateDelta": round(py_rate - node_rate, 3),
|
||
"pyRealSrcMultifile": py_real_src,
|
||
"spikePass": spike_pass,
|
||
"verdict": "范式可移植坐实(Python 产真 src/ 多文件且九门过门率不低于 Node)" if spike_pass
|
||
else "未达标(Python 产物非真 src/ 多文件,或九门过门率低于 Node)",
|
||
}
|
||
|
||
|
||
def _played(gates: dict) -> bool:
|
||
"""这一局是否真玩起来了 = H_progress 过(有真实游戏进展;两路同卡菜单时此门为 False)。"""
|
||
return gates.get(_PLAY_GATE) is True
|
||
|
||
|
||
def _mean(xs) -> float:
|
||
xs = list(xs)
|
||
return round(statistics.fmean(xs), 3) if xs else 0.0
|
||
|
||
|
||
def judge_genre(py_runs: list, node_runs: list) -> dict:
|
||
"""逐品类两层判据 + 绝对地板 + 双低标红(R3）。
|
||
|
||
py_runs/node_runs:每元素 = {"gates": {门: pass_bool}, "realSrc": bool}(一局)。
|
||
|
||
status:
|
||
· inconclusive(标红):两路真玩比例都 < 0.5 —— 两路都没真玩起来,对照不充分,不计坐实。
|
||
· equivalent(坐实):Python 每局产真 src/ 且 过门率不低于 Node(相对层)且 Python 真玩比例 ≥0.6(绝对地板)。
|
||
· regressed:Python 明显差(非真 src/ / 过门率低 / 真玩比例不达地板)。
|
||
"""
|
||
py_rate = _mean(_pass_rate(r["gates"]) for r in py_runs)
|
||
node_rate = _mean(_pass_rate(r["gates"]) for r in node_runs)
|
||
py_play = _mean(1.0 if _played(r["gates"]) else 0.0 for r in py_runs)
|
||
node_play = _mean(1.0 if _played(r["gates"]) else 0.0 for r in node_runs)
|
||
py_real = bool(py_runs) and all(r.get("realSrc") for r in py_runs)
|
||
|
||
relative_ok = py_rate >= node_rate - _EPS
|
||
double_low = py_play < _DOUBLE_LOW_PLAY and node_play < _DOUBLE_LOW_PLAY
|
||
floor_ok = py_play >= _FLOOR_PLAY
|
||
|
||
if double_low:
|
||
status, reason = "inconclusive", "两路真玩比例都低(同卡/同低分)→ 对照不充分、标红、不计坐实"
|
||
elif py_real and relative_ok and floor_ok:
|
||
status, reason = "equivalent", "Python 产真 src/ + 过门率不低于 Node + 真玩比例达绝对地板 → 坐实等价"
|
||
else:
|
||
bits = []
|
||
if not py_real:
|
||
bits.append("Python 非每局真 src/")
|
||
if not relative_ok:
|
||
bits.append(f"过门率低于 Node({py_rate}<{node_rate})")
|
||
if not floor_ok:
|
||
bits.append(f"Python 真玩比例 {py_play}<{_FLOOR_PLAY} 地板")
|
||
status, reason = "regressed", "回退:" + "、".join(bits)
|
||
|
||
return {
|
||
"pyPassRate": py_rate, "nodePassRate": node_rate, "rateDelta": round(py_rate - node_rate, 3),
|
||
"pyPlayRatio": py_play, "nodePlayRatio": node_play, "pyRealSrcAll": py_real,
|
||
"doubleLowRed": double_low, "status": status, "reason": reason,
|
||
}
|
||
|
||
|
||
def _read_verdict(game_id: str):
|
||
vp = cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json"
|
||
if vp.exists():
|
||
try:
|
||
return json.loads(vp.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _clamp_n(n: int) -> int:
|
||
"""n>5 小批口径:夹到 [1, 5],不滑向 n≥30 统计批跑(plan R5)。"""
|
||
try:
|
||
n = int(n)
|
||
except (TypeError, ValueError):
|
||
n = 3
|
||
return max(1, min(_MAX_N, n))
|
||
|
||
|
||
# ───────────────────────── 金标 spec 注入(staged 路径、强制覆写)─────────────────────────
|
||
|
||
def golden_spec_path(genre: dict) -> Path:
|
||
return _GOLDEN_DIR / genre["golden"]
|
||
|
||
|
||
def _write_spec_atomic(dst_path: Path, content: str) -> None:
|
||
"""原子写 play-spec.json(写 tmp 后 rename,防 play 读半成品)。"""
|
||
dst_path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = dst_path.parent / (dst_path.name + ".tmp")
|
||
tmp.write_text(content, encoding="utf-8")
|
||
tmp.replace(dst_path)
|
||
|
||
|
||
def inject_golden(game_id: str, golden: Path) -> Path:
|
||
"""强制覆写金标 spec 到 staged games/_wg1-gen/<id>/play-spec.json(play.cdp.cjs 真实读取处)。
|
||
|
||
与 ensure_play_spec 的「不覆盖」不同:这里**强制覆写**两路各自生成时产的自动 spec,把驱动器恒定到金标
|
||
—— 这是对照公平的落点(Codex C1/C4)。返回写入的 staged spec 路径。
|
||
"""
|
||
dst = cheap_run.wg1_game_dir(game_id) / "play-spec.json"
|
||
_write_spec_atomic(dst, golden.read_text(encoding="utf-8"))
|
||
return dst
|
||
|
||
|
||
# ───────────────────────── 真跑两路(生成 / play 拆步、金标注入、串行隔离)─────────────────────────
|
||
|
||
def load_brief(genre: dict) -> str:
|
||
"""品类 brief 取自 base 样本 run-summary(单一来源);缺失则空串(由调用方兜底)。"""
|
||
rs = cheap_run.game_dir(genre["base"]) / "evidence" / "run-summary.json"
|
||
if rs.exists():
|
||
try:
|
||
return json.loads(rs.read_text(encoding="utf-8")).get("brief") or ""
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _gen_node(brief: str, game_id: str, port: int = 4320, cdp_port: int = 9222,
|
||
timeout: float = 900.0) -> dict:
|
||
"""Node 旧路 generation-only:gen.mjs --mode saa(scaffold+stage+done 门 smoke,SAA 不跑九门)。不在此 play。
|
||
|
||
传 --port/--cdp:gen.mjs done 门内部 smoke-boot 用这俩端口(行 142-143/249),并发时每槽独立端口防撞。
|
||
"""
|
||
r = subprocess.run(
|
||
["node", str(cheap_run._AMODEL_GEN / "gen.mjs"), "--mode", "saa", "--game-id", game_id,
|
||
"--brief", brief, "--port", str(port), "--cdp", str(cdp_port)],
|
||
cwd=str(cheap_run._GAME_RUNTIME), capture_output=True, text=True, timeout=timeout,
|
||
env=cheap_run._shell_env(), check=False,
|
||
)
|
||
return {"rc": r.returncode, "raw": (r.stdout + r.stderr)[-1200:]}
|
||
|
||
|
||
def run_pair_sync(genre: dict, brief: str, py_id: str, node_id: str,
|
||
port: int = 4320, cdp_port: int = 9222) -> dict:
|
||
"""一局对照(同步、线程内整对跑):两路各 generation-only → 注入同一份金标 spec → 各自单独 play。
|
||
|
||
设计为同步,由 run_multi 经 asyncio.to_thread 并发调度(每对一个线程、独立端口);线程内 run_studio 用自带
|
||
event loop(asyncio.run)。返回逐门 + 形态。
|
||
"""
|
||
golden = golden_spec_path(genre)
|
||
|
||
# Python 路:generation-only(run_gates=False,内部不 play)→ 注入金标 → play。
|
||
asyncio.run(cheap_studio.run_studio(py_id, brief, run_gates=False, port=port, cdp_port=cdp_port))
|
||
inject_golden(py_id, golden)
|
||
py_play = cheap_run.play(py_id, port=port, cdp_port=cdp_port)
|
||
|
||
# Node 路:gen.mjs --mode saa(不 play,内部 smoke 用同槽端口)→ 注入金标(覆写自动 spec)→ play。
|
||
node_gen = _gen_node(brief, node_id, port=port, cdp_port=cdp_port)
|
||
inject_golden(node_id, golden)
|
||
node_play = cheap_run.play(node_id, port=port, cdp_port=cdp_port)
|
||
|
||
py_gates = gates_from_verdict(py_play.get("verdict"))
|
||
node_gates = gates_from_verdict(node_play.get("verdict"))
|
||
return {
|
||
"pyId": py_id, "nodeId": node_id, "nodeGenRc": node_gen["rc"],
|
||
"py": {"gates": py_gates, "realSrc": src_shape(cheap_run.game_dir(py_id))["realSrcMultifile"]},
|
||
"node": {"gates": node_gates, "realSrc": src_shape(cheap_run.game_dir(node_id))["realSrcMultifile"]},
|
||
}
|
||
|
||
|
||
def aggregate(genre_runs: dict) -> dict:
|
||
"""按品类聚合:每品类的多局 runs → judge_genre 结论 + 顶层退役就绪范围。
|
||
|
||
genre_runs: {genreKey: [run, ...]}，run = run_pair 返回。
|
||
"""
|
||
per_genre = []
|
||
for key, runs in genre_runs.items():
|
||
py_runs = [r["py"] for r in runs]
|
||
node_runs = [r["node"] for r in runs]
|
||
judge = judge_genre(py_runs, node_runs)
|
||
per_genre.append({"genre": key, "n": len(runs), "judge": judge,
|
||
"runs": [{"pyId": r["pyId"], "nodeId": r["nodeId"],
|
||
"pyGates": r["py"]["gates"], "nodeGates": r["node"]["gates"]} for r in runs]})
|
||
equivalent = [g["genre"] for g in per_genre if g["judge"]["status"] == "equivalent"]
|
||
inconclusive = [g["genre"] for g in per_genre if g["judge"]["status"] == "inconclusive"]
|
||
regressed = [g["genre"] for g in per_genre if g["judge"]["status"] == "regressed"]
|
||
return {
|
||
"perGenre": per_genre,
|
||
"summary": {
|
||
"equivalentGenres": equivalent,
|
||
"inconclusiveGenres": inconclusive,
|
||
"regressedGenres": regressed,
|
||
# 「框架等价证据(金标 spec 下)」:坐实等价的品类 = Node 旧路在该品类可退役就绪(金标 spec 下);
|
||
# 面向生产的退役授权另 gate 在 WU-C 自动 spec 达金标同等过门率(auto-vs-golden),不在本计划。
|
||
"retireReadyUnderGolden": equivalent,
|
||
"note": "本对照在金标 spec 下采证 = 框架等价;生产退役授权须另核对 WU-C 自动 spec 达金标同等(plan 对账表)",
|
||
},
|
||
}
|
||
|
||
|
||
def _pair_ids(genre_key: str, k: int) -> tuple:
|
||
"""一局两路的 gameId(前缀隔离:cheap-* vs node-*,品类×k 不撞)。"""
|
||
return f"cheap-{genre_key}-{k}", f"node-{genre_key}-{k}"
|
||
|
||
|
||
def _port_pool(conc: int, base_port: int = 4320, base_cdp: int = 9222) -> list:
|
||
"""每并发槽一对独立 (server_port, cdp_port):两两不撞、server 段(43xx)与 cdp 段(92xx)不重叠。
|
||
|
||
并发的真正约束 = 九门 play/smoke 的固定端口 + Chrome 实例;每槽独立端口才能并发不撞(gameId 前缀已隔离产物)。
|
||
"""
|
||
conc = max(1, conc)
|
||
return [(base_port + i * 2, base_cdp + i * 2) for i in range(conc)]
|
||
|
||
|
||
def _next_index(stems: list) -> int:
|
||
"""从已有 compare-multi-<N> 文件名算下一个序号(不覆盖历史)。"""
|
||
nums = [int(s.rsplit("-", 1)[-1]) for s in stems if s.rsplit("-", 1)[-1].isdigit()]
|
||
return (max(nums) + 1) if nums else 1
|
||
|
||
|
||
def _next_report_path() -> Path:
|
||
"""报告分批次文件、不覆盖历史(plan U2)。"""
|
||
_RESULTS_DIR.mkdir(exist_ok=True)
|
||
idx = _next_index([p.stem for p in _RESULTS_DIR.glob("compare-multi-*.json")])
|
||
return _RESULTS_DIR / f"compare-multi-{idx}.json"
|
||
|
||
|
||
def print_multi_report(report: dict) -> None:
|
||
print("\n=== 三品类对照报告:Python 新路 vs Node 旧路(金标 spec 下)===", file=sys.stderr)
|
||
for g in report["perGenre"]:
|
||
j = g["judge"]
|
||
flag = " ⛔RED" if j["doubleLowRed"] else ""
|
||
print(f"[{g['genre']}] n={g['n']} status={j['status']}{flag} "
|
||
f"py_rate={j['pyPassRate']} node_rate={j['nodePassRate']} "
|
||
f"py_play={j['pyPlayRatio']} node_play={j['nodePlayRatio']} — {j['reason']}", file=sys.stderr)
|
||
s = report["summary"]
|
||
print(f"\n>>> 坐实等价(金标 spec 下,Node 可退役就绪): {s['equivalentGenres']}", file=sys.stderr)
|
||
print(f">>> 对照不充分(标红): {s['inconclusiveGenres']} 回退: {s['regressedGenres']}", file=sys.stderr)
|
||
print(f">>> 注:{s['note']}\n", file=sys.stderr)
|
||
|
||
|
||
async def run_multi(genre_keys: list, n: int, conc: int = 1,
|
||
base_port: int = 4320, base_cdp: int = 9222) -> dict:
|
||
"""三品类 × n 对照:逐局生成→注入金标→play,按品类聚合。
|
||
|
||
conc=1 串行;conc>1 有界并发 —— 端口池(每槽独立 port/cdp)+ 信号量限并发(≤conc 同时跑),每对经
|
||
asyncio.to_thread 在独立线程跑(线程内 run_studio 自带 loop)。前台进程内有界并发,非后台子代理。
|
||
"""
|
||
n = _clamp_n(n)
|
||
conc = max(1, conc)
|
||
pool = asyncio.Queue() # 端口池:size=conc,get 空则阻塞 → 天然限并发到 ≤conc
|
||
for pp in _port_pool(conc, base_port, base_cdp):
|
||
pool.put_nowait(pp)
|
||
|
||
# 收集每品类的 runs(线程并发写不同 key 的 list,key 互斥、无共享写)。
|
||
genre_runs = {}
|
||
valid = []
|
||
for key in genre_keys:
|
||
genre = _GENRE_BY_KEY.get(key)
|
||
if genre is None:
|
||
print(f"[skip] 未知品类 {key}", file=sys.stderr)
|
||
continue
|
||
brief = load_brief(genre)
|
||
if not brief:
|
||
print(f"[skip] {key} 无 brief(base {genre['base']} run-summary 缺失)", file=sys.stderr)
|
||
continue
|
||
valid.append((key, genre, brief))
|
||
genre_runs[key] = []
|
||
|
||
async def one(key, genre, brief, k):
|
||
py_id, node_id = _pair_ids(key, k)
|
||
port, cdp = await pool.get() # 取一对独立端口(无空闲则等)
|
||
try:
|
||
print(f"[{key} {k + 1}/{n}] py={py_id} node={node_id} port={port}/{cdp} 生成→注入金标→play …",
|
||
file=sys.stderr)
|
||
r = await asyncio.to_thread(run_pair_sync, genre, brief, py_id, node_id, port, cdp)
|
||
genre_runs[key].append(r)
|
||
except Exception as e: # noqa: BLE001 单局失败不拖垮整批,记录后继续
|
||
print(f"[{key} {k + 1}/{n}] 异常:{type(e).__name__}: {e}", file=sys.stderr)
|
||
finally:
|
||
pool.put_nowait((port, cdp)) # 归还端口
|
||
|
||
tasks = [one(key, genre, brief, k) for (key, genre, brief) in valid for k in range(n)]
|
||
print(f"[run] {len(valid)} 品类 × n={n} = {len(tasks)} 对,conc={conc}(端口段 {base_port}.. / {base_cdp}..)",
|
||
file=sys.stderr)
|
||
await asyncio.gather(*tasks)
|
||
genre_runs = {k: v for k, v in genre_runs.items() if v} # 去掉全失败的品类
|
||
|
||
report = aggregate(genre_runs)
|
||
out = _next_report_path()
|
||
out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"[report] → {out}", file=sys.stderr)
|
||
print_multi_report(report)
|
||
return report
|
||
|
||
|
||
# ───────────────────────── 单品类对照(spike U6 起点,保留)─────────────────────────
|
||
|
||
async def run_python(brief: str, game_id: str) -> dict:
|
||
"""Python 新路单局(spike：run_studio 内部已跑九门)。"""
|
||
t0 = time.time()
|
||
summary = await cheap_studio.run_studio(game_id, brief)
|
||
return {"summary": summary, "wallSec": round(time.time() - t0, 1)}
|
||
|
||
|
||
def run_node(brief: str, game_id: str) -> dict:
|
||
"""Node 旧路单局(spike：gen.mjs --mode saa,再跑同款九门 play 用自动 spec)。"""
|
||
t0 = time.time()
|
||
gen = _gen_node(brief, game_id)
|
||
pr = cheap_run.play(game_id)
|
||
return {"nodeRC": gen["rc"], "wallSec": round(time.time() - t0, 1), "playVerdict": pr["verdict"], "raw": gen["raw"]}
|
||
|
||
|
||
def build_report(brief: str, py_id: str, node_id: str) -> dict:
|
||
py_v = _read_verdict(py_id)
|
||
node_v = _read_verdict(node_id)
|
||
py_gates = gates_from_verdict(py_v)
|
||
node_gates = gates_from_verdict(node_v)
|
||
py_shape = src_shape(cheap_run.game_dir(py_id))
|
||
node_shape = src_shape(cheap_run.game_dir(node_id))
|
||
cmp = compare_logic(py_gates, node_gates, py_shape["realSrcMultifile"])
|
||
return {
|
||
"brief": brief, "pyId": py_id, "nodeId": node_id,
|
||
"pyGates": py_gates, "nodeGates": node_gates,
|
||
"pyShape": py_shape, "nodeShape": node_shape,
|
||
"pyOverallPass": bool(py_v and py_v.get("pass")),
|
||
"nodeOverallPass": bool(node_v and node_v.get("pass")),
|
||
"compare": cmp,
|
||
}
|
||
|
||
|
||
# ───────────────────────── CLI ─────────────────────────
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
ap = argparse.ArgumentParser(description="便宜档 Python 路 vs Node 旧路三品类对照验证(金标 spec 下)")
|
||
ap.add_argument("--genres", default="click-score,whack-mole,shop-serve",
|
||
help="逗号分隔品类键(默认三 tap-targets 代表品类)")
|
||
ap.add_argument("--n", type=int, default=3, help="每品类重复次数(n>5 小批,上限夹 5)")
|
||
ap.add_argument("--conc", type=int, default=1, help="并发对数(1=串行;>1 端口池+线程有界并发,本机 Chrome 建议 ≤4)")
|
||
a = ap.parse_args()
|
||
keys = [k.strip() for k in a.genres.split(",") if k.strip()]
|
||
rep = asyncio.run(run_multi(keys, a.n, conc=a.conc))
|
||
# 退出码:有任一品类坐实等价即 0(机制验证,非达标判定)。
|
||
sys.exit(0 if rep["summary"]["equivalentGenres"] else 1)
|