gplan-2/tools/skill_rhythm_audit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""技能池出招节奏审计（消费 playtest_*.jsonl 的 enemy_action 行）。

对应 06 待办「单拍行为环的体验验证（改型：技能池节奏体验验证）」的机器半边。
判读带（2026-09-19 用户拍板）：
  · 过涝 = 敌群连续出特殊技、无普攻基底——读数 = 单敌群同场战斗内最长连续特殊技连击
    （连击 ≥3 记过涝信号，连击阈值 3 为代拍板默认：保底 M=4 与冷却 2-4 语义下 2 连属正常波动）
  · 过旱 = 按战斗整体：一场战斗里所有敌人都没用过特殊技记过旱样本；任一敌人用过即正常
    （战斗内敌人从未行动过的「首轮全灭/跳战」场单列，不与"有行动无特殊技"混计）
机器读数：
  · 含闸速率 = 特殊技出招占全部出招的份额（对表值权重 = 静态份额比对）
  · 旱涝间隔 = 相邻两次特殊技之间的普攻连投数（cast 间隔）与回合间隔（round 间隔）
  · 保底 M=4 语义：连续 4 拍普攻 → 下拍必出特殊技 → cast 间隔样本 ≤4 合法

用法：
  python tools/skill_rhythm_audit.py <playtest_*.jsonl> [...]
"""
import collections
import glob
import json
import re
import sys

SUFFIX_RE = re.compile(r"（[^）]*）|＋[^＋]*$")


def actor_of(display_name: str) -> str:
    # "蛛化卓尔精灵 毒爆（被压制）＋附加2（必中）" → 敌群前缀取第一个空格前段（无空格 = 状态行，丢弃）
    if " " not in display_name:
        return ""
    return display_name.split(" ")[0]


def pct(samples, q):
    if not samples:
        return "-"
    s = sorted(samples)
    idx = min(len(s) - 1, max(0, round(q * (len(s) - 1))))
    return s[idx]


def main(paths):
    battles = []
    cur = None
    for pattern in paths:
        for path in sorted(glob.glob(pattern)):
            with open(path, encoding="utf-8") as f:
                for line in f:
                    try:
                        d = json.loads(line)
                    except json.JSONDecodeError:
                        continue
                    kind = d.get("kind")
                    if kind == "node":
                        cur = {"actors": collections.defaultdict(list)}
                        battles.append(cur)
                    elif (kind == "step" and d.get("source") == "enemy_action"
                          and d.get("skill_id") and cur is not None):
                        actor = actor_of(d.get("display_name", ""))
                        if actor:
                            cur["actors"][actor].append(
                                (d.get("round_index", 0), d.get("skill_id", "attack")))

    agg = collections.defaultdict(lambda: {
        "battles": 0, "casts": 0, "specials": 0,
        "cast_gaps": [], "round_gaps": [], "streaks": [],
    })
    # 过旱/过涝判读带（用户拍板 2026-09-19）——按战斗整体计
    drought_battles = 0          # 过旱：全战斗零特殊技（有敌行动）
    no_action_battles = 0        # 首轮全灭/跳战：敌从未行动（单列不混计）
    normal_battles = 0
    flood_signals = 0            # 过涝信号：单敌群连续特殊技 ≥3
    flood_max = 0
    for b in battles:
        actors = b["actors"]
        if not actors:
            no_action_battles += 1
            continue
        has_special = False
        for actor, casts in actors.items():
            a = agg[actor]
            a["battles"] += 1
            a["casts"] += len(casts)
            special_idx = [i for i, (_, sk) in enumerate(casts) if sk != "attack"]
            a["specials"] += len(special_idx)
            if special_idx:
                has_special = True
                # 过涝：最长连续特殊技连击
                run, best = 0, 0
                for _, sk in casts:
                    if sk != "attack":
                        run += 1
                        best = max(best, run)
                    else:
                        run = 0
                a["streaks"].append(best)
                flood_max = max(flood_max, best)
                if best >= 3:
                    flood_signals += 1
                for prev, nxt in zip(special_idx, special_idx[1:]):
                    a["cast_gaps"].append(nxt - prev - 1)
                    a["round_gaps"].append(casts[nxt][0] - casts[prev][0])
        if has_special:
            normal_battles += 1
        else:
            drought_battles += 1

    print(f"{'敌群':<10}{'场数':>5}{'出招':>6}{'特殊':>6}{'份额':>7}"
          f"{'cast隔p50':>10}{'p90':>5}{'max':>5}"
          f"{'round隔p50':>11}{'p90':>5}{'max':>5}{'连击max':>8}")
    for actor, a in sorted(agg.items()):
        share = a["specials"] / a["casts"] if a["casts"] else 0
        print(f"{actor:<10}{a['battles']:>5}{a['casts']:>6}{a['specials']:>6}{share:>7.1%}"
              f"{str(pct(a['cast_gaps'],0.5)):>10}{str(pct(a['cast_gaps'],0.9)):>5}"
              f"{str(max(a['cast_gaps']) if a['cast_gaps'] else '-'):>5}"
              f"{str(pct(a['round_gaps'],0.5)):>11}{str(pct(a['round_gaps'],0.9)):>5}"
              f"{str(max(a['round_gaps']) if a['round_gaps'] else '-'):>5}"
              f"{str(max(a['streaks']) if a['streaks'] else 0):>8}")
    total_battles = len(battles)
    print(f"\n战斗总数 {total_battles}｜跳战（敌未行动）{no_action_battles}"
          f"｜过旱（有行动·全战斗零特殊技）{drought_battles}"
          f"｜正常 {normal_battles}")
    print(f"过涝信号（单敌群连续特殊技≥3）{flood_signals} 处｜全场最长连击 {flood_max}")
    # 保底越界样本：cast 间隔 > M（M=4 → 间隔 ≥5 普攻才越界；间隔 4 = 连续 4 拍普攻后保底必出，合法）
    total_over = sum(1 for a in agg.values() for g in a["cast_gaps"] if g >= 5)
    total_gaps = sum(len(a["cast_gaps"]) for a in agg.values())
    print(f"保底越界（相邻特殊技间隔≥5 普攻）：{total_over}/{total_gaps}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)
    main(sys.argv[1:])