mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-17 23:13:03 +08:00
refactor(autoscript): 使用 ASTokenizer 和 NodeVisitor 重构解析与执行流程
This commit is contained in:
+177
-167
@@ -8,10 +8,29 @@ You may use, modify, and distribute this file under the terms of the MIT License
|
||||
See the LICENSE file for details.
|
||||
"""
|
||||
import re
|
||||
from datetime import datetime, timedelta, date, time
|
||||
from datetime import (
|
||||
datetime,
|
||||
timedelta,
|
||||
date,
|
||||
time
|
||||
)
|
||||
|
||||
from .ASObject import ASObject, _META_VARS, _inferType
|
||||
from .ASObject import (
|
||||
ASObject,
|
||||
_META_VARS,
|
||||
_inferType
|
||||
)
|
||||
from .ASOperator import ASOperator
|
||||
from .ASTokenizer import (
|
||||
ASTokenizer,
|
||||
NodeVisitor,
|
||||
Script,
|
||||
IfNode,
|
||||
SetNode,
|
||||
OpNode,
|
||||
PassNode,
|
||||
UnrecogNode
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["execute", "addTargetVar"]
|
||||
@@ -24,70 +43,30 @@ _TARGET_VARS = {}
|
||||
_SCRIPT_VARS = {}
|
||||
# Name -> ASObject lookup map built from _META_VARS, _TARGET_VARS, and display names
|
||||
_FIELD_MAP = {}
|
||||
# Current line number for error reporting
|
||||
_CUR_LINE = 0
|
||||
|
||||
|
||||
def _errPos(
|
||||
line: int,
|
||||
message: str
|
||||
) -> str:
|
||||
"""
|
||||
Format an error message with the current script line number.
|
||||
Format an error message with a script line number.
|
||||
|
||||
Args:
|
||||
line (int): The script line number where the error occurred.
|
||||
message (str): The error description.
|
||||
|
||||
Returns:
|
||||
str: A formatted error string like "AutoScript syntax error(line X): message".
|
||||
"""
|
||||
return f"AutoScript 语法错误(第{_CUR_LINE}行): {message}"
|
||||
return f"AutoScript 语法错误(第{line}行): {message}"
|
||||
|
||||
|
||||
def _findConditionBegin(
|
||||
upper_line: str
|
||||
) -> int:
|
||||
"""
|
||||
Find the position of the opening parenthesis that starts a condition.
|
||||
|
||||
Args:
|
||||
upper_line (str): The uppercased IF / ELSE IF line.
|
||||
|
||||
Returns:
|
||||
int: Index of '(' or -1 if not found.
|
||||
"""
|
||||
return upper_line.find("(")
|
||||
|
||||
|
||||
def _findConditionEnd(
|
||||
upper_line: str,
|
||||
start_pos: int
|
||||
) -> int:
|
||||
"""
|
||||
Find the matching closing parenthesis for a condition expression.
|
||||
|
||||
Handles nested parentheses and optionally strips a trailing "THEN" keyword.
|
||||
|
||||
Args:
|
||||
upper_line (str): The uppercased IF / ELSE IF line.
|
||||
start_pos (int): Index of the opening '('.
|
||||
|
||||
Returns:
|
||||
int: Index of the matching ')' or -1 if unbalanced.
|
||||
"""
|
||||
|
||||
line = upper_line.rstrip()
|
||||
if line.endswith(" THEN"):
|
||||
line = line[:-5].rstrip()
|
||||
depth = 1
|
||||
for i in range(start_pos + 1, len(line)):
|
||||
ch = line[i]
|
||||
if ch == "(":
|
||||
depth += 1
|
||||
elif ch == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return i
|
||||
return -1
|
||||
# Pre-compiled regex patterns for value resolution
|
||||
_RE_TIME = re.compile(r"^TIME\((\d{1,2}):(\d{2})\)$", re.IGNORECASE)
|
||||
_RE_DATE = re.compile(r"^DATE\((\d{4})-(\d{2})-(\d{2})\)$", re.IGNORECASE)
|
||||
_RE_CUR_DATE_OFFSET = re.compile(r"^CURRENT_DATE\s*\+\s*(\d+)$", re.IGNORECASE)
|
||||
_RE_CUR_TIME_OFFSET = re.compile(r"^CURRENT_TIME\s*\+\s*(\d+)$", re.IGNORECASE)
|
||||
|
||||
|
||||
def _splitTopLevel(
|
||||
@@ -203,10 +182,10 @@ def _resolveValue(
|
||||
"""
|
||||
|
||||
s = value_str.strip()
|
||||
m = re.match(r"^TIME\((\d{1,2}):(\d{2})\)$", s, re.IGNORECASE)
|
||||
m = _RE_TIME.match(s)
|
||||
if m:
|
||||
return time(int(m.group(1)), int(m.group(2)))
|
||||
m = re.match(r"^DATE\((\d{4})-(\d{2})-(\d{2})\)$", s, re.IGNORECASE)
|
||||
m = _RE_DATE.match(s)
|
||||
if m:
|
||||
return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||||
up = s.upper()
|
||||
@@ -218,11 +197,11 @@ def _resolveValue(
|
||||
return s[1:-1].replace("''", "'")
|
||||
if s.startswith('"') and s.endswith('"'):
|
||||
return s[1:-1]
|
||||
m = re.match(r"^CURRENT_DATE\s*\+\s*(\d+)$", s, re.IGNORECASE)
|
||||
m = _RE_CUR_DATE_OFFSET.match(s)
|
||||
if m:
|
||||
days = int(m.group(1))
|
||||
return datetime.now().date() + timedelta(days=days)
|
||||
m = re.match(r"^CURRENT_TIME\s*\+\s*(\d+)$", s, re.IGNORECASE)
|
||||
m = _RE_CUR_TIME_OFFSET.match(s)
|
||||
if m:
|
||||
hours = int(m.group(1))
|
||||
return (datetime.now() + timedelta(hours=hours)).time()
|
||||
@@ -273,7 +252,8 @@ def _resolveAsObject(
|
||||
|
||||
def _evaluateCondition(
|
||||
condition_str: str,
|
||||
target_data: dict
|
||||
target_data: dict,
|
||||
line: int = 0
|
||||
) -> bool:
|
||||
"""
|
||||
Evaluate a condition expression and return a boolean result.
|
||||
@@ -304,18 +284,18 @@ def _evaluateCondition(
|
||||
or_parts = _splitTopLevel(s, ".OR.")
|
||||
if len(or_parts) > 1:
|
||||
return any(
|
||||
_evaluateCondition(p.strip(), target_data)
|
||||
_evaluateCondition(p.strip(), target_data, line)
|
||||
for p in or_parts
|
||||
)
|
||||
and_parts = _splitTopLevel(s, ".AND.")
|
||||
if len(and_parts) > 1:
|
||||
return all(
|
||||
_evaluateCondition(p.strip(), target_data)
|
||||
_evaluateCondition(p.strip(), target_data, line)
|
||||
for p in and_parts
|
||||
)
|
||||
s = s.strip()
|
||||
if s.startswith("(") and s.endswith(")"):
|
||||
return _evaluateCondition(s[1:-1], target_data)
|
||||
return _evaluateCondition(s[1:-1], target_data, line)
|
||||
up = s.upper()
|
||||
if up == ".TRUE.":
|
||||
return True
|
||||
@@ -331,13 +311,14 @@ def _evaluateCondition(
|
||||
right_obj = _resolveAsObject(right_raw, target_data)
|
||||
return ASOperator.compare(left_obj, right_obj, op, target_data)
|
||||
raise ValueError(
|
||||
_errPos(f"无法识别的条件表达式 '{condition_str}'")
|
||||
_errPos(line, f"无法识别的条件表达式 '{condition_str}'")
|
||||
)
|
||||
|
||||
|
||||
def _executeSet(
|
||||
line: str,
|
||||
target_data: dict
|
||||
line_text: str,
|
||||
target_data: dict,
|
||||
line: int = 0
|
||||
):
|
||||
"""
|
||||
Execute a SET statement to assign a value to a field or script variable.
|
||||
@@ -354,7 +335,7 @@ def _executeSet(
|
||||
ValueError: If the value string contains unexpected extra tokens.
|
||||
"""
|
||||
|
||||
rest = line[3:].strip()
|
||||
rest = line_text[3:].strip()
|
||||
eq_idx = rest.find("=")
|
||||
if eq_idx < 0:
|
||||
return
|
||||
@@ -365,7 +346,7 @@ def _executeSet(
|
||||
resolved = _resolveValue(value_str, target_data)
|
||||
stripped = value_str.strip()
|
||||
if resolved == "" and stripped not in ("''", '""') and len(stripped.split()) > 1:
|
||||
raise ValueError(_errPos(f"SET 值中存在多余内容 '{stripped}'"))
|
||||
raise ValueError(_errPos(line, f"SET 值中存在多余内容 '{stripped}'"))
|
||||
upper_name = field_name.upper().strip()
|
||||
obj = _FIELD_MAP.get(upper_name)
|
||||
if not obj:
|
||||
@@ -385,8 +366,9 @@ def _executeSet(
|
||||
|
||||
|
||||
def _executeOperation(
|
||||
line: str,
|
||||
target_data: dict
|
||||
line_text: str,
|
||||
target_data: dict,
|
||||
line: int = 0
|
||||
):
|
||||
"""
|
||||
Execute a field operation statement: "FIELD .ADD. N" or "FIELD .SUB. N".
|
||||
@@ -403,42 +385,23 @@ def _executeOperation(
|
||||
or the type does not support the operation.
|
||||
"""
|
||||
|
||||
parts = line.split()
|
||||
parts = line_text.split()
|
||||
if len(parts) < 3:
|
||||
return
|
||||
if len(parts) > 3:
|
||||
raise ValueError(
|
||||
_errPos(f"操作语句中存在多余内容 '{' '.join(parts[3:])}'")
|
||||
_errPos(line, f"操作语句中存在多余内容 '{' '.join(parts[3:])}'")
|
||||
)
|
||||
field_name = parts[0].upper().strip()
|
||||
op = parts[1].upper().strip()
|
||||
raw_value = parts[2].strip()
|
||||
target = _resolveFieldObj(field_name)
|
||||
if target is None:
|
||||
raise ValueError(_errPos(f"未知字段 '{field_name}'"))
|
||||
raise ValueError(_errPos(line, f"未知字段 '{field_name}'"))
|
||||
operand = _resolveAsObject(raw_value, target_data)
|
||||
ASOperator.apply(target, operand, op, target_data)
|
||||
|
||||
|
||||
def _assertInIf(
|
||||
if_stack: list,
|
||||
line: str
|
||||
):
|
||||
"""
|
||||
Assert that an executable statement is inside an IF block.
|
||||
|
||||
Args:
|
||||
if_stack (list): The current IF nesting stack.
|
||||
line (str): The statement line (used for error message).
|
||||
|
||||
Raises:
|
||||
ValueError: If if_stack is empty (statement is outside any IF block).
|
||||
"""
|
||||
|
||||
if not if_stack:
|
||||
raise ValueError(_errPos(f"可执行语句必须位于 IF 块内: {line}"))
|
||||
|
||||
|
||||
def addTargetVar(
|
||||
name: str,
|
||||
var_type: str,
|
||||
@@ -472,6 +435,125 @@ def addTargetVar(
|
||||
_TARGET_VARS[upper_name] = obj
|
||||
|
||||
|
||||
class _EngineExecutor(NodeVisitor):
|
||||
"""
|
||||
AST visitor that executes AutoScript against target_data.
|
||||
Walks the AST and dispatches SET / ADD / SUB operations
|
||||
via visitScript / visitIf / visitSet / visitOp / visitPass / visitUnrecog.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
target_data: dict
|
||||
):
|
||||
|
||||
super().__init__()
|
||||
self._target_data = target_data
|
||||
self._cur_line = 0
|
||||
|
||||
@property
|
||||
def _line(self) -> int:
|
||||
"""Return current line number for _errPos calls."""
|
||||
|
||||
return self._cur_line
|
||||
|
||||
def _incLine(
|
||||
self
|
||||
):
|
||||
|
||||
self._cur_line += 1
|
||||
|
||||
def visitScript(
|
||||
self,
|
||||
_node: Script
|
||||
):
|
||||
|
||||
for child in _node.body:
|
||||
child.accept(self)
|
||||
|
||||
def visitIf(
|
||||
self,
|
||||
_node: IfNode
|
||||
):
|
||||
|
||||
self._incLine()
|
||||
if not _node.closed:
|
||||
raise ValueError(_errPos(self._line, "IF 与 ENDIF / END IF 不匹配"))
|
||||
matched = _evaluateCondition(_node.condition, self._target_data, self._line)
|
||||
if matched:
|
||||
for child in _node.body:
|
||||
child.accept(self)
|
||||
else:
|
||||
executed = False
|
||||
for elif_node in _node.elif_branches:
|
||||
self._incLine()
|
||||
if _evaluateCondition(elif_node.condition, self._target_data, self._line):
|
||||
for child in elif_node.body:
|
||||
child.accept(self)
|
||||
executed = True
|
||||
break
|
||||
if not executed and _node.else_body:
|
||||
self._incLine()
|
||||
for child in _node.else_body:
|
||||
child.accept(self)
|
||||
|
||||
def visitSet(
|
||||
self,
|
||||
_node: SetNode
|
||||
):
|
||||
|
||||
self._incLine()
|
||||
full_line = f"SET {_node.target} = {_node.value}"
|
||||
_executeSet(full_line, self._target_data, self._line)
|
||||
|
||||
def visitOp(
|
||||
self,
|
||||
_node: OpNode
|
||||
):
|
||||
|
||||
self._incLine()
|
||||
op_upper = _node.op_type.upper()
|
||||
full_line = f"{_node.target} .{op_upper}. {_node.value}"
|
||||
_executeOperation(full_line, self._target_data, self._line)
|
||||
|
||||
def visitPass(
|
||||
self,
|
||||
_node: PassNode
|
||||
):
|
||||
|
||||
self._incLine()
|
||||
|
||||
def visitUnrecog(
|
||||
self,
|
||||
_node: UnrecogNode
|
||||
):
|
||||
|
||||
self._incLine()
|
||||
upper = _node.raw_line.upper().strip()
|
||||
if upper.startswith("IF"):
|
||||
paren_open = upper.find("(")
|
||||
if paren_open < 0:
|
||||
raise ValueError(_errPos(self._line, "IF 缺少左括号"))
|
||||
depth = 1
|
||||
for ci in range(paren_open + 1, len(upper)):
|
||||
if upper[ci] == "(":
|
||||
depth += 1
|
||||
elif upper[ci] == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
remaining = upper[ci + 1:].strip()
|
||||
if remaining and remaining != "THEN":
|
||||
raise ValueError(_errPos(self._line, f"IF 条件后存在多余内容 '{remaining}'"))
|
||||
break
|
||||
if depth > 0:
|
||||
raise ValueError(_errPos(self._line, "IF 缺少右括号"))
|
||||
elif upper.startswith("ELSE IF"):
|
||||
paren_open = upper.find("(")
|
||||
if paren_open < 0:
|
||||
raise ValueError(_errPos(self._line, "ELSE IF 缺少左括号"))
|
||||
_executeOperation(_node.raw_line, self._target_data, self._line)
|
||||
|
||||
|
||||
def execute(
|
||||
script_text: str,
|
||||
target_data: dict
|
||||
@@ -479,9 +561,9 @@ def execute(
|
||||
"""
|
||||
Execute an AutoScript on the given target data.
|
||||
|
||||
Parses the script line by line, maintaining an IF nesting stack to control
|
||||
which blocks are active. Supports IF / ELSE IF / ELSE / END IF control flow,
|
||||
SET assignments, PASS no-ops, and .ADD. / .SUB. field operations.
|
||||
Parses the script into an AST via ASTokenizer.parse(),
|
||||
then walks the tree with a visitor to evaluate conditions
|
||||
and dispatch SET / ADD / SUB operations.
|
||||
|
||||
Args:
|
||||
script_text (str): The AutoScript source code.
|
||||
@@ -489,85 +571,13 @@ def execute(
|
||||
|
||||
Raises:
|
||||
ValueError: On syntax errors, unbalanced IF/END IF, unknown fields, etc.
|
||||
|
||||
Example:
|
||||
>>> data = {"reserve_info": {"date": "2026-05-01"}}
|
||||
>>> execute(
|
||||
... "IF(.TRUE.)\\n"
|
||||
... " RESERVE_DATE .ADD. 1\\n"
|
||||
... "END IF",
|
||||
... data
|
||||
... )
|
||||
>>> data["reserve_info"]["date"]
|
||||
'2026-05-02'
|
||||
"""
|
||||
|
||||
global _CUR_LINE
|
||||
|
||||
_buildFieldMap()
|
||||
if not script_text or not script_text.strip():
|
||||
return
|
||||
lines = [l.strip() for l in script_text.split("\n") if l.strip()]
|
||||
if not lines:
|
||||
ast = ASTokenizer.parse(script_text)
|
||||
if not ast.body:
|
||||
return
|
||||
if_stack = []
|
||||
for _CUR_LINE, line in enumerate(lines, 1):
|
||||
upper_line = line.upper().strip()
|
||||
if upper_line.startswith("IF"):
|
||||
paren_open = _findConditionBegin(upper_line)
|
||||
if paren_open < 0:
|
||||
raise ValueError(_errPos("IF 缺少左括号"))
|
||||
cond_end = _findConditionEnd(upper_line, paren_open)
|
||||
if cond_end < 0:
|
||||
raise ValueError(_errPos("IF 缺少右括号"))
|
||||
remaining = upper_line[cond_end + 1:].strip()
|
||||
if remaining and remaining.upper() != "THEN":
|
||||
raise ValueError(_errPos(f"IF 条件后存在多余内容 '{remaining}'"))
|
||||
condition_str = line[paren_open + 1:cond_end].strip()
|
||||
matched = _evaluateCondition(condition_str, target_data)
|
||||
if_stack.append([matched, matched])
|
||||
elif upper_line.startswith("ELSE IF"):
|
||||
if not if_stack:
|
||||
raise ValueError(_errPos("ELSE IF 前缺少 IF"))
|
||||
paren_open = _findConditionBegin(upper_line)
|
||||
if paren_open < 0:
|
||||
raise ValueError(_errPos("ELSE IF 缺少左括号"))
|
||||
cond_end = _findConditionEnd(upper_line, paren_open)
|
||||
if cond_end < 0:
|
||||
raise ValueError(_errPos("ELSE IF 缺少右括号"))
|
||||
remaining = upper_line[cond_end + 1:].strip()
|
||||
if remaining and remaining.upper() != "THEN":
|
||||
raise ValueError(_errPos(f"ELSE IF 条件后存在多余内容 '{remaining}'"))
|
||||
_, branch_matched = if_stack[-1]
|
||||
if not branch_matched:
|
||||
condition_str = line[paren_open + 1:cond_end].strip()
|
||||
matched = _evaluateCondition(condition_str, target_data)
|
||||
if_stack[-1] = [matched, matched]
|
||||
else:
|
||||
if_stack[-1][0] = False
|
||||
elif upper_line == "ELSE":
|
||||
if not if_stack:
|
||||
raise ValueError(_errPos("ELSE 前缺少 IF"))
|
||||
_, branch_matched = if_stack[-1]
|
||||
if not branch_matched:
|
||||
if_stack[-1] = [True, True]
|
||||
else:
|
||||
if_stack[-1][0] = False
|
||||
elif upper_line in ("ENDIF", "END IF"):
|
||||
if not if_stack:
|
||||
raise ValueError(_errPos("ENDIF / END IF 前缺少 IF"))
|
||||
if_stack.pop()
|
||||
elif upper_line.startswith("SET "):
|
||||
_assertInIf(if_stack, line)
|
||||
if all(ctx[0] for ctx in if_stack):
|
||||
_executeSet(line, target_data)
|
||||
elif upper_line == "PASS":
|
||||
continue
|
||||
else:
|
||||
_assertInIf(if_stack, line)
|
||||
if all(ctx[0] for ctx in if_stack):
|
||||
_executeOperation(line, target_data)
|
||||
if if_stack:
|
||||
raise ValueError(
|
||||
"AutoScript 语法错误: IF 与 ENDIF / END IF 不匹配"
|
||||
)
|
||||
executor = _EngineExecutor(target_data)
|
||||
ast.accept(executor)
|
||||
|
||||
@@ -8,10 +8,18 @@ You may use, modify, and distribute this file under the terms of the MIT License
|
||||
See the LICENSE file for details.
|
||||
"""
|
||||
import re
|
||||
from datetime import datetime, date, time
|
||||
from datetime import (
|
||||
datetime,
|
||||
date,
|
||||
time
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["ASObject", "_META_VARS", "_inferType"]
|
||||
__all__ = [
|
||||
"ASObject",
|
||||
"_META_VARS",
|
||||
"_inferType"
|
||||
]
|
||||
|
||||
|
||||
# Default values for each supported type when no value is present
|
||||
@@ -21,7 +29,7 @@ _TYPE_DEFAULTS = {
|
||||
"Boolean": False,
|
||||
"Date": None,
|
||||
"Time": None,
|
||||
"String": "",
|
||||
"String": ""
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,12 @@ This software is provided "as is", without any warranty of any kind.
|
||||
You may use, modify, and distribute this file under the terms of the MIT License.
|
||||
See the LICENSE file for details.
|
||||
"""
|
||||
from datetime import datetime, timedelta, date, time
|
||||
from datetime import (
|
||||
datetime,
|
||||
timedelta,
|
||||
date,
|
||||
time
|
||||
)
|
||||
|
||||
from .ASObject import ASObject
|
||||
|
||||
@@ -51,8 +56,9 @@ class ASOperator:
|
||||
}
|
||||
_ARITH_TYPES = {"Date", "Time", "Int", "Float"}
|
||||
|
||||
@staticmethod
|
||||
@classmethod
|
||||
def apply(
|
||||
cls,
|
||||
target: ASObject,
|
||||
operand: ASObject,
|
||||
op: str,
|
||||
@@ -73,7 +79,7 @@ class ASOperator:
|
||||
|
||||
tp = target.var_type
|
||||
op_tp = operand.var_type
|
||||
if tp not in ASOperator._ARITH_TYPES:
|
||||
if tp not in cls._ARITH_TYPES:
|
||||
raise ValueError(f"'{tp}' 类型字段不支持操作运算")
|
||||
if op_tp not in ("Int", "Float"):
|
||||
raise ValueError(f"操作数类型 '{op_tp}' 不能用于运算,需要数值类型 (Int / Float)")
|
||||
@@ -83,73 +89,70 @@ class ASOperator:
|
||||
if target_val is None:
|
||||
raise ValueError(f"'{target.name}' 的值为空,无法进行运算")
|
||||
if op == ".ADD.":
|
||||
ASOperator._arithAdd(target, target_val, operand, target_data)
|
||||
cls._arithAdd(target, target_val, operand, target_data)
|
||||
elif op == ".SUB.":
|
||||
ASOperator._arithSub(target, target_val, operand, target_data)
|
||||
cls._arithSub(target, target_val, operand, target_data)
|
||||
else:
|
||||
raise ValueError(f"不支持的操作 '{op}'")
|
||||
|
||||
@staticmethod
|
||||
@classmethod
|
||||
def _arithBinary(
|
||||
cls,
|
||||
target: ASObject,
|
||||
target_val,
|
||||
operand: ASObject,
|
||||
target_data: dict,
|
||||
sign: int
|
||||
):
|
||||
"""Apply ADD (sign=1) or SUB (sign=-1) per type."""
|
||||
|
||||
tp = target.var_type
|
||||
raw_op = operand._value
|
||||
op_name = "ADD" if sign == 1 else "SUB"
|
||||
|
||||
if tp == "Date":
|
||||
if not isinstance(target_val, date):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效日期")
|
||||
new_val = target_val + timedelta(days=int(raw_op)) * sign
|
||||
elif tp == "Time":
|
||||
if not isinstance(target_val, time):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效时间")
|
||||
delta = timedelta(hours=int(raw_op)) * sign
|
||||
dt = datetime.combine(datetime.today(), target_val) + delta
|
||||
new_val = dt.time()
|
||||
elif tp == "Int":
|
||||
new_val = int(target_val) + int(raw_op) * sign
|
||||
elif tp == "Float":
|
||||
new_val = float(target_val) + float(raw_op) * sign
|
||||
else:
|
||||
raise ValueError(f"'{tp}' 类型不支持 {op_name} 操作")
|
||||
target.setValue(new_val, target_data)
|
||||
|
||||
@classmethod
|
||||
def _arithAdd(
|
||||
cls,
|
||||
target: ASObject,
|
||||
target_val,
|
||||
operand: ASObject,
|
||||
target_data: dict
|
||||
):
|
||||
"""Dispatch ADD per type."""
|
||||
cls._arithBinary(target, target_val, operand, target_data, 1)
|
||||
|
||||
tp = target.var_type
|
||||
raw_op = operand._value
|
||||
|
||||
if tp == "Date":
|
||||
if not isinstance(target_val, date):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效日期")
|
||||
new_val = target_val + timedelta(days=int(raw_op))
|
||||
elif tp == "Time":
|
||||
if not isinstance(target_val, time):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效时间")
|
||||
dt = datetime.combine(datetime.today(), target_val) + timedelta(hours=int(raw_op))
|
||||
new_val = dt.time()
|
||||
elif tp == "Int":
|
||||
new_val = int(target_val) + int(raw_op)
|
||||
elif tp == "Float":
|
||||
new_val = float(target_val) + float(raw_op)
|
||||
else:
|
||||
raise ValueError(f"'{tp}' 类型不支持 ADD 操作")
|
||||
target.setValue(new_val, target_data)
|
||||
|
||||
@staticmethod
|
||||
@classmethod
|
||||
def _arithSub(
|
||||
cls,
|
||||
target: ASObject,
|
||||
target_val,
|
||||
operand: ASObject,
|
||||
target_data: dict
|
||||
):
|
||||
|
||||
"""Dispatch SUB per type."""
|
||||
cls._arithBinary(target, target_val, operand, target_data, -1)
|
||||
|
||||
tp = target.var_type
|
||||
raw_op = operand._value
|
||||
|
||||
if tp == "Date":
|
||||
if not isinstance(target_val, date):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效日期")
|
||||
new_val = target_val - timedelta(days=int(raw_op))
|
||||
elif tp == "Time":
|
||||
if not isinstance(target_val, time):
|
||||
raise ValueError(f"'{target.name}' 的值 '{target_val}' 不是有效时间")
|
||||
dt = datetime.combine(datetime.today(), target_val) - timedelta(hours=int(raw_op))
|
||||
new_val = dt.time()
|
||||
elif tp == "Int":
|
||||
new_val = int(target_val) - int(raw_op)
|
||||
elif tp == "Float":
|
||||
new_val = float(target_val) - float(raw_op)
|
||||
else:
|
||||
raise ValueError(f"'{tp}' 类型不支持 SUB 操作")
|
||||
target.setValue(new_val, target_data)
|
||||
|
||||
@staticmethod
|
||||
@classmethod
|
||||
def compare(
|
||||
cls,
|
||||
left: ASObject,
|
||||
right: ASObject,
|
||||
op: str,
|
||||
@@ -171,7 +174,7 @@ class ASOperator:
|
||||
ValueError: If the types are incompatible for comparison.
|
||||
"""
|
||||
|
||||
cmp_func = ASOperator._COMPARE.get(op)
|
||||
cmp_func = cls._COMPARE.get(op)
|
||||
if cmp_func is None:
|
||||
raise ValueError(f"未知的比较操作 '{op}'")
|
||||
left_val = left.getValue(target_data)
|
||||
|
||||
@@ -0,0 +1,478 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Copyright (c) 2026 KenanZhu.
|
||||
All rights reserved.
|
||||
|
||||
This software is provided "as is", without any warranty of any kind.
|
||||
You may use, modify, and distribute this file under the terms of the MIT License.
|
||||
See the LICENSE file for details.
|
||||
"""
|
||||
import re
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ASTokenizer",
|
||||
"Stmt",
|
||||
"Script",
|
||||
"IfNode",
|
||||
"ElifNode",
|
||||
"SetNode",
|
||||
"OpNode",
|
||||
"PassNode",
|
||||
"UnrecogNode",
|
||||
"NodeVisitor",
|
||||
"LineStrategy"
|
||||
]
|
||||
|
||||
|
||||
# Token kind constants
|
||||
K_IF = "IF"
|
||||
K_ELSE_IF = "ELSE IF"
|
||||
K_ELSE = "ELSE"
|
||||
K_ENDIF = "ENDIF"
|
||||
K_SET = "SET"
|
||||
K_ADD = "ADD"
|
||||
K_SUB = "SUB"
|
||||
K_PASS = "PASS"
|
||||
|
||||
# Op-type constants
|
||||
OP_SET = "set"
|
||||
OP_ADD = "add"
|
||||
OP_SUB = "sub"
|
||||
|
||||
# Compiled line patterns
|
||||
_RE_IF = re.compile(r"^IF\((.+)\)(?:\s+THEN\s*)?$", re.IGNORECASE)
|
||||
_RE_ELSE_IF = re.compile(r"^ELSE\s+IF\((.+)\)(?:\s+THEN\s*)?$", re.IGNORECASE)
|
||||
_RE_ELSE = re.compile(r"^ELSE\s*$", re.IGNORECASE)
|
||||
_RE_ENDIF = re.compile(r"^(ENDIF|END IF)$", re.IGNORECASE)
|
||||
_RE_SET = re.compile(r"^SET\s+(\w+)\s*=\s*(.+)$", re.IGNORECASE)
|
||||
_RE_ADD = re.compile(r"^(\w+)\s+\.ADD\.\s+(\d+)$", re.IGNORECASE)
|
||||
_RE_SUB = re.compile(r"^(\w+)\s+\.SUB\.\s+(\d+)$", re.IGNORECASE)
|
||||
_RE_PASS = re.compile(r"^\s*PASS\s*$", re.IGNORECASE)
|
||||
|
||||
|
||||
class Script:
|
||||
"""
|
||||
Root AST node for an entire AutoScript.
|
||||
Contains an ordered list of top-level statement nodes.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
body: list = None
|
||||
):
|
||||
|
||||
self.body = body or []
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitScript(self)
|
||||
|
||||
|
||||
class IfNode:
|
||||
"""
|
||||
IF conditional block with optional ELSE IF / ELSE branches.
|
||||
|
||||
Attributes:
|
||||
condition (str): Raw condition expression.
|
||||
body (list): Statements executed when condition is true.
|
||||
elif_branches (list[ElifNode]): ELSE IF branches in order.
|
||||
else_body (list): Statements executed for the ELSE branch.
|
||||
closed (bool): Whether this IF has a matching ENDIF token.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
condition: str = "",
|
||||
body: list = None,
|
||||
elif_branches: list = None,
|
||||
else_body: list = None,
|
||||
closed: bool = True
|
||||
):
|
||||
|
||||
self.condition = condition
|
||||
self.body = body or []
|
||||
self.elif_branches = elif_branches or []
|
||||
self.else_body = else_body or []
|
||||
self.closed = closed
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitIf(self)
|
||||
|
||||
|
||||
class ElifNode:
|
||||
"""
|
||||
ELSE IF branch within an IfNode.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
condition: str = "",
|
||||
body: list = None
|
||||
):
|
||||
|
||||
self.condition = condition
|
||||
self.body = body or []
|
||||
|
||||
|
||||
class SetNode:
|
||||
"""
|
||||
SET assignment statement.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
target: str = "",
|
||||
value: str = ""
|
||||
):
|
||||
|
||||
self.target = target
|
||||
self.value = value
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitSet(self)
|
||||
|
||||
|
||||
class OpNode:
|
||||
"""
|
||||
.ADD. / .SUB. operation statement.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
op_type: str = "",
|
||||
target: str = "",
|
||||
value: str = ""
|
||||
):
|
||||
|
||||
self.op_type = op_type
|
||||
self.target = target
|
||||
self.value = value
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitOp(self)
|
||||
|
||||
|
||||
class PassNode:
|
||||
"""
|
||||
PASS no-op statement.
|
||||
"""
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitPass(self)
|
||||
|
||||
|
||||
class UnrecogNode:
|
||||
"""
|
||||
Unrecognised line preserved for downstream error reporting.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
raw_line: str = ""
|
||||
):
|
||||
|
||||
self.raw_line = raw_line
|
||||
|
||||
def accept(
|
||||
self,
|
||||
visitor
|
||||
):
|
||||
|
||||
return visitor.visitUnrecog(self)
|
||||
|
||||
|
||||
class NodeVisitor:
|
||||
"""
|
||||
Base visitor for the AutoScript AST.
|
||||
|
||||
Subclass and override visit* methods to implement
|
||||
custom traversal logic. Default walks tree depth-first.
|
||||
"""
|
||||
|
||||
def visitScript(
|
||||
self,
|
||||
_node: Script
|
||||
):
|
||||
|
||||
for child in _node.body:
|
||||
child.accept(self)
|
||||
|
||||
def visitIf(
|
||||
self,
|
||||
_node: IfNode
|
||||
):
|
||||
|
||||
for child in _node.body:
|
||||
child.accept(self)
|
||||
for elif_node in _node.elif_branches:
|
||||
for child in elif_node.body:
|
||||
child.accept(self)
|
||||
for child in _node.else_body:
|
||||
child.accept(self)
|
||||
|
||||
def visitSet(
|
||||
self,
|
||||
_node: SetNode
|
||||
):
|
||||
|
||||
pass
|
||||
|
||||
def visitOp(
|
||||
self,
|
||||
_node: OpNode
|
||||
):
|
||||
|
||||
pass
|
||||
|
||||
def visitPass(
|
||||
self,
|
||||
_node: PassNode
|
||||
):
|
||||
|
||||
pass
|
||||
|
||||
def visitUnrecog(
|
||||
self,
|
||||
_node: UnrecogNode
|
||||
):
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class LineStrategy:
|
||||
"""
|
||||
Encapsulates a regex pattern and its data-extraction handler.
|
||||
Used by the tokenizer to classify a single line.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
pattern,
|
||||
handler
|
||||
):
|
||||
|
||||
self.pattern = pattern
|
||||
self.handler = handler
|
||||
|
||||
def match(
|
||||
self,
|
||||
line: str
|
||||
):
|
||||
|
||||
m = self.pattern.match(line)
|
||||
if m:
|
||||
return self.handler(m)
|
||||
return None
|
||||
|
||||
|
||||
# Strategy instances — one per recognised AutoScript syntax form
|
||||
_LINE_STRATEGIES = [
|
||||
LineStrategy(_RE_IF, lambda m: (K_IF, m.group(1))),
|
||||
LineStrategy(_RE_ELSE_IF, lambda m: (K_ELSE_IF, m.group(1))),
|
||||
LineStrategy(_RE_ELSE, lambda m: (K_ELSE, None)),
|
||||
LineStrategy(_RE_ENDIF, lambda m: (K_ENDIF, None)),
|
||||
LineStrategy(_RE_SET, lambda m: (K_SET, (m.group(1).strip(), m.group(2).strip()))),
|
||||
LineStrategy(_RE_ADD, lambda m: (K_ADD, (m.group(1).strip(), m.group(2).strip()))),
|
||||
LineStrategy(_RE_SUB, lambda m: (K_SUB, (m.group(1).strip(), m.group(2).strip()))),
|
||||
LineStrategy(_RE_PASS, lambda m: (K_PASS, None)),
|
||||
]
|
||||
|
||||
|
||||
class Stmt:
|
||||
"""
|
||||
Flat statement container, backward-compatible with the original
|
||||
tokenize() output and the orchestration dialog's _classifyLine.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
kind: str | None = None,
|
||||
condition: str | None = None,
|
||||
target: str | None = None,
|
||||
value: str | None = None,
|
||||
op_type: str | None = None,
|
||||
raw_line: str = ""
|
||||
):
|
||||
|
||||
self.kind = kind
|
||||
self.condition = condition
|
||||
self.target = target
|
||||
self.value = value
|
||||
self.op_type = op_type
|
||||
self.raw_line = raw_line
|
||||
|
||||
|
||||
class ASTokenizer:
|
||||
"""
|
||||
Tokenizer / parser for the AutoScript DSL.
|
||||
|
||||
Provides three entry points:
|
||||
- classifyLine(line) — single-line classifier.
|
||||
- tokenize(script) — flat Stmt list.
|
||||
- parse(script) — structured AST (Script root).
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def _matchLine(
|
||||
cls,
|
||||
stripped: str
|
||||
):
|
||||
|
||||
for strategy in _LINE_STRATEGIES:
|
||||
result = strategy.match(stripped)
|
||||
if result:
|
||||
return result
|
||||
return (None, None)
|
||||
|
||||
@classmethod
|
||||
def classifyLine(
|
||||
cls,
|
||||
stripped: str
|
||||
):
|
||||
|
||||
kind, data = cls._matchLine(stripped)
|
||||
if kind is None or kind == K_PASS:
|
||||
return None
|
||||
return (kind, data)
|
||||
|
||||
@classmethod
|
||||
def tokenize(
|
||||
cls,
|
||||
script: str
|
||||
) -> list:
|
||||
|
||||
statements = []
|
||||
for raw_line in script.split("\n"):
|
||||
stripped = raw_line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
kind, data = cls._matchLine(stripped)
|
||||
stmt = Stmt(kind=kind, raw_line=stripped)
|
||||
|
||||
if kind == K_IF or kind == K_ELSE_IF:
|
||||
stmt.condition = data
|
||||
elif kind == K_SET:
|
||||
stmt.target, stmt.value = data
|
||||
stmt.op_type = OP_SET
|
||||
elif kind == K_ADD:
|
||||
stmt.target, stmt.value = data
|
||||
stmt.op_type = OP_ADD
|
||||
elif kind == K_SUB:
|
||||
stmt.target, stmt.value = data
|
||||
stmt.op_type = OP_SUB
|
||||
statements.append(stmt)
|
||||
return statements
|
||||
|
||||
@classmethod
|
||||
def parse(
|
||||
cls,
|
||||
script: str
|
||||
) -> Script:
|
||||
|
||||
tokens = cls.tokenize(script)
|
||||
body = []
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
tok = tokens[i]
|
||||
kind = tok.kind
|
||||
|
||||
if kind == K_IF:
|
||||
node, consumed = cls._parseIfBlock(tokens, i)
|
||||
body.append(node)
|
||||
i += consumed
|
||||
elif kind in (K_ELSE_IF, K_ELSE, K_ENDIF):
|
||||
i += 1
|
||||
elif kind == K_SET:
|
||||
body.append(SetNode(target=tok.target, value=tok.value))
|
||||
i += 1
|
||||
elif kind in (K_ADD, K_SUB):
|
||||
body.append(OpNode(
|
||||
op_type=tok.op_type,
|
||||
target=tok.target,
|
||||
value=tok.value
|
||||
))
|
||||
i += 1
|
||||
elif kind == K_PASS:
|
||||
body.append(PassNode())
|
||||
i += 1
|
||||
else:
|
||||
body.append(UnrecogNode(raw_line=tok.raw_line))
|
||||
i += 1
|
||||
return Script(body=body)
|
||||
|
||||
@classmethod
|
||||
def _parseIfBlock(
|
||||
cls,
|
||||
tokens: list,
|
||||
start: int
|
||||
):
|
||||
|
||||
first = tokens[start]
|
||||
node = IfNode(condition=first.condition or "")
|
||||
body = []
|
||||
elif_branches = []
|
||||
else_body = []
|
||||
current_target = body
|
||||
i = start + 1
|
||||
|
||||
while i < len(tokens):
|
||||
tok = tokens[i]
|
||||
kind = tok.kind
|
||||
if kind == K_IF:
|
||||
sub_node, consumed = cls._parseIfBlock(tokens, i)
|
||||
current_target.append(sub_node)
|
||||
i += consumed
|
||||
elif kind == K_ELSE_IF:
|
||||
elif_branches.append(ElifNode(condition=tok.condition or ""))
|
||||
current_target = elif_branches[-1].body
|
||||
i += 1
|
||||
elif kind == K_ELSE:
|
||||
else_body = []
|
||||
current_target = else_body
|
||||
i += 1
|
||||
elif kind == K_ENDIF:
|
||||
node.body = body
|
||||
node.elif_branches = elif_branches
|
||||
node.else_body = else_body
|
||||
return (node, i - start + 1)
|
||||
elif kind == K_SET:
|
||||
current_target.append(SetNode(target=tok.target, value=tok.value))
|
||||
i += 1
|
||||
elif kind in (K_ADD, K_SUB):
|
||||
current_target.append(OpNode(
|
||||
op_type=tok.op_type,
|
||||
target=tok.target,
|
||||
value=tok.value
|
||||
))
|
||||
i += 1
|
||||
elif kind == K_PASS:
|
||||
current_target.append(PassNode())
|
||||
i += 1
|
||||
else:
|
||||
current_target.append(UnrecogNode(raw_line=tok.raw_line))
|
||||
i += 1
|
||||
node.body = body
|
||||
node.elif_branches = elif_branches
|
||||
node.else_body = else_body
|
||||
node.closed = False
|
||||
return (node, i - start)
|
||||
@@ -11,14 +11,37 @@
|
||||
- registerDefaultTargetVars(): Register all built-in target variables.
|
||||
- META_VARS: dict of built-in read-only meta variables.
|
||||
- ALL_VARIABLES: dict of all available variables (display_name -> (name, type)).
|
||||
- ASTokenizer: Unified tokenizer for the orchestration dialog and engine.
|
||||
"""
|
||||
|
||||
from autoscript.ASEngine import execute, addTargetVar
|
||||
from autoscript.ASTokenizer import (
|
||||
ASTokenizer,
|
||||
Stmt,
|
||||
ElifNode,
|
||||
Script,
|
||||
IfNode,
|
||||
SetNode,
|
||||
OpNode,
|
||||
)
|
||||
from autoscript.ASEngine import (
|
||||
execute,
|
||||
addTargetVar,
|
||||
)
|
||||
from autoscript.ASObject import _META_VARS as META_VARS
|
||||
|
||||
__all__ = [
|
||||
"execute", "addTargetVar", "registerDefaultTargetVars",
|
||||
"META_VARS", "ALL_VARIABLES",
|
||||
"execute",
|
||||
"addTargetVar",
|
||||
"registerDefaultTargetVars",
|
||||
"buildMockTargetData",
|
||||
"META_VARS",
|
||||
"ALL_VARIABLES",
|
||||
"ASTokenizer",
|
||||
"Stmt",
|
||||
"Script",
|
||||
"IfNode",
|
||||
"SetNode",
|
||||
"OpNode",
|
||||
"ElifNode"
|
||||
]
|
||||
|
||||
# All variables available to scripts (display_name -> (name, type)).
|
||||
@@ -43,9 +66,33 @@ _TARGET_VAR_DEFS = [
|
||||
("RESERVE_BEGIN_TIME", "Time", ["reserve_info", "begin_time", "time"], "预约开始时间"),
|
||||
("RESERVE_END_TIME", "Time", ["reserve_info", "end_time", "time"], "预约结束时间"),
|
||||
]
|
||||
_MOCK_TYPE_VALUES = {
|
||||
"String": "__mock__",
|
||||
"Boolean": True,
|
||||
"Date": "2099-01-01",
|
||||
"Time": "00:00",
|
||||
"Int": 0,
|
||||
"Float": 0.0,
|
||||
}
|
||||
|
||||
|
||||
def registerDefaultTargetVars() -> None:
|
||||
def buildMockTargetData(
|
||||
) -> dict:
|
||||
"""
|
||||
Build a target_data dict filled with type-appropriate mock values
|
||||
for all registered target variables.
|
||||
"""
|
||||
data = {}
|
||||
for _, var_type, key_path, _ in _TARGET_VAR_DEFS:
|
||||
d = data
|
||||
for key in key_path[:-1]:
|
||||
d = d.setdefault(key, {})
|
||||
d[key_path[-1]] = _MOCK_TYPE_VALUES.get(var_type, "")
|
||||
return data
|
||||
|
||||
|
||||
def registerDefaultTargetVars(
|
||||
) -> None:
|
||||
"""
|
||||
Register all built-in target variables with the engine.
|
||||
This must be called before any script execution.
|
||||
|
||||
Reference in New Issue
Block a user