aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/example
diff options
context:
space:
mode:
Diffstat (limited to 'example')
-rw-r--r--example/plugins/HydroRoll/__init__.py20
-rw-r--r--example/plugins/HydroRoll/command.py6
-rw-r--r--example/plugins/HydroRoll/config.py92
-rw-r--r--example/plugins/HydroRoll/typing.py34
-rw-r--r--example/plugins/HydroRoll/utils.py127
-rw-r--r--example/plugins/r.py10
-rw-r--r--example/plugins/show.py2
-rw-r--r--example/plugins/tf/__init__.py52
-rw-r--r--example/plugins/tf/config.py0
9 files changed, 90 insertions, 253 deletions
diff --git a/example/plugins/HydroRoll/__init__.py b/example/plugins/HydroRoll/__init__.py
index d8c87f5..6739d94 100644
--- a/example/plugins/HydroRoll/__init__.py
+++ b/example/plugins/HydroRoll/__init__.py
@@ -58,6 +58,7 @@ class HydroRoll(Plugin):
logger.info(f"Command {current_cmd} not found with flag {flag}")
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:
@@ -80,9 +81,8 @@ class HydroRoll(Plugin):
@BODY: lexer module will return a list of tokens, parser module will parse the tokens into a tree, and executor module will execute the tokens with a stack with a bool return value.
"""
logger.info("loading psi...")
- if (
- not self.bot.global_state["HydroRoll"].get("hola")
- and not os.path.exists(join(BASE_DIR, "HydroRoll"))
+ if not self.bot.global_state["HydroRoll"].get("hola") and not os.path.exists(
+ join(BASE_DIR, "HydroRoll")
):
# hola = self.models["hola"]
# _, max_similarity = find_max_similarity(
@@ -126,17 +126,3 @@ class HydroRoll(Plugin):
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
diff --git a/example/plugins/HydroRoll/command.py b/example/plugins/HydroRoll/command.py
index 5a31534..dcb10b0 100644
--- a/example/plugins/HydroRoll/command.py
+++ b/example/plugins/HydroRoll/command.py
@@ -1,9 +1,7 @@
-from .utils import CommandPluginBase
-
-class Get(CommandPluginBase):
+class Get:
...
-class Set(CommandPluginBase):
+class Set:
... \ No newline at end of file
diff --git a/example/plugins/HydroRoll/config.py b/example/plugins/HydroRoll/config.py
index 788c27a..97591fe 100644
--- a/example/plugins/HydroRoll/config.py
+++ b/example/plugins/HydroRoll/config.py
@@ -9,30 +9,14 @@ import os
from typing import Set, Optional
from iamai import ConfigModel
import datetime
-# 创建全局 ArgumentParser 对象
-global_parser = argparse.ArgumentParser(description="HydroRoll[水系] 全局命令参数")
+from typing import Set
-class Color:
- # 定义ANSI转义序列
- RESET = '\033[0m'
- BLUE_BASE = '\033[36m'
- BLUE_DARK = '\033[34m'
- BLUE_DARKER = '\033[32m'
- BLUE_DARKEST = '\033[30m'
+from pydantic import Field
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}"
+ message_str: str = "{message}"
"""最终发送消息的格式。"""
@@ -41,16 +25,25 @@ class RegexPluginConfig(BasePluginConfig):
class CommandPluginConfig(RegexPluginConfig):
- command_prefix: Set[str] = {":"}
+ command_prefix: Set[str] = Field(default_factory=lambda: {".", "。"})
"""命令前缀。"""
- command: Set[str] = {}
+ command: Set[str] = Field(default_factory=set)
"""命令文本。"""
ignore_case: bool = True
"""忽略大小写。"""
+class Color:
+ # 定义ANSI转义序列
+ RESET = "\033[0m"
+ BLUE_BASE = "\033[36m"
+ BLUE_DARK = "\033[34m"
+ BLUE_DARKER = "\033[32m"
+ BLUE_DARKEST = "\033[30m"
+
+
# 定义全局配置类
-class GlobalConfig(CommandPluginConfig):
+class GlobalConfig:
_name = "HydroRoll[水系]"
_version = "0.1.0"
_svn = "2"
@@ -73,39 +66,6 @@ class GlobalConfig(CommandPluginConfig):
Github: https://github.com/HydroRoll-Team
Under the MIT License, see LICENSE for more details.
"""
- _rg = Generator(AESCounter())
- SEED = _rg.random() # 随机数种子
-
- # 定义系统组件
-
- class HydroSystem:
- def __init__(self):
- self.parser = argparse.ArgumentParser(
- description="HydroRoll[水系].system 系统命令参数"
- )
- self.subparsers = self.parser.add_subparsers()
- self.status_parser = self.subparsers.add_parser(
- "status", aliases=["stat", "state"], help="系统状态"
- )
- self.reload_parser = self.subparsers.add_parser(
- "reload", aliases=["rld"], help="重新加载系统"
- )
- self.restart_parser = self.subparsers.add_parser(
- "restart", aliases=["rst"], help="重启系统"
- )
- self.collect_parser = self.subparsers.add_parser(
- "collect", aliases=["gc"], help="释放 python 内存"
- )
- self.help = "\n".join(
- self.parser.format_help()
- .replace("\r\n", "\n")
- .replace("\r", "")
- .split("\n")[2:-3]
- )
-
- class HydroBot:
- def __init__(self) -> None:
- self.parser = argparse.ArgumentParser(description="Bot命令")
class Directory(object):
@@ -113,16 +73,15 @@ class Directory(object):
self.current_path = _path
def get_dice_dir_list(self, _prefix: str) -> list:
-
return [
- os.path.join(self.current_path, f'{_prefix}', *dirs)
+ os.path.join(self.current_path, f"{_prefix}", *dirs)
for dirs in [
- ['config'],
- ['data'],
- ['rules'],
- ['scripts'],
- ['web', 'frontend'],
- ['web', 'backend'],
+ ["config"],
+ ["data"],
+ ["rules"],
+ ["scripts"],
+ ["web", "frontend"],
+ ["web", "backend"],
]
]
@@ -133,8 +92,8 @@ class FileManager(object):
def get_file_list(self, _dir: str):
return {
- 'web;frontend': 'index.html',
- 'data': 'censor.json',
+ "web;frontend": "index.html",
+ "data": "censor.json",
}
@@ -142,8 +101,7 @@ class Models:
"""模型管理类"""
def __init__(self) -> None:
- self.builtin_models = {'hola': 'hola.pkl'}
+ self.builtin_models = {"hola": "hola.pkl"}
def get_models_dict(self) -> dict:
return self.builtin_models
-
diff --git a/example/plugins/HydroRoll/typing.py b/example/plugins/HydroRoll/typing.py
deleted file mode 100644
index 876fa92..0000000
--- a/example/plugins/HydroRoll/typing.py
+++ /dev/null
@@ -1,34 +0,0 @@
-"""HydroRoll 类型提示支持。
-
-此模块定义了部分 HydroRoll 使用的类型。
-"""
-
-from typing import TYPE_CHECKING, TypeVar, Callable, NoReturn, Awaitable
-
-from iamai.message import T_MS, T_Message, T_MessageSegment
-
-if TYPE_CHECKING:
- from iamai.bot import Bot # noqa
- from iamai.event import Event # noqa
- from iamai.plugin import Plugin # noqa
- from iamai.config import ConfigModel # noqa
-
-__all__ = [
- "T_State",
- "T_Event",
- "T_Plugin",
- "T_Config",
- "T_Message",
- "T_MessageSegment",
- "T_MS",
- "T_BotHook",
- "T_EventHook",
-]
-
-T_State = TypeVar("T_State")
-T_Event = TypeVar("T_Event", bound="Event")
-T_Plugin = TypeVar("T_Plugin", bound="Plugin")
-T_Config = TypeVar("T_Config", bound="ConfigModel")
-
-T_BotHook = Callable[["Bot"], Awaitable[NoReturn]]
-T_EventHook = Callable[[T_Event], Awaitable[NoReturn]] \ No newline at end of file
diff --git a/example/plugins/HydroRoll/utils.py b/example/plugins/HydroRoll/utils.py
index 2665580..2b4c7bf 100644
--- a/example/plugins/HydroRoll/utils.py
+++ b/example/plugins/HydroRoll/utils.py
@@ -5,100 +5,94 @@ import random
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
+import re
+from abc import ABC, abstractmethod
+from typing import Any, Generic, TypeVar
+
+from iamai import MessageEvent, Plugin
+from iamai.typing import StateT
-T_Config = TypeVar("T_Config", bound=BasePluginConfig)
-T_RegexPluginConfig = TypeVar("T_RegexPluginConfig", bound=RegexPluginConfig)
-T_CommandPluginConfig = TypeVar("T_CommandPluginConfig", bound=CommandPluginConfig)
+from .config import BasePluginConfig, CommandPluginConfig, RegexPluginConfig
+
+ConfigT = TypeVar("ConfigT", bound=BasePluginConfig)
+RegexPluginConfigT = TypeVar("RegexPluginConfigT", bound=RegexPluginConfig)
+CommandPluginConfigT = TypeVar("CommandPluginConfigT", bound=CommandPluginConfig)
class BasePlugin(
- Plugin[Union[PrivateMessageEvent, GroupMessageEvent], T_State, T_Config],
+ Plugin[MessageEvent[Any], StateT, ConfigT],
ABC,
- Generic[T_State, T_Config],
+ Generic[StateT, ConfigT],
):
- Config: Type[T_Config] = BasePluginConfig # type: ignore
-
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,
+ user_name=self.get_event_sender_name(),
+ user_id=self.get_event_sender_id(),
)
- async def rule(self) -> bool:
- is_bot_off = False
+ def get_event_sender_name(self) -> str:
+ from iamai.adapter.onebot11.event import MessageEvent as OneBotMessageEvent
- 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(
- rf"^\[CQ:at,qq={self.event.self_id}\]", "", match_str
- )
- elif self.event.message.startswith(f"[CQ:at,qq={self.event.self_tiny_id}]"):
- match_str = re.sub(
- rf"^\[CQ:at,qq={self.event.self_tiny_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
+ 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
+
+ if isinstance(self.event, OneBotMessageEvent):
+ if self.event.sender.user_id is not None:
+ return str(self.event.sender.user_id)
+ return ""
+ return ""
+
+ async def rule(self) -> bool:
+ return isinstance(self.event, MessageEvent) and self.str_match(
+ self.event.get_plain_text()
+ )
@abstractmethod
def str_match(self, msg_str: str) -> bool:
- raise NotImplemented
+ raise NotImplementedError
-class RegexPluginBase(BasePlugin[T_State, T_RegexPluginConfig], ABC):
- msg_match: re.Match
- re_pattern: re.Pattern
- Config: Type[T_RegexPluginConfig] = RegexPluginConfig
+class RegexPluginBase(BasePlugin[StateT, RegexPluginConfigT], ABC):
+ msg_match: re.Match[str]
+ re_pattern: re.Pattern[str]
def str_match(self, msg_str: str) -> bool:
msg_str = msg_str.strip()
- self.msg_match = self.re_pattern.fullmatch(msg_str)
+ msg_match = self.re_pattern.fullmatch(msg_str)
+ if msg_match is None:
+ return False
+ self.msg_match = msg_match
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
+class CommandPluginBase(RegexPluginBase[StateT, CommandPluginConfigT], ABC):
+ command_match: re.Match[str]
+ command_re_pattern: re.Pattern[str]
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)})'
+ if not hasattr(self, "re_pattern"):
+ self.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:
+ msg_match = self.re_pattern.fullmatch(msg_str)
+ if not msg_match:
return False
- self.msg_match = self.re_pattern.fullmatch(
- self.command_match.group("command_args")
- )
- return bool(self.msg_match)
+ self.msg_match = msg_match
+ command_match = self.re_pattern.fullmatch(self.msg_match.group("command_args"))
+ if not command_match:
+ return False
+ self.command_match = command_match
+ return True
class PseudoRandomGenerator:
@@ -153,9 +147,7 @@ class HydroDice:
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}"
- )
+ return f"{total} = {rolls_str}" if is_reversed else f"{rolls_str} = {total}"
def find_max_similarity(input_string, string_list):
@@ -174,6 +166,5 @@ def find_max_similarity(input_string, string_list):
def check_file(filename: str) -> bool:
"""根据给定参数校验文件夹内文件完整性"""
-
-
- return False \ No newline at end of file
+
+ return False
diff --git a/example/plugins/r.py b/example/plugins/r.py
deleted file mode 100644
index 2f05af9..0000000
--- a/example/plugins/r.py
+++ /dev/null
@@ -1,10 +0,0 @@
-from iamai import Plugin
-
-
-class Roll(Plugin):
- async def handle(self) -> None:
- await self.event.reply("""Attack: 25
-Damage: 9""")
-
- async def rule(self) -> bool:
- return self.event.type == "message" and self.event.message.startswith("/r")
diff --git a/example/plugins/show.py b/example/plugins/show.py
index e43a241..a9cf446 100644
--- a/example/plugins/show.py
+++ b/example/plugins/show.py
@@ -1,6 +1,6 @@
from iamai import Plugin
from numpy.random import Generator
-from iamai.adapter.cqhttp.message import CQHTTPMessage, CQHTTPMessageSegment
+from iamai.adapter.onebot11.message import CQHTTPMessage, CQHTTPMessageSegment
ms = CQHTTPMessageSegment
diff --git a/example/plugins/tf/__init__.py b/example/plugins/tf/__init__.py
deleted file mode 100644
index ce78d97..0000000
--- a/example/plugins/tf/__init__.py
+++ /dev/null
@@ -1,52 +0,0 @@
-
-from iamai import Plugin
-from iamai.log import logger as _logger
-from HydroRoll.models.cos_sim import cosSim
-import jieba
-
-logger = _logger
-
-
-texts = [
- "你好 HydroRoll",
- "你好 水系",
- "hi 水系",
- "hi HydroRoll",
- "hello 水系",
- "hello HydroRoll",
- "hola 水系",
- "hola HydroRoll",
-]
-
-cos_Sim = cosSim(texts)
-logger.info(f"{cos_Sim.calcuSim('你好')}")
-
-cos_Sim.save('cos.h5')
-
-model = cosSim.load('cos.h5')
-
-logger.info(f"{model}")
-
-# class Sim(Plugin):
-# async def handle(self) -> None:
-# try:
-
-# txt_list = eval(self.event.message.get_plain_text()[5:])
-# if len(txt_list) == 1:
-# await self.event.reply(f"{cos_Sim.CalcuSim(txt_list)}")
-# elif len(txt_list) == 2:
-# corpuss = [" ".join(jieba.cut(text))
-# for text in eval(self.event.message.get_plain_text()[5:])]
-# await self.event.reply(str(corpuss))
-# vocabulary = cos_Sim.getVocabulary(corpuss)
-# v = cos_Sim.getVectors(corpuss, vocabulary=vocabulary)
-# await self.event.reply(f"weight\n=========\n{v}")
-# await self.event.reply(f"相似度\n=========\n{cos_Sim.cos_sim(v[0], v[1])}")
-# except Exception as e:
-# await self.event.reply(f"{e!r}")
-
-# async def rule(self) -> bool:
-# return self.event.type == "message" and self.event.message.startswith(".cos")
-
-
-
diff --git a/example/plugins/tf/config.py b/example/plugins/tf/config.py
deleted file mode 100644
index e69de29..0000000
--- a/example/plugins/tf/config.py
+++ /dev/null
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-21 19:43:44 +0000