"""
事件处理
========
NoneBot 内部处理并按优先级分发事件给所有事件响应器,提供了多个插槽以进行事件的预处理等。
"""
import asyncio
from datetime import datetime
from contextlib import AsyncExitStack
from typing import (
TYPE_CHECKING,
Any,
Set,
Dict,
List,
Type,
Optional,
Coroutine,
)
from nonebot import params
from nonebot.log import logger
from nonebot.rule import TrieRule
from nonebot.utils import escape_tag
from nonebot.dependencies import Dependent
from nonebot.matcher import Matcher, matchers
from nonebot.exception import (
NoLogException,
StopPropagation,
IgnoredException,
SkippedException,
)
from nonebot.typing import (
T_State,
T_Handler,
T_DependencyCache,
T_RunPreProcessor,
T_RunPostProcessor,
T_EventPreProcessor,
T_EventPostProcessor,
)
if TYPE_CHECKING:
from nonebot.adapters import Bot, Event
_event_preprocessors: Set[Dependent[None]] = set()
_event_postprocessors: Set[Dependent[None]] = set()
_run_preprocessors: Set[Dependent[None]] = set()
_run_postprocessors: Set[Dependent[None]] = set()
EVENT_PCS_PARAMS = [
params.DependParam,
params.BotParam,
params.EventParam,
params.StateParam,
params.DefaultParam,
]
RUN_PREPCS_PARAMS = [
params.DependParam,
params.BotParam,
params.EventParam,
params.StateParam,
params.ArgParam,
params.MatcherParam,
params.DefaultParam,
]
RUN_POSTPCS_PARAMS = [
params.DependParam,
params.ExceptionParam,
params.BotParam,
params.EventParam,
params.StateParam,
params.ArgParam,
params.MatcherParam,
params.DefaultParam,
]
def event_preprocessor(func: T_EventPreProcessor) -> T_EventPreProcessor:
"""
:说明:
事件预处理。装饰一个函数,使它在每次接收到事件并分发给各响应器之前执行。
"""
_event_preprocessors.add(
Dependent[None].parse(call=func, allow_types=EVENT_PCS_PARAMS)
)
return func
def event_postprocessor(func: T_EventPostProcessor) -> T_EventPostProcessor:
"""
:说明:
事件后处理。装饰一个函数,使它在每次接收到事件并分发给各响应器之后执行。
"""
_event_postprocessors.add(
Dependent[None].parse(call=func, allow_types=EVENT_PCS_PARAMS)
)
return func
def run_preprocessor(func: T_RunPreProcessor) -> T_RunPreProcessor:
"""
:说明:
运行预处理。装饰一个函数,使它在每次事件响应器运行前执行。
"""
_run_preprocessors.add(
Dependent[None].parse(call=func, allow_types=RUN_PREPCS_PARAMS)
)
return func
def run_postprocessor(func: T_RunPostProcessor) -> T_RunPostProcessor:
"""
:说明:
运行后处理。装饰一个函数,使它在每次事件响应器运行后执行。
"""
_run_postprocessors.add(
Dependent[None].parse(call=func, allow_types=RUN_POSTPCS_PARAMS)
)
return func
async def _run_coro_with_catch(coro: Coroutine[Any, Any, Any]) -> Any:
try:
return await coro
except SkippedException:
pass
async def _check_matcher(
priority: int,
Matcher: Type[Matcher],
bot: "Bot",
event: "Event",
state: T_State,
stack: Optional[AsyncExitStack] = None,
dependency_cache: Optional[T_DependencyCache] = None,
) -> None:
if Matcher.expire_time and datetime.now() > Matcher.expire_time:
try:
matchers[priority].remove(Matcher)
except Exception:
pass
return
try:
if not await Matcher.check_perm(
bot, event, stack, dependency_cache
) or not await Matcher.check_rule(bot, event, state, stack, dependency_cache):
return
except Exception as e:
logger.opt(colors=True, exception=e).error(
f"Rule check failed for {Matcher}."
)
return
if Matcher.temp:
try:
matchers[priority].remove(Matcher)
except Exception:
pass
await _run_matcher(Matcher, bot, event, state, stack, dependency_cache)
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:
logger.info(f"Event will be handled by {Matcher}")
matcher = Matcher()
coros = list(
map(
lambda x: _run_coro_with_catch(
x(
matcher=matcher,
bot=bot,
event=event,
state=state,
stack=stack,
dependency_cache=dependency_cache,
)
),
_run_preprocessors,
)
)
if coros:
try:
await asyncio.gather(*coros)
except IgnoredException:
logger.opt(colors=True).info(
f"Matcher {matcher} running is cancelled"
)
return
except Exception as e:
logger.opt(colors=True, exception=e).error(
"Error when running RunPreProcessors. "
"Running cancelled!"
)
return
exception = None
try:
logger.debug(f"Running matcher {matcher}")
await matcher.run(bot, event, state, stack, dependency_cache)
except Exception as e:
logger.opt(colors=True, exception=e).error(
f"Running matcher {matcher} failed."
)
exception = e
coros = list(
map(
lambda x: _run_coro_with_catch(
x(
matcher=matcher,
exception=exception,
bot=bot,
event=event,
state=state,
stack=stack,
dependency_cache=dependency_cache,
)
),
_run_postprocessors,
)
)
if coros:
try:
await asyncio.gather(*coros)
except Exception as e:
logger.opt(colors=True, exception=e).error(
"Error when running RunPostProcessors"
)
if matcher.block:
raise StopPropagation
return
async def handle_event(bot: "Bot", event: "Event") -> None:
"""
:说明:
处理一个事件。调用该函数以实现分发事件。
:参数:
* ``bot: Bot``: Bot 对象
* ``event: Event``: Event 对象
:示例:
.. code-block:: python
import asyncio
asyncio.create_task(handle_event(bot, event))
"""
show_log = True
log_msg = f"{escape_tag(bot.type.upper())} {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 = {}
async with AsyncExitStack() as stack:
coros = list(
map(
lambda x: _run_coro_with_catch(
x(
bot=bot,
event=event,
state=state,
stack=stack,
dependency_cache=dependency_cache,
)
),
_event_preprocessors,
)
)
if coros:
try:
if show_log:
logger.debug("Running PreProcessors...")
await asyncio.gather(*coros)
except IgnoredException as e:
logger.opt(colors=True).info(
f"Event {escape_tag(event.get_event_name())} is ignored"
)
return
except Exception as e:
logger.opt(colors=True, exception=e).error(
"Error when running EventPreProcessors. "
"Event ignored!"
)
return
# Trie Match
TrieRule.get_value(bot, event, state)
break_flag = False
for priority in sorted(matchers.keys()):
if break_flag:
break
if show_log:
logger.debug(f"Checking for matchers in priority {priority}...")
pending_tasks = [
_check_matcher(
priority, matcher, bot, event, state.copy(), stack, dependency_cache
)
for matcher in matchers[priority]
]
results = await asyncio.gather(*pending_tasks, return_exceptions=True)
for result in results:
if not isinstance(result, Exception):
continue
if isinstance(result, StopPropagation):
break_flag = True
logger.debug("Stop event propagation")
else:
logger.opt(colors=True, exception=result).error(
"Error when checking Matcher."
)
coros = list(
map(
lambda x: _run_coro_with_catch(
x(
bot=bot,
event=event,
state=state,
stack=stack,
dependency_cache=dependency_cache,
)
),
_event_postprocessors,
)
)
if coros:
try:
if show_log:
logger.debug("Running PostProcessors...")
await asyncio.gather(*coros)
except Exception as e:
logger.opt(colors=True, exception=e).error(
"Error when running EventPostProcessors"
)