629 lines
29 KiB
Python
629 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""试玩日志分析器（垂直切片 tools/ 专用，纯标准库，零第三方依赖）。
|
||
|
||
消费 presentation/playtest_logger.gd 产生的 playtest_*.jsonl，
|
||
产出 关卡 / 战斗 / 轮次 三级统计，并为护甲等级（AC）掷点批预留验收统计面。
|
||
|
||
已知测量盲区（本工具的核心补偿，勿绕过）：
|
||
settlement 快照在衰减段之后发出、且现有状态寿命全部为 1 结算
|
||
→ 快照 buffs 字段恒为空。战斗期真实增益只能从 step 行（type=effect，
|
||
文本 增伤/削弱/减伤/…）按结算轮窗口复原。本工具从不把快照 buffs
|
||
当战斗期状态使用，只将其并列输出以持续暴露盲区。
|
||
|
||
AC 掷点批预留位：
|
||
落批后伤害 step 行会新增掷点字段（读数/档位/暴击，字段名以落批实际为准）。
|
||
本工具按 FIELD_CANDIDATES 探测：字段存在 → 激活掷点统计面；
|
||
不存在（AC 批前旧日志）→ 降级输出"不适用（AC 批前日志）"。
|
||
同时所有 step 行的未知键都会进"数据质量"观测面——落批后先看那里
|
||
拿到真实字段名，再在 FIELD_CANDIDATES 里补一行适配即可。
|
||
|
||
用法：
|
||
python log_analysis.py <jsonl 文件 | 目录 | 通配> [--per-run] [--rounds]
|
||
目录/通配 = 批量汇总；--per-run 附加逐局明细；--rounds 附加逐轮流水。
|
||
"""
|
||
|
||
import argparse
|
||
import glob
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from collections import Counter, defaultdict
|
||
|
||
# ---------------------------------------------------------------- 常量与映射
|
||
|
||
# step 行已知键（schema 事实源 = playtest_logger.gd _on_step_row）
|
||
STEP_KNOWN_KEYS = {"kind", "type", "source", "cell", "round", "text", "value", "highlight"}
|
||
|
||
# AC 掷点批候选字段名（落批后按实际字段名在此登记——预留的一行适配点）
|
||
ROLL_FIELD_CANDIDATES = ("roll", "d20", "dice", "nat", "nat_roll", "roll_value")
|
||
TIER_FIELD_CANDIDATES = ("tier", "discount", "discount_tier", "damage_tier", "bracket")
|
||
CRIT_FIELD_CANDIDATES = ("crit", "critical", "is_crit", "nat20")
|
||
AC_FIELD_CANDIDATES = ("ac", "target_ac", "defense_ac")
|
||
|
||
# effect 行显示文本 → 增益语义键（未登记文本原样作键，新增益自动浮现）
|
||
TEXT_TO_BUFF = {"增伤": "attack_up", "削弱": "weaken", "减伤": "defend_up"}
|
||
|
||
# 档位语义：缺口 = AC − 掷点读数（读数达标 = 全额）
|
||
TIER_MULTIPLIER = {"全额": 1.0, "×0.75": 0.75, "×0.5": 0.5, "×0.25": 0.25}
|
||
THEORY_EV_REDUCTION_AC10 = 0.2125 # AC10 理论 EV 减免（任务口径）
|
||
THEORY_CRIT_RATE = 0.05 # nat20 暴击率理论值
|
||
CRIT_TOLERANCE = 0.02 # 健康检查宽容度 ±2 个百分点
|
||
|
||
BAND_NORMAL = 0.20 # 承伤带：普通场 ≤20% max_hp
|
||
BAND_BOSS = 0.50 # 承伤带：Boss 场 ≤50% max_hp
|
||
|
||
BATTLE_TYPES = ("battle", "boss")
|
||
|
||
|
||
def _first(row, candidates):
|
||
"""返回 row 中第一个命中的候选字段值。"""
|
||
for c in candidates:
|
||
if c in row:
|
||
return row[c]
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- 逐局状态机
|
||
|
||
class BattleAcc:
|
||
"""一场战斗（battle/boss 节点）的累计器。"""
|
||
|
||
def __init__(self, node_type, depth):
|
||
self.node_type = node_type
|
||
self.depth = depth
|
||
self.rounds = [] # 每结算一个 dict（见 RunAcc._finalize_round）
|
||
self.damage = 0 # Σ 敌方攻击落地伤害（enemy_action × type=damage 的 value）
|
||
self.heals = 0 # Σ 主角治疗（type=heal 的 value）
|
||
self.attack_rounds = 0 # 有敌方攻击落地的轮数
|
||
self.max_hp = None
|
||
self.start_hp = None # 首结算快照 HP + 首轮承伤 − 首轮治疗 反推的进场 HP
|
||
self.last_hp = None
|
||
|
||
|
||
class RunAcc:
|
||
"""一局（run 行切分；无 run 行的旧日志 = 整文件一局，种子取 session 行）。"""
|
||
|
||
def __init__(self, src_file, session_seed):
|
||
self.src_file = src_file
|
||
self.session_seed = session_seed
|
||
self.seed = session_seed
|
||
self.policy = ""
|
||
self.level_index = None
|
||
self.result = None
|
||
self.end_hp = None
|
||
self.max_hp = None
|
||
self.round_steps = [] # 当前结算轮窗口内的 step 行缓冲
|
||
self.battle = None
|
||
self.battles = []
|
||
self.battle_fifo = [] # battle_end 延迟发出且不带 depth，按 FIFO 对账
|
||
self.late_battle_ends = 0 # 未能配对的 battle_end（数据质量）
|
||
self.boss_node_seen = False
|
||
self.rounds_total = 0
|
||
self.rounds_with_enemy = 0
|
||
self.damage_total = 0
|
||
self.heals_total = 0
|
||
self.zero_narrow = 0 # 零承伤轮（窄口径，AC 验收主指标）
|
||
self.attack_landed_rounds = 0
|
||
self.zero_loss_wide = 0 # 零掉血轮（宽口径，参考）
|
||
self.snapshot_buffs_nonempty = 0 # 盲区暴露面：快照 buffs 恒空计数
|
||
self.snapshot_buffs_total = 0
|
||
self.round_ledger = [] # --rounds 逐轮流水
|
||
self.roll_rows = [] # 命中掷点候选字段的 step 行
|
||
self.refresh_count = 0 # 受理的弃图刷新次数（action 行 op=refresh accepted=True）
|
||
self._prev_hp = None # 上一结算快照的主角 HP（宽口径逐轮差分用）
|
||
self._last_enemy_present = False
|
||
|
||
# -- 事件入口 ---------------------------------------------------------
|
||
|
||
def on_run_row(self, e):
|
||
self.seed = e.get("seed", self.session_seed)
|
||
self.policy = "" if e.get("policy") is None else str(e.get("policy"))
|
||
self.level_index = e.get("level_index")
|
||
|
||
def on_node(self, e):
|
||
t = e.get("type", "")
|
||
if t in BATTLE_TYPES:
|
||
# 先收掉上一场（战斗的 settlement 严格落在相邻 node 行之间）
|
||
self._close_battle()
|
||
self.battle = BattleAcc(t, e.get("depth"))
|
||
if t == "boss":
|
||
self.boss_node_seen = True
|
||
# rest / reward 节点不切战斗上下文（与 battle 同 index 出现）
|
||
|
||
def on_step(self, e):
|
||
self.round_steps.append(e)
|
||
if any(c in e for c in ROLL_FIELD_CANDIDATES):
|
||
self.roll_rows.append(e)
|
||
|
||
def on_settlement(self, e):
|
||
b = self.battle
|
||
self.rounds_total += 1
|
||
hero = e.get("hero", {})
|
||
hp = int(hero.get("hp", 0))
|
||
if hero.get("max_hp"):
|
||
self.max_hp = int(hero.get("max_hp"))
|
||
enemy_present = bool(e.get("enemies"))
|
||
if enemy_present:
|
||
self.rounds_with_enemy += 1
|
||
# 盲区暴露面：快照 buffs 只计数对照，从不当作战斗期状态
|
||
self.snapshot_buffs_total += 1
|
||
if e.get("buffs"):
|
||
self.snapshot_buffs_nonempty += 1
|
||
self._finalize_round(e, b, hp, enemy_present)
|
||
|
||
def on_battle_end(self, e):
|
||
# 延迟发出、不带 depth：只做 FIFO 对账（数据质量），核心统计不依赖它
|
||
if self.battle_fifo:
|
||
self.battle_fifo.pop(0)
|
||
else:
|
||
self.late_battle_ends += 1
|
||
|
||
def on_level_end(self, e):
|
||
self._close_battle()
|
||
self.result = e.get("result", "")
|
||
self.end_hp = int(e.get("hp", 0))
|
||
|
||
def on_action(self, e):
|
||
# 刷新经济监控（07 清单⑦）：只计受理成功的弃图刷新——被拒尝试不构成弃图。
|
||
if e.get("op") == "refresh" and e.get("accepted"):
|
||
self.refresh_count += 1
|
||
|
||
# -- 轮窗口收口（核心口径都在这里）------------------------------------
|
||
|
||
def _finalize_round(self, snap, battle, hp, enemy_present):
|
||
steps, self.round_steps = self.round_steps, []
|
||
dmg = sum(int(s.get("value", 0)) for s in steps
|
||
if s.get("type") == "damage" and s.get("source") == "enemy_action")
|
||
heal = sum(int(s.get("value", 0)) for s in steps if s.get("type") == "heal")
|
||
attacks = sum(1 for s in steps
|
||
if s.get("type") == "damage" and s.get("source") == "enemy_action")
|
||
charges = sum(1 for s in steps if s.get("type") == "charge")
|
||
# 蓄力拍（type=charge，含空拍/白喂）不算"攻击落地"——验收口径硬性排除。
|
||
# 战斗期增益复原：本结算轮窗口内的 effect 行（快照 buffs 恒空，禁用）
|
||
buffs = Counter()
|
||
for s in steps:
|
||
if s.get("type") == "effect":
|
||
key = TEXT_TO_BUFF.get(s.get("text", ""), s.get("text", ""))
|
||
buffs[key or "<未登记effect文本>"] += 1
|
||
|
||
# 宽口径（整轮掉血 0）：首轮用进场反推 HP 差分，之后用相邻快照差
|
||
if self._prev_hp is None:
|
||
entry_hp = hp + dmg - heal
|
||
loss = entry_hp - hp
|
||
else:
|
||
loss = max(0, self._prev_hp - hp)
|
||
self._prev_hp = hp
|
||
|
||
zero_narrow = attacks > 0 and dmg == 0 # 窄口径：有攻击落地且掉血 0
|
||
zero_wide = loss == 0 # 宽口径：整轮 HP 不降（聚合时限敌战轮）
|
||
|
||
self.damage_total += dmg
|
||
self.heals_total += heal
|
||
if attacks > 0:
|
||
self.attack_landed_rounds += 1
|
||
if zero_narrow:
|
||
self.zero_narrow += 1
|
||
if zero_wide:
|
||
self.zero_loss_wide += 1
|
||
|
||
rec = {
|
||
"type": battle.node_type if battle else "?",
|
||
"depth": snap.get("depth"), "round": snap.get("round"),
|
||
"dmg": dmg, "heal": heal, "hp": hp,
|
||
"attacks": attacks, "charges": charges,
|
||
"zero_narrow": zero_narrow, "zero_wide": zero_wide,
|
||
"buffs": dict(buffs),
|
||
"path": (snap.get("round_facts") or {}).get("path", ""),
|
||
"enemy_present": enemy_present,
|
||
}
|
||
self.round_ledger.append(rec)
|
||
if battle is not None:
|
||
battle.rounds.append(rec)
|
||
battle.damage += dmg
|
||
battle.heals += heal
|
||
battle.attack_rounds += 1 if attacks > 0 else 0
|
||
battle.max_hp = self.max_hp
|
||
battle.last_hp = hp
|
||
if battle.start_hp is None:
|
||
battle.start_hp = hp + dmg - heal
|
||
|
||
def _close_battle(self):
|
||
if self.battle is None:
|
||
return
|
||
b, self.battle = self.battle, None
|
||
if b.rounds:
|
||
# 有过敌人的战斗（任一轮出现过敌人或攻击落地）——宽口径分母用其全部轮数。
|
||
# 不能按单轮快照 enemies 判定：敌人当轮被歼的轮快照为空，但轮仍属于敌战。
|
||
b.enemy_flag = any(r["attacks"] > 0 or r["enemy_present"] for r in b.rounds)
|
||
for rec in b.rounds:
|
||
rec["enemy_battle"] = b.enemy_flag
|
||
self.battles.append(b)
|
||
self.battle_fifo.append(b)
|
||
|
||
# -- 派生量 -----------------------------------------------------------
|
||
|
||
def boss_battle(self):
|
||
for b in self.battles:
|
||
if b.node_type == "boss":
|
||
return b
|
||
return None
|
||
|
||
def boss_enter_hp(self):
|
||
"""进 Boss HP = Boss 首结算快照 HP + 首轮承伤 − 首轮治疗（精确反推）。"""
|
||
b = self.boss_battle()
|
||
if b is None or not b.rounds:
|
||
return None
|
||
r0 = b.rounds[0]
|
||
return r0["hp"] + r0["dmg"] - r0["heal"]
|
||
|
||
|
||
# ---------------------------------------------------------------- 文件解析
|
||
|
||
class FileResult:
|
||
def __init__(self, path):
|
||
self.path = path
|
||
self.lines = 0
|
||
self.bad_lines = 0
|
||
self.parse_seconds = 0.0
|
||
self.session_seeds = []
|
||
self.runs = []
|
||
self.unknown_step_keys = Counter() # 未知 step 键 → 次数（AC 落批观察面）
|
||
self.unknown_step_samples = {} # 未知 step 键 → 样例值
|
||
self.dice_active = False
|
||
|
||
|
||
def parse_file(path):
|
||
res = FileResult(path)
|
||
t0 = time.perf_counter()
|
||
run = None
|
||
with open(path, encoding="utf-8", errors="replace") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
res.lines += 1
|
||
try:
|
||
e = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
res.bad_lines += 1
|
||
continue
|
||
kind = e.get("kind", "")
|
||
if kind == "session":
|
||
res.session_seeds.append(e.get("seed"))
|
||
if run is None:
|
||
run = RunAcc(os.path.basename(path), e.get("seed"))
|
||
res.runs.append(run)
|
||
elif kind == "run":
|
||
if run is not None and (run.rounds_total or run.result or run.battles):
|
||
run = RunAcc(os.path.basename(path), res.session_seeds[-1] if res.session_seeds else None)
|
||
res.runs.append(run)
|
||
elif run is None:
|
||
run = RunAcc(os.path.basename(path), None)
|
||
res.runs.append(run)
|
||
run.on_run_row(e)
|
||
elif run is None:
|
||
# 无 session/run 头的残缺文件：兜底开一局，保证数据不丢
|
||
run = RunAcc(os.path.basename(path), None)
|
||
res.runs.append(run)
|
||
_feed(run, res, kind, e)
|
||
else:
|
||
_feed(run, res, kind, e)
|
||
if run is not None:
|
||
run._close_battle()
|
||
res.parse_seconds = time.perf_counter() - t0
|
||
res.dice_active = any(r.roll_rows for r in res.runs)
|
||
return res
|
||
|
||
|
||
def _feed(run, res, kind, e):
|
||
if kind == "node":
|
||
run.on_node(e)
|
||
elif kind == "step":
|
||
run.on_step(e)
|
||
for k in e:
|
||
if k not in STEP_KNOWN_KEYS:
|
||
res.unknown_step_keys[k] += 1
|
||
res.unknown_step_samples.setdefault(k, repr(e[k])[:60])
|
||
elif kind == "settlement":
|
||
run.on_settlement(e)
|
||
elif kind == "battle_end":
|
||
run.on_battle_end(e)
|
||
elif kind == "level_end":
|
||
run.on_level_end(e)
|
||
elif kind == "action":
|
||
run.on_action(e)
|
||
# ux / diagnosis 等其余 kind：本工具当前无消费面，静默跳过
|
||
|
||
|
||
def resolve_inputs(arg):
|
||
"""文件 / 目录 / 通配 → 排序后的 playtest_*.jsonl 列表。"""
|
||
if os.path.isdir(arg):
|
||
return sorted(glob.glob(os.path.join(arg, "playtest_*.jsonl")))
|
||
if any(ch in arg for ch in "*?["):
|
||
return sorted(glob.glob(arg))
|
||
return [arg]
|
||
|
||
|
||
# ---------------------------------------------------------------- 聚合与渲染
|
||
|
||
def fmt(x, nd=1):
|
||
return "—" if x is None else f"{x:.{nd}f}"
|
||
|
||
|
||
def aggregate_levels(runs):
|
||
"""关卡级：按（关卡, 档位）聚合。"""
|
||
groups = defaultdict(list)
|
||
for r in runs:
|
||
groups[(r.level_index, r.policy)].append(r)
|
||
rows = []
|
||
for (lvl, policy), rs in sorted(groups.items(), key=lambda kv: (str(kv[0][0]), str(kv[0][1]))):
|
||
n = len(rs)
|
||
wins = sum(1 for r in rs if r.result == "victory")
|
||
boss_rs = [r for r in rs if r.boss_enter_hp() is not None]
|
||
win_rs = [r for r in rs if r.result == "victory"]
|
||
total_rounds = sum(r.rounds_total for r in rs)
|
||
total_refresh = sum(r.refresh_count for r in rs)
|
||
rows.append({
|
||
"level": lvl, "policy": policy or "—", "runs": n,
|
||
"clear_rate": wins / n if n else None,
|
||
"avg_rounds": sum(r.rounds_total for r in rs) / n if n else None,
|
||
"avg_boss_hp": (sum(r.boss_enter_hp() for r in boss_rs) / len(boss_rs)) if boss_rs else None,
|
||
"avg_end_hp": (sum(r.end_hp for r in win_rs) / len(win_rs)) if win_rs else None,
|
||
"avg_dmg": sum(r.damage_total for r in rs) / n if n else None,
|
||
"refresh_per_run": total_refresh / n if n else None, # 弃图率（与 MC 刷新/局同口径）
|
||
"refresh_per_round": total_refresh / total_rounds if total_rounds else None, # 每轮刷新率
|
||
})
|
||
return rows
|
||
|
||
|
||
def aggregate_battles(runs):
|
||
"""战斗级：按（关卡, 节点类型, 深度）分组，普通场与 Boss 场分开对照承伤带。"""
|
||
groups = defaultdict(list)
|
||
for r in runs:
|
||
for b in r.battles:
|
||
if b.max_hp:
|
||
groups[(r.level_index, b.node_type, b.depth)].append(b)
|
||
rows = []
|
||
for (lvl, ntype, depth), bs in sorted(
|
||
groups.items(), key=lambda kv: (str(kv[0][0]), kv[0][1], str(kv[0][2]))):
|
||
n = len(bs)
|
||
avg_dmg = sum(b.damage for b in bs) / n
|
||
avg_ratio = sum(b.damage / b.max_hp for b in bs) / n
|
||
band = BAND_BOSS if ntype == "boss" else BAND_NORMAL
|
||
rows.append({
|
||
"level": lvl, "type": ntype, "depth": depth, "count": n,
|
||
"avg_dmg": avg_dmg, "avg_ratio": avg_ratio,
|
||
"band": band, "pass": avg_ratio <= band,
|
||
"avg_rounds": sum(len(b.rounds) for b in bs) / n,
|
||
"avg_attack_rounds": sum(b.attack_rounds for b in bs) / n,
|
||
})
|
||
return rows
|
||
|
||
|
||
def aggregate_rounds(runs):
|
||
t = sum(r.rounds_total for r in runs)
|
||
e = sum(r.rounds_with_enemy for r in runs)
|
||
a = sum(r.attack_landed_rounds for r in runs)
|
||
zn = sum(r.zero_narrow for r in runs)
|
||
# 宽口径分母 = 有过敌人的战斗的全部轮数（含敌人当轮被歼的收尾轮）
|
||
enemy_battle_rounds = 0
|
||
zw = 0
|
||
for r in runs:
|
||
for b in r.battles:
|
||
if getattr(b, "enemy_flag", False):
|
||
enemy_battle_rounds += len(b.rounds)
|
||
zw += sum(1 for rec in b.rounds if rec["zero_wide"])
|
||
return {"rounds": t, "with_enemy": e, "attack_landed": a,
|
||
"zero_narrow": zn, "zero_loss_wide": zw,
|
||
"enemy_battle_rounds": enemy_battle_rounds}
|
||
|
||
|
||
def derive_tier(roll, ac):
|
||
"""缺口 = AC − 掷点读数 → 档位（1-3/4-7/8+ → 75/50/25%，达标=全额）。"""
|
||
gap = ac - roll
|
||
if gap <= 0:
|
||
return "全额"
|
||
if gap <= 3:
|
||
return "×0.75"
|
||
if gap <= 7:
|
||
return "×0.5"
|
||
return "×0.25"
|
||
|
||
|
||
def aggregate_dice(res_files):
|
||
"""掷点面（AC 批落地后激活）。字段名以 FIELD_CANDIDATES 命中为准。"""
|
||
rolls = [] # (direction, roll, tier, crit)
|
||
for res in res_files:
|
||
for run in res.runs:
|
||
for row in run.roll_rows:
|
||
roll = _first(row, ROLL_FIELD_CANDIDATES)
|
||
if not isinstance(roll, int):
|
||
continue
|
||
direction = "主角→敌" if row.get("source") == "hero_attack" else "敌→主角"
|
||
ac = _first(row, AC_FIELD_CANDIDATES)
|
||
ac = int(ac) if isinstance(ac, int) else 10 # 基线 AC10
|
||
tier_v = _first(row, TIER_FIELD_CANDIDATES)
|
||
tier = tier_v if tier_v in TIER_MULTIPLIER else derive_tier(roll, ac)
|
||
crit_v = _first(row, CRIT_FIELD_CANDIDATES)
|
||
crit = bool(crit_v) if crit_v is not None else (roll == 20)
|
||
rolls.append((direction, roll, tier, crit))
|
||
if not rolls:
|
||
return None
|
||
hist = Counter(r for _, r, _, _ in rolls)
|
||
tiers = Counter(t for _, _, t, _ in rolls)
|
||
crits = sum(1 for *_, c in rolls if c)
|
||
ev_red = 1 - sum(TIER_MULTIPLIER[t] for _, _, t, _ in rolls) / len(rolls)
|
||
by_dir = defaultdict(list)
|
||
for d, r, t, c in rolls:
|
||
by_dir[d].append((r, t, c))
|
||
return {"hist": hist, "tiers": tiers, "n": len(rolls),
|
||
"crit_rate": crits / len(rolls), "ev_reduction": ev_red, "by_dir": by_dir}
|
||
|
||
|
||
def render_dice(dice, out):
|
||
print("== 掷点面（AC 批验收统计）==", file=out)
|
||
if dice is None:
|
||
print(" 不适用（AC 批前日志）：本日志无掷点字段。", file=out)
|
||
print(" 其余指标照常输出；落批后重跑本工具，先看【数据质量】的未知字段观测面。", file=out)
|
||
return
|
||
print(f" 样本 {dice['n']} 次掷点", file=out)
|
||
print(" 掷点直方图（d20）:", file=out)
|
||
for v in range(1, 21):
|
||
c = dice["hist"].get(v, 0)
|
||
if c:
|
||
print(f" {v:2d}: {'#' * min(c, 60)}{c if c > 60 else ''}", file=out)
|
||
n = dice["n"]
|
||
print(" 档位占比:", file=out)
|
||
for t in ("全额", "×0.75", "×0.5", "×0.25"):
|
||
c = dice["tiers"].get(t, 0)
|
||
print(f" {t:>6}: {c:6d} ({c / n * 100:5.1f}%)", file=out)
|
||
cr = dice["crit_rate"]
|
||
ok = abs(cr - THEORY_CRIT_RATE) <= CRIT_TOLERANCE
|
||
print(f" nat20 暴击率: {cr * 100:.2f}%（理论 {THEORY_CRIT_RATE * 100:.0f}% ±{CRIT_TOLERANCE * 100:.0f}pp）"
|
||
f" → {'健康' if ok else '超界，需排查'}", file=out)
|
||
print(f" 观察 EV 减免: {dice['ev_reduction'] * 100:.2f}%"
|
||
f"（理论 AC10 = {THEORY_EV_REDUCTION_AC10 * 100:.2f}%）", file=out)
|
||
print(" 按攻击方向:", file=out)
|
||
for d, items in sorted(dice["by_dir"].items()):
|
||
rr = [r for r, _, _ in items]
|
||
cc = sum(1 for *_, c in items if c)
|
||
red = 1 - sum(TIER_MULTIPLIER[t] for _, t, _ in items) / len(items)
|
||
print(f" {d}: n={len(items)} 均值掷点={sum(rr) / len(rr):.2f} "
|
||
f"暴击率={cc / len(items) * 100:.2f}% EV减免={red * 100:.2f}%", file=out)
|
||
|
||
|
||
def render(res_files, args, out):
|
||
total_lines = sum(r.lines for r in res_files)
|
||
total_secs = sum(r.parse_seconds for r in res_files)
|
||
all_runs = [run for res in res_files for run in res.runs]
|
||
|
||
print("== 输入概况 ==", file=out)
|
||
print(f" 文件 {len(res_files)} 个，共 {total_lines} 行，解析 {total_secs:.2f}s"
|
||
f"（大文件单文件耗时见下）", file=out)
|
||
for res in res_files:
|
||
dice_tag = "掷点面：激活" if res.dice_active else "掷点面：本日志无掷点字段（AC 批前日志，已降级）"
|
||
print(f" {os.path.basename(res.path)}: {res.lines} 行 / {res.parse_seconds:.2f}s / "
|
||
f"{len(res.runs)} 局 / {dice_tag}", file=out)
|
||
print(file=out)
|
||
|
||
render_dice(aggregate_dice(res_files), out)
|
||
print(file=out)
|
||
|
||
print("== 关卡级统计 ==", file=out)
|
||
print(" 口径：局=run 行切分（无 run 行的旧日志整文件一局）；通关=result=victory；", file=out)
|
||
print(" 进BossHP=Boss首结算快照HP+首轮承伤−首轮治疗（精确反推）；平均轮数=每局settlement数。", file=out)
|
||
print(" 弃图率=每局受理刷新次数（07⑦ 基线读数，与 MC 刷新/局同口径）；", file=out)
|
||
print(" 每轮刷新率=刷新次数÷结算轮数（组内加权）。阈值数字归调参批，本表只出基线。", file=out)
|
||
hdr = (f" {'关卡':>6} {'档位':>8} {'局数':>5} {'通关率':>7} {'平均轮数':>8} {'进BossHP均':>10} "
|
||
f"{'通关剩余HP均':>12} {'整局承伤均':>10} {'弃图率':>7} {'每轮刷新率':>9}")
|
||
print(hdr, file=out)
|
||
for row in aggregate_levels(all_runs):
|
||
print(f" {str(row['level']):>6} {row['policy']:>8} {row['runs']:>5} "
|
||
f"{fmt(row['clear_rate'] * 100 if row['clear_rate'] is not None else None):>6}% "
|
||
f"{fmt(row['avg_rounds']):>8} {fmt(row['avg_boss_hp']):>10} "
|
||
f"{fmt(row['avg_end_hp']):>12} {fmt(row['avg_dmg']):>10} "
|
||
f"{fmt(row['refresh_per_run'], 2):>7} {fmt(row['refresh_per_round'], 3):>9}", file=out)
|
||
print(file=out)
|
||
|
||
print("== 战斗级统计（按 关卡/节点类型/深度 分组；对照承伤带 普通≤20%、Boss≤50% max_hp）==", file=out)
|
||
print(f" {'关卡':>6} {'类型':>6} {'深度':>4} {'场数':>5} {'场均轮数':>8} {'场均承伤':>8} {'承伤/max_hp':>11} {'承伤带':>7} {'判定':>4}", file=out)
|
||
for row in aggregate_battles(all_runs):
|
||
print(f" {str(row['level']):>6} {row['type']:>6} {str(row['depth']):>4} {row['count']:>5} "
|
||
f"{fmt(row['avg_rounds']):>8} {fmt(row['avg_dmg']):>8} {fmt(row['avg_ratio'] * 100):>10}% "
|
||
f"{'≤' + format(row['band'] * 100, '.0f') + '%':>7} {'✓' if row['pass'] else '✗':>4}", file=out)
|
||
print(file=out)
|
||
|
||
rr = aggregate_rounds(all_runs)
|
||
print("== 轮级统计 ==", file=out)
|
||
print(f" 总轮数 {rr['rounds']}｜敌战轮（有过敌人的战斗全部轮数）{rr['enemy_battle_rounds']}｜有攻击落地轮 {rr['attack_landed']}", file=out)
|
||
zn_pct = rr['zero_narrow'] / rr['attack_landed'] * 100 if rr['attack_landed'] else 0.0
|
||
ebr = rr['enemy_battle_rounds']
|
||
zw_pct = rr['zero_loss_wide'] / ebr * 100 if ebr else 0.0
|
||
print(f" 零承伤轮（窄口径，AC 验收主指标）: {rr['zero_narrow']} / 有攻击落地轮 {rr['attack_landed']}"
|
||
f"（{zn_pct:.1f}%）——AC 批落地含'伤害下限 1'后该值应趋零", file=out)
|
||
print(f" 零掉血轮（宽口径，参考）: {rr['zero_loss_wide']} / 敌战轮 {ebr}"
|
||
f"（{zw_pct:.1f}%）", file=out)
|
||
print(file=out)
|
||
|
||
# 增益战斗期复原
|
||
buff_totals = Counter()
|
||
buff_settlements = 0
|
||
for run in all_runs:
|
||
seen_in_run = False
|
||
for rec in run.round_ledger:
|
||
for k, v in rec["buffs"].items():
|
||
buff_totals[k] += v
|
||
seen_in_run = True
|
||
if seen_in_run:
|
||
buff_settlements += sum(1 for rec in run.round_ledger if rec["buffs"])
|
||
snap_total = sum(r.snapshot_buffs_total for r in all_runs)
|
||
snap_nonempty = sum(r.snapshot_buffs_nonempty for r in all_runs)
|
||
print("== 增益战斗期复原（来自 step 行，按结算轮窗口聚合）==", file=out)
|
||
print(f" 各增益出现总层数: {dict(buff_totals) if buff_totals else '（无 effect 行）'}", file=out)
|
||
print(f" 有增益激活的结算数: {buff_settlements}（每结算层数明细见 --rounds 流水）", file=out)
|
||
print(f" 快照 buffs 非空: {snap_nonempty}/{snap_total}"
|
||
f"（已知盲区：快照在衰减段后发出且状态寿命=1 结算 → 恒空；此行仅作盲区持续暴露）", file=out)
|
||
print(file=out)
|
||
|
||
if args.per_run:
|
||
print("== 逐局明细（--per-run）==", file=out)
|
||
print(f" {'文件':<28} {'种子':>6} {'档位':>8} {'结果':>8} {'轮数':>4} {'承伤':>5} {'零承伤轮':>8} {'进BossHP':>8} {'终局HP':>6} {'刷新':>4}", file=out)
|
||
for run in all_runs:
|
||
print(f" {run.src_file:<28} {str(run.seed):>6} {run.policy or '—':>8} "
|
||
f"{run.result or '—':>8} {run.rounds_total:>4} {run.damage_total:>5} "
|
||
f"{run.zero_narrow:>8} {fmt(run.boss_enter_hp(), 0):>8} {str(run.end_hp):>6} {run.refresh_count:>4}", file=out)
|
||
print(file=out)
|
||
|
||
if args.rounds:
|
||
print("== 逐轮流水（--rounds）==", file=out)
|
||
print(" 列: 种子 节点 深度 轮 | 攻击落地 蓄力 承伤 治疗 结算HP | 零承伤 零掉血 | 增益复原 | 出路", file=out)
|
||
for run in all_runs:
|
||
for rec in run.round_ledger:
|
||
buffs = ",".join(f"{k}×{v}" for k, v in sorted(rec["buffs"].items())) or "—"
|
||
print(f" {run.seed} {rec['type']} {rec['depth']} r{rec['round']} | "
|
||
f"{rec['attacks']} {rec['charges']} {rec['dmg']} {rec['heal']} {rec['hp']:>3} | "
|
||
f"{'Y' if rec['zero_narrow'] else '-'} {'Y' if rec['zero_wide'] else '-'} | "
|
||
f"{buffs} | {rec['path']}", file=out)
|
||
print(file=out)
|
||
|
||
print("== 数据质量 ==", file=out)
|
||
bad = sum(r.bad_lines for r in res_files)
|
||
late = sum(run.late_battle_ends for res in res_files for run in res.runs)
|
||
print(f" 解析失败行: {bad}｜未配对 battle_end（延迟发出越局）: {late}", file=out)
|
||
if res_files and any(res.unknown_step_keys for res in res_files):
|
||
print(" step 行未知字段观测面（AC 批落批后的适配入口——新掷点字段会先出现在这里）:", file=out)
|
||
for res in res_files:
|
||
for k, c in res.unknown_step_keys.most_common():
|
||
print(f" {os.path.basename(res.path)}: {k} ×{c} 样例 {res.unknown_step_samples[k]}", file=out)
|
||
else:
|
||
print(" step 行未知字段: 无（当前日志 = 2026-09-16 schema，无掷点/分量字段）", file=out)
|
||
print(" AC 批落批后：新字段先出现在上面观测面 → 在 FIELD_CANDIDATES 登记真实字段名 → 掷点面自动激活。", file=out)
|
||
|
||
|
||
# ---------------------------------------------------------------- 入口
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="试玩 JSONL 日志三级统计分析器（关卡/战斗/轮次 + AC 掷点预留面）")
|
||
ap.add_argument("input", help="jsonl 文件 | 目录 | 通配（目录按 playtest_*.jsonl 收取）")
|
||
ap.add_argument("--per-run", action="store_true", help="附加逐局明细")
|
||
ap.add_argument("--rounds", action="store_true", help="附加逐轮流水")
|
||
args = ap.parse_args(argv)
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
|
||
files = resolve_inputs(args.input)
|
||
files = [f for f in files if os.path.isfile(f)]
|
||
if not files:
|
||
print(f"未找到输入：{args.input}", file=sys.stderr)
|
||
return 2
|
||
res_files = [parse_file(f) for f in files]
|
||
render(res_files, args, sys.stdout)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|