iamai.bot module#

iamai 机器人对象。

iamai 的基础模块,每一个 iamai 机器人即是一个 Bot 实例。

class iamai.bot.Bot(*, config_file: str | None = 'config.toml', config_dict: Dict[str, Any] | None = None, hot_reload: bool = False)[source]#

Bases: object

iamai 机器人对象,定义了机器人的基本方法。

读取并储存配置 Config,加载适配器 Adapter 和插件 Plugin,并进行事件分发。

config#

机器人配置。

Type:

iamai.config.MainConfig

should_exit#

机器人是否应该进入准备退出状态。

Type:

asyncio.locks.Event

adapters#

当前已经加载的适配器的列表。

Type:

List[iamai.adapter.Adapter[Any, Any]]

plugins_priority_dict#

插件优先级字典。

Type:

Dict[int, List[Type[iamai.plugin.Plugin[Any, Any, Any]]]]

plugin_state#

插件状态。

Type:

Dict[str, Any]

global_state#

全局状态。

Type:

Dict[Any, Any]

adapter_run_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器运行时的函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

adapter_shutdown_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器关闭时的函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

adapter_startup_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器初始化时的函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

adapters: List[Adapter[Any, Any]]#
bot_exit_hook(func: Callable[[Bot], Awaitable[None]]) Callable[[Bot], Awaitable[None]][source]#

注册一个 Bot 退出时的函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

bot_run_hook(func: Callable[[Bot], Awaitable[None]]) Callable[[Bot], Awaitable[None]][source]#

注册一个 Bot 启动时的函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

config: MainConfig#
error_or_exception(message: str, exception: Exception) None[source]#

根据当前 Bot 的配置输出 error 或者 exception 日志。

Parameters:
  • message – 消息。

  • exception – 异常。

event_postprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) Callable[[Event[Any]], Awaitable[None]][source]#

注册一个事件后处理函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

event_preprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) Callable[[Event[Any]], Awaitable[None]][source]#

注册一个事件预处理函数。

Parameters:

func – 被注册的函数。

Returns:

被注册的函数。

async get(func: Callable[[Event[Any]], bool | Awaitable[bool]] | None = None, *, event_type: None = None, adapter_type: None = None, max_try_times: int | None = None, timeout: int | float | None = None) Event[Any][source]#
async get(func: Callable[[EventT], bool | Awaitable[bool]] | None = None, *, event_type: None = None, adapter_type: Type[Adapter[EventT, Any]], max_try_times: int | None = None, timeout: int | float | None = None) EventT
async get(func: Callable[[EventT], bool | Awaitable[bool]] | None = None, *, event_type: Type[EventT], adapter_type: Type[AdapterT] | None = None, max_try_times: int | None = None, timeout: int | float | None = None) EventT

获取满足指定条件的的事件,协程会等待直到适配器接收到满足条件的事件、超过最大事件数或超时。

Parameters:
  • func – 协程或者函数,函数会被自动包装为协程执行。 要求接受一个事件作为参数,返回布尔值。当协程返回 True 时返回当前事件。 当为 None 时相当于输入对于任何事件均返回真的协程,即返回适配器接收到的下一个事件。

  • event_type – 当指定时,只接受指定类型的事件,先于 func 条件生效。默认为 None

  • adapter_type – 当指定时,只接受指定适配器产生的事件,先于 func 条件生效。默认为 None

  • max_try_times – 最大事件数。

  • timeout – 超时时间。

Returns:

返回满足 func 条件的事件。

Raises:

GetEventTimeout – 超过最大事件数或超时。

get_adapter(adapter: str) Adapter[Any, Any][source]#
get_adapter(adapter: Type[AdapterT]) AdapterT

按照名称或适配器类获取已经加载的适配器。

Parameters:

adapter – 适配器名称或适配器类。

Returns:

获取到的适配器对象。

Raises:

LookupError – 找不到此名称的适配器对象。

get_plugin(name: str) Type[Plugin[Any, Any, Any]][source]#

按照名称获取已经加载的插件类。

Parameters:

name – 插件名称

Returns:

获取到的插件类。

Raises:

LookupError – 找不到此名称的插件类。

global_state: Dict[Any, Any]#
async handle_event(current_event: Event[Any], *, handle_get: bool = True, show_log: bool = True) None[source]#

被适配器对象调用,根据优先级分发事件给所有插件,并处理插件的 stopskip 等信号。

此方法不应该被用户手动调用。

Parameters:
  • current_event – 当前待处理的 Event

  • handle_get – 当前事件是否可以被 get 方法捕获,默认为 True

  • show_log – 是否在日志中显示,默认为 True

load_adapters(*adapters: Type[Adapter[Any, Any]] | str) None[source]#

加载适配器。

Parameters:

*adapters

适配器类或适配器名称,类型可以是 Type[Adapter]str。 如果为 Type[Adapter] 类型时,将作为适配器类进行加载。 如果为 str 类型时,将作为适配器模块名称进行加载,格式和 Python import 语句相同。

例如:path.of.adapter

load_plugins(*plugins: Type[Plugin[Any, Any, Any]] | str | Path) None[source]#

加载插件。

Parameters:

*plugins

插件类、插件模块名称或者插件模块文件路径。 类型可以是 Type[Plugin], strpathlib.Path。 如果为 Type[Plugin] 类型时,将作为插件类进行加载。 如果为 str 类型时,将作为插件模块名称进行加载,格式和 Python import 语句相同。

例如:path.of.plugin

如果为 pathlib.Path 类型时,将作为插件模块文件路径进行加载。

例如:pathlib.Path(“path/of/plugin”)

load_plugins_from_dirs(*dirs: Path) None[source]#

从目录中加载插件,以 _ 开头的模块中的插件不会被导入。路径可以是相对路径或绝对路径。

Parameters:

*dirs – 储存包含插件的模块的模块路径。 例如:pathlib.Path(“path/of/plugins/”)

plugin_state: Dict[str, Any]#
property plugins: List[Type[Plugin[Any, Any, Any]]]#

当前已经加载的插件的列表。

plugins_priority_dict: Dict[int, List[Type[Plugin[Any, Any, Any]]]]#
reload_plugins() None[source]#

手动重新加载所有插件。

restart() None[source]#

退出并重新运行 iamai。

run() None[source]#

运行 iamai,监听并拦截系统退出信号,更新机器人配置。

should_exit: Event#