melobot.handle¶

处理流¶

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

基类：LogMixin

处理流

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

__init__(name: str, *edge_maps: Iterable[Iterable[FlowNode] | FlowNode], priority: int = 0) → None[源代码]¶

初始化处理流

参数:

name (str) -- 处理流的标识
edge_maps (Iterable[Iterable[FlowNode] | FlowNode]) -- 边映射，遵循 melobot 的 graph edges 表示方法
priority (int) -- 处理流的优先级

返回类型:

None

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

更新处理流优先级

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

dismiss() → None[源代码]¶

停用处理流

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

返回类型:: None

is_active() → bool[源代码]¶

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

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

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

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

参数:

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

返回:

新的处理流

返回类型:

Flow

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: Checker | None | Callable[[Event], bool] = None, matcher: Matcher | None = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, rule: Rule[Event] | None = None) → None[源代码]¶

处理流装饰器

参数:

checker (Checker | None | Callable[[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[源代码]¶

实现了此方法，因此可被当作装饰器使用

参数:: func (SyncOrAsyncCallable[..., bool | None]) -- 被装饰的函数
返回:: 处理流对象
返回类型:: Flow

melobot.handle.on_event(checker: Checker | None | Callable[[Event], bool] = None, matcher: Matcher | None = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, rule: Rule[Event] | None = None) → FlowDecorator[源代码]¶

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

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

参数:

checker (Checker | None | Callable[[Event], bool]) -- 检查器
matcher (Matcher | None) -- 匹配器
parser (Parser | None) -- 解析器
priority (int) -- 优先级
block (bool) -- 是否阻断向低优先级传播
temp (bool) -- 是否临时使用（处理一次事件后停用）
decos (Sequence[Callable[[Callable], Callable]] | None) -- 装饰器组
rule (Rule[Event] | None) -- 会话规则

返回:

处理流装饰器

返回类型:

FlowDecorator

melobot.handle.on_text(checker: Checker | None | Callable[[TextEvent], bool] = None, matcher: Matcher | None = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

checker (Checker | None | Callable[[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 | list[str], logic_mode: LogicMode = LogicMode.OR, checker: Checker | None | Callable[[TextEvent], bool] = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

target (str | list[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | Callable[[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 | list[str], logic_mode: LogicMode = LogicMode.OR, checker: Checker | None | Callable[[TextEvent], bool] = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

target (str | list[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | Callable[[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 | list[str], logic_mode: LogicMode = LogicMode.OR, checker: Checker | None | Callable[[TextEvent], bool] = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

target (str | list[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | Callable[[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 | list[str], logic_mode: LogicMode = LogicMode.OR, checker: Checker | None | Callable[[TextEvent], bool] = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

target (str | list[str]) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | Callable[[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, logic_mode: LogicMode = LogicMode.OR, checker: Checker | None | Callable[[TextEvent], bool] = None, parser: Parser | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

target (str) -- 匹配的目标字符串
logic_mode (LogicMode) -- 匹配的逻辑模式
checker (Checker | None | Callable[[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 | list[str], cmd_sep: str | list[str], targets: str | list[str], fmtters: list[CmdArgFormatter | None] | None = None, checker: Checker | None | Callable[[TextEvent], bool] = None, matcher: Matcher | None = None, priority: int = 0, block: bool = False, temp: bool = False, decos: Sequence[Callable[[Callable], Callable]] | None = None, legacy_session: bool = False) → FlowDecorator[源代码]¶

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

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

参数:

cmd_start (str | list[str]) -- 命令起始符
cmd_sep (str | list[str]) -- 命令分隔符
targets (str | list[str]) -- 解析匹配的命令名
fmtters (list[CmdArgFormatter | None] | None) -- 命令参数格式化器
checker (Checker | None | Callable[[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