diff options
| -rw-r--r-- | HydroRoll/bot.py | 64 | ||||
| -rw-r--r-- | HydroRoll/plugins/HydroRoll_plugin_base/__init__.py | 93 | ||||
| -rw-r--r-- | HydroRoll/plugins/HydroRoll_plugin_base/config.py | 30 | ||||
| -rw-r--r-- | HydroRoll/test.py | 0 |
4 files changed, 187 insertions, 0 deletions
diff --git a/HydroRoll/bot.py b/HydroRoll/bot.py new file mode 100644 index 0000000..c2b99e6 --- /dev/null +++ b/HydroRoll/bot.py @@ -0,0 +1,64 @@ +from iamai import Bot as _Bot +from typing import Optional, Dict, List, Type, Any, Union +from pathlib import Path +import os +# 获取当前目录路径 +current_dir = Path.cwd() + +# 获取当前脚本文件路径 +script_file = current_dir.resolve() / __file__ + +# 获取当前脚本文件所在目录路径 +script_dir = script_file.parent + +__all__ = ["Bot"] + +class Bot: + def __init__( + self, + *, + config_file: Optional[str] = "config.toml", + config_dict: Optional[Dict] = None, + hot_reload: bool = False, + ) -> None: + self.bot = _Bot(hot_reload=hot_reload, + config_file=config_file, + config_dict=config_dict + ) + self.bot.load_plugins_from_dirs(Path(f"{script_dir}/plugins")) + self.create_folders() + self.init_data() + self.init_conf() + self.init_webui() + + def run(self) -> None: + self.bot.run() + + def restart(self) -> None: + self.bot.restart() + + def create_folders(self): + folder_path = os.path.dirname(os.path.abspath('__file__')) # 获取main.py所在文件夹路径 + if not os.path.isdir(os.path.join(folder_path, 'user')): + os.mkdir(os.path.join(folder_path, 'user')) + if not os.path.isdir(os.path.join(folder_path, 'data')): + os.mkdir(os.path.join(folder_path, 'data')) + if not os.path.isdir(os.path.join(folder_path, 'models')): + os.mkdir(os.path.join(folder_path, 'models')) + if not os.path.isdir(os.path.join(folder_path, 'web')): + os.mkdir(os.path.join(folder_path, 'web')) + if not os.path.isdir(os.path.join(folder_path, 'config')): + os.mkdir(os.path.join(folder_path, 'config')) + if not os.path.isdir(os.path.join(folder_path, 'logs')): + os.mkdir(os.path.join(folder_path, 'logs')) + if not os.path.isdir(os.path.join(folder_path, 'rules')): + os.mkdir(os.path.join(folder_path, 'rules')) + + def init_data(self): + ... + + def init_webui(self): + ... + + def init_conf(self): + ...
\ No newline at end of file diff --git a/HydroRoll/plugins/HydroRoll_plugin_base/__init__.py b/HydroRoll/plugins/HydroRoll_plugin_base/__init__.py new file mode 100644 index 0000000..a050ae1 --- /dev/null +++ b/HydroRoll/plugins/HydroRoll_plugin_base/__init__.py @@ -0,0 +1,93 @@ +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: + is_bot_off = True + + if self.event.adapter.name != "cqhttp": + return False + if self.event.type != "message": + return False + match_str = self.event.message.get_plain_text() + if is_bot_off: + if self.event.message.startswith(f'[CQ:at,qq={self.event.self_id}]'): + match_str = re.sub(fr'^\[CQ:at,qq={self.event.self_id}\]', '', match_str) + else: + return False + if self.config.handle_all_message: + return self.str_match(match_str) + elif self.config.handle_friend_message: + if self.event.message_type == "private": + return self.str_match(match_str) + 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(match_str) + 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)
\ No newline at end of file diff --git a/HydroRoll/plugins/HydroRoll_plugin_base/config.py b/HydroRoll/plugins/HydroRoll_plugin_base/config.py new file mode 100644 index 0000000..311874c --- /dev/null +++ b/HydroRoll/plugins/HydroRoll_plugin_base/config.py @@ -0,0 +1,30 @@ +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/HydroRoll/test.py b/HydroRoll/test.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/HydroRoll/test.py |