init: 剧情事件协作 Web 编辑器独立仓(从 SGame/tools/event_authoring 拆出)

This commit is contained in:
bia
2026-06-08 16:50:27 +08:00
commit f5669dc01d
32 changed files with 3497 additions and 0 deletions

14
ir_core/__init__.py Normal file
View File

@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
"""ir_coreStory IR 校验 + 编译 + 词典的共享内核。
编译器 CLIir_compile.pyM5Web 后端共用此包,保证试走/校验/导出口径一致。
"""
from .dictionary import CompileError, Dictionary, load_dictionary
from .compile import compile_ir, extract_texts, expand_out_refs
from .validate import validate, collect_point_refs
__all__ = [
"CompileError", "Dictionary", "load_dictionary",
"compile_ir", "extract_texts", "expand_out_refs",
"validate", "collect_point_refs",
]

269
ir_core/compile.py Normal file
View File

@ -0,0 +1,269 @@
# -*- coding: utf-8 -*-
"""Story IR -> 扁平 List[GameEventData] 编译器。
产出 events.json绕过 StorylineMgr.ReplaceStoryEventData自己做 step/前缀/展开)。
M4 新增out_ref 子序列复用(编译前预展开)、选项 skip押注跳过→直接结算
"""
from .dictionary import CompileError
OUT_REF_SEP = "__" # out_ref 展开前缀分隔符(与节点 id 内的下划线区分)
# ============ out_ref 预展开 ============
def expand_out_refs(ir):
"""把 out_ref 节点摊平成普通节点,返回新的 nodes 列表(不改原 ir
对齐 StorylineMgr.ReplaceStoryEventData 的语义:
- 子序列每个节点 id 加 {refId}__ 前缀;内部跳转同步加前缀;
- 子序列出口next 为空的尾节点)的 next 接到 out_ref 节点的 next
- 引用该 out_ref 节点的跳转改写为子序列入口 {refId}__{entry}
多个 out_ref 引用同一子序列 → 前缀不同 → id 天然不撞(复用本意)。
M4 仅支持一层(子序列内不得再含 out_ref
"""
seqs = {s["id"]: s for s in ir.get("sequences", [])}
src_nodes = ir.get("nodes", [])
out_refs = [n for n in src_nodes if n.get("kind") == "out_ref"]
if not out_refs:
return list(src_nodes)
expanded = [] # 展开出来的子序列节点
ref_entry = {} # refId -> 子序列入口节点 id含前缀
for r in out_refs:
rid, ref = r["id"], r.get("ref")
seq = seqs.get(ref)
if seq is None:
raise CompileError("out_ref 节点 %s 引用了不存在的子序列 %r" % (rid, ref))
sub = seq.get("nodes", [])
if not sub:
raise CompileError("子序列 %r 为空(被 %s 引用)" % (ref, rid))
sub_ids = {n["id"] for n in sub}
prefix = rid + OUT_REF_SEP
ref_entry[rid] = prefix + sub[0]["id"]
def remap(t, is_next):
"""重写子序列内部跳转目标。"""
if t in sub_ids:
return prefix + t
if not t: # 空 next = 子序列出口 -> 接回 caller
return r.get("next", "") if is_next else t
return t # 指向子序列外(共享结局等)保持原样
for n in sub:
if n.get("kind") == "out_ref":
raise CompileError("子序列 %r 含嵌套 out_ref节点 %sM4 暂不支持" % (ref, n["id"]))
c = dict(n)
c["id"] = prefix + n["id"]
if "next" in c:
c["next"] = remap(c.get("next"), True)
elif n.get("kind") not in ("choice", "choice_once", "random", "fight"):
# 链尾叶子(无 next 字段)-> 出口接回 caller.next
c["next"] = r.get("next", "")
if "options" in c:
c["options"] = [_remap_opt(o, remap) for o in c["options"]]
if "branches" in c:
c["branches"] = [dict(b, goto=remap(b.get("goto"), False)) for b in c["branches"]]
if c.get("kind") == "fight":
c["win"] = remap(c.get("win"), False)
c["lose"] = remap(c.get("lose"), False)
expanded.append(c)
# 主图节点:去掉 out_ref 占位,并把指向它们的跳转改写到子序列入口
def deref(t):
return ref_entry.get(t, t)
result = []
for n in src_nodes:
if n.get("kind") == "out_ref":
continue
c = dict(n)
if "next" in c:
c["next"] = deref(c.get("next"))
if "options" in c:
c["options"] = [dict(o, goto=deref(o.get("goto"))) for o in c["options"]]
if "branches" in c:
c["branches"] = [dict(b, goto=deref(b.get("goto"))) for b in c["branches"]]
if c.get("kind") == "fight":
c["win"] = deref(c.get("win"))
c["lose"] = deref(c.get("lose"))
result.append(c)
return result + expanded
def _remap_opt(o, remap):
c = dict(o)
c["goto"] = remap(o.get("goto"), False)
if o.get("skip"):
sk = dict(o["skip"])
sk["node"] = remap(o["skip"].get("node"), False)
c["skip"] = sk
return c
# ============ 编译 ============
def compile_ir(ir, dic):
group = ir["id"]
src_nodes = expand_out_refs(ir)
nodes = {n["id"]: n for n in src_nodes}
endings = {e["id"]: e for e in ir.get("endings", [])}
def entry(tid):
"""目标节点的入口行 id含 group 前缀)。"""
n = nodes.get(tid)
if n and n.get("kind") in ("choice", "choice_once", "random"):
return "%s_%s_o0" % (group, tid)
return "%s_%s" % (group, tid)
def children(tid):
n = nodes.get(tid)
if not n:
return []
res = []
if n.get("next"):
res.append(n["next"])
res += [o["goto"] for o in n.get("options", [])]
res += [b["goto"] for b in n.get("branches", [])]
if n.get("kind") == "fight":
res += [n.get("win"), n.get("lose")]
return [r for r in res if r]
indeg = {i: 0 for i in list(nodes) + list(endings)}
for nid in nodes:
for c in children(nid):
if c in indeg:
indeg[c] += 1
roots = [i for i in nodes if indeg[i] == 0] or list(nodes)[:1]
order, seen = [], set()
def dfs(tid):
if tid in seen or tid not in indeg:
return
seen.add(tid)
order.append(tid)
for c in children(tid):
dfs(c)
for r in roots:
dfs(r)
for i in list(nodes) + list(endings):
if i not in seen:
order.append(i); seen.add(i)
rows, step = [], 0
def newrow(**kw):
nonlocal step
step += 1
row = {"group": group, "step": step, "type": 0}
row.update(kw)
rows.append(row)
return row
for tid in order:
if tid in endings:
e = endings[tid]
rid_str, rac = dic.compile_grants(e.get("grants"))
status = {"success": 1, "fail": 2, "end": 3}.get(e.get("result", "success"))
if status is None:
raise CompileError("结局 %s 的 result 非法: %r(仅 success/fail/end"
% (tid, e.get("result")))
end_id = "%s_%s__end" % (group, tid)
r = newrow(id="%s_%s" % (group, tid), missionText=e.get("summary", ""),
nextStepId=end_id)
if rid_str:
r["resultRewardIds"] = rid_str
if rac:
r["roleActionCode"] = rac
newrow(id=end_id, gameTaskStatus=status)
continue
n = nodes[tid]
kind = n["kind"]
if kind == "narration":
spk = n.get("speaker", "P1")
newrow(id="%s_%s" % (group, tid), points=[spk], content=n.get("text", ""),
nextStepId=entry(n["next"]) if n.get("next") else "")
elif kind == "dialogue":
newrow(id="%s_%s" % (group, tid), points=[n["speaker"]],
content=n.get("text", ""),
nextStepId=entry(n["next"]) if n.get("next") else "")
elif kind in ("choice", "choice_once"):
opt_type = 1 if kind == "choice" else 2
for i, o in enumerate(n.get("options", [])):
rid_str, rac = dic.compile_grants((o.get("reward") or {}).get("grants"))
r = newrow(id="%s_%s_o%d" % (group, tid, i), type=opt_type,
choose=o.get("text", ""), points=["P1"],
nextStepId=entry(o["goto"]))
if o.get("condition"):
r["condition"] = dic.compile_cond(o["condition"])
if rid_str:
r["resultRewardIds"] = rid_str
if rac:
r["roleActionCode"] = rac
# skip押注跳过——选该项时把彩头注入 skip.node 行runtime 自拼 {group}_
if o.get("skip"):
sk = o["skip"]
sk_str, _ = dic.compile_grants((sk.get("reward") or {}).get("grants"))
r["skipNodeId"] = sk["node"]
r["skipReward"] = sk_str
elif kind == "random":
for i, b in enumerate(n.get("branches", [])):
newrow(id="%s_%s_o%d" % (group, tid, i), type=3,
weight=b.get("weight", 1), nextStepId=entry(b["goto"]))
elif kind == "fight":
newrow(id="%s_%s" % (group, tid),
points=n.get("camp1") or ["P1"],
camp2Fighters=n.get("camp2", []),
fightStatus=[str(n["fight_type"]), entry(n["win"]), entry(n["lose"])])
elif kind == "move":
if n.get("mode") == "teleport":
ms = ["1"]
elif n.get("mode") == "remove":
ms = ["2"]
else:
ms = [str(n.get("speed", 6)), n.get("ani", "")]
newrow(id="%s_%s" % (group, tid), points=[n["actor"]],
movePoint=[n["to"]], moveStatus=ms,
nextStepId=entry(n["next"]) if n.get("next") else "")
elif kind == "anim":
anis = [n["ani"]] + ([str(n["angle"])] if n.get("angle") is not None else [])
newrow(id="%s_%s" % (group, tid), points=[n["actor"]], anis=anis,
nextStepId=entry(n["next"]) if n.get("next") else "")
elif kind == "reward":
rid_str, rac = dic.compile_grants(n.get("grants"))
r = newrow(id="%s_%s" % (group, tid),
nextStepId=entry(n["next"]) if n.get("next") else "")
if rid_str:
r["resultRewardIds"] = rid_str
if rac:
r["roleActionCode"] = rac
else:
raise CompileError("未知节点 kind: %r (节点 %s)" % (kind, tid))
ids = [r["id"] for r in rows]
dup = {x for x in ids if ids.count(x) > 1}
if dup:
raise CompileError("行 id 重复: %s" % dup)
return rows
def extract_texts(ir):
"""抽取所有玩家可见文本(去重保序),供本地化翻译。
含 sequences 子序列内文本。"""
texts, seen = [], set()
def add(t):
if t and t not in seen:
seen.add(t); texts.append(t)
def walk_nodes(node_list):
for n in node_list:
add(n.get("text"))
for o in n.get("options", []):
add(o.get("text"))
walk_nodes(ir.get("nodes", []))
for s in ir.get("sequences", []):
walk_nodes(s.get("nodes", []))
for e in ir.get("endings", []):
add(e.get("summary"))
return texts

93
ir_core/dictionary.py Normal file
View File

@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
"""语义词典condition/grant kind -> 真实 ID/形态。外置 ir_dictionary.json
编译器与M5Web 后端共读。新增查表类 kind 只改 JSON零代码改动。
form 决定 grant 编译串形态:
money -> "{id},{value}" (银两;正给负扣)
item -> "{item},{value}" 道具item 由 grant 指定 drop_item_data ID
friend -> "{target},{id},{value}" (友好度,需 target NPC slot
join -> roleActionCode "{code}={target}"(入门,需 target
"""
import json
import os
_DICT_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"ir_dictionary.json")
class CompileError(Exception):
pass
class Dictionary(object):
def __init__(self, data):
self.conditions = data.get("conditions", {})
self.grants = data.get("grants", {})
# ---- 查询 ----
def is_known_cond(self, kind):
return kind in self.conditions
def is_known_grant(self, kind):
return kind in self.grants
def cond_spec(self, kind):
return self.conditions.get(kind)
def grant_spec(self, kind):
return self.grants.get(kind)
def grant_needs_target(self, kind):
spec = self.grants.get(kind) or {}
return bool(spec.get("needs_target"))
# ---- 编译 ----
def compile_cond(self, c):
"""condition -> condition 字段字符串。"""
if not c:
return ""
kind = c.get("kind")
spec = self.conditions.get(kind)
if not spec:
raise CompileError("不支持的 condition kind: %r" % kind)
op = c.get("op")
ops = spec.get("ops", {})
if op not in ops:
raise CompileError("condition %r 不支持比较符 %r" % (kind, op))
return "%s,%s,%s" % (spec["id"], ops[op], c["value"])
def compile_grants(self, grants):
"""grants -> (resultRewardIds 串, roleActionCode 串)。"""
rewards, rac = [], ""
for g in grants or []:
k = g.get("kind")
spec = self.grants.get(k)
if not spec:
raise CompileError("不支持的 grant kind: %r" % k)
form = spec.get("form")
if form == "money":
rewards.append("%s,%s" % (spec["id"], g["value"]))
elif form == "item":
rewards.append("%s,%s" % (g["item"], g["value"]))
elif form == "friend":
rewards.append("%s,%s,%s" % (g["target"], spec["id"], g["value"]))
elif form == "join":
rac = "%s=%s" % (spec["code"], g["target"])
else:
raise CompileError("grant kind %r 的 form %r 未实现" % (k, form))
return ";".join(rewards), rac
_cached = None
def load_dictionary(path=None):
"""加载词典(默认缓存同目录 ir_dictionary.json"""
global _cached
if path is None:
if _cached is None:
with open(_DICT_PATH, encoding="utf-8") as f:
_cached = Dictionary(json.load(f))
return _cached
with open(path, encoding="utf-8") as f:
return Dictionary(json.load(f))

185
ir_core/validate.py Normal file
View File

@ -0,0 +1,185 @@
# -*- coding: utf-8 -*-
"""Story IR 配置期校验。挡掉调研发现的卡流程 bug校验不过不产出任何配置。
返回 (errors, warnings)errors 非空即拒绝编译warnings 仅提示(如点位集缺失)。
M4 新增out_ref 结构校验、选项 skip 校验、点位集坐标引用校验。
"""
import os
import re
from .compile import expand_out_refs
from .dictionary import CompileError
ITEM_RE = re.compile(r"^[VP]\d+$")
# 点位集位置:本文件在 SGame/tools/event_authoring/ir_core/ -> 上溯 4 级到项目根
_POINTS_DIR = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
"Assets", "StreamingAssets", "Story", "PointSets")
def validate(ir, dic, points_dir=None, strict_points=False):
errs, warns = [], []
# ---- out_ref 结构预检(在展开前;展开依赖此处通过)----
seqs = {s["id"]: s for s in ir.get("sequences", [])}
for n in ir.get("nodes", []):
if n.get("kind") == "out_ref":
ref = n.get("ref")
if ref not in seqs:
errs.append("[out_ref失效] 节点 %s 引用了不存在的子序列 %r" % (n["id"], ref))
elif not seqs[ref].get("nodes"):
errs.append("[空子序列] 子序列 %r 没有任何节点(被 %s 引用)" % (ref, n["id"]))
for sn in seqs.get(ref, {}).get("nodes", []):
if sn.get("kind") == "out_ref":
errs.append("[嵌套out_ref] 子序列 %r 节点 %s 含嵌套 out_refM4 暂不支持"
% (ref, sn["id"]))
if errs:
return errs, warns # 结构坏了就别展开了,避免 KeyError 噪声
# ---- 展开 out_ref 后做节点级校验goto/win/lose/skip 目标含子序列内 id----
try:
nodes_list = expand_out_refs(ir)
except CompileError as e:
return ["[展开失败] %s" % e], warns
nodes = {n["id"]: n for n in nodes_list}
endings = {e["id"]: e for e in ir.get("endings", [])}
all_ids = set(nodes) | set(endings)
slots = {r["slot"] for r in ir.get("roles", [])}
def check_target(src, tid, field):
if tid not in all_ids:
errs.append("[跳转失效] 节点 %s%s 指向不存在的目标 %r" % (src, field, tid))
def check_slot(src, s, field):
if s and s not in slots:
errs.append("[未声明角色] 节点 %s%s 引用了未在 roles 声明的 slot %r" % (src, field, s))
def check_grants(src, grants):
for g in grants or []:
k = g.get("kind")
if not dic.is_known_grant(k):
errs.append("[未知奖励] 节点 %s 出现未登记的 grant kind %r" % (src, k)); continue
if k == "道具" and not ITEM_RE.match(str(g.get("item", ""))):
errs.append("[非法道具ID] 节点 %s 的道具 %r 不是合法 ID" % (src, g.get("item")))
spec = dic.grant_spec(k)
if spec.get("form") in ("money", "item", "friend") and \
not isinstance(g.get("value"), (int, float)):
errs.append("[非法数值] 节点 %s 的奖励数值 %r 不是数字" % (src, g.get("value")))
if dic.grant_needs_target(k):
check_slot(src, g.get("target"), "奖励target")
def check_cond(src, c):
if not c:
return
if not dic.is_known_cond(c.get("kind")):
errs.append("[非法条件] 节点 %s 的 condition %r 无法编译" % (src, c)); return
spec = dic.cond_spec(c.get("kind"))
if c.get("op") not in spec.get("ops", {}):
errs.append("[非法条件] 节点 %s 的 condition %r 比较符不支持" % (src, c))
elif not isinstance(c.get("value"), (int, float)):
errs.append("[非法条件数值] 节点 %s 的 condition 阈值 %r 不是数字" % (src, c.get("value")))
for nid, n in nodes.items():
kind = n.get("kind")
if n.get("next"):
check_target(nid, n["next"], "next")
if n.get("speaker"):
check_slot(nid, n["speaker"], "speaker")
if n.get("actor"):
check_slot(nid, n["actor"], "actor")
if kind in ("choice", "choice_once"):
opts = n.get("options", [])
if not opts:
errs.append("[空选择] 节点 %s 没有任何选项" % nid)
if not any("condition" not in o for o in opts):
errs.append("[选项无兜底] 节点 %s 的所有选项都带条件,可能全部不满足导致空面板卡死" % nid)
for o in opts:
check_target(nid, o.get("goto"), "option.goto")
check_cond(nid, o.get("condition"))
if o.get("reward"):
check_grants(nid, o["reward"].get("grants"))
if o.get("skip"):
sk = o["skip"]
check_target(nid, sk.get("node"), "option.skip.node")
grants = (sk.get("reward") or {}).get("grants")
if not grants:
errs.append("[空skip奖励] 节点 %s 选项 skip 缺 reward.grants押注无彩头" % nid)
check_grants(nid, grants)
if kind == "random":
for b in n.get("branches", []):
check_target(nid, b.get("goto"), "branch.goto")
if not isinstance(b.get("weight"), (int, float)):
errs.append("[非法权重] 节点 %s 的随机权重 %r 不是数字" % (nid, b.get("weight")))
if kind == "fight":
if n.get("fight_type") not in (1, 2):
errs.append("[非法战斗类型] 节点 %s 的 fight_type 必须是 1(击倒)/2(死斗)" % nid)
if not n.get("camp2"):
errs.append("[战斗缺敌方] 节点 %s 的 camp2 为空" % nid)
for s in (n.get("camp2") or []) + (n.get("camp1") or []):
check_slot(nid, s, "fight阵营")
check_target(nid, n.get("win"), "fight.win")
check_target(nid, n.get("lose"), "fight.lose")
if kind == "move":
check_slot(nid, n.get("actor"), "move.actor")
if kind == "reward":
check_grants(nid, n.get("grants"))
for e in ir.get("endings", []):
check_grants(e["id"], e.get("grants"))
# ---- 点位集坐标引用校验 ----
_check_point_set(ir, errs, warns, points_dir, strict_points)
return errs, warns
def collect_point_refs(ir):
"""收集 IR 里**实际需要物理站位**的点位名(角色出生点/移动目标/镜头点/战斗阵营)。
仅声明、从未在节点出现的 role如未登场的旁系角色不计——运行时不为其摆人。
"""
refs = set()
all_nodes = list(ir.get("nodes", []))
for s in ir.get("sequences", []):
all_nodes += s.get("nodes", [])
for n in all_nodes:
for f in ("speaker", "actor"):
v = n.get(f)
if v and v != "P1":
refs.add(v)
if n.get("kind") == "move" and n.get("to") and n["to"] != "P1":
refs.add(n["to"])
if n.get("camera"):
refs.add(n["camera"])
if n.get("kind") == "fight":
for s in (n.get("camp1") or []) + (n.get("camp2") or []):
if s and s != "P1":
refs.add(s)
return refs
def _check_point_set(ir, errs, warns, points_dir, strict_points):
import json
name = (ir.get("stage") or {}).get("point_set") or ir["id"]
base = points_dir or _POINTS_DIR
path = os.path.join(base, name + ".points.json")
refs = collect_point_refs(ir)
if not os.path.exists(path):
msg = "[点位集缺失] 未找到 %s.points.json%s);坐标校验跳过" % (name, path)
if strict_points:
errs.append(msg + "--strict-points 下视为错)")
else:
warns.append(msg)
return
try:
with open(path, encoding="utf-8") as f:
ps = json.load(f)
except Exception as e:
errs.append("[点位集损坏] %s 解析失败: %s" % (path, e)); return
have = {p.get("name") for p in ps.get("points", [])}
for r in sorted(refs):
if r not in have:
errs.append("[点位缺失] 点位 %r 在点位集 %s.points.json 中不存在(需取点工具补录)"
% (r, name))