melobot.bot

bot 层组件

class melobot.bot.Bot[源代码]

基类：HookMixin[BotLifeSpan]

bot 类

变量:

  • name (str) -- bot 对象的名称

  • logger (GenericLogger) -- bot 对象使用的日志器

__init__(name: str = 'melobot', /, logger: GenericLogger | None = None) → None[源代码]

初始化 bot

参数:
返回类型:

None

add_input(src: AbstractInSource) → Bot[源代码]

绑定输入源

参数:

src (AbstractInSource) -- 输入源

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

add_output(src: AbstractOutSource) → Bot[源代码]

绑定输出源

参数:

src (AbstractOutSource) -- 输出源

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

add_io(src: AbstractIOSource) → Bot[源代码]

绑定输入输出源

参数:

src (AbstractIOSource) -- 输入输出源

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

add_adapter(adapter: Adapter) → Bot[源代码]

绑定适配器

参数:

adapter (Adapter) -- 适配器对象

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

add_protocol(pstack: ProtocolStack | type[ProtocolStack]) → Bot[源代码]

绑定完整的协议栈，这包含了一组协同工作的输入源、输出源和适配器

参数:

pstack (ProtocolStack | type[ProtocolStack]) -- 协议栈对象或协议栈类

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

load_plugin(plugin: ModuleType | str | PathLike[str] | PluginPlanner, load_depth: int = 1) → Bot[源代码]

加载插件，非线程安全

参数:

  • plugin (ModuleType | str | PathLike[str] | PluginPlanner) -- 可以被加载为插件的对象（插件目录对应的模块，插件的目录路径，可直接 import 包名称，插件管理器对象）

  • load_depth (int) -- 插件加载时的相对引用深度，默认值 1 只支持向上引用到插件目录一级。 增加为 2 可以引用到插件目录的父目录一级，依此类推。 此参数只在 plugin 参数为插件的目录路径时有效。

返回:

bot 对象，因此支持链式调用

返回类型:

Bot

load_plugins(plugins: Iterable[ModuleType | str | PathLike[str] | PluginPlanner], load_depth: int = 1) → None[源代码]

与 load_plugin() 行为类似，但是参数变为可迭代对象

此方法同样不是线程安全的

参数:
返回类型:

None

load_plugins_dir(pdir: str | PathLike[str], load_depth: int = 1) → None[源代码]

与 load_plugin() 行为类似，但是参数变为插件目录的父目录，本方法可以加载单个目录下的多个插件

此方法同样不是线程安全的

参数:
返回类型:

None

load_plugins_dirs(pdirs: Iterable[str | PathLike[str]], load_depth: int = 1) → None[源代码]

与 load_plugins_dir() 行为类似，但是参数变为可迭代对象，每个元素为包含插件目录的父目录。 本方法可以加载多个目录下的多个插件

此方法同样不是线程安全的

参数:
返回类型:

None

run(debug: bool = False, strict_log: bool = False, use_exc_handler: bool = True, loop_factory: Callable[[], AbstractEventLoop] | None = None, eager_task: bool = True) → None[源代码]

运行 bot 的阻塞方法（将在内部创建事件循环），这适用于只运行单一 bot 的情况

参数:

  • debug (bool) -- 是否启用 asyncio 的调试模式，但是这不会更改 asyncio 日志器的日志等级

  • strict_log (bool) -- 是否启用严格日志，启用后事件循环中的未捕获异常都会输出错误日志，否则未捕获异常将只输出调试日志

  • use_exc_handler (bool) -- 是否使用内置的事件循环异常处理器来处理未捕获异常，为 False 时 strict_log 参数无效。 设置为 True 时，未捕获异常会被记录到日志中。并在运行结束后重置为调用前的异常处理器

  • loop_factory (Callable[[], AbstractEventLoop] | None) -- 使用此参数提供事件循环工厂。 Python 3.14 版本开始弃用事件循环策略，替代方案为事件循环工厂。 当然在 < 3.16 的版本，你依然可以不提供此参数，继续使用事件循环策略设置事件循环类型

  • eager_task (bool) -- 是否启用主动任务模式（Python 3.12+ 可用），启用后任务会在创建时立即执行协程直到第一个 await 表达式

返回类型:

None

async run_async(strict_log: bool = False, use_exc_handler: bool = True, reserved_tasks: set[Task] | None = None, pre_loop_sig_handlers: list[tuple[int, Callable[[...], object]]] | None = None, shutdown_asyncgens: bool = False, shutdown_default_executor: bool = False) → None[源代码]

运行 bot 的异步方法（将在已有的事件循环中运行），这适用于只运行单一 bot 的情况

我们仅推荐在测试环境中使用此方法，因为部分测试框架的异步测试会先运行一个事件循环（如 pytest-asyncio）。 在复杂的生产环境中，无法 100% 保证运行前后事件循环的状态不被破坏

参数:

  • strict_log (bool) -- 是否启用严格日志，启用后事件循环中的未捕获异常都会输出错误日志，否则未捕获异常将只输出调试日志

  • use_exc_handler (bool) -- 是否使用内置的事件循环异常处理器来处理未捕获异常，为 False 时 strict_log 参数无效。 设置为 True 时，未捕获异常会被记录到日志中。此方法运行结束后重置为调用前的异常处理器

  • reserved_tasks (set[Task] | None) -- bot 运行结束时，会取消事件循环中除此方法调用栈上的其他任务，此参数提供不需要取消的任务的集合。 这一般适用于在调用此方法时，已存在其他非本栈上的异步任务的情况

  • pre_loop_sig_handlers (list[tuple[int, Callable[[...], object]]] | None) -- bot 运行时会尝试覆盖原有的信号处理函数，对于 Windows 平台，在运行结束后自动重置为调用前的信号处理函数。 但对于其他平台，内部使用 add_signal_handler 方法添加信号处理函数， 无法获取到先前的信号处理函数。如果需要重置，请使用此参数提供调用前的信号处理函数列表

  • shutdown_asyncgens (bool) -- 运行结束后是否关闭事件循环中所有异步生成器

  • shutdown_default_executor (bool) -- 运行结束后是否关闭事件循环的默认执行器

返回类型:

None

classmethod start(*bots: Bot, debug: bool = False, strict_log: bool = False) → None[源代码]

安全地同时运行多个 bot 的阻塞方法

参数:

  • bots (Bot) -- 要运行的 bot 对象

  • debug (bool) -- 参见 run() 同名参数

  • strict_log (bool) -- 参见 run() 同名参数

返回类型:

None

async close() → None[源代码]

停止并关闭当前 bot

返回类型:

None

is_restartable() → bool[源代码]

判断当前 bot 是否可以重启

返回:

是否可以重启

返回类型:

bool

async restart() → NoReturn[源代码]

重启当前 bot，需要通过模块运行模式启动 bot 主脚本：

python3 -m melobot run xxx.py

另外请注意，重启功能只在启动了一个 bot 时生效，多个 bot 同时运行时无法重启

返回类型:

NoReturn

get_adapter(type: LiteralString | type[Adapter] | None = None, filter: Callable[[Adapter], bool] | None = None) → Adapter | None[源代码]

获取 bot 所绑定的适配器

参数:

  • type (LiteralString | type[Adapter] | None) -- 适配器的类型（可传入协议字符串或适配器类型），为空时才使用 filter 参数

  • filter (Callable[[Adapter], bool] | None) -- 过滤函数，返回 True 则表明需要该适配器。为空则不使用

返回:

适配器或空

返回类型:

Adapter | None

get_adapters(filter: Callable[[Adapter], bool] | None = None) → set[Adapter][源代码]

获取一组适配器

参数:

filter (Callable[[Adapter], bool] | None) -- 参见 get_adapter() 同名参数。但此处为空时直接获取所有适配器

返回:

适配器的集合

返回类型:

set[Adapter]

get_plugins() → list[str][源代码]

获取所有绑定的插件的名称

返回:

所绑定的插件的名称列表

返回类型:

list[str]

get_share(plugin: str, share: str) → SyncShare | AsyncShare[源代码]

获取绑定的插件中的共享对象

参数:

  • plugin (str) -- 插件名称

  • share (str) -- 共享对象的标识

返回:

共享对象

返回类型:

SyncShare | AsyncShare

add_flows(*flows: Flow) → None[源代码]

添加处理流，非线程安全

参数:

flows (Flow) -- 流对象

返回类型:

None

async wait_finish(event: Event) → None[源代码]

等待事件被所有处理流处理完成

参数:

event (Event) -- 事件对象

返回类型:

None

property on_loaded: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.LOADED 阶段 hook 的装饰器

property on_reloaded: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.RESTARTED 阶段 hook 的装饰器

property on_restarted: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.RESTARTED 阶段 hook 的装饰器

get_hook_evoke_time(hook_type: HookEnumT) → float

获取指定 hook 最后一次触发的时间戳（秒）

若从未触发过，返回 -1

参数:

hook_type (HookEnumT) -- hook 类型

返回:

触发时间

返回类型:

float

on(*periods: HookEnumT) → Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

注册一个 hook

参数:

periods (HookEnumT) -- 要绑定的 hook 类型

返回:

装饰器

返回类型:

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

property on_started: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.STARTED 阶段 hook 的装饰器

property on_close: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.CLOSE 阶段 hook 的装饰器

property on_stopped: Callable[[SyncOrAsyncCallable[P, None]], AsyncCallable[P, None]]

给 bot 注册 BotLifeSpan.STOPPED 阶段 hook 的装饰器

class melobot.bot.BotLifeSpan[源代码]

基类：Enum

bot 生命周期阶段的枚举

LOADED = 'l'

所有插件加载之后

STARTED = 'sta'

bot 启动完成之后（此时所有适配器和源已经开始工作）

RESTARTED = 'r'

bot 重新启动完成之后

CLOSE = 'c'

bot 手动停止即将发生前（仅限 bot.close() 发起的，信号触发不会调用此类型，出现异常可能不会调用此类型）

STOPPED = 'sto'

bot 停止完成后（包含任何情况导致的停止）

melobot.bot.get_bot() → Bot[源代码]

获得当前上下文中的 bot 对象

返回:

bot 对象

返回类型:

Bot

上下文动态变量

melobot.bot.bot

当前上下文中的 bot 实例，类型为 Bot