refactor(command): 命令词法解析器 (#82)

* refactor(command): 重构命令路由

* feat(lua): 包装异步方法`self.event.reply` => `msg:echo`

* feat(lua): 包装异步输入流方法`self.event.ask` => `msg:ask`

* 'Refactored by Sourcery' (#83)

Co-authored-by: Sourcery AI <>

* feat(Token|Lexer): 添加`Token`与`Lexer`类

* refactor(Lexer): 词法分析器添加`advance`方法

* chore: lint code

* refactor: sync gensokyo adapter

* feat: Cli parser (#85)

* feat(cli): 添加`Cli`类，解析命令行参数

* fix: 修复错误的`dest`与`action`

* feat(cli): 实现`install_package` 与 `build_template`

* feat(cli): 实现`-c|--config`指令配置镜像常量等

* feat(cli): 使用高效率的异步网络库`aiohttp`

* fix(cli): `TYPE_CHECKING` with partially module `typing`

* refactor!: examples, tests, src... rewrite in rust

---------

Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>

.gitignore

@@ -108,4 +108,5 @@ docs/.next/
 
 # example
 example/config.toml
-.pdm-python
+.pdm-python
+bottles.db

examples/config.toml

@@ -0,0 +1,45 @@
+[bot]
+plugins = []
+plugin_dirs = ["plugins"]
+rules = []
+rule_dirs = ["rules"]
+adapters = [
+    "iamai.adapter.onebot11",
+    # "iamai.adapter.gensokyo",
+    # "iamai.adapter.apscheduler",
+    # "iamai.adapter.dingtalk"        
+]
+
+[bot.log]
+level = "INFO"
+verbose_exception = true
+
+[adapter.onebot11]
+adapter_type = "reverse-ws"
+host = "127.0.0.1"
+port = 8080
+url = "/cqhttp/ws"
+show_raw = true
+
+[adapter.gensokyo]
+adapter_type = "reverse-ws"
+host = "127.0.0.1"
+port = 8081
+url = "/gsk/ws"
+show_raw = true
+
+[adapter.dingtalk]
+adapter_type = "stream"
+host = "127.0.0.1"
+port = 15700
+url = "/dingtalk"
+app_secret = "FnQU_a88xRpmcs3oPNXSgoQgm4TidGduVqKhLHR7_NgF6MLBUUbwYdE6MkOFWZFb"
+app_key = "dingo7xu5djthkxpoick"
+
+[adapter.apscheduler]
+scheduler_config = { "apscheduler.timezone" = "Asia/Shanghai" }
+
+[plugin.HydroRoll]
+uid = ''
+rules = []
+rule_dirs = ["rules"]

example/plugins/HydroRoll/__init__.py → examples/plugins/HydroRoll/__init__.py

@@ -5,6 +5,7 @@
 import os
 import shutil
 
+import oneroll
 from iamai import ConfigModel, Plugin
 from iamai.log import logger
 from iamai.exceptions import GetEventTimeout
@@ -14,7 +15,7 @@
 from .config import Directory, GlobalConfig, Models
 from .utils import *
 from .models.Transformer import query
-from .command import Set, Get
+
 from .config import (
     BasePluginConfig,
     CommandPluginConfig,
@@ -45,8 +46,8 @@ class Dice(Plugin[MessageEvent, Annotated[dict, {}], RegexPluginConfig]):
 
     priority = 0
 
-    # TODO: HydroRollCore should be able to handle all signals and tokens from Psi.
-    logger.info("Loading HydroRollCore...")
+    # TODO: infini should be able to handle all signals and tokens from Psi.
+    logger.info("Loading infini...")
 
     def __post_init__(self):
         self.state = {}
@@ -62,22 +63,36 @@ def __post_init__(self):
 
     async def handle(self) -> None:
         """
-        @TODO: HydroRollCore should be able to handle all signals and tokens from Psi.
-        @BODY: HydroRollCore actives the rule-packages.
+        @TODO: infini should be able to handle all signals and tokens from Psi.
+        @BODY: infini actives the rule-packages.
         """
         global flag
 
         args = self.event.get_plain_text().split(" ")
-        command_list = [".root", ".roots", ".core", ".set", ".get", ".test"]
+        command_list = ["/r", ".root", ".roots", ".core", ".set", ".get", ".test"]
         current_cmd = args[0]
+        text = (
+            self.event.get_plain_text()[2:]
+            if len(self.event.get_plain_text()) >= 2
+            else None
+        )
         flag = True in [cmd.startswith(current_cmd) for cmd in command_list]
-        logger.info(f"Command {current_cmd} not found with flag {flag}")
+        # logger.info(f"Command {current_cmd} not found with flag {flag}")
+        logger.info(f"text: {text}")
+        if text and self.event.get_plain_text().startswith("/r"):
+            logger.info(text)
+            try:
+                await self.event.reply(f"{oneroll.roll(text)}")
+            except Exception as e:
+                await self.event.reply(f"{e!r}")
         if args[0] in [".root", ".roots"]:
             try:
                 import aiohttp
 
                 async with aiohttp.ClientSession() as session:
-                    async with session.get("https://api.hydroroll.team/api/roots") as response:
+                    async with session.get(
+                        "https://api.hydroroll.team/api/roots"
+                    ) as response:
                         data = await response.json()
                         await self.event.reply(data["line"])
             except Exception as e:

examples/plugins/HydroRoll/command/__init__.py

@@ -0,0 +1,74 @@
+SET, GET, ALIAS = ("SET", "GET", "ALIAS")
+INTEGER = "INTEGER"
+EOF = "EOF"
+
+class Token(object):
+    def __init__(self, type, value):
+        self.type = type
+        self.value = value
+
+    def __str__(self):
+        return f"Token({self.type}, {self.value}"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+
+class Lexer(object):
+    def __init__(self, text):
+        self.text = text
+        self.pos = 0
+        self.current_char = self.text[self.pos]
+
+    def error(self):
+        raise Exception("Invalid Character")
+
+    def advance(self):
+        """Advance the `pos` pointer and set the `current_char` variable."""
+        self.pos += 1
+        if self.pos > len(self.text) - 1:
+            self.current_char = None  # Indicates end of input
+        else:
+            self.current_char = self.text[self.pos]
+
+    def skip_whitespace(self):
+        while self.current_char is not None and self.current_char.isspace():
+            self.advance()
+
+    def integer(self):
+        """Return a (multidigit) integer consumed from the input."""
+        result = ""
+        while self.current_char is not None and self.current_char.isdigit():
+            result += self.current_char
+            self.advance()
+        return int(result)
+
+    def get_next_token(self):
+        """Lexical analyzer (also known as scanner or tokenizer)
+
+        This method is responsible for breaking a sentence
+        apart into tokens. One token at a time.
+        """
+        while self.current_char is not None:
+            if self.current_char.isspace():
+                self.skip_whitespace()
+                continue
+
+            if self.current_char.isdigit():
+                return Token(INTEGER, self.integer())
+
+            if self.current_char == "get":
+                self.advance()
+                return Token(GET, "get")
+
+            if self.current_char == "set":
+                self.advance()
+                return Token(SET, "set")
+
+            if self.current_char == "Alias":
+                self.advance()
+                return Token(ALIAS, "Alias")
+
+            self.error()
+
+        return Token(EOF, None)

examples/plugins/HydroRoll/command/alias_command.py

@@ -0,0 +1,4 @@
+from .typing import CommandBase
+
+class AliasCommand(CommandBase):
+    ...

examples/plugins/HydroRoll/command/get_command.py

@@ -0,0 +1,4 @@
+from .typing import CommandBase
+
+class GetCommand(CommandBase):
+    ...

examples/plugins/HydroRoll/command/set_command.py

@@ -0,0 +1,4 @@
+from .typing import CommandBase
+
+class SetCommand(CommandBase):
+    ...

examples/plugins/HydroRoll/command/typing.py

@@ -0,0 +1,4 @@
+from pydantic import BaseModel
+
+class CommandBase(BaseModel):
+    ...

example/plugins/HydroRoll/utils.py → examples/plugins/HydroRoll/utils.py

@@ -33,14 +33,14 @@ def format_str(self, format_str: str, message_str: str = "") -> str:
         )
 
     def get_event_sender_name(self) -> str:
-        from iamai.adapter.onebot11.event import MessageEvent as OneBotMessageEvent
+        from iamai.adapter.gensokyo.event import MessageEvent as OneBotMessageEvent
 
         if isinstance(self.event, OneBotMessageEvent):
             return self.event.sender.nickname or ""
         return ""
 
     def get_event_sender_id(self) -> str:
-        from iamai.adapter.onebot11.event import MessageEvent as OneBotMessageEvent
+        from iamai.adapter.gensokyo.event import MessageEvent as OneBotMessageEvent
 
         if isinstance(self.event, OneBotMessageEvent):
             if self.event.sender.user_id is not None:

examples/plugins/cachetool.py

@@ -0,0 +1,33 @@
+from cachetools import cached
+import time
+
+from iamai import Plugin
+from iamai.event import MessageEvent
+
+
+class CachedPlugin(Plugin):
+    async def handle(self) -> None:
+        # without cached
+        def fib(n):
+            return n if n < 2 else fib(n - 1) + fib(n - 2)
+
+        s = time.time()
+        await self.event.reply(f"{fib(36)}")
+        await self.event.reply(f"Time Taken: {time.time() - s}")
+
+        # Now using cached
+        s = time.time()
+
+        # Use this decorator to enable caching
+        @cached(cache={})
+        def fib(n):
+            return n if n < 2 else fib(n - 1) + fib(n - 2)
+
+        await self.event.reply(f"{fib(36)}")
+        await self.event.reply(f"Time Taken(cached): {time.time() - s}")
+
+    async def rule(self) -> bool:
+        return (
+            isinstance(self.event, MessageEvent)
+            and self.event.get_plain_text() == ".cachetools"
+        )

examples/plugins/draftbottles/__init__.py

@@ -0,0 +1,48 @@
+from typing import Union
+from iamai import Plugin, Event, Depends
+from iamai.log import logger
+from .config import Config
+from iamai.event import MessageEvent
+from .database import Database
+from .permission import Permission
+from .workroutes import WorkRoutes
+from .inspector import Inspector
+
+
+class Bottles(Plugin, config=Config):
+    database: Database = Depends()
+    permission: Permission = Depends()
+    workroutes: WorkRoutes = Depends()
+    inspector: Inspector = Depends()
+
+    def __init__(self):
+        self.text = None
+        self.prefix = None
+        self.suffix = None
+
+    async def handle(self) -> None:
+        self.namespace = next(
+            (
+                key
+                for key, value in self.config.command_list.items()
+                if value == self.prefix
+            ),
+            "",
+        )
+        if method := getattr(self.inspector, self.namespace, None):
+            result = await method(self.suffix, self.config)
+            if result:
+                await self.event.reply(result)
+
+    async def rule(self) -> bool:
+        if not isinstance(self.event, MessageEvent):
+            return False
+        if not self.permission.is_admin():
+            return False
+        self.text = self.event.get_plain_text()
+        for prefix in list(self.config.command_list.values()):
+            if self.text.startswith(prefix):
+                self.prefix = prefix
+                self.suffix = self.text[len(self.prefix) + 1 :]
+                return True
+        return False