Function handle_event

nonebot/message.py:519–606

处理一个事件。调用该函数以实现分发事件。 参数: bot: Bot 对象 event: Event 对象 用法: ```python driver.task_group.start_soon(handle_event, bot, event) ```

(bot: "Bot", event: "Event")

async def handle_event(bot: "Bot", event: "Event") -> None:
    """处理一个事件。调用该函数以实现分发事件。

    参数:
        bot: Bot 对象
        event: Event 对象

    用法:
        ```python
        driver.task_group.start_soon(handle_event, bot, event)
        ```
    """
    show_log = True
    log_msg = f"<m>{escape_tag(bot.type)} {escape_tag(bot.self_id)}</m> | "
    try:
        log_msg += event.get_log_string()
    except NoLogException:
        show_log = False
    if show_log:
        logger.opt(colors=True).success(log_msg)

    state: dict[Any, Any] = {}
    dependency_cache: T_DependencyCache = {}

    # create event scope context
    async with AsyncExitStack() as stack:
        if not await _apply_event_preprocessors(
            bot=bot,
            event=event,
            state=state,
            stack=stack,
            dependency_cache=dependency_cache,
        ):
            return

        # Trie Match
        try:
            TrieRule.get_value(bot, event, state)
        except Exception as e:
            logger.opt(colors=True, exception=e).warning(
                "Error while parsing command for event"
            )

        break_flag = False

        def _handle_stop_propagation(exc_group: BaseExceptionGroup) -> None:
            nonlocal break_flag

            break_flag = True
            logger.debug("Stop event propagation")

        # iterate through all priority until stop propagation
        for priority in sorted(matchers.keys()):
            if break_flag:
                break

            if show_log:
                logger.debug(f"Checking for matchers in priority {priority}...")