Add runtime variables and data-driven control flow

Pre-execution interpolate.py only resolved ${var} placeholders once against
a static mapping; scripts had no way to mutate state during execution.
VariableScope is a runtime mapping the executor exposes to flow-control
commands so AC_set_var / AC_inc_var / AC_get_var, AC_if_var (with
eq/ne/lt/le/gt/ge/contains/startswith/endswith), and AC_for_each can read
and write the same bag the runtime interpolator consults.

The executor now resolves ${var} per command call (not pre-flattened), so
nested body/then/else lists keep their placeholders and re-bind each time
they execute — letting AC_for_each iterate over a list while the body sees
the current item.

Author: JE-Chen

je_auto_control/__init__.py

@@ -104,6 +104,7 @@
 from je_auto_control.utils.script_vars.interpolate import (
     interpolate_actions, interpolate_value, load_vars_from_json,
 )
+from je_auto_control.utils.script_vars.scope import VariableScope
 # Watchers (headless)
 from je_auto_control.utils.watcher.watcher import (
     LogTail, MouseWatcher, PixelWatcher,
@@ -213,6 +214,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "Scheduler", "ScheduledJob", "default_scheduler",
     # Script variables
     "interpolate_actions", "interpolate_value", "load_vars_from_json",
+    "VariableScope",
     # Watchers
     "MouseWatcher", "PixelWatcher", "LogTail",
     # Window manager

je_auto_control/utils/executor/action_executor.py

@@ -31,7 +31,10 @@
     wait_for_text as ocr_wait_for_text,
 )
 from je_auto_control.utils.run_history.history_store import default_history_store
-from je_auto_control.utils.script_vars.interpolate import interpolate_actions
+from je_auto_control.utils.script_vars.interpolate import (
+    interpolate_actions, interpolate_value,
+)
+from je_auto_control.utils.script_vars.scope import VariableScope
 from je_auto_control.utils.generate_report.generate_html_report import generate_html, generate_html_report
 from je_auto_control.utils.generate_report.generate_json_report import generate_json, generate_json_report
 from je_auto_control.utils.generate_report.generate_xml_report import generate_xml, generate_xml_report
@@ -157,8 +160,13 @@ class Executor:
     - 支援流程控制指令 (AC_loop, AC_if_image_found 等)
     """
 
+    # Args keys that hold nested action lists; runtime interpolation must
+    # leave them untouched so each iteration re-reads current variable state.
+    _DEFERRED_ARG_KEYS: frozenset = frozenset({"body", "then", "else"})
+
     def __init__(self):
         self._block_commands = BLOCK_COMMANDS
+        self.variables = VariableScope()
         # 事件字典，對應字串名稱到函式
         self.event_dict: dict = {
             # Mouse 滑鼠相關
@@ -258,6 +266,27 @@ def known_commands(self) -> set:
         """Return the set of all command names the executor recognises."""
         return set(self.event_dict.keys()) | set(self._block_commands.keys())
 
+    def _resolve_runtime_args(self, args: Any) -> Any:
+        """Interpolate ``${var}`` placeholders against the current scope.
+
+        Keys inside :attr:`_DEFERRED_ARG_KEYS` (``body``/``then``/``else``)
+        are left as-is so nested action lists keep their placeholders for
+        per-iteration evaluation.
+        """
+        if not self.variables:
+            return args
+        if isinstance(args, dict):
+            resolved: Dict[str, Any] = {}
+            for key, value in args.items():
+                if key in self._DEFERRED_ARG_KEYS:
+                    resolved[key] = value
+                else:
+                    resolved[key] = interpolate_value(value, self.variables)
+            return resolved
+        if isinstance(args, list):
+            return [interpolate_value(item, self.variables) for item in args]
+        return args
+
     def _execute_event(self, action: list) -> Any:
         """
         執行單一事件
@@ -271,16 +300,17 @@ def _execute_event(self, action: list) -> Any:
                 raise AutoControlActionException(
                     f"{name} requires a dict of arguments"
                 )
-            return block_handler(self, args)
+            return block_handler(self, self._resolve_runtime_args(args))
 
         event = self.event_dict.get(name)
         if event is None:
             raise AutoControlActionException(f"Unknown action: {name}")
 
         if len(action) == 2:
-            if isinstance(action[1], dict):
-                return event(**action[1])
-            return event(*action[1])
+            resolved = self._resolve_runtime_args(action[1])
+            if isinstance(resolved, dict):
+                return event(**resolved)
+            return event(*resolved)
         if len(action) == 1:
             return event()
         raise AutoControlActionException(cant_execute_action_error_message + " " + str(action))
@@ -393,6 +423,12 @@ def execute_files(execute_files_list: list) -> List[Dict[str, str]]:
 
 def execute_action_with_vars(action_list: list, variables: dict
                              ) -> Dict[str, str]:
-    """Interpolate ``${name}`` placeholders with ``variables`` and execute."""
+    """Interpolate ``${name}`` placeholders with ``variables`` and execute.
+
+    The same mapping seeds the runtime variable scope so flow-control
+    commands (``AC_set_var``/``AC_if_var``/...) can read and mutate the
+    same values during execution.
+    """
     resolved = interpolate_actions(action_list, variables)
+    executor.variables.update_many(variables)
     return executor.execute_action(resolved)

je_auto_control/utils/executor/action_schema.py

@@ -12,9 +12,11 @@
 FLOW_BODY_KEYS = {
     "AC_if_image_found": ("then", "else"),
     "AC_if_pixel": ("then", "else"),
+    "AC_if_var": ("then", "else"),
     "AC_loop": ("body",),
     "AC_while_image": ("body",),
     "AC_retry": ("body",),
+    "AC_for_each": ("body",),
 }
 
 

je_auto_control/utils/executor/flow_control.py

@@ -178,9 +178,95 @@ def exec_continue(executor: Any, args: Mapping[str, Any]) -> None:
     raise LoopContinue()
 
 
+def exec_set_var(executor: Any, args: Mapping[str, Any]) -> Any:
+    """Store ``value`` under ``name`` in the executor's variable scope."""
+    name = args["name"]
+    value = args.get("value")
+    executor.variables.set(name, value)
+    return value
+
+
+def exec_get_var(executor: Any, args: Mapping[str, Any]) -> Any:
+    """Return the variable named ``name`` (or ``default`` if missing)."""
+    return executor.variables.get_value(args["name"], args.get("default"))
+
+
+def exec_inc_var(executor: Any, args: Mapping[str, Any]) -> Any:
+    """Increment a numeric variable by ``by`` (default 1) and return new value."""
+    name = args["name"]
+    delta = args.get("by", 1)
+    current = executor.variables.get_value(name, 0)
+    try:
+        new_value = current + delta
+    except TypeError as error:
+        raise AutoControlActionException(
+            f"AC_inc_var: variable {name!r} is not numeric: {current!r}"
+        ) from error
+    executor.variables.set(name, new_value)
+    return new_value
+
+
+_COMPARATORS: Dict[str, Callable[[Any, Any], bool]] = {
+    "eq": lambda a, b: a == b,
+    "ne": lambda a, b: a != b,
+    "lt": lambda a, b: a < b,
+    "le": lambda a, b: a <= b,
+    "gt": lambda a, b: a > b,
+    "ge": lambda a, b: a >= b,
+    "contains": lambda a, b: b in a,
+    "startswith": lambda a, b: isinstance(a, str) and a.startswith(b),
+    "endswith": lambda a, b: isinstance(a, str) and a.endswith(b),
+}
+
+
+def exec_if_var(executor: Any, args: Mapping[str, Any]) -> Any:
+    """Run ``then`` when ``variable op value`` holds, else run ``else``."""
+    name = args["name"]
+    op = args.get("op", "eq")
+    comparator = _COMPARATORS.get(op)
+    if comparator is None:
+        raise AutoControlActionException(
+            f"AC_if_var: unsupported op {op!r}; "
+            f"expected one of {sorted(_COMPARATORS)}"
+        )
+    current = executor.variables.get_value(name)
+    try:
+        matched = comparator(current, args.get("value"))
+    except TypeError as error:
+        raise AutoControlActionException(
+            f"AC_if_var: cannot compare {current!r} {op} {args.get('value')!r}"
+        ) from error
+    key = "then" if matched else "else"
+    return _run_branch(executor, args.get(key))
+
+
+def exec_for_each(executor: Any, args: Mapping[str, Any]) -> int:
+    """Bind each item in ``items`` to ``as`` and execute ``body``."""
+    items = args["items"]
+    if not isinstance(items, (list, tuple)):
+        raise AutoControlActionException(
+            f"AC_for_each: items must be a list, got {type(items).__name__}"
+        )
+    var_name = args.get("as", "item")
+    body = args.get("body") or []
+    iterations = 0
+    for item in items:
+        executor.variables.set(var_name, item)
+        try:
+            executor.execute_action(body, _validated=True)
+        except LoopContinue:
+            iterations += 1
+            continue
+        except LoopBreak:
+            break
+        iterations += 1
+    return iterations
+
+
 BLOCK_COMMANDS: Dict[str, Callable[[Any, Mapping[str, Any]], Any]] = {
     "AC_if_image_found": exec_if_image_found,
     "AC_if_pixel": exec_if_pixel,
+    "AC_if_var": exec_if_var,
     "AC_wait_image": exec_wait_image,
     "AC_wait_pixel": exec_wait_pixel,
     "AC_sleep": exec_sleep,
@@ -189,4 +275,8 @@ def exec_continue(executor: Any, args: Mapping[str, Any]) -> None:
     "AC_retry": exec_retry,
     "AC_break": exec_break,
     "AC_continue": exec_continue,
+    "AC_set_var": exec_set_var,
+    "AC_get_var": exec_get_var,
+    "AC_inc_var": exec_inc_var,
+    "AC_for_each": exec_for_each,
 }

je_auto_control/utils/script_vars/__init__.py

@@ -2,5 +2,9 @@
 from je_auto_control.utils.script_vars.interpolate import (
     interpolate_actions, interpolate_value, load_vars_from_json,
 )
+from je_auto_control.utils.script_vars.scope import VariableScope
 
-__all__ = ["interpolate_actions", "interpolate_value", "load_vars_from_json"]
+__all__ = [
+    "VariableScope", "interpolate_actions", "interpolate_value",
+    "load_vars_from_json",
+]

je_auto_control/utils/script_vars/scope.py

@@ -0,0 +1,60 @@
+"""Runtime variable scope for the action executor.
+
+Pre-execution interpolation in :mod:`interpolate` replaces ``${var}``
+placeholders once, against a static mapping. Some scripts need to mutate
+state during execution — counters in loops, captured OCR/locator results,
+``for_each`` items. ``VariableScope`` is a thin mutable container the
+executor exposes to flow-control commands so those commands can read and
+write the same bag the runtime interpolator consults.
+"""
+from typing import Any, Dict, Iterator, Mapping, MutableMapping, Optional
+
+
+class VariableScope(MutableMapping[str, Any]):
+    """Mutable mapping of script variables shared across action execution."""
+
+    __slots__ = ("_vars",)
+
+    def __init__(self, initial: Optional[Mapping[str, Any]] = None) -> None:
+        self._vars: Dict[str, Any] = dict(initial) if initial else {}
+
+    def __getitem__(self, key: str) -> Any:
+        return self._vars[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        if not isinstance(key, str) or not key:
+            raise ValueError("variable name must be a non-empty string")
+        self._vars[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self._vars[key]
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._vars)
+
+    def __len__(self) -> int:
+        return len(self._vars)
+
+    def __contains__(self, key: object) -> bool:
+        return key in self._vars
+
+    def set(self, name: str, value: Any) -> None:
+        """Assign ``name`` to ``value``."""
+        self[name] = value
+
+    def get_value(self, name: str, default: Any = None) -> Any:
+        """Return the variable, or ``default`` when missing."""
+        return self._vars.get(name, default)
+
+    def update_many(self, mapping: Mapping[str, Any]) -> None:
+        """Bulk-assign from a mapping."""
+        for key, value in mapping.items():
+            self[key] = value
+
+    def as_dict(self) -> Dict[str, Any]:
+        """Return a shallow copy as a plain dict (safe for interpolation)."""
+        return dict(self._vars)
+
+    def clear(self) -> None:
+        """Drop every stored variable."""
+        self._vars.clear()