✨ simple interrupt of Analyser

entry_test.py

@@ -9,3 +9,5 @@
 if __name__ == '__main__':
     import pytest
     pytest.main([__file__, "-v"])
+
+

src/arclet/alconna/analysis/analyser.py

@@ -2,12 +2,16 @@
 import traceback
 from weakref import finalize
 from copy import copy
-from typing import Dict, Union, List, Optional, TYPE_CHECKING, Tuple, Any, Generic, TypeVar, Set, Callable, ClassVar
+from typing import (
+    Dict, Union, List, Optional, TYPE_CHECKING, Tuple, Any, Generic, TypeVar, Set, Callable, ClassVar
+)
 from nepattern import pattern_map, type_parser, BasePattern
 from nepattern.util import TPattern
 
 from ..manager import command_manager
-from ..exceptions import NullMessage, ParamsUnmatched, ArgumentMissing, FuzzyMatchSuccess, CompletionTriggered
+from ..exceptions import (
+    NullMessage, ParamsUnmatched, ArgumentMissing, FuzzyMatchSuccess, CompletionTriggered, PauseTriggered
+)
 from ..args import Args, ArgUnit
 from ..base import Option, Subcommand, Sentence, StrMounter
 from ..arpamar import Arpamar
@@ -194,6 +198,20 @@ def reset(self):
         self.origin_data, self.header, self.context = None, None, None
         self.head_pos = (0, 0)
 
+    def push(self, *data: Union[str, Any]):
+        for d in data:
+            if not d:
+                continue
+            if isinstance(d, str) and isinstance(self.raw_data[-1], StrMounter):
+                if self.current_index == self.ndata:
+                    self.current_index -= 1
+                    self.content_index = len(self.raw_data[-1]) - 1
+                self.raw_data[-1].append(d)
+            else:
+                self.raw_data.append(StrMounter([d]) if isinstance(d, str) else d)
+                self.ndata += 1
+        return self
+
     def popitem(self, separate: Optional[Set[str]] = None, move: bool = True) -> Tuple[Union[str, Any], bool]:
         """获取解析需要的下个数据"""
         if self.current_index == self.ndata:
@@ -288,7 +306,11 @@ def process(self, data: DataCollection[Union[str, Any]]) -> 'Analyser':
         "--shortcut": handle_shortcut, "-sct": handle_shortcut
     }
 
-    def analyse(self, message: Union[DataCollection[Union[str, Any]], None] = None) -> Arpamar:
+    def analyse(
+        self,
+        message: Union[DataCollection[Union[str, Any]], None] = None,
+        interrupt: bool = False
+    ) -> Arpamar:
         """主体解析函数, 应针对各种情况进行解析"""
         if command_manager.is_disable(self.alconna):
             return self.export(fail=True)
@@ -362,6 +384,8 @@ def analyse(self, message: Union[DataCollection[Union[str, Any]], None] = None)
                         return handle_completion(self, self.context)
                     if handler := self._special.get(rest[-1]):
                         return handler(self)
+                if interrupt and isinstance(e1, ArgumentMissing):
+                    raise PauseTriggered from e1
                 if self.raise_exception:
                     raise
                 return self.export(fail=True, exception=e1)
@@ -382,6 +406,8 @@ def analyse(self, message: Union[DataCollection[Union[str, Any]], None] = None)
             exc = ParamsUnmatched(config.lang.analyser_param_unmatched.format(target=self.popitem(move=False)[0]))
         else:
             exc = ArgumentMissing(config.lang.analyser_param_missing)
+        if interrupt and isinstance(exc, ArgumentMissing):
+            raise PauseTriggered
         if self.raise_exception:
             raise exc
         return self.export(fail=True, exception=exc)

src/arclet/alconna/analysis/base.py

@@ -22,7 +22,7 @@ def _compile_opts(option: Option, data: Dict[str, Union[Sentence, List[Option]]]
             data[alias] = [option]
 
 
-def default_params_compiler(analyser: "Analyser"):
+def default_params_parser(analyser: "Analyser"):
     require_len = 0
     for opts in analyser.alconna.options:
         if isinstance(opts, Option):
@@ -52,9 +52,9 @@ def default_params_compiler(analyser: "Analyser"):
         )
 
 
-def compile(alconna: "Alconna", params_compiler: Callable[[Analyser], None] = default_params_compiler) -> Analyser:
+def compile(alconna: "Alconna", params_parser: Callable[[Analyser], None] = default_params_parser) -> Analyser:
     _analyser = alconna.analyser_type(alconna)
-    params_compiler(_analyser)
+    params_parser(_analyser)
     return _analyser
 
 
@@ -130,7 +130,7 @@ def analyse_option(option: Option, command: DataCollection[Union[str, Any]], rai
     _analyser.need_main_args = False
     _analyser.raise_exception = True
     _analyser.alconna.options.append(option)
-    default_params_compiler(_analyser)
+    default_params_parser(_analyser)
     _analyser.alconna.options.clear()
     try:
         _analyser.process(command)
@@ -148,7 +148,7 @@ def analyse_subcommand(subcommand: Subcommand, command: DataCollection[Union[str
     _analyser.need_main_args = False
     _analyser.raise_exception = True
     _analyser.alconna.options.append(subcommand)
-    default_params_compiler(_analyser)
+    default_params_parser(_analyser)
     _analyser.alconna.options.clear()
     try:
         _analyser.process(command)

src/arclet/alconna/core.py

@@ -12,6 +12,7 @@
 from .typing import TDataCollection
 from .manager import command_manager
 from .arpamar import Arpamar
+from .exceptions import PauseTriggered
 from .analysis.analyser import TAnalyser, Analyser
 from .components.action import ActionHandler, ArgAction
 from .components.output import TextFormatter
@@ -349,24 +350,33 @@ def add(self, name: str, *alias: str, args: Optional[Args] = None, sep: str = "
 
     @overload
     def parse(
-            self, message: TDataCollection, duplication: Type[T_Duplication], static: bool = True,
+        self, message: TDataCollection, duplication: Type[T_Duplication], static: bool = True, interrupt: bool = False
     ) -> T_Duplication:
         ...
 
     @overload
     def parse(
-            self, message: TDataCollection, duplication=None, static: bool = True
+        self, message: TDataCollection, duplication=None, static: bool = True, interrupt: bool = False
     ) -> Arpamar[TDataCollection]:
         ...
 
+    @overload
     def parse(
-            self, message: TDataCollection, duplication: Optional[Type[T_Duplication]] = None,
-            static: bool = True,
+        self, message: TDataCollection, duplication=..., static: bool = True, interrupt: bool = True
+    ) -> TAnalyser:
+        ...
+
+    def parse(
+        self, message: TDataCollection, duplication: Optional[Type[T_Duplication]] = None,
+        static: bool = True, interrupt: bool = False
     ):
         """命令分析功能, 传入字符串或消息链, 返回一个特定的数据集合类"""
         analyser = command_manager.require(self) if static else compile(self)
         analyser.process(message)
-        arp = analyser.analyse()
+        try:
+            arp: Arpamar[TDataCollection] = analyser.analyse(interrupt=interrupt)
+        except PauseTriggered:
+            return analyser
         if arp.matched:
             arp = arp.execute()
         if duplication:

src/arclet/alconna/exceptions.py

@@ -43,3 +43,7 @@ class FuzzyMatchSuccess(Exception):
 
 class CompletionTriggered(Exception):
     """补全触发"""
+
+
+class PauseTriggered(Exception):
+    """解析状态保存触发"""

test_alconna/core_test.py

@@ -342,6 +342,14 @@ def test_completion():
     alc20_1.parse("core20_1 -cp")
 
 
+def test_interrupt():
+    alc21 = Alconna("core21", Args.foo[int], Args.bar[str])
+    print("\n", alc21.parse("core21"))
+    print("\n", ana := alc21.parse("core21", interrupt=True))
+
+    assert ana.push("1", "a").analyse().matched
+
+
 if __name__ == "__main__":
     import pytest