melobot.utils¶

基础工具¶

melobot.utils.get_obj_name(obj: Any, otype: Literal['callable', 'class', 'object'] | str = 'object', default: str = '<anonymous %s>') → str[源代码]¶

获取一个对象的限定名称或名称，这适用于一些类型较宽的参数。

无法获取有效名称时，产生一个 default % otype 字符串

例如某处接受一个 Callable 类型的参数，对于一般函数来说，使用 __qualname__ 或 __name__ 可获得名称，但某些可调用对象这些值可能为 None 或不存在。使用此方法可保证一定返回字符串

def _(a: Callable) -> None:
    valid_str: str = get_obj_name(a, otype="callable")

def _(a: type) -> None:
    valid_str: str = get_obj_name(a, otype="class")

def _(a: Any) -> None:
    valid_str: str = get_obj_name(a, otype="type of a, only for str concat")

参数:

obj (Any) -- 对象
otype (Literal['callable', 'class', 'object'] | str) -- 预期的对象类型
default (str) -- 无法获取任何有效名称时的默认字符串

返回:

对象名称或默认字符串

返回类型:

str

melobot.utils.singleton(cls: type[T] | Callable[[P], T]) → type[T] | Callable[[P], T][源代码]¶

单例装饰器

参数:: cls (type[T] | Callable[[~P], T]) -- 需要被单例化的可调用对象
返回:: 需要被单例化的可调用对象
返回类型:: type[T] | Callable[[~P], T]

class melobot.utils.RWContext[源代码]¶

基类：object

异步读写上下文

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

使用方法：

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

__init__(read_limit: int | None = None) → None[源代码]¶

初始化异步读写上下文

参数:: read_limit (int | None) -- 读取的数量限制，为空则不限制
返回类型:: None

read() → AsyncGenerator[None, None][源代码]¶

上下文管理器，展开一个关于该对象的安全异步读上下文

返回类型:: AsyncGenerator[None, None]

write() → AsyncGenerator[None, None][源代码]¶

上下文管理器，展开一个关于该对象的安全异步写上下文

返回类型:: AsyncGenerator[None, None]

melobot.utils.get_id() → str[源代码]¶

从 melobot 内部 id 获取器获得一个 id 值，不保证线程安全。算法使用雪花算法

返回:: id 值
返回类型:: str

melobot.utils.to_async(obj: SyncOrAsyncCallable[P, T] | Awaitable[T]) → Callable[[P], Coroutine[Any, Any, T]][源代码]¶

异步包装函数

将一个可调用对象或可等待对象装饰为异步函数

参数:: obj (SyncOrAsyncCallable[~P, T] | Awaitable[T]) -- 需要转换的可调用对象或可等待对象
返回:: 异步函数
返回类型:: Callable[[~P], Coroutine[Any, Any, T]]

melobot.utils.to_coro(obj: SyncOrAsyncCallable[P, T] | Awaitable[T], *args: Any, **kwargs: Any) → Coroutine[Any, Any, T][源代码]¶

协程包装函数

将一个可调用对象或可等待对象装饰为异步函数，并返回对应的协程

参数:

obj (SyncOrAsyncCallable[~P, T] | Awaitable[T]) -- 需要包装的可调用对象或可等待对象
args (Any) -- 需要使用的位置参数
kwargs (Any) -- 需要使用的关键字参数

返回:

协程

返回类型:

Coroutine[Any, Any, T]

melobot.utils.to_sync(obj: SyncOrAsyncCallable[P, Any] | Awaitable[Any]) → Callable[[P], None][源代码]¶

同步包装函数

将一个可调用对象或可等待对象装饰为同步函数，但同步函数无法异步等待，包装后无法获取返回值

因此仅用于接口兼容，如果提供了异步可调用对象，需要自行捕获内部可能的异常

参数:: obj (SyncOrAsyncCallable[~P, Any] | Awaitable[Any]) -- 需要转换的可调用对象或可等待对象
返回:: 同步函数
返回类型:: Callable[[~P], None]

melobot.utils.if_not(condition: SyncOrAsyncCallable[(), U] | U, reject: SyncOrAsyncCallable[(), None], give_up: bool = False, accept: SyncOrAsyncCallable[U, None] | None = None) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | None]][源代码]¶

条件判断装饰器

参数:

condition (SyncOrAsyncCallable[(), U] | U) -- 用于判断的条件（如果是可调用对象，则先求值再转为 bool 值）
reject (SyncOrAsyncCallable[(), None]) -- 当条件为 False 时，执行的回调
give_up (bool) -- 在条件为 False 时，是否放弃执行被装饰函数
accept (SyncOrAsyncCallable[(~U,), None] | None) -- 当条件为 True 时，执行的回调

返回类型:

Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | None]]

melobot.utils.unfold_ctx(getter: SyncOrAsyncCallable[(), ContextManager | AsyncContextManager]) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T]][源代码]¶

上下文装饰器

展开一个上下文，供被装饰函数使用。但注意此装饰器不支持获取上下文管理器 yield 的值

参数:: getter (SyncOrAsyncCallable[(), ContextManager | AsyncContextManager]) -- 上下文管理器或上下文管理器获取方法
返回类型:: Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T]]

melobot.utils.lock(callback: SyncOrAsyncCallable[(), U] | None = None) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | U]][源代码]¶

锁装饰器

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

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

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

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

参数:: callback (SyncOrAsyncCallable[(), U] | None) -- 获取锁冲突时的回调
返回类型:: Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | U]]

melobot.utils.cooldown(busy_callback: ~melobot.typ.base.SyncOrAsyncCallable[(), ~melobot.typ.base.U] | None = None, cd_callback: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'float'>, ), ~melobot.typ.base.V] | None = None, interval: float = 5) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | U | V]][源代码]¶

冷却装饰器

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

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

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

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

cd_callback 不为空：使用 cd_callback 生成回调并执行。
cd_callback 为空，被装饰函数持续等待，直至冷却结束再执行。

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

参数:

busy_callback (SyncOrAsyncCallable[(), U] | None) -- 已运行时的回调
cd_callback (SyncOrAsyncCallable[(<class 'float'>,), V] | None) -- 冷却时间未结束的回调
interval (float) -- 冷却时间

返回类型:

Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | U | V]]

melobot.utils.semaphore(callback: SyncOrAsyncCallable[(), U] | None = None, value: int = -1) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | U]][源代码]¶

信号量装饰器

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

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

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

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

参数:

callback (SyncOrAsyncCallable[(), U] | None) -- 信号量无法立即获取的回调
value (int) -- 信号量阈值

返回类型:

Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | U]]

melobot.utils.timelimit(callback: SyncOrAsyncCallable[(), U] | None = None, timeout: float = 5) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | U]][源代码]¶

时间限制装饰器

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

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

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

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

参数:

callback (SyncOrAsyncCallable[(), U] | None) -- 超时时的回调
timeout (float) -- 超时时间

返回类型:

Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | U]]

melobot.utils.speedlimit(callback: SyncOrAsyncCallable[(), U] | None = None, limit: int = 60, duration: int = 60) → Callable[[SyncOrAsyncCallable[P, T]], AsyncCallable[P, T | U]][源代码]¶

流量/速率限制装饰器（使用固定窗口算法）

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

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

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

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

参数:

callback (SyncOrAsyncCallable[(), U] | None) -- 超出速率限制时的回调
limit (int) -- duration 秒内允许调用多少次
duration (int) -- 时长区间

返回类型:

Callable[[SyncOrAsyncCallable[~P, T]], AsyncCallable[~P, T | U]]

melobot.utils.call_later(callback: Callable[[], None], delay: float) → TimerHandle[源代码]¶

同步函数延迟调度

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

参数:

callback (Callable[[], None]) -- 同步函数
delay (float) -- 多长时间后调度

返回:

asyncio.TimerHandle 对象

返回类型:

TimerHandle

melobot.utils.call_at(callback: Callable[[], None], timestamp: float) → TimerHandle[源代码]¶

同步函数指定时间调度

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

参数:

callback (Callable[[], None]) -- 同步函数
timestamp (float) -- 在什么时刻调度

返回:

asyncio.TimerHandle 对象

返回类型:

TimerHandle

melobot.utils.async_later(callback: Coroutine[Any, Any, T], delay: float) → Task[T][源代码]¶

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

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

返回一个 asyncio.Task 对象，等待 asyncio.Task 即是等待 callback 的返回值。

参数:

callback (Coroutine[Any, Any, T]) -- 协程（可有返回值）
delay (float) -- 多长时间后调度

返回:

asyncio.Task 对象

返回类型:

Task[T]

melobot.utils.async_at(callback: Coroutine[Any, Any, T], timestamp: float) → Task[T][源代码]¶

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

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

返回一个 asyncio.Task 对象，等待 asyncio.Task 即是等待 callback 的返回值。

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

参数:

callback (Coroutine[Any, Any, T]) -- 协程（可有返回值）
timestamp (float) -- 在什么时刻调度

返回:

asyncio.Task 对象

返回类型:

Task[T]

melobot.utils.async_interval(callback: Callable[[], Coroutine[Any, Any, None]], interval: float) → Task[None][源代码]¶

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

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

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

参数:

callback (Callable[[], Coroutine[Any, Any, None]]) -- 异步函数
interval (float) -- 调度的间隔

返回:

asyncio.Task 对象

返回类型:

Task[None]

检查/验证¶

class melobot.utils.check.Checker[源代码]¶

基类：Generic[EventT], BetterABC

检查器基类

abstract async check(event: EventT) → bool[源代码]¶

检查器检查方法

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

参数:: event (EventT) -- 给定的事件
返回:: 检查是否通过
返回类型:: bool

static new(func: Callable[[EventT], bool]) → Checker[EventT][源代码]¶

从可调用对象创建检查器

参数:: func (Callable[[EventT], bool]) -- 可调用对象
返回:: 检查器对象
返回类型:: Checker[EventT]

class melobot.utils.check.WrappedChecker[源代码]¶

基类：Checker[EventT]

合并检查器

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

set_fail_cb(fail_cb: SyncOrAsyncCallable[(), None] | None) → None[源代码]¶

参数:: fail_cb (SyncOrAsyncCallable[(), None] | None)
返回类型:: None

基础检查/验证工具¶

melobot.utils.check.checker_join(*checkers: Checker | None | Callable[[Any], bool]) → Checker[源代码]¶

合并检查器

相比于使用 | & ^ ~ 运算符，此函数可以接受一个检查器序列，并返回一个合并检查器。检查器序列可以为检查器对象，检查函数或空值

返回:: 合并后的检查器对象
参数:: checkers (Checker | None | Callable[[Any], bool])
返回类型:: Checker

匹配¶

class melobot.utils.match.Matcher[源代码]¶

基类：BetterABC

匹配器基类

abstract async match(text: str) → bool[源代码]¶

匹配器匹配方法

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

参数:: text (str) -- 消息事件的文本内容
返回:: 是否匹配
返回类型:: bool

class melobot.utils.match.WrappedMatcher[源代码]¶

基类：Matcher

合并匹配器

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

基础匹配工具¶

class melobot.utils.match.StartMatcher[源代码]¶

基类：Matcher

字符串起始匹配器

__init__(target: str | list[str], mode: LogicMode = LogicMode.OR) → None[源代码]¶

初始化一个字符串起始匹配器

target 为字符串时，只进行一次起始匹配，即判断是否匹配成功。 target 为字符串列表时，所有字符串都进行起始匹配，再将所有结果使用给定 mode 计算是否匹配成功。

参数:

target (str | list[str]) -- 匹配目标
mode (LogicMode) -- 匹配模式

返回类型:

None

class melobot.utils.match.ContainMatcher[源代码]¶

基类：Matcher

字符串包含匹配器

__init__(target: str | list[str], mode: LogicMode = LogicMode.OR) → None[源代码]¶

初始化一个字符串包含匹配器

target 为字符串时，只进行一次包含匹配，即判断是否匹配成功。 target 为字符串列表时，所有字符串都进行包含匹配，再将所有结果使用给定 mode 计算是否匹配成功。

参数:

target (str | list[str]) -- 匹配目标
mode (LogicMode) -- 匹配模式

返回类型:

None

class melobot.utils.match.EndMatcher[源代码]¶

基类：Matcher

字符串结尾匹配器

__init__(target: str | list[str], mode: LogicMode = LogicMode.OR) → None[源代码]¶

初始化一个字符串结尾匹配器

target 为字符串时，只进行一次结尾匹配，即判断是否匹配成功。 target 为字符串列表时，所有字符串都进行结尾匹配，再将所有结果使用给定 mode 计算是否匹配成功。

参数:

target (str | list[str]) -- 匹配目标
mode (LogicMode) -- 匹配模式

返回类型:

None

class melobot.utils.match.FullMatcher[源代码]¶

基类：Matcher

字符串全匹配器

__init__(target: str | list[str], mode: LogicMode = LogicMode.OR) → None[源代码]¶

初始化一个字符串全匹配器

target 为字符串时，只进行一次全匹配，即判断是否匹配成功。 target 为字符串列表时，所有字符串都进行全匹配，再将所有结果使用给定 mode 计算是否匹配成功。

参数:

target (str | list[str]) -- 匹配目标
mode (LogicMode) -- 匹配模式

返回类型:

None

class melobot.utils.match.RegexMatcher[源代码]¶

基类：Matcher

字符串正则匹配器

__init__(regex_pattern: str, regex_flags: Any = 0) → None[源代码]¶

初始化一个字符串正则匹配器

参数:

regex_pattern (str) -- 正则 pattern
regex_flags (Any) -- 正则 flag，默认不使用

返回类型:

None

解析¶

class melobot.utils.parse.Parser[源代码]¶

基类：BetterABC

解析器基类

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

abstract async parse(text: str) → AbstractParseArgs | None[源代码]¶

解析方法

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

参数:: text (str) -- 消息文本内容
返回:: 解析结果，为空代表没有有效的解析参数
返回类型:: AbstractParseArgs | None

class melobot.utils.parse.AbstractParseArgs[源代码]¶

基类：object

解析参数抽象类

子类需要把以下属性按 abstractattr() 的要求实现

vals: Any¶

解析值

基础解析工具¶

class melobot.utils.parse.CmdParser[源代码]¶

基类：Parser

命令解析器

通过解析命令名和命令参数的形式，解析字符串。

初始化一个命令解析器

注意

命令起始符和命令间隔符不允许包含：引号，各种括号，反斜杠，数字，英文，控制字符及各类空白字符。
命令起始符不能是命令间隔符的子序列，反之亦然。

参数:

cmd_start (str | list[str]) -- 命令起始符（可以是字符串或字符串列表）
cmd_sep (str | list[str]) -- 命令间隔符（可以是字符串或字符串列表）
targets (str | list[str]) -- 匹配的命令名
formatters -- 格式化器列表（列表可以包含空值，即此位置的参数无格式化）
tag (str | None) -- 标签，此标签将被填充给本解析器产生的 CmdArgs 对象的 tag 属性
fmtters (list[CmdArgFormatter | None] | None)

返回类型:

None

class melobot.utils.parse.CmdArgs[源代码]¶

基类：AbstractParseArgs

命令参数对象

name: str¶

tag: str | None¶

class melobot.utils.parse.CmdParserFactory[源代码]¶

基类：object

命令解析器的工厂

预先存储命令起始符和命令间隔符，指定匹配的命令名后返回一个命令解析器。

__init__(cmd_start: str | list[str], cmd_sep: str | list[str]) → None[源代码]¶

初始化一个命令解析器的工厂

注意

命令起始符和命令间隔符不允许包含：引号，各种括号，反斜杠，数字，英文，控制字符及各类空白字符。
命令起始符不能是命令间隔符的子序列，反之亦然。

参数:

cmd_start (str | list[str]) -- 命令起始符（可以是字符串或字符串列表）
cmd_sep (str | list[str]) -- 命令间隔符（可以是字符串或字符串列表）

返回类型:

None

get(targets: str | list[str], formatters: list[CmdArgFormatter | None] | None = None, tag: str | None = None) → CmdParser[源代码]¶

生成匹配指定命令名的命令解析器

参数:

targets (str | list[str]) -- 匹配的命令名
formatters (list[CmdArgFormatter | None] | None) -- 格式化器列表（列表可以包含空值，即此位置的参数无格式化选项）
tag (str | None) -- 标签，此标签将被填充给解析器产生的 CmdArgs 对象的 tag 属性

返回类型:

CmdParser

class melobot.utils.parse.CmdArgFormatter[源代码]¶

基类：object

命令参数格式化器

用于格式化命令解析器解析出的命令参数。搭配命令解析器 CmdParser 使用

__init__(convert: ~typing.Callable[[str], ~typing_extensions.Any] | None = None, validate: ~typing.Callable[[~typing_extensions.Any], bool] | None = None, src_desc: str | None = None, src_expect: str | None = None, default: ~typing_extensions.Any = VoidType.VOID, default_replace_flag: str | None = None, convert_fail: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None = None, validate_fail: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None = None, arg_lack: ~melobot.typ.base.SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None = None) → None[源代码]¶

初始化一个命令参数格式化器

参数:

convert (Callable[[str], Any] | None) -- 类型转换方法，为空则不进行类型转换
validate (Callable[[Any], bool] | None) -- 值验证方法，为空则不对值进行验证
src_desc (str | None) -- 命令参数值的功能描述
src_expect (str | None) -- 命令参数值的值期待描述
default (Any) -- 命令参数值的默认值（默认值 Void 表示无值，而不是 None 表达的空值）
default_replace_flag (str | None) -- 命令参数使用默认值的标志
convert_fail (SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None) -- 类型转换失败的回调，为空则使用默认回调
validate_fail (SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None) -- 值验证失败的回调，为空则使用默认回调
arg_lack (SyncOrAsyncCallable[(<class 'melobot.utils.parse.cmd.CmdArgFormatInfo'>,), None] | None) -- 参数缺失的回调，为空则执行默认规则

返回类型:

None

async format(group_id: str, args: CmdArgs, idx: int) → bool[源代码]¶

参数:

group_id (str)
args (CmdArgs)
idx (int)

返回类型:

bool

class melobot.utils.parse.CmdArgFormatInfo[源代码]¶

基类：object

命令参数格式化信息对象

用于在命令参数格式化异常时传递信息。

name: str¶: 命令参数所属命令的命令名

src: str | VoidType¶: 命令参数格式化前的原值，参数缺失时是 VoidType.VOID

src_desc: str | None¶: 命令参数值的功能描述

src_expect: str | None¶: 命令参数值的值期待描述

idx: int¶: 命令参数值的顺序（从 0 开始索引）

exc: Exception¶: 命令参数格式化异常时的异常对象

exc_tb: TracebackType | None¶: 命令参数格式化异常时的调用栈信息