第二节 Matcher、Rule、Permission的实例化

机器人>NoneBot
realhuhu 242 0

本文介绍Matche如何实例化、handler如何注册到Matcher，顺便介绍Rule和Permission的实例化

注意区分handlers和handler

  • handler：被形如@xxx.handle()等装饰的函数，类型是函数
  • handlers：特指Matcher.handlers，类型为List[Dependent[Any]]，每个元素都是handler转化的Denpendent handler

Matcher的实例化

Matcher通过各种on函数生成，如on、on_metaevent、on_message等
这里以on()为例进行分析

def on(
    type: str = "",
    rule: Optional[Union[Rule, T_RuleChecker]] = None,
    permission: Optional[Union[Permission, T_PermissionChecker]] = None,
    *,
    handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
    temp: bool = False,
    expire_time: Optional[Union[datetime, timedelta]] = None,
    priority: int = 1,
    block: bool = False,
    state: Optional[T_State] = None,
    _depth: int = 0,
) -> Type[Matcher]:
    plugin_chain = _current_plugin_chain.get() # 姑且当作Matcher所在插件，具体分析要到第四章杂项
    matcher = Matcher.new(
        type,
        Rule() & rule,
        Permission() | permission,
        temp=temp,
        expire_time=expire_time,
        priority=priority,
        block=block,
        handlers=handlers,
        plugin=plugin_chain[-1] if plugin_chain else None,
        module=_get_matcher_module(_depth + 1),
        default_state=state,
    )
    _store_matcher(matcher) # 姑且认为Matcher被存到所在plugin中，具体分析要到第四章杂项
    return matcher

显然核心代码是Matcher.new

class Matcher(metaclass=MatcherMeta):
    @classmethod
    def new(
        cls,
        type_: str = "",
        rule: Optional[Rule] = None,
        permission: Optional[Permission] = None,
        handlers: Optional[List[Union[T_Handler, Dependent[Any]]]] = None,
        temp: bool = False,
        priority: int = 1,
        block: bool = False,
        *,
        plugin: Optional["Plugin"] = None,
        module: Optional[ModuleType] = None,
        expire_time: Optional[Union[datetime, timedelta]] = None,
        default_state: Optional[T_State] = None,
        default_type_updater: Optional[Union[T_TypeUpdater, Dependent[str]]] = None,
        default_permission_updater: Optional[
            Union[T_PermissionUpdater, Dependent[Permission]]
        ] = None,
    ) -> Type["Matcher"]:
        NewMatcher = type(
            "Matcher",
            (Matcher,),
            {
                "plugin": plugin,
                "module": module,
                "plugin_name": plugin and plugin.name,
                "module_name": module and module.__name__,
                "type": type_,
                "rule": rule or Rule(),
                "permission": permission or Permission(),
                "handlers": [
                    handler
                    if isinstance(handler, Dependent)
                    else Dependent[Any].parse(
                        call=handler, allow_types=cls.HANDLER_PARAM_TYPES
                    )
                    for handler in handlers
                ]
                if handlers
                else [],
                "temp": temp,
                "expire_time": (
                    expire_time
                    and (
                        expire_time
                        if isinstance(expire_time, datetime)
                        else datetime.now() + expire_time
                    )
                ),
                "priority": priority,
                "block": block,
                "_default_state": default_state or {},
                "_default_type_updater": (
                    default_type_updater
                    and (
                        default_type_updater
                        if isinstance(default_type_updater, Dependent)
                        else Dependent[str].parse(
                            call=default_type_updater,
                            allow_types=cls.HANDLER_PARAM_TYPES,
                        )
                    )
                ),
                "_default_permission_updater": (
                    default_permission_updater
                    and (
                        default_permission_updater
                        if isinstance(default_permission_updater, Dependent)
                        else Dependent[Permission].parse(
                            call=default_permission_updater,
                            allow_types=cls.HANDLER_PARAM_TYPES,
                        )
                    )
                ),
            },
        )
        matchers[priority].append(NewMatcher) # 根据优先级将Matcher子类储存
        return NewMatcher

Matcher的元类是MatcherMeta，可以说卵用没有，唯一的作用是实现Matcher的自定义打印
type的作用是生成类，例如

class A:
    def class_name(self):
        return self.__class__

    @staticmethod
    def number():
        return 1

    @property
    def string(self):
        return "233"

def new_class_name(self):
    return str(self.__class__) + "new"

def new_number():
    return 2

def new_string(self):
    return "666"

B = type(
    "B", # 类名
    (A,), # 父类
    {
        "class_name": new_class_name,
        "number": staticmethod(new_number),
        "string": property(new_string),
        "show": lambda self: print(self.class_name(), self.number(), self.string)
    } # 类方法、对象方法、类属性
)

B().show()

实际上与下面写法相同

class B(A):
    def class_name(self):
        return str(self.__class__) + "new"

    @staticmethod
    def number():
        return 2

    @property
    def string(self):
        return "666"

    def show(self):
        print(self.class_name(), self.number(), self.string)

总之随便你怎么折腾都行
可以看出，Matcher.new生成的是Matcher子类，重写了父类的以下类属性

plugin: Optional["Plugin"] = None
"""Matcher所在插件，通过_current_plugin_chain.get获取，具体代码将在第四章分析"""

module: Optional[ModuleType] = None
"""Matcher所在插件模块，通过_get_matcher_module获取，具体代码将在第四章分析"""

plugin_name: Optional[str] = None
"""Matcher所在插件名，用plugin.name或plugin类作为名称"""

module_name: Optional[str] = None
"""Matcher所在点分割插件模块路径，module.\__name__或module类作为名称"""

type: str = ""
"""
Matcher类型，Matcher的类型决定了事件匹配规则
例如on_metaevent的类型是“meta_event”、on_message的类型是“message”等，可以再nonebot.plugin.on中查看
"""

rule: Rule = Rule()
"""Matcher匹配规则，见后文详细介绍"""

permission: Permission = Permission()
"""Matcher触发权限，见后文详细介绍"""

handlers: List[Dependent[Any]] = []
"""
Matcher拥有的事件处理函数列表，可以在on函数注册，如on_message(handlers=[handler1,handler2])
也可以通过Matcher的其它方法注册，见后文
"""

priority: int = 1
"""Matcher优先级，顾名思义，优先级越高小越先运行，同一优先级的同时运行（详见第三节）"""

block: bool = False
"""Matcher是否阻止事件传播，无论Matcher是否被触发"""

temp: bool = False
"""Matcher是否为临时，无论Matcher是否被触发"""

expire_time: Optional[datetime] = None
"""Matcher过期时间，可以是datetime或timedelta"""

_default_state: T_State = {}
"""
Matcher默认状态，用于存储上下文，例如对话系统可能用到，默认为空字典
你也可以在初始化的时候放一些值进去
"""

_default_type_updater: Optional[Dependent[str]] = None
"""
Matcher类型更新函数，Matcher.pause, Matcher.reject（第三节介绍）时被运行
用于更新Matcher的type。默认会更新为message
"""

_default_permission_updater: Optional[Dependent[Permission]] = None
"""
Matcher权限更新函数，在Matcher.pause, Matcher.reject（第三节介绍）时被运行
用于更新会话对象权限。默认会更新为当前事件的触发对象
"""

Matcher的HANDLER_PARAM_TYPES有DependParam、BotParam、EventParam、StateParam、ArgParam、MatcherParam、DefaultParam

Handlers的注册

Matcher.append_handler

class Matcher(metaclass=MatcherMeta):
    ...
    @classmethod
    def append_handler(
        cls, handler: T_Handler, parameterless: Optional[Iterable[Any]] = None
    ) -> Dependent[Any]:
        handler_ = Dependent[Any].parse( #将handler转化为Dependent
            call=handler,
            parameterless=parameterless,
            allow_types=cls.HANDLER_PARAM_TYPES,
        )
        cls.handlers.append(handler_) # 添加到Matcher.handlers列表中
        return handler_ # 返回值是Dependent handler，不是handler，和后面三种注册方法不同
    ...
    # 例如有多个handler需要注册，可以循环使用append_handler。一般不作为装饰器使用

Matcher.handle

class Matcher(metaclass=MatcherMeta):
    ...
    @classmethod
    def handle(
        cls, parameterless: Optional[Iterable[Any]] = None
    ) -> Callable[[T_Handler], T_Handler]:
        def _decorator(func: T_Handler) -> T_Handler:
            cls.append_handler(func, parameterless=parameterless)
            return func
        return _decorator
    ...
    # 和append_handler几乎相同，作为装饰器使用，返回值是handler本身

Matcher.receive

class Matcher(metaclass=MatcherMeta):
    ...
    @classmethod
    def receive(
        cls, id: str = "", parameterless: Optional[Iterable[Any]] = None
    ) -> Callable[[T_Handler], T_Handler]:
        async def _receive(event: Event, matcher: "Matcher") -> Union[None, NoReturn]:
            matcher.set_target(RECEIVE_KEY.format(id=id))
            if matcher.get_target() == RECEIVE_KEY.format(id=id):
                matcher.set_receive(id, event)
                return
            if matcher.get_receive(id, ...) is not ...:
                return
            await matcher.reject()

        ```
        额外添加Depends(_receive)到parameterless，那么有什么用呢？
        由于我们现在连parameterless是干什么的都不知道，因此下一节再介绍Depends(_receive)
        ```
        _parameterless = (Depends(_receive), *(parameterless or tuple()))

        def _decorator(func: T_Handler) -> T_Handler:
            if cls.handlers and cls.handlers[-1].call is func:
                ```
                如果Matcher的handlers不为空，且handlers[-1]的func也是当前传入的handler
                例如
                @xxx.receive() receive要在handle等的上面
                @xxx.handle()
                async def(xxx:bot):
                    pass
                ```
                func_handler = cls.handlers[-1]
                new_handler = Dependent(
                    call=func_handler.call,
                    params=func_handler.params,
                    parameterless=Dependent.parse_parameterless( # 将参数转化为Param子类
                        tuple(_parameterless), cls.HANDLER_PARAM_TYPES
                    )
                    + func_handler.parameterless,
                )
                cls.handlers[-1] = new_handler # 将handlers[-1]的parameterless加上receive的parameterless和Depends(_receive)
            else:
                ```
                其他情况，按正常流程注册
                @xxx.receive()
                async def(xxx:bot):
                    pass
                ```
                cls.append_handler(func, parameterless=_parameterless)
            return func

        return _decorator
    ...
    # 可见Matcher.receive注册的Dependent handler的核心在于parameterless的第一位是Depends(_receive)，其它与普通Dependent handler没区别

Matcher.got

class Matcher(metaclass=MatcherMeta):
    ...
    @classmethod
    def got(
        cls,
        key: str,
        prompt: Optional[Union[str, Message, MessageSegment, MessageTemplate]] = None,
        parameterless: Optional[Iterable[Any]] = None,
    ) -> Callable[[T_Handler], T_Handler]:
        async def _key_getter(event: Event, matcher: "Matcher"):
            matcher.set_target(ARG_KEY.format(key=key))
            if matcher.get_target() == ARG_KEY.format(key=key):
                matcher.set_arg(key, event.get_message())
                return
            if matcher.get_arg(key, ...) is not ...:
                return
            await matcher.reject(prompt)

        _parameterless = (Depends(_key_getter), *(parameterless or tuple()))

        def _decorator(func: T_Handler) -> T_Handler:
            if cls.handlers and cls.handlers[-1].call is func:
                ```
                用法一
                @xxx.got() got要在handle等的上面
                @xxx.handle()
                async def(xxx:bot):
                    pass
                ```
                func_handler = cls.handlers[-1]
                new_handler = Dependent(
                    call=func_handler.call,
                    params=func_handler.params,
                    parameterless=Dependent.parse_parameterless(
                        tuple(_parameterless), cls.HANDLER_PARAM_TYPES
                    )
                    + func_handler.parameterless,
                )
                cls.handlers[-1] = new_handler
            else:
                ```
                用法二
                @xxx.got()
                async def(xxx:bot):
                    pass
                ```
                cls.append_handler(func, parameterless=_parameterless)
            return func

        return _decorator
    # 和Matcher.receive几乎一样，只不过parameterless第一位是Depends(_key_getter)，同样下节再介绍

Rule

只介绍Rule的实例化，关于Rule的运行将在下节介绍

class Rule:
    HANDLER_PARAM_TYPES = [
        DependParam,
        BotParam,
        EventParam,
        StateParam,
        DefaultParam,
    ]

    def __init__(self, *checkers: Union[T_RuleChecker, Dependent[bool]]) -> None:
        ```
        T_RuleChecker = _DependentCallable[bool]
        _DependentCallable = Union[Callable[..., T], Callable[..., Awaitable[T]]]
        所以返回值为bool的同步函数、异步函数和对应Dependent都可以作为Rule
        ```
        self.checkers: Set[Dependent[bool]] = { # 把不是Dependent的Rule转化为Dependent Rule
            checker
            if isinstance(checker, Dependent)
            else Dependent[bool].parse(
                call=checker, allow_types=self.HANDLER_PARAM_TYPES # 相较于Matcher，Rule的参数不支持ArgParam和MatcherParam，因为此时还没有运行到Matcher
            )
            for checker in checkers
        }

    def __and__(self, other: Optional[Union["Rule", T_RuleChecker]]) -> "Rule":
        ```
        支持左 & 运算
        Rule() & Rule()
        Rule() & Union[T_RuleChecker, Dependent[bool]]类型
        最终生成一个新的Rule
        Rule() & None 返回自身，防止意外None值
        ```
        if other is None:
            return self
        elif isinstance(other, Rule):
            return Rule(*self.checkers, *other.checkers)
        else:
            return Rule(*self.checkers, other)

    def __rand__(self, other: Optional[Union["Rule", T_RuleChecker]]) -> "Rule":
        ```
        支持右 & 运算
        Rule() & Rule()
        Union[T_RuleChecker, Dependent[bool]]类型 & Rule()
        最终生成一个新的Rule
        None & Rule() 返回自身，防止意外None值
        ```
        if other is None:
            return self
        elif isinstance(other, Rule):
            return Rule(*other.checkers, *self.checkers)
        else:
            return Rule(other, *self.checkers)

    def __or__(self, other: object) -> NoReturn:
        ```
        不支持 | 运算
        ```
        raise RuntimeError("Or operation between rules is not allowed.")

Permission

class Permission:
    HANDLER_PARAM_TYPES = [
        DependParam,
        BotParam,
        EventParam,
        DefaultParam,
    ]

    def __init__(self, *checkers: Union[T_PermissionChecker, Dependent[bool]]) -> None:
        ```
        T_PermissionChecker = _DependentCallable[bool]
        _DependentCallable = Union[Callable[..., T], Callable[..., Awaitable[T]]]
        ```
        self.checkers: Set[Dependent[bool]] = {
            checker
            if isinstance(checker, Dependent)
            else Dependent[bool].parse(
                call=checker, allow_types=self.HANDLER_PARAM_TYPES
            )
            for checker in checkers
        }
        # 逻辑和Rule几乎一样，不过HANDLER_PARAM_TYPES又比Rule少了StateParam，因为此时还没有上下文

    def __and__(self, other: object) -> NoReturn:
        ```
        不支持 & 运算
        ```
        raise RuntimeError("And operation between Permissions is not allowed.")

    def __or__(
        self, other: Optional[Union["Permission", T_PermissionChecker]]
    ) -> "Permission":
        ```
        支持左 | 运算
        Permission() | Permission()
        Permission() | Union[T_PermissionChecker, Dependent[bool]]类型
        最终生成一个新的Permission
        Permission() | None 返回自身，防止意外None值
        ```
        if other is None:
            return self
        elif isinstance(other, Permission):
            return Permission(*self.checkers, *other.checkers)
        else:
            return Permission(*self.checkers, other)

    def __ror__(
        self, other: Optional[Union["Permission", T_PermissionChecker]]
    ) -> "Permission":
        ```
        支持右 | 运算
        Permission() | Permission()
        Union[T_PermissionChecker, Dependent[bool]]类型 | Permission()
        最终生成一个新的Permission
        None | Permission() 返回自身，防止意外None值
        ```
        if other is None:
            return self
        elif isinstance(other, Permission):
            return Permission(*other.checkers, *self.checkers)
        else:
            return Permission(other, *self.checkers)

on、on_metaevent、on_message、on_notice、on_request

用于生成Matcher子类
分别可以生成Matcher.type为任意、meta_event、message、notice、request的Matcher子类

```
主要代码已在前文介绍
默认不block
```
def on(
	type: str = "",
	rule: Optional[Union[Rule, T_RuleChecker]] = None,
	permission: Optional[Union[Permission, T_PermissionChecker]] = None,
	*,
	handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
	temp: bool = False,
	expire_time: Optional[Union[datetime, timedelta]] = None,
	priority: int = 1,
	block: bool = False,
	state: Optional[T_State] = None,
	_depth: int = 0,
) -> Type[Matcher]:
	plugin_chain = _current_plugin_chain.get()
	matcher = Matcher.new(
		type,
		Rule() & rule,
		Permission() | permission,
		temp=temp,
		expire_time=expire_time,
		priority=priority,
		block=block,
		handlers=handlers,
		plugin=plugin_chain[-1] if plugin_chain else None,
		module=_get_matcher_module(_depth + 1),
		default_state=state,
	)
	_store_matcher(matcher)
	return matcher

```
Matcher.type为meta_event
没有permission，因为这不是用户触发的事件
默认不block
```
def on_metaevent(
	rule: Optional[Union[Rule, T_RuleChecker]] = None,
	*,
	handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
	temp: bool = False,
	expire_time: Optional[Union[datetime, timedelta]] = None,
	priority: int = 1,
	block: bool = False,
	state: Optional[T_State] = None,
	_depth: int = 0,
) -> Type[Matcher]:
	plugin_chain = _current_plugin_chain.get()
	matcher = Matcher.new(
		"meta_event",
		Rule() & rule,
		Permission(),
		temp=temp,
		expire_time=expire_time,
		priority=priority,
		block=block,
		handlers=handlers,
		plugin=plugin_chain[-1] if plugin_chain else None,
		module=_get_matcher_module(_depth + 1),
		default_state=state,
	)
	_store_matcher(matcher)
	return matcher

```
Matcher.type为on_message
默认会block
```
def on_message(
	rule: Optional[Union[Rule, T_RuleChecker]] = None,
	permission: Optional[Union[Permission, T_PermissionChecker]] = None,
	*,
	handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
	temp: bool = False,
	expire_time: Optional[Union[datetime, timedelta]] = None,
	priority: int = 1,
	block: bool = True,
	state: Optional[T_State] = None,
	_depth: int = 0,
) -> Type[Matcher]:
	plugin_chain = _current_plugin_chain.get()
	matcher = Matcher.new(
		"message",
		Rule() & rule,
		Permission() | permission,
		temp=temp,
		expire_time=expire_time,
		priority=priority,
		block=block,
		handlers=handlers,
		plugin=plugin_chain[-1] if plugin_chain else None,
		module=_get_matcher_module(_depth + 1),
		default_state=state,
	)
	_store_matcher(matcher)
	return matcher

```
Matcher.type为notice
没有permission，因为这不是用户触发的事件
默认会block
```
def on_notice(
	rule: Optional[Union[Rule, T_RuleChecker]] = None,
	*,
	handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
	temp: bool = False,
	expire_time: Optional[Union[datetime, timedelta]] = None,
	priority: int = 1,
	block: bool = True,
	state: Optional[T_State] = None,
	_depth: int = 0,
) -> Type[Matcher]:
	plugin_chain = _current_plugin_chain.get()
	matcher = Matcher.new(
		"notice",
		Rule() & rule,
		Permission(),
		temp=temp,
		expire_time=expire_time,
		priority=priority,
		block=block,
		handlers=handlers,
		plugin=plugin_chain[-1] if plugin_chain else None,
		module=_get_matcher_module(_depth + 1),
		default_state=state,
	)
	_store_matcher(matcher)
	return matcher

```
Matcher.type为request
没有permission，因为这不是用户触发的事件
默认不block
```
def on_request(
	rule: Optional[Union[Rule, T_RuleChecker]] = None,
	*,
	handlers: Optional[List[Union[T_Handler, Dependent]]] = None,
	temp: bool = False,
	expire_time: Optional[Union[datetime, timedelta]] = None,
	priority: int = 1,
	block: bool = False,
	state: Optional[T_State] = None,
	_depth: int = 0,
) -> Type[Matcher]:
	plugin_chain = _current_plugin_chain.get()
	matcher = Matcher.new(
		"request",
		Rule() & rule,
		Permission(),
		temp=temp,
		expire_time=expire_time,
		priority=priority,
		block=block,
		handlers=handlers,
		plugin=plugin_chain[-1] if plugin_chain else None,
		module=_get_matcher_module(_depth + 1),
		default_state=state,
	)
	_store_matcher(matcher)
	return matcher

对于其它on函数将在下一节介绍

发表评论 取消回复
表情 图片 链接 代码

分享