feat: Input stream detective for builtin command

example/plugins/HydroRoll/__init__.py

@@ -10,6 +10,7 @@
 from .config import Directory, GlobalConfig, Models
 from .utils import *
 from .models.Transformer import query
+from .command import Set, Get
 from iamai.exceptions import GetEventTimeout
 
 BASE_DIR = dirname(abspath("__file__"))
@@ -48,27 +49,46 @@ async def handle(self) -> None:
         @BODY: HydroRollCore actives the rule-packages.
         """
         args = self.event.get_plain_text().split(" ")
-        try:
-            event = await self.event.adapter.get(
-                lambda x: x.type == "message", timeout=10
-            )
-        except GetEventTimeout:
-            return
-        else:
-            if args[0] == ".core":
-                ...
-            if args[0].startswith(".set"):
-                ...
-            elif args[0].startswith(".get"):
-                ...
-            elif args[0].startswith(".test"):
-                try:
-                    result = eval(
-                        self.event.message.get_plain_text()[5:]
-                    )  # literal_eval(self.event.message.get_plain_text()[5:])
-                    await self.event.reply(str(result))
-                except Exception as error:
-                    await self.event.reply(f"{error!r}")
+        command_list = [".root", ".roots", ".core", ".set", ".get", ".test"]
+        for cmd in command_list:
+            if cmd.startswith(args[0]):
+                logger.info(cmd)
+                if args[0] != cmd:
+                    try:
+                        flag = True
+                        current_cmd = args[0]
+                        while flag:
+                            flag = False
+                            event = await self.ask(ask_text=None, timeout=3)
+                            current_cmd = current_cmd + event.get_plain_text()
+                            if cmd.startswith(current_cmd):
+                                if current_cmd != cmd:
+                                    flag = True
+                                else:
+                                    await self.event.reply(f"{cmd}")
+                                    flag = False
+                    except GetEventTimeout:
+                        continue
+                else:
+                    await self.event.reply(cmd)
+        # if args[0] in [".root", ".roots"]:
+        #     import requests
+
+        #     data = requests.get("https://vercel-hitokoto.vercel.app/api/roots").json()
+        #     await self.event.reply(data["line"])
+        # else:
+        #     if args[0] == ".core":
+        #         ...
+        #     if args[0].startswith(".set"):
+        #         resolve = Set(args[1:])  # TODO: handle multiple sets
+        #     elif args[0].startswith(".get"):
+        #         resolve = Get(args[1:])  # TODO: handle multiple gets
+        #     elif args[0].startswith(".test"):
+        #         try:
+        #             result = eval(self.event.message.get_plain_text()[5:])
+        #             await self.event.reply(str(result))
+        #         except Exception as error:
+        #             await self.event.reply(f"{error!r}")
 
     async def rule(self) -> bool:
         """
@@ -124,3 +144,17 @@ def _load_models(self, model_path_list: list, model_dict: dict) -> dict:
 
     def load_models(self):
         self.models = self._load_models(self.model_path_list, self.model_dict)
+
+    async def ask(self, ask_text: str | None, timeout: int = 10) -> None:
+        if ask_text:
+            await self.event.reply(ask_text)
+        try:
+            event = await self.event.adapter.get(
+                lambda x: x.type == "message"
+                and x.group_id == self.event.group_id
+                and x.user_id == self.event.user_id,
+                timeout=timeout,
+            )
+            return event
+        except GetEventTimeout as e:
+            raise GetEventTimeout from e

example/plugins/HydroRoll/command.py

@@ -0,0 +1,9 @@
+from .utils import CommandPluginBase
+
+class Get(CommandPluginBase):
+    ...
+
+
+
+class Set(CommandPluginBase):
+    ...

example/plugins/HydroRoll/config.py

@@ -41,7 +41,7 @@ class RegexPluginConfig(BasePluginConfig):
 
 
 class CommandPluginConfig(RegexPluginConfig):
-    command_prefix: Set[str] = {":", "你妈", "👅", "约瑟夫妥斯妥耶夫斯基戴安那只鸡🐔"}
+    command_prefix: Set[str] = {":"}
     """命令前缀。"""
     command: Set[str] = {}
     """命令文本。"""
@@ -146,3 +146,4 @@ def __init__(self) -> None:
 
     def get_models_dict(self) -> dict:
         return self.builtin_models
+

example/plugins/HydroRoll/utils.py

@@ -20,7 +20,7 @@ class BasePlugin(
     ABC,
     Generic[T_State, T_Config],
 ):
-    Config: Type[T_Config] = BasePluginConfig
+    Config: Type[T_Config] = BasePluginConfig # type: ignore
 
     def format_str(self, format_str: str, message_str: str = "") -> str:
         return format_str.format(
@@ -60,9 +60,6 @@ async def rule(self) -> bool:
                     or self.event.group_id in self.config.accept_group
                 ):
                     return self.str_match(match_str)
-        elif self.config.handle_group_message:
-            if self.event.message_type == "guild":
-                return self.str_match(match_str)
         return False
 
     @abstractmethod
@@ -153,14 +150,12 @@ def roll_dice(
 
         if streamline:
             return str(total)
-        else:
-            if len(rolls) > int(threshold):
-                return str(total)
-            rolls_str = " + ".join(str(r) for r in rolls)
-            result_str = (
-                f"{total} = {rolls_str}" if is_reversed else f"{rolls_str} = {total}"
-            )
-            return result_str
+        if len(rolls) > int(threshold):
+            return str(total)
+        rolls_str = " + ".join(str(r) for r in rolls)
+        return (
+            f"{total} = {rolls_str}" if is_reversed else f"{rolls_str} = {total}"
+        )
 
 
 def find_max_similarity(input_string, string_list):