"""Grafmotorn: kör ett nod/kant-flöde genom att följa kanterna.
Startar på `start`-noden (eller den enda utan inkommande kant), följer `out`-
handtaget efter varje vanlig nod. `if`-noder följer `case{i}`/`else`, `loop`-
noder `each` (per varv) / `done` (efter sista varvet); loop-kroppen kopplar
tillbaka till loop-nodens `in`.
Stegprimitiverna (_execute_step, _evaluate_condition, ...) återanvänds från
rpa/flow_engine.py - inget dupliceras.
"""
from __future__ import annotations
from collections import defaultdict
from .flow_engine import (
FlowError,
_DESKTOP_STEPS,
_NO_BROWSER_STEPS,
_evaluate_condition,
_execute_step,
_interpolate,
)
MAX_NODE_STEPS = 100_000 # skydd mot oändliga loop-back-cykler
MAX_CALL_DEPTH = 20 # skydd mot djupa/cirkulära call_flow-anrop
def _adjacency(edges: list) -> dict[str, dict[str, list[str]]]:
out: dict[str, dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
for e in edges:
h = e.get("sourceHandle") or "out"
out[e["source"]][h].append(e["target"])
return out
def _start_node(nodes: list, edges: list, explicit: str | None) -> str | None:
if explicit:
return explicit
targeted = {e["target"] for e in edges}
starts = [n["id"] for n in nodes if n["id"] not in targeted]
if starts:
return starts[0]
return nodes[0]["id"] if nodes else None
def graph_needs_browser(nodes: list, resolve=None, _seen: frozenset = frozenset()) -> bool:
"""Platt variant av flow_engine._flow_needs_browser för nod-grafen. `resolve`
(target -> (nodes, edges, start, inputs)) låter oss titta in i call_flow-mål;
saknas den antar vi konservativt att webbläsaren behövs."""
for n in nodes:
if n.get("enabled") is False:
continue
t = n.get("type")
if t in _DESKTOP_STEPS or t in _NO_BROWSER_STEPS or t == "loop":
continue
if t == "if":
for br in n.get("branches", []):
if br.get("condition", {}).get("kind", "element_exists") in (
"element_exists", "element_not_exists"
):
return True
continue
if t == "call_flow":
target = str(n.get("target", ""))
if resolve is None or target in _seen:
return True
try:
sub_nodes, *_ = resolve(target)
except Exception: # noqa: BLE001 - okänt mål: låt körningen ge felet
return True
if graph_needs_browser(sub_nodes, resolve, _seen | {target}):
return True
continue
return True
return False
def first_browser_node(nodes: list) -> str | None:
for n in nodes:
if n.get("enabled") is False:
continue
t = n.get("type")
if t in _DESKTOP_STEPS or t in _NO_BROWSER_STEPS or t == "loop":
continue
if t == "if":
if any(
br.get("condition", {}).get("kind", "element_exists")
in ("element_exists", "element_not_exists")
for br in n.get("branches", [])
):
return f"if{(' ' + n['label']) if n.get('label') else ''}"
continue
if t == "call_flow":
return f"call_flow → {n.get('target', '?')}"
return t + (f" '{n['label']}'" if n.get("label") else "")
return None
def run_graph(
nodes, edges, log, *, page, download_dir, vars_, desktop_holder, start=None,
resolve_call=None, _call_stack=(),
) -> None:
"""Kör grafen. Undantag från ett steg propagerar om inte nodens on_error =
"continue". Muterar vars_ / desktop_holder.
resolve_call(target) -> (nodes, edges, start, inputs) används av call_flow-noder
för att köra ett annat flöde/en lokal subgraf som ett steg."""
nodes_by_id = {n["id"]: n for n in nodes}
out = _adjacency(edges)
def go(node_id: str, handle: str) -> str | None:
targets = out.get(node_id, {}).get(handle, [])
return targets[0] if targets else None
cursor = _start_node(nodes, edges, start)
loop_state: dict[str, dict] = {}
# [(loop_node_id, locator|None)] - översta = aktuell "item"-scope
item_stack: list[list] = []
steps_run = 0
while cursor is not None:
steps_run += 1
if steps_run > MAX_NODE_STEPS:
raise FlowError("För många steg i körningen - avbryter (troligen en oändlig loop).")
node = nodes_by_id.get(cursor)
if node is None:
raise FlowError(f"En kant pekar på en okänd nod: {cursor!r}")
t = node.get("type")
if node.get("enabled") is False:
log(f"(Inaktiverat, hoppar över: {t})")
cursor = go(cursor, "out")
continue
item_locator = item_stack[-1][1] if item_stack else None
# --- if ---
if t == "if":
chosen = "else"
for i, br in enumerate(node.get("branches", [])):
try:
if _evaluate_condition(page, item_locator, vars_, br.get("condition", {})):
chosen = f"case{i}"
log(f"if: gren {i} matchade")
break
except Exception as exc: # noqa: BLE001
if node.get("on_error") == "continue":
log(f"if-villkoret misslyckades - fortsätter (else): {exc}")
break
raise
else:
log("if: ingen gren matchade -> else")
cursor = go(cursor, chosen)
continue
# --- loop ---
if t == "loop":
st = loop_state.get(cursor)
if st is None:
st = _loop_init(node, vars_)
loop_state[cursor] = st
item_stack.append([cursor, item_locator]) # ärver ev. yttre scope
total = st["total"]
if st["i"] >= total:
loop_state.pop(cursor, None)
if item_stack and item_stack[-1][0] == cursor:
item_stack.pop()
log(f"loop klar ({total} varv)")
cursor = go(cursor, "done")
continue
idx = st["i"]
st["i"] += 1
vars_[st["index_var"]] = idx + 1
if st["kind"] == "dom":
loc = page.locator(st["selector"]).nth(idx)
if item_stack and item_stack[-1][0] == cursor:
item_stack[-1][1] = loc
else:
vars_[st["item_var"]] = st["items"][idx]
log(f"-- Varv {idx + 1}/{total} --")
nxt = go(cursor, "each")
cursor = nxt if nxt is not None else cursor # ingen kropp -> loopa vidare
continue
# --- call_flow: kör ett annat flöde / lokal subgraf som ett steg ---
if t == "call_flow":
try:
_run_call_flow(node, log, page, download_dir, vars_, desktop_holder,
resolve_call, _call_stack)
except Exception as exc: # noqa: BLE001
if node.get("on_error") == "continue":
log(f"call_flow misslyckades - fortsätter (on_error=continue): {exc}")
else:
raise
cursor = go(cursor, "out")
continue
# --- vanligt steg (desktop hanteras internt av _execute_step) ---
try:
_execute_step(node, t, page, log, vars_, item_locator, download_dir, desktop_holder)
except Exception as exc: # noqa: BLE001
if node.get("on_error") == "continue":
log(f"Steget '{t}' misslyckades - fortsätter (on_error=continue): {exc}")
else:
raise
cursor = go(cursor, "out")
log("Flöde klart.")
def _run_call_flow(node, log, page, download_dir, vars_, desktop_holder, resolve_call, call_stack) -> None:
target = str(node.get("target", "")).strip()
if not target:
raise FlowError("call_flow saknar 'target'.")
if resolve_call is None:
raise FlowError(f"call_flow '{target}' kan inte köras här (ingen flödesupplösare).")
if target in call_stack:
raise FlowError(f"Cirkulärt call_flow-anrop: {' -> '.join([*call_stack, target])}.")
if len(call_stack) >= MAX_CALL_DEPTH:
raise FlowError(f"För djupa call_flow-anrop (>{MAX_CALL_DEPTH}) - avbryter.")
sub_nodes, sub_edges, sub_start, sub_inputs = resolve_call(target)
sub_vars: dict = {}
for inp in sub_inputs or []:
name = inp.get("name") if isinstance(inp, dict) else None
if name and inp.get("default") not in (None, ""):
sub_vars[str(name)] = str(inp["default"])
for m in node.get("in", []) or []:
if isinstance(m, dict) and m.get("name"):
sub_vars[str(m["name"])] = _interpolate(str(m.get("value", "")), vars_)
log(f"call_flow -> {target} (in: {', '.join(sorted(sub_vars)) or '-'})")
run_graph(
sub_nodes, sub_edges, log, page=page, download_dir=download_dir,
vars_=sub_vars, desktop_holder=desktop_holder, start=sub_start,
resolve_call=resolve_call, _call_stack=(*call_stack, target),
)
for m in node.get("out", []) or []:
if isinstance(m, dict) and m.get("from") and m.get("to"):
vars_[str(m["to"])] = sub_vars.get(str(m["from"]))
log(f"call_flow <- {target} klart")
def _loop_init(node: dict, vars_: dict) -> dict:
over_name = node.get("over")
over = vars_.get(over_name)
if over is None:
raise FlowError(
f"Loop refererar okänd variabel '{over_name}' (kör 'Lista poster' / "
"'Lägg till i lista' / 'Läs in variabel' innan loopen)."
)
index_var = node.get("index_var") or "index"
if isinstance(over, dict) and "selector" in over and "count" in over:
return {"kind": "dom", "selector": over["selector"], "total": over["count"],
"i": 0, "index_var": index_var}
if isinstance(over, list):
return {"kind": "list", "items": over, "total": len(over), "i": 0,
"index_var": index_var, "item_var": node.get("item_var") or "item"}
raise FlowError(
f"Variabel '{over_name}' kan inte loopas över (varken en lista eller ett "
"'Lista poster'-resultat)."
)