"""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)." )