refactor(autoscript): 使用观察者模式解耦解析与预检查/编排流程

src/autoscript/ASObserver.py

@@ -0,0 +1,78 @@
+# -*- 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.
+"""
+
+
+class ParsingObserver:
+    """
+        Base observer for AutoScript parsing events.
+
+        Subclass and override the relevant methods to react to
+        tokenization / parsing events produced by ASTokenizer.
+        This is the core abstraction that lets pre-check and
+        orchestration modules subscribe to the same parsing pipeline.
+    """
+
+    def onParseStart(
+        self,
+        script_text: str
+    ):
+        """
+            Called when tokenization of a new script begins.
+
+            Args:
+                script_text (str): The full script text being parsed.
+        """
+        pass
+
+    def onTokenParsed(
+        self,
+        kind: str | None,
+        data,
+        line_num: int,
+        raw_line: str
+    ):
+        """
+            Called after each script line has been classified as a token.
+
+            Args:
+                kind (str | None): Token kind (K_IF, K_ELSE_IF, K_ELSE,
+                                   K_ENDIF, K_SET, K_ADD, K_SUB) or
+                                   None if unrecognised.
+                data:      Token payload — condition string for IF/ELSE IF,
+                           (target, value) tuple for SET/ADD/SUB,
+                           None for ELSE/ENDIF/unrecognised.
+                line_num (int): 1-based line number.
+                raw_line (str): The stripped raw line text.
+        """
+        pass
+
+    def onParseComplete(
+        self,
+        statements: list
+    ):
+        """
+            Called when flat tokenization is complete.
+
+            Args:
+                statements (list[Stmt]): The list of parsed Stmt objects.
+        """
+        pass
+
+    def onASTReady(
+        self,
+        ast
+    ):
+        """
+            Called when full AST construction is complete (after parse()).
+
+            Args:
+                ast (Script): The root Script AST node.
+        """
+        pass

src/autoscript/ASTokenizer.py

@@ -326,12 +326,27 @@ class ASTokenizer:
     """
         Tokenizer / parser for the AutoScript DSL.
 
-        Provides three entry points:
+        Main class-level entry points (engine-facing):
           - classifyLine(line)   — single-line classifier.
           - tokenize(script)      — flat Stmt list.
           - parse(script)         — structured AST (Script root).
+
+        Observer-enabled API (used by pre-check & orchestration):
+          >>> obs = ScriptPrecheckObserver()
+          >>> stmts = ASTokenizer.tokenizeWithObservers(script, [obs])
     """
 
+    @classmethod
+    def _notifyObservers(
+        cls,
+        observers: list,
+        method: str,
+        *args
+    ):
+
+        for obs in observers:
+            getattr(obs, method)(*args)
+
     @classmethod
     def _matchLine(
         cls,
@@ -345,30 +360,14 @@ class ASTokenizer:
         return (None, None)
 
     @classmethod
-    def classifyLine(
+    def _buildStmt(
         cls,
-        stripped: str
-    ):
+        stripped: str,
+        kind: str | None,
+        data
+    ) -> Stmt:
 
-        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:
@@ -380,16 +379,29 @@ class ASTokenizer:
         elif kind == K_SUB:
             stmt.target, stmt.value = data
             stmt.op_type = OP_SUB
-            statements.append(stmt)
+        return stmt
+
+    @classmethod
+    def _tokenizeImpl(
+        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)
+            statements.append(cls._buildStmt(stripped, kind, data))
         return statements
 
     @classmethod
-    def parse(
+    def _parseTokens(
         cls,
-        script: str
+        tokens: list
     ) -> Script:
 
-        tokens = cls.tokenize(script)
         body = []
         i = 0
         while i < len(tokens):
@@ -420,6 +432,77 @@ class ASTokenizer:
                 i += 1
         return Script(body=body)
 
+    @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:
+
+        return cls._tokenizeImpl(script)
+
+    @classmethod
+    def parse(
+        cls,
+        script: str
+    ) -> Script:
+
+        return cls._parseTokens(cls._tokenizeImpl(script))
+
+    @classmethod
+    def tokenizeWithObservers(
+        cls,
+        script: str,
+        observers: list
+    ) -> list:
+        """
+            Tokenize and notify observers for each classified line.
+
+            Fires onParseStart, onTokenParsed, and onParseComplete
+            events to each observer.  This is the single tokenization
+            pipeline shared by pre-check and orchestration modules.
+        """
+
+        cls._notifyObservers(observers, "onParseStart", script)
+        statements = []
+        for i, raw_line in enumerate(script.split("\n"), 1):
+            stripped = raw_line.strip()
+            if not stripped:
+                continue
+            kind, data = cls._matchLine(stripped)
+            cls._notifyObservers(observers, "onTokenParsed", kind, data, i, stripped)
+            statements.append(cls._buildStmt(stripped, kind, data))
+        cls._notifyObservers(observers, "onParseComplete", statements)
+        return statements
+
+    @classmethod
+    def parseWithObservers(
+        cls,
+        script: str,
+        observers: list
+    ) -> Script:
+        """
+            Parse and notify observers throughout the pipeline.
+
+            Calls tokenizeWithObservers (which fires per-token events),
+            then builds the AST and fires onASTReady.
+        """
+
+        tokens = cls.tokenizeWithObservers(script, observers)
+        ast = cls._parseTokens(tokens)
+        cls._notifyObservers(observers, "onASTReady", ast)
+        return ast
+
     @classmethod
     def _parseIfBlock(
         cls,

src/autoscript/__init__.py

@@ -1,17 +1,6 @@
 """
     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)).
-    - ASTokenizer: Unified tokenizer for the orchestration dialog and engine.
+    A lightweight scripting DSL for preprocessing user reservation data.
 """
 from autoscript.ASTokenizer import (
     ASTokenizer,
@@ -27,6 +16,7 @@ from autoscript.ASEngine import (
     addTargetVar,
 )
 from autoscript.ASObject import _META_VARS as META_VARS
+from autoscript.ASObserver import ParsingObserver
 
 __all__ = [
     "execute",
@@ -41,7 +31,8 @@ __all__ = [
     "IfNode",
     "SetNode",
     "OpNode",
-    "ElifNode"
+    "ElifNode",
+    "ParsingObserver",
 ]
 
 # All variables available to scripts (display_name -> (name, type)).