Module aiocqhttp.default

此模块提供了默认 bot 对象及用于控制和使用它的相关函数、对象、和装饰器。

"""
此模块提供了默认 bot 对象及用于控制和使用它的相关函数、对象、和装饰器。
"""

from . import CQHttp
from .api_impl import LazyApi

__all__ = [
    'default_bot',
    'default_server_app',
    'api',
    'sync_api',
    'reconfigure_default_bot',
    'run',
    'send',
    'before_sending',
    'on',
    'on_message',
    'on_notice',
    'on_request',
    'on_meta_event',
    'before',
    'before_message',
    'before_notice',
    'before_request',
    'before_meta_event',
]

__pdoc__ = {}

default_bot = CQHttp()
"""默认 bot 对象。"""

default_server_app = default_bot.server_app
"""默认 bot 对象的 Quart app。"""

api = default_bot.api
"""默认 bot 对象的 `aiocqhttp.api.AsyncApi` 对象，用于异步地调用 OneBot API。"""

sync_api = LazyApi(lambda: default_bot.sync)
"""默认 bot 对象的 `aiocqhttp.api.SyncApi` 对象，用于同步地调用 OneBot API。"""

reconfigure_default_bot = default_bot._configure
__pdoc__['reconfigure_default_bot'] = """
重新配置默认 bot 对象。
"""

run = default_bot.run
__pdoc__['run'] = """
运行默认 bot 对象。
"""

send = default_bot.send
__pdoc__['send'] = """
使用默认 bot 对象发送消息。
"""

before_sending = default_bot.before_sending
__pdoc__['before_sending'] = """
注册默认 bot 对象发送消息前的钩子函数，用作装饰器。
"""

on = default_bot.on
__pdoc__['on'] = """
注册默认 bot 对象的事件处理函数，用作装饰器。
"""

on_message = default_bot.on_message
__pdoc__['on_message'] = """
注册默认 bot 对象的消息事件处理函数，用作装饰器。
"""

on_notice = default_bot.on_notice
__pdoc__['on_notice'] = """
注册默认 bot 对象的通知事件处理函数，用作装饰器。
"""

on_request = default_bot.on_request
__pdoc__['on_request'] = """
注册默认 bot 对象的请求事件处理函数，用作装饰器。
"""

on_meta_event = default_bot.on_meta_event
__pdoc__['on_meta_event'] = """
注册默认 bot 对象的元事件处理函数，用作装饰器。
"""

before = default_bot.before
__pdoc__['before'] = """
注册默认 bot 对象事件处理前的钩子函数，用作装饰器。
"""

before_message = default_bot.before_message
__pdoc__['before_message'] = """
注册默认 bot 对象消息事件处理前的钩子函数，用作装饰器。
"""

before_notice = default_bot.before_notice
__pdoc__['before_notice'] = """
注册默认 bot 对象的通知事件处理前的钩子函数，用作装饰器。
"""

before_request = default_bot.before_request
__pdoc__['before_request'] = """
注册默认 bot 对象的请求事件处理前的钩子函数，用作装饰器。
"""

before_meta_event = default_bot.before_meta_event
__pdoc__['before_meta_event'] = """
注册默认 bot 对象的元事件处理前的钩子函数，用作装饰器。
"""

Global variables

var default_bot : Any

默认 bot 对象。

var default_server_app : None

默认 bot 对象的 Quart app。

var api : Any

默认 bot 对象的 AsyncApi 对象，用于异步地调用 OneBot API。

var sync_api : Union[Awaitable[Any], Any]

默认 bot 对象的 SyncApi 对象，用于同步地调用 OneBot API。

Functions

def reconfigure_default_bot(api_root: Union[str, NoneType] = None, access_token: Union[str, NoneType] = None, secret: Union[~AnyStr, NoneType] = None, message_class: Union[type, NoneType] = None, api_timeout_sec: Union[float, NoneType] = None)

重新配置默认 bot 对象。

Expand source code

def _configure(self,
               api_root: Optional[str] = None,
               access_token: Optional[str] = None,
               secret: Optional[AnyStr] = None,
               message_class: Optional[type] = None,
               api_timeout_sec: Optional[float] = None):
    self._message_class = message_class
    api_timeout_sec = api_timeout_sec or 60  # wait for 60 secs by default
    self._access_token = access_token
    self._secret = secret
    self._api._http_api = HttpApi(api_root, access_token, api_timeout_sec)
    self._wsr_api_clients = {}  # connected wsr api clients
    self._wsr_event_clients = set()
    self._api._wsr_api = WebSocketReverseApi(self._wsr_api_clients,
                                             self._wsr_event_clients,
                                             api_timeout_sec)

def run(host: str = '127.0.0.1', port: int = 8080, *args, **kwargs) ‑> NoneType

运行默认 bot 对象。

Expand source code

def run(self,
        host: str = '127.0.0.1',
        port: int = 8080,
        *args,
        **kwargs) -> None:
    """运行 bot 对象，实际就是运行 Quart app，参数与 `Quart.run` 一致。"""
    if 'use_reloader' not in kwargs:
        kwargs['use_reloader'] = False
    self._server_app.run(host=host, port=port, *args, **kwargs)

async def send(event: Event, message: Union[str, Dict[str, Any], List[Dict[str, Any]], ForwardRef('MessageSegment'), ForwardRef('Message')], **kwargs) ‑> Union[Dict[str, Any], NoneType]

使用默认 bot 对象发送消息。

Expand source code

async def send(self, event: Event, message: Message_T,
               **kwargs) -> Optional[Dict[str, Any]]:
    """
    向触发事件的主体发送消息。

    ``event`` 参数为事件对象，``message`` 参数为要发送的消息。可额外传入 ``at_sender``
    命名参数用于控制是否 at 事件的触发者，默认为 `False`。其它命名参数作为
    OneBot API ``send_msg`` 的参数直接传递。
    """
    msg = message if isinstance(message, Message) else Message(message)
    await run_async_funcs(self._before_sending_funcs, event, msg, kwargs)

    at_sender = kwargs.pop('at_sender', False) and ('user_id' in event)

    keys = {'message_type', 'user_id', 'group_id', 'discuss_id'}
    params = {k: v for k, v in event.items() if k in keys}
    params['message'] = msg
    params.update(kwargs)

    if 'message_type' not in params:
        if 'group_id' in params:
            params['message_type'] = 'group'
        elif 'discuss_id' in params:
            params['message_type'] = 'discuss'
        elif 'user_id' in params:
            params['message_type'] = 'private'

    if at_sender and params['message_type'] != 'private':
        params['message'] = MessageSegment.at(params['user_id']) + \
                            MessageSegment.text(' ') + params['message']

    return await self.send_msg(**params)

def before_sending(func: Callable) ‑> Callable

注册默认 bot 对象发送消息前的钩子函数，用作装饰器。

Expand source code

def before_sending(self, func: Callable) -> Callable:
    """
    注册发送消息前的钩子函数，用作装饰器，例如：

    ```py
    @bot.before_sending
    async def hook(event: Event, message: Message, kwargs: Dict[str, Any]):
        message.clear()
        message.append(MessageSegment.text('hooked!'))
    ```

    该钩子函数在刚进入 `CQHttp.send` 函数时运行，用户可在钩子函数中修改要发送的
    ``message`` 和发送参数 ``kwargs``。
    """
    self._before_sending_funcs.add(ensure_async(func))
    return func

def on(*event_names: str) ‑> Callable

注册默认 bot 对象的事件处理函数，用作装饰器。

Expand source code

def on(self, *event_names: str) -> Callable:
    """
    注册事件处理函数，用作装饰器，例如：

    ```py
    @bot.on('notice.group_decrease', 'notice.group_increase')
    async def handler(event):
        pass
    ```

    参数为要注册的事件名，格式是点号分割的各级事件类型，见 `Event.name`。

    可以多次调用，一个函数可作为多个事件的处理函数，一个事件也可以有多个处理函数。

    可以按不同粒度注册处理函数，例如：

    ```py
    @bot.on('message')
    async def handle_message(event):
        pass

    @bot.on('message.private')
    async def handle_private_message(event):
        pass

    @bot.on('message.private.friend')
    async def handle_friend_private_message(event):
        pass
    ```

    当收到好友私聊消息时，会首先运行 ``handle_friend_private_message``，然后运行
    ``handle_private_message``，最后运行 ``handle_message``。
    """

    def deco(func: Callable) -> Callable:
        for name in event_names:
            self.subscribe(name, func)
        return func

    return deco

def on_message(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的消息事件处理函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def on_notice(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的通知事件处理函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def on_request(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的请求事件处理函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def on_meta_event(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的元事件处理函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def before(*event_names: str) ‑> Callable

注册默认 bot 对象事件处理前的钩子函数，用作装饰器。

Expand source code

def before(self, *event_names: str) -> Callable:
    """
    注册事件处理前的钩子函数，用作装饰器，例如：

    ```py
    @bot.before('notice.group_decrease', 'notice.group_increase')
    async def hook(event):
        pass
    ```

    参数为要注册的事件名，格式是点号分割的各级事件类型，见 `Event.name`。

    钩子函数的注册方法和事件处理函数几乎完全一致，只需将 ``on`` 改为 ``before``。

    各级 before 钩子函数全部运行完成后，才会运行事件处理函数。
    """

    def deco(func: Callable) -> Callable:
        for name in event_names:
            self.hook_before(name, func)
        return func

    return deco

def before_message(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象消息事件处理前的钩子函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def before_notice(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的通知事件处理前的钩子函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def before_request(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的请求事件处理前的钩子函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco

def before_meta_event(arg: Union[str, Callable, NoneType] = None, *sub_event_names: str) ‑> Callable

注册默认 bot 对象的元事件处理前的钩子函数，用作装饰器。

Expand source code

def deco_deco(self,
              arg: Optional[Union[str, Callable]] = None,
              *sub_event_names: str) -> Callable:

    def deco(func: Callable) -> Callable:
        if isinstance(arg, str):
            e = [type_ + '.' + e for e in [arg] + list(sub_event_names)]
            # self.on(*e)(func)
            deco_method(self, *e)(func)
        else:
            # self.on(type_)(func)
            deco_method(self, type_)(func)
        return func

    if callable(arg):
        return deco(arg)
    return deco