🔀 Merge pull request #683

📝 prepare docstring for autodoc
This commit is contained in:
Ju4tCode 2022-01-13 11:14:03 +08:00 committed by GitHub
commit a8a1b4d65d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
29 changed files with 859 additions and 1767 deletions

View File

@ -1,32 +1,31 @@
"""
快捷导入
========
## 快捷导入
为方便使用``nonebot`` 模块从子模块导入了部分内容
为方便使用`nonebot` 模块从子模块导入了部分内容
- ``on_message`` => ``nonebot.plugin.on_message``
- ``on_notice`` => ``nonebot.plugin.on_notice``
- ``on_request`` => ``nonebot.plugin.on_request``
- ``on_metaevent`` => ``nonebot.plugin.on_metaevent``
- ``on_startswith`` => ``nonebot.plugin.on_startswith``
- ``on_endswith`` => ``nonebot.plugin.on_endswith``
- ``on_keyword`` => ``nonebot.plugin.on_keyword``
- ``on_command`` => ``nonebot.plugin.on_command``
- ``on_shell_command`` => ``nonebot.plugin.on_shell_command``
- ``on_regex`` => ``nonebot.plugin.on_regex``
- ``CommandGroup`` => ``nonebot.plugin.CommandGroup``
- ``Matchergroup`` => ``nonebot.plugin.MatcherGroup``
- ``load_plugin`` => ``nonebot.plugin.load_plugin``
- ``load_plugins`` => ``nonebot.plugin.load_plugins``
- ``load_all_plugins`` => ``nonebot.plugin.load_all_plugins``
- ``load_from_json`` => ``nonebot.plugin.load_from_json``
- ``load_from_toml`` => ``nonebot.plugin.load_from_toml``
- ``load_builtin_plugin`` => ``nonebot.plugin.load_builtin_plugin``
- ``load_builtin_plugins`` => ``nonebot.plugin.load_builtin_plugins``
- ``get_plugin`` => ``nonebot.plugin.get_plugin``
- ``get_loaded_plugins`` => ``nonebot.plugin.get_loaded_plugins``
- ``export`` => ``nonebot.plugin.export``
- ``require`` => ``nonebot.plugin.require``
- `on_message` => `nonebot.plugin.on_message`
- `on_notice` => `nonebot.plugin.on_notice`
- `on_request` => `nonebot.plugin.on_request`
- `on_metaevent` => `nonebot.plugin.on_metaevent`
- `on_startswith` => `nonebot.plugin.on_startswith`
- `on_endswith` => `nonebot.plugin.on_endswith`
- `on_keyword` => `nonebot.plugin.on_keyword`
- `on_command` => `nonebot.plugin.on_command`
- `on_shell_command` => `nonebot.plugin.on_shell_command`
- `on_regex` => `nonebot.plugin.on_regex`
- `CommandGroup` => `nonebot.plugin.CommandGroup`
- `Matchergroup` => `nonebot.plugin.MatcherGroup`
- `load_plugin` => `nonebot.plugin.load_plugin`
- `load_plugins` => `nonebot.plugin.load_plugins`
- `load_all_plugins` => `nonebot.plugin.load_all_plugins`
- `load_from_json` => `nonebot.plugin.load_from_json`
- `load_from_toml` => `nonebot.plugin.load_from_toml`
- `load_builtin_plugin` => `nonebot.plugin.load_builtin_plugin`
- `load_builtin_plugins` => `nonebot.plugin.load_builtin_plugins`
- `get_plugin` => `nonebot.plugin.get_plugin`
- `get_loaded_plugins` => `nonebot.plugin.get_loaded_plugins`
- `export` => `nonebot.plugin.export`
- `require` => `nonebot.plugin.require`
"""
import importlib
@ -53,24 +52,18 @@ _driver: Optional[Driver] = None
def get_driver() -> Driver:
"""
:说明:
获取全局 Driver 对象可用于在计划任务的回调中获取当前 Driver 对象
获取全局 Driver 对象可用于在计划任务的回调中获取当前 Driver 对象
返回:
Driver: 全局 Driver 对象
:返回:
* ``Driver``: 全局 Driver 对象
:异常:
* ``ValueError``: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
:用法:
.. code-block:: python
异常:
ValueError: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
用法:
```python
driver = nonebot.get_driver()
```
"""
if _driver is None:
raise ValueError("NoneBot has not been initialized.")
@ -79,24 +72,18 @@ def get_driver() -> Driver:
def get_app() -> Any:
"""
:说明:
获取全局 Driver 对应 Server App 对象
获取全局 Driver 对应 Server App 对象
返回:
Any: Server App 对象
:返回:
* ``Any``: Server App 对象
:异常:
* ``ValueError``: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
:用法:
.. code-block:: python
异常:
ValueError: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
用法:
```python
app = nonebot.get_app()
```
"""
driver = get_driver()
assert isinstance(
@ -107,24 +94,18 @@ def get_app() -> Any:
def get_asgi() -> Any:
"""
:说明:
获取全局 Driver 对应 Asgi 对象
获取全局 Driver 对应 Asgi 对象
返回:
Any: Asgi 对象
:返回:
* ``Any``: Asgi 对象
:异常:
* ``ValueError``: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
:用法:
.. code-block:: python
异常:
ValueError: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
用法:
```python
asgi = nonebot.get_asgi()
```
"""
driver = get_driver()
assert isinstance(
@ -135,31 +116,25 @@ def get_asgi() -> Any:
def get_bot(self_id: Optional[str] = None) -> Bot:
"""
:说明:
当提供 self_id 此函数是 get_bots()[self_id] 的简写当不提供时返回一个 Bot
当提供 self_id 此函数是 get_bots()[self_id] 的简写当不提供时返回一个 Bot
参数:
self_id: 用来识别 Bot ID
:参数:
返回:
Bot: Bot 对象
* ``self_id: Optional[str]``: 用来识别 Bot ID
:返回:
* ``Bot``: Bot 对象
:异常:
* ``KeyError``: 对应 ID Bot 不存在
* ``ValueError``: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
* ``ValueError``: 没有传入 ID 且没有 Bot 可用
:用法:
.. code-block:: python
异常:
KeyError: 对应 ID Bot 不存在
ValueError: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
ValueError: 没有传入 ID 且没有 Bot 可用
用法:
```python
assert nonebot.get_bot('12345') == nonebot.get_bots()['12345']
another_unspecified_bot = nonebot.get_bot()
```
"""
bots = get_bots()
if self_id is not None:
@ -173,24 +148,18 @@ def get_bot(self_id: Optional[str] = None) -> Bot:
def get_bots() -> Dict[str, Bot]:
"""
:说明:
获取所有通过 ws 连接 NoneBot Bot 对象
获取所有通过 ws 连接 NoneBot Bot 对象
返回:
Dict[str, Bot]: 一个以字符串 ID 为键Bot 对象为值的字典
:返回:
* ``Dict[str, Bot]``: 一个以字符串 ID 为键Bot 对象为值的字典
:异常:
* ``ValueError``: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
:用法:
.. code-block:: python
异常:
ValueError: 全局 Driver 对象尚未初始化 (nonebot.init 尚未调用)
用法:
```python
bots = nonebot.get_bots()
```
"""
driver = get_driver()
return driver.bots
@ -229,29 +198,20 @@ def _resolve_combine_expr(obj_str: str) -> Type[Driver]:
def init(*, _env_file: Optional[str] = None, **kwargs):
"""
:说明:
初始化 NoneBot 以及 全局 Driver 对象
初始化 NoneBot 以及 全局 Driver 对象
NoneBot 将会从 .env 文件中读取环境信息并使用相应的 env 文件配置
NoneBot 将会从 .env 文件中读取环境信息并使用相应的 env 文件配置
你也可以传入自定义的 _env_file 来指定 NoneBot 从该文件读取配置
你也可以传入自定义的 _env_file 来指定 NoneBot 从该文件读取配置
:参数:
* ``_env_file: Optional[str]``: 配置文件名默认从 .env.{env_name} 中读取配置
* ``**kwargs``: 任意变量将会存储到 Config 对象里
:返回:
- ``None``
:用法:
.. code-block:: python
参数:
_env_file: 配置文件名默认从 .env.{env_name} 中读取配置
**kwargs: 任意变量将会存储到 Config 对象里
用法:
```python
nonebot.init(database=Database(...))
```
"""
global _driver
if not _driver:
@ -277,25 +237,16 @@ def init(*, _env_file: Optional[str] = None, **kwargs):
def run(*args: Any, **kwargs: Any) -> None:
"""
:说明:
启动 NoneBot即运行全局 Driver 对象
启动 NoneBot即运行全局 Driver 对象
:参数:
* ``*args``: 传入 Driver.run 的位置参数
* ``**kwargs``: 传入 Driver.run 的命名参数
:返回:
- ``None``
:用法:
.. code-block:: python
参数:
*args: 传入 Driver.run 的位置参数
**kwargs: 传入 Driver.run 的命名参数
用法:
```python
nonebot.run(host="127.0.0.1", port=8080)
```
"""
logger.success("Running NoneBot...")
get_driver().run(*args, **kwargs)

View File

@ -1,8 +1,7 @@
"""
协议适配基类
============
## 协议适配基类
各协议请继承以下基类并使用 ``driver.register_adapter`` 注册适配器
各协议请继承以下基类并使用 `driver.register_adapter` 注册适配器
"""
from typing import Iterable

View File

@ -63,13 +63,10 @@ class Adapter(abc.ABC):
@abc.abstractmethod
async def _call_api(self, bot: Bot, api: str, **data) -> Any:
"""
:说明:
`adapter` 实际调用 api 的逻辑实现函数实现该方法以调用 api
``adapter`` 实际调用 api 的逻辑实现函数实现该方法以调用 api
:参数:
* ``api: str``: API 名称
* ``**data``: API 数据
参数:
api: API 名称
**data: API 数据
"""
raise NotImplementedError

View File

@ -27,21 +27,18 @@ class Bot(abc.ABC):
_calling_api_hook: Set[T_CallingAPIHook] = set()
"""
:类型: ``Set[T_CallingAPIHook]``
:说明: call_api 时执行的函数
call_api 时执行的函数
"""
_called_api_hook: Set[T_CalledAPIHook] = set()
"""
:类型: ``Set[T_CalledAPIHook]``
:说明: call_api 后执行的函数
call_api 后执行的函数
"""
def __init__(self, adapter: "Adapter", self_id: str):
"""
:参数:
* ``self_id: str``: 机器人 ID
* ``request: HTTPConnection``: request 连接对象
参数:
self_id: 机器人 ID
request: request 连接对象
"""
self.adapter: "Adapter" = adapter
self.self_id: str = self_id
@ -60,21 +57,17 @@ class Bot(abc.ABC):
async def call_api(self, api: str, **data: Any) -> Any:
"""
:说明:
调用机器人 API 接口可以通过该函数或直接通过 bot 属性进行调用
调用机器人 API 接口可以通过该函数或直接通过 bot 属性进行调用
:参数:
* ``api: str``: API 名称
* ``**data``: API 数据
:示例:
.. code-block:: python
参数:
api: API 名称
**data: API 数据
用法:
```python
await bot.call_api("send_msg", message="hello world")
await bot.send_msg(message="hello world")
```
"""
result: Any = None
@ -131,30 +124,23 @@ class Bot(abc.ABC):
self, event: "Event", message: Union[str, "Message", "MessageSegment"], **kwargs
) -> Any:
"""
:说明:
调用机器人基础发送消息接口
调用机器人基础发送消息接口
:参数:
* ``event: Event``: 上报事件
* ``message: Union[str, Message, MessageSegment]``: 要发送的消息
* ``**kwargs``
参数:
event: 上报事件
message: 要发送的消息
"""
raise NotImplementedError
@classmethod
def on_calling_api(cls, func: T_CallingAPIHook) -> T_CallingAPIHook:
"""
:说明:
调用 api 预处理
调用 api 预处理
:参数:
* ``bot: Bot``: 当前 bot 对象
* ``api: str``: 调用的 api 名称
* ``data: Dict[str, Any]``: api 调用的参数字典
参数:
bot: 当前 bot 对象
api: 调用的 api 名称
data: api 调用的参数字典
"""
cls._calling_api_hook.add(func)
return func
@ -162,17 +148,14 @@ class Bot(abc.ABC):
@classmethod
def on_called_api(cls, func: T_CalledAPIHook) -> T_CalledAPIHook:
"""
:说明:
调用 api 后处理
调用 api 后处理
:参数:
* ``bot: Bot``: 当前 bot 对象
* ``exception: Optional[Exception]``: 调用 api 时发生的错误
* ``api: str``: 调用的 api 名称
* ``data: Dict[str, Any]``: api 调用的参数字典
* ``result: Any``: api 调用的返回
参数:
bot: 当前 bot 对象
exception: 调用 api 时发生的错误
api: 调用的 api 名称
data: api 调用的参数字典
result: api 调用的返回
"""
cls._called_api_hook.add(func)
return func

View File

@ -16,40 +16,21 @@ class Event(abc.ABC, BaseModel):
@abc.abstractmethod
def get_type(self) -> str:
"""
:说明:
获取事件类型的方法类型通常为 NoneBot 内置的四种类型
:返回:
* ``Literal["message", "notice", "request", "meta_event"]``
* 其他自定义 ``str``
获取事件类型的方法类型通常为 NoneBot 内置的四种类型
"""
raise NotImplementedError
@abc.abstractmethod
def get_event_name(self) -> str:
"""
:说明:
获取事件名称的方法
:返回:
* ``str``
获取事件名称的方法
"""
raise NotImplementedError
@abc.abstractmethod
def get_event_description(self) -> str:
"""
:说明:
获取事件描述的方法通常为事件具体内容
:返回:
* ``str``
获取事件描述的方法通常为事件具体内容
"""
raise NotImplementedError
@ -58,80 +39,43 @@ class Event(abc.ABC, BaseModel):
def get_log_string(self) -> str:
"""
:说明:
获取事件日志信息的方法通常你不需要修改这个方法只有当希望 NoneBot 隐藏该事件日志时可以抛出 `NoLogException` 异常
获取事件日志信息的方法通常你不需要修改这个方法只有当希望 NoneBot 隐藏该事件日志时可以抛出 ``NoLogException`` 异常
:返回:
* ``str``
:异常:
- ``NoLogException``
异常:
NoLogException
"""
return f"[{self.get_event_name()}]: {self.get_event_description()}"
@abc.abstractmethod
def get_user_id(self) -> str:
"""
:说明:
获取事件主体 id 的方法通常是用户 id
:返回:
* ``str``
获取事件主体 id 的方法通常是用户 id
"""
raise NotImplementedError
@abc.abstractmethod
def get_session_id(self) -> str:
"""
:说明:
获取会话 id 的方法用于判断当前事件属于哪一个会话通常是用户 id群组 id 组合
:返回:
* ``str``
获取会话 id 的方法用于判断当前事件属于哪一个会话通常是用户 id群组 id 组合
"""
raise NotImplementedError
@abc.abstractmethod
def get_message(self) -> "Message":
"""
:说明:
获取事件消息内容的方法
:返回:
* ``Message``
获取事件消息内容的方法
"""
raise NotImplementedError
def get_plaintext(self) -> str:
"""
:说明:
获取消息纯文本的方法通常不需要修改默认通过 ``get_message().extract_plain_text`` 获取
:返回:
* ``str``
获取消息纯文本的方法通常不需要修改默认通过 `get_message().extract_plain_text` 获取
"""
return self.get_message().extract_plain_text()
@abc.abstractmethod
def is_tome(self) -> bool:
"""
:说明:
获取事件是否与机器人有关的方法
:返回:
* ``bool``
获取事件是否与机器人有关的方法
"""
raise NotImplementedError

View File

@ -26,13 +26,11 @@ class MessageSegment(Mapping, abc.ABC, Generic[TM]):
type: str
"""
- 类型: ``str``
- 说明: 消息段类型
消息段类型
"""
data: Dict[str, Any] = field(default_factory=lambda: {})
"""
- 类型: ``Dict[str, Union[str, list]]``
- 说明: 消息段数据
消息段数据
"""
@classmethod
@ -99,9 +97,8 @@ class Message(List[TMS], abc.ABC):
**kwargs,
):
"""
:参数:
* ``message: Union[str, list, dict, MessageSegment, Message, Any]``: 消息内容
参数:
message: 消息内容
"""
super().__init__(*args, **kwargs)
if message is None:
@ -116,15 +113,12 @@ class Message(List[TMS], abc.ABC):
@classmethod
def template(cls: Type[TM], format_string: Union[str, TM]) -> MessageTemplate[TM]:
"""
:说明:
根据创建消息模板, 用法和 `str.format` 大致相同, 但是可以输出消息对象, 并且支持以 `Message` 对象作为消息模板
根据创建消息模板, 用法和 ``str.format`` 大致相同, 但是可以输出消息对象, 并且支持以 ``Message`` 对象作为消息模板
并且提供了拓展的格式化控制符, 可以用适用于该消息类型的 ``MessageSegment`` 的工厂方法创建消息
:示例:
.. code-block:: python
并且提供了拓展的格式化控制符, 可以用适用于该消息类型的 `MessageSegment` 的工厂方法创建消息
用法:
```python
>>> Message.template("{} {}").format("hello", "world") # 基础演示
Message(MessageSegment(type='text', data={'text': 'hello world'}))
>>> Message.template("{} {}").format(MessageSegment.image("file///..."), "world") # 支持消息段等对象
@ -137,14 +131,13 @@ class Message(List[TMS], abc.ABC):
MessageSegment(type='text', data={'text': 'test hello world'}))
>>> Message.template("{link:image}").format(link='https://...') # 支持拓展格式化控制符
Message(MessageSegment(type='image', data={'file': 'https://...'}))
```
:参数:
参数:
format_string: 格式化字符串
* ``format_string: str``: 格式化字符串
:返回:
- ``MessageFormatter[TM]``: 消息格式化器
返回:
MessageFormatter[TM]: 消息格式化器
"""
return MessageTemplate(format_string, cls)
@ -189,13 +182,10 @@ class Message(List[TMS], abc.ABC):
def append(self: TM, obj: Union[str, TMS]) -> TM:
"""
:说明:
添加一个消息段到消息数组末尾
添加一个消息段到消息数组末尾
:参数:
* ``obj: Union[str, MessageSegment]``: 要添加的消息段
参数:
obj: 要添加的消息段
"""
if isinstance(obj, MessageSegment):
super(Message, self).append(obj)
@ -207,13 +197,10 @@ class Message(List[TMS], abc.ABC):
def extend(self: TM, obj: Union[TM, Iterable[TMS]]) -> TM:
"""
:说明:
拼接一个消息数组或多个消息段到消息数组末尾
拼接一个消息数组或多个消息段到消息数组末尾
:参数:
* ``obj: Union[Message, Iterable[MessageSegment]]``: 要添加的消息数组
参数:
obj: 要添加的消息数组
"""
for segment in obj:
self.append(segment)
@ -224,9 +211,7 @@ class Message(List[TMS], abc.ABC):
def extract_plain_text(self: "Message[MessageSegment]") -> str:
"""
:说明:
提取消息内纯文本消息
提取消息内纯文本消息
"""
return "".join(str(seg) for seg in self if seg.is_text())

View File

@ -47,14 +47,11 @@ class MessageTemplate(Formatter, Generic[TF]):
def __init__(self, template, factory=str) -> None:
"""
:说明:
创建一个模板
创建一个模板
:参数:
* ``template: Union[str, Message]``: 模板
* ``factory: Union[str, Message]``: 消息构造类型默认为 `str`
参数:
template: 模板
factory: 消息构造类型默认为 `str`
"""
self.template: TF = template
self.factory: Type[TF] = factory
@ -71,9 +68,7 @@ class MessageTemplate(Formatter, Generic[TF]):
def format(self, *args: Any, **kwargs: Any) -> TF:
"""
:说明:
根据模板和参数生成消息对象
根据模板和参数生成消息对象
"""
msg = self.factory()
if isinstance(self.template, str):

View File

@ -1,17 +1,9 @@
"""
配置
====
## 配置
NoneBot 使用 `pydantic`_ 以及 `python-dotenv`_ 来读取配置
NoneBot 使用 [`pydantic`](https://pydantic-docs.helpmanual.io/) 以及 [`python-dotenv`](https://saurabh-kumar.com/python-dotenv/) 来读取配置
配置项需符合特殊格式或 json 序列化格式详情见 `pydantic Field Type`_ 文档
.. _pydantic:
https://pydantic-docs.helpmanual.io/
.. _python-dotenv:
https://saurabh-kumar.com/python-dotenv/
.. _pydantic Field Type:
https://pydantic-docs.helpmanual.io/usage/types/
配置项需符合特殊格式或 json 序列化格式详情见 [`pydantic Field Type`](https://pydantic-docs.helpmanual.io/usage/types/) 文档
"""
import os
from pathlib import Path
@ -132,16 +124,12 @@ class Env(BaseConfig):
"""
运行环境配置大小写不敏感
将会从 ``nonebot.init 参数`` > ``环境变量`` > ``.env 环境配置文件`` 的优先级读取配置
将会从 `nonebot.init 参数` > `环境变量` > `.env 环境配置文件` 的优先级读取配置
"""
environment: str = "prod"
"""
- **类型**: ``str``
- **默认值**: ``"prod"``
:说明:
当前环境名 NoneBot 将从 ``.env.{environment}`` 文件中加载配置
当前环境名 NoneBot 将从 `.env.{environment}` 文件中加载配置
"""
class Config:
@ -153,8 +141,8 @@ class Config(BaseConfig):
"""
NoneBot 主要配置大小写不敏感
除了 NoneBot 的配置项外还可以自行添加配置项到 ``.env.{environment}`` 文件中
这些配置将会在 json 反序列化后一起带入 ``Config`` 类中
除了 NoneBot 的配置项外还可以自行添加配置项到 `.env.{environment}` 文件中
这些配置将会在 json 反序列化后一起带入 `Config` 类中
"""
_common_config: dict
@ -163,125 +151,69 @@ class Config(BaseConfig):
# nonebot configs
driver: str = "~fastapi"
"""
- **类型**: ``str``
- **默认值**: ``"~fastapi"``
NoneBot 运行所使用的 `Driver` 继承自 `nonebot.drivers.Driver`
:说明:
配置格式为 `<module>[:<Driver>][+<module>[:<Mixin>]]*`
NoneBot 运行所使用的 ``Driver`` 继承自 ``nonebot.drivers.Driver``
配置格式为 ``<module>[:<Driver>][+<module>[:<Mixin>]]*``
``~`` ``nonebot.drivers.`` 的缩写
`~` `nonebot.drivers.` 的缩写
"""
host: IPvAnyAddress = IPv4Address("127.0.0.1") # type: ignore
"""
- **类型**: ``IPvAnyAddress``
- **默认值**: ``127.0.0.1``
:说明:
NoneBot HTTP WebSocket 服务端监听的 IP/主机名
NoneBot HTTP WebSocket 服务端监听的 IP/主机名
"""
port: int = 8080
"""
- **类型**: ``int``
- **默认值**: ``8080``
:说明:
NoneBot HTTP WebSocket 服务端监听的端口
NoneBot HTTP WebSocket 服务端监听的端口
"""
log_level: Union[int, str] = "INFO"
"""
- **类型**: ``Union[int, str]``
- **默认值**: ``INFO``
:说明:
配置 NoneBot 日志输出等级可以为 ``int`` 类型等级或等级名称参考 `loguru 日志等级`_
:示例:
.. code-block:: default
配置 NoneBot 日志输出等级可以为 `int` 类型等级或等级名称参考 [`loguru 日志等级`](https://loguru.readthedocs.io/en/stable/api/logger.html#levels)。
用法:
```conf
LOG_LEVEL=25
LOG_LEVEL=INFO
.. _loguru 日志等级:
https://loguru.readthedocs.io/en/stable/api/logger.html#levels
```
"""
# bot connection configs
api_timeout: Optional[float] = 30.0
"""
- **类型**: ``Optional[float]``
- **默认值**: ``30.``
:说明:
API 请求超时时间单位:
API 请求超时时间单位:
"""
# bot runtime configs
superusers: Set[str] = set()
"""
- **类型**: ``Set[str]``
- **默认值**: ``set()``
:说明:
机器人超级用户
:示例:
.. code-block:: default
机器人超级用户
用法:
```conf
SUPERUSERS=["12345789"]
```
"""
nickname: Set[str] = set()
"""
- **类型**: ``Set[str]``
- **默认值**: ``set()``
:说明:
机器人昵称
机器人昵称
"""
command_start: Set[str] = {"/"}
"""
- **类型**: ``Set[str]``
- **默认值**: ``{"/"}``
:说明:
命令的起始标记用于判断一条消息是不是命令
命令的起始标记用于判断一条消息是不是命令
"""
command_sep: Set[str] = {"."}
"""
- **类型**: ``Set[str]``
- **默认值**: ``{"."}``
:说明:
命令的分隔标记用于将文本形式的命令切分为元组实际的命令名
命令的分隔标记用于将文本形式的命令切分为元组实际的命令名
"""
session_expire_timeout: timedelta = timedelta(minutes=2)
"""
- **类型**: ``timedelta``
- **默认值**: ``timedelta(minutes=2)``
:说明:
等待用户回复的超时时间
:示例:
.. code-block:: default
等待用户回复的超时时间
用法:
```conf
SESSION_EXPIRE_TIMEOUT=120 # 单位: 秒
SESSION_EXPIRE_TIMEOUT=[DD ][HH:MM]SS[.ffffff]
SESSION_EXPIRE_TIMEOUT=P[DD]DT[HH]H[MM]M[SS]S # ISO 8601
```
"""
# adapter configs

View File

@ -1,6 +1,5 @@
"""
依赖注入处理模块
================
## 依赖注入处理模块
该模块实现了依赖注入的定义与处理
"""

View File

@ -1,6 +1,5 @@
"""
后端驱动适配基类
=================
## 后端驱动适配基类
各驱动请继承以下基类
"""
@ -41,66 +40,51 @@ class Driver(abc.ABC):
_adapters: Dict[str, "Adapter"] = {}
"""
:类型: ``Dict[str, Adapter]``
:说明: 已注册的适配器列表
已注册的适配器列表
"""
_bot_connection_hook: Set[T_BotConnectionHook] = set()
"""
:类型: ``Set[T_BotConnectionHook]``
:说明: Bot 连接建立时执行的函数
Bot 连接建立时执行的函数
"""
_bot_disconnection_hook: Set[T_BotDisconnectionHook] = set()
"""
:类型: ``Set[T_BotDisconnectionHook]``
:说明: Bot 连接断开时执行的函数
Bot 连接断开时执行的函数
"""
def __init__(self, env: Env, config: Config):
"""
:参数:
* ``env: Env``: 包含环境信息的 Env 对象
* ``config: Config``: 包含配置信息的 Config 对象
参数:
env: 包含环境信息的 Env 对象
config: 包含配置信息的 Config 对象
"""
self.env: str = env.environment
"""
:类型: ``str``
:说明: 环境名称
环境名称
"""
self.config: Config = config
"""
:类型: ``Config``
:说明: 配置对象
配置对象
"""
self._clients: Dict[str, "Bot"] = {}
"""
:类型: ``Dict[str, Bot]``
:说明: 已连接的 Bot
已连接的 Bot
"""
@property
def bots(self) -> Dict[str, "Bot"]:
"""
:类型:
``Dict[str, Bot]``
:说明:
获取当前所有已连接的 Bot
获取当前所有已连接的 Bot
"""
return self._clients
def register_adapter(self, adapter: Type["Adapter"], **kwargs) -> None:
"""
:说明:
注册一个协议适配器
注册一个协议适配器
:参数:
* ``name: str``: 适配器名称用于在连接时进行识别
* ``adapter: Type[Bot]``: 适配器 Class
* ``**kwargs``: 其他传递给适配器的参数
参数:
name: 适配器名称用于在连接时进行识别
adapter: 适配器 Class
**kwargs: 其他传递给适配器的参数
"""
name = adapter.get_name()
if name in self._adapters:
@ -128,13 +112,7 @@ class Driver(abc.ABC):
@abc.abstractmethod
def run(self, *args, **kwargs):
"""
:说明:
启动驱动框架
:参数:
* ``*args``
* ``**kwargs``
启动驱动框架
"""
logger.opt(colors=True).debug(
f"<g>Loaded adapters: {escape_tag(', '.join(self._adapters))}</g>"
@ -152,26 +130,20 @@ class Driver(abc.ABC):
def on_bot_connect(self, func: T_BotConnectionHook) -> T_BotConnectionHook:
"""
:说明:
装饰一个函数使他在 bot 通过 WebSocket 连接成功时执行
装饰一个函数使他在 bot 通过 WebSocket 连接成功时执行
:函数参数:
* ``bot: Bot``: 当前连接上的 Bot 对象
参数:
bot: 当前连接上的 Bot 对象
"""
self._bot_connection_hook.add(func)
return func
def on_bot_disconnect(self, func: T_BotDisconnectionHook) -> T_BotDisconnectionHook:
"""
:说明:
装饰一个函数使他在 bot 通过 WebSocket 连接断开时执行
装饰一个函数使他在 bot 通过 WebSocket 连接断开时执行
:函数参数:
* ``bot: Bot``: 当前连接上的 Bot 对象
参数:
bot: 当前连接上的 Bot 对象
"""
self._bot_disconnection_hook.add(func)
return func

View File

@ -27,7 +27,7 @@ class BlockDriver(Driver):
@property
@overrides(Driver)
def type(self) -> str:
"""驱动名称: ``block_driver``"""
"""驱动名称: `block_driver`"""
return "block_driver"
@property
@ -39,13 +39,7 @@ class BlockDriver(Driver):
@overrides(Driver)
def on_startup(self, func: STARTUP_FUNC) -> STARTUP_FUNC:
"""
:说明:
注册一个启动时执行的函数
:参数:
* ``func: Callable[[], Awaitable[None]]``
注册一个启动时执行的函数
"""
self.startup_funcs.add(func)
return func
@ -53,13 +47,7 @@ class BlockDriver(Driver):
@overrides(Driver)
def on_shutdown(self, func: SHUTDOWN_FUNC) -> SHUTDOWN_FUNC:
"""
:说明:
注册一个停止时执行的函数
:参数:
* ``func: Callable[[], Awaitable[None]]``
注册一个停止时执行的函数
"""
self.shutdown_funcs.add(func)
return func

View File

@ -168,8 +168,7 @@ class WebSocket(abc.ABC):
@abc.abstractmethod
def closed(self) -> bool:
"""
:类型: ``bool``
:说明: 连接是否已经关闭
连接是否已经关闭
"""
raise NotImplementedError

View File

@ -1,6 +1,5 @@
"""
AIOHTTP 驱动适配
================
## AIOHTTP 驱动适配
本驱动仅支持客户端连接
"""

View File

@ -1,13 +1,9 @@
"""
FastAPI 驱动适配
================
## FastAPI 驱动适配
本驱动同时支持服务端以及客户端连接
后端使用方法请参考: `FastAPI 文档`_
.. _FastAPI 文档:
https://fastapi.tiangolo.com/
后端使用方法请参考: [`FastAPI 文档`](https://fastapi.tiangolo.com/)
"""
import logging
@ -48,93 +44,39 @@ class Config(BaseSettings):
fastapi_openapi_url: Optional[str] = None
"""
:类型:
``Optional[str]``
:说明:
``openapi.json`` 地址默认为 ``None`` 即关闭
`openapi.json` 地址默认为 `None` 即关闭
"""
fastapi_docs_url: Optional[str] = None
"""
:类型:
``Optional[str]``
:说明:
``swagger`` 地址默认为 ``None`` 即关闭
`swagger` 地址默认为 `None` 即关闭
"""
fastapi_redoc_url: Optional[str] = None
"""
:类型:
``Optional[str]``
:说明:
``redoc`` 地址默认为 ``None`` 即关闭
`redoc` 地址默认为 `None` 即关闭
"""
fastapi_include_adapter_schema: bool = True
"""
:类型:
``bool``
:说明:
是否包含适配器路由的 schema默认为 ``True``
是否包含适配器路由的 schema默认为 `True`
"""
fastapi_reload: bool = False
"""
:类型:
``bool``
:说明:
开启/关闭冷重载
开启/关闭冷重载
"""
fastapi_reload_dirs: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
重载监控文件夹列表默认为 uvicorn 默认值
重载监控文件夹列表默认为 uvicorn 默认值
"""
fastapi_reload_delay: Optional[float] = None
"""
:类型:
``Optional[float]``
:说明:
重载延迟默认为 uvicorn 默认值
重载延迟默认为 uvicorn 默认值
"""
fastapi_reload_includes: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
"""
fastapi_reload_excludes: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
不要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
不要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
"""
class Config:
@ -158,19 +100,19 @@ class Driver(ReverseDriver):
@property
@overrides(ReverseDriver)
def type(self) -> str:
"""驱动名称: ``fastapi``"""
"""驱动名称: `fastapi`"""
return "fastapi"
@property
@overrides(ReverseDriver)
def server_app(self) -> FastAPI:
"""``FastAPI APP`` 对象"""
"""`FastAPI APP` 对象"""
return self._server_app
@property
@overrides(ReverseDriver)
def asgi(self) -> FastAPI:
"""``FastAPI APP`` 对象"""
"""`FastAPI APP` 对象"""
return self._server_app
@property
@ -222,7 +164,7 @@ class Driver(ReverseDriver):
app: Optional[str] = None,
**kwargs,
):
"""使用 ``uvicorn`` 启动 FastAPI"""
"""使用 `uvicorn` 启动 FastAPI"""
super().run(host, port, app, **kwargs)
LOGGING_CONFIG = {
"version": 1,

View File

@ -1,11 +1,7 @@
"""
Quart 驱动适配
================
## Quart 驱动适配
后端使用方法请参考: `Quart 文档`_
.. _Quart 文档:
https://pgjones.gitlab.io/quart/index.html
后端使用方法请参考: [`Quart 文档`](https://pgjones.gitlab.io/quart/index.html)
"""
import asyncio
@ -56,53 +52,23 @@ class Config(BaseSettings):
quart_reload: bool = False
"""
:类型:
``bool``
:说明:
开启/关闭冷重载
开启/关闭冷重载
"""
quart_reload_dirs: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
重载监控文件夹列表默认为 uvicorn 默认值
重载监控文件夹列表默认为 uvicorn 默认值
"""
quart_reload_delay: Optional[float] = None
"""
:类型:
``Optional[float]``
:说明:
重载延迟默认为 uvicorn 默认值
重载延迟默认为 uvicorn 默认值
"""
quart_reload_includes: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
"""
quart_reload_excludes: Optional[List[str]] = None
"""
:类型:
``Optional[List[str]]``
:说明:
不要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
不要监听的文件列表支持 glob pattern默认为 uvicorn 默认值
"""
class Config:
@ -124,19 +90,19 @@ class Driver(ReverseDriver):
@property
@overrides(ReverseDriver)
def type(self) -> str:
"""驱动名称: ``quart``"""
"""驱动名称: `quart`"""
return "quart"
@property
@overrides(ReverseDriver)
def server_app(self) -> Quart:
"""``Quart`` 对象"""
"""`Quart` 对象"""
return self._server_app
@property
@overrides(ReverseDriver)
def asgi(self):
"""``Quart`` 对象"""
"""`Quart` 对象"""
return self._server_app
@property
@ -170,16 +136,12 @@ class Driver(ReverseDriver):
@overrides(ReverseDriver)
def on_startup(self, func: _AsyncCallable) -> _AsyncCallable:
"""参考文档: `Startup and Shutdown`_
.. _Startup and Shutdown:
https://pgjones.gitlab.io/quart/how_to_guides/startup_shutdown.html
"""
"""参考文档: [`Startup and Shutdown`](https://pgjones.gitlab.io/quart/how_to_guides/startup_shutdown.html)"""
return self.server_app.before_serving(func) # type: ignore
@overrides(ReverseDriver)
def on_shutdown(self, func: _AsyncCallable) -> _AsyncCallable:
"""参考文档: `Startup and Shutdown`_"""
"""参考文档: [`Startup and Shutdown`](https://pgjones.gitlab.io/quart/how_to_guides/startup_shutdown.html)"""
return self.server_app.after_serving(func) # type: ignore
@overrides(ReverseDriver)
@ -191,7 +153,7 @@ class Driver(ReverseDriver):
app: Optional[str] = None,
**kwargs,
):
"""使用 ``uvicorn`` 启动 Quart"""
"""使用 `uvicorn` 启动 Quart"""
super().run(host, port, app, **kwargs)
LOGGING_CONFIG = {
"version": 1,

View File

@ -1,6 +1,5 @@
"""
异常
====
## 异常
下列文档中的异常是所有 NoneBot 运行时可能会抛出的
这些异常并非所有需要用户处理 NoneBot 内部运行时被捕获并进行对应操作
@ -13,23 +12,14 @@ from pydantic.fields import ModelField
class NoneBotException(Exception):
"""
:说明:
所有 NoneBot 发生的异常基类
所有 NoneBot 发生的异常基类
"""
# Rule Exception
class ParserExit(NoneBotException):
"""
:说明:
``shell command`` 处理消息失败时返回的异常
:参数:
* ``status``
* ``message``
`shell command` 处理消息失败时返回的异常
"""
def __init__(self, status: int = 0, message: Optional[str] = None):
@ -46,21 +36,16 @@ class ParserExit(NoneBotException):
# Processor Exception
class ProcessException(NoneBotException):
"""
:说明:
事件处理过程中发生的异常基类
事件处理过程中发生的异常基类
"""
class IgnoredException(ProcessException):
"""
:说明:
指示 NoneBot 应该忽略该事件可由 PreProcessor 抛出
指示 NoneBot 应该忽略该事件可由 PreProcessor 抛出
:参数:
* ``reason``: 忽略事件的原因
参数:
reason: 忽略事件的原因
"""
def __init__(self, reason):
@ -75,13 +60,10 @@ class IgnoredException(ProcessException):
class MockApiException(ProcessException):
"""
:说明:
指示 NoneBot 阻止本次 API 调用或修改本次调用返回值并返回自定义内容可由 api hook 抛出
指示 NoneBot 阻止本次 API 调用或修改本次调用返回值并返回自定义内容可由 api hook 抛出
:参数:
* ``result``: 返回的内容
参数:
result: 返回的内容
"""
def __init__(self, result: Any):
@ -96,42 +78,32 @@ class MockApiException(ProcessException):
class StopPropagation(ProcessException):
"""
:说明:
指示 NoneBot 终止事件向下层传播
指示 NoneBot 终止事件向下层传播
:用法:
``Matcher.block == True`` 时抛出
用法:
`Matcher.block == True` 时抛出
"""
# Matcher Exceptions
class MatcherException(NoneBotException):
"""
:说明:
所有 Matcher 发生的异常基类
所有 Matcher 发生的异常基类
"""
class SkippedException(MatcherException):
"""
:说明:
指示 NoneBot 立即结束当前 `Handler` 的处理继续处理下一个 `Handler`
指示 NoneBot 立即结束当前 ``Handler`` 的处理继续处理下一个 ``Handler``
:用法:
可以在 ``Handler`` 中通过 ``Matcher.skip()`` 抛出
用法:
可以在 `Handler` 中通过 `Matcher.skip()` 抛出
"""
class TypeMisMatch(SkippedException):
"""
:说明:
当前 ``Handler`` 的参数类型不匹配
当前 `Handler` 的参数类型不匹配
"""
def __init__(self, param: ModelField, value: Any):
@ -147,53 +119,41 @@ class TypeMisMatch(SkippedException):
class PausedException(MatcherException):
"""
:说明:
指示 NoneBot 结束当前 ``Handler`` 并等待下一条消息后继续下一个 ``Handler``
指示 NoneBot 结束当前 `Handler` 并等待下一条消息后继续下一个 `Handler`
可用于用户输入新信息
:用法:
可以在 ``Handler`` 中通过 ``Matcher.pause()`` 抛出
用法:
可以在 `Handler` 中通过 `Matcher.pause()` 抛出
"""
class RejectedException(MatcherException):
"""
:说明:
指示 NoneBot 结束当前 ``Handler`` 并等待下一条消息后重新运行当前 ``Handler``
指示 NoneBot 结束当前 `Handler` 并等待下一条消息后重新运行当前 `Handler`
可用于用户重新输入
:用法:
可以在 ``Handler`` 中通过 ``Matcher.reject()`` 抛出
用法:
可以在 `Handler` 中通过 `Matcher.reject()` 抛出
"""
class FinishedException(MatcherException):
"""
:说明:
指示 NoneBot 结束当前 ``Handler`` 且后续 ``Handler`` 不再被运行
指示 NoneBot 结束当前 `Handler` 且后续 `Handler` 不再被运行
可用于结束用户会话
:用法:
可以在 ``Handler`` 中通过 ``Matcher.finish()`` 抛出
用法:
可以在 `Handler` 中通过 `Matcher.finish()` 抛出
"""
# Adapter Exceptions
class AdapterException(NoneBotException):
"""
:说明:
代表 `Adapter` 抛出的异常所有的 `Adapter` 都要在内部继承自这个 `Exception`
代表 ``Adapter`` 抛出的异常所有的 ``Adapter`` 都要在内部继承自这个 ``Exception``
:参数:
* ``adapter_name: str``: 标识 adapter
参数:
adapter_name: 标识 adapter
"""
def __init__(self, adapter_name: str) -> None:
@ -202,9 +162,7 @@ class AdapterException(NoneBotException):
class NoLogException(AdapterException):
"""
:说明:
指示 NoneBot 对当前 ``Event`` 进行处理但不显示 Log 信息可在 ``get_log_string`` 时抛出
指示 NoneBot 对当前 `Event` 进行处理但不显示 Log 信息可在 `get_log_string` 时抛出
"""
pass
@ -212,9 +170,7 @@ class NoLogException(AdapterException):
class ApiNotAvailable(AdapterException):
"""
:说明:
API 连接不可用时抛出
API 连接不可用时抛出
"""
pass
@ -222,9 +178,7 @@ class ApiNotAvailable(AdapterException):
class NetworkError(AdapterException):
"""
:说明:
在网络出现问题时抛出: API 请求地址不正确, API 请求无返回或返回状态非正常等
在网络出现问题时抛出: API 请求地址不正确, API 请求无返回或返回状态非正常等
"""
pass
@ -232,9 +186,7 @@ class NetworkError(AdapterException):
class ActionFailed(AdapterException):
"""
:说明:
API 请求成功返回数据 API 操作失败
API 请求成功返回数据 API 操作失败
"""
pass
@ -243,17 +195,13 @@ class ActionFailed(AdapterException):
# Driver Exceptions
class DriverException(NoneBotException):
"""
:说明:
``Driver`` 抛出的异常基类
`Driver` 抛出的异常基类
"""
class WebSocketClosed(DriverException):
"""
:说明:
WebSocket 连接已关闭
WebSocket 连接已关闭
"""
def __init__(self, code: int, reason: Optional[str] = None):

View File

@ -1,13 +1,11 @@
"""
日志
====
## 日志
NoneBot 使用 `loguru`_ 来记录日志信息
NoneBot 使用 [`loguru`][loguru] 来记录日志信息
自定义 logger 请参考 `loguru`_ 文档
自定义 logger 请参考 [`loguru`][loguru] 文档
.. _loguru:
https://github.com/Delgan/loguru
[loguru]: https://github.com/Delgan/loguru
"""
import sys
@ -24,21 +22,18 @@ if TYPE_CHECKING:
# logger = logging.getLogger("nonebot")
logger: "Logger" = loguru.logger
"""
:说明:
NoneBot 日志记录器对象
NoneBot 日志记录器对象
默认信息:
:默认信息:
* 格式: ``[%(asctime)s %(name)s] %(levelname)s: %(message)s``
* 等级: ``INFO`` 根据 ``config.log_level`` 配置改变
* 输出: 输出至 stdout
:用法:
.. code-block:: python
- 格式: `[%(asctime)s %(name)s] %(levelname)s: %(message)s`
- 等级: `INFO` 根据 `config.log_level` 配置改变
- 输出: 输出至 stdout
用法:
```python
from nonebot.log import logger
```
"""
# default_handler = logging.StreamHandler(sys.stdout)

View File

@ -1,6 +1,5 @@
"""
事件响应器
==========
## 事件响应器
该模块实现事件响应器的创建与运行并提供一些快捷方法来帮助用户更好的与机器人进行对话
"""
@ -66,8 +65,7 @@ T = TypeVar("T")
matchers: Dict[int, List[Type["Matcher"]]] = defaultdict(list)
"""
:类型: ``Dict[int, List[Type[Matcher]]]``
:说明: 用于存储当前所有的事件响应器
用于存储当前所有的事件响应器
"""
current_bot: ContextVar[Bot] = ContextVar("current_bot")
current_event: ContextVar[Event] = ContextVar("current_event")
@ -106,81 +104,66 @@ class Matcher(metaclass=MatcherMeta):
plugin: Optional["Plugin"] = None
"""
:类型: ``Optional[Plugin]``
:说明: 事件响应器所在插件
事件响应器所在插件
"""
module: Optional[ModuleType] = None
"""
:类型: ``Optional[ModuleType]``
:说明: 事件响应器所在插件模块
事件响应器所在插件模块
"""
plugin_name: Optional[str] = None
"""
:类型: ``Optional[str]``
:说明: 事件响应器所在插件名
事件响应器所在插件名
"""
module_name: Optional[str] = None
"""
:类型: ``Optional[str]``
:说明: 事件响应器所在点分割插件模块路径
事件响应器所在点分割插件模块路径
"""
type: str = ""
"""
:类型: ``str``
:说明: 事件响应器类型
事件响应器类型
"""
rule: Rule = Rule()
"""
:类型: ``Rule``
:说明: 事件响应器匹配规则
事件响应器匹配规则
"""
permission: Permission = Permission()
"""
:类型: ``Permission``
:说明: 事件响应器触发权限
事件响应器触发权限
"""
handlers: List[Dependent[Any]] = []
"""
:类型: ``List[Handler]``
:说明: 事件响应器拥有的事件处理函数列表
事件响应器拥有的事件处理函数列表
"""
priority: int = 1
"""
:类型: ``int``
:说明: 事件响应器优先级
事件响应器优先级
"""
block: bool = False
"""
:类型: ``bool``
:说明: 事件响应器是否阻止事件传播
事件响应器是否阻止事件传播
"""
temp: bool = False
"""
:类型: ``bool``
:说明: 事件响应器是否为临时
事件响应器是否为临时
"""
expire_time: Optional[datetime] = None
"""
:类型: ``Optional[datetime]``
:说明: 事件响应器过期时间点
事件响应器过期时间点
"""
_default_state: T_State = {}
"""
:类型: ``T_State``
:说明: 事件响应器默认状态
事件响应器默认状态
"""
_default_type_updater: Optional[Dependent[str]] = None
"""
:类型: ``Optional[Dependent]``
:说明: 事件响应器类型更新函数
事件响应器类型更新函数
"""
_default_permission_updater: Optional[Dependent[Permission]] = None
"""
:类型: ``Optional[Dependent]``
:说明: 事件响应器权限更新函数
事件响应器权限更新函数
"""
HANDLER_PARAM_TYPES = [
@ -226,27 +209,23 @@ class Matcher(metaclass=MatcherMeta):
default_permission_updater: Optional[T_PermissionUpdater] = None,
) -> Type["Matcher"]:
"""
:说明:
创建一个新的事件响应器并存储至 `matchers <#matchers>`_
创建一个新的事件响应器并存储至 `matchers <#matchers>`_
参数:
type_: 事件响应器类型 `event.get_type()` 一致时触发空字符串表示任意
rule: 匹配规则
permission: 权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器即触发一次后删除
priority: 响应优先级
block: 是否阻止事件向更低优先级的响应器传播
plugin: 事件响应器所在插件
module: 事件响应器所在模块
default_state: 默认状态 `state`
expire_time: 事件响应器最终有效时间点过时即被删除
:参数:
* ``type_: str``: 事件响应器类型 ``event.get_type()`` 一致时触发空字符串表示任意
* ``rule: Optional[Rule]``: 匹配规则
* ``permission: Optional[Permission]``: 权限
* ``handlers: Optional[List[T_Handler]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器即触发一次后删除
* ``priority: int``: 响应优先级
* ``block: bool``: 是否阻止事件向更低优先级的响应器传播
* ``plugin: Optional[Plugin]``: 事件响应器所在插件
* ``module: Optional[ModuleType]``: 事件响应器所在模块
* ``default_state: Optional[T_State]``: 默认状态 ``state``
* ``expire_time: Optional[datetime]``: 事件响应器最终有效时间点过时即被删除
:返回:
- ``Type[Matcher]``: 新的事件响应器类
返回:
Type[Matcher]: 新的事件响应器类
"""
NewMatcher = type(
"Matcher",
@ -294,18 +273,14 @@ class Matcher(metaclass=MatcherMeta):
dependency_cache: Optional[T_DependencyCache] = None,
) -> bool:
"""
:说明:
检查是否满足触发权限
检查是否满足触发权限
参数:
bot: Bot 对象
event: 上报事件
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: 上报事件
:返回:
- ``bool``: 是否满足权限
返回:
bool: 是否满足权限
"""
event_type = event.get_type()
return event_type == (cls.type or event_type) and await cls.permission(
@ -322,19 +297,15 @@ class Matcher(metaclass=MatcherMeta):
dependency_cache: Optional[T_DependencyCache] = None,
) -> bool:
"""
:说明:
检查是否满足匹配规则
检查是否满足匹配规则
参数:
bot: Bot 对象
event: 上报事件
state: 当前状态
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: 上报事件
* ``state: T_State``: 当前状态
:返回:
- ``bool``: 是否满足匹配规则
返回:
bool: 是否满足匹配规则
"""
event_type = event.get_type()
return event_type == (cls.type or event_type) and await cls.rule(
@ -344,13 +315,10 @@ class Matcher(metaclass=MatcherMeta):
@classmethod
def type_updater(cls, func: T_TypeUpdater) -> T_TypeUpdater:
"""
:说明:
装饰一个函数来更改当前事件响应器的默认响应事件类型更新函数
装饰一个函数来更改当前事件响应器的默认响应事件类型更新函数
:参数:
* ``func: T_TypeUpdater``: 响应事件类型更新函数
参数:
func: 响应事件类型更新函数
"""
cls._default_type_updater = Dependent[str].parse(
call=func, allow_types=cls.HANDLER_PARAM_TYPES
@ -360,13 +328,10 @@ class Matcher(metaclass=MatcherMeta):
@classmethod
def permission_updater(cls, func: T_PermissionUpdater) -> T_PermissionUpdater:
"""
:说明:
装饰一个函数来更改当前事件响应器的默认会话权限更新函数
装饰一个函数来更改当前事件响应器的默认会话权限更新函数
:参数:
* ``func: T_PermissionUpdater``: 会话权限更新函数
参数:
func: 会话权限更新函数
"""
cls._default_permission_updater = Dependent[Permission].parse(
call=func, allow_types=cls.HANDLER_PARAM_TYPES
@ -390,13 +355,10 @@ class Matcher(metaclass=MatcherMeta):
cls, parameterless: Optional[List[Any]] = None
) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
装饰一个函数来向事件响应器直接添加一个处理函数
装饰一个函数来向事件响应器直接添加一个处理函数
:参数:
* ``parameterless: Optional[List[Any]]``: 非参数类型依赖列表
参数:
parameterless: 非参数类型依赖列表
"""
def _decorator(func: T_Handler) -> T_Handler:
@ -410,14 +372,11 @@ class Matcher(metaclass=MatcherMeta):
cls, id: str = "", parameterless: Optional[List[Any]] = None
) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
装饰一个函数来指示 NoneBot 在接收用户新的一条消息后继续运行该函数
装饰一个函数来指示 NoneBot 在接收用户新的一条消息后继续运行该函数
:参数:
* ``id: str``: 消息 ID
* ``parameterless: Optional[List[Any]]``: 非参数类型依赖列表
参数:
id: 消息 ID
parameterless: 非参数类型依赖列表
"""
async def _receive(event: Event, matcher: "Matcher") -> Union[None, NoReturn]:
@ -452,16 +411,12 @@ class Matcher(metaclass=MatcherMeta):
parameterless: Optional[List[Any]] = None,
) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
装饰一个函数来指示 NoneBot 当要获取的 `key` 不存在时接收用户新的一条消息再运行该函数如果 `key` 已存在则直接继续运行
装饰一个函数来指示 NoneBot 当要获取的 ``key`` 不存在时接收用户新的一条消息并经过 ``ArgsParser`` 处理后再运行该函数如果 ``key`` 已存在则直接继续运行
:参数:
* ``key: str``: 参数名
* ``prompt: Optional[Union[str, Message, MessageSegment, MessageFormatter]]``: 在参数不存在时向用户发送的消息
* ``args_parser: Optional[T_ArgsParser]``: 可选参数解析函数空则使用默认解析函数
* ``parameterless: Optional[List[Any]]``: 非参数类型依赖列表
参数:
key: 参数名
prompt: 在参数不存在时向用户发送的消息
parameterless: 非参数类型依赖列表
"""
async def _key_getter(event: Event, matcher: "Matcher"):
@ -498,14 +453,11 @@ class Matcher(metaclass=MatcherMeta):
**kwargs: Any,
) -> Any:
"""
:说明:
发送一条消息给当前交互用户
发送一条消息给当前交互用户
:参数:
* ``message: Union[str, Message, MessageSegment]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
message: 消息内容
**kwargs: `bot.send` 的参数请参考对应 adapter bot 对象 api
"""
bot = current_bot.get()
event = current_event.get()
@ -523,14 +475,11 @@ class Matcher(metaclass=MatcherMeta):
**kwargs,
) -> NoReturn:
"""
:说明:
发送一条消息给当前交互用户并结束当前事件响应器
发送一条消息给当前交互用户并结束当前事件响应器
:参数:
* ``message: Union[str, Message, MessageSegment, MessageTemplate]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
message: 消息内容
**kwargs: `bot.send` 的参数请参考对应 adapter bot 对象 api
"""
if message is not None:
await cls.send(message, **kwargs)
@ -543,14 +492,11 @@ class Matcher(metaclass=MatcherMeta):
**kwargs,
) -> NoReturn:
"""
:说明:
发送一条消息给当前交互用户并暂停事件响应器在接收用户新的一条消息后继续下一个处理函数
发送一条消息给当前交互用户并暂停事件响应器在接收用户新的一条消息后继续下一个处理函数
:参数:
* ``prompt: Union[str, Message, MessageSegment, MessageTemplate]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
prompt: 消息内容
**kwargs`: bot.send` 的参数请参考对应 adapter bot 对象 api
"""
if prompt is not None:
await cls.send(prompt, **kwargs)
@ -563,15 +509,11 @@ class Matcher(metaclass=MatcherMeta):
**kwargs,
) -> NoReturn:
"""
:说明:
最近使用 `got` / `receive` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器在接收用户新的一条消息后继续当前处理函数
最近使用 ``got`` / ``receive`` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器
在接收用户新的一条消息后继续当前处理函数
:参数:
* ``prompt: Union[str, Message, MessageSegment, MessageTemplate]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
prompt: 消息内容
**kwargs: `bot.send` 的参数请参考对应 adapter bot 对象 api
"""
if prompt is not None:
await cls.send(prompt, **kwargs)
@ -585,16 +527,12 @@ class Matcher(metaclass=MatcherMeta):
**kwargs,
) -> NoReturn:
"""
:说明:
最近使用 `got` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器在接收用户新的一条消息后继续当前处理函数
最近使用 ``got`` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器
在接收用户新的一条消息后继续当前处理函数
:参数:
* ``key: str``: 参数名
* ``prompt: Union[str, Message, MessageSegment, MessageTemplate]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
key: 参数名
prompt: 消息内容
**kwargs: `bot.send` 的参数请参考对应 adapter bot 对象 api
"""
matcher = current_matcher.get()
matcher.set_target(ARG_KEY.format(key=key))
@ -610,16 +548,12 @@ class Matcher(metaclass=MatcherMeta):
**kwargs,
) -> NoReturn:
"""
:说明:
最近使用 `got` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器在接收用户新的一条消息后继续当前处理函数
最近使用 ``got`` 接收的消息不符合预期发送一条消息给当前交互用户并暂停事件响应器
在接收用户新的一条消息后继续当前处理函数
:参数:
* ``id: str``: 消息 id
* ``prompt: Union[str, Message, MessageSegment, MessageTemplate]``: 消息内容
* ``**kwargs``: 其他传递给 ``bot.send`` 的参数请参考对应 adapter bot 对象 api
参数:
id: 消息 id
prompt: 消息内容
**kwargs: `bot.send` 的参数请参考对应 adapter bot 对象 api
"""
matcher = current_matcher.get()
matcher.set_target(RECEIVE_KEY.format(id=id))
@ -658,9 +592,7 @@ class Matcher(metaclass=MatcherMeta):
def stop_propagation(self):
"""
:说明:
阻止事件传播
阻止事件传播
"""
self.block = True

View File

@ -1,6 +1,5 @@
"""
事件处理
========
## 事件处理
NoneBot 内部处理并按优先级分发事件给所有事件响应器提供了多个插槽以进行事件的预处理等
"""
@ -69,9 +68,7 @@ RUN_POSTPCS_PARAMS = [
def event_preprocessor(func: T_EventPreProcessor) -> T_EventPreProcessor:
"""
:说明:
事件预处理装饰一个函数使它在每次接收到事件并分发给各响应器之前执行
事件预处理装饰一个函数使它在每次接收到事件并分发给各响应器之前执行
"""
_event_preprocessors.add(
Dependent[None].parse(call=func, allow_types=EVENT_PCS_PARAMS)
@ -81,9 +78,7 @@ def event_preprocessor(func: T_EventPreProcessor) -> T_EventPreProcessor:
def event_postprocessor(func: T_EventPostProcessor) -> T_EventPostProcessor:
"""
:说明:
事件后处理装饰一个函数使它在每次接收到事件并分发给各响应器之后执行
事件后处理装饰一个函数使它在每次接收到事件并分发给各响应器之后执行
"""
_event_postprocessors.add(
Dependent[None].parse(call=func, allow_types=EVENT_PCS_PARAMS)
@ -93,9 +88,7 @@ def event_postprocessor(func: T_EventPostProcessor) -> T_EventPostProcessor:
def run_preprocessor(func: T_RunPreProcessor) -> T_RunPreProcessor:
"""
:说明:
运行预处理装饰一个函数使它在每次事件响应器运行前执行
运行预处理装饰一个函数使它在每次事件响应器运行前执行
"""
_run_preprocessors.add(
Dependent[None].parse(call=func, allow_types=RUN_PREPCS_PARAMS)
@ -105,9 +98,7 @@ def run_preprocessor(func: T_RunPreProcessor) -> T_RunPreProcessor:
def run_postprocessor(func: T_RunPostProcessor) -> T_RunPostProcessor:
"""
:说明:
运行后处理装饰一个函数使它在每次事件响应器运行后执行
运行后处理装饰一个函数使它在每次事件响应器运行后执行
"""
_run_postprocessors.add(
Dependent[None].parse(call=func, allow_types=RUN_POSTPCS_PARAMS)
@ -242,21 +233,17 @@ async def _run_matcher(
async def handle_event(bot: "Bot", event: "Event") -> None:
"""
:说明:
处理一个事件调用该函数以实现分发事件
处理一个事件调用该函数以实现分发事件
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
:示例:
.. code-block:: python
参数:
bot: Bot 对象
event: Event 对象
用法:
```python
import asyncio
asyncio.create_task(handle_event(bot, event))
```
"""
show_log = True
log_msg = f"<m>{escape_tag(bot.type.upper())} {escape_tag(bot.self_id)}</m> | "

View File

@ -56,17 +56,14 @@ def Depends(
use_cache: bool = True,
) -> Any:
"""
:说明:
参数依赖注入装饰器
参数依赖注入装饰器
:参数:
* ``dependency: Optional[Callable[..., Any]] = None``: 依赖函数默认为参数的类型注释
* ``use_cache: bool = True``: 是否使用缓存默认为 ``True``
.. code-block:: python
参数:
dependency: 依赖函数默认为参数的类型注释
use_cache: 是否使用缓存默认为 `True`
用法:
```python
def depend_func() -> Any:
return ...
@ -78,6 +75,7 @@ def Depends(
async def handler(param_name: Any = Depends(depend_func), gen: Any = Depends(depend_gen_func)):
...
```
"""
return DependsInner(dependency, use_cache=use_cache)

View File

@ -1,12 +1,11 @@
r"""
权限
====
"""
## 权限
每个 ``Matcher`` 拥有一个 ``Permission`` 其中是 ``PermissionChecker`` 的集合只要有一个 ``PermissionChecker`` 检查结果为 ``True`` 时就会继续运行
每个 `Matcher` 拥有一个 `Permission` 其中是 `PermissionChecker` 的集合只要有一个 `PermissionChecker` 检查结果为 `True` 时就会继续运行
\:\:\:tip 提示
``PermissionChecker`` 既可以是 async function 也可以是 sync function
\:\:\:
::: tip 提示
`PermissionChecker` 既可以是 async function 也可以是 sync function
:::
"""
import asyncio
@ -35,18 +34,15 @@ async def _run_coro_with_catch(coro: Coroutine[Any, Any, Any]):
class Permission:
"""
:说明:
``Matcher`` 规则类当事件传递时 ``Matcher`` 运行前进行检查
:示例:
.. code-block:: python
`Matcher` 规则类当事件传递时 `Matcher` 运行前进行检查
用法:
```python
Permission(async_function) | sync_function
# 等价于
from nonebot.utils import run_sync
Permission(async_function, run_sync(sync_function))
```
"""
__slots__ = ("checkers",)
@ -60,9 +56,8 @@ class Permission:
def __init__(self, *checkers: Union[T_PermissionChecker, Dependent[bool]]) -> None:
"""
:参数:
* ``*checkers: Union[T_PermissionChecker, Dependent[bool]``: PermissionChecker
参数:
*checkers: PermissionChecker
"""
self.checkers: Set[Dependent[bool]] = set(
@ -74,13 +69,7 @@ class Permission:
for checker in checkers
)
"""
:说明:
存储 ``PermissionChecker``
:类型:
* ``Set[Dependent[bool]]``
存储 `PermissionChecker`
"""
async def __call__(
@ -91,20 +80,13 @@ class Permission:
dependency_cache: Optional[T_DependencyCache] = None,
) -> bool:
"""
:说明:
检查是否满足某个权限
检查是否满足某个权限
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``stack: Optional[AsyncExitStack]``: 异步上下文栈
* ``dependency_cache: Optional[CacheDict[T_Handler, Any]]``: 依赖缓存
:返回:
- ``bool``
参数:
bot: Bot 对象
event: Event 对象
stack: 异步上下文栈
dependency_cache: 依赖缓存
"""
if not self.checkers:
return True
@ -159,19 +141,19 @@ class MetaEvent:
MESSAGE = Permission(Message())
"""
- **说明**: 匹配任意 ``message`` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 message type Matcher
匹配任意 `message` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 message type Matcher
"""
NOTICE = Permission(Notice())
"""
- **说明**: 匹配任意 ``notice`` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 notice type Matcher
匹配任意 `notice` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 notice type Matcher
"""
REQUEST = Permission(Request())
"""
- **说明**: 匹配任意 ``request`` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 request type Matcher
匹配任意 `request` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 request type Matcher
"""
METAEVENT = Permission(MetaEvent())
"""
- **说明**: 匹配任意 ``meta_event`` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 meta_event type Matcher
匹配任意 `meta_event` 类型事件仅在需要同时捕获不同类型事件时使用优先使用 meta_event type Matcher
"""
@ -191,14 +173,11 @@ class User:
def USER(*users: str, perm: Optional[Permission] = None):
"""
:说明:
`event` `session_id` 在白名单内且满足 perm
``event`` ``session_id`` 在白名单内且满足 perm
:参数:
* ``*user: str``: 白名单
* ``perm: Optional[Permission]``: 需要同时满足的权限
参数:
*user: 白名单
perm: 需要同时满足的权限
"""
return Permission(User(users, perm))
@ -215,5 +194,5 @@ class SuperUser:
SUPERUSER = Permission(SuperUser())
"""
- **说明**: 匹配任意超级用户消息类型事件
匹配任意超级用户消息类型事件
"""

View File

@ -1,6 +1,5 @@
"""
插件
====
## 插件
NoneBot 插件开发提供便携的定义函数
"""

View File

@ -3,14 +3,10 @@ from . import _current_plugin
class Export(dict):
"""
:说明:
插件导出内容以使得其他插件可以获得
:示例:
.. code-block:: python
插件导出内容以使得其他插件可以获得
用法:
```python
nonebot.export().default = "bar"
@nonebot.export()
@ -25,6 +21,7 @@ class Export(dict):
@nonebot.export().sub
def something_else():
pass
```
"""
def __call__(self, func, **kwargs):
@ -46,13 +43,7 @@ class Export(dict):
def export() -> Export:
"""
:说明:
获取插件的导出内容对象
:返回:
- ``Export``
获取插件的导出内容对象
"""
plugin = _current_plugin.get()
if not plugin:

View File

@ -12,17 +12,10 @@ from .plugin import Plugin, get_plugin
def load_plugin(module_path: str) -> Optional[Plugin]:
"""
:说明:
使用 `PluginManager` 加载单个插件可以是本地插件或是通过 `pip` 安装的插件
使用 ``PluginManager`` 加载单个插件可以是本地插件或是通过 ``pip`` 安装的插件
:参数:
* ``module_path: str``: 插件名称 ``path.to.your.plugin``
:返回:
- ``Optional[Plugin]``
参数:
module_path: 插件名称 `path.to.your.plugin`
"""
manager = PluginManager([module_path])
@ -32,17 +25,10 @@ def load_plugin(module_path: str) -> Optional[Plugin]:
def load_plugins(*plugin_dir: str) -> Set[Plugin]:
"""
:说明:
导入目录下多个插件 `_` 开头的插件不会被导入
导入目录下多个插件 ``_`` 开头的插件不会被导入
:参数:
- ``*plugin_dir: str``: 插件路径
:返回:
- ``Set[Plugin]``
参数:
plugin_dir: 插件路径
"""
manager = PluginManager(search_path=plugin_dir)
_managers.append(manager)
@ -53,18 +39,11 @@ def load_all_plugins(
module_path: Iterable[str], plugin_dir: Iterable[str]
) -> Set[Plugin]:
"""
:说明:
导入指定列表中的插件以及指定目录下多个插件 `_` 开头的插件不会被导入
导入指定列表中的插件以及指定目录下多个插件 ``_`` 开头的插件不会被导入
:参数:
- ``module_path: Iterable[str]``: 指定插件集合
- ``plugin_dir: Iterable[str]``: 指定插件路径集合
:返回:
- ``Set[Plugin]``
参数:
module_path: 指定插件集合
plugin_dir: 指定插件路径集合
"""
manager = PluginManager(module_path, plugin_dir)
_managers.append(manager)
@ -73,18 +52,11 @@ def load_all_plugins(
def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
"""
:说明:
导入指定 json 文件中的 `plugins` 以及 `plugin_dirs` 下多个插件 `_` 开头的插件不会被导入
导入指定 json 文件中的 ``plugins`` 以及 ``plugin_dirs`` 下多个插件 ``_`` 开头的插件不会被导入
:参数:
- ``file_path: str``: 指定 json 文件路径
- ``encoding: str``: 指定 json 文件编码
:返回:
- ``Set[Plugin]``
参数:
file_path: 指定 json 文件路径
encoding: 指定 json 文件编码
"""
with open(file_path, "r", encoding=encoding) as f:
data = json.load(f)
@ -97,19 +69,12 @@ def load_from_json(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
def load_from_toml(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
"""
:说明:
导入指定 toml 文件 `[tool.nonebot]` 中的 `plugins` 以及 `plugin_dirs` 下多个插件
`_` 开头的插件不会被导入
导入指定 toml 文件 ``[tool.nonebot]`` 中的 ``plugins`` 以及 ``plugin_dirs`` 下多个插件
``_`` 开头的插件不会被导入
:参数:
- ``file_path: str``: 指定 toml 文件路径
- ``encoding: str``: 指定 toml 文件编码
:返回:
- ``Set[Plugin]``
参数:
file_path: 指定 toml 文件路径
encoding: 指定 toml 文件编码
"""
with open(file_path, "r", encoding=encoding) as f:
data = tomlkit.parse(f.read()) # type: ignore
@ -133,46 +98,27 @@ def load_from_toml(file_path: str, encoding: str = "utf-8") -> Set[Plugin]:
def load_builtin_plugin(name: str) -> Optional[Plugin]:
"""
:说明:
导入 NoneBot 内置插件
:返回:
- ``Plugin``
导入 NoneBot 内置插件
"""
return load_plugin(f"nonebot.plugins.{name}")
def load_builtin_plugins(*plugins) -> Set[Plugin]:
"""
:说明:
导入多个 NoneBot 内置插件
:返回:
- ``Set[Plugin]``
导入多个 NoneBot 内置插件
"""
return load_all_plugins([f"nonebot.plugins.{p}" for p in plugins], [])
def require(name: str) -> Export:
"""
:说明:
获取一个插件的导出内容
获取一个插件的导出内容
参数:
name: 插件名 `load_plugin` 参数一致如果为 `load_plugins` 导入的插件则为文件()
:参数:
* ``name: str``: 插件名 ``load_plugin`` 参数一致如果为 ``load_plugins`` 导入的插件则为文件()
:返回:
- ``Export``
:异常:
- ``RuntimeError``: 插件无法加载
异常:
RuntimeError: 插件无法加载
"""
plugin = get_plugin(name) or load_plugin(name)
if not plugin:

View File

@ -55,24 +55,17 @@ def on(
_depth: int = 0,
) -> Type[Matcher]:
"""
:说明:
注册一个基础事件响应器可自定义类型
注册一个基础事件响应器可自定义类型
:参数:
* ``type: str``: 事件响应器类型
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
type: 事件响应器类型
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
matcher = Matcher.new(
type,
@ -101,22 +94,15 @@ def on_metaevent(
_depth: int = 0,
) -> Type[Matcher]:
"""
:说明:
注册一个元事件响应器
注册一个元事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
matcher = Matcher.new(
"meta_event",
@ -146,23 +132,16 @@ def on_message(
_depth: int = 0,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器
注册一个消息事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
matcher = Matcher.new(
"message",
@ -191,22 +170,15 @@ def on_notice(
_depth: int = 0,
) -> Type[Matcher]:
"""
:说明:
注册一个通知事件响应器
注册一个通知事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
matcher = Matcher.new(
"notice",
@ -235,22 +207,15 @@ def on_request(
_depth: int = 0,
) -> Type[Matcher]:
"""
:说明:
注册一个请求事件响应器
注册一个请求事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
matcher = Matcher.new(
"request",
@ -276,25 +241,18 @@ def on_startswith(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息的**文本部分**以指定内容开头时响应
注册一个消息事件响应器并且当消息的**文本部分**以指定内容开头时响应
:参数:
* ``msg: Union[str, Tuple[str, ...]]``: 指定消息开头内容
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``ignorecase: bool``: 是否忽略大小写
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
msg: 指定消息开头内容
rule: 事件响应规则
ignorecase: 是否忽略大小写
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
return on_message(startswith(msg, ignorecase) & rule, **kwargs, _depth=_depth + 1)
@ -307,25 +265,18 @@ def on_endswith(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息的**文本部分**以指定内容结尾时响应
注册一个消息事件响应器并且当消息的**文本部分**以指定内容结尾时响应
:参数:
* ``msg: Union[str, Tuple[str, ...]]``: 指定消息结尾内容
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``ignorecase: bool``: 是否忽略大小写
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
msg: 指定消息结尾内容
rule: 事件响应规则
ignorecase: 是否忽略大小写
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
return on_message(endswith(msg, ignorecase) & rule, **kwargs, _depth=_depth + 1)
@ -337,24 +288,17 @@ def on_keyword(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息纯文本部分包含关键词时响应
注册一个消息事件响应器并且当消息纯文本部分包含关键词时响应
:参数:
* ``keywords: Set[str]``: 关键词列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
keywords: 关键词列表
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
return on_message(keyword(*keywords) & rule, **kwargs, _depth=_depth + 1)
@ -367,27 +311,20 @@ def on_command(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息以指定命令开头时响应
注册一个消息事件响应器并且当消息以指定命令开头时响应
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
cmd: 指定命令内容
rule: 事件响应规则
aliases: 命令别名
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
commands = set([cmd]) | (aliases or set())
@ -406,30 +343,23 @@ def on_shell_command(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个支持 `shell_like` 解析参数的命令消息事件响应器
注册一个支持 ``shell_like`` 解析参数的命令消息事件响应器
与普通的 `on_command` 不同的是在添加 `parser` 参数时, 响应器会自动处理消息
与普通的 ``on_command`` 不同的是在添加 ``parser`` 参数时, 响应器会自动处理消息
并将用户输入的原始参数列表保存在 `state["argv"]`, `parser` 处理的参数保存在 `state["args"]`
并将用户输入的原始参数列表保存在 ``state["argv"]``, ``parser`` 处理的参数保存在 ``state["args"]``
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``parser: Optional[ArgumentParser]``: ``nonebot.rule.ArgumentParser`` 对象
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
cmd: 指定命令内容
rule: 事件响应规则
aliases: 命令别名
parser: `nonebot.rule.ArgumentParser` 对象
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
commands = set([cmd]) | (aliases or set())
@ -448,27 +378,20 @@ def on_regex(
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息匹配正则表达式时响应
注册一个消息事件响应器并且当消息匹配正则表达式时响应
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
:参数:
* ``pattern: str``: 正则表达式
* ``flags: Union[int, re.RegexFlag]``: 正则匹配标志
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
pattern: 正则表达式
flags: 正则匹配标志
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
return on_message(regex(pattern, flags) & rule, **kwargs, _depth=_depth + 1)
@ -478,38 +401,28 @@ class CommandGroup:
def __init__(self, cmd: Union[str, Tuple[str, ...]], **kwargs):
"""
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 命令前缀
* ``**kwargs``: 其他传递给 ``on_command`` 的参数默认值参考 `on_command <#on-command-cmd-rule-none-aliases-none-kwargs>`_
参数:
cmd: 命令前缀
**kwargs: `on_command` 的参数默认值参考 `on_command <#on-command-cmd-rule-none-aliases-none-kwargs>`_
"""
self.basecmd: Tuple[str, ...] = (cmd,) if isinstance(cmd, str) else cmd
"""
- **类型**: ``Tuple[str, ...]``
- **说明**: 命令前缀
命令前缀
"""
if "aliases" in kwargs:
del kwargs["aliases"]
self.base_kwargs: Dict[str, Any] = kwargs
"""
- **类型**: ``Dict[str, Any]``
- **说明**: 其他传递给 ``on_command`` 的参数默认值
其他传递给 `on_command` 的参数默认值
"""
def command(self, cmd: Union[str, Tuple[str, ...]], **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个新的命令
注册一个新的命令
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 命令前缀
* ``**kwargs``: 其他传递给 ``on_command`` 的参数将会覆盖命令组默认值
:返回:
- ``Type[Matcher]``
参数:
cmd: 命令前缀
**kwargs: `on_command` 的参数将会覆盖命令组默认值
"""
sub_cmd = (cmd,) if isinstance(cmd, str) else cmd
cmd = self.basecmd + sub_cmd
@ -522,18 +435,11 @@ class CommandGroup:
self, cmd: Union[str, Tuple[str, ...]], **kwargs
) -> Type[Matcher]:
"""
:说明:
注册一个新的命令
注册一个新的命令
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 命令前缀
* ``**kwargs``: 其他传递给 ``on_shell_command`` 的参数将会覆盖命令组默认值
:返回:
- ``Type[Matcher]``
参数:
cmd: 命令前缀
**kwargs: `on_shell_command` 的参数将会覆盖命令组默认值
"""
sub_cmd = (cmd,) if isinstance(cmd, str) else cmd
cmd = self.basecmd + sub_cmd
@ -544,45 +450,34 @@ class CommandGroup:
class MatcherGroup:
"""事件响应器组合,统一管理。为 ``Matcher`` 创建提供默认属性。"""
"""事件响应器组合,统一管理。为 `Matcher` 创建提供默认属性。"""
def __init__(self, **kwargs):
"""
:说明:
创建一个事件响应器组合参数为默认值 ``on`` 一致
创建一个事件响应器组合参数为默认值 `on` 一致
"""
self.matchers: List[Type[Matcher]] = []
"""
:类型: ``List[Type[Matcher]]``
:说明: 组内事件响应器列表
组内事件响应器列表
"""
self.base_kwargs: Dict[str, Any] = kwargs
"""
- **类型**: ``Dict[str, Any]``
- **说明**: 其他传递给 ``on`` 的参数默认值
其他传递给 `on` 的参数默认值
"""
def on(self, **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个基础事件响应器可自定义类型
注册一个基础事件响应器可自定义类型
:参数:
* ``type: str``: 事件响应器类型
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
type: 事件响应器类型
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -592,22 +487,15 @@ class MatcherGroup:
def on_metaevent(self, **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个元事件响应器
注册一个元事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -619,23 +507,16 @@ class MatcherGroup:
def on_message(self, **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器
注册一个消息事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -646,22 +527,15 @@ class MatcherGroup:
def on_notice(self, **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个通知事件响应器
注册一个通知事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -672,22 +546,15 @@ class MatcherGroup:
def on_request(self, **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个请求事件响应器
注册一个请求事件响应器
:参数:
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
rule: 事件响应规则
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -700,25 +567,18 @@ class MatcherGroup:
self, msg: Union[str, Tuple[str, ...]], **kwargs
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息的**文本部分**以指定内容开头时响应
注册一个消息事件响应器并且当消息的**文本部分**以指定内容开头时响应
:参数:
* ``msg: Union[str, Tuple[str, ...]]``: 指定消息开头内容
* ``ignorecase: bool``: 是否忽略大小写
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
msg: 指定消息开头内容
ignorecase: 是否忽略大小写
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -729,25 +589,18 @@ class MatcherGroup:
def on_endswith(self, msg: Union[str, Tuple[str, ...]], **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息的**文本部分**以指定内容结尾时响应
注册一个消息事件响应器并且当消息的**文本部分**以指定内容结尾时响应
:参数:
* ``msg: Union[str, Tuple[str, ...]]``: 指定消息结尾内容
* ``ignorecase: bool``: 是否忽略大小写
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
msg: 指定消息结尾内容
ignorecase: 是否忽略大小写
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -758,24 +611,17 @@ class MatcherGroup:
def on_keyword(self, keywords: Set[str], **kwargs) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息纯文本部分包含关键词时响应
注册一个消息事件响应器并且当消息纯文本部分包含关键词时响应
:参数:
* ``keywords: Set[str]``: 关键词列表
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
keywords: 关键词列表
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -791,27 +637,20 @@ class MatcherGroup:
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息以指定命令开头时响应
注册一个消息事件响应器并且当消息以指定命令开头时响应
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
命令匹配规则参考: `命令形式匹配 <rule.md#command-command>`_
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
cmd: 指定命令内容
aliases: 命令别名
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -828,30 +667,23 @@ class MatcherGroup:
**kwargs,
) -> Type[Matcher]:
"""
:说明:
注册一个支持 `shell_like` 解析参数的命令消息事件响应器
注册一个支持 ``shell_like`` 解析参数的命令消息事件响应器
与普通的 `on_command` 不同的是在添加 `parser` 参数时, 响应器会自动处理消息
与普通的 ``on_command`` 不同的是在添加 ``parser`` 参数时, 响应器会自动处理消息
并将用户输入的原始参数列表保存在 `state["argv"]`, `parser` 处理的参数保存在 `state["args"]`
并将用户输入的原始参数列表保存在 ``state["argv"]``, ``parser`` 处理的参数保存在 ``state["args"]``
:参数:
* ``cmd: Union[str, Tuple[str, ...]]``: 指定命令内容
* ``aliases: Optional[Set[Union[str, Tuple[str, ...]]]]``: 命令别名
* ``parser: Optional[ArgumentParser]``: ``nonebot.rule.ArgumentParser`` 对象
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
cmd: 指定命令内容
aliases: 命令别名
parser: `nonebot.rule.ArgumentParser` 对象
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)
@ -866,27 +698,20 @@ class MatcherGroup:
self, pattern: str, flags: Union[int, re.RegexFlag] = 0, **kwargs
) -> Type[Matcher]:
"""
:说明:
注册一个消息事件响应器并且当消息匹配正则表达式时响应
注册一个消息事件响应器并且当消息匹配正则表达式时响应
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
命令匹配规则参考: `正则匹配 <rule.md#regex-regex-flags-0>`_
:参数:
* ``pattern: str``: 正则表达式
* ``flags: Union[int, re.RegexFlag]``: 正则匹配标志
* ``rule: Optional[Union[Rule, T_RuleChecker]]``: 事件响应规则
* ``permission: Optional[Union[Permission, T_PermissionChecker]] =]]``: 事件响应权限
* ``handlers: Optional[List[Union[T_Handler, Dependent]]]``: 事件处理函数列表
* ``temp: bool``: 是否为临时事件响应器仅执行一次
* ``priority: int``: 事件响应器优先级
* ``block: bool``: 是否阻止事件向更低优先级传递
* ``state: Optional[T_State]``: 默认 state
:返回:
- ``Type[Matcher]``
参数:
pattern: 正则表达式
flags: 正则匹配标志
rule: 事件响应规则
permission: 事件响应权限
handlers: 事件处理函数列表
temp: 是否为临时事件响应器仅执行一次
priority: 事件响应器优先级
block: 是否阻止事件向更低优先级传递
state: 默认 state
"""
final_kwargs = self.base_kwargs.copy()
final_kwargs.update(kwargs)

View File

@ -10,8 +10,7 @@ if TYPE_CHECKING:
plugins: Dict[str, "Plugin"] = {}
"""
:类型: ``Dict[str, Plugin]``
:说明: 已加载的插件
已加载的插件
"""
@ -21,72 +20,51 @@ class Plugin(object):
name: str
"""
- **类型**: ``str``
- **说明**: 插件名称使用 文件/文件夹 名称作为插件名
插件名称使用 文件/文件夹 名称作为插件名
"""
module: ModuleType
"""
- **类型**: ``ModuleType``
- **说明**: 插件模块对象
插件模块对象
"""
module_name: str
"""
- **类型**: ``str``
- **说明**: 点分割模块路径
点分割模块路径
"""
manager: "PluginManager"
"""
- **类型**: ``PluginManager``
- **说明**: 导入该插件的插件管理器
导入该插件的插件管理器
"""
export: Export = field(default_factory=Export)
"""
- **类型**: ``Export``
- **说明**: 插件内定义的导出内容
插件内定义的导出内容
"""
matcher: Set[Type[Matcher]] = field(default_factory=set)
"""
- **类型**: ``Set[Type[Matcher]]``
- **说明**: 插件内定义的 ``Matcher``
插件内定义的 `Matcher`
"""
parent_plugin: Optional["Plugin"] = None
"""
- **类型**: ``Optional[Plugin]``
- **说明**: 父插件
父插件
"""
sub_plugins: Set["Plugin"] = field(default_factory=set)
"""
- **类型**: ``Set[Plugin]``
- **说明**: 子插件集合
子插件集合
"""
def get_plugin(name: str) -> Optional[Plugin]:
"""
:说明:
获取当前导入的某个插件
获取当前导入的某个插件
:参数:
* ``name: str``: 插件名 ``load_plugin`` 参数一致如果为 ``load_plugins`` 导入的插件则为文件()
:返回:
- ``Optional[Plugin]``
参数:
name: 插件名 `load_plugin` 参数一致如果为 `load_plugins` 导入的插件则为文件()
"""
return plugins.get(name)
def get_loaded_plugins() -> Set[Plugin]:
"""
:说明:
获取当前已导入的所有插件
:返回:
- ``Set[Plugin]``
获取当前已导入的所有插件
"""
return set(plugins.values())

View File

@ -1,12 +1,11 @@
r"""
规则
====
"""
## 规则
每个事件响应器 ``Matcher`` 拥有一个匹配规则 ``Rule`` 其中是 ``RuleChecker`` 的集合只有当所有 ``RuleChecker`` 检查结果为 ``True`` 时继续运行
每个事件响应器 `Matcher` 拥有一个匹配规则 `Rule` 其中是 `RuleChecker` 的集合只有当所有 `RuleChecker` 检查结果为 `True` 时继续运行
\:\:\:tip 提示
``RuleChecker`` 既可以是 async function 也可以是 sync function
\:\:\:
::: tip 提示
`RuleChecker` 既可以是 async function 也可以是 sync function
:::
"""
import re
@ -63,18 +62,15 @@ CMD_RESULT = TypedDict(
class Rule:
"""
:说明:
``Matcher`` 规则类当事件传递时 ``Matcher`` 运行前进行检查
:示例:
.. code-block:: python
`Matcher` 规则类当事件传递时 `Matcher` 运行前进行检查
用法:
```python
Rule(async_function) & sync_function
# 等价于
from nonebot.utils import run_sync
Rule(async_function, run_sync(sync_function))
```
"""
__slots__ = ("checkers",)
@ -89,9 +85,8 @@ class Rule:
def __init__(self, *checkers: Union[T_RuleChecker, Dependent[bool]]) -> None:
"""
:参数:
* ``*checkers: Union[T_RuleChecker, Dependent[bool]]``: RuleChecker
参数:
*checkers: RuleChecker
"""
self.checkers: Set[Dependent[bool]] = set(
@ -103,13 +98,7 @@ class Rule:
for checker in checkers
)
"""
:说明:
存储 ``RuleChecker``
:类型:
* ``Set[Dependent[bool]]``
存储 `RuleChecker`
"""
async def __call__(
@ -121,21 +110,14 @@ class Rule:
dependency_cache: Optional[T_DependencyCache] = None,
) -> bool:
"""
:说明:
检查是否符合所有规则
检查是否符合所有规则
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
* ``state: T_State``: 当前 State
* ``stack: Optional[AsyncExitStack]``: 异步上下文栈
* ``dependency_cache: Optional[CacheDict[T_Handler, Any]]``: 依赖缓存
:返回:
- ``bool``
参数:
bot: Bot 对象
event: Event 对象
state: 当前 State
stack: 异步上下文栈
dependency_cache: 依赖缓存
"""
if not self.checkers:
return True
@ -224,13 +206,10 @@ class StartswithRule:
def startswith(msg: Union[str, Tuple[str, ...]], ignorecase: bool = False) -> Rule:
"""
:说明:
匹配消息开头
匹配消息开头
:参数:
* ``msg: str``: 消息开头字符串
参数:
msg: 消息开头字符串
"""
if isinstance(msg, str):
msg = (msg,)
@ -259,13 +238,10 @@ class EndswithRule:
def endswith(msg: Union[str, Tuple[str, ...]], ignorecase: bool = False) -> Rule:
"""
:说明:
匹配消息结尾
匹配消息结尾
:参数:
* ``msg: str``: 消息结尾字符串
参数:
msg: 消息结尾字符串
"""
if isinstance(msg, str):
msg = (msg,)
@ -287,13 +263,10 @@ class KeywordsRule:
def keyword(*keywords: str) -> Rule:
"""
:说明:
匹配消息关键词
匹配消息关键词
:参数:
* ``*keywords: str``: 关键词
参数:
*keywords: 关键词
"""
return Rule(KeywordsRule(*keywords))
@ -311,27 +284,23 @@ class CommandRule:
def command(*cmds: Union[str, Tuple[str, ...]]) -> Rule:
r"""
:说明:
"""
命令形式匹配根据配置里提供的 `command_start`, `command_sep` 判断消息是否为命令
命令形式匹配根据配置里提供的 ``command_start``, ``command_sep`` 判断消息是否为命令
可以通过 `state["_prefix"]["command"]` 获取匹配成功的命令`("test",)`通过 `state["_prefix"]["raw_command"]` 获取匹配成功的原始命令文本`"/test"`
可以通过 ``state["_prefix"]["command"]`` 获取匹配成功的命令``("test",)``通过 ``state["_prefix"]["raw_command"]`` 获取匹配成功的原始命令文本``"/test"``
参数:
*cmds: 命令内容
:参数:
用法:
使用默认 `command_start`, `command_sep` 配置
* ``*cmds: Union[str, Tuple[str, ...]]``: 命令内容
命令 `("test",)` 可以匹配`/test` 开头的消息
命令 `("test", "sub")` 可以匹配`/test.sub` 开头的消息
:示例:
使用默认 ``command_start``, ``command_sep`` 配置
命令 ``("test",)`` 可以匹配``/test`` 开头的消息
命令 ``("test", "sub")`` 可以匹配``/test.sub`` 开头的消息
\:\:\:tip 提示
::: tip 提示
命令内容与后续消息间无需空格
\:\:\:
:::
"""
config = get_driver().config
@ -356,9 +325,7 @@ def command(*cmds: Union[str, Tuple[str, ...]]) -> Rule:
class ArgumentParser(ArgParser):
"""
:说明:
``shell_like`` 命令参数解析器解析出错时不会退出程序
`shell_like` 命令参数解析器解析出错时不会退出程序
"""
def _print_message(self, message, file=None):
@ -411,38 +378,34 @@ class ShellCommandRule:
def shell_command(
*cmds: Union[str, Tuple[str, ...]], parser: Optional[ArgumentParser] = None
) -> Rule:
r"""
:说明:
"""
支持 `shell_like` 解析参数的命令形式匹配根据配置里提供的 `command_start`, `command_sep` 判断消息是否为命令
支持 ``shell_like`` 解析参数的命令形式匹配根据配置里提供的 ``command_start``, ``command_sep`` 判断消息是否为命令
可以通过 `state["_prefix"]["command"]` 获取匹配成功的命令`("test",)`通过 `state["_prefix"]["raw_command"]` 获取匹配成功的原始命令文本`"/test"`
可以通过 ``state["_prefix"]["command"]`` 获取匹配成功的命令``("test",)``通过 ``state["_prefix"]["raw_command"]`` 获取匹配成功的原始命令文本``"/test"``
可以通过 `state["argv"]` 获取用户输入的原始参数列表
可以通过 ``state["argv"]`` 获取用户输入的原始参数列表
添加 `parser` 参数后, 可以自动处理消息并将结果保存在 `state["args"]`
添加 ``parser`` 参数后, 可以自动处理消息并将结果保存在 ``state["args"]``
参数:
*cmds: 命令内容
parser: `nonebot.rule.ArgumentParser` 对象
:参数:
* ``*cmds: Union[str, Tuple[str, ...]]``: 命令内容
* ``parser: Optional[ArgumentParser]``: ``nonebot.rule.ArgumentParser`` 对象
:示例:
使用默认 ``command_start``, ``command_sep`` 配置更多示例参考 ``argparse`` 标准库文档
.. code-block:: python
用法:
使用默认 `command_start`, `command_sep` 配置更多示例参考 `argparse` 标准库文档
```python
from nonebot.rule import ArgumentParser
parser = ArgumentParser()
parser.add_argument("-a", action="store_true")
rule = shell_command("ls", parser=parser)
```
\:\:\:tip 提示
::: tip 提示
命令内容与后续消息间无需空格
\:\:\:
:::
"""
if parser is not None and not isinstance(parser, ArgumentParser):
raise TypeError("`parser` must be an instance of nonebot.rule.ArgumentParser")
@ -491,22 +454,19 @@ class RegexRule:
def regex(regex: str, flags: Union[int, re.RegexFlag] = 0) -> Rule:
r"""
:说明:
"""
根据正则表达式进行匹配
根据正则表达式进行匹配
可以通过 `state["_matched"]` `state["_matched_groups"]` `state["_matched_dict"]`
获取正则表达式匹配成功的文本
可以通过 ``state["_matched"]`` ``state["_matched_groups"]`` ``state["_matched_dict"]``
获取正则表达式匹配成功的文本
参数:
regex: 正则表达式
flags: 正则标志
:参数:
* ``regex: str``: 正则表达式
* ``flags: Union[int, re.RegexFlag]``: 正则标志
\:\:\:tip 提示
正则表达式匹配使用 search 而非 match如需从头匹配请使用 ``r"^xxx"`` 来确保匹配开头
\:\:\:
::: tip 提示
正则表达式匹配使用 search 而非 match如需从头匹配请使用 `r"^xxx"` 来确保匹配开头
:::
"""
return Rule(RegexRule(regex, flags))
@ -519,13 +479,7 @@ class ToMeRule:
def to_me() -> Rule:
"""
:说明:
通过 ``event.is_tome()`` 判断事件是否与机器人有关
:参数:
*
通过 `event.is_tome()` 判断事件是否与机器人有关
"""
return Rule(ToMeRule())

View File

@ -1,21 +1,11 @@
"""
类型
====
## 类型
下面的文档中类型部分使用 Python Type Hint 语法 `PEP 484`_`PEP 526`_ `typing`_
下面的文档中类型部分使用 Python Type Hint 语法 [`PEP 484`](https://www.python.org/dev/peps/pep-0484/)[`PEP 526`](https://www.python.org/dev/peps/pep-0526/) [`typing`](https://docs.python.org/3/library/typing.html)
除了 Python 内置的类型下面还出现了如下 NoneBot 自定类型实际上它们是 Python 内置类型的别名
以下类型均可从 nonebot.typing 模块导入
.. _PEP 484:
https://www.python.org/dev/peps/pep-0484/
.. _PEP 526:
https://www.python.org/dev/peps/pep-0526/
.. _typing:
https://docs.python.org/3/library/typing.html
"""
from typing import (
TYPE_CHECKING,
@ -47,193 +37,135 @@ def overrides(InterfaceClass: object):
T_State = Dict[Any, Any]
"""
:类型: ``Dict[Any, Any]``
:说明:
事件处理状态 State 类型
事件处理状态 State 类型
"""
T_BotConnectionHook = Callable[["Bot"], Awaitable[None]]
"""
:类型: ``Callable[[Bot], Awaitable[None]]``
:说明:
Bot 连接建立时执行的函数
Bot 连接建立时执行的函数
"""
T_BotDisconnectionHook = Callable[["Bot"], Awaitable[None]]
"""
:类型: ``Callable[[Bot], Awaitable[None]]``
:说明:
Bot 连接断开时执行的函数
Bot 连接断开时执行的函数
"""
T_CallingAPIHook = Callable[["Bot", str, Dict[str, Any]], Awaitable[None]]
"""
:类型: ``Callable[[Bot, str, Dict[str, Any]], Awaitable[None]]``
:说明:
``bot.call_api`` 时执行的函数
`bot.call_api` 时执行的函数
"""
T_CalledAPIHook = Callable[
["Bot", Optional[Exception], str, Dict[str, Any], Any], Awaitable[None]
]
"""
:类型: ``Callable[[Bot, Optional[Exception], str, Dict[str, Any], Any], Awaitable[None]]``
:说明:
``bot.call_api`` 后执行的函数参数分别为 bot, exception, api, data, result
`bot.call_api` 后执行的函数参数分别为 bot, exception, api, data, result
"""
T_EventPreProcessor = Callable[..., Union[None, Awaitable[None]]]
"""
:类型: ``Callable[..., Union[None, Awaitable[None]]]``
事件预处理函数 EventPreProcessor 类型
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
事件预处理函数 EventPreProcessor 类型
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- DefaultParam: 带有默认值的参数
"""
T_EventPostProcessor = Callable[..., Union[None, Awaitable[None]]]
"""
:类型: ``Callable[..., Union[None, Awaitable[None]]]``
事件预处理函数 EventPostProcessor 类型
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
事件预处理函数 EventPostProcessor 类型
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- DefaultParam: 带有默认值的参数
"""
T_RunPreProcessor = Callable[..., Union[None, Awaitable[None]]]
"""
:类型: ``Callable[..., Union[None, Awaitable[None]]]``
事件响应器运行前预处理函数 RunPreProcessor 类型
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``MatcherParam``: Matcher 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
事件响应器运行前预处理函数 RunPreProcessor 类型
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_RunPostProcessor = Callable[..., Union[None, Awaitable[None]]]
"""
:类型: ``Callable[..., Union[None, Awaitable[None]]]``
事件响应器运行前预处理函数 RunPostProcessor 类型第二个参数为运行时产生的错误如果存在
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``MatcherParam``: Matcher 对象
* ``ExceptionParam``: 异常对象可能为 None
* ``DefaultParam``: 带有默认值的参数
:说明:
事件响应器运行前预处理函数 RunPostProcessor 类型第二个参数为运行时产生的错误如果存在
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- MatcherParam: Matcher 对象
- ExceptionParam: 异常对象可能为 None
- DefaultParam: 带有默认值的参数
"""
T_RuleChecker = Callable[..., Union[bool, Awaitable[bool]]]
"""
:类型: ``Callable[..., Union[bool, Awaitable[bool]]]``
RuleChecker 即判断是否响应事件的处理函数
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
RuleChecker 即判断是否响应事件的处理函数
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- DefaultParam: 带有默认值的参数
"""
T_PermissionChecker = Callable[..., Union[bool, Awaitable[bool]]]
"""
:类型: ``Callable[..., Union[bool, Awaitable[bool]]]``
RuleChecker 即判断是否响应消息的处理函数
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
RuleChecker 即判断是否响应消息的处理函数
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- DefaultParam: 带有默认值的参数
"""
T_Handler = Callable[..., Any]
"""
:类型: ``Callable[..., Any]``
:说明:
Handler 处理函数
Handler 处理函数
"""
T_TypeUpdater = Callable[..., Union[str, Awaitable[str]]]
"""
:类型: ``Callable[..., Union[None, Awaitable[None]]]``
TypeUpdater Matcher.pause, Matcher.reject 时被运行用于更新响应的事件类型默认会更新为 `message`
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``MatcherParam``: Matcher 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
TypeUpdater Matcher.pause, Matcher.reject 时被运行用于更新响应的事件类型默认会更新为 ``message``
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_PermissionUpdater = Callable[..., Union["Permission", Awaitable["Permission"]]]
"""
:类型: ``Callable[..., Union[Permission, Awaitable[Permission]]]``
PermissionUpdater Matcher.pause, Matcher.reject 时被运行用于更新会话对象权限默认会更新为当前事件的触发对象
:依赖参数:
依赖参数:
* ``DependParam``: 子依赖参数
* ``BotParam``: Bot 对象
* ``EventParam``: Event 对象
* ``StateParam``: State 对象
* ``MatcherParam``: Matcher 对象
* ``DefaultParam``: 带有默认值的参数
:说明:
PermissionUpdater Matcher.pause, Matcher.reject 时被运行用于更新会话对象权限默认会更新为当前事件的触发对象
- DependParam: 子依赖参数
- BotParam: Bot 对象
- EventParam: Event 对象
- StateParam: State 对象
- MatcherParam: Matcher 对象
- DefaultParam: 带有默认值的参数
"""
T_DependencyCache = Dict[Callable[..., Any], "Task[Any]"]
"""
:类型: ``Dict[Callable[..., Any], Task[Any]]``
:说明:
依赖缓存, 用于存储依赖函数的返回值
依赖缓存, 用于存储依赖函数的返回值
"""

View File

@ -31,17 +31,10 @@ V = TypeVar("V")
def escape_tag(s: str) -> str:
"""
:说明:
用于记录带颜色日志时转义 `<tag>` 类型特殊标签
用于记录带颜色日志时转义 ``<tag>`` 类型特殊标签
:参数:
* ``s: str``: 需要转义的字符串
:返回:
- ``str``
参数:
s: 需要转义的字符串
"""
return re.sub(r"</?((?:[fb]g\s)?[^<>\s]*)>", r"\\\g<0>", s)
@ -90,17 +83,10 @@ def is_async_gen_callable(call: Callable[..., Any]) -> bool:
def run_sync(call: Callable[P, R]) -> Callable[P, Awaitable[R]]:
"""
:说明:
一个用于包装 sync function async function 的装饰器
一个用于包装 sync function async function 的装饰器
:参数:
* ``call: Callable[P, R]``: 被装饰的同步函数
:返回:
- ``Callable[P, Awaitable[R]]``
参数:
call: 被装饰的同步函数
"""
@wraps(call)
@ -135,9 +121,7 @@ def get_name(obj: Any) -> str:
class DataclassEncoder(json.JSONEncoder):
"""
:说明:
在JSON序列化 ``Message`` (List[Dataclass]) 时使用的 ``JSONEncoder``
在JSON序列化 `Message` (List[Dataclass]) 时使用的 `JSONEncoder`
"""
@overrides(json.JSONEncoder)
@ -149,15 +133,12 @@ class DataclassEncoder(json.JSONEncoder):
def logger_wrapper(logger_name: str):
"""
:说明:
用于打印 adapter 的日志
:log 参数:
* ``level: Literal["CRITICAL", "WARNING", "INFO", "DEBUG", "TRACE"]``: 日志等级
* ``message: str``: 日志信息
* ``exception: Optional[Exception]``: 异常信息
参数:
level: 日志等级
message: 日志信息
exception: 异常信息
"""
def log(level: str, message: str, exception: Optional[Exception] = None):