1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
"""iamai 插件。
所有 iamai 插件的基类。所有用户编写的插件必须继承自 `Plugin` 类。
"""
from enum import Enum
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Type, Generic, NoReturn, Optional
from hydrorollcore.config import ConfigModel
from hydrorollcore.utils import is_config_class
from hydrorollcore.typing import T_Event, T_State, T_Config
from hydrorollcore.exceptions import SkipException, StopException
if TYPE_CHECKING:
from iamai.bot import Bot
__all__ = ["Rule", "RuleLoadType"]
class RuleLoadType(Enum):
"""插件加载类型。"""
DIR = "dir"
NAME = "name"
FILE = "file"
CLASS = "class"
class Rule(ABC, Generic[T_Event, T_State, T_Config]):
"""所有 iamai 插件的基类。
Attributes:
event: 当前正在被此插件处理的事件。
priority: 插件的优先级，数字越小表示优先级越高，默认为 0。
block: 插件执行结束后是否阻止事件的传播。True 表示阻止。
__plugin_load_type__: 插件加载类型，由 iamai 自动设置，反映了此插件是如何被加载的。
__plugin_file_path__: 当插件加载类型为 `PluginLoadType.CLASS` 时为 `None`，
否则为定义插件在的 Python 模块的位置。
"""
event: T_Event
priority: int = 0
block: bool = False
Config: Type[ConfigModel]
__plugin_load_type__: PluginLoadType
__plugin_file_path__: Optional[str]
def __init__(self, event: T_Event):
self.event = event
if not hasattr(self, "priority"):
self.priority = 0
if not hasattr(self, "block"):
self.block = False
self.get = self.bot.get
self.__post_init__()
def __post_init__(self):
"""用于初始化后处理，被 `__init__()` 方法调用。"""
pass
@property
def name(self) -> str:
"""插件类名称。"""
return self.__class__.__name__
@property
def bot(self) -> "Bot":
"""机器人对象。"""
return self.event.adapter.bot
@property
def config(self) -> Optional[T_Config]:
"""插件配置。"""
config_class: ConfigModel = getattr(self, "Config", None)
if is_config_class(config_class):
return getattr(self.bot.config.plugin, config_class.__config_name__, None)
return None
def stop(self) -> NoReturn:
"""停止当前事件传播。"""
raise StopException()
def skip(self) -> NoReturn:
"""跳过自身继续当前事件传播。"""
raise SkipException()
@property
def state(self) -> T_State:
"""插件状态。"""
return self.bot.plugin_state[self.name]
@state.setter
def state(self, value: T_State):
self.bot.plugin_state[self.name] = value
@abstractmethod
async def handle(self) -> None:
"""处理事件的方法。当 `rule()` 方法返回 `True` 时 iamai 会调用此方法。每个插件必须实现此方法。"""
raise NotImplementedError
@abstractmethod
async def rule(self) -> bool:
"""匹配事件的方法。事件处理时，会按照插件的优先级依次调用此方法，当此方法返回 `True` 时将事件交由此插件处理。每个插件必须实现此方法。
注意：不建议直接在此方法内实现对事件的处理，事件的具体处理请交由 `handle()` 方法。
"""
raise NotImplementedError
|