aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/tests/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'tests/plugins')
-rw-r--r--tests/plugins/exc.py16
-rw-r--r--tests/plugins/iamai_plugin_base/__init__.py85
-rw-r--r--tests/plugins/iamai_plugin_base/config.py30
-rw-r--r--tests/plugins/iamai_plugin_reply/__init__.py43
-rw-r--r--tests/plugins/iamai_plugin_reply/config.py11
-rw-r--r--tests/plugins/test.py35
6 files changed, 51 insertions, 169 deletions
diff --git a/tests/plugins/exc.py b/tests/plugins/exc.py
new file mode 100644
index 0000000..7e453ed
--- /dev/null
+++ b/tests/plugins/exc.py
@@ -0,0 +1,16 @@
+from iamai import Plugin
+import os
+class Exec(Plugin):
+ async def handle(self) -> None:
+ from hydroroll.config import ConfigManager, GlobalConfig
+ tmpConf = ConfigManager(os.path.join(GlobalConfig._current_path, "config", "userConf.dat"))
+ tmpConf.load()
+ await self.event.reply(str(tmpConf.properties))
+
+
+ async def rule(self) -> bool:
+ if self.event.adapter.name != "cqhttp":
+ return False
+ if self.event.type != "message":
+ return False
+ return self.event.message.startswith("t") \ No newline at end of file
diff --git a/tests/plugins/iamai_plugin_base/__init__.py b/tests/plugins/iamai_plugin_base/__init__.py
deleted file mode 100644
index 8fe559b..0000000
--- a/tests/plugins/iamai_plugin_base/__init__.py
+++ /dev/null
@@ -1,85 +0,0 @@
-import re
-from abc import ABC, abstractmethod
-from typing import Type, Union, Generic, TypeVar
-
-from iamai import Plugin
-from iamai.typing import T_State
-from iamai.adapter.cqhttp.event import GroupMessageEvent, PrivateMessageEvent
-
-from .config import BasePluginConfig, RegexPluginConfig, CommandPluginConfig
-
-T_Config = TypeVar("T_Config", bound=BasePluginConfig)
-T_RegexPluginConfig = TypeVar("T_RegexPluginConfig", bound=RegexPluginConfig)
-T_CommandPluginConfig = TypeVar("T_CommandPluginConfig", bound=CommandPluginConfig)
-
-
-class BasePlugin(
- Plugin[Union[PrivateMessageEvent, GroupMessageEvent], T_State, T_Config],
- ABC,
- Generic[T_State, T_Config],
-):
- Config: Type[T_Config] = BasePluginConfig
-
- def format_str(self, format_str: str, message_str: str = "") -> str:
- return format_str.format(
- message=message_str,
- user_name=self.event.sender.nickname,
- user_id=self.event.sender.user_id,
- )
-
- async def rule(self) -> bool:
- if self.event.adapter.name != "cqhttp":
- return False
- if self.event.type != "message_sent":
- return False
- if self.config.handle_all_message:
- return self.str_match(self.event.message.get_plain_text())
- elif self.config.handle_friend_message:
- if self.event.message_type == "private":
- return self.str_match(self.event.message.get_plain_text())
- elif self.config.handle_group_message:
- if self.event.message_type == "group":
- if (
- self.config.accept_group is None
- or self.event.group_id in self.config.accept_group
- ):
- return self.str_match(self.event.message.get_plain_text())
- return False
-
- @abstractmethod
- def str_match(self, msg_str: str) -> bool:
- raise NotImplemented
-
-
-class RegexPluginBase(BasePlugin[T_State, T_RegexPluginConfig], ABC):
- msg_match: re.Match
- re_pattern: re.Pattern
- Config: Type[T_RegexPluginConfig] = RegexPluginConfig
-
- def str_match(self, msg_str: str) -> bool:
- msg_str = msg_str.strip()
- self.msg_match = self.re_pattern.fullmatch(msg_str)
- return bool(self.msg_match)
-
-
-class CommandPluginBase(RegexPluginBase[T_State, T_CommandPluginConfig], ABC):
- command_match: re.Match
- command_re_pattern: re.Pattern
- Config: Type[T_CommandPluginConfig] = CommandPluginConfig
-
- def str_match(self, msg_str: str) -> bool:
- if not hasattr(self, "command_re_pattern"):
- self.command_re_pattern = re.compile(
- f'[{"".join(self.config.command_prefix)}]'
- f'({"|".join(self.config.command)})'
- r"\s*(?P<command_args>.*)",
- flags=re.I if self.config.ignore_case else 0,
- )
- msg_str = msg_str.strip()
- self.command_match = self.command_re_pattern.fullmatch(msg_str)
- if not self.command_match:
- return False
- self.msg_match = self.re_pattern.fullmatch(
- self.command_match.group("command_args")
- )
- return bool(self.msg_match)
diff --git a/tests/plugins/iamai_plugin_base/config.py b/tests/plugins/iamai_plugin_base/config.py
deleted file mode 100644
index 04924e0..0000000
--- a/tests/plugins/iamai_plugin_base/config.py
+++ /dev/null
@@ -1,30 +0,0 @@
-from typing import Set, Optional
-
-from iamai import ConfigModel
-
-
-class BasePluginConfig(ConfigModel):
- __config_name__ = ""
- handle_all_message: bool = True
- """是否处理所有类型的消息，此配置为 True 时会覆盖 handle_friend_message 和 handle_group_message。"""
- handle_friend_message: bool = True
- """是否处理好友消息。"""
- handle_group_message: bool = True
- """是否处理群消息。"""
- accept_group: Optional[Set[int]] = None
- """处理消息的群号，仅当 handle_group_message 为 True 时生效，留空表示处理所有群。"""
- message_str: str = "{user_name}: {message}"
- """最终发送消息的格式。"""
-
-
-class RegexPluginConfig(BasePluginConfig):
- pass
-
-
-class CommandPluginConfig(RegexPluginConfig):
- command_prefix: Set[str] = {".", "。","!",":"}
- """命令前缀。"""
- command: Set[str] = {}
- """命令文本。"""
- ignore_case: bool = True
- """忽略大小写。"""
diff --git a/tests/plugins/iamai_plugin_reply/__init__.py b/tests/plugins/iamai_plugin_reply/__init__.py
deleted file mode 100644
index 6835ca8..0000000
--- a/tests/plugins/iamai_plugin_reply/__init__.py
+++ /dev/null
@@ -1,43 +0,0 @@
-import re
-import json
-
-from plugins.iamai_plugin_base import BasePlugin
-
-from .config import Config
-
-
-class Reply(BasePlugin[None, Config]):
- priority: int = 1
- Config = Config
-
- def __post_init__(self):
- with open(self.config.data_file, "r",encoding="utf-8") as fp:
- if self.config.data_type == "json":
- json_data = json.load(fp)
- else:
- raise ValueError(f"data_type must be json, not {self.config.data_type}")
- self.rule_to_message = {
- item["rule"]: item["message"]
- for item in json_data
- if isinstance(item, dict)
- and "rule" in item.keys()
- and "message" in item.keys()
- }
-
- async def handle(self) -> None:
- msg = self.rule_to_message[self.msg_match.re.pattern]
- if isinstance(msg, str):
- await self.event.reply(self.format_str(msg, self.msg_match.string))
- else:
- await self.event.reply(msg)
-
- def str_match(self, msg_str: str) -> bool:
- msg_str = msg_str.strip()
- for rule in self.rule_to_message.keys():
- msg_match = re.fullmatch(
- rule, msg_str, flags=re.I if self.config.ignore_case else 0
- )
- if msg_match:
- self.msg_match = msg_match
- return bool(self.msg_match)
- return False
diff --git a/tests/plugins/iamai_plugin_reply/config.py b/tests/plugins/iamai_plugin_reply/config.py
deleted file mode 100644
index e024fb6..0000000
--- a/tests/plugins/iamai_plugin_reply/config.py
+++ /dev/null
@@ -1,11 +0,0 @@
-from plugins.iamai_plugin_base import CommandPluginConfig
-
-
-class Config(CommandPluginConfig):
- __config_name__ = "plugin_reply"
- data_type: str = "json"
- """数据类型，目前只支持 json。"""
- data_file: str = "data/reply_data.json"
- """数据文件位置。"""
- ignore_case: bool = True
- """是否忽略大小写，默认为 True。"""
diff --git a/tests/plugins/test.py b/tests/plugins/test.py
new file mode 100644
index 0000000..6f3bc3b
--- /dev/null
+++ b/tests/plugins/test.py
@@ -0,0 +1,35 @@
+from iamai import Plugin
+from iamai.exceptions import GetEventTimeout
+
+
+class Weather(Plugin):
+ async def handle(self) -> None:
+ args = self.event.get_plain_text().split(" ")
+ if len(args) >= 2:
+ await self.event.reply(await self.get_weather(args[1]))
+ else:
+ await self.event.reply("请输入想要查询天气的城市：")
+ try:
+ city_event = await self.event.adapter.get(
+ lambda x: x.type == "message", timeout=10
+ )
+ except GetEventTimeout:
+ return
+ else:
+ await self.event.reply(
+ await self.get_weather(city_event.get_plain_text())
+ )
+
+ async def rule(self) -> bool:
+ if self.event.adapter.name != "cqhttp":
+ return False
+ if self.event.type != "message":
+ return False
+ return self.event.message.startswith("天气")
+
+ @staticmethod
+ async def get_weather(city):
+ if city not in ["北京", "上海"]:
+ return "你想查询的城市暂不支持！"
+ return f"{city}的天气是..."
+
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-21 09:34:02 +0000