melobot.base

abc 模块

本模块包含 melobot 重要的抽象类或基类。

class melobot.base.abc.AbstractConnector

抽象连接器类

提示

一般无需手动实例化该类，多数情况会直接使用本类对象，或将本类用作类型注解。

logger: BotLogger

连接器的日志器

slack: bool

是否在 slack 状态

cd_time

连接器发送行为操作的冷却时间

allow_reconn

是否允许重新连接

class melobot.base.abc.BotEvent

事件基类

提示

一般无需手动实例化该类，多数情况会直接使用本类对象，或将本类用作类型注解。

raw: dict

从 onebot 实现获得的事件原始值

abstract property time: int

事件发生的时间

abstract property type: Literal['message', 'request', 'notice', 'meta']

事件类型

is_msg_event() → bool

判断是否是消息事件

is_req_event() → bool

判断是否是请求事件

is_notice_event() → bool

判断是否是通知事件

is_meta_event() → bool

判断是否是元事件

class melobot.base.abc.BotAction

行为类

每个行为操作应该产生一个行为对象

提示

一般无需手动实例化该类，多数情况会直接使用本类对象，或将本类用作类型注解。

type: str

行为类型。对应 onebot 标准中的 API 终结点名称

params: dict

行为参数。对应 onebot 标准中向 API 传送的数据的 params 字段

trigger: BotEvent | None

行为的触发事件，一般是行为生成时所在会话的事件（大多数行为操作函数生成 BotAction 对象时，会自动填充此属性，但少数情况下此属性可能为空）

class melobot.base.abc.SessionRule

会话规则基类

会话规则用于判断两事件是否在同一会话中。

__init__() → None

abstract compare(e1: Event_T, e2: Event_T) → bool

判断两事件是否在同一会话中的判断方法

任何会话规则应该实现此抽象方法。

参数:

• e1 – 判断时的事件1

• e2 – 判断时的事件2

返回:

判断结果

class melobot.base.abc.BotChecker

检查器基类

__init__() → None

abstract async check(event: Event_T) → bool

检查器检查方法

任何检查器应该实现此抽象方法。

参数:

event – 给定的事件

返回:

检查是否通过

class melobot.base.abc.WrappedChecker

合并检查器

在两个 BotChecker 对象间使用 | & ^ ~ 运算符即可返回合并检查器。

提示

一般无需手动实例化该类，多数情况会直接使用本类对象，或将本类用作类型注解。

class melobot.base.abc.BotMatcher

匹配器基类

__init__() → None

abstract async match(text: str) → bool

匹配器匹配方法

任何匹配器应该实现此抽象方法。

参数:

text – 消息事件的文本内容

返回:

是否匹配

class melobot.base.abc.WrappedMatcher

合并匹配器

在两个 BotMatcher 对象间使用 | & ^ ~ 运算符即可返回合并匹配器

提示

一般无需手动实例化该类，多数情况会直接使用本类对象，或将本类用作类型注解。

class melobot.base.abc.BotParser

解析器基类

解析器一般用作从消息文本中按规则批量提取参数

__init__() → None

abstract async parse(text: str) → ParseArgs | None

解析方法

任何解析器应该实现此抽象方法

参数:

text – 消息文本内容

返回:

解析结果

exceptions 模块

本模块包含 melobot 的异常类，此处仅展示对外部有利用价值的异常类。

class melobot.base.exceptions.BotException

Bot 异常基类

在使用 melobot 编写代码时，若需要抛出异常，可以有意抛出该类。 表明这是你设计的异常情况，而不是完全预期之外的异常。

class melobot.base.exceptions.BotSessionTimeout

会话暂停的超时异常

tools 模块

本模块包含了 melobot 内部使用的工具类和工具函数，它们也可被外部使用。

class melobot.base.tools.RWController

异步读写控制器

提供异步安全的读写上下文。在读取时可以多读，同时读写互斥。

使用方法：

rwc = RWController()
# 读时使用此控制器的安全读上下文：
async with rwc.safe_read():
    ...
# 写时使用此控制器的安全写上下文：
async with rwc.safe_write():
    ...

__init__(read_limit: int | None = None) → None

初始化一个异步读写控制器

参数:

read_limit – 读取的数量限制，为空则不限制

melobot.base.tools.get_id() → str

从 melobot 内部 id 获取器获得一个 id 值，不保证线程安全。

返回:

id 值

melobot.base.tools.this_dir(*relative_path: str) → str

包内 py 脚本通过该方法可获得所在目录的绝对路径。提供参数，还可拼接路径。

使用方法：

this_dir() # 获取 ./ 的绝对路径
this_dir('abc', 'env.toml') # 获取 ./abc/env.toml 的绝对路径

一些注意事项如下：

使用 this_dir()，只能这样导入：（导入语句后可以使用 as 子句）

from melobot import this_dir
# 或
from melobot.base.tools import this_dir

若 B.py 从 A.py 导入包含 this_dir() 调用的结构， 导入前 this_dir() 必须已运行，而不能延迟求值

A.py 中：

class Foo:
    DIR = this_dir()
    LAMBDA_DIR = lambda: this_dir()
    GET_DIR = lambda: this_dir()
OUTER_DIR = Foo.LAMBDA_DIR()  # Ok

B.py 中：

from .A import Foo, OUTER_DIR
OUTER_DIR      # Ok
Foo.DIR        # Ok
Foo.GET_DIR()  # Error

参数:

relative_path – 可用于拼接的路径部分

返回:

拼接后的绝对路径

melobot.base.tools.to_async(func: Callable[[], T]) → Callable[[], Coroutine[Any, Any, T]]

异步包装函数

将一个同步函数包装为异步函数，保留返回值。如果需要传参使用 partial 包裹。

参数:

func – 需要转换的函数

返回:

异步函数

melobot.base.tools.to_coro(obj: Callable[[], T] | Future[T]) → Coroutine[Any, Any, T]

协程包装函数

将一个同步函数或 Future 对象包装为协程，保留返回值。如果需要传参使用 partial 包裹。

参数:

obj – 需要包装的同步函数或 Future 对象

返回:

协程

melobot.base.tools.lock(callback: Callable[[], Awaitable[T1]] | None = None)

锁装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数加锁。

在获取锁冲突时，调用 callback 获得一个回调并执行。回调执行完毕后直接返回。

callback 参数为空，只应用 asyncio.Lock 的锁功能。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

callback – 获取锁冲突时的回调

melobot.base.tools.cooldown(busy_callback: Callable[[], Awaitable[T1]] | None = None, cd_callback: Callable[[float], Awaitable[T2]] | None = None, interval: float = 5)

冷却装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加 cd 时间。

如果被装饰函数已有一个在运行，此时调用 busy_callback 生成回调并执行。回调执行完毕后直接返回。

busy_callback 参数为空，则等待已运行的运行完成。随后执行下面的“冷却”处理逻辑。

当被装饰函数没有在运行的，但冷却时间未结束：

• cd_callback 不为空：使用 cd_callback 生成回调并执行。

• cd_callback 为空，被装饰函数持续等待，直至冷却结束再执行。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

• busy_callback – 已运行时的回调

• cd_callback – 冷却时间未结束的回调

• interval – 冷却时间

melobot.base.tools.semaphore(callback: Callable[[], Awaitable[T1]] | None = None, value: int = -1)

信号量装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加信号量控制。

在信号量无法立刻获取时，将调用 callback 获得回调并执行。回调执行完毕后直接返回。

callback 参数为空，只应用 asyncio.Semaphore 的信号量功能。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

• callback – 信号量无法立即获取的回调

• value – 信号量阈值

melobot.base.tools.timelimit(callback: Callable[[], Awaitable[T1]] | None = None, timeout: float = 5)

时间限制装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加超时控制。

超时之后，调用 callback 获得回调并执行，同时取消原任务。

callback 参数为空，如果超时，则抛出 asyncio.TimeoutError 异常。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

• callback – 超时时的回调

• timeout – 超时时间

melobot.base.tools.speedlimit(callback: Callable[[], Awaitable[T1]] | None = None, limit: int = 60, duration: int = 60)

流量/速率限制装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加流量控制：duration 秒内只允许 limit 次调用。

超出调用速率限制后，调用 callback 获得回调并执行，同时取消原任务。

callback 参数为空，等待直至满足速率控制要求再调用。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值。

参数:

• callback – 超出速率限制时的回调

• limit – duration 秒内允许调用多少次

• duration – 时长区间

melobot.base.tools.call_later(callback: Callable[[], None], delay: float)

同步函数延迟调度

在指定的 delay 后调度一个 callback 执行。callback 应该是同步方法。

参数:

• callback – 同步函数

• delay – 多长时间后调度

返回:

asyncio.TimerHandle 对象

melobot.base.tools.call_at(callback: Callable[[], None], timestamp: float)

同步函数指定时间调度

在指定的时间戳调度一个 callback 执行。callback 应该是同步方法。timestamp <= 当前时刻回调立即执行

参数:

• callback – 同步函数

• timestamp – 在什么时刻调度

返回:

asyncio.TimerHandle 对象

melobot.base.tools.async_later(callback: Coroutine[Any, Any, Any], delay: float) → Future[T]

异步函数延迟调度（可自主选择是否等待）

在指定的 delay 后调度一个 callback 执行。callback 是协程。

返回一个 asyncio.Future 对象，你可以选择等待或不等待。等待 asyncio.Future 即是等待 callback 的返回值。

注意：如果 callback 未完成就被取消，需要捕获 asyncio.CancelledError。

参数:

• callback – 异步函数（可有返回值）

• delay – 多长时间后调度

返回:

asyncio.Future 对象

melobot.base.tools.async_at(callback: Coroutine[Any, Any, Any], timestamp: float) → Future[T]

异步函数指定时间调度（可自主选择是否等待）

在指定的时间戳调度一个 callback 执行。callback 是协程。

返回一个 asyncio.Future 对象，你可以选择等待或不等待。等待 asyncio.Future 即是等待 callback 的返回值。

注意：如果 callback 未完成就被取消，需要捕获 asyncio.CancelledError。

参数:

• callback – 异步函数（可有返回值）

• timestamp – 在什么时刻调度

返回:

asyncio.Future 对象

melobot.base.tools.async_interval(callback: Callable[[], Coroutine[Any, Any, None]], interval: float) → Task[None]

异步函数间隔调度（类似 JavaScript 的 setInterval）

每过时间间隔执行 callback 一次。callback 是返回协程的可调用对象（异步函数或 lambda 函数等）。

返回一个 asyncio.Task 对象，可使用该 task 取消调度过程。

参数:

• callback – 异步函数

• interval – 调度的间隔

返回:

asyncio.Task 对象

typing 模块

本模块包含了 melobot 自定义的、用于类型注解的类型。可供外部参考。

class melobot.base.typing.MsgSegment

onebot 标准的消息段对象

type: str

data: dict[str, float | int | str]

class melobot.base.typing.StdCustomNodeData

onebot 标准的自定义消息结点数据

nickname: str

user_id: str

content: list[MsgSegment]

class melobot.base.typing.GocqCustomNodeData

gocq 风格的自定义消息结点数据

name: str

uin: str

content: list[MsgSegment]

seq: NotRequired[list[MsgSegment]]

class melobot.base.typing.ReferNodeData

onebot 标准的引用消息结点数据

id: str

class melobot.base.typing.MsgNode

onebot 标准的转发消息结点

type: Literal['node']

data: StdCustomNodeData | GocqCustomNodeData | ReferNodeData

class melobot.base.typing.ParseArgs

解析参数类

__init__(values: list[Any]) → None

实例化一组解析参数，对应一次解析的结果

参数:

values – 参数值的列表（表示无可用参数用空列表，而不是 None 值）

vals: list[Any]

保存的一组解析参数值

class melobot.base.typing.LogicMode

逻辑模式枚举类型

AND = 1

OR = 2

NOT = 3

XOR = 4

class melobot.base.typing.User

用户权限等级枚举

OWNER = 10000

SU = 1000

WHITE = 100

USER = 10

BLACK = -1

class melobot.base.typing.PriorLevel

事件处理优先级枚举

为方便进行优先级设置，有 MIN, MAX, MEAN 三个枚举值

MIN = 0

MAX = 1000

MEAN = 500

class melobot.base.typing.BotLife

bot 实例的生命周期枚举

LOADED = 1

FIRST_CONNECTED = 2

RECONNECTED = 3

BEFORE_CLOSE = 4

BEFORE_STOP = 5

EVENT_BUILT = 6

ACTION_PRESEND = 7

melobot.base.typing.T

泛型 T，无约束

melobot.base.typing.T1

泛型 T1，无约束

melobot.base.typing.T2

泛型 T2，无约束

melobot.base.typing.T3

泛型 T3，无约束

melobot.base.typing.P

ParamSpec 泛型 P，无约束

melobot.base.typing.AsyncCallable

用法：AsyncCallable[P, T]

是该类型的别名：Callable[P, Awaitable[T]]

class melobot.base.typing.Void

“无值”对象，而不是 None 代表的“空值”

使用方法：直接使用这个类即可。

这个对象的类型标注是：VoidType。

melobot.base.typing.VoidType

“无值”对象的类型标注，定义如下：

VoidType: TypeAlias = Type[Void]