276 lines
11 KiB
Python
276 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""拖拽编程画布——图编译为 script_engine 可执行脚本（Python，script_type=0）。
|
||
|
||
编译策略：
|
||
- 变量声明统一提升到脚本头部（var_create），保证使用前已定义
|
||
- 函数定义（logic_function_def）生成 def + 函数体链
|
||
- 每个事件块作为独立入口分支
|
||
- 逻辑块生成控制流（for / while / if / else / 并行顺序展开）
|
||
- 动作块生成动作函数调用（move/rotate/scale/set_property/play_animation/
|
||
play_sound/show_hide/camera/wait），由宿主运行时注册到 script_engine 白名单
|
||
- 产物仅含白名单兼容语法：赋值 / def / for / while / if-else / 函数调用 / 注释，
|
||
无 import、无 class、无 lambda、无对象方法调用、无 async
|
||
- 表达式内 `@变量名` 引用统一替换为 Python 标识符（变量提升到头部后按名引用）
|
||
"""
|
||
|
||
import re
|
||
|
||
from .blocks import BLOCK_DEFS, get_dynamic_outputs
|
||
from .validator import validate_graph
|
||
|
||
_IDENT_RE = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')
|
||
_SAFE_EXPR = re.compile(r"^[A-Za-z0-9_+\-*/%()<>=!&| .'\"]+$")
|
||
_VAR_REF = re.compile(r'@([A-Za-z_][A-Za-z0-9_]*)')
|
||
|
||
|
||
def _ident(name):
|
||
s = re.sub(r'\W', '_', str(name or ''))
|
||
if not s:
|
||
return '_v'
|
||
if s[0].isdigit():
|
||
s = '_' + s
|
||
return s
|
||
|
||
|
||
def _num(s):
|
||
s = str(s or '').strip()
|
||
try:
|
||
float(s)
|
||
return s
|
||
except (TypeError, ValueError):
|
||
return '0'
|
||
|
||
|
||
def _str(s):
|
||
return "'%s'" % str(s or '').replace("'", "\\'")
|
||
|
||
|
||
def _expr(s):
|
||
"""表达式安全转义：表达式内 @var 引用 → 标识符；纯数字/安全表达式原样；其余当字符串。"""
|
||
s = str(s or '').strip()
|
||
s = _VAR_REF.sub(lambda m: _ident(m.group(1)), s)
|
||
try:
|
||
float(s)
|
||
return s
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if (_SAFE_EXPR.match(s) and 'import' not in s and '__' not in s and ';' not in s
|
||
and not s.startswith('lambda') and not s.startswith('class')):
|
||
return s
|
||
return _str(s)
|
||
|
||
|
||
def _val(s):
|
||
"""值转义：@var → 标识符；数字 → 原样；true/false → 布尔；其余 → 字符串字面量。"""
|
||
s = str(s or '').strip()
|
||
if s.startswith('@') and _IDENT_RE.match(s[1:]):
|
||
return _ident(s[1:])
|
||
try:
|
||
f = float(s)
|
||
return str(int(f)) if f == int(f) else s
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if s in ('true', 'True', '1'):
|
||
return 'True'
|
||
if s in ('false', 'False', '0'):
|
||
return 'False'
|
||
return _str(s)
|
||
|
||
|
||
def _block_label(btype):
|
||
bdef = BLOCK_DEFS.get(btype) or {}
|
||
return bdef.get("label", btype)
|
||
|
||
|
||
def _block_code(b):
|
||
"""生成单个块的执行代码行；返回 None 表示该块不生成独立代码（结构块/已提升）。"""
|
||
btype = b.get("type")
|
||
p = b.get("params") or {}
|
||
if btype == "event_start":
|
||
return "# 事件: 开始"
|
||
if btype == "event_click":
|
||
return "# 事件: 点击 %s (%s)" % (p.get("target", ''), p.get("button", 'left'))
|
||
if btype == "event_hover":
|
||
return "# 事件: 悬停 %s %s" % (p.get("target", ''), p.get("state", 'enter'))
|
||
if btype == "event_timer":
|
||
return "# 事件: 定时 每 %s 秒 × %s 次" % (p.get("interval", ''), p.get("repeat", '-1'))
|
||
if btype == "event_collision":
|
||
return "# 事件: 碰撞 %s × %s" % (p.get("entity_a", ''), p.get("entity_b", ''))
|
||
if btype == "event_keyboard":
|
||
return "# 事件: 键盘 %s (%s)" % (p.get("key", ''), p.get("action", 'keydown'))
|
||
if btype == "logic_sequence":
|
||
return "# 顺序执行"
|
||
if btype == "logic_wait":
|
||
return "wait(seconds=%s)" % _num(p.get("seconds", '0'))
|
||
if btype == "logic_function_call":
|
||
return "%s()" % _ident(p.get("name", 'fn'))
|
||
if btype == "var_create":
|
||
return None # 已提升到头部变量声明区
|
||
if btype == "var_assign":
|
||
return "%s = %s" % (_ident(p.get("name", '_v')), _val(p.get("value", '0')))
|
||
if btype == "var_read":
|
||
return "%s = %s" % (_ident(p.get("result", '_r')), _ident(p.get("name", '_v')))
|
||
if btype == "expr_arith":
|
||
return "%s = %s %s %s" % (_ident(p.get("result", '_r')), _expr(p.get("operand_a", '0')),
|
||
p.get("operator", '+'), _expr(p.get("operand_b", '0')))
|
||
if btype == "expr_compare":
|
||
return "%s = %s %s %s" % (_ident(p.get("result", '_r')), _expr(p.get("operand_a", '0')),
|
||
p.get("operator", '=='), _expr(p.get("operand_b", '0')))
|
||
if btype == "expr_logic":
|
||
op = p.get("operator", 'and')
|
||
a, bv = _expr(p.get("operand_a", 'True')), _expr(p.get("operand_b", 'True'))
|
||
body = "not %s" % a if op == 'not' else "%s %s %s" % (a, op, bv)
|
||
return "%s = %s" % (_ident(p.get("result", '_r')), body)
|
||
# 动作
|
||
if btype == "action_move":
|
||
return "move(entity=%s, x=%s, y=%s, duration=%s)" % (
|
||
_str(p.get("entity", '')), _num(p.get("x", '0')), _num(p.get("y", '0')), _num(p.get("duration", '0')))
|
||
if btype == "action_rotate":
|
||
return "rotate(entity=%s, angle=%s, duration=%s)" % (
|
||
_str(p.get("entity", '')), _num(p.get("angle", '0')), _num(p.get("duration", '0')))
|
||
if btype == "action_scale":
|
||
return "scale(entity=%s, scale_x=%s, scale_y=%s, duration=%s)" % (
|
||
_str(p.get("entity", '')), _num(p.get("scale_x", '1')), _num(p.get("scale_y", '1')),
|
||
_num(p.get("duration", '0')))
|
||
if btype == "action_set_property":
|
||
return "set_property(entity=%s, property=%s, value=%s)" % (
|
||
_str(p.get("entity", '')), _str(p.get("property", '')), _str(p.get("value", '')))
|
||
if btype == "action_play_animation":
|
||
return "play_animation(entity=%s, animation=%s)" % (
|
||
_str(p.get("entity", '')), _str(p.get("animation", '')))
|
||
if btype == "action_play_sound":
|
||
return "play_sound(entity=%s, sound=%s)" % (_str(p.get("entity", '')), _str(p.get("sound", '')))
|
||
if btype == "action_show_hide":
|
||
return "show_hide(entity=%s, visible=%s)" % (_str(p.get("entity", '')), _val(p.get("visible", 'true')))
|
||
if btype == "action_camera":
|
||
return "camera(camera_id=%s, mode=%s, x=%s, y=%s, zoom=%s)" % (
|
||
_str(p.get("camera_id", 'main')), _str(p.get("mode", 'follow')),
|
||
_num(p.get("x", '0')), _num(p.get("y", '0')), _num(p.get("zoom", '1')))
|
||
return None
|
||
|
||
|
||
def _emit_chain(bid, bmap, adj, indent, visited=None):
|
||
"""沿输出端口生成执行代码（每块仅访问一次防环）。返回代码行列表。"""
|
||
if visited is None:
|
||
visited = set()
|
||
if bid in visited or bid not in bmap:
|
||
return []
|
||
visited.add(bid)
|
||
b = bmap[bid]
|
||
btype = b.get("type")
|
||
p = b.get("params") or {}
|
||
ind = ' ' * indent
|
||
lines = []
|
||
outs = adj.get(bid, [])
|
||
|
||
if btype == "logic_condition":
|
||
lines.append(ind + "if %s:" % _expr(p.get("condition", 'True')))
|
||
t = [c for c in outs if c.get("fromPort") == 'true']
|
||
if t:
|
||
lines.extend(_emit_chain(t[0].get("to"), bmap, adj, indent + 1, visited))
|
||
else:
|
||
lines.append(ind + " pass")
|
||
lines.append(ind + "else:")
|
||
f = [c for c in outs if c.get("fromPort") == 'false']
|
||
if f:
|
||
lines.extend(_emit_chain(f[0].get("to"), bmap, adj, indent + 1, visited))
|
||
else:
|
||
lines.append(ind + " pass")
|
||
return lines
|
||
|
||
if btype == "logic_loop":
|
||
mode = p.get("mode", 'count')
|
||
if mode == 'while':
|
||
lines.append(ind + "while %s:" % _expr(p.get("condition", 'True')))
|
||
else:
|
||
lines.append(ind + "for _i in range(%s):" % _num(p.get("times", '3')))
|
||
nxt = [c for c in outs if c.get("fromPort") == 'out']
|
||
if nxt:
|
||
lines.extend(_emit_chain(nxt[0].get("to"), bmap, adj, indent + 1, visited))
|
||
else:
|
||
lines.append(ind + " pass")
|
||
return lines
|
||
|
||
if btype in ("logic_branch", "logic_parallel"):
|
||
prefix = 'branch' if btype == "logic_branch" else 'p'
|
||
names = get_dynamic_outputs(btype, p) or []
|
||
for i, port in enumerate(names, 1):
|
||
lines.append(ind + "# %s分支%d" % ("并行" if btype == "logic_parallel" else "", i))
|
||
br = [c for c in outs if c.get("fromPort") == port]
|
||
if br:
|
||
lines.extend(_emit_chain(br[0].get("to"), bmap, adj, indent, visited))
|
||
else:
|
||
lines.append(ind + "pass")
|
||
return lines
|
||
|
||
# 普通块
|
||
code = _block_code(b)
|
||
if code is not None:
|
||
lines.append(ind + code)
|
||
# 单输出继续
|
||
nxt = [c for c in outs if c.get("fromPort") == 'out']
|
||
if nxt:
|
||
lines.extend(_emit_chain(nxt[0].get("to"), bmap, adj, indent, visited))
|
||
return lines
|
||
|
||
|
||
def compile_graph(graph, meta=None):
|
||
"""编译画布 → {success, content, warnings, entry_points} 或 {success:False, errors:[...]}"""
|
||
v = validate_graph(graph)
|
||
if v["errors"]:
|
||
return {"success": False, "valid": False, "errors": v["errors"]}
|
||
|
||
blocks = graph.get("blocks") or []
|
||
conns = graph.get("connections") or []
|
||
bmap = {b.get("id"): b for b in blocks if b.get("id")}
|
||
adj = {}
|
||
for c in conns:
|
||
adj.setdefault(c.get("from"), []).append(c)
|
||
|
||
events = [b for b in blocks if (BLOCK_DEFS.get(b.get("type")) or {}).get("category") == "event"]
|
||
|
||
lines = []
|
||
lines.append("# -*- coding: utf-8 -*-")
|
||
lines.append("# 拖拽编程画布编译产物 (drag compiler v1)")
|
||
if meta:
|
||
lines.append("# 画布: %s (id=%s)" % (meta.get("name", ''), meta.get("id", '')))
|
||
lines.append("# 入口事件: %s" % ", ".join(
|
||
"%s[%s]" % (e.get("id"), _block_label(e.get("type"))) for e in events))
|
||
lines.append("")
|
||
|
||
# 变量声明区（提升）
|
||
var_decls = []
|
||
for b in blocks:
|
||
if b.get("type") == "var_create":
|
||
p = b.get("params") or {}
|
||
var_decls.append("%s = %s" % (_ident(p.get("name", '_v')), _val(p.get("initial_value", '0'))))
|
||
if var_decls:
|
||
lines.append("# ===== 变量声明 =====")
|
||
lines.extend(var_decls)
|
||
lines.append("")
|
||
|
||
# 函数定义区
|
||
funcs = [b for b in blocks if b.get("type") == "logic_function_def"]
|
||
if funcs:
|
||
lines.append("# ===== 函数定义 =====")
|
||
for fb in funcs:
|
||
p = fb.get("params") or {}
|
||
lines.append("def %s():" % _ident(p.get("name", 'fn')))
|
||
body = _emit_chain(fb.get("id"), bmap, adj, 1, set())
|
||
lines.extend(body or [" pass"])
|
||
lines.append("")
|
||
|
||
# 入口执行区（每个事件块一个分支）
|
||
if not events:
|
||
lines.append("# (无入口事件)")
|
||
for ev in events:
|
||
lines.append("# ===== 入口: %s [%s] =====" % (ev.get("id"), _block_label(ev.get("type"))))
|
||
body = _emit_chain(ev.get("id"), bmap, adj, 0, set())
|
||
lines.extend(body or ["pass"])
|
||
lines.append("")
|
||
|
||
content = "\n".join(lines).rstrip() + "\n"
|
||
return {"success": True, "valid": True, "content": content,
|
||
"warnings": v["warnings"],
|
||
"entry_points": [e.get("id") for e in events]}
|