ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2024-11-27 18:45:05 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ finish typing change

Browse Source
This commit is contained in:
yanyongyu 2020-12-17 21:09:30 +08:00
parent 6bd69e6708
commit 799144e64d
19 changed files with 309 additions and 315 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nonebot/config.py
View File

@ -216,9 +216,9 @@ class Config(BaseConfig):
"""
# bot runtime configs
superusers: Set[int] = set()
superusers: Set[str] = set()
"""
- **类型**: ``Set[int]``
- **类型**: ``Set[str]``
- **默认值**: ``set()``
:说明:
@ -229,7 +229,7 @@ class Config(BaseConfig):
.. code-block:: default
SUPER_USERS=[12345789]
SUPER_USERS=["12345789"]
"""
nickname: Set[str] = set()
"""

2
nonebot/drivers/fastapi.py
View File

@ -205,7 +205,7 @@ class WebSocket(BaseWebSocket):
def __init__(self, websocket: FastAPIWebSocket):
super().__init__(websocket)
self._closed = None
self._closed = False
@property
@overrides(BaseWebSocket)

68
nonebot/matcher.py
View File

@ -15,7 +15,7 @@ from typing import Type, List, Dict, Union, Callable, Optional, NoReturn, TYPE_C
from nonebot.rule import Rule
from nonebot.log import logger
from nonebot.permission import Permission, USER
from nonebot.typing import State, Handler, ArgsParser
from nonebot.typing import T_State, T_Handler, T_ArgsParser
from nonebot.exception import PausedException, RejectedException, FinishedException
if TYPE_CHECKING:
@ -64,9 +64,9 @@ class Matcher(metaclass=MatcherMeta):
:类型: ``Permission``
:说明: 事件响应器触发权限
"""
handlers: List[Handler] = []
handlers: List[T_Handler] = []
"""
:类型: ``List[Handler]``
:类型: ``List[T_Handler]``
:说明: 事件响应器拥有的事件处理函数列表
"""
priority: int = 1
@ -90,15 +90,15 @@ class Matcher(metaclass=MatcherMeta):
:说明: 事件响应器过期时间点
"""
_default_state: State = {}
_default_state: T_State = {}
"""
:类型: ``State``
:类型: ``T_State``
:说明: 事件响应器默认状态
"""
_default_parser: Optional[ArgsParser] = None
_default_parser: Optional[T_ArgsParser] = None
"""
:类型: ``Optional[ArgsParser]``
:类型: ``Optional[T_ArgsParser]``
:说明: 事件响应器默认参数解析函数
"""
@ -119,13 +119,13 @@ class Matcher(metaclass=MatcherMeta):
type_: str = "",
rule: Optional[Rule] = None,
permission: Optional[Permission] = None,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
*,
module: Optional[str] = None,
default_state: Optional[State] = None,
default_state: Optional[T_State] = None,
expire_time: Optional[datetime] = None) -> Type["Matcher"]:
"""
:说明:
@ -137,12 +137,12 @@ class Matcher(metaclass=MatcherMeta):
* ``type_: str``: 事件响应器类型，与 ``event.type`` 一致时触发，空字符串表示任意
* ``rule: Optional[Rule]``: 匹配规则
* ``permission: Optional[Permission]``: 权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器，即触发一次后删除
* ``priority: int``: 响应优先级
* ``block: bool``: 是否阻止事件向更低优先级的响应器传播
* ``module: Optional[str]``: 事件响应器所在模块名称
* ``default_state: Optional[State]``: 默认状态 ``state``
* ``default_state: Optional[T_State]``: 默认状态 ``state``
* ``expire_time: Optional[datetime]``: 事件响应器最终有效时间点，过时即被删除
:返回:
@ -187,7 +187,8 @@ class Matcher(metaclass=MatcherMeta):
return await cls.permission(bot, event)
@classmethod
async def check_rule(cls, bot: "Bot", event: "Event", state: State) -> bool:
async def check_rule(cls, bot: "Bot", event: "Event",
state: T_State) -> bool:
"""
:说明:
@ -197,7 +198,7 @@ class Matcher(metaclass=MatcherMeta):
* ``bot: Bot``: Bot 对象
* ``event: Event``: 上报事件
* ``state: State``: 当前状态
* ``state: T_State``: 当前状态
:返回:
@ -208,7 +209,7 @@ class Matcher(metaclass=MatcherMeta):
await cls.rule(bot, event, state))
@classmethod
def args_parser(cls, func: ArgsParser) -> ArgsParser:
def args_parser(cls, func: T_ArgsParser) -> T_ArgsParser:
"""
:说明:
@ -216,13 +217,13 @@ class Matcher(metaclass=MatcherMeta):
:参数:
* ``func: ArgsParser``: 参数解析函数
* ``func: T_ArgsParser``: 参数解析函数
"""
cls._default_parser = func
return func
@staticmethod
def process_handler(handler: Handler) -> Handler:
def process_handler(handler: T_Handler) -> T_Handler:
signature = inspect.signature(handler, follow_wrapped=False)
bot = signature.parameters.get("bot")
event = signature.parameters.get("event")
@ -232,17 +233,17 @@ class Matcher(metaclass=MatcherMeta):
handler.__params__ = {
"bot": bot.annotation,
"event": event.annotation if event else None,
"state": State if state else None
"state": T_State if state else None
}
return handler
@classmethod
def append_handler(cls, handler: Handler) -> None:
def append_handler(cls, handler: T_Handler) -> None:
# Process handler first
cls.handlers.append(cls.process_handler(handler))
@classmethod
def handle(cls) -> Callable[[Handler], Handler]:
def handle(cls) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
@ -253,14 +254,14 @@ class Matcher(metaclass=MatcherMeta):
* 无
"""
def _decorator(func: Handler) -> Handler:
def _decorator(func: T_Handler) -> T_Handler:
cls.append_handler(func)
return func
return _decorator
@classmethod
def receive(cls) -> Callable[[Handler], Handler]:
def receive(cls) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
@ -272,14 +273,14 @@ class Matcher(metaclass=MatcherMeta):
"""
async def _receive(bot: "Bot", event: "Event",
state: State) -> NoReturn:
state: T_State) -> NoReturn:
raise PausedException
if cls.handlers:
# 已有前置handlers则接受一条新的消息，否则视为接收初始消息
cls.append_handler(_receive)
def _decorator(func: Handler) -> Handler:
def _decorator(func: T_Handler) -> T_Handler:
if not cls.handlers or cls.handlers[-1] is not func:
cls.append_handler(func)
@ -292,8 +293,8 @@ class Matcher(metaclass=MatcherMeta):
cls,
key: str,
prompt: Optional[Union[str, "Message", "MessageSegment"]] = None,
args_parser: Optional[ArgsParser] = None
) -> Callable[[Handler], Handler]:
args_parser: Optional[T_ArgsParser] = None
) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
@ -303,10 +304,10 @@ class Matcher(metaclass=MatcherMeta):
* ``key: str``: 参数名
* ``prompt: Optional[Union[str, Message, MessageSegment]]``: 在参数不存在时向用户发送的消息
* ``args_parser: Optional[ArgsParser]``: 可选参数解析函数，空则使用默认解析函数
* ``args_parser: Optional[T_ArgsParser]``: 可选参数解析函数，空则使用默认解析函数
"""
async def _key_getter(bot: "Bot", event: "Event", state: State):
async def _key_getter(bot: "Bot", event: "Event", state: T_State):
state["_current_key"] = key
if key not in state:
if prompt:
@ -316,12 +317,13 @@ class Matcher(metaclass=MatcherMeta):
else:
state["_skip_key"] = True
async def _key_parser(bot: "Bot", event: "Event", state: State):
async def _key_parser(bot: "Bot", event: "Event", state: T_State):
if key in state and state.get("_skip_key"):
del state["_skip_key"]
return
parser = args_parser or cls._default_parser
if parser:
# parser = cast(T_ArgsParser["Bot", "Event"], parser)
await parser(bot, event, state)
else:
state[state["_current_key"]] = str(event.get_message())
@ -329,13 +331,13 @@ class Matcher(metaclass=MatcherMeta):
cls.append_handler(_key_getter)
cls.append_handler(_key_parser)
def _decorator(func: Handler) -> Handler:
def _decorator(func: T_Handler) -> T_Handler:
if not hasattr(cls.handlers[-1], "__wrapped__"):
cls.process_handler(func)
parser = cls.handlers.pop()
@wraps(func)
async def wrapper(bot: "Bot", event: "Event", state: State):
async def wrapper(bot: "Bot", event: "Event", state: T_State):
await cls.run_handler(parser, bot, event, state)
await cls.run_handler(func, bot, event, state)
if "_current_key" in state:
@ -428,8 +430,8 @@ class Matcher(metaclass=MatcherMeta):
raise RejectedException
@classmethod
async def run_handler(cls, handler: Handler, bot: "Bot", event: "Event",
state: State):
async def run_handler(cls, handler: T_Handler, bot: "Bot", event: "Event",
state: T_State):
if not hasattr(handler, "__params__"):
cls.process_handler(handler)
params = getattr(handler, "__params__")
@ -445,7 +447,7 @@ class Matcher(metaclass=MatcherMeta):
**{k: v for k, v in args.items() if params[k] is not None})
# 运行handlers
async def run(self, bot: "Bot", event: "Event", state: State):
async def run(self, bot: "Bot", event: "Event", state: T_State):
b_t = current_bot.set(bot)
e_t = current_event.set(event)
try:

32
nonebot/message.py
View File

@ -13,18 +13,18 @@ from nonebot.log import logger
from nonebot.rule import TrieRule
from nonebot.matcher import matchers, Matcher
from nonebot.exception import IgnoredException, StopPropagation, NoLogException
from nonebot.typing import State, EventPreProcessor, RunPreProcessor, EventPostProcessor, RunPostProcessor
from nonebot.typing import T_State, T_EventPreProcessor, T_RunPreProcessor, T_EventPostProcessor, T_RunPostProcessor
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
_event_preprocessors: Set[EventPreProcessor] = set()
_event_postprocessors: Set[EventPostProcessor] = set()
_run_preprocessors: Set[RunPreProcessor] = set()
_run_postprocessors: Set[RunPostProcessor] = set()
_event_preprocessors: Set[T_EventPreProcessor] = set()
_event_postprocessors: Set[T_EventPostProcessor] = set()
_run_preprocessors: Set[T_RunPreProcessor] = set()
_run_postprocessors: Set[T_RunPostProcessor] = set()
def event_preprocessor(func: EventPreProcessor) -> EventPreProcessor:
def event_preprocessor(func: T_EventPreProcessor) -> T_EventPreProcessor:
"""
:说明:
@ -36,13 +36,13 @@ def event_preprocessor(func: EventPreProcessor) -> EventPreProcessor:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: State``: 当前 State
* ``state: T_State``: 当前 State
"""
_event_preprocessors.add(func)
return func
def event_postprocessor(func: EventPostProcessor) -> EventPostProcessor:
def event_postprocessor(func: T_EventPostProcessor) -> T_EventPostProcessor:
"""
:说明:
@ -54,13 +54,13 @@ def event_postprocessor(func: EventPostProcessor) -> EventPostProcessor:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: State``: 当前事件运行前 State
* ``state: T_State``: 当前事件运行前 State
"""
_event_postprocessors.add(func)
return func
def run_preprocessor(func: RunPreProcessor) -> RunPreProcessor:
def run_preprocessor(func: T_RunPreProcessor) -> T_RunPreProcessor:
"""
:说明:
@ -73,13 +73,13 @@ def run_preprocessor(func: RunPreProcessor) -> RunPreProcessor:
* ``matcher: Matcher``: 当前要运行的事件响应器
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: State``: 当前 State
* ``state: T_State``: 当前 State
"""
_run_preprocessors.add(func)
return func
def run_postprocessor(func: RunPostProcessor) -> RunPostProcessor:
def run_postprocessor(func: T_RunPostProcessor) -> T_RunPostProcessor:
"""
:说明:
@ -93,18 +93,18 @@ def run_postprocessor(func: RunPostProcessor) -> RunPostProcessor:
* ``exception: Optional[Exception]``: 事件响应器运行错误（如果存在）
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: State``: 当前 State
* ``state: T_State``: 当前 State
"""
_run_postprocessors.add(func)
return func
async def _check_matcher(priority: int, bot: "Bot", event: "Event",
state: State) -> Iterable[Type[Matcher]]:
state: T_State) -> Iterable[Type[Matcher]]:
current_matchers = matchers[priority].copy()
async def _check(Matcher: Type[Matcher], bot: "Bot", event: "Event",
state: State) -> Optional[Type[Matcher]]:
state: T_State) -> Optional[Type[Matcher]]:
try:
if (not Matcher.expire_time or datetime.now() <= Matcher.expire_time
) and await Matcher.check_perm(
@ -139,7 +139,7 @@ async def _check_matcher(priority: int, bot: "Bot", event: "Event",
async def _run_matcher(Matcher: Type[Matcher], bot: "Bot", event: "Event",
state: State) -> None:
state: T_State) -> None:
logger.info(f"Event will be handled by {Matcher}")
matcher = Matcher()

4
nonebot/permission.py
View File

@ -13,7 +13,7 @@ import asyncio
from typing import Union, Optional, Callable, NoReturn, Awaitable, TYPE_CHECKING
from nonebot.utils import run_sync
from nonebot.typing import PermissionChecker
from nonebot.typing import T_PermissionChecker
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
@ -67,7 +67,7 @@ class Permission:
def __or__(
self, other: Optional[Union["Permission",
PermissionChecker]]) -> "Permission":
T_PermissionChecker]]) -> "Permission":
checkers = self.checkers.copy()
if other is None:
return self

181
nonebot/plugin.py
View File

@ -18,7 +18,7 @@ from typing import Any, Set, List, Dict, Type, Tuple, Union, Optional
from nonebot.log import logger
from nonebot.matcher import Matcher
from nonebot.permission import Permission
from nonebot.typing import State, Handler, RuleChecker
from nonebot.typing import T_State, T_Handler, T_RuleChecker
from nonebot.rule import Rule, startswith, endswith, keyword, command, regex
plugins: Dict[str, "Plugin"] = {}
@ -101,14 +101,14 @@ class Plugin(object):
def on(type: str = "",
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
permission: Optional[Permission] = None,
*,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
"""
:说明:
@ -117,13 +117,13 @@ def on(type: str = "",
:参数:
* ``type: str``: 事件响应器类型
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -141,13 +141,13 @@ def on(type: str = "",
return matcher
def on_metaevent(rule: Optional[Union[Rule, RuleChecker]] = None,
def on_metaevent(rule: Optional[Union[Rule, T_RuleChecker]] = None,
*,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
"""
:说明:
@ -155,12 +155,12 @@ def on_metaevent(rule: Optional[Union[Rule, RuleChecker]] = None,
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -178,14 +178,14 @@ def on_metaevent(rule: Optional[Union[Rule, RuleChecker]] = None,
return matcher
def on_message(rule: Optional[Union[Rule, RuleChecker]] = None,
def on_message(rule: Optional[Union[Rule, T_RuleChecker]] = None,
permission: Optional[Permission] = None,
*,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = True,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
"""
:说明:
@ -193,13 +193,13 @@ def on_message(rule: Optional[Union[Rule, RuleChecker]] = None,
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -217,13 +217,13 @@ def on_message(rule: Optional[Union[Rule, RuleChecker]] = None,
return matcher
def on_notice(rule: Optional[Union[Rule, RuleChecker]] = None,
def on_notice(rule: Optional[Union[Rule, T_RuleChecker]] = None,
*,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
"""
:说明:
@ -231,12 +231,12 @@ def on_notice(rule: Optional[Union[Rule, RuleChecker]] = None,
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -254,13 +254,13 @@ def on_notice(rule: Optional[Union[Rule, RuleChecker]] = None,
return matcher
def on_request(rule: Optional[Union[Rule, RuleChecker]] = None,
def on_request(rule: Optional[Union[Rule, T_RuleChecker]] = None,
*,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
"""
:说明:
@ -268,12 +268,12 @@ def on_request(rule: Optional[Union[Rule, RuleChecker]] = None,
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -292,7 +292,7 @@ def on_request(rule: Optional[Union[Rule, RuleChecker]] = None,
def on_startswith(msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = None,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -302,13 +302,13 @@ def on_startswith(msg: str,
:参数:
* ``msg: str``: 指定消息开头内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -318,7 +318,7 @@ def on_startswith(msg: str,
def on_endswith(msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = None,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -328,13 +328,13 @@ def on_endswith(msg: str,
:参数:
* ``msg: str``: 指定消息结尾内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -344,7 +344,7 @@ def on_endswith(msg: str,
def on_keyword(keywords: Set[str],
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -354,13 +354,13 @@ def on_keyword(keywords: Set[str],
:参数:
* ``keywords: Set[str]``: 关键词列表
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -370,7 +370,7 @@ def on_keyword(keywords: Set[str],
def on_command(cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
**kwargs) -> Type[Matcher]:
"""
@ -383,21 +383,21 @@ def on_command(cmd: Union[str, Tuple[str, ...]],
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
- ``Type[Matcher]``
"""
async def _strip_cmd(bot, event, state: State):
async def _strip_cmd(bot, event, state: T_State):
message = event.message
event.message = message.__class__(
str(message)[len(state["_prefix"]["raw_command"]):].strip())
@ -424,13 +424,13 @@ def on_regex(pattern: str,
* ``pattern: str``: 正则表达式
* ``flags: Union[int, re.RegexFlag]``: 正则匹配标志
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -515,13 +515,13 @@ class MatcherGroup:
:参数:
* ``type: str``: 事件响应器类型
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -542,12 +542,12 @@ class MatcherGroup:
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -569,13 +569,13 @@ class MatcherGroup:
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -597,12 +597,12 @@ class MatcherGroup:
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -624,12 +624,12 @@ class MatcherGroup:
:参数:
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -645,7 +645,8 @@ class MatcherGroup:
def on_startswith(self,
msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = None,
rule: Optional[Optional[Union[Rule,
T_RuleChecker]]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -655,13 +656,13 @@ class MatcherGroup:
:参数:
* ``msg: str``: 指定消息开头内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -671,7 +672,7 @@ class MatcherGroup:
def on_endswith(self,
msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = None,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -681,13 +682,13 @@ class MatcherGroup:
:参数:
* ``msg: str``: 指定消息结尾内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -697,7 +698,7 @@ class MatcherGroup:
def on_keyword(self,
keywords: Set[str],
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
**kwargs) -> Type[Matcher]:
"""
:说明:
@ -707,13 +708,13 @@ class MatcherGroup:
:参数:
* ``keywords: Set[str]``: 关键词列表
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -723,7 +724,7 @@ class MatcherGroup:
def on_command(self,
cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None,
**kwargs) -> Type[Matcher]:
"""
@ -736,21 +737,21 @@ class MatcherGroup:
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
- ``Type[Matcher]``
"""
async def _strip_cmd(bot, event, state: State):
async def _strip_cmd(bot, event, state: T_State):
message = event.message
event.message = message.__class__(
str(message)[len(state["_prefix"]["raw_command"]):].strip())
@ -779,13 +780,13 @@ class MatcherGroup:
* ``pattern: str``: 正则表达式
* ``flags: Union[int, re.RegexFlag]``: 正则匹配标志
* ``rule: Optional[Union[Rule, RuleChecker]]``: 事件响应规则
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Permission]``: 事件响应权限
* ``handlers: Optional[List[Handler]]``: 事件处理函数列表
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器（仅执行一次）
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[State]``: 默认的 state
* ``state: Optional[T_State]``: 默认的 state
:返回:
@ -819,7 +820,7 @@ def load_plugin(module_path: str) -> Optional[Plugin]:
logger.warning(
f"Module {module_path} has been loaded by other plugins! Ignored"
)
return
return None
module = importlib.import_module(module_path)
for m in _tmp_matchers.get():
m.module = module_path
@ -859,15 +860,15 @@ def load_plugins(*plugin_dir: str) -> Set[Plugin]:
_export.set(Export())
name = module_info.name
if name.startswith("_"):
return
return None
spec = module_info.module_finder.find_spec(name, None)
if spec.name in plugins:
return
return None
elif spec.name in sys.modules:
logger.warning(
f"Module {spec.name} has been loaded by other plugin! Ignored")
return
return None
try:
module = _load(spec)

137
nonebot/plugin.pyi
View File

@ -6,7 +6,7 @@ from typing import Any, Set, List, Dict, Type, Tuple, Union, Optional
from nonebot.rule import Rule
from nonebot.matcher import Matcher
from nonebot.permission import Permission
from nonebot.typing import State, Handler, RuleChecker
from nonebot.typing import T_State, T_Handler, T_RuleChecker
plugins: Dict[str, "Plugin"] = ...
@ -34,104 +34,104 @@ class Plugin(object):
def on(type: str = ...,
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Permission] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_metaevent(rule: Optional[Union[Rule, RuleChecker]] = ...,
def on_metaevent(rule: Optional[Union[Rule, T_RuleChecker]] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_message(rule: Optional[Union[Rule, RuleChecker]] = ...,
def on_message(rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Permission] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_notice(rule: Optional[Union[Rule, RuleChecker]] = ...,
def on_notice(rule: Optional[Union[Rule, T_RuleChecker]] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_request(rule: Optional[Union[Rule, RuleChecker]] = ...,
def on_request(rule: Optional[Union[Rule, T_RuleChecker]] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_startswith(msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = ...,
*,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_endswith(msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = ...,
*,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_keyword(keywords: Set[str],
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = ...,
*,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_command(cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
*,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
@ -140,11 +140,11 @@ def on_regex(pattern: str,
rule: Optional[Rule] = ...,
*,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
@ -180,28 +180,28 @@ class CommandGroup:
def __init__(self,
cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Permission] = ...,
*,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...):
state: Optional[T_State] = ...):
self.basecmd: Tuple[str, ...] = ...
self.base_kwargs: Dict[str, Any] = ...
def command(self,
cmd: Union[str, Tuple[str, ...]],
*,
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
@ -210,115 +210,116 @@ class MatcherGroup:
def __init__(self,
*,
type: str = ...,
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...):
state: Optional[T_State] = ...):
...
def on(self,
*,
type: str = ...,
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_metaevent(self,
*,
rule: Optional[Union[Rule, RuleChecker]] = None,
handlers: Optional[List[Handler]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
...
def on_message(self,
*,
rule: Optional[Union[Rule, RuleChecker]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
permission: Optional[Permission] = None,
handlers: Optional[List[Handler]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = True,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
...
def on_notice(self,
*,
rule: Optional[Union[Rule, RuleChecker]] = None,
handlers: Optional[List[Handler]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
...
def on_request(self,
*,
rule: Optional[Union[Rule, RuleChecker]] = None,
handlers: Optional[List[Handler]] = None,
rule: Optional[Union[Rule, T_RuleChecker]] = None,
handlers: Optional[List[T_Handler]] = None,
temp: bool = False,
priority: int = 1,
block: bool = False,
state: Optional[State] = None) -> Type[Matcher]:
state: Optional[T_State] = None) -> Type[Matcher]:
...
def on_startswith(self,
*,
msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule,
T_RuleChecker]]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_endswith(self,
*,
msg: str,
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_keyword(self,
*,
keywords: Set[str],
rule: Optional[Optional[Union[Rule, RuleChecker]]] = ...,
rule: Optional[Optional[Union[Rule, T_RuleChecker]]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_command(self,
*,
cmd: Union[str, Tuple[str, ...]],
rule: Optional[Union[Rule, RuleChecker]] = ...,
rule: Optional[Union[Rule, T_RuleChecker]] = ...,
aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...
def on_regex(self,
@ -327,9 +328,9 @@ class MatcherGroup:
flags: Union[int, re.RegexFlag] = 0,
rule: Optional[Rule] = ...,
permission: Optional[Permission] = ...,
handlers: Optional[List[Handler]] = ...,
handlers: Optional[List[T_Handler]] = ...,
temp: bool = ...,
priority: int = ...,
block: bool = ...,
state: Optional[State] = ...) -> Type[Matcher]:
state: Optional[T_State] = ...) -> Type[Matcher]:
...

12
nonebot/plugins/base.py
View File

@ -1,24 +1,24 @@
from functools import reduce
from nonebot.rule import to_me
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.plugin import on_command
from nonebot.permission import SUPERUSER
from nonebot.adapters import Bot, Event, MessageSegment
from nonebot.adapters import Bot, Event, Message, MessageSegment
say = on_command("say", to_me(), permission=SUPERUSER)
@say.handle()
async def say_unescape(bot: Bot, event: Event, state: State):
Message = event.get_message().__class__
async def say_unescape(bot: Bot, event: Event, state: T_State):
MessageImpl = event.get_message().__class__
def _unescape(message: Message, segment: MessageSegment):
if segment.is_text():
return message.append(str(segment))
return message.append(segment)
message = reduce(_unescape, event.get_message(), Message()) # type: ignore
message = reduce(_unescape, event.get_message(), MessageImpl())
await bot.send(message=message, event=event)
@ -26,5 +26,5 @@ echo = on_command("echo", to_me())
@echo.handle()
async def echo_escape(bot: Bot, event: Event, state: State):
async def echo_escape(bot: Bot, event: Event, state: T_State):
await bot.send(message=event.get_message(), event=event)

29
nonebot/rule.py
View File

@ -19,7 +19,7 @@ from pygtrie import CharTrie
from nonebot import get_driver
from nonebot.log import logger
from nonebot.utils import run_sync
from nonebot.typing import State, RuleChecker
from nonebot.typing import T_State, T_RuleChecker
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
@ -43,12 +43,12 @@ class Rule:
__slots__ = ("checkers",)
def __init__(
self, *checkers: Callable[["Bot", "Event", State],
self, *checkers: Callable[["Bot", "Event", T_State],
Awaitable[bool]]) -> None:
"""
:参数:
* ``*checkers: Callable[[Bot, Event, State], Awaitable[bool]]``: **异步** RuleChecker
* ``*checkers: Callable[[Bot, Event, T_State], Awaitable[bool]]``: **异步** RuleChecker
"""
self.checkers = set(checkers)
@ -59,10 +59,11 @@ class Rule:
:类型:
* ``Set[Callable[[Bot, Event, State], Awaitable[bool]]]``
* ``Set[Callable[[Bot, Event, T_State], Awaitable[bool]]]``
"""
async def __call__(self, bot: "Bot", event: "Event", state: State) -> bool:
async def __call__(self, bot: "Bot", event: "Event",
state: T_State) -> bool:
"""
:说明:
@ -72,7 +73,7 @@ class Rule:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: State``: 当前 State
* ``state: T_State``: 当前 State
:返回:
@ -82,7 +83,7 @@ class Rule:
*map(lambda c: c(bot, event, state), self.checkers))
return all(results)
def __and__(self, other: Optional[Union["Rule", RuleChecker]]) -> "Rule":
def __and__(self, other: Optional[Union["Rule", T_RuleChecker]]) -> "Rule":
checkers = self.checkers.copy()
if other is None:
return self
@ -118,7 +119,7 @@ class TrieRule:
@classmethod
def get_value(cls, bot: "Bot", event: "Event",
state: State) -> Tuple[Dict[str, Any], Dict[str, Any]]:
state: T_State) -> Tuple[Dict[str, Any], Dict[str, Any]]:
if event.get_type() != "message":
state["_prefix"] = {"raw_command": None, "command": None}
state["_suffix"] = {"raw_command": None, "command": None}
@ -182,7 +183,7 @@ def startswith(msg: str) -> Rule:
* ``msg: str``: 消息开头字符串
"""
async def _startswith(bot: "Bot", event: "Event", state: State) -> bool:
async def _startswith(bot: "Bot", event: "Event", state: T_State) -> bool:
if event.get_type() != "message":
return False
text = event.get_plaintext()
@ -202,7 +203,7 @@ def endswith(msg: str) -> Rule:
* ``msg: str``: 消息结尾字符串
"""
async def _endswith(bot: "Bot", event: "Event", state: State) -> bool:
async def _endswith(bot: "Bot", event: "Event", state: T_State) -> bool:
if event.get_type() != "message":
return False
return event.get_plaintext().endswith(msg)
@ -221,7 +222,7 @@ def keyword(*keywords: str) -> Rule:
* ``*keywords: str``: 关键词
"""
async def _keyword(bot: "Bot", event: "Event", state: State) -> bool:
async def _keyword(bot: "Bot", event: "Event", state: T_State) -> bool:
if event.get_type() != "message":
return False
text = event.get_plaintext()
@ -269,7 +270,7 @@ def command(*cmds: Union[str, Tuple[str, ...]]) -> Rule:
for start, sep in product(command_start, command_sep):
TrieRule.add_prefix(f"{start}{sep.join(command)}", command)
async def _command(bot: "Bot", event: "Event", state: State) -> bool:
async def _command(bot: "Bot", event: "Event", state: T_State) -> bool:
return state["_prefix"]["command"] in commands
return Rule(_command)
@ -295,7 +296,7 @@ def regex(regex: str, flags: Union[int, re.RegexFlag] = 0) -> Rule:
pattern = re.compile(regex, flags)
async def _regex(bot: "Bot", event: "Event", state: State) -> bool:
async def _regex(bot: "Bot", event: "Event", state: T_State) -> bool:
if event.get_type() != "message":
return False
matched = pattern.search(str(event.get_message()))
@ -320,7 +321,7 @@ def to_me() -> Rule:
* 无
"""
async def _to_me(bot: "Bot", event: "Event", state: State) -> bool:
async def _to_me(bot: "Bot", event: "Event", state: T_State) -> bool:
return event.is_tome()
return Rule(_to_me)

67
nonebot/typing.py
View File

@ -18,8 +18,7 @@
https://docs.python.org/3/library/typing.html
"""
from functools import singledispatch
from typing import Any, Dict, Union, overload, Optional, Callable, NoReturn, Awaitable, TYPE_CHECKING
from typing import Any, Dict, Union, TypeVar, Protocol, overload, Optional, Callable, NoReturn, Awaitable, TYPE_CHECKING
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
@ -36,7 +35,7 @@ def overrides(InterfaceClass: object):
return overrider
State = Dict[Any, Any]
T_State = Dict[Any, Any]
"""
:类型: ``Dict[Any, Any]``
@ -44,49 +43,52 @@ State = Dict[Any, Any]
事件处理状态 State 类型
"""
EventPreProcessor = Callable[["Bot", "Event", State], Awaitable[None]]
T_EventPreProcessor = Callable[["Bot", "Event", T_State], Awaitable[None]]
"""
:类型: ``Callable[[Bot, Event, State], Awaitable[None]]``
:类型: ``Callable[[Bot, Event, T_State], Awaitable[None]]``
:说明:
事件预处理函数 EventPreProcessor 类型
"""
EventPostProcessor = Callable[["Bot", "Event", State], Awaitable[None]]
T_EventPostProcessor = Callable[["Bot", "Event", T_State], Awaitable[None]]
"""
:类型: ``Callable[[Bot, Event, State], Awaitable[None]]``
:类型: ``Callable[[Bot, Event, T_State], Awaitable[None]]``
:说明:
事件预处理函数 EventPostProcessor 类型
"""
RunPreProcessor = Callable[["Matcher", "Bot", "Event", State], Awaitable[None]]
T_RunPreProcessor = Callable[["Matcher", "Bot", "Event", T_State],
Awaitable[None]]
"""
:类型: ``Callable[[Matcher, Bot, Event, State], Awaitable[None]]``
:类型: ``Callable[[Matcher, Bot, Event, T_State], Awaitable[None]]``
:说明:
事件响应器运行前预处理函数 RunPreProcessor 类型
"""
RunPostProcessor = Callable[
["Matcher", Optional[Exception], "Bot", "Event", State], Awaitable[None]]
T_RunPostProcessor = Callable[
["Matcher", Optional[Exception], "Bot", "Event", T_State], Awaitable[None]]
"""
:类型: ``Callable[[Matcher, Optional[Exception], Bot, Event, State], Awaitable[None]]``
:类型: ``Callable[[Matcher, Optional[Exception], Bot, Event, T_State], Awaitable[None]]``
:说明:
事件响应器运行前预处理函数 RunPostProcessor 类型，第二个参数为运行时产生的错误（如果存在）
"""
RuleChecker = Callable[["Bot", "Event", State], Union[bool, Awaitable[bool]]]
T_RuleChecker = Callable[["Bot", "Event", T_State], Union[bool,
Awaitable[bool]]]
"""
:类型: ``Callable[[Bot, Event, State], Union[bool, Awaitable[bool]]]``
:类型: ``Callable[[Bot, Event, T_State], Union[bool, Awaitable[bool]]]``
:说明:
RuleChecker 即判断是否响应事件的处理函数。
"""
PermissionChecker = Callable[["Bot", "Event"], Union[bool, Awaitable[bool]]]
T_PermissionChecker = Callable[["Bot", "Event"], Union[bool, Awaitable[bool]]]
"""
:类型: ``Callable[[Bot, Event], Union[bool, Awaitable[bool]]]``
@ -95,40 +97,27 @@ PermissionChecker = Callable[["Bot", "Event"], Union[bool, Awaitable[bool]]]
RuleChecker 即判断是否响应消息的处理函数。
"""
# @overload
# async def Handler(bot: "Bot") -> None:
# ...
# @overload
# async def Handler(bot: "Bot", event: "Event") -> None:
# ...
# @overload
# async def Handler(bot: "Bot", state: State) -> None:
# ...
# @overload
# async def Handler(bot: Any, event: Any, state: State) -> None:
# ...
Handler = Union[Callable[["Bot", "Event", State], Union[Awaitable[None],
T_Handler = Union[Callable[[Any, Any, Any], Union[Awaitable[None],
Awaitable[NoReturn]]],
Callable[["Bot", State], Union[Awaitable[None],
Callable[[Any, Any], Union[Awaitable[None],
Awaitable[NoReturn]]],
Callable[["Bot", "Event"], Union[Awaitable[None],
Awaitable[NoReturn]]],
Callable[["Bot"], Union[Awaitable[None], Awaitable[NoReturn]]]]
Callable[[Any], Union[Awaitable[None], Awaitable[NoReturn]]]]
"""
:类型: ``Callable[[Bot, Event, State], Union[Awaitable[None], Awaitable[NoReturn]]]``
:类型:
* ``Callable[[Bot, Event, T_State], Union[Awaitable[None], Awaitable[NoReturn]]]``
* ``Callable[[Bot, Event], Union[Awaitable[None], Awaitable[NoReturn]]]``
* ``Callable[[Bot, T_State], Union[Awaitable[None], Awaitable[NoReturn]]]``
* ``Callable[[Bot], Union[Awaitable[None], Awaitable[NoReturn]]]``
:说明:
Handler 即事件的处理函数。
"""
ArgsParser = Callable[["Bot", "Event", State], Union[Awaitable[None],
T_ArgsParser = Callable[["Bot", "Event", T_State], Union[Awaitable[None],
Awaitable[NoReturn]]]
"""
:类型: ``Callable[[Bot, Event, State], Union[Awaitable[None], Awaitable[NoReturn]]]``
:类型: ``Callable[[Bot, Event, T_State], Union[Awaitable[None], Awaitable[NoReturn]]]``
:说明:

2
tests/.env.dev
View File

@ -3,7 +3,7 @@ HOST=0.0.0.0
PORT=2333
DEBUG=true
SUPERUSERS=[123123123]
SUPERUSERS=["123123123"]
NICKNAME=["bot"]
COMMAND_START=["", "/", "#"]

12
tests/test_plugins/test_delete.py
View File

@ -1,24 +1,24 @@
import asyncio
from nonebot import on_message
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.permission import USER
from nonebot.adapters import Bot, Event
a = on_message(priority=0, permission=USER(123123123), temp=True)
a = on_message(priority=0, permission=USER("123123123"), temp=True)
@a.handle()
async def test_a(bot: Bot, event: Event, state: State):
async def test_a(bot: Bot, event: Event, state: T_State):
print("======== A Received ========")
print("======== A Running Completed ========")
b = on_message(priority=0, permission=USER(123456789), temp=True)
b = on_message(priority=0, permission=USER("123456789"), temp=True)
@b.handle()
async def test_b(bot: Bot, event: Event, state: State):
async def test_b(bot: Bot, event: Event, state: T_State):
print("======== B Received ========")
await asyncio.sleep(10)
print("======== B Running Completed ========")
@ -28,5 +28,5 @@ c = on_message(priority=0, permission=USER(1111111111))
@c.handle()
async def test_c(bot: Bot, event: Event, state: State):
async def test_c(bot: Bot, event: Event, state: T_State):
print("======== C Received ========")

8
tests/test_plugins/test_group/commands.py
View File

@ -1,12 +1,10 @@
from nonebot.typing import State
from nonebot.adapters import Bot, Event
from nonebot.permission import GROUP_OWNER
from . import cmd
test_1 = cmd.command("1", aliases={"test"}, permission=GROUP_OWNER)
test_1 = cmd.command("1", aliases={"test"})
@test_1.handle()
async def test1(bot: Bot, event: Event, state: State):
await test_1.finish(event.raw_message)
async def test1(bot: Bot, event: Event):
await test_1.finish(event.get_message())

9
tests/test_plugins/test_group/matches.py
View File

@ -1,16 +1,17 @@
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.adapters import Bot, Event
from nonebot.adapters.cqhttp import HeartbeatMetaEvent
from . import match
async def heartbeat(bot: Bot, event: Event, state: State) -> bool:
return event.detail_type == "heartbeat"
async def heartbeat(bot: Bot, event: Event, state: T_State) -> bool:
return isinstance(event, HeartbeatMetaEvent)
test = match.on_metaevent(rule=heartbeat)
@test.receive()
async def handle_heartbeat(bot: Bot, event: Event, state: State):
async def handle_heartbeat(bot: Bot):
print("[i] Heartbeat")

11
tests/test_plugins/test_metaevent.py
View File

@ -1,15 +1,16 @@
from nonebot.typing import State
from nonebot.adapters import Bot, Event
from nonebot.typing import T_State
from nonebot.plugin import on_metaevent
from nonebot.adapters import Bot, Event
from nonebot.adapters.cqhttp import HeartbeatMetaEvent
async def heartbeat(bot: Bot, event: Event, state: State) -> bool:
return event.detail_type == "heartbeat"
async def heartbeat(bot: Bot, event: Event, state: T_State) -> bool:
return isinstance(event, HeartbeatMetaEvent)
test_matcher = on_metaevent(heartbeat)
@test_matcher.receive()
async def handle_heartbeat(bot: Bot, event: Event, state: dict):
async def handle_heartbeat(bot: Bot, event: Event, state: T_State):
print("[i] Heartbeat")

10
tests/test_plugins/test_package/test_command.py
View File

@ -1,14 +1,14 @@
from nonebot.rule import to_me
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.plugin import on_command
from nonebot.adapters.cqhttp import Bot, Event
from nonebot.adapters import Bot, Event
test_command = on_command("帮助", to_me())
@test_command.handle()
async def test_handler(bot: Bot, event: Event, state: State):
args = str(event.message).strip()
async def test_handler(bot: Bot, event: Event, state: T_State):
args = str(event.get_message()).strip()
print("[!] Command:", state["_prefix"], "Args:", args)
if args:
state["help"] = args
@ -17,7 +17,7 @@ async def test_handler(bot: Bot, event: Event, state: State):
@test_command.got("help", prompt="你要帮助的命令是？")
async def test_handler(bot: Bot, event: Event, state: State):
async def test_handler(bot: Bot, event: Event, state: T_State):
print("[!] Command 帮助:", state["help"])
if state["help"] not in ["test1", "test2"]:
await test_command.reject(f"{state['help']} 不支持，请重新输入！")

14
tests/test_plugins/test_permission.py
View File

@ -1,18 +1,18 @@
from nonebot.rule import to_me
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.plugin import on_startswith
from nonebot.permission import GROUP_ADMIN
from nonebot.adapters.ding import Bot as DingBot, Event as DingEvent
from nonebot.adapters.cqhttp import Bot as CQHTTPBot, Event as CQHTTPEvent
from nonebot.permission import SUPERUSER
from nonebot.adapters.ding import Bot as DingBot
from nonebot.adapters.cqhttp import Bot as CQHTTPBot
test_command = on_startswith("hello", to_me(), permission=GROUP_ADMIN)
test_command = on_startswith("hello", to_me(), permission=SUPERUSER)
@test_command.handle()
async def test_handler(bot: CQHTTPBot, event: CQHTTPEvent, state: State):
async def test_handler(bot: CQHTTPBot):
await test_command.finish("cqhttp hello")
@test_command.handle()
async def test_handler(bot: DingBot, event: DingEvent, state: State):
async def test_handler(bot: DingBot):
await test_command.finish("ding hello")

6
tests/test_plugins/test_processor.py
View File

@ -1,15 +1,15 @@
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.matcher import Matcher
from nonebot.adapters import Bot, Event
from nonebot.message import event_preprocessor, run_preprocessor
@event_preprocessor
async def handle(bot: Bot, event: Event, state: State):
async def handle(bot: Bot, event: Event, state: T_State):
state["preprocessed"] = True
print(type(event), event)
@run_preprocessor
async def run(matcher: Matcher, bot: Bot, event: Event, state: State):
async def run(matcher: Matcher, bot: Bot, event: Event, state: T_State):
print(matcher)

6
tests/test_plugins/test_weather.py
View File

@ -1,13 +1,13 @@
from nonebot import on_command
from nonebot.rule import to_me
from nonebot.typing import State
from nonebot.typing import T_State
from nonebot.adapters import Bot, Event
weather = on_command("天气", rule=to_me(), priority=1)
@weather.handle()
async def handle_first_receive(bot: Bot, event: Event, state: State):
async def handle_first_receive(bot: Bot, event: Event, state: T_State):
args = str(event.get_message()).strip() # 首次发送命令时跟随的参数，例：/天气 上海，则args为上海
print(f"==={args}===")
if args:
@ -15,7 +15,7 @@ async def handle_first_receive(bot: Bot, event: Event, state: State):
@weather.got("city", prompt="你想查询哪个城市的天气呢？")
async def handle_city(bot: Bot, state: State):
async def handle_city(bot: Bot, state: T_State):
city = state["city"]
if city not in ["上海", "北京"]:
await weather.reject("你想查询的城市暂不支持，请重新输入！")