mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
refactor(autoscript): 替换 dsl 包为 autoscript 引擎模块
This commit is contained in:
@@ -0,0 +1,573 @@
|
||||
# -*- 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
|
||||
from datetime import datetime, timedelta, date, time
|
||||
|
||||
from .ASObject import ASObject, _META_VARS, _inferType
|
||||
from .ASOperator import ASOperator
|
||||
|
||||
|
||||
__all__ = ["execute", "addTargetVar"]
|
||||
|
||||
|
||||
# Engine state
|
||||
# User-registered target variables (bound to target_data paths)
|
||||
_TARGET_VARS = {}
|
||||
# Free-form script variables (not bound to target_data)
|
||||
_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(
|
||||
message: str
|
||||
) -> str:
|
||||
"""
|
||||
Format an error message with the current script line number.
|
||||
|
||||
Args:
|
||||
message (str): The error description.
|
||||
|
||||
Returns:
|
||||
str: A formatted error string like "AutoScript syntax error(line X): message".
|
||||
"""
|
||||
return f"AutoScript 语法错误(第{_CUR_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
|
||||
|
||||
|
||||
def _splitTopLevel(
|
||||
text: str,
|
||||
delimiter: str
|
||||
) -> list:
|
||||
"""
|
||||
Split a condition expression by a delimiter (.AND. / .OR.), respecting parentheses.
|
||||
|
||||
Only splits at the top nesting level; delimiters inside parentheses are ignored.
|
||||
|
||||
Args:
|
||||
text (str): The condition expression to split.
|
||||
delimiter (str): The delimiter string, e.g. ".OR." or ".AND.".
|
||||
|
||||
Returns:
|
||||
list: A list of sub-expression strings (stripped of leading/trailing whitespace).
|
||||
"""
|
||||
|
||||
parts = []
|
||||
depth = 0
|
||||
buf = ""
|
||||
i = 0
|
||||
text_upper = text.upper()
|
||||
delim_upper = delimiter.upper()
|
||||
dlen = len(delim_upper)
|
||||
while i < len(text):
|
||||
if text[i] == "(":
|
||||
depth += 1
|
||||
buf += text[i]
|
||||
elif text[i] == ")":
|
||||
depth -= 1
|
||||
buf += text[i]
|
||||
elif depth == 0 and text_upper[i:i + dlen] == delim_upper:
|
||||
parts.append(buf)
|
||||
buf = ""
|
||||
i += dlen
|
||||
continue
|
||||
else:
|
||||
buf += text[i]
|
||||
i += 1
|
||||
if buf.strip():
|
||||
parts.append(buf)
|
||||
return parts
|
||||
|
||||
|
||||
def _buildFieldMap():
|
||||
"""
|
||||
Rebuild the _FIELD_MAP lookup from _META_VARS and _TARGET_VARS.
|
||||
|
||||
Each variable is registered under both its canonical name (uppercased)
|
||||
and its display_name (if present), so that scripts can refer to either.
|
||||
"""
|
||||
|
||||
_FIELD_MAP.clear()
|
||||
for ch_name, obj in _META_VARS.items():
|
||||
_FIELD_MAP[obj.name.upper()] = obj
|
||||
_FIELD_MAP[ch_name.upper().strip()] = obj
|
||||
for obj in _TARGET_VARS.values():
|
||||
_FIELD_MAP[obj.name.upper()] = obj
|
||||
if obj.display_name:
|
||||
_FIELD_MAP[obj.display_name.upper().strip()] = obj
|
||||
|
||||
|
||||
def _resolveFieldObj(
|
||||
field_name: str
|
||||
):
|
||||
"""
|
||||
Resolve a field name to its ASObject by looking up _FIELD_MAP then _SCRIPT_VARS.
|
||||
|
||||
Unlike getting a raw value, this returns the ASObject instance itself,
|
||||
preserving type information for operations and comparisons.
|
||||
|
||||
Args:
|
||||
field_name (str): The field name (case-insensitive).
|
||||
|
||||
Returns:
|
||||
ASObject or None: The resolved ASObject, or None if not found.
|
||||
"""
|
||||
|
||||
upper_name = field_name.upper().strip()
|
||||
obj = _FIELD_MAP.get(upper_name)
|
||||
if obj:
|
||||
return obj
|
||||
obj = _SCRIPT_VARS.get(upper_name)
|
||||
if obj:
|
||||
return obj
|
||||
return None
|
||||
|
||||
|
||||
def _resolveValue(
|
||||
value_str: str,
|
||||
target_data: dict
|
||||
):
|
||||
"""
|
||||
Parse and resolve a value string from a script into a Python object.
|
||||
|
||||
Supports the following literal forms:
|
||||
- TIME(hh:mm)
|
||||
- DATE(yyyy-mm-dd)
|
||||
- .TRUE. / .FALSE.
|
||||
- Single/double quoted strings (with escaped single quotes)
|
||||
- CURRENT_DATE + N / CURRENT_TIME + N (relative offsets)
|
||||
- Numeric literals (int / float)
|
||||
- Field references (resolved via _resolveFieldObj)
|
||||
|
||||
Args:
|
||||
value_str (str): The raw value string from the script.
|
||||
target_data (dict): The application data dict.
|
||||
|
||||
Returns:
|
||||
The resolved Python value.
|
||||
"""
|
||||
|
||||
s = value_str.strip()
|
||||
m = re.match(r"^TIME\((\d{1,2}):(\d{2})\)$", s, re.IGNORECASE)
|
||||
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)
|
||||
if m:
|
||||
return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||||
up = s.upper()
|
||||
if up == ".TRUE.":
|
||||
return True
|
||||
if up == ".FALSE.":
|
||||
return False
|
||||
if s.startswith("'") and s.endswith("'"):
|
||||
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)
|
||||
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)
|
||||
if m:
|
||||
hours = int(m.group(1))
|
||||
return (datetime.now() + timedelta(hours=hours)).time()
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
return float(s)
|
||||
except ValueError:
|
||||
pass
|
||||
obj = _resolveFieldObj(s)
|
||||
if obj:
|
||||
return obj.getValue(target_data)
|
||||
return ""
|
||||
|
||||
|
||||
def _resolveAsObject(
|
||||
expr: str,
|
||||
target_data: dict
|
||||
) -> ASObject:
|
||||
"""
|
||||
Resolve a value expression to an ASObject.
|
||||
|
||||
- If the expression is a registered field name, returns its ASObject directly.
|
||||
- If the expression is a literal (number, string, DATE(), TIME(), bool),
|
||||
creates a temporary ASObject with the inferred type.
|
||||
|
||||
This is the key function that ensures all internal operations work
|
||||
with typed ASObject instances rather than raw Python values.
|
||||
|
||||
Args:
|
||||
expr (str): The raw expression string from the script.
|
||||
target_data (dict): The application data dict.
|
||||
|
||||
Returns:
|
||||
ASObject: A registered or temporary ASObject representing the expression value.
|
||||
"""
|
||||
|
||||
s = expr.strip()
|
||||
obj = _resolveFieldObj(s)
|
||||
if obj is not None:
|
||||
return obj
|
||||
value = _resolveValue(s, target_data)
|
||||
inferred = _inferType(value, s)
|
||||
return ASObject._makeTemp(value, inferred)
|
||||
|
||||
|
||||
def _evaluateCondition(
|
||||
condition_str: str,
|
||||
target_data: dict
|
||||
) -> bool:
|
||||
"""
|
||||
Evaluate a condition expression and return a boolean result.
|
||||
|
||||
Supports:
|
||||
- Boolean literals: .TRUE., .FALSE.
|
||||
- .AND. / .OR. operators (lowest precedence)
|
||||
- Parenthesised sub-expressions
|
||||
- Comparison operators: .EQ., .NEQ., .BGT., .BLT., .BGE., .BLE.
|
||||
|
||||
All operands are resolved as ASObject instances (via _resolveAsObject)
|
||||
and comparisons are delegated to ASOperator.compare().
|
||||
|
||||
Args:
|
||||
condition_str (str): The raw condition expression from the script.
|
||||
target_data (dict): The application data dict.
|
||||
|
||||
Returns:
|
||||
bool: The evaluation result.
|
||||
|
||||
Raises:
|
||||
ValueError: If the expression contains unrecognised tokens or type-mismatched comparisons.
|
||||
"""
|
||||
|
||||
s = condition_str.strip()
|
||||
if not s:
|
||||
return False
|
||||
or_parts = _splitTopLevel(s, ".OR.")
|
||||
if len(or_parts) > 1:
|
||||
return any(
|
||||
_evaluateCondition(p.strip(), target_data)
|
||||
for p in or_parts
|
||||
)
|
||||
and_parts = _splitTopLevel(s, ".AND.")
|
||||
if len(and_parts) > 1:
|
||||
return all(
|
||||
_evaluateCondition(p.strip(), target_data)
|
||||
for p in and_parts
|
||||
)
|
||||
s = s.strip()
|
||||
if s.startswith("(") and s.endswith(")"):
|
||||
return _evaluateCondition(s[1:-1], target_data)
|
||||
up = s.upper()
|
||||
if up == ".TRUE.":
|
||||
return True
|
||||
if up == ".FALSE.":
|
||||
return False
|
||||
for op in ASOperator._COMPARE:
|
||||
idx = up.find(op.upper())
|
||||
if idx < 0:
|
||||
continue
|
||||
left_raw = s[:idx].strip()
|
||||
right_raw = s[idx + len(op):].strip()
|
||||
left_obj = _resolveAsObject(left_raw, target_data)
|
||||
right_obj = _resolveAsObject(right_raw, target_data)
|
||||
return ASOperator.compare(left_obj, right_obj, op, target_data)
|
||||
raise ValueError(
|
||||
_errPos(f"无法识别的条件表达式 '{condition_str}'")
|
||||
)
|
||||
|
||||
|
||||
def _executeSet(
|
||||
line: str,
|
||||
target_data: dict
|
||||
):
|
||||
"""
|
||||
Execute a SET statement to assign a value to a field or script variable.
|
||||
|
||||
Parses the line as "SET field_name = value_expr", resolves the value,
|
||||
and assigns it. If the target field does not exist, a new script variable
|
||||
is created with an inferred type.
|
||||
|
||||
Args:
|
||||
line (str): The raw SET line from the script.
|
||||
target_data (dict): The application data dict.
|
||||
|
||||
Raises:
|
||||
ValueError: If the value string contains unexpected extra tokens.
|
||||
"""
|
||||
|
||||
rest = line[3:].strip()
|
||||
eq_idx = rest.find("=")
|
||||
if eq_idx < 0:
|
||||
return
|
||||
field_name = rest[:eq_idx].strip()
|
||||
value_str = rest[eq_idx + 1:].strip()
|
||||
if not field_name:
|
||||
return
|
||||
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}'"))
|
||||
upper_name = field_name.upper().strip()
|
||||
obj = _FIELD_MAP.get(upper_name)
|
||||
if not obj:
|
||||
obj = _SCRIPT_VARS.get(upper_name)
|
||||
if obj:
|
||||
obj.setValue(resolved, target_data)
|
||||
return
|
||||
inferred_type = _inferType(resolved, stripped)
|
||||
new_var = ASObject(
|
||||
upper_name,
|
||||
inferred_type,
|
||||
read_only=False,
|
||||
is_config=False,
|
||||
default_value=resolved
|
||||
)
|
||||
_SCRIPT_VARS[upper_name] = new_var
|
||||
|
||||
|
||||
def _executeOperation(
|
||||
line: str,
|
||||
target_data: dict
|
||||
):
|
||||
"""
|
||||
Execute a field operation statement: "FIELD .ADD. N" or "FIELD .SUB. N".
|
||||
|
||||
Resolves the left side as a registered ASObject and the right side
|
||||
as a temporary numeric ASObject, then delegates to ASOperator.apply().
|
||||
|
||||
Args:
|
||||
line (str): The raw operation line from the script (e.g. "RESERVE_DATE .ADD. 1").
|
||||
target_data (dict): The application data dict.
|
||||
|
||||
Raises:
|
||||
ValueError: If the field is unknown, the operand is invalid,
|
||||
or the type does not support the operation.
|
||||
"""
|
||||
|
||||
parts = line.split()
|
||||
if len(parts) < 3:
|
||||
return
|
||||
if len(parts) > 3:
|
||||
raise ValueError(
|
||||
_errPos(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}'"))
|
||||
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,
|
||||
key_path: list,
|
||||
display_name: str = None
|
||||
):
|
||||
"""
|
||||
Register a new target variable bound to a path in the application data dict.
|
||||
|
||||
Once registered, the variable can be read, written, and operated on in scripts
|
||||
using its canonical name or display_name.
|
||||
|
||||
Args:
|
||||
name (str): The canonical variable name (e.g. "RESERVE_DATE").
|
||||
var_type (str): The type ("Int", "Float", "Boolean", "Date", "Time", "String").
|
||||
key_path (list): The nested path into target_data, e.g. ["reserve_info", "date"].
|
||||
display_name (str): An optional Chinese alias for use in script conditions.
|
||||
|
||||
Example:
|
||||
>>> addTargetVar("MY_FIELD", "String", ["custom", "field"], display_name="自定义字段")
|
||||
"""
|
||||
|
||||
upper_name = name.upper().strip()
|
||||
obj = ASObject(
|
||||
upper_name,
|
||||
var_type,
|
||||
is_config=True,
|
||||
key_path=key_path,
|
||||
display_name=display_name
|
||||
)
|
||||
_TARGET_VARS[upper_name] = obj
|
||||
|
||||
|
||||
def execute(
|
||||
script_text: str,
|
||||
target_data: dict
|
||||
):
|
||||
"""
|
||||
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.
|
||||
|
||||
Args:
|
||||
script_text (str): The AutoScript source code.
|
||||
target_data (dict): The application data dict to read from / write to.
|
||||
|
||||
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:
|
||||
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 不匹配"
|
||||
)
|
||||
@@ -0,0 +1,251 @@
|
||||
# -*- 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
|
||||
from datetime import datetime, date, time
|
||||
|
||||
|
||||
__all__ = ["ASObject", "_META_VARS", "_inferType"]
|
||||
|
||||
|
||||
# Default values for each supported type when no value is present
|
||||
_TYPE_DEFAULTS = {
|
||||
"Int": 0,
|
||||
"Float": 0.0,
|
||||
"Boolean": False,
|
||||
"Date": None,
|
||||
"Time": None,
|
||||
"String": "",
|
||||
}
|
||||
|
||||
|
||||
class ASObject:
|
||||
"""
|
||||
Represents a variable object used throughout the AutoScript engine.
|
||||
|
||||
An ASObject can be a meta variable (read-only, e.g. CURRENT_DATE),
|
||||
a target variable (bound to a config data dict via key_path),
|
||||
or a script variable (free-form, stored internally).
|
||||
|
||||
Args:
|
||||
name (str): The canonical name of the variable (case-insensitive).
|
||||
var_type (str): One of "Int", "Float", "Boolean", "Date", "Time", "String".
|
||||
read_only (bool): Whether the variable is read-only (default: False).
|
||||
is_config (bool): Whether the variable maps to a target_data path (default: False).
|
||||
key_path (list): The nested key path into target_data, e.g. ["reserve_info", "date"].
|
||||
default_value: The fallback value when no target_data value is found.
|
||||
display_name (str): An alias for use in script conditions and assignments.
|
||||
|
||||
Example:
|
||||
>>> obj = ASObject("MY_DATE", "Date", is_config=True,
|
||||
... key_path=["reserve_info", "date"],
|
||||
... display_name="预约日期")
|
||||
>>> obj.getValue({"reserve_info": {"date": "2026-05-01"}})
|
||||
datetime.date(2026, 5, 1)
|
||||
"""
|
||||
|
||||
_TEMP_COUNTER = 0
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
var_type: str, *,
|
||||
read_only: bool = False,
|
||||
is_config: bool = False,
|
||||
key_path: list = None,
|
||||
default_value=None,
|
||||
display_name: str = None
|
||||
):
|
||||
|
||||
self.name = name
|
||||
self.var_type = var_type
|
||||
self.read_only = read_only
|
||||
self.is_config = is_config
|
||||
self.key_path = key_path or []
|
||||
self._value = default_value
|
||||
self.display_name = display_name
|
||||
|
||||
@classmethod
|
||||
def _makeTemp(
|
||||
cls,
|
||||
value,
|
||||
inferred_type: str
|
||||
):
|
||||
"""
|
||||
Create a temporary unnamed ASObject from a literal value.
|
||||
|
||||
Temporary objects are used for inline script literals (e.g. 42,
|
||||
'hello', DATE(2026-01-01)) so they can participate in typed
|
||||
operations alongside registered variables.
|
||||
|
||||
Args:
|
||||
value: The resolved Python value.
|
||||
inferred_type (str): The AutoScript type name.
|
||||
|
||||
Returns:
|
||||
ASObject: A temporary, non-config, read-write ASObject.
|
||||
"""
|
||||
|
||||
cls._TEMP_COUNTER += 1
|
||||
return cls(
|
||||
f"__TMP_{cls._TEMP_COUNTER}",
|
||||
inferred_type,
|
||||
read_only=False,
|
||||
is_config=False,
|
||||
default_value=value
|
||||
)
|
||||
|
||||
|
||||
def _typeEmpty(
|
||||
self
|
||||
):
|
||||
"""
|
||||
Return the type-appropriate empty / default value.
|
||||
"""
|
||||
|
||||
return _TYPE_DEFAULTS.get(self.var_type, "")
|
||||
|
||||
|
||||
def getValue(
|
||||
self,
|
||||
target_data: dict = None
|
||||
):
|
||||
"""
|
||||
Retrieve the current value of this variable.
|
||||
|
||||
For read-only variables (CURRENT_DATE, CURRENT_TIME), returns the
|
||||
live datetime. For config variables, traverses the key_path into
|
||||
target_data and parses Date/Time strings. Otherwise returns the
|
||||
internal _value.
|
||||
|
||||
Args:
|
||||
target_data (dict): The application data dict (required for config vars).
|
||||
|
||||
Returns:
|
||||
The resolved value, or a type-appropriate default if missing.
|
||||
"""
|
||||
|
||||
if self.read_only:
|
||||
if self.name == "CURRENT_DATE":
|
||||
return datetime.now().date()
|
||||
if self.name == "CURRENT_TIME":
|
||||
return datetime.now().time()
|
||||
return self._value
|
||||
if self.is_config and target_data is not None and self.key_path:
|
||||
d = target_data
|
||||
for key in self.key_path[:-1]:
|
||||
d = d.get(key, {})
|
||||
if not isinstance(d, dict):
|
||||
return self._typeEmpty()
|
||||
raw = d.get(self.key_path[-1])
|
||||
if raw is None:
|
||||
return self._typeEmpty()
|
||||
if self.var_type == "Date" and isinstance(raw, str):
|
||||
try:
|
||||
return datetime.strptime(raw, "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
return self._typeEmpty()
|
||||
if self.var_type == "Time" and isinstance(raw, str):
|
||||
try:
|
||||
return datetime.strptime(raw, "%H:%M").time()
|
||||
except ValueError:
|
||||
return self._typeEmpty()
|
||||
return raw
|
||||
return self._value
|
||||
|
||||
|
||||
def setValue(
|
||||
self,
|
||||
value,
|
||||
target_data: dict = None
|
||||
):
|
||||
"""
|
||||
Assign a new value to this variable, with type coercion.
|
||||
|
||||
Performs coercion for Boolean (string -> bool), Int, and Float types.
|
||||
For config variables, dates/times are converted back to strings before
|
||||
writing into target_data.
|
||||
|
||||
Args:
|
||||
value: The value to assign.
|
||||
target_data (dict): The application data dict (required for config vars).
|
||||
|
||||
Raises:
|
||||
ValueError: If the variable is read-only or value cannot be coerced.
|
||||
"""
|
||||
|
||||
if self.read_only:
|
||||
raise ValueError(f"不能修改只读变量 '{self.name}'")
|
||||
if self.var_type == "Boolean" and not isinstance(value, bool):
|
||||
value = (str(value).upper() == "TRUE")
|
||||
if self.var_type == "Int" and not isinstance(value, int):
|
||||
try:
|
||||
value = int(value)
|
||||
except (ValueError, TypeError):
|
||||
raise ValueError(f"无法将值 '{value}' 转换为 Int 类型")
|
||||
if self.var_type == "Float" and not isinstance(value, float):
|
||||
try:
|
||||
value = float(value)
|
||||
except (ValueError, TypeError):
|
||||
raise ValueError(f"无法将值 '{value}' 转换为 Float 类型")
|
||||
if self.is_config:
|
||||
if self.var_type == "Date" and isinstance(value, date):
|
||||
value = value.strftime("%Y-%m-%d")
|
||||
if self.var_type == "Time" and isinstance(value, time):
|
||||
value = value.strftime("%H:%M")
|
||||
if self.is_config and target_data is not None and self.key_path:
|
||||
d = target_data
|
||||
for key in self.key_path[:-1]:
|
||||
d = d.setdefault(key, {})
|
||||
d[self.key_path[-1]] = value
|
||||
else:
|
||||
self._value = value
|
||||
|
||||
|
||||
# Built-in read-only meta variables available to all scripts
|
||||
_META_VARS = {
|
||||
"CURRENT_DATE": ASObject("CURRENT_DATE", "Date", read_only=True, display_name="当前日期"),
|
||||
"CURRENT_TIME": ASObject("CURRENT_TIME", "Time", read_only=True, display_name="当前时间"),
|
||||
}
|
||||
|
||||
|
||||
def _inferType(
|
||||
value,
|
||||
raw_expr: str = None
|
||||
) -> str:
|
||||
"""
|
||||
Infer the ASObject type string from a Python value or raw expression.
|
||||
|
||||
When the Python type is ambiguous (e.g. int can be Int or a component
|
||||
of Date), the raw_expr is used as a hint.
|
||||
|
||||
Args:
|
||||
value: The resolved Python value.
|
||||
raw_expr (str): The original expression string from the script (optional).
|
||||
|
||||
Returns:
|
||||
str: One of "Boolean", "Int", "Float", "Date", "Time", "String".
|
||||
"""
|
||||
|
||||
if isinstance(value, bool):
|
||||
return "Boolean"
|
||||
if isinstance(value, int):
|
||||
return "Int"
|
||||
if isinstance(value, float):
|
||||
return "Float"
|
||||
if isinstance(value, date):
|
||||
return "Date"
|
||||
if isinstance(value, time):
|
||||
return "Time"
|
||||
if raw_expr:
|
||||
if re.match(r"^DATE\(\d{4}-\d{2}-\d{2}\)$", raw_expr, re.IGNORECASE):
|
||||
return "Date"
|
||||
if re.match(r"^TIME\(\d{1,2}:\d{2}\)$", raw_expr, re.IGNORECASE):
|
||||
return "Time"
|
||||
return "String"
|
||||
@@ -0,0 +1,185 @@
|
||||
# -*- 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.
|
||||
"""
|
||||
from datetime import datetime, timedelta, date, time
|
||||
|
||||
from .ASObject import ASObject
|
||||
|
||||
|
||||
__all__ = ["ASOperator"]
|
||||
|
||||
|
||||
class ASOperator:
|
||||
"""
|
||||
Centralised type-safe operations for AutoScript engine types.
|
||||
|
||||
All arithmetic (ADD / SUB) and comparison operators are routed through
|
||||
this class, which dispatches to the correct Python-level logic based on
|
||||
the ASObject's var_type. This keeps type-specific branching in one
|
||||
place instead of scattering it across the engine.
|
||||
|
||||
Args:
|
||||
op (str): One of ".ADD.", ".SUB.", ".EQ.", ".NEQ.", ".BGT.",
|
||||
".BLT.", ".BGE.", ".BLE.".
|
||||
|
||||
Example:
|
||||
>>> obj = ASObject("X", "Int", default_value=10)
|
||||
>>> ASOperator.apply(obj, ASObject._makeTemp(5, "Int"), ".ADD.", None)
|
||||
>>> obj.getValue()
|
||||
15
|
||||
>>> ASOperator.compare(
|
||||
... obj,
|
||||
... ASObject._makeTemp(15, "Int"),
|
||||
... ".EQ.", None
|
||||
... )
|
||||
True
|
||||
"""
|
||||
|
||||
_COMPARE = {
|
||||
".EQ." : lambda a, b: a == b,
|
||||
".NEQ.": lambda a, b: a != b,
|
||||
".BGT.": lambda a, b: a > b,
|
||||
".BLT.": lambda a, b: a < b,
|
||||
".BGE.": lambda a, b: a >= b,
|
||||
".BLE.": lambda a, b: a <= b,
|
||||
}
|
||||
_ARITH_TYPES = {"Date", "Time", "Int", "Float"}
|
||||
|
||||
@staticmethod
|
||||
def apply(
|
||||
target: ASObject,
|
||||
operand: ASObject,
|
||||
op: str,
|
||||
target_data: dict
|
||||
):
|
||||
"""
|
||||
Apply ADD or SUB to a target ASObject, modifying it in place.
|
||||
|
||||
Args:
|
||||
target (ASObject): The variable to modify.
|
||||
operand (ASObject): The operand (numeric value for Date/Time/Int/Float).
|
||||
op (str): ".ADD." or ".SUB.".
|
||||
target_data (dict): Application data dict (passed through to getValue/setValue).
|
||||
|
||||
Raises:
|
||||
ValueError: If the type does not support the operation or values are invalid.
|
||||
"""
|
||||
|
||||
tp = target.var_type
|
||||
op_tp = operand.var_type
|
||||
if tp not in ASOperator._ARITH_TYPES:
|
||||
raise ValueError(f"'{tp}' 类型字段不支持操作运算")
|
||||
if op_tp not in ("Int", "Float"):
|
||||
raise ValueError(f"操作数类型 '{op_tp}' 不能用于运算，需要数值类型 (Int / Float)")
|
||||
if tp in ("Date", "Time") and op_tp != "Int":
|
||||
raise ValueError(f"'{tp}' 类型的加减法操作数必须为 Int 类型，不允许 Float")
|
||||
target_val = target.getValue(target_data)
|
||||
if target_val is None:
|
||||
raise ValueError(f"'{target.name}' 的值为空，无法进行运算")
|
||||
if op == ".ADD.":
|
||||
ASOperator._arithAdd(target, target_val, operand, target_data)
|
||||
elif op == ".SUB.":
|
||||
ASOperator._arithSub(target, target_val, operand, target_data)
|
||||
else:
|
||||
raise ValueError(f"不支持的操作 '{op}'")
|
||||
|
||||
@staticmethod
|
||||
def _arithAdd(
|
||||
target: ASObject,
|
||||
target_val,
|
||||
operand: ASObject,
|
||||
target_data: dict
|
||||
):
|
||||
"""Dispatch ADD per type."""
|
||||
|
||||
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
|
||||
def _arithSub(
|
||||
target: ASObject,
|
||||
target_val,
|
||||
operand: ASObject,
|
||||
target_data: dict
|
||||
):
|
||||
|
||||
"""Dispatch SUB per type."""
|
||||
|
||||
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
|
||||
def compare(
|
||||
left: ASObject,
|
||||
right: ASObject,
|
||||
op: str,
|
||||
target_data: dict
|
||||
) -> bool:
|
||||
"""
|
||||
Compare two ASObjects using the given comparison operator.
|
||||
|
||||
Args:
|
||||
left (ASObject): Left-hand side.
|
||||
right (ASObject): Right-hand side.
|
||||
op (str): One of ".EQ.", ".NEQ.", ".BGT.", ".BLT.", ".BGE.", ".BLE.".
|
||||
target_data (dict): Application data dict.
|
||||
|
||||
Returns:
|
||||
bool: The comparison result.
|
||||
|
||||
Raises:
|
||||
ValueError: If the types are incompatible for comparison.
|
||||
"""
|
||||
|
||||
cmp_func = ASOperator._COMPARE.get(op)
|
||||
if cmp_func is None:
|
||||
raise ValueError(f"未知的比较操作 '{op}'")
|
||||
left_val = left.getValue(target_data)
|
||||
right_val = right.getValue(target_data)
|
||||
try:
|
||||
return cmp_func(left_val, right_val)
|
||||
except TypeError:
|
||||
raise ValueError(
|
||||
f"无法比较 '{left.name}' ({left.var_type}) "
|
||||
f"与 '{right.name}' ({right.var_type})"
|
||||
)
|
||||
@@ -0,0 +1,55 @@
|
||||
"""
|
||||
AutoScript module for the AutoLibrary project.
|
||||
|
||||
A lightweight scripting DSL for preprocessing user reservation data
|
||||
in repeatable timer tasks. Supports IF/ELSE IF/ELSE/END IF control
|
||||
flow, SET assignments, .ADD./.SUB. operations, and rich comparisons.
|
||||
|
||||
Public API:
|
||||
- execute(script_text, target_data): Execute an AutoScript.
|
||||
- addTargetVar(name, var_type, key_path, display_name): Register a variable.
|
||||
- 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)).
|
||||
"""
|
||||
|
||||
from autoscript.ASEngine import execute, addTargetVar
|
||||
from autoscript.ASObject import _META_VARS as META_VARS
|
||||
|
||||
__all__ = [
|
||||
"execute", "addTargetVar", "registerDefaultTargetVars",
|
||||
"META_VARS", "ALL_VARIABLES",
|
||||
]
|
||||
|
||||
# All variables available to scripts (display_name -> (name, type)).
|
||||
# This mirrors the old AutoScriptEngine.VARIABLE_META for backward
|
||||
# compatibility in the UI orchestration dialog.
|
||||
ALL_VARIABLES: dict = {
|
||||
"用户名": ("USERNAME", "String"),
|
||||
"用户启用": ("USER_ENABLE", "Boolean"),
|
||||
"预约日期": ("RESERVE_DATE", "Date"),
|
||||
"预约开始时间": ("RESERVE_BEGIN_TIME", "Time"),
|
||||
"预约结束时间": ("RESERVE_END_TIME", "Time"),
|
||||
"当前时间": ("CURRENT_TIME", "Time"),
|
||||
"当前日期": ("CURRENT_DATE", "Date"),
|
||||
}
|
||||
|
||||
# Key paths into target_data dict for each target variable.
|
||||
# (name, type, key_path, display_name)
|
||||
_TARGET_VAR_DEFS = [
|
||||
("USERNAME", "String",["username"], "用户名"),
|
||||
("USER_ENABLE", "Boolean",["enabled"], "用户启用"),
|
||||
("RESERVE_DATE", "Date", ["reserve_info", "date"], "预约日期"),
|
||||
("RESERVE_BEGIN_TIME", "Time", ["reserve_info", "begin_time", "time"], "预约开始时间"),
|
||||
("RESERVE_END_TIME", "Time", ["reserve_info", "end_time", "time"], "预约结束时间"),
|
||||
]
|
||||
|
||||
|
||||
def registerDefaultTargetVars() -> None:
|
||||
"""
|
||||
Register all built-in target variables with the engine.
|
||||
This must be called before any script execution.
|
||||
Calling multiple times is idempotent (re-registers same keys).
|
||||
"""
|
||||
for name, var_type, key_path, display_name in _TARGET_VAR_DEFS:
|
||||
addTargetVar(name, var_type, key_path, display_name)
|
||||
@@ -1,386 +0,0 @@
|
||||
# -*- 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
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
class AutoScriptEngine:
|
||||
"""
|
||||
AutoScript script engine.
|
||||
|
||||
Parses and executes AutoScript — a lightweight scripting DSL
|
||||
used in repeatable timer tasks to preprocess user reservation
|
||||
data before automation runs.
|
||||
|
||||
Supports IF/ELSE IF/ELSE/END IF control flow, SET assignments,
|
||||
.ADD./.SUB. operations on Date/Time fields, and rich comparison
|
||||
operators (.EQ. .NEQ. .BGT. .BLT. .BGE. .BLE.).
|
||||
|
||||
Examples:
|
||||
>>> engine = AutoScriptEngine
|
||||
>>> user = {
|
||||
... "username": "test",
|
||||
... "enabled": True,
|
||||
... "reserve_info": {"date": "2026-05-07"}
|
||||
... }
|
||||
>>> engine.execute(
|
||||
... 'IF(CURRENT_TIME .BGT. TIME(19:00))\\n'
|
||||
... ' RESERVE_DATE .ADD. 1\\n'
|
||||
... 'END IF',
|
||||
... user
|
||||
... )
|
||||
"""
|
||||
COMPARE_OPS = { # compare operators
|
||||
".EQ." : lambda a, b: a == b,
|
||||
".NEQ.": lambda a, b: a != b,
|
||||
".BGT.": lambda a, b: a > b,
|
||||
".BLT.": lambda a, b: a < b,
|
||||
".BGE.": lambda a, b: a >= b,
|
||||
".BLE.": lambda a, b: a <= b,
|
||||
}
|
||||
VARIABLE_META = { # variable metadata
|
||||
"预约开始时间": ("RESERVE_BEGIN_TIME", "Time"),
|
||||
"预约结束时间": ("RESERVE_END_TIME", "Time"),
|
||||
"预约日期": ("RESERVE_DATE", "Date"),
|
||||
"用户名": ("USERNAME", "String"),
|
||||
"用户启用": ("USER_ENABLE", "Boolean"),
|
||||
"当前时间": ("CURRENT_TIME", "Time"),
|
||||
"当前日期": ("CURRENT_DATE", "Date"),
|
||||
}
|
||||
_FIELD_TYPE_MAP = {meta[0]: meta[1] for meta in VARIABLE_META.values()}
|
||||
|
||||
@staticmethod
|
||||
def execute(
|
||||
script_text: str,
|
||||
user_data: dict
|
||||
):
|
||||
"""
|
||||
Execute an AutoScript against the given user data.
|
||||
|
||||
The script is parsed line-by-line. All modifications are
|
||||
applied directly to ``user_data`` in-place.
|
||||
|
||||
Args:
|
||||
script_text (str): Raw AutoScript source code.
|
||||
user_data (dict): User data dictionary to read from and
|
||||
write to. Must conform to the standard user profile
|
||||
structure (username, enabled, reserve_info, etc.).
|
||||
|
||||
Raises:
|
||||
ValueError: On any syntax or type error encountered
|
||||
during parsing or execution.
|
||||
"""
|
||||
|
||||
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:
|
||||
return
|
||||
if_stack = []
|
||||
|
||||
for line in lines:
|
||||
upper_line = line.upper().strip()
|
||||
if upper_line.startswith("IF("):
|
||||
cond_end = _findConditionEnd(upper_line)
|
||||
if cond_end < 0:
|
||||
raise ValueError("AutoScript 语法错误: IF 缺少右括号")
|
||||
condition_str = line[3:cond_end].strip()
|
||||
matched = AutoScriptEngine._evaluateCondition(
|
||||
condition_str, user_data
|
||||
)
|
||||
if_stack.append([matched, matched])
|
||||
elif upper_line.startswith("ELSE IF("):
|
||||
if not if_stack:
|
||||
raise ValueError("AutoScript 语法错误: ELSE IF 前缺少 IF")
|
||||
cond_end = _findConditionEnd(upper_line)
|
||||
if cond_end < 0:
|
||||
raise ValueError("AutoScript 语法错误: ELSE IF 缺少右括号")
|
||||
condition_str = line[8:cond_end].strip()
|
||||
_, has_matched = if_stack[-1]
|
||||
if not has_matched:
|
||||
matched = AutoScriptEngine._evaluateCondition(
|
||||
condition_str, user_data
|
||||
)
|
||||
if_stack[-1] = [matched, matched]
|
||||
else:
|
||||
if_stack[-1][0] = False
|
||||
elif upper_line == "ELSE":
|
||||
if not if_stack:
|
||||
raise ValueError("AutoScript 语法错误: ELSE 前缺少 IF")
|
||||
_, has_matched = if_stack[-1]
|
||||
if not has_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("AutoScript 语法错误: ENDIF/END IF 前缺少 IF")
|
||||
if_stack.pop()
|
||||
elif upper_line.startswith("SET "):
|
||||
should_execute = (
|
||||
all(ctx[0] for ctx in if_stack) if if_stack else True
|
||||
)
|
||||
if should_execute:
|
||||
AutoScriptEngine._executeSet(line, user_data)
|
||||
elif upper_line == "PASS":
|
||||
continue
|
||||
else:
|
||||
should_execute = (
|
||||
all(ctx[0] for ctx in if_stack) if if_stack else True
|
||||
)
|
||||
if should_execute:
|
||||
AutoScriptEngine._executeOperation(line, user_data)
|
||||
if if_stack:
|
||||
raise ValueError("AutoScript 语法错误: IF 与 ENDIF/END IF 不匹配")
|
||||
|
||||
@staticmethod
|
||||
def _resolveField(
|
||||
field_name: str,
|
||||
user_data: dict
|
||||
):
|
||||
|
||||
upper_name = field_name.upper().strip()
|
||||
if upper_name == "CURRENT_DATE":
|
||||
return datetime.now().strftime("%Y-%m-%d")
|
||||
elif upper_name == "CURRENT_TIME":
|
||||
return datetime.now().strftime("%H:%M")
|
||||
elif upper_name == "USERNAME":
|
||||
return user_data.get("username", "")
|
||||
elif upper_name == "USER_ENABLE":
|
||||
return user_data.get("enabled", False)
|
||||
elif upper_name == "RESERVE_DATE":
|
||||
return user_data.get("reserve_info", {}).get("date", "")
|
||||
elif upper_name == "RESERVE_BEGIN_TIME":
|
||||
return (
|
||||
user_data
|
||||
.get("reserve_info", {})
|
||||
.get("begin_time", {})
|
||||
.get("time", "")
|
||||
)
|
||||
elif upper_name == "RESERVE_END_TIME":
|
||||
return (
|
||||
user_data
|
||||
.get("reserve_info", {})
|
||||
.get("end_time", {})
|
||||
.get("time", "")
|
||||
)
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _resolveValue(
|
||||
value_str: str,
|
||||
user_data: dict
|
||||
):
|
||||
|
||||
s = value_str.strip()
|
||||
time_match = re.match(r"^TIME\((\d{1,2}):(\d{2})\)$", s, re.IGNORECASE)
|
||||
if time_match:
|
||||
h, m = time_match.group(1), time_match.group(2)
|
||||
return f"{int(h):02d}:{int(m):02d}"
|
||||
date_match = re.match(r"^DATE\((\d{4})-(\d{2})-(\d{2})\)$", s, re.IGNORECASE)
|
||||
if date_match:
|
||||
y, mo, d = date_match.group(1), date_match.group(2), date_match.group(3)
|
||||
return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
|
||||
if s.upper() == ".TRUE.":
|
||||
return True
|
||||
if s.upper() == ".FALSE.":
|
||||
return False
|
||||
if s.startswith("'") and s.endswith("'"):
|
||||
inner = s[1:-1].replace("''", "'")
|
||||
return inner
|
||||
if s.startswith('"') and s.endswith('"'):
|
||||
return s[1:-1]
|
||||
relDate = re.match(r"^CURRENT_DATE\s*\+\s*(\d+)$", s, re.IGNORECASE)
|
||||
if relDate:
|
||||
days = int(relDate.group(1))
|
||||
return (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d")
|
||||
relTime = re.match(r"^CURRENT_TIME\s*\+\s*(\d+)$", s, re.IGNORECASE)
|
||||
if relTime:
|
||||
hours = int(relTime.group(1))
|
||||
return (datetime.now() + timedelta(hours=hours)).strftime("%H:%M")
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
return float(s)
|
||||
except ValueError:
|
||||
pass
|
||||
resolved = AutoScriptEngine._resolveField(s, user_data)
|
||||
return resolved
|
||||
|
||||
@staticmethod
|
||||
def _setField(
|
||||
field_name: str,
|
||||
value: str,
|
||||
user_data: dict
|
||||
):
|
||||
upper_name = field_name.upper().strip()
|
||||
if upper_name == "RESERVE_DATE":
|
||||
user_data.setdefault("reserve_info", {})["date"] = value
|
||||
elif upper_name == "RESERVE_BEGIN_TIME":
|
||||
ri = user_data.setdefault("reserve_info", {})
|
||||
ri.setdefault("begin_time", {})["time"] = value
|
||||
elif upper_name == "RESERVE_END_TIME":
|
||||
ri = user_data.setdefault("reserve_info", {})
|
||||
ri.setdefault("end_time", {})["time"] = value
|
||||
elif upper_name == "USERNAME":
|
||||
user_data["username"] = value
|
||||
elif upper_name == "USER_ENABLE":
|
||||
if isinstance(value, bool):
|
||||
user_data["enabled"] = value
|
||||
else:
|
||||
user_data["enabled"] = (str(value).upper() == "TRUE")
|
||||
|
||||
@staticmethod
|
||||
def _evaluateCondition(
|
||||
condition_str: str,
|
||||
user_data: dict
|
||||
) -> bool:
|
||||
|
||||
for op, cmp_func in AutoScriptEngine.COMPARE_OPS.items():
|
||||
if op not in condition_str.upper():
|
||||
continue
|
||||
idx = condition_str.upper().find(op)
|
||||
parts = [condition_str[:idx], condition_str[idx + len(op):]]
|
||||
if len(parts) != 2:
|
||||
continue
|
||||
field_name = parts[0].strip()
|
||||
value_str = parts[1].strip()
|
||||
left_val = AutoScriptEngine._resolveField(field_name, user_data)
|
||||
right_val = AutoScriptEngine._resolveValue(value_str, user_data)
|
||||
try:
|
||||
return cmp_func(left_val, right_val)
|
||||
except TypeError:
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: 无法比较 "
|
||||
f"'{field_name}' ({type(left_val).__name__}) "
|
||||
f"与 '{value_str}' ({type(right_val).__name__})"
|
||||
)
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _executeSet(
|
||||
line: str,
|
||||
user_data: dict
|
||||
):
|
||||
rest = line[3:].strip()
|
||||
eq_idx = rest.find("=")
|
||||
if eq_idx < 0:
|
||||
return
|
||||
field_name = rest[:eq_idx].strip()
|
||||
value_str = rest[eq_idx + 1:].strip()
|
||||
if not field_name:
|
||||
return
|
||||
resolved = AutoScriptEngine._resolveValue(value_str, user_data)
|
||||
AutoScriptEngine._setField(field_name, resolved, user_data)
|
||||
|
||||
@staticmethod
|
||||
def _executeOperation(
|
||||
line: str,
|
||||
user_data: dict
|
||||
):
|
||||
|
||||
parts = line.split()
|
||||
if len(parts) < 3:
|
||||
return
|
||||
field_name = parts[0].upper().strip()
|
||||
op = parts[1].upper().strip()
|
||||
raw_value = parts[2].strip()
|
||||
field_type = AutoScriptEngine._FIELD_TYPE_MAP.get(field_name)
|
||||
if not field_type:
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: 未知字段 '{field_name}'"
|
||||
)
|
||||
try:
|
||||
num_value = float(raw_value) if "." in raw_value else int(raw_value)
|
||||
except (ValueError, TypeError):
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: 无效操作数 '{raw_value}'"
|
||||
)
|
||||
if field_type == "Date":
|
||||
date_str = AutoScriptEngine._resolveField(field_name, user_data)
|
||||
if not date_str:
|
||||
return
|
||||
try:
|
||||
date_obj = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
except (ValueError, TypeError):
|
||||
return
|
||||
if op == ".ADD.":
|
||||
date_obj += timedelta(days=num_value)
|
||||
elif op == ".SUB.":
|
||||
date_obj -= timedelta(days=num_value)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: Date 类型不支持操作 '{op}'"
|
||||
)
|
||||
AutoScriptEngine._setField(
|
||||
field_name, date_obj.strftime("%Y-%m-%d"), user_data
|
||||
)
|
||||
elif field_type == "Time":
|
||||
time_str = AutoScriptEngine._resolveField(field_name, user_data)
|
||||
if not time_str:
|
||||
return
|
||||
try:
|
||||
time_obj = datetime.strptime(time_str, "%H:%M")
|
||||
except (ValueError, TypeError):
|
||||
return
|
||||
if op == ".ADD.":
|
||||
time_obj += timedelta(hours=num_value)
|
||||
elif op == ".SUB.":
|
||||
time_obj -= timedelta(hours=num_value)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: Time 类型不支持操作 '{op}'"
|
||||
)
|
||||
AutoScriptEngine._setField(
|
||||
field_name, time_obj.strftime("%H:%M"), user_data
|
||||
)
|
||||
elif field_type in ("String", "Boolean"):
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: '{field_type}' 类型字段不支持操作运算"
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"AutoScript 语法错误: 未知字段类型 '{field_type}'"
|
||||
)
|
||||
|
||||
|
||||
def _findConditionEnd(
|
||||
upper_line: str
|
||||
) -> int:
|
||||
"""
|
||||
Find the index of the closing parenthesis that matches the
|
||||
opening parenthesis in a condition expression, handling nested
|
||||
parentheses and optional ``THEN`` keyword.
|
||||
|
||||
Args:
|
||||
upper_line (str): The uppercased line text containing the
|
||||
condition, e.g. ``"IF(A .BGT. B) THEN"``.
|
||||
|
||||
Returns:
|
||||
int: Index of the matching ``)``, or ``-1`` if no match
|
||||
is found.
|
||||
"""
|
||||
|
||||
line = upper_line.rstrip()
|
||||
if line.endswith(" THEN"):
|
||||
line = line[:-5].rstrip()
|
||||
paren_depth = 0
|
||||
start_found = False
|
||||
for i, ch in enumerate(line):
|
||||
if ch == "(":
|
||||
paren_depth += 1
|
||||
start_found = True
|
||||
elif ch == ")":
|
||||
paren_depth -= 1
|
||||
if start_found and paren_depth == 0:
|
||||
return i
|
||||
return -1
|
||||
@@ -1,9 +0,0 @@
|
||||
"""
|
||||
DSL module for the AutoLibrary project.
|
||||
|
||||
Contains the AutoScript DSL engine and related components
|
||||
for preprocessing user reservation data in timer tasks.
|
||||
|
||||
Classes:
|
||||
- AutoScriptEngine: AutoScript script engine class.
|
||||
"""
|
||||
Reference in New Issue
Block a user