807 lines
32 KiB
Python
807 lines
32 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
__all__ = [
|
||
"_OUT_MACRO_RE",
|
||
"_VAR_MACRO_RE",
|
||
"_PROMPT_MACRO_RE",
|
||
"_OUT_SHORT_RE",
|
||
"_BARE_MACRO_RE",
|
||
"_BRACES_RE",
|
||
"_split_path",
|
||
"_get_by_path",
|
||
"_stringify_for_template",
|
||
"_deep_find_text",
|
||
"_best_text_from_outputs",
|
||
"render_template_simple",
|
||
"eval_condition_expr",
|
||
]
|
||
|
||
# Regex-макросы (общие для бэка)
|
||
_OUT_MACRO_RE = re.compile(r"\[\[\s*OUT\s*[:\s]\s*([^\]]+?)\s*\]\]", re.IGNORECASE)
|
||
_VAR_MACRO_RE = re.compile(r"\[\[\s*VAR\s*[:\s]\s*([^\]]+?)\s*\]\]", re.IGNORECASE)
|
||
# STORE: постоянное хранилище переменных (пер-пайплайн)
|
||
_STORE_MACRO_RE = re.compile(r"\[\[\s*STORE\s*[:\s]\s*([^\]]+?)\s*\]\]", re.IGNORECASE)
|
||
# Единый фрагмент PROMPT (провайдеро-специфичный JSON-фрагмент)
|
||
_PROMPT_MACRO_RE = re.compile(r"\[\[\s*PROMPT\s*\]\]", re.IGNORECASE)
|
||
# Короткая форма: [[OUT1]] — best-effort текст из ноды n1
|
||
_OUT_SHORT_RE = re.compile(r"\[\[\s*OUT\s*(\d+)\s*\]\]", re.IGNORECASE)
|
||
# Голые переменные: [[NAME]] или [[path.to.value]] — сначала ищем в vars, затем в контексте
|
||
_BARE_MACRO_RE = re.compile(r"\[\[\s*([A-Za-z_][A-Za-z0-9_]*(?:\.[^\]]+?)?)\s*\]\]")
|
||
# Подстановки {{ ... }} (включая простейший фильтр |default(...))
|
||
# Разбираем выражение до ближайшего '}}', допускаем '}' внутри (например в JSON-литералах)
|
||
_BRACES_RE = re.compile(r"\{\{\s*(.*?)\s*\}\}", re.DOTALL)
|
||
|
||
# Сокращённый синтаксис: img(mime?)[[...]] → data:<mime>;base64,<resolved_inner_macro>
|
||
# Пример: img()[[OUT1]] → data:image/png;base64,{{resolved OUT1}}
|
||
# img(jpeg)[[OUT:n1.result...]] → data:image/jpeg;base64,{{resolved}}
|
||
_IMG_WRAPPER_RE = re.compile(r"(?is)img\(\s*([^)]+?)?\s*\)\s*\[\[\s*(.+?)\s*\]\]")
|
||
|
||
|
||
def _split_path(path: str) -> List[str]:
|
||
return [p.strip() for p in str(path).split(".") if str(p).strip()]
|
||
|
||
|
||
def _get_by_path(obj: Any, path: Optional[str]) -> Any:
|
||
if path is None or path == "":
|
||
return obj
|
||
cur = obj
|
||
for seg in _split_path(path):
|
||
if isinstance(cur, dict):
|
||
if seg in cur:
|
||
cur = cur[seg]
|
||
else:
|
||
return None
|
||
elif isinstance(cur, list):
|
||
try:
|
||
idx = int(seg)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
if 0 <= idx < len(cur):
|
||
cur = cur[idx]
|
||
else:
|
||
return None
|
||
else:
|
||
return None
|
||
return cur
|
||
|
||
|
||
def _stringify_for_template(val: Any) -> str:
|
||
if val is None:
|
||
return ""
|
||
if isinstance(val, bool):
|
||
# JSON-friendly booleans (useful when embedding into JSON-like templates)
|
||
return "true" if val else "false"
|
||
if isinstance(val, (dict, list)):
|
||
try:
|
||
return json.dumps(val, ensure_ascii=False)
|
||
except Exception: # noqa: BLE001
|
||
return str(val)
|
||
return str(val)
|
||
|
||
|
||
def _deep_find_text(obj: Any, max_nodes: int = 5000) -> Optional[str]:
|
||
"""
|
||
Best-effort поиск первого текстового значения в глубине структуры JSON.
|
||
Сначала пытаемся по ключам content/text, затем общий обход.
|
||
"""
|
||
try:
|
||
# Быстрые ветки
|
||
if isinstance(obj, str):
|
||
return obj
|
||
if isinstance(obj, dict):
|
||
c = obj.get("content")
|
||
if isinstance(c, str):
|
||
return c
|
||
t = obj.get("text")
|
||
if isinstance(t, str):
|
||
return t
|
||
parts = obj.get("parts")
|
||
if isinstance(parts, list) and parts:
|
||
for p in parts:
|
||
if isinstance(p, dict) and isinstance(p.get("text"), str):
|
||
return p.get("text")
|
||
|
||
# Общий нерекурсивный обход в ширину
|
||
queue: List[Any] = [obj]
|
||
seen = 0
|
||
while queue and seen < max_nodes:
|
||
cur = queue.pop(0)
|
||
seen += 1
|
||
if isinstance(cur, str):
|
||
return cur
|
||
if isinstance(cur, dict):
|
||
# часто встречающиеся поля
|
||
for k in ("text", "content"):
|
||
v = cur.get(k)
|
||
if isinstance(v, str):
|
||
return v
|
||
# складываем все значения
|
||
for v in cur.values():
|
||
queue.append(v)
|
||
elif isinstance(cur, list):
|
||
for it in cur:
|
||
queue.append(it)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _best_text_from_outputs(node_out: Any) -> str:
|
||
"""
|
||
Унифицированное извлечение "текста" из выхода ноды.
|
||
Поддерживает:
|
||
- PromptTemplate: {"text": ...}
|
||
- LLMInvoke: {"response_text": ...}
|
||
- ProviderCall/RawForward: {"result": <provider_json>}, извлекаем текст для openai/gemini/claude
|
||
- Общий глубокий поиск текста, если специфичные ветки не сработали
|
||
"""
|
||
# Строка сразу
|
||
if isinstance(node_out, str):
|
||
return node_out
|
||
|
||
if not isinstance(node_out, dict):
|
||
return ""
|
||
|
||
# Явные короткие поля
|
||
if isinstance(node_out.get("response_text"), str) and node_out.get("response_text"):
|
||
return str(node_out["response_text"])
|
||
if isinstance(node_out.get("text"), str) and node_out.get("text"):
|
||
return str(node_out["text"])
|
||
|
||
res = node_out.get("result")
|
||
base = res if isinstance(res, (dict, list)) else node_out
|
||
|
||
# OpenAI
|
||
try:
|
||
if isinstance(base, dict):
|
||
ch0 = (base.get("choices") or [{}])[0]
|
||
msg = ch0.get("message") or {}
|
||
c = msg.get("content")
|
||
if isinstance(c, str):
|
||
return c
|
||
except Exception:
|
||
pass
|
||
|
||
# Gemini
|
||
try:
|
||
if isinstance(base, dict):
|
||
cands = base.get("candidates") or []
|
||
texts: List[str] = []
|
||
for cand in cands:
|
||
try:
|
||
content = cand.get("content") or {}
|
||
parts = content.get("parts") or []
|
||
for p in parts:
|
||
if isinstance(p, dict):
|
||
t = p.get("text")
|
||
if isinstance(t, str) and t.strip():
|
||
texts.append(t.strip())
|
||
except Exception:
|
||
continue
|
||
if texts:
|
||
return "\n".join(texts)
|
||
except Exception:
|
||
pass
|
||
|
||
# Claude
|
||
try:
|
||
if isinstance(base, dict):
|
||
blocks = base.get("content") or []
|
||
texts = [b.get("text") for b in blocks if isinstance(b, dict) and isinstance(b.get("text"), str)]
|
||
if texts:
|
||
return "\n".join(texts)
|
||
except Exception:
|
||
pass
|
||
|
||
# Общий глубокий поиск
|
||
txt = _deep_find_text(base)
|
||
return txt or ""
|
||
|
||
|
||
def render_template_simple(template: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> str:
|
||
"""
|
||
Простая подстановка:
|
||
- {{ path }} — берёт из context (или {{ OUT.node.path }} для выходов)
|
||
- Поддержка фильтра по умолчанию: {{ path|default(value) }}
|
||
value может быть числом, строкой ('..'/".."), массивом/объектом в виде литерала.
|
||
- [[VAR:path]] — берёт из context
|
||
- [[OUT:nodeId(.path)*]] — берёт из out_map
|
||
- [[STORE:path]] — берёт из постоянного хранилища (context.store)
|
||
Возвращает строку.
|
||
"""
|
||
if template is None:
|
||
return ""
|
||
s = str(template)
|
||
|
||
# 0) Сокращённый синтаксис: img(mime?)[[...]] → data:<mime>;base64,<resolved>
|
||
# Выполняем до развёртки обычных [[...]] макросов, чтобы внутри можно было использовать любой квадратный макрос.
|
||
def _normalize_mime(m: str) -> str:
|
||
mm = (m or "").strip().lower()
|
||
if not mm:
|
||
return "image/png"
|
||
if "/" in mm:
|
||
return mm
|
||
return {
|
||
"png": "image/png",
|
||
"jpg": "image/jpeg",
|
||
"jpeg": "image/jpeg",
|
||
"webp": "image/webp",
|
||
"gif": "image/gif",
|
||
"svg": "image/svg+xml",
|
||
"bmp": "image/bmp",
|
||
"tif": "image/tiff",
|
||
"tiff": "image/tiff",
|
||
}.get(mm, mm)
|
||
|
||
def _repl_imgwrap(m: re.Match) -> str:
|
||
mime_raw = m.group(1) or ""
|
||
inner = m.group(2) or ""
|
||
mime = _normalize_mime(mime_raw)
|
||
try:
|
||
val = _resolve_square_macro_value(inner, context, out_map)
|
||
except Exception:
|
||
val = ""
|
||
if isinstance(val, (dict, list, bool)) or val is None:
|
||
val = _stringify_for_template(val)
|
||
else:
|
||
val = str(val)
|
||
return f"data:{mime};base64,{val}"
|
||
|
||
# Поддерживаем много вхождений — повторяем до исчерпания (на случай каскадных макросов)
|
||
while True:
|
||
ns, cnt = _IMG_WRAPPER_RE.subn(_repl_imgwrap, s)
|
||
s = ns
|
||
if cnt == 0:
|
||
break
|
||
|
||
# 1) Макросы [[VAR:...]] / [[OUT:...]] / [[STORE:...]]
|
||
def repl_var(m: re.Match) -> str:
|
||
path = m.group(1).strip()
|
||
val = _get_by_path(context, path)
|
||
return _stringify_for_template(val)
|
||
|
||
def repl_out(m: re.Match) -> str:
|
||
body = m.group(1).strip()
|
||
if "." in body:
|
||
node_id, rest = body.split(".", 1)
|
||
node_val = out_map.get(node_id)
|
||
val = _get_by_path(node_val, rest)
|
||
else:
|
||
val = out_map.get(body)
|
||
return _stringify_for_template(val)
|
||
|
||
def repl_store(m: re.Match) -> str:
|
||
path = m.group(1).strip()
|
||
store = context.get("store") or {}
|
||
val = _get_by_path(store, path)
|
||
return _stringify_for_template(val)
|
||
|
||
s = _VAR_MACRO_RE.sub(repl_var, s)
|
||
s = _OUT_MACRO_RE.sub(repl_out, s)
|
||
s = _STORE_MACRO_RE.sub(repl_store, s)
|
||
|
||
# [[OUT1]] → текст из ноды n1 (best-effort)
|
||
def repl_out_short(m: re.Match) -> str:
|
||
try:
|
||
num = int(m.group(1))
|
||
node_id = f"n{num}"
|
||
node_out = out_map.get(node_id)
|
||
txt = _best_text_from_outputs(node_out)
|
||
return _stringify_for_template(txt)
|
||
except Exception:
|
||
return ""
|
||
s = _OUT_SHORT_RE.sub(repl_out_short, s)
|
||
|
||
# [[PROMPT]] — провайдеро-специфичный JSON-фрагмент, подготовленный в context["PROMPT"]
|
||
s = _PROMPT_MACRO_RE.sub(lambda _m: str(context.get("PROMPT") or ""), s)
|
||
|
||
# 1.5) Голые [[NAME]] / [[path.to.value]]
|
||
def repl_bare(m: re.Match) -> str:
|
||
name = m.group(1).strip()
|
||
# Зарезервированные формы уже обработаны выше; бережно пропускаем похожие
|
||
if name.upper() in {"OUT", "VAR", "PROMPT"} or re.fullmatch(r"OUT\d+", name.upper() or ""):
|
||
return m.group(0)
|
||
# Сначала пользовательские переменные
|
||
vmap = context.get("vars") or {}
|
||
if isinstance(vmap, dict) and name in vmap:
|
||
return _stringify_for_template(vmap.get(name))
|
||
# Затем путь из общего контекста
|
||
val = _get_by_path(context, name)
|
||
return _stringify_for_template(val)
|
||
s = _BARE_MACRO_RE.sub(repl_bare, s)
|
||
|
||
# 2) Подстановки {{ ... }} (+ simple default filter)
|
||
def repl_braces(m: re.Match) -> str:
|
||
expr = m.group(1).strip()
|
||
|
||
def eval_path(p: str) -> Any:
|
||
p = p.strip()
|
||
# Приоритет пользовательских переменных для простых идентификаторов {{ NAME }}
|
||
vmap = context.get("vars") or {}
|
||
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", p) and isinstance(vmap, dict) and p in vmap:
|
||
return vmap.get(p)
|
||
if p.startswith("OUT."):
|
||
body = p[4:]
|
||
if "." in body:
|
||
node_id, rest = body.split(".", 1)
|
||
node_val = out_map.get(node_id)
|
||
return _get_by_path(node_val, rest)
|
||
return out_map.get(body)
|
||
# STORE.* — из постоянного хранилища
|
||
if p.startswith("STORE.") or p.startswith("store."):
|
||
body = p.split(".", 1)[1] if "." in p else ""
|
||
store = context.get("store") or {}
|
||
return _get_by_path(store, body)
|
||
return _get_by_path(context, p)
|
||
|
||
default_match = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", expr)
|
||
if default_match:
|
||
base_path = default_match.group(1).strip()
|
||
fallback_raw = default_match.group(2).strip()
|
||
|
||
# Рекурсивная обработка вложенных default(...) и путей
|
||
def eval_default(raw: str) -> Any:
|
||
raw = raw.strip()
|
||
# Вложенный default: a|default(b)
|
||
dm = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", raw)
|
||
if dm:
|
||
base2 = dm.group(1).strip()
|
||
fb2 = dm.group(2).strip()
|
||
v2 = eval_path(base2)
|
||
if v2 not in (None, ""):
|
||
return v2
|
||
return eval_default(fb2)
|
||
# Пробуем как путь
|
||
v = eval_path(raw)
|
||
if v not in (None, ""):
|
||
return v
|
||
# Явная строка в кавычках
|
||
if len(raw) >= 2 and ((raw[0] == '"' and raw[-1] == '"') or (raw[0] == "'" and raw[-1] == "'")):
|
||
return raw[1:-1]
|
||
# Пробуем распарсить как JSON литерал (число/объект/массив/true/false/null)
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
# Последний вариант: вернуть сырой текст. Для строк рекомендуется default('...') с кавычками.
|
||
return raw
|
||
|
||
raw_val = eval_path(base_path)
|
||
val = raw_val if raw_val not in (None, "") else eval_default(fallback_raw)
|
||
else:
|
||
val = eval_path(expr)
|
||
|
||
return _stringify_for_template(val)
|
||
|
||
s = _BRACES_RE.sub(repl_braces, s)
|
||
return s
|
||
|
||
|
||
# --- Boolean condition evaluator for If-node ---------------------------------
|
||
# Поддерживает:
|
||
# - Операторы: &&, ||, !, ==, !=, <, <=, >, >=, contains
|
||
# - Скобки (...)
|
||
# - Токены-литералы: числа (int/float), строки "..." (без escape-сложностей)
|
||
# - Макросы: [[VAR:...]], [[OUT:...]], [[OUT1]], [[NAME]] (vars/context),
|
||
# {{ path }} и {{ path|default(...) }} — типобезопасно (числа остаются числами)
|
||
# Возвращает bool. Бросает ValueError при синтаксической/семантической ошибке.
|
||
def eval_condition_expr(expr: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> bool:
|
||
import ast
|
||
|
||
if expr is None:
|
||
return False
|
||
s = str(expr)
|
||
|
||
# Tokenize into a flat list of tokens and build value bindings for macros/braces.
|
||
tokens, bindings = _tokenize_condition_expr(s, context, out_map)
|
||
|
||
# Transform infix "contains" into function form contains(a,b)
|
||
tokens = _transform_contains(tokens)
|
||
|
||
# Join into python-like boolean expression and map logical ops.
|
||
py_expr = _tokens_to_python_expr(tokens)
|
||
|
||
# Evaluate safely via AST with strict whitelist
|
||
result = _safe_eval_bool(py_expr, bindings)
|
||
return bool(result)
|
||
|
||
|
||
def _tokens_to_python_expr(tokens: List[str]) -> str:
|
||
# Уже нормализовано на этапе токенизации, просто склеиваем с пробелами
|
||
return " ".join(tokens)
|
||
|
||
|
||
def _transform_contains(tokens: List[str]) -> List[str]:
|
||
# Заменяет "... A contains B ..." на "contains(A, B)" с учётом скобок.
|
||
i = 0
|
||
out: List[str] = tokens[:] # копия
|
||
# Итерируем, пока встречается 'contains'
|
||
while True:
|
||
try:
|
||
idx = out.index("contains")
|
||
except ValueError:
|
||
break
|
||
|
||
# Левая часть
|
||
lstart = idx - 1
|
||
if lstart >= 0 and out[lstart] == ")":
|
||
# найти соответствующую открывающую "("
|
||
bal = 0
|
||
j = lstart
|
||
while j >= 0:
|
||
if out[j] == ")":
|
||
bal += 1
|
||
elif out[j] == "(":
|
||
bal -= 1
|
||
if bal == 0:
|
||
lstart = j
|
||
break
|
||
j -= 1
|
||
if bal != 0:
|
||
# несбалансированные скобки
|
||
raise ValueError("Unbalanced parentheses around left operand of contains")
|
||
# Правая часть
|
||
rend = idx + 1
|
||
if rend < len(out) and out[rend] == "(":
|
||
bal = 0
|
||
j = rend
|
||
while j < len(out):
|
||
if out[j] == "(":
|
||
bal += 1
|
||
elif out[j] == ")":
|
||
bal -= 1
|
||
if bal == 0:
|
||
rend = j
|
||
break
|
||
j += 1
|
||
if bal != 0:
|
||
raise ValueError("Unbalanced parentheses around right operand of contains")
|
||
# Если нет скобок — однотокенный операнд
|
||
left_tokens = out[lstart:idx]
|
||
right_tokens = out[idx + 1:rend + 1] if (idx + 1 < len(out) and out[idx + 1] == "(") else out[idx + 1:idx + 2]
|
||
if not left_tokens or not right_tokens:
|
||
raise ValueError("contains requires two operands")
|
||
|
||
left_str = " ".join(left_tokens)
|
||
right_str = " ".join(right_tokens)
|
||
|
||
# Синтезируем вызов и заменяем диапазон
|
||
new_tok = f"contains({left_str}, {right_str})"
|
||
out = out[:lstart] + [new_tok] + out[(rend + 1) if (idx + 1 < len(out) and out[idx + 1] == "(") else (idx + 2):]
|
||
return out
|
||
|
||
|
||
def _tokenize_condition_expr(expr: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> tuple[List[str], Dict[str, Any]]:
|
||
tokens: List[str] = []
|
||
bindings: Dict[str, Any] = {}
|
||
i = 0
|
||
n = len(expr)
|
||
vcount = 0
|
||
|
||
def add_binding(val: Any) -> str:
|
||
nonlocal vcount
|
||
name = f"__v{vcount}"
|
||
vcount += 1
|
||
bindings[name] = val
|
||
return name
|
||
|
||
while i < n:
|
||
ch = expr[i]
|
||
# Пробелы
|
||
if ch.isspace():
|
||
i += 1
|
||
continue
|
||
|
||
# Операторы двойные
|
||
if expr.startswith("&&", i):
|
||
tokens.append("and")
|
||
i += 2
|
||
continue
|
||
if expr.startswith("||", i):
|
||
tokens.append("or")
|
||
i += 2
|
||
continue
|
||
if expr.startswith(">=", i) or expr.startswith("<=", i) or expr.startswith("==", i) or expr.startswith("!=", i):
|
||
tokens.append(expr[i:i+2])
|
||
i += 2
|
||
continue
|
||
|
||
# Одинарные операторы
|
||
if ch in "()<>":
|
||
tokens.append(ch)
|
||
i += 1
|
||
continue
|
||
if ch == "!":
|
||
# уже обработали "!=" как двойной
|
||
tokens.append("not")
|
||
i += 1
|
||
continue
|
||
|
||
# Строковые литералы "...." и '....' (простая версия: без экранирования)
|
||
if ch == '"':
|
||
j = i + 1
|
||
while j < n and expr[j] != '"':
|
||
# простая версия: без экранирования
|
||
j += 1
|
||
if j >= n:
|
||
raise ValueError('Unterminated string literal')
|
||
content = expr[i+1:j]
|
||
# Конвертируем в безопасный Python-литерал
|
||
tokens.append(repr(content))
|
||
i = j + 1
|
||
continue
|
||
# Одинарные кавычки
|
||
if ch == "'":
|
||
j = i + 1
|
||
while j < n and expr[j] != "'":
|
||
# простая версия: без экранирования
|
||
j += 1
|
||
if j >= n:
|
||
raise ValueError('Unterminated string literal')
|
||
content = expr[i+1:j]
|
||
tokens.append(repr(content))
|
||
i = j + 1
|
||
continue
|
||
|
||
# Макросы [[...]]
|
||
if expr.startswith("[[", i):
|
||
j = expr.find("]]", i + 2)
|
||
if j < 0:
|
||
raise ValueError("Unterminated [[...]] macro")
|
||
body = expr[i+2:j]
|
||
val = _resolve_square_macro_value(body, context, out_map)
|
||
name = add_binding(val)
|
||
tokens.append(name)
|
||
i = j + 2
|
||
continue
|
||
|
||
# Скобки {{ ... }}
|
||
if expr.startswith("{{", i):
|
||
j = expr.find("}}", i + 2)
|
||
if j < 0:
|
||
raise ValueError("Unterminated {{ ... }} expression")
|
||
body = expr[i+2:j]
|
||
val = _resolve_braces_value(body, context, out_map)
|
||
name = add_binding(val)
|
||
tokens.append(name)
|
||
i = j + 2
|
||
continue
|
||
|
||
# Ключевое слово contains
|
||
if expr[i:i+8].lower() == "contains":
|
||
tokens.append("contains")
|
||
i += 8
|
||
continue
|
||
|
||
# Число
|
||
if ch.isdigit():
|
||
j = i + 1
|
||
dot_seen = False
|
||
while j < n and (expr[j].isdigit() or (expr[j] == "." and not dot_seen)):
|
||
if expr[j] == ".":
|
||
dot_seen = True
|
||
j += 1
|
||
tokens.append(expr[i:j])
|
||
i = j
|
||
continue
|
||
|
||
# Идентификатор (на всякий — пропускаем последовательность букв/подчёрк/цифр)
|
||
if ch.isalpha() or ch == "_":
|
||
j = i + 1
|
||
while j < n and (expr[j].isalnum() or expr[j] in "._"):
|
||
j += 1
|
||
word = expr[i:j]
|
||
lw = word.lower()
|
||
# Литералы: true/false/null (любая раскладка) → Python-константы
|
||
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", word) and lw in {"true", "false", "null"}:
|
||
tokens.append("True" if lw == "true" else ("False" if lw == "false" else "None"))
|
||
i = j
|
||
continue
|
||
# Поддержка «голых» идентификаторов из vars: cycleindex, WAS_ERROR и т.п.
|
||
# Если это простой идентификатор (без точек) и он есть в context.vars — биндим его значением.
|
||
try:
|
||
vmap = context.get("vars") or {}
|
||
except Exception:
|
||
vmap = {}
|
||
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", word) and isinstance(vmap, dict) and word in vmap:
|
||
name = add_binding(vmap.get(word))
|
||
tokens.append(name)
|
||
else:
|
||
# Логические в словах не поддерживаем (используйте &&, ||, !)
|
||
tokens.append(word)
|
||
i = j
|
||
continue
|
||
|
||
# Иное — ошибка
|
||
raise ValueError(f"Unexpected character in expression: {ch!r}")
|
||
|
||
return tokens, bindings
|
||
|
||
|
||
def _resolve_square_macro_value(body: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> Any:
|
||
# Тело без [[...]]
|
||
b = str(body or "").strip()
|
||
# [[OUT1]]
|
||
m = re.fullmatch(r"(?is)OUT\s*(\d+)", b)
|
||
if m:
|
||
try:
|
||
num = int(m.group(1))
|
||
node_id = f"n{num}"
|
||
node_out = out_map.get(node_id)
|
||
return _best_text_from_outputs(node_out)
|
||
except Exception:
|
||
return ""
|
||
|
||
# [[VAR: ...]]
|
||
m = re.fullmatch(r"(?is)VAR\s*[:]\s*(.+)", b)
|
||
if m:
|
||
path = m.group(1).strip()
|
||
return _get_by_path(context, path)
|
||
|
||
# [[OUT: node.path]]
|
||
m = re.fullmatch(r"(?is)OUT\s*[:]\s*(.+)", b)
|
||
if m:
|
||
body2 = m.group(1).strip()
|
||
if "." in body2:
|
||
node_id, rest = body2.split(".", 1)
|
||
node_val = out_map.get(node_id.strip())
|
||
return _get_by_path(node_val, rest.strip())
|
||
return out_map.get(body2)
|
||
|
||
# [[STORE: path]]
|
||
m = re.fullmatch(r"(?is)STORE\s*[:]\s*(.+)", b)
|
||
if m:
|
||
path = m.group(1).strip()
|
||
store = context.get("store") or {}
|
||
return _get_by_path(store, path)
|
||
|
||
# [[NAME]] — «голая» переменная: сначала vars, потом context по пути/ключу
|
||
name = b
|
||
vmap = context.get("vars") or {}
|
||
if isinstance(vmap, dict) and (name in vmap):
|
||
return vmap.get(name)
|
||
return _get_by_path(context, name)
|
||
|
||
|
||
def _resolve_braces_value(body: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> Any:
|
||
# Логика совместима с {{ path|default(value) }}, возврат — типобезопасный
|
||
expr = str(body or "").strip()
|
||
|
||
def eval_path(p: str) -> Any:
|
||
p = p.strip()
|
||
vmap = context.get("vars") or {}
|
||
# Простой идентификатор — сначала в vars
|
||
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", p) and isinstance(vmap, dict) and p in vmap:
|
||
return vmap.get(p)
|
||
if p.startswith("OUT."):
|
||
body2 = p[4:].strip()
|
||
if "." in body2:
|
||
node_id, rest = body2.split(".", 1)
|
||
node_val = out_map.get(node_id.strip())
|
||
return _get_by_path(node_val, rest.strip())
|
||
return out_map.get(body2)
|
||
return _get_by_path(context, p)
|
||
|
||
m = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", expr)
|
||
if m:
|
||
base_path = m.group(1).strip()
|
||
fallback_raw = m.group(2).strip()
|
||
|
||
def eval_default(raw: str) -> Any:
|
||
raw = raw.strip()
|
||
dm = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", raw)
|
||
if dm:
|
||
base2 = dm.group(1).strip()
|
||
fb2 = dm.group(2).strip()
|
||
v2 = eval_path(base2)
|
||
if v2 not in (None, ""):
|
||
return v2
|
||
return eval_default(fb2)
|
||
# Пробуем как путь
|
||
v = eval_path(raw)
|
||
if v not in (None, ""):
|
||
return v
|
||
# Строка в кавычках
|
||
if len(raw) >= 2 and ((raw[0] == '"' and raw[-1] == '"') or (raw[0] == "'" and raw[-1] == "'")):
|
||
return raw[1:-1]
|
||
# JSON литерал
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
return raw
|
||
|
||
raw_val = eval_path(base_path)
|
||
return raw_val if raw_val not in (None, "") else eval_default(fallback_raw)
|
||
else:
|
||
return eval_path(expr)
|
||
|
||
|
||
def _stringify_for_contains(val: Any) -> str:
|
||
# Для contains на строках — совместим со строкификацией шаблона
|
||
return _stringify_for_template(val)
|
||
|
||
|
||
def _safe_eval_bool(py_expr: str, bindings: Dict[str, Any]) -> bool:
|
||
import ast
|
||
import operator as op
|
||
|
||
def contains_fn(a: Any, b: Any) -> bool:
|
||
# Семантика: список/кортеж/множество — membership, иначе — подстрока по строковому представлению
|
||
if isinstance(a, (list, tuple, set)):
|
||
return b in a
|
||
sa = _stringify_for_contains(a)
|
||
sb = _stringify_for_contains(b)
|
||
return sb in sa
|
||
|
||
allowed_boolops = {ast.And, ast.Or}
|
||
allowed_unary = {ast.Not}
|
||
allowed_cmp = {ast.Eq: op.eq, ast.NotEq: op.ne, ast.Lt: op.lt, ast.LtE: op.le, ast.Gt: op.gt, ast.GtE: op.ge}
|
||
|
||
def eval_node(node: ast.AST) -> Any:
|
||
if isinstance(node, ast.Expression):
|
||
return eval_node(node.body)
|
||
if isinstance(node, ast.Constant):
|
||
return node.value
|
||
if isinstance(node, ast.Name):
|
||
if node.id == "contains":
|
||
# Возврат специальной метки, реально обрабатывается в Call
|
||
return ("__fn__", "contains")
|
||
if node.id in bindings:
|
||
return bindings[node.id]
|
||
# Неизвестные имена запрещены
|
||
raise ValueError(f"Unknown name: {node.id}")
|
||
if isinstance(node, ast.UnaryOp) and isinstance(node.op, tuple(allowed_unary)):
|
||
val = bool(eval_node(node.operand))
|
||
if isinstance(node.op, ast.Not):
|
||
return (not val)
|
||
if isinstance(node, ast.BoolOp) and isinstance(node.op, tuple(allowed_boolops)):
|
||
# Короткое замыкание:
|
||
# AND — при первом False прекращаем и возвращаем False; иначе True
|
||
# OR — при первом True прекращаем и возвращаем True; иначе False
|
||
if isinstance(node.op, ast.And):
|
||
for v in node.values:
|
||
if not bool(eval_node(v)):
|
||
return False
|
||
return True
|
||
if isinstance(node.op, ast.Or):
|
||
for v in node.values:
|
||
if bool(eval_node(v)):
|
||
return True
|
||
return False
|
||
if isinstance(node, ast.Compare):
|
||
left = eval_node(node.left)
|
||
for opnode, comparator in zip(node.ops, node.comparators):
|
||
if type(opnode) not in allowed_cmp:
|
||
raise ValueError("Unsupported comparison operator")
|
||
right = eval_node(comparator)
|
||
if not allowed_cmp[type(opnode)](left, right):
|
||
return False
|
||
left = right
|
||
return True
|
||
if isinstance(node, ast.Call):
|
||
# Разрешаем только contains(a,b)
|
||
if node.keywords or len(node.args) != 2:
|
||
raise ValueError("Only contains(a,b) call is allowed")
|
||
fn = node.func
|
||
# Форма contains(...) может прийти как Name('contains') или как ("__fn__","contains")
|
||
if isinstance(fn, ast.Name) and fn.id == "contains":
|
||
a = eval_node(node.args[0])
|
||
b = eval_node(node.args[1])
|
||
return contains_fn(a, b)
|
||
# Дополнительно: если парсер вернул константу-маркер
|
||
if isinstance(fn, ast.Constant) and fn.value == ("__fn__", "contains"):
|
||
a = eval_node(node.args[0])
|
||
b = eval_node(node.args[1])
|
||
return contains_fn(a, b)
|
||
raise ValueError("Function calls are not allowed")
|
||
# Запрещаем имена, атрибуты, индексации и прочее
|
||
raise ValueError("Expression construct not allowed")
|
||
|
||
try:
|
||
tree = ast.parse(py_expr, mode="eval")
|
||
except Exception as exc:
|
||
raise ValueError(f"Condition parse error: {exc}") from exc
|
||
return bool(eval_node(tree)) |