melobot.handle¶

处理流¶

class melobot.handle.Flow[源代码]¶

基类：object

处理流

变量:: name (str) -- 处理流的标识

__init__(name: str, *edge_maps: ~typing.Iterable[~typing.Iterable[~melobot.handle.base.FlowNode] | ~melobot.handle.base.FlowNode], priority: int = 0, guard: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool | None] | None = None) → None[源代码]¶

初始化处理流

参数:

name (str) -- 处理流的标识
edge_maps (Iterable[Iterable[FlowNode] | FlowNode]) -- 边映射，遵循 melobot 的 graph edges 表示方法
priority (int) -- 处理流的优先级
guard (SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool | None] | None) -- 守卫函数。在处理流运行前调用，返回 True 不再继续运行处理流。默认不启用

返回类型:

None

update_priority(priority: int) → None[源代码]¶

更新处理流优先级

参数:: priority (int) -- 新优先级
返回类型:: None

dismiss() → None[源代码]¶

停用处理流

停用后将无法处理任何新事件，也无法再次恢复使用

返回类型:: None

is_active() → bool[源代码]¶

判断处理流是否处于可用状态

返回:: 是否可用
返回类型:: bool

set_guard(guard: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool | None]) → None[源代码]¶

设置或重设守卫函数

参数:: guard (SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool | None]) -- 守卫函数
返回类型:: None

link(flow: Flow, priority: int | None = None) → Flow[源代码]¶

连接另一处理流返回新处理流，并设置新优先级

新处理流守卫函数为空，使用 set_guard() 自行添加

参数:

flow (Flow) -- 连接的新流
priority (int | None) -- 新优先级，若为空，则使用两者中较小的优先级

返回:

新的处理流

返回类型:

Flow

start(node: FlowNode) → FlowNode[源代码]¶

设置处理流起始结点的装饰器

参数:: node (FlowNode) -- 起始结点
返回:: 起始结点
返回类型:: FlowNode

after(node: FlowNode) → Callable[[FlowNode], FlowNode][源代码]¶

在处理流某一参照结点后，添加新结点的装饰器函数

参数:: node (FlowNode) -- 参照结点
返回:: 被装饰的结点
返回类型:: Callable[[FlowNode], FlowNode]

before(node: FlowNode) → Callable[[FlowNode], FlowNode][源代码]¶

在处理流某一参照结点前，添加新结点的装饰器函数

参数:: node (FlowNode) -- 参照结点
返回:: 被装饰的结点
返回类型:: Callable[[FlowNode], FlowNode]

merge(*nodes: FlowNode) → Callable[[FlowNode], FlowNode][源代码]¶

将处理流某几结点的控制流，在当前结点合并的装饰器函数

返回:: 被装饰的结点
参数:: nodes (FlowNode)
返回类型:: Callable[[FlowNode], FlowNode]

fork(*nodes: FlowNode) → Callable[[FlowNode], FlowNode][源代码]¶

将处理流当前结点的控制流，分流到某几个结点的装饰器函数

返回:: 被装饰的结点
参数:: nodes (FlowNode)
返回类型:: Callable[[FlowNode], FlowNode]

class melobot.handle.FlowNode[源代码]¶

基类：object

处理流结点

处理结点¶

melobot.handle.node(func: SyncOrAsyncCallable[..., bool | None]) → FlowNode[源代码]¶

处理结点装饰器，将当前异步可调用对象装饰为一个处理结点

参数:: func (SyncOrAsyncCallable[..., bool | None])
返回类型:: FlowNode

melobot.handle.no_deps_node(func: SyncOrAsyncCallable[..., bool | None]) → FlowNode[源代码]¶

与 node() 类似，但是不自动为结点标记依赖注入。

需要后续使用 inject_deps() 手动标记依赖注入，这适用于某些对处理结点进行再装饰的情况

参数:: func (SyncOrAsyncCallable[..., bool | None])
返回类型:: FlowNode

流装饰器¶

流装饰相关的组件，可以将一个普通函数装饰为一个处理流。

这些组件，在 melobot 的教程中，早期我们称它们为“事件绑定方法”或“事件绑定函数”。

class melobot.handle.FlowDecorator[源代码]¶

基类：object

__init__(checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool] = None, matcher: ~melobot.utils.match.base.Matcher | None = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, rule: ~melobot.session.option.Rule[~melobot.adapter.model.Event] | None = None) → None[源代码]¶

处理流装饰器

参数:

checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool]) -- 检查器
matcher (Matcher | None) -- 匹配器（指定匹配器，需要先验证已经是文本事件）
parser (Parser | None) -- 解析器（指定解析器，需要先验证已经是文本事件）
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
rule (Rule[Event] | None) -- 会话规则

返回类型:

None

__call__(func: SyncOrAsyncCallable[..., bool | None]) → Flow[源代码]¶

Call self as a function.

参数:: func (SyncOrAsyncCallable[..., bool | None])
返回类型:: Flow

melobot.handle.on_event(checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool] = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, rule: ~melobot.session.option.Rule[~melobot.adapter.model.Event] | None = None) → FlowDecorator[源代码]¶

绑定任意事件的处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.Event'>,), bool]) -- 检查器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
rule (Rule[Event] | None) -- 会话规则

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_text(checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, matcher: ~melobot.utils.match.base.Matcher | None = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
matcher (Matcher | None) -- 匹配器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_start_match(target: str | ~typing.Sequence[str], logic_mode: ~melobot.typ._enum.LogicMode = LogicMode.OR, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“字符串起始匹配”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

target (str | Sequence[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_contain_match(target: str | ~typing.Sequence[str], logic_mode: ~melobot.typ._enum.LogicMode = LogicMode.OR, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“字符串包含匹配”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

target (str | Sequence[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_end_match(target: str | ~typing.Sequence[str], logic_mode: ~melobot.typ._enum.LogicMode = LogicMode.OR, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“字符串末尾匹配”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

target (str | Sequence[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_full_match(target: str | ~typing.Sequence[str], logic_mode: ~melobot.typ._enum.LogicMode = LogicMode.OR, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“字符串完整匹配”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

target (str | Sequence[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_regex_match(target: str, regex_flags: ~typing_extensions.Any = 0, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, parser: ~melobot.utils.parse.base.Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“正则匹配”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

target (str) -- 匹配的目标字符串
regex_flags (Any) -- 正则匹配的 flags
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_command(cmd_start: str | ~typing.Iterable[str], cmd_sep: str | ~typing.Iterable[str], targets: str | ~typing.Sequence[str], fmtters: ~typing.Sequence[~melobot.utils.parse.cmd.CmdArgFormatter | None] | None = None, checker: ~melobot.utils.check.base.Checker | None | ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool] = None, matcher: ~melobot.utils.match.base.Matcher | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: ~typing.Sequence[~typing.Callable[[~typing.Callable], ~typing.Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

绑定文本事件的“命令解析”处理流装饰方法

在前期的教程中，处理流装饰方法也称为绑定方法

参数:

cmd_start (str | Iterable[str]) -- 命令起始符
cmd_sep (str | Iterable[str]) -- 命令分隔符
targets (str | Sequence[str]) -- 解析匹配的命令名
fmtters (Sequence[CmdArgFormatter | None] | None) -- 命令参数格式化器
checker (Checker | None | SyncOrAsyncCallable[(<class 'melobot.adapter.model.TextEvent'>,), bool]) -- 检查器
matcher (Matcher | None) -- 匹配器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
legacy_session (bool) -- 是否自动启用传统会话

返回:

处理流装饰器

返回类型:

FlowDecorator

处理流控制¶

async melobot.handle.nextn() → None[源代码]¶

运行下一处理结点（在处理流中使用）

返回类型:: None

async melobot.handle.block() → None[源代码]¶

阻止当前事件向更低优先级的处理流传播（在处理流中使用）

返回类型:: None

async melobot.handle.bypass() → NoReturn[源代码]¶

立即跳过当前处理结点剩下的步骤，运行下一处理结点（在处理流中使用）

返回类型:: NoReturn

async melobot.handle.rewind() → NoReturn[源代码]¶

立即重新运行当前处理结点（在处理流中使用）

返回类型:: NoReturn

async melobot.handle.stop() → NoReturn[源代码]¶

立即停止当前处理流（在处理流中使用）

返回类型:: NoReturn

async melobot.handle.flow_to(flow: Flow) → None[源代码]¶

立即进入一个其他处理流（在处理流中使用）

参数:: flow (Flow)
返回类型:: None

处理流状态¶

class melobot.handle.FlowStore[源代码]¶

基类：dict[str, Any]

流存储，将会在流运行前初始化，运行结束后销毁

class melobot.handle.FlowRecordStage[源代码]¶

基类：Enum

流记录阶段的枚举

FLOW_START = 'fs'¶

FLOW_EARLY_FINISH = 'fef'¶

FLOW_FINISH = 'ff'¶

NODE_START = 'ns'¶

DEPENDS_NOT_MATCH = 'dnm'¶

BLOCK = 'bl'¶

STOP = 'st'¶

BYPASS = 'by'¶

REWIND = 're'¶

NODE_EARLY_FINISH = 'nef'¶

NODE_FINISH = 'nf'¶

class melobot.handle.FlowRecord[源代码]¶

基类：object

流记录

stage: FlowRecordStage¶

flow_name: str¶

node_name: str¶

event: model.Event¶

prompt: str = ''¶

melobot.handle.get_flow_store() → FlowStore[源代码]¶

获取当前上下文中的流存储

返回:: 流存储
返回类型:: FlowStore

melobot.handle.get_flow_records() → tuple[FlowRecord, ...][源代码]¶

获取当前上下文中的流记录

返回:: 流记录
返回类型:: tuple[FlowRecord, ...]

melobot.handle.get_event() → Event[源代码]¶

获取当前上下文中的事件

返回:: 事件
返回类型:: Event

melobot.handle.try_get_event() → Event | None[源代码]¶

尝试获取当前上下文中的事件

返回:: 事件或空
返回类型:: Event | None

上下文动态变量¶

melobot.handle.f_records¶: 当前上下文中的流记录，类型为 tuple[FlowRecord, ...]

melobot.handle.f_store¶: 当前上下文中的流存储，类型为 FlowStore

melobot.handle.event¶: 当前上下文中的事件，类型为 Event