iamai.adapter.gensokyo package#

Submodules#

Module contents#

gensokyo *ob11 协议适配器。

本适配器适配了 gensokyo obv11 协议。 协议详情请参考：[OneBot](https://github.com/howmanybots/onebot/blob/master/README.md)。

class iamai.adapter.gensokyo.GSKAdapter(bot: Bot)[source]#

Bases: WebSocketAdapter[GSKEvent, Config]

GSK 协议适配器。

class Config(*, adapter_type: Literal['ws', 'reverse-ws', 'ws-reverse'] = 'reverse-ws', host: str = '127.0.0.1', port: int = 8080, url: str = '/gsk/ws', reconnect_interval: int = 3, api_timeout: int = 1000, app_id: str = '', app_secret: str = '', token: str = '', access_token: str = '', **extra_data: Any)#

Bases: ConfigModel

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

adapter_type#

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

Type:

Literal[‘ws’, ‘reverse-ws’, ‘ws-reverse’]

host#

本机域名。

Type:

str

port#

监听的端口。

Type:

int

url#

WebSocket 路径，需和协议端配置相同。

Type:

str

reconnect_interval#

重连等待时间。

Type:

int

api_timeout#

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

Type:

int

access_token#

鉴权。

Type:

str

access_token: str#
adapter_type: Literal['ws', 'reverse-ws', 'ws-reverse']#
api_timeout: int#
app_id: str#
app_secret: str#
host: str#
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', 'reverse-ws', 'ws-reverse'], required=False, default='reverse-ws'), 'api_timeout': FieldInfo(annotation=int, required=False, default=1000), 'app_id': FieldInfo(annotation=str, required=False, default=''), 'app_secret': FieldInfo(annotation=str, required=False, default=''), 'host': FieldInfo(annotation=str, required=False, default='127.0.0.1'), 'port': FieldInfo(annotation=int, required=False, default=8080), 'reconnect_interval': FieldInfo(annotation=int, required=False, default=3), 'token': FieldInfo(annotation=str, required=False, default=''), 'url': FieldInfo(annotation=str, required=False, default='/gsk/ws')}#

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

This replaces Model.__fields__ from Pydantic V1.

port: int#
reconnect_interval: int#
token: str#
url: str#
classmethod add_event_model(event_model: Type[GSKEvent]) → None[source]#

添加自定义事件模型，事件模型类必须继承于 GSKEvent。

Parameters:

event_model – 事件模型类。

async call_api(api: str, **params: Any) → Any[source]#

调用 GSK API，协程会等待直到获得 API 响应。

Parameters:

  • api – API 名称。

  • **params – API 参数。

Returns:

API 响应中的 data 字段。

Raises:
event_models: ClassVar[Dict[Tuple[str | None, str | None, str | None], Type[GSKEvent]]] = {('message', 'group', None): <class 'iamai.adapter.gensokyo.event.GroupMessageEvent'>, ('message', 'private', None): <class 'iamai.adapter.gensokyo.event.PrivateMessageEvent'>, ('message', None, None): <class 'iamai.adapter.gensokyo.event.MessageEvent'>, ('meta_event', 'heartbeat', None): <class 'iamai.adapter.gensokyo.event.HeartbeatMetaEvent'>, ('meta_event', 'lifecycle', None): <class 'iamai.adapter.gensokyo.event.LifecycleMetaEvent'>, ('meta_event', None, None): <class 'iamai.adapter.gensokyo.event.MetaEvent'>, ('notice', 'friend_add', None): <class 'iamai.adapter.gensokyo.event.FriendAddNoticeEvent'>, ('notice', 'friend_recall', None): <class 'iamai.adapter.gensokyo.event.FriendRecallNoticeEvent'>, ('notice', 'group_admin', None): <class 'iamai.adapter.gensokyo.event.GroupAdminNoticeEvent'>, ('notice', 'group_ban', None): <class 'iamai.adapter.gensokyo.event.GroupBanNoticeEvent'>, ('notice', 'group_decrease', None): <class 'iamai.adapter.gensokyo.event.GroupDecreaseNoticeEvent'>, ('notice', 'group_increase', None): <class 'iamai.adapter.gensokyo.event.GroupIncreaseNoticeEvent'>, ('notice', 'group_recall', None): <class 'iamai.adapter.gensokyo.event.GroupRecallNoticeEvent'>, ('notice', 'group_upload', None): <class 'iamai.adapter.gensokyo.event.GroupUploadNoticeEvent'>, ('notice', 'notify', 'honor'): <class 'iamai.adapter.gensokyo.event.GroupHonorNotifyEvent'>, ('notice', 'notify', 'lucky_king'): <class 'iamai.adapter.gensokyo.event.GroupLuckyKingNotifyEvent'>, ('notice', 'notify', 'poke'): <class 'iamai.adapter.gensokyo.event.PokeNotifyEvent'>, ('notice', 'notify', None): <class 'iamai.adapter.gensokyo.event.NotifyEvent'>, ('notice', None, None): <class 'iamai.adapter.gensokyo.event.NoticeEvent'>, ('request', 'friend', None): <class 'iamai.adapter.gensokyo.event.FriendRequestEvent'>, ('request', 'group', None): <class 'iamai.adapter.gensokyo.event.GroupRequestEvent'>, ('request', None, None): <class 'iamai.adapter.gensokyo.event.RequestEvent'>, (None, None, None): <class 'iamai.adapter.gensokyo.event.GSKEvent'>}#
async get_access_token() → str[source]#

异步获取登录凭证

https://bots.qq.com/app/getAppAccessToken 属性 类型 必填 说明 appId string 是 在开放平台管理端上获得。 clientSecret string 是 在开放平台管理端上获得。

classmethod get_event_model(post_type: str | None, detail_type: str | None, sub_type: str | None) → Type[GSKEvent][source]#

根据接收到的消息类型返回对应的事件类。

Parameters:

  • post_type – 请求类型。

  • detail_type – 事件类型。

  • sub_type – 子类型。

Returns:

对应的事件类。

async handle_gsk_event(msg: Dict[str, Any]) → None[source]#

处理 GSK 事件。

Parameters:

msg – 接收到的信息。

async handle_websocket_msg(msg: WSMessage) → None[source]#

处理 WebSocket 消息。

name: str = 'gensokyo'#
async reverse_ws_connection_hook() → None[source]#

反向 WebSocket 连接建立时的钩子函数。

async send(message_: List[GSKMessageSegment] | GSKMessageSegment | str | Mapping[str, Any], message_type: Literal['private', 'group'], id_: int) → Any[source]#

发送消息，调用 send_private_msg 或 send_group_msg API 发送消息。

Parameters:

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

  • message_type – 消息类型。应该是 “private” 或者 “group”。

  • id – 发送对象的 ID， QQ 号码或者群号码。

Returns:

API 响应。

Raises:

  • TypeError – message_type 不是 “private” 或 “group”。

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

async startup() → None[source]#

初始化适配器。

async websocket_connect() → None[source]#

创建正向 WebSocket 连接。