iamai.adapter.kook package#

Subpackages#

Submodules#

Module contents#

Kook 协议适配器。

本适配器适配了 Kook 协议。 协议详情请参考: [Kook 开发者平台](https://developer.kookapp.cn/) 。

class iamai.adapter.kook.KookAdapter(bot: Bot)[source]#

Bases: WebSocketAdapter[KookEvent, Config]

Kook 协议适配器。

class Config(*, adapter_type: Literal['ws', 'wb'] = 'ws', reconnect_interval: int = 3, api_timeout: int = 1000, access_token: str = '', compress: Literal[0, 1] = 0, show_raw: bool = False, report_self_message: bool = False, **extra_data: Any)#

Bases: ConfigModel

Kook 配置类，将在适配器被加载时被混入到机器人主配置中。

adapter_type#

适配器类型，需要和协议端配置相同。

Type:

Literal[‘ws’, ‘wb’]

reconnect_interval#

重连等待时间。

Type:

int

api_timeout#

进行 API 调用时等待返回响应的超时时间。

Type:

int

access_token#

鉴权。

Type:

str

compress#

是否启用压缩，默认为 0，不启用。

Type:

Literal[0, 1]

show_raw#

是否显示原始数据，默认为 False，不显示。

Type:

bool

access_token: str#
adapter_type: Literal['ws', 'wb']#
api_timeout: int#
compress: Literal[0, 1]#
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'access_token': FieldInfo(annotation=str, required=False, default=''), 'adapter_type': FieldInfo(annotation=Literal['ws', 'wb'], required=False, default='ws'), 'api_timeout': FieldInfo(annotation=int, required=False, default=1000), 'compress': FieldInfo(annotation=Literal[0, 1], required=False, default=0), 'reconnect_interval': FieldInfo(annotation=int, required=False, default=3), 'report_self_message': FieldInfo(annotation=bool, required=False, default=False), 'show_raw': FieldInfo(annotation=bool, required=False, default=False)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

reconnect_interval: int#
report_self_message: bool#
show_raw: bool#
api_root = 'https://www.kookapp.cn/api/v3/'#
async call_api(api: str, **data: dict) → Any[source]#
async handle_kook_event(data: Dict[str, Any])[source]#

处理 kook 事件。

Parameters:

msg – 接收到的信息。

async handle_websocket_msg(msg: WSMessage)[source]#

处理 WebSocket 消息。

name: str = 'kook'#
async send(message_: T_KookMSG, message_type: Literal['GROUP', 'PERSON'], id_: int) → Dict[str, Any][source]#

发送消息，调用 message/create 或 direct-message/create API 发送消息。

Parameters:

  • message – 消息内容，可以是 str, Mapping, Iterable[Mapping], ‘KookMessageSegment’, ‘KookMessage’。 将使用 KookMessage 进行封装。

  • message_type – 消息类型。应该是 GROUP 或者 PERSON。

  • id – 发送对象的 ID ，Kook 用户码或者Kook频道码。

Returns:

API 响应。

Raises:

  • TypeError – message_type 不是 ‘PERSON’ 或 ‘GROUP’。

  • ... – 同 call_api() 方法。

async start_heartbeat(session: str) → None[source]#

每30s一次心跳 :return:

async startup()[source]#

初始化适配器。

async websocket_connect()[source]#

创建正向 WebSocket 连接。