Files
gplan-2/tools/balance_verifier.py
T
DKChan d5783a5740 refactor(tools): 测量面去单关直入口——难度读数改冒险条件口径(用户指令 2026-09-23)
- verifier_batch.gd: level 段(直入口单关基线 360 局/全量)整段移除——满血零上下文进关
  与真实冒险条件通关率脱节(直入口 L2 普通 5% vs 营地检查点 L2 条件 90% 同期并存),
  曾误导曲线调参;逐关读数由 adventure 段行内 levels_cleared/stopped_at 派生,零补采
- balance_verifier.py: 直入口验收带(LEVEL_BANDS)退役,新增①逐关条件通关率表
  (通关数/到达数,±95%CI,带待重锚只测不判——重锚归曲线调参批);营地检查点判定同批
  暂停(旧带值锚直入口读数不可比);--n 参数随段删除;--from-raw 兼容旧 raw(level 行忽略)
- 退役 level_monte_carlo.gd + level_debug.gd(单关扫描/单关诊断——同口径产 misleading 数据,
  替代面 = balance_verifier 冒险段 / adventure_monte_carlo)
- level_autoplay 头注:run_level 降格机制夹具/单战锚点专用——通关率/难度类测量禁用
- 保留面(非测量,零改动):smoke_all 机制夹具(治疗数学/选路状态机/L2 首遇战 12 种子锚行)、
  rhythm_batch/defense_scan/damage_probe 单战机制扫描——断言对象是具体机制非通关率数据
- 文档同步:难度曲线与Boss v0.19(带口径退役注记:调参表行+验收标准行+版本历史)/
  entities.yaml(clear_rate_target 两轮过期带值对齐退役注记+tool_carrier 换主载体)/
  CLAUDE.md(工具行两处)/ README(headless 工具节)/ 07-体验验证清单 / docs/code-conventions
  §5.2 / board_stat_probe 注释
- 验证:py_compile 过;冒烟 553 项 0 失败三闸全过;balance_verifier --quick 端到端 36 局
  2.7 分钟出报告(完美档逐关 87.5%→42.9%→0% 冒险条件口径首读数);doc 三闸失败集与
  master 逐项一致(lint 8/死链 7/term 1 均既有)
2026-09-23 05:24:13 +08:00

483 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""数值验证工具(判定端)——自动化评判 + 元进度爬升测量,替代「亲手试玩再反馈」循环。
两段采集(数据源 = smoke/verifier_batch.gd,本文件驱动它并做全部判定):
adventure 段 全冒险连探基线(零永久强化)——逐关条件通关率派生 + 止步分布 + 败退节点归因(层/节点/敌人)
campaign 段 营地循环战役(隔离档真实元进度账本:注入→冒险→收束入账→买锻造)——爬升曲线
评判标准(溯源):
① 逐关条件通关率(冒险口径派生——打通 k-1 关才到达第 k 关,通关数/到达数):单关直入口口径
(分带批 2026-09-20,难度曲线与Boss v0.16)满血零上下文进关,与真实冒险的条件通关率脱节,
曾误导曲线调参——已随单关试玩入口一并退役(2026-09-23 用户拍板);旧带值
(L1 普通 [35,55]/L2 普通 [25,45]/完美档 [35,65])锚定直入口读数不可比,
带重锚归曲线调参批——重锚前本表只测不判。
② 元循环目标(用户提案 2026-09-22,数字待确认):累计通过 5 次第 k 层的冒险后,第 k+1 层的
条件通关率(滑窗)应进入该层验收带——判定同①随口径退役暂停(重锚前只给滑窗读数与触发点)。
③ 魔法道具面:道具工坊六件按通关次数解锁(阈值 1/1/2/2/3/3),局内效果未实装(归 Alpha 魔法道具批)——
本工具只报解锁时点,不入判定。
表值硬编码溯源(单一事实源仍在 .tres / GDD,变更须同步):
· 锻造三档 flat(meta.tres):生命 +3 / 攻击 +1 / 护甲 +1
· 钥石入账(adventure.tres + meta.tres):每层通关 10 + Boss 首杀/重复杀 5/1 · 8/2 · 12/3
· 工坊阈值(meta.tres):显影尘1 · 工兵令1 · 探宝罗盘2 · 探巢摆钟2 · 地穴全图卷3 · 爆破工兵包3
用法:
python tools/balance_verifier.py # 全段默认规模
python tools/balance_verifier.py --quick # 小规模快跑(读数精度低)
python tools/balance_verifier.py --na 60 --k 60 --seed0 9000 --spend atk_first
python tools/balance_verifier.py --from-raw <raw.jsonl> # 免重跑,对既有采集重新判定/出报告
输出:控制台判定结论 + production/qa/balance-verifier-<时间戳>.md(报告)+ 同名 .raw.jsonl(原始逐局数据)。
兼容注:2026-09-23 前的旧 raw.jsonl 含 level 段(直入口单关)行——本版判定自动忽略,
逐关读数一律由冒险行派生。
"""
import argparse
import json
import math
import subprocess
import sys
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
SLICE = ROOT / "prototypes" / "dungeon-sweeper-vertical-slice"
GODOT = r"D:\code\godot\Godot_v4.7.2-stable_win64_console.exe"
SCRIPT = "res://smoke/verifier_batch.gd"
QA_DIR = ROOT / "production" / "qa"
VJSON_PREFIX = "VJSON "
ERROR_MARKERS = ("SCRIPT ERROR", "Parse Error")
# ---- 判定标准区(溯源见文件头) ----
# 直入口验收带(原 LEVEL_BANDS)已随单关直入口口径退役(2026-09-23 用户拍板)——
# 逐关/检查点判定全部暂停(judge_band 传 band=None = 只测不判),带重锚归曲线调参批。
CAMPAIGN_CLEARS_TARGET = 5 # 元循环检查点:累计通过第 k 层的冒险次数(用户提案 2026-09-22,待确认)
CAMPAIGN_WINDOW = 10 # 检查点滑窗局数(条件通关率的窗口)
CAMPAIGN_MIN_ATTEMPTS = 5 # 滑窗内少于此局数 = 数据不足(不出判定)
LEVEL_CN = {1: "第一层", 2: "第二层", 3: "第三层"}
NODE_TYPE_CN = {"battle": "战斗", "boss": "Boss", "rest": "恢复", "reward": "奖励", "shop": "商店房"}
FORGE_FLAT = {"max_hp": 3, "atk": 1, "armor_class": 1} # meta.tres 锻造三档每级 flat 值
WORKSHOP_CN = {"probe_dust": "显影尘", "sapper_order": "工兵令", "loot_compass": "探宝罗盘",
"nest_pendulum": "探巢摆钟", "catacomb_scroll": "地穴全图卷", "blast_kit": "爆破工兵包"}
def die(msg):
print(f"[数值验证] FAIL:{msg}")
sys.exit(1)
def disp_w(s):
return sum(2 if ord(c) > 0x2E7F else 1 for c in str(s))
def pad(s, w, right=False):
s = str(s)
gap = w - disp_w(s)
return (" " * gap + s) if right else (s + " " * gap)
# ---------------------------------------------------------------- 采集
def collect(args):
segs = args.only.split(",")
cmd = [args.godot, "--headless", "--path", str(SLICE), "--script", SCRIPT, "--",
f"na={args.na}", f"k={args.k}", f"seed0={args.seed0}",
f"only={args.only}", f"spend={args.spend}"]
print(f"[数值验证] 采集启动({' · '.join(segs)},约 {args.est_min:.0f} 分钟)…")
t0 = time.time()
try:
proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=args.timeout)
except subprocess.TimeoutExpired:
die(f"外部超时(>{args.timeout}s)——可加大 --timeout 或缩小规模")
out = (proc.stdout or "") + (proc.stderr or "")
if proc.returncode != 0:
print(out[-3000:])
die(f"采集进程退出码 {proc.returncode}")
hits = [ln.strip() for ln in out.splitlines() if any(k in ln for k in ERROR_MARKERS)]
if hits:
print("\n".join(hits[:5]))
die(f"脚本错误标记命中 {len(hits)} 行(脚本错误静默中断断言/采集,必须人工排查)")
rows, bad = [], 0
for ln in out.splitlines():
if ln.startswith(VJSON_PREFIX):
try:
rows.append(json.loads(ln[len(VJSON_PREFIX):]))
except json.JSONDecodeError:
bad += 1
if bad:
die(f"VJSON 行解析失败 {bad} 行(采集端输出契约被破坏)")
if not rows:
die("采集零数据行(verifier_batch.gd 未按契约输出)")
print(f"[数值验证] 采集完成:{len(rows)} 局,耗时 {(time.time() - t0) / 60:.1f} 分钟")
return rows, out
def load_raw(path):
rows = []
with open(path, encoding="utf-8") as f:
for ln in f:
ln = ln.strip()
if ln:
rows.append(json.loads(ln))
return rows
# ---------------------------------------------------------------- 判定件
def binom_ci(wins, n):
if n == 0:
return 0.0
p = wins / n
return 1.96 * math.sqrt(p * (1 - p) / n)
def judge_band(rate_pct, band):
"""落带判定:返回 (结论, 是否达标)。band=None = 带未定(直入口口径退役后待重锚),只报读数。"""
if band is None:
return "带未定(只测不判)", None
lo, hi = band
if rate_pct < lo:
return f"越下界(低于 {lo}% 达 {lo - rate_pct:.1f} 点)", False
if rate_pct > hi:
return f"越上界(高于 {hi}% 达 {rate_pct - hi:.1f} 点)", False
return f"落带 [{lo},{hi}] ✓", True
def node_label(defeat):
"""败退节点人读名:第X层·节点N·类型·敌人(敌人 = 死亡遭遇显示名,最多 3 个)。"""
if not defeat:
return "—"
t = NODE_TYPE_CN.get(defeat.get("node_type", ""), defeat.get("node_type", "?"))
s = f"{LEVEL_CN.get(defeat.get('level_index', -1), '?层')}·节点{defeat.get('node_id', '?')}·{t}"
enemies = defeat.get("enemies") or []
if enemies:
seen, names = set(), []
for e in enemies:
if e not in seen:
seen.add(e)
names.append(e)
s += "·" + "、".join(names[:3])
return s
# ---------------------------------------------------------------- 分段分析
def valid_adventure_runs(rows, policy):
"""冒险段有效局(summary 在场——levels_cleared >= 0);编排异常局单列。"""
runs = [r for r in rows if r.get("seg") == "adventure" and r.get("policy") == policy
and int(r.get("levels_cleared", -1)) >= 0]
invalid = sum(1 for r in rows if r.get("seg") == "adventure" and r.get("policy") == policy
and int(r.get("levels_cleared", -1)) < 0)
return runs, invalid
def analyze_adventure_levels(rows):
"""逐关条件通关率(冒险口径派生——直入口口径退役 2026-09-23):
到达第 k 关 ⇔ levels_cleared >= k-1;通关第 k 关 ⇔ levels_cleared >= k。
条件通关率 = 通关数 / 到达数(L1 的到达数 = 全部有效局——冒险即从第一关开始)。"""
out = {}
for policy in ("normal", "perfect"):
runs, invalid = valid_adventure_runs(rows, policy)
levels = {}
for k in (1, 2, 3):
reached = sum(1 for r in runs if int(r.get("levels_cleared", 0)) >= k - 1)
cleared = sum(1 for r in runs if int(r.get("levels_cleared", 0)) >= k)
levels[k] = {"reached": reached, "cleared": cleared}
out[policy] = {"n": len(runs), "invalid": invalid, "levels": levels}
return out
def analyze_adventure(rows):
out = {}
for policy in ("normal", "perfect"):
runs, invalid = valid_adventure_runs(rows, policy)
stopped, death_nodes = {}, {}
for r in runs:
sa = int(r.get("stopped_at", -1))
if sa == 0:
stopped["通关"] = stopped.get("通关", 0) + 1
elif sa > 0:
stopped[f"止步第{sa}层"] = stopped.get(f"止步第{sa}层", 0) + 1
d = r.get("defeat") or {}
if d:
lb = node_label(d)
death_nodes[lb] = death_nodes.get(lb, 0) + 1
out[policy] = {"n": len(runs), "invalid": invalid, "stopped": stopped, "death_nodes": death_nodes}
return out
def campaign_series(runs):
"""逐局爬升序列:条件通关尝试流(第 k 层的尝试 = levels_cleared ≥ k-1 的局,按局序)。"""
return {
k: [(r["run_no"], int(r.get("levels_cleared", -1)) >= k, r)
for r in sorted(runs, key=lambda x: x["run_no"])
if int(r.get("levels_cleared", -1)) >= k - 1]
for k in (1, 2, 3)
}
def window_rate(series, end_idx, window):
"""尝试流截至 end_idx(含)往前 window 次尝试的通关率(%)。"""
chunk = series[max(0, end_idx - window + 1): end_idx + 1]
if not chunk:
return None, 0
return 100.0 * sum(1 for _, ok, _ in chunk if ok) / len(chunk), len(chunk)
def analyze_campaign(rows):
out = {}
for policy in ("normal", "perfect"):
runs = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
and not r.get("anomaly")]
anomalies = [r for r in rows if r.get("seg") == "campaign" and r.get("policy") == policy
and r.get("anomaly")]
res = {"n": len(runs), "anomalies": len(anomalies), "runs": runs}
if not runs:
out[policy] = res
continue
series = campaign_series(runs)
res["series"] = series
# 检查点:累计通过第 k 层达目标后,第 k+1 层滑窗条件通关率——带待重锚(直入口口径退役)只报读数
checkpoints = []
for k in (1, 2):
clears = 0
trigger_run = None
for r in sorted(runs, key=lambda x: x["run_no"]):
if int(r.get("levels_cleared", 0)) >= k:
clears += 1
if clears >= CAMPAIGN_CLEARS_TARGET and trigger_run is None:
trigger_run = r["run_no"]
cp = {"level": k, "trigger_run": trigger_run, "clears": clears}
if trigger_run is None:
cp["verdict"] = f"数据不足:全程通过{LEVEL_CN[k]}仅 {clears} 次(目标 {CAMPAIGN_CLEARS_TARGET})——加大 --k 重跑"
else:
attempts = [(rn, ok, r) for rn, ok, r in series.get(k + 1, []) if rn > trigger_run]
chunk = attempts[:CAMPAIGN_WINDOW]
if len(chunk) < CAMPAIGN_MIN_ATTEMPTS:
cp["verdict"] = (f"触发于第 {trigger_run} 局,但其后通过{LEVEL_CN[k + 1]}的尝试仅 "
f"{len(chunk)} 局(不足 {CAMPAIGN_MIN_ATTEMPTS})——加大 --k 重跑")
else:
rate = 100.0 * sum(1 for _, ok, _ in chunk if ok) / len(chunk)
cp["rate"] = rate
cp["attempts"] = len(chunk)
cp["verdict"], cp["pass"] = judge_band(rate, None) # 带重锚归曲线调参批
cp["forge_at"] = chunk[0][2].get("override", {})
checkpoints.append(cp)
res["checkpoints"] = checkpoints
# 工坊解锁时点(魔法道具面——只报不判)
unlocks = {}
for r in runs:
for iid in r.get("new_unlocks", []):
unlocks[WORKSHOP_CN.get(iid, iid)] = r["run_no"]
res["unlocks"] = unlocks
# 累计钥石与锻造终态
last = max(runs, key=lambda x: x["run_no"])
res["forge_final"] = last.get("forge", {})
res["keystone_final"] = int(last.get("keystone_after", -1))
res["income_total"] = sum(int(r.get("keystone_income", 0)) for r in runs)
out[policy] = res
return out
# ---------------------------------------------------------------- 报告
def fmt_table(header, rows):
widths = [max(disp_w(h), max((disp_w(r[i]) for r in rows), default=0)) for i, h in enumerate(header)]
def line(cells):
return "| " + " | ".join(pad(c, w) for c, w in zip(cells, widths)) + " |"
sep = "|" + "|".join("-" * (w + 2) for w in widths) + "|"
out = [line(header), sep]
out += [line(r) for r in rows]
return "\n".join(out)
def report(args, rows, meta):
lines = []
A = lines.append
A(f"# 数值验证报告({meta['ts']})")
A("")
A(f"- 采集规模:基线冒险 {args.na} 局/画像 · 战役 {args.k} 局/画像"
f" · 种子 {args.seed0} 起逐局独立 · 锻造花费策略 = {args.spend}")
A(f"- 切片版本:{meta['commit']} · 口径:逐关读数 = 冒险条件口径(单关直入口口径已退役 2026-09-23,"
f"带待重锚只测不判——重锚归曲线调参批)")
A("")
# ① 逐关条件通关率(冒险口径派生)
A("## ① 逐关条件通关率(冒险口径——打通 k-1 关才到达第 k 关;带待重锚只测不判)")
A("")
alv = analyze_adventure_levels(rows)
tbl = []
for policy in ("normal", "perfect"):
d = alv[policy]
cn = "普通档" if policy == "normal" else "完美档"
for k in (1, 2, 3):
lv = d["levels"][k]
if lv["reached"] <= 0:
tbl.append([LEVEL_CN[k], cn, f"—/{d['n']}", "—(未到达)", "未定"])
continue
rate = 100.0 * lv["cleared"] / lv["reached"]
ci = binom_ci(lv["cleared"], lv["reached"]) * 100
tbl.append([LEVEL_CN[k], cn, f"{lv['cleared']}/{lv['reached']}",
f"{rate:.1f}% ±{ci:.1f}", "未定"])
A(fmt_table(["层", "画像", "通关/到达", "条件通关率±95%CI", "验收带"], tbl))
A("")
# ② 冒险基线:止步分布与败退节点
A("## ② 零永久强化冒险:止步分布与败退节点(用户判据③)")
A("")
adv = analyze_adventure(rows)
for policy in ("normal", "perfect"):
d = adv[policy]
cn = "普通档" if policy == "normal" else "完美档"
A(f"### {cn}({d['n']} 局)")
A("")
stop_line = "、".join(f"{k} {v}({100.0 * v / max(1, d['n']):.0f}%)" for k, v in sorted(d["stopped"].items()))
A(f"- 止步分布:{stop_line or '—'}")
A("- 败退节点分布(失败局归因):")
for lb, c in sorted(d["death_nodes"].items(), key=lambda x: -x[1]):
A(f" - {lb} ×{c}")
A("")
# ③ 元循环爬升
A("## ③ 营地循环爬升(用户判据④:5 次通关 → 下一层可过)")
A("")
camp = analyze_campaign(rows)
for policy in ("normal", "perfect"):
d = camp[policy]
cn = "普通档" if policy == "normal" else "完美档"
A(f"### {cn}({d['n']} 局战役)")
A("")
if not d["runs"]:
A("- 无数据")
A("")
continue
A(f"- 钥石:全程入账 {d['income_total']} · 终态余额 {d['keystone_final']}"
f" · 锻造终态 = 生命+{d['forge_final'].get('max_hp', 0) * FORGE_FLAT['max_hp']}"
f"/攻击+{d['forge_final'].get('atk', 0) * FORGE_FLAT['atk']}"
f"/护甲+{d['forge_final'].get('armor_class', 0) * FORGE_FLAT['armor_class']}")
if d["unlocks"]:
A("- 道具工坊解锁时点(魔法道具面,局内效果未实装只报不判):" +
"、".join(f"{k} 第 {v} 局" for k, v in sorted(d["unlocks"].items(), key=lambda x: x[1])))
else:
A("- 道具工坊解锁时点:无(未达阈值)")
if d.get("anomalies"):
A(f"- ⚠ 编排异常局 {d['anomalies']} 局(未结算零入账——timeout/advance_failed,须排查)")
A("")
for cp in d.get("checkpoints", []):
A(f"- 判据:累计通过{LEVEL_CN[cp['level']]} {CAMPAIGN_CLEARS_TARGET} 次后,{LEVEL_CN[cp['level'] + 1]}"
f"条件通关率(滑窗 {CAMPAIGN_WINDOW} 局)——{cp['verdict']}"
+ (f"(触发时注入 = 生命+{cp['forge_at'].get('max_hp', 0)}/攻击+{cp['forge_at'].get('atk', 0)}"
f"/护甲+{cp['forge_at'].get('armor_class', 0)})" if "forge_at" in cp else ""))
A("")
tbl = []
series = d.get("series", {})
for r in d["runs"][::max(1, len(d["runs"]) // 20)]:
cells = [f"第{r['run_no']}局",
f"血{r['override'].get('max_hp', 0)}/攻{r['override'].get('atk', 0)}/甲{r['override'].get('armor_class', 0)}",
str(r.get("keystone_after", "-"))]
for k in (1, 2, 3):
s = series.get(k, [])
idx = next((i for i, (rn, _, _) in enumerate(s) if rn == r["run_no"]), None)
if idx is None:
cells.append("—")
else:
rate, cnt = window_rate(s, idx, CAMPAIGN_WINDOW)
cells.append(f"{rate:.0f}%(n={cnt})" if rate is not None else "—")
sa = int(r.get("stopped_at", -1))
cells.append("通关" if sa == 0 else (f"止步第{sa}层" if sa > 0 else str(r.get("result", "?"))))
tbl.append(cells)
A(fmt_table(["局", "注入(flat)", "钥石", "第1层率(窗)", "第2层率(窗)", "第3层率(窗)", "终局"], tbl))
A("")
return "\n".join(lines)
def console_verdict(args, rows):
"""控制台判定结论(先读数后细节)。"""
alv = analyze_adventure_levels(rows)
print("\n判定结论(detail 见报告文件)")
for policy in ("normal", "perfect"):
d = alv[policy]
cn = "普通档" if policy == "normal" else "完美档"
cells = []
for k in (1, 2, 3):
lv = d["levels"][k]
if lv["reached"] > 0:
rate = 100.0 * lv["cleared"] / lv["reached"]
cells.append(f"{LEVEL_CN[k]} {rate:.1f}%({lv['cleared']}/{lv['reached']})")
print(f" [·] 逐关条件通关率·{cn}:{' · '.join(cells) or '—'}(冒险口径——带待重锚只测不判)")
adv = analyze_adventure(rows)
for policy in ("normal", "perfect"):
d = adv[policy]
cn = "普通档" if policy == "normal" else "完美档"
top = sorted(d["death_nodes"].items(), key=lambda x: -x[1])[:2]
print(f" [·] 零强化冒险·{cn}:止步分布 {'、'.join(f'{k}{v}' for k, v in sorted(d['stopped'].items()))}"
+ (f";主要败退点 = {';'.join(f'{lb} ×{c}' for lb, c in top)}" if top else ""))
camp = analyze_campaign(rows)
for policy in ("normal", "perfect"):
d = camp[policy]
cn = "普通档" if policy == "normal" else "完美档"
if not d["runs"]:
print(f" [·] 营地循环·{cn}:无数据")
continue
for cp in d.get("checkpoints", []):
mark = {True: "✓", False: "✗", None: "·"}.get(cp.get("pass"), "·")
print(f" [{mark}] 营地循环·{cn}·{LEVEL_CN[cp['level']]}→{LEVEL_CN[cp['level'] + 1]}:{cp['verdict']}")
print()
# ---------------------------------------------------------------- 入口
def main():
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
ap = argparse.ArgumentParser(description="数值验证工具(判定端)")
ap.add_argument("--godot", default=GODOT)
ap.add_argument("--na", type=int, default=60, help="基线冒险局数/画像(±95%CI 约 ±9 点——判读须看 CI)")
ap.add_argument("--k", type=int, default=60, help="战役局数/画像")
ap.add_argument("--seed0", type=int, default=8000)
ap.add_argument("--only", default="adventure,campaign")
ap.add_argument("--spend", default="hp_first", choices=["hp_first", "atk_first"])
ap.add_argument("--timeout", type=int, default=5400)
ap.add_argument("--quick", action="store_true", help="小规模快跑(8/10)")
ap.add_argument("--from-raw", default=None, help="跳过采集,对既有 raw.jsonl 重新判定出报告")
args = ap.parse_args()
if args.quick:
args.na, args.k = 8, 10
args.est_min = (2 * args.na * 3.5 + 2 * args.k * 4.0) / 60
ts = time.strftime("%Y%m%d-%H%M%S")
if args.from_raw:
rows = load_raw(args.from_raw)
raw_path = Path(args.from_raw)
print(f"[数值验证] 已载入既有采集 {len(rows)} 局:{raw_path}")
else:
rows, _ = collect(args)
QA_DIR.mkdir(parents=True, exist_ok=True)
raw_path = QA_DIR / f"balance-verifier-{ts}.raw.jsonl"
with open(raw_path, "w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
commit = "?"
try:
commit = subprocess.run(["git", "-C", str(SLICE), "rev-parse", "--short", "HEAD"],
capture_output=True, text=True).stdout.strip()
except OSError:
pass
meta = {"ts": ts, "commit": commit}
console_verdict(args, rows)
md = report(args, rows, meta)
if args.from_raw:
md_path = raw_path.with_suffix(".md")
else:
md_path = QA_DIR / f"balance-verifier-{ts}.md"
md_path.write_text(md, encoding="utf-8")
print(f"[数值验证] 报告:{md_path}")
print(f"[数值验证] 原始逐局数据:{raw_path}(--from-raw 可免重跑重判)")
return 0
if __name__ == "__main__":
sys.exit(main())