"""本模块定义了事件处理主要流程。 NoneBot 内部处理并按优先级分发事件给所有事件响应器,提供了多个插槽以进行事件的预处理等。 FrontMatter: mdx: format: md sidebar_position: 2 description: nonebot.message 模块 """ import contextlib from datetime import datetime from contextlib import AsyncExitStack from typing import TYPE_CHECKING, Any, Callable, Optional import anyio from exceptiongroup import BaseExceptionGroup, catch from nonebot.log import logger from nonebot.rule import TrieRule from nonebot.dependencies import Dependent from nonebot.matcher import Matcher, matchers from nonebot.exception import ( NoLogException, StopPropagation, IgnoredException, SkippedException, ) from nonebot.utils import ( escape_tag, run_coro_with_catch, run_coro_with_shield, flatten_exception_group, ) from nonebot.typing import ( T_State, T_DependencyCache, T_RunPreProcessor, T_RunPostProcessor, T_EventPreProcessor, T_EventPostProcessor, ) from nonebot.internal.params import ( ArgParam, BotParam, EventParam, StateParam, DependParam, DefaultParam, MatcherParam, ExceptionParam, ) if TYPE_CHECKING: from nonebot.adapters import Bot, Event _event_preprocessors: set[Dependent[Any]] = set() _event_postprocessors: set[Dependent[Any]] = set() _run_preprocessors: set[Dependent[Any]] = set() _run_postprocessors: set[Dependent[Any]] = set() EVENT_PCS_PARAMS = ( DependParam, BotParam, EventParam, StateParam, DefaultParam, ) RUN_PREPCS_PARAMS = ( DependParam, BotParam, EventParam, StateParam, ArgParam, MatcherParam, DefaultParam, ) RUN_POSTPCS_PARAMS = ( DependParam, ExceptionParam, BotParam, EventParam, StateParam, ArgParam, MatcherParam, DefaultParam, ) def event_preprocessor(func: T_EventPreProcessor) -> T_EventPreProcessor: """事件预处理。 装饰一个函数,使它在每次接收到事件并分发给各响应器之前执行。 """ _event_preprocessors.add( Dependent[Any].parse(call=func, allow_types=EVENT_PCS_PARAMS) ) return func def event_postprocessor(func: T_EventPostProcessor) -> T_EventPostProcessor: """事件后处理。 装饰一个函数,使它在每次接收到事件并分发给各响应器之后执行。 """ _event_postprocessors.add( Dependent[Any].parse(call=func, allow_types=EVENT_PCS_PARAMS) ) return func def run_preprocessor(func: T_RunPreProcessor) -> T_RunPreProcessor: """运行预处理。 装饰一个函数,使它在每次事件响应器运行前执行。 """ _run_preprocessors.add( Dependent[Any].parse(call=func, allow_types=RUN_PREPCS_PARAMS) ) return func def run_postprocessor(func: T_RunPostProcessor) -> T_RunPostProcessor: """运行后处理。 装饰一个函数,使它在每次事件响应器运行后执行。 """ _run_postprocessors.add( Dependent[Any].parse(call=func, allow_types=RUN_POSTPCS_PARAMS) ) return func def _handle_ignored_exception(msg: str) -> Callable[[BaseExceptionGroup], None]: def _handle(exc_group: BaseExceptionGroup[IgnoredException]) -> None: logger.opt(colors=True).info(msg) return _handle def _handle_exception(msg: str) -> Callable[[BaseExceptionGroup], None]: def _handle(exc_group: BaseExceptionGroup[Exception]) -> None: for exc in flatten_exception_group(exc_group): logger.opt(colors=True, exception=exc).error(msg) return _handle async def _apply_event_preprocessors( bot: "Bot", event: "Event", state: T_State, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, show_log: bool = True, ) -> bool: """运行事件预处理。 参数: bot: Bot 对象 event: Event 对象 state: 会话状态 stack: 异步上下文栈 dependency_cache: 依赖缓存 show_log: 是否显示日志 返回: 是否继续处理事件 """ if not _event_preprocessors: return True if show_log: logger.debug("Running PreProcessors...") with catch( { IgnoredException: _handle_ignored_exception( f"Event {escape_tag(event.get_event_name())} is ignored" ), Exception: _handle_exception( "Error when running EventPreProcessors. " "Event ignored!" ), } ): async with anyio.create_task_group() as tg: for proc in _event_preprocessors: tg.start_soon( run_coro_with_catch, proc( bot=bot, event=event, state=state, stack=stack, dependency_cache=dependency_cache, ), (SkippedException,), ) return True return False async def _apply_event_postprocessors( bot: "Bot", event: "Event", state: T_State, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, show_log: bool = True, ) -> None: """运行事件后处理。 参数: bot: Bot 对象 event: Event 对象 state: 会话状态 stack: 异步上下文栈 dependency_cache: 依赖缓存 show_log: 是否显示日志 """ if not _event_postprocessors: return if show_log: logger.debug("Running PostProcessors...") with catch( { Exception: _handle_exception( "Error when running EventPostProcessors" ) } ): async with anyio.create_task_group() as tg: for proc in _event_postprocessors: tg.start_soon( run_coro_with_catch, proc( bot=bot, event=event, state=state, stack=stack, dependency_cache=dependency_cache, ), (SkippedException,), ) async def _apply_run_preprocessors( bot: "Bot", event: "Event", state: T_State, matcher: Matcher, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, ) -> bool: """运行事件响应器运行前处理。 参数: bot: Bot 对象 event: Event 对象 state: 会话状态 matcher: 事件响应器 stack: 异步上下文栈 dependency_cache: 依赖缓存 返回: 是否继续处理事件 """ if not _run_preprocessors: return True # ensure matcher function can be correctly called with ( matcher.ensure_context(bot, event), catch( { IgnoredException: _handle_ignored_exception( f"{matcher} running is cancelled" ), Exception: _handle_exception( "Error when running RunPreProcessors. " "Running cancelled!" ), } ), ): async with anyio.create_task_group() as tg: for proc in _run_preprocessors: tg.start_soon( run_coro_with_catch, proc( matcher=matcher, bot=bot, event=event, state=state, stack=stack, dependency_cache=dependency_cache, ), (SkippedException,), ) return True return False async def _apply_run_postprocessors( bot: "Bot", event: "Event", matcher: Matcher, exception: Optional[Exception] = None, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, ) -> None: """运行事件响应器运行后处理。 参数: bot: Bot 对象 event: Event 对象 matcher: 事件响应器 exception: 事件响应器运行异常 stack: 异步上下文栈 dependency_cache: 依赖缓存 """ if not _run_postprocessors: return with ( matcher.ensure_context(bot, event), catch( { Exception: _handle_exception( "Error when running RunPostProcessors" "" ) } ), ): async with anyio.create_task_group() as tg: for proc in _run_postprocessors: tg.start_soon( run_coro_with_catch, proc( matcher=matcher, exception=exception, bot=bot, event=event, state=matcher.state, stack=stack, dependency_cache=dependency_cache, ), (SkippedException,), ) async def _check_matcher( Matcher: type[Matcher], bot: "Bot", event: "Event", state: T_State, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, ) -> bool: """检查事件响应器是否符合运行条件。 请注意,过时的事件响应器将被**销毁**。对于未过时的事件响应器,将会一次检查其响应类型、权限和规则。 参数: Matcher: 要检查的事件响应器 bot: Bot 对象 event: Event 对象 state: 会话状态 stack: 异步上下文栈 dependency_cache: 依赖缓存 返回: bool: 是否符合运行条件 """ if Matcher.expire_time and datetime.now() > Matcher.expire_time: with contextlib.suppress(Exception): Matcher.destroy() return False try: if not await Matcher.check_perm(bot, event, stack, dependency_cache): logger.trace(f"Permission conditions not met for {Matcher}") return False except Exception as e: logger.opt(colors=True, exception=e).error( f"Permission check failed for {Matcher}." ) return False try: if not await Matcher.check_rule(bot, event, state, stack, dependency_cache): logger.trace(f"Rule conditions not met for {Matcher}") return False except Exception as e: logger.opt(colors=True, exception=e).error( f"Rule check failed for {Matcher}." ) return False return True async def _run_matcher( Matcher: type[Matcher], bot: "Bot", event: "Event", state: T_State, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, ) -> None: """运行事件响应器。 临时事件响应器将在运行前被**销毁**。 参数: Matcher: 事件响应器 bot: Bot 对象 event: Event 对象 state: 会话状态 stack: 异步上下文栈 dependency_cache: 依赖缓存 异常: StopPropagation: 阻止事件继续传播 """ logger.info(f"Event will be handled by {Matcher}") if Matcher.temp: with contextlib.suppress(Exception): Matcher.destroy() matcher = Matcher() if not await _apply_run_preprocessors( bot=bot, event=event, state=state, matcher=matcher, stack=stack, dependency_cache=dependency_cache, ): return exception = None logger.debug(f"Running {matcher}") try: await matcher.run(bot, event, state, stack, dependency_cache) except Exception as e: logger.opt(colors=True, exception=e).error( f"Running {matcher} failed." ) exception = e await _apply_run_postprocessors( bot=bot, event=event, matcher=matcher, exception=exception, stack=stack, dependency_cache=dependency_cache, ) if matcher.block: raise StopPropagation async def check_and_run_matcher( Matcher: type[Matcher], bot: "Bot", event: "Event", state: T_State, stack: Optional[AsyncExitStack] = None, dependency_cache: Optional[T_DependencyCache] = None, ) -> None: """检查并运行事件响应器。 参数: Matcher: 事件响应器 bot: Bot 对象 event: Event 对象 state: 会话状态 stack: 异步上下文栈 dependency_cache: 依赖缓存 """ if not await _check_matcher( Matcher=Matcher, bot=bot, event=event, state=state, stack=stack, dependency_cache=dependency_cache, ): return await _run_matcher( Matcher=Matcher, bot=bot, event=event, state=state, stack=stack, dependency_cache=dependency_cache, ) 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"{escape_tag(bot.type)} {escape_tag(bot.self_id)} | " 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}...") if not (priority_matchers := matchers[priority]): continue with catch( { StopPropagation: _handle_stop_propagation, Exception: _handle_exception( "Error when checking Matcher." ), } ): async with anyio.create_task_group() as tg: for matcher in priority_matchers: tg.start_soon( run_coro_with_shield, check_and_run_matcher( matcher, bot, event, state.copy(), stack, dependency_cache, ), ) if show_log: logger.debug("Checking for matchers completed") await _apply_event_postprocessors(bot, event, state, stack, dependency_cache)