11.3 对话树的遍历与校验

#!/usr/bin/env python3
# file: sim/na_11_3_dialogue_traversal.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/na_11_3_dialogue_traversal.py
"""卷一 11.3 对话树的遍历与校验（S 层）：把 48 个对话节点当成一张有向图，
用 DFS / BFS 三种遍历各查一类结构缺陷。

三类缺陷（本节的口径）：
  1) 环：沿 goto 边走一圈能回到自己 —— DFS 三色标记检出；
  2) 不可达节点：从起点出发 BFS 走不到的节点 —— 从未进入任何路径；
  3) 死路：既无选项、又不是结局的节点 —— 走到这里故事停住但没给结局。

同时给出 BFS 最短距离（起点0）与另行按拓扑序计算的 DAG 最长边数。
本案例实测三类缺陷均为 0，与 na_00_spec_check.py 的 BFS 穷举口径互证。

产出 out/na_11_3_dialogue_traversal.csv（逐节点层号表）。
"""
from __future__ import annotations

import csv
import json
import os
import sys
from collections import deque

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
CSV_PATH = os.path.join(OUT, "na_11_3_dialogue_traversal.csv")


def load_spec():
    with open(os.path.join(ROOT, "config/story_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)


def build_graph(dlg):
    """返回 (邻接表, 节点表, 边集)。邻接表按 id 排序，保证遍历确定。"""
    adj = {n["id"]: sorted({c["goto"] for c in (n.get("choices") or [])}) for n in dlg}
    node = {n["id"]: n for n in dlg}
    edges = {(n["id"], c["goto"]) for n in dlg for c in (n.get("choices") or [])}
    return adj, node, edges


def find_cycle(adj):
    """DFS 三色标记找环，返回一条环上的节点序列（无环返回 []）。"""
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {k: WHITE for k in adj}
    parent: dict[str, str] = {}

    for root in sorted(adj):
        if color[root] != WHITE:
            continue
        stack = [(root, iter(adj[root]))]
        color[root] = GRAY
        while stack:
            u, it = stack[-1]
            advanced = False
            for v in it:
                if color[v] == GRAY:
                    path, x = [v], u
                    while x != v:
                        path.append(x)
                        x = parent[x]
                    return list(reversed(path + [v]))[:-1]
                if color[v] == WHITE:
                    color[v] = GRAY
                    parent[v] = u
                    stack.append((v, iter(adj[v])))
                    advanced = True
                    break
            if not advanced:
                color[u] = BLACK
                stack.pop()
    return []


def bfs_layers(adj, start):
    """从 start 出发的 BFS 层号；不可达节点不入表。"""
    layer = {start: 0}
    q = deque([start])
    while q:
        u = q.popleft()
        for v in adj[u]:
            if v not in layer:
                layer[v] = layer[u] + 1
                q.append(v)
    return layer


def longest_dag_depths(adj, start):
    """可达子图按 Kahn 拓扑序松弛，返回从 start 出发的最长边数。"""
    if start not in adj or any(v not in adj for vs in adj.values() for v in vs):
        raise ValueError("起点或目标节点不存在")
    reachable = set(bfs_layers(adj, start))
    indegree = {u: 0 for u in reachable}
    for u in reachable:
        for v in adj[u]:
            indegree[v] += 1
    queue = deque(sorted(u for u, degree in indegree.items() if degree == 0))
    depth = {start: 0}
    processed = 0
    while queue:
        u = queue.popleft()
        processed += 1
        for v in adj[u]:
            depth[v] = max(depth.get(v, -1), depth[u] + 1)
            indegree[v] -= 1
            if indegree[v] == 0:
                queue.append(v)
    if processed != len(reachable):
        raise ValueError("可达子图有环，不能按 DAG 计算最长路")
    return depth


def main() -> int:
    spec = load_spec()
    dlg = spec["dialogue"]
    start = spec["meta"]["start"]
    adj, node, edges = build_graph(dlg)

    print(f"spec={spec['meta']['title']}  起点 {start}")
    print(f"有向图：节点 {len(adj)}  选项 {sum(len(n.get('choices') or []) for n in dlg)} 条"
          f"，去重后唯一跳转 {len(edges)} 条（多个选项汇到同一节点时会合并）")

    # 1) 环
    cyc = find_cycle(adj)
    print(f"1) DFS 找环：检出环 {len(cyc)} 个 {cyc if cyc else '（无环，图是 DAG）'}")

    # 2) 不可达
    layer = bfs_layers(adj, start)
    unreach = sorted(set(adj) - set(layer))
    print(f"2) BFS 可达性：从起点可达 {len(layer)} / {len(adj)} 个节点；不可达 {len(unreach)} 个 {unreach}")

    # 3) 死路
    dead = sorted(n for n in adj if not adj[n] and not node[n].get("ending"))
    sinks = sorted(n for n in adj if not adj[n])
    print(f"3) 死路检测：出度为 0 的节点 {len(sinks)} 个；其中非结局（真死路）{len(dead)} 个 {dead}")
    print(f"   5 个结局节点：{sinks}")

    # 最短距离分层，不能把这个顺序当作拓扑序
    max_layer = max(layer.values())
    hist: dict[int, int] = {}
    for v in layer.values():
        hist[v] = hist.get(v, 0) + 1
    print(f"层号（BFS 最短跳数）：最深 {max_layer} 层；逐层节点数 "
          + "  ".join(f"L{k}={hist.get(k, 0)}" for k in range(max_layer + 1)))
    back = [(u, v) for u in adj for v in adj[u] if layer[v] > layer[u] + 1]
    print(f"分层合法性：每条边满足 layer[v] <= layer[u]+1 的反例 {len(back)} 条")
    depth = longest_dag_depths(adj, start)
    print(f"拓扑深度（最长路径，图无环故良定义）：最深 {max(depth.values())} 层")
    reach_end = sorted(n for n in layer if node[n].get("ending"))
    print(f"可达的结局节点 {len(reach_end)} / {len(sinks)}：{reach_end}")

    rows = [{"id": n["id"], "layer": layer.get(n["id"], -1), "out_degree": len(adj[n["id"]]),
             "kind": "ending" if node[n["id"]].get("ending") else "choice"}
            for n in sorted(dlg, key=lambda x: (layer.get(x["id"], 99), x["id"]))]

    # 与 na_00 的口径互证：三者必须全为 0
    ok = (len(cyc) == 0) and (len(unreach) == 0) and (len(dead) == 0)
    print(f"三类缺陷（环 / 不可达 / 死路）计数：{len(cyc)} / {len(unreach)} / {len(dead)}"
          f"；与 na_00_spec_check.py 的 BFS 口径一致：{ok}")
    if not ok:
        print("FAIL 对话图存在结构缺陷")
        return 1

    os.makedirs(OUT, exist_ok=True)
    with open(CSV_PATH, "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=["id", "layer", "out_degree", "kind"])
        w.writeheader()
        w.writerows(rows)
    print(f"wrote out/na_11_3_dialogue_traversal.csv rows={len(rows)}")
    print("OK na_11_3_dialogue_traversal：无环、全可达、无死路，DAG 层号已输出")
    return 0


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

spec=雾港夜航  起点 d_pier_1
有向图：节点 48  选项 115 条，去重后唯一跳转 56 条（多个选项汇到同一节点时会合并）
1) DFS 找环：检出环 0 个 （无环，图是 DAG）
2) BFS 可达性：从起点可达 48 / 48 个节点；不可达 0 个 []
3) 死路检测：出度为 0 的节点 5 个；其中非结局（真死路）0 个 []
   5 个结局节点：['e_dark', 'e_lit', 'e_loss', 'e_silence', 'e_truth']
层号（BFS 最短跳数）：最深 22 层；逐层节点数 L0=1  L1=1  L2=2  L3=2  L4=2  L5=1  L6=2  L7=2  L8=2  L9=1  L10=2  L11=2  L12=2  L13=3  L14=3  L15=3  L16=1  L17=2  L18=2  L19=2  L20=2  L21=2  L22=6
分层合法性：每条边满足 layer[v] <= layer[u]+1 的反例 0 条
拓扑深度（最长路径，图无环故良定义）：最深 31 层
可达的结局节点 5 / 5：['e_dark', 'e_lit', 'e_loss', 'e_silence', 'e_truth']
三类缺陷（环 / 不可达 / 死路）计数：0 / 0 / 0；与 na_00_spec_check.py 的 BFS 口径一致：True
wrote out/na_11_3_dialogue_traversal.csv rows=48
OK na_11_3_dialogue_traversal：无环、全可达、无死路，DAG 层号已输出

# file: sim/na_11_3_graph_contract.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/na_11_3_graph_contract.py
import unittest
from functools import lru_cache
from na_11_3_dialogue_traversal import bfs_layers, build_graph, load_spec, longest_dag_depths


class GraphContract(unittest.TestCase):
    def test_shortcut_is_not_topological_order(self):
        graph = {"s": ["a", "b"], "a": ["end"], "b": ["a"], "end": []}
        self.assertEqual(bfs_layers(graph, "s")["end"], 2)
        self.assertEqual(longest_dag_depths(graph, "s")["end"], 3)
        # 改存储/邻接顺序不应改变答案。
        flipped = {key: list(reversed(graph[key])) for key in reversed(graph)}
        self.assertEqual(longest_dag_depths(flipped, "s")["end"], 3)

    def test_actual_spec_against_recursive_reference(self):
        spec = load_spec()
        graph, _, _ = build_graph(spec["dialogue"])
        @lru_cache(None)
        def suffix(node):
            return max((1 + suffix(v) for v in graph[node]), default=0)
        self.assertEqual(suffix(spec["meta"]["start"]), 31)
        self.assertEqual(max(longest_dag_depths(graph, spec["meta"]["start"]).values()), 31)
        self.assertEqual(max(bfs_layers(graph, spec["meta"]["start"]).values()), 22)

    def test_cycle_has_bfs_distance_but_no_dag_longest_path(self):
        graph = {"s": ["a"], "a": ["s"]}
        self.assertEqual(bfs_layers(graph, "s"), {"s": 0, "a": 1})
        with self.assertRaises(ValueError):
            longest_dag_depths(graph, "s")

    def test_bad_reference_rejected(self):
        with self.assertRaises(ValueError):
            longest_dag_depths({"s": ["missing"]}, "s")


if __name__ == "__main__":
    unittest.main()