277 lines
11 KiB
Python
277 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""拖拽编程画布——图校验器。
|
||
|
||
校验项（错误码见 ERR_*）：
|
||
- 无入口（没有任何事件块）／多入口（多个事件块，v1 允许，各自为独立分支）
|
||
- 孤立块（无任何连线）／不可达块（有连线但不在任何入口链上）
|
||
—— 可达性根 = 事件块 + 函数定义块（函数体经函数定义可达）
|
||
—— 豁免孤立检查：声明类块（var_create / logic_function_def，编译时提升/独立成区）
|
||
- 连线引用不存在的块或端口 / 自环 / 循环引用 / 重复连线
|
||
- 事件块被作为连线目标
|
||
- 块参数校验：必填缺失 / 类型错误（number、boolean、select）/ 数值越界 / 非法标识符
|
||
- 变量引用校验（@var 引用未创建变量）/ 函数调用引用未定义的函数
|
||
|
||
返回统一结构：{valid, errors: [{code, message, block_id, field}], warnings: [...]}
|
||
"""
|
||
|
||
import re
|
||
|
||
from .blocks import BLOCK_DEFS, get_dynamic_outputs
|
||
|
||
# 错误码
|
||
ERR_MISSING_ID = "BLOCK_MISSING_ID"
|
||
ERR_DUP_ID = "BLOCK_DUP_ID"
|
||
ERR_UNKNOWN_TYPE = "BLOCK_UNKNOWN_TYPE"
|
||
ERR_NO_ENTRY = "NO_ENTRY"
|
||
ERR_ORPHAN = "ORPHAN_BLOCK"
|
||
ERR_UNREACHABLE = "UNREACHABLE_BLOCK"
|
||
ERR_CONN_BLOCK_NOT_FOUND = "CONN_BLOCK_NOT_FOUND"
|
||
ERR_CONN_PORT_NOT_FOUND = "CONN_PORT_NOT_FOUND"
|
||
ERR_CONN_DUP = "CONN_DUPLICATE"
|
||
ERR_CONN_SELF = "CONN_SELF_LOOP"
|
||
ERR_CONN_CYCLE = "CONN_CYCLE"
|
||
ERR_EVENT_AS_TARGET = "EVENT_AS_TARGET"
|
||
ERR_PARAM_REQUIRED = "PARAM_REQUIRED"
|
||
ERR_PARAM_TYPE = "PARAM_TYPE"
|
||
ERR_PARAM_RANGE = "PARAM_RANGE"
|
||
ERR_PARAM_OPTION = "PARAM_OPTION"
|
||
ERR_IDENTIFIER = "INVALID_IDENTIFIER"
|
||
ERR_VAR_UNDEFINED = "VAR_UNDEFINED"
|
||
ERR_FUNC_UNDEFINED = "FUNC_UNDEFINED"
|
||
ERR_BAD_GRAPH = "BAD_GRAPH"
|
||
|
||
_IDENT_RE = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')
|
||
|
||
# 声明类块：编译时被提升（变量）或独立成区（函数定义），不需要参与孤立检查
|
||
_DECLARATION_TYPES = ('var_create', 'logic_function_def')
|
||
|
||
|
||
def _err(code, message, block_id=None, field=None):
|
||
e = {"code": code, "message": message}
|
||
if block_id is not None:
|
||
e["block_id"] = block_id
|
||
if field is not None:
|
||
e["field"] = field
|
||
return e
|
||
|
||
|
||
def _is_number(s):
|
||
try:
|
||
float(s)
|
||
return True
|
||
except (TypeError, ValueError):
|
||
return False
|
||
|
||
|
||
def _is_bool(s):
|
||
return s in (True, False, 'true', 'false', '1', '0', 1, 0)
|
||
|
||
|
||
def _block_outputs(btype, params):
|
||
"""块的全部输出端口名（含动态端口）。"""
|
||
bdef = BLOCK_DEFS.get(btype) or {}
|
||
names = [p["name"] for p in (bdef.get("ports") or {}).get("outputs", [])]
|
||
names += get_dynamic_outputs(btype, params)
|
||
return names
|
||
|
||
|
||
def _block_inputs(btype):
|
||
bdef = BLOCK_DEFS.get(btype) or {}
|
||
return [p["name"] for p in (bdef.get("ports") or {}).get("inputs", [])]
|
||
|
||
|
||
def _block_category(btype):
|
||
bdef = BLOCK_DEFS.get(btype) or {}
|
||
return bdef.get("category")
|
||
|
||
|
||
def validate_graph(graph):
|
||
"""graph: {blocks: [...], connections: [...]} → {valid, errors, warnings}"""
|
||
errors, warnings = [], []
|
||
if not isinstance(graph, dict):
|
||
return {"valid": False, "errors": [_err(ERR_BAD_GRAPH, "画布结构非法：必须是对象")], "warnings": []}
|
||
blocks = graph.get("blocks") or []
|
||
conns = graph.get("connections") or []
|
||
if not isinstance(blocks, list) or not isinstance(conns, list):
|
||
return {"valid": False, "errors": [_err(ERR_BAD_GRAPH, "画布结构非法：blocks/connections 必须是数组")], "warnings": []}
|
||
|
||
bmap = {}
|
||
for b in blocks:
|
||
bid = b.get("id")
|
||
if not bid:
|
||
errors.append(_err(ERR_MISSING_ID, "块缺少 id"))
|
||
continue
|
||
if bid in bmap:
|
||
errors.append(_err(ERR_DUP_ID, "块 id 重复：%s" % bid, bid))
|
||
continue
|
||
bmap[bid] = b
|
||
btype = b.get("type")
|
||
if btype not in BLOCK_DEFS:
|
||
errors.append(_err(ERR_UNKNOWN_TYPE, "未知块类型：%s" % btype, bid))
|
||
|
||
if not blocks:
|
||
return {"valid": False, "errors": [_err(ERR_NO_ENTRY, "画布为空：没有块")], "warnings": []}
|
||
|
||
# ── 连线基础校验 ──
|
||
seen = set()
|
||
for c in conns:
|
||
frm, to = c.get("from"), c.get("to")
|
||
fp, tp = c.get("fromPort"), c.get("toPort")
|
||
key = "%s:%s->%s:%s" % (frm, fp, to, tp)
|
||
if frm == to:
|
||
errors.append(_err(ERR_CONN_SELF, "自环连线：%s" % frm, frm))
|
||
if key in seen:
|
||
errors.append(_err(ERR_CONN_DUP, "重复连线：%s -> %s" % (frm, to), frm))
|
||
seen.add(key)
|
||
if frm not in bmap:
|
||
errors.append(_err(ERR_CONN_BLOCK_NOT_FOUND, "连线起点块不存在：%s" % frm, None))
|
||
continue
|
||
if to not in bmap:
|
||
errors.append(_err(ERR_CONN_BLOCK_NOT_FOUND, "连线终点块不存在：%s" % to, None))
|
||
continue
|
||
fb, tb = bmap[frm], bmap[to]
|
||
fbtype, tbtype = fb.get("type"), tb.get("type")
|
||
# 事件块不能作为目标
|
||
if _block_category(tbtype) == "event":
|
||
errors.append(_err(ERR_EVENT_AS_TARGET, "事件块不能作为连线目标：%s" % to, to))
|
||
# 端口存在性
|
||
if fp not in _block_outputs(fbtype, fb.get("params") or {}):
|
||
errors.append(_err(ERR_CONN_PORT_NOT_FOUND, "起点端口不存在：%s.%s" % (frm, fp), frm))
|
||
if tp not in _block_inputs(tbtype):
|
||
errors.append(_err(ERR_CONN_PORT_NOT_FOUND, "终点端口不存在：%s.%s" % (to, tp), to))
|
||
|
||
# ── 入口检查 ──
|
||
events = [b for b in blocks if _block_category(b.get("type")) == "event"]
|
||
if not events:
|
||
errors.append(_err(ERR_NO_ENTRY, "没有入口：画布必须包含至少一个事件块（开始/点击/悬停/定时/碰撞/键盘）"))
|
||
|
||
# ── 孤立块 / 可达性 ──
|
||
connected_ids = set()
|
||
for c in conns:
|
||
connected_ids.add(c.get("from"))
|
||
connected_ids.add(c.get("to"))
|
||
|
||
def _exempt_orphan(b):
|
||
"""孤立检查豁免：事件入口 + 声明类块。"""
|
||
btype = b.get("type")
|
||
return _block_category(btype) == "event" or btype in _DECLARATION_TYPES
|
||
|
||
for b in blocks:
|
||
if b.get("id") not in connected_ids and not _exempt_orphan(b):
|
||
errors.append(_err(ERR_ORPHAN, "孤立块（没有任何连线）：%s [%s]" % (b.get("id"), b.get("type")), b.get("id")))
|
||
|
||
# BFS 可达性：根 = 事件块 + 函数定义块（函数体经函数定义可达）
|
||
adj = {}
|
||
for c in conns:
|
||
adj.setdefault(c.get("from"), []).append(c)
|
||
roots = [b.get("id") for b in events if b.get("id")] + \
|
||
[b.get("id") for b in blocks if b.get("type") == "logic_function_def" and b.get("id")]
|
||
visited = set()
|
||
stack = list(roots)
|
||
while stack:
|
||
bid = stack.pop()
|
||
if bid in visited:
|
||
continue
|
||
visited.add(bid)
|
||
for c in adj.get(bid, []):
|
||
if c.get("to") not in visited:
|
||
stack.append(c.get("to"))
|
||
for b in blocks:
|
||
bid = b.get("id")
|
||
if bid and bid not in visited and not _exempt_orphan(b):
|
||
errors.append(_err(ERR_UNREACHABLE, "不可达块（不在任何事件/函数链上）：%s [%s]" % (bid, b.get("type")), bid))
|
||
|
||
# ── 环检测（DFS 三色） ──
|
||
WHITE, GRAY, BLACK = 0, 1, 2
|
||
color = {bid: WHITE for bid in bmap}
|
||
|
||
def dfs(bid):
|
||
color[bid] = GRAY
|
||
for c in adj.get(bid, []):
|
||
nxt = c.get("to")
|
||
if nxt in color and color[nxt] == GRAY:
|
||
return True
|
||
if nxt in color and color[nxt] == WHITE:
|
||
if dfs(nxt):
|
||
return True
|
||
color[bid] = BLACK
|
||
return False
|
||
|
||
for bid in list(bmap):
|
||
if color.get(bid) == WHITE and dfs(bid):
|
||
errors.append(_err(ERR_CONN_CYCLE, "检测到循环引用（环形连线），无法编译", bid))
|
||
break
|
||
|
||
# ── 参数校验 ──
|
||
defined_vars = set()
|
||
defined_funcs = set()
|
||
for b in blocks:
|
||
btype = b.get("type")
|
||
if btype == "var_create":
|
||
nm = (b.get("params") or {}).get("name")
|
||
if nm:
|
||
defined_vars.add(str(nm).strip())
|
||
if btype == "logic_function_def":
|
||
nm = (b.get("params") or {}).get("name")
|
||
if nm:
|
||
defined_funcs.add(str(nm).strip())
|
||
|
||
for b in blocks:
|
||
bid = b.get("id")
|
||
btype = b.get("type")
|
||
bdef = BLOCK_DEFS.get(btype)
|
||
if not bdef:
|
||
continue
|
||
params = b.get("params") or {}
|
||
if not isinstance(params, dict):
|
||
errors.append(_err(ERR_PARAM_TYPE, "块参数必须是对象：%s" % bid, bid))
|
||
params = {}
|
||
for pdef in bdef.get("params", []):
|
||
pname = pdef["name"]
|
||
val = params.get(pname)
|
||
# 必填
|
||
if pdef.get("required") and (val is None or str(val).strip() == ""):
|
||
errors.append(_err(ERR_PARAM_REQUIRED, "参数「%s」必填" % pdef["label"], bid, pname))
|
||
continue
|
||
if val is None or str(val).strip() == "":
|
||
continue
|
||
val = str(val).strip()
|
||
# 类型
|
||
ut = pdef.get("uitype")
|
||
if ut == "number":
|
||
if not _is_number(val):
|
||
errors.append(_err(ERR_PARAM_TYPE, "参数「%s」必须是数字" % pdef["label"], bid, pname))
|
||
else:
|
||
f = float(val)
|
||
if "min" in pdef and f < pdef["min"]:
|
||
errors.append(_err(ERR_PARAM_RANGE, "参数「%s」不能小于 %s" % (pdef["label"], pdef["min"]), bid, pname))
|
||
if "max" in pdef and f > pdef["max"]:
|
||
errors.append(_err(ERR_PARAM_RANGE, "参数「%s」不能大于 %s" % (pdef["label"], pdef["max"]), bid, pname))
|
||
elif ut == "boolean":
|
||
if not _is_bool(val):
|
||
errors.append(_err(ERR_PARAM_TYPE, "参数「%s」必须是布尔值" % pdef["label"], bid, pname))
|
||
elif ut == "select":
|
||
opts = pdef.get("options") or []
|
||
if opts and val not in [o["value"] for o in opts]:
|
||
errors.append(_err(ERR_PARAM_OPTION, "参数「%s」取值非法：%s" % (pdef["label"], val), bid, pname))
|
||
# 标识符类参数
|
||
if pname in ("name", "result") and ut in ("text", "variable"):
|
||
if not _IDENT_RE.match(val):
|
||
errors.append(_err(ERR_IDENTIFIER, "参数「%s」必须是合法标识符（字母/数字/下划线，不能以数字开头）：%s" % (pdef["label"], val), bid, pname))
|
||
# 变量引用校验
|
||
if ut in ("expression", "variable"):
|
||
for token in re.findall(r'@([A-Za-z_][A-Za-z0-9_]*)', val):
|
||
if token not in defined_vars:
|
||
errors.append(_err(ERR_VAR_UNDEFINED, "引用了未创建的变量：@%s" % token, bid, pname))
|
||
# 函数调用校验
|
||
if btype == "logic_function_call":
|
||
fname = (params.get("name") or "").strip()
|
||
if fname and fname not in defined_funcs:
|
||
errors.append(_err(ERR_FUNC_UNDEFINED, "调用了未定义的函数：%s" % fname, bid, "name"))
|
||
if btype == "logic_function_def":
|
||
fname = (params.get("name") or "").strip()
|
||
if not fname:
|
||
errors.append(_err(ERR_PARAM_REQUIRED, "函数名必填", bid, "name"))
|
||
|
||
return {"valid": not errors, "errors": errors, "warnings": warnings}
|