7.3 节奏冲突的检测

#!/usr/bin/env python3
# file: sim/nm_07_3_rhythm_conflict_detection.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/nm_07_3_rhythm_conflict_detection.py
"""卷三 7.3 节奏冲突的检测（S 层）：密度与张力曲线的错配，检出并计数。

7.1 量了节奏的快慢，7.2 把密度维持在可读区间；这一节查**两者与张力的错配**。
可判定口径（写死在代码里）：

  张力 = 命中节拍的 beats[].tension（1~5 整数）；
  局部密度 = 该节拍**首次进入的场景**所在节点起、W=5 节点的滑窗密度（千分），
             事件口径同 7.2（新场景 + 新 flag + 新物品 + 有意义选择）。
  规则 A（绝对错配）：张力峰值 tension>=4 而密度 < DENS_LO（峰处极疏）；
                      张力谷   tension<=2 而密度 > DENS_HI（谷处极密）。
  规则 B（方向错配，相对本路径基线）：峰值处密度 < 本路均值；谷处密度 > 本路均值。

用 Lehmer MINSTD（rng = rng * 48271 % 2147483647）采样 500 条路径。
冲突计数是 **proxy**：它只说明「结构上张力的峰谷与信息疏密不同向」，不评价叙事好坏，
也不预测玩家是否感到「该紧不紧」。产出 out/nm_07_3_conflict.csv + .json。
"""
from __future__ import annotations

import csv
import itertools
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "../out")
LO, HI = 0, 100
SEED = 20261016
N_PATHS = 500
W = 5
DENS_LO = 250
DENS_HI = 750
PEAK = 4         # 张力峰值阈值
TROUGH = 2       # 张力谷阈值


def cond_ok(c, s):
    if c is None:
        return True
    k = c[0]
    if k == "flag":
        return s["flags"].get(c[1], 0) == c[2]
    if k == "item":
        return s["items"].get(c[1], 0) == c[2]
    if k == "aff":
        v, op, n = s["aff"].get(c[1], 0), c[2], c[3]
        return v >= n if op == ">=" else v < n if op == "<" else v == n if op == "==" else v != n
    if k == "and":
        return cond_ok(c[1], s) and cond_ok(c[2], s)
    if k == "or":
        return cond_ok(c[1], s) or cond_ok(c[2], s)
    return not cond_ok(c[1], s)


def apply_eff(eff, s):
    aff, fl, it = dict(s["aff"]), dict(s["flags"]), dict(s["items"])
    for k, v in (eff.get("set") or {}).items():
        fl[k] = v
    for k, v in (eff.get("add") or {}).items():
        aff[k] = LO if aff.get(k, 0) + v < LO else HI if aff.get(k, 0) + v > HI else aff.get(k, 0) + v
    for k, v in (eff.get("grant") or {}).items():
        it[k] = v
    for k in (eff.get("revoke") or {}):
        it[k] = 0
    return aff, fl, it


def advance_key(c, s):
    a, f, i = apply_eff(c.get("effects") or {}, s)
    return (c["goto"], tuple(sorted(a.items())), tuple(sorted(f.items())), tuple(sorted(i.items())))


def main() -> int:
    with open(os.path.join(HERE, "../config/story_spec.json"), encoding="utf-8") as fh:
        spec = json.load(fh)
    nodes = spec["dialogue"]
    by = {n["id"]: n for n in nodes}
    start = spec["meta"]["start"]
    vars_ = sorted(spec["state"]["vars"])
    flags = sorted(spec["state"]["flags"])
    items = sorted(spec["state"]["items"])
    scene_of_node = {n["id"]: n["scene"] for n in nodes}
    beat_by_scene = {}
    for b in spec["beats"]:
        beat_by_scene.setdefault(b["scene"], []).append(b)

    def init_state():
        aff = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
               for v in vars_}
        return {"node": start, "aff": aff,
                "flags": {f: 0 for f in flags}, "items": {i: 0 for i in items}}

    rng = SEED
    rows = []
    peak_dens, trough_dens, base_dens = [], [], []
    for p in range(N_PATHS):
        s = init_state()
        seen_scenes = {scene_of_node[start]}
        path = [start]
        events = []
        while True:
            cur = by[s["node"]]
            if cur.get("ending"):
                break
            opts = [c for c in (cur.get("choices") or []) if cond_ok(c.get("when"), s)]
            if not opts:
                break
            rng = (rng * 48271) % 2147483647
            c = opts[rng % len(opts)]
            k = advance_key(c, s)
            fake = any(o is not c and advance_key(o, s) == k for o in opts)
            aff, fl, it = apply_eff(c.get("effects") or {}, s)
            sc = scene_of_node[c["goto"]]
            ev = 0
            if sc not in seen_scenes:
                seen_scenes.add(sc)
                ev += 1
            if any(fl[f] == 1 and s["flags"].get(f, 0) == 0 for f in flags):
                ev += 1
            if any(it[i] == 1 and s["items"].get(i, 0) == 0 for i in items):
                ev += 1
            if not fake:
                ev += 1
            events.append(ev)
            s = {"node": c["goto"], "aff": aff, "flags": fl, "items": it}
            path.append(s["node"])

        # 每个节点位置上的滑窗密度（起点窗口从 0 开始）
        wins = [sum(events[i:i + W]) * 1000 // (4 * W) for i in range(max(1, len(events) - W + 1))]
        base = sum(wins) // len(wins) if wins else 0
        base_dens.append(base)

        # 节拍命中：按首次进入场景的顺序，取该节拍的张力与该处的窗口密度
        first_pos, seen = {}, set()
        for i, nid in enumerate(path):
            sc = scene_of_node[nid]
            if sc not in seen:
                seen.add(sc)
                first_pos[sc] = i
        # 第 i 个节点对应的事件下标是 i-1（起点无事件）；窗口密度取该处
        conf_a, conf_b, pairs = 0, 0, []
        for sc, i in first_pos.items():
            if sc not in beat_by_scene:
                continue
            t = beat_by_scene[sc][0]["tension"]
            wi = max(0, i - 1)
            d = wins[wi] if wi < len(wins) else wins[-1]
            pairs.append((t, d))
            if t >= PEAK:
                peak_dens.append(d)
                if d < DENS_LO:
                    conf_a += 1
                if d < base:
                    conf_b += 1
            elif t <= TROUGH:
                trough_dens.append(d)
                if d > DENS_HI:
                    conf_a += 1
                if d > base:
                    conf_b += 1
        rows.append({
            "path": p, "nodes": len(path), "beats": len(pairs),
            "base_density": base, "conflict_a": conf_a, "conflict_b": conf_b,
            "peak_beats": sum(1 for t, _ in pairs if t >= PEAK),
            "trough_beats": sum(1 for t, _ in pairs if t <= TROUGH),
        })

    n = len(rows)
    tot_beats = sum(r["beats"] for r in rows)
    tot_a = sum(r["conflict_a"] for r in rows)
    tot_b = sum(r["conflict_b"] for r in rows)
    tot_peaks = sum(r["peak_beats"] for r in rows)
    tot_troughs = sum(r["trough_beats"] for r in rows)

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "nm_07_3_conflict.csv"), "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=list(rows[0]))
        w.writeheader()
        w.writerows(rows)

    summary = {
        "paths": n, "seed": SEED, "window": W,
        "readable_band_permille": [DENS_LO, DENS_HI],
        "peak_threshold": PEAK, "trough_threshold": TROUGH,
        "beat_hits_total": tot_beats,
        "peak_beats": tot_peaks, "trough_beats": tot_troughs,
        "conflict_a_total": tot_a,
        "conflict_b_total": tot_b,
        "conflict_a_per_mille": tot_a * 1000 // tot_beats,
        "conflict_b_per_mille": tot_b * 1000 // tot_beats,
        "paths_with_conflict_a": sum(1 for r in rows if r["conflict_a"]),
        "paths_with_conflict_b": sum(1 for r in rows if r["conflict_b"]),
        "max_conflict_a_per_path": max(r["conflict_a"] for r in rows),
        "max_conflict_b_per_path": max(r["conflict_b"] for r in rows),
        "peak_beat_share_permille": tot_peaks * 1000 // tot_beats,
        "trough_beat_share_permille": tot_troughs * 1000 // tot_beats,
        "peak_density_mean": sum(peak_dens) // len(peak_dens) if peak_dens else 0,
        "trough_density_mean": sum(trough_dens) // len(trough_dens) if trough_dens else 0,
        "base_density_mean": sum(base_dens) // len(base_dens) if base_dens else 0,
    }
    with open(os.path.join(OUT, "nm_07_3_conflict.json"), "w", encoding="utf-8") as fh:
        json.dump(summary, fh, ensure_ascii=False, indent=1)
        fh.write("\n")

    print(f"采样 {n} 条路径（seed={SEED}），节拍命中 {tot_beats} 次")
    print(f"张力阈值：峰值 >={PEAK}  谷值 <={TROUGH}；密度可读区间 [{DENS_LO}, {DENS_HI}] 千分")
    print(f"峰值节拍 {tot_peaks} 次 = {summary['peak_beat_share_permille']} 千分"
          f"  谷值节拍 {tot_troughs} 次 = {summary['trough_beat_share_permille']} 千分")
    print(f"规则 A 绝对错配（峰处极疏 / 谷处极密）：{tot_a} 次 = {summary['conflict_a_per_mille']} 千分")
    print(f"规则 B 方向错配（峰处低于本路均值 / 谷处高于本路均值）：{tot_b} 次 = "
          f"{summary['conflict_b_per_mille']} 千分")
    print(f"含 A 冲突的路径 {summary['paths_with_conflict_a']} / {n}"
          f"  含 B 冲突的路径 {summary['paths_with_conflict_b']} / {n}")
    print(f"单条路径最多 A 冲突 {summary['max_conflict_a_per_path']}"
          f"  最多 B 冲突 {summary['max_conflict_b_per_path']}")
    print(f"平均密度：峰值节拍处 {summary['peak_density_mean']} 千分"
          f"  谷值节拍处 {summary['trough_density_mean']} 千分"
          f"  全路径基线 {summary['base_density_mean']} 千分")
    print("wrote out/nm_07_3_conflict.csv / .json")

    if tot_beats == 0:
        print("FAIL 采样未命中任何节拍")
        return 1
    print("OK nm_07_3_rhythm_conflict_detection：密度-张力错配已检出并计数（软告警）")
    return 0


if __name__ == "__main__":
    sys.exit(main())

采样 500 条路径（seed=20261016），节拍命中 4310 次
张力阈值：峰值 >=4  谷值 <=2；密度可读区间 [250, 750] 千分
峰值节拍 1928 次 = 447 千分  谷值节拍 1500 次 = 348 千分
规则 A 绝对错配（峰处极疏 / 谷处极密）：3 次 = 0 千分
规则 B 方向错配（峰处低于本路均值 / 谷处高于本路均值）：2810 次 = 651 千分
含 A 冲突的路径 3 / 500  含 B 冲突的路径 500 / 500
单条路径最多 A 冲突 1  最多 B 冲突 8
平均密度：峰值节拍处 373 千分  谷值节拍处 502 千分  全路径基线 406 千分
wrote out/nm_07_3_conflict.csv / .json
OK nm_07_3_rhythm_conflict_detection：密度-张力错配已检出并计数（软告警）