laoxong/AstrBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: Implement Pydantic data models for the KOOK adapter to enhance data retrieval and message schema validation (#5719)

Browse Source 
* refactor: 给kook适配器添加kook事件数据类

* format: 使用StrEnum替换kook适配器中的(str,enum)
This commit is contained in:
shuiping233
2026-03-17 18:05:58 +08:00
committed by GitHub
parent dcffb5269a
commit f5ba1a026a
23 changed files with 1036 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
astrbot/core/config/default.py
-6
View File
@@ -463,7 +463,6 @@ CONFIG_METADATA_2 = {
"type": "kook",
"enable": False,
"kook_bot_token": "",
"kook_bot_nickname": "",
"kook_reconnect_delay": 1,
"kook_max_reconnect_delay": 60,
"kook_max_retry_delay": 60,
@@ -875,11 +874,6 @@ CONFIG_METADATA_2 = {
"type": "string",
"hint": "必填项。从 KOOK 开发者平台获取的机器人 Token。",
},
"kook_bot_nickname": {
"description": "Bot Nickname",
"type": "string",
"hint": "可选项。若发送者昵称与此值一致，将忽略该消息以避免广播风暴。",
},
"kook_reconnect_delay": {
"description": "重连延迟",
"type": "int",
astrbot/core/platform/sources/kook/kook_adapter.py
+130 -85
View File
@@ -13,11 +13,28 @@ from astrbot.api.platform import (
PlatformMetadata,
register_platform_adapter,
)
from astrbot.core.message.components import File, Record, Video
from astrbot.core.platform.astr_message_event import MessageSesion
from .kook_client import KookClient
from .kook_config import KookConfig
from .kook_event import KookEvent
from .kook_types import (
ContainerModule,
FileModule,
HeaderModule,
ImageGroupModule,
KmarkdownElement,
KookCardMessageContainer,
KookChannelType,
KookMessageEventData,
KookMessageType,
KookModuleType,
PlainTextElement,
SectionModule,
)
KOOK_AT_SELECTOR_REGEX = re.compile(r"\(met\)([^()]+)\(met\)")
@register_platform_adapter(
@@ -57,35 +74,26 @@ class KookPlatformAdapter(Platform):
name="kook", description="KOOK 适配器", id=self.kook_config.id
)
def _should_ignore_event_by_bot_nickname(self, payload: dict) -> bool:
bot_nickname = self.kook_config.bot_nickname.strip()
if not bot_nickname:
return False
def _should_ignore_event_by_bot_nickname(self, author_id: str) -> bool:
return self.client.bot_id == author_id
author = payload.get("extra", {}).get("author", {})
if not isinstance(author, dict):
return False
author_nickname = author.get("nickname") or author.get("username") or ""
if not isinstance(author_nickname, str):
author_nickname = str(author_nickname)
return author_nickname.strip().casefold() == bot_nickname.casefold()
async def _on_received(self, data: dict):
logger.debug(f"KOOK 收到数据: {data}")
if "d" in data and data["s"] == 0:
payload = data["d"]
event_type = payload.get("type")
# 支持type=9（文本）和type=10（卡片）
if event_type in (9, 10):
if self._should_ignore_event_by_bot_nickname(payload):
return
try:
abm = await self.convert_message(payload)
await self.handle_msg(abm)
except Exception as e:
logger.error(f"[KOOK] 消息处理异常: {e}")
async def _on_received(self, event: KookMessageEventData):
logger.debug(
f'[KOOK] 收到来自"{event.channel_type.name}"渠道的消息, 消息类型为: {event.type.name}({event.type.value})'
)
event_type = event.type
if event_type in (KookMessageType.KMARKDOWN, KookMessageType.CARD):
if self._should_ignore_event_by_bot_nickname(event.author_id):
logger.debug("[KOOK] 收到来自机器人自身的消息, 忽略此消息")
return
try:
abm = await self.convert_message(event)
await self.handle_msg(abm)
except Exception as e:
logger.error(f"[KOOK] 消息处理异常: {e}")
elif event_type == KookMessageType.SYSTEM:
logger.debug(f'[KOOK] 消息为系统通知, 通知类型为: "{event.extra.type}"')
logger.debug(f"[KOOK] 原始消息数据: {event.to_json()}")
async def run(self):
"""主运行循环"""
@@ -184,18 +192,26 @@ class KookPlatformAdapter(Platform):
logger.info("[KOOK] 资源清理完成")
def _parse_kmarkdown_text_message(
self, data: dict, self_id: str
self, data: KookMessageEventData, self_id: str
) -> tuple[list, str]:
kmarkdown = data.get("extra", {}).get("kmarkdown", {})
content = data.get("content") or ""
raw_content = kmarkdown.get("raw_content") or content
kmarkdown = data.extra.kmarkdown
content = data.content or ""
if kmarkdown is None:
logger.error(
f'[KOOK] 无法转换"{KookMessageType.KMARKDOWN.name}"消息, 消息中找不到kmarkdown字段'
)
logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
return [], ""
raw_content = kmarkdown.raw_content or content
if not isinstance(content, str):
content = str(content)
if not isinstance(raw_content, str):
raw_content = str(raw_content)
# TODO 后面的pydantic类型替换,以后再来探索吧 :(
mention_name_map: dict[str, str] = {}
mention_part = kmarkdown.get("mention_part", [])
mention_part = kmarkdown.mention_part
if isinstance(mention_part, list):
for item in mention_part:
if not isinstance(item, dict):
@@ -207,7 +223,7 @@ class KookPlatformAdapter(Platform):
components = []
cursor = 0
for match in re.finditer(r"\(met\)([^()]+)\(met\)", content):
for match in KOOK_AT_SELECTOR_REGEX.finditer(content):
if match.start() > cursor:
plain_text = content[cursor : match.start()]
if plain_text:
@@ -254,77 +270,109 @@ class KookPlatformAdapter(Platform):
return components, message_str
def _parse_card_message(self, data: dict) -> tuple[list, str]:
content = data.get("content", "[]")
def _parse_card_message(self, data: KookMessageEventData) -> tuple[list, str]:
content = data.content
if not isinstance(content, str):
content = str(content)
card_list = json.loads(content)
card_list = KookCardMessageContainer.from_dict(json.loads(content))
text_parts: list[str] = []
images: list[str] = []
files: list[tuple[KookModuleType, str, str]] = []
for card in card_list:
if not isinstance(card, dict):
continue
for module in card.get("modules", []):
if not isinstance(module, dict):
continue
for module in card.modules:
match module:
case SectionModule():
if content := self._handle_section_text(module):
text_parts.append(content)
module_type = module.get("type")
if module_type == "section":
section_text = module.get("text", {}).get("content", "")
if section_text:
text_parts.append(str(section_text))
continue
case ContainerModule() | ImageGroupModule():
urls = self._handle_image_group(module)
images.extend(urls)
text_parts.append(" [image]" * len(urls))
if module_type != "container":
continue
case HeaderModule():
text_parts.append(module.text.content)
for element in module.get("elements", []):
if not isinstance(element, dict):
continue
if element.get("type") != "image":
continue
case FileModule():
files.append((module.type, module.title, module.src))
text_parts.append(f" [{module.type.value}]")
image_src = element.get("src")
if not isinstance(image_src, str):
logger.warning(
f'[KOOK] 处理卡片中的图片时发生错误,图片url "{image_src}" 应该为str类型, 而不是 "{type(image_src)}" '
)
continue
if not image_src.startswith(("http://", "https://")):
logger.warning(f"[KOOK] 屏蔽非http图片url: {image_src}")
continue
images.append(image_src)
case _:
logger.debug(f"[KOOK] 跳过或未处理模块: {module.type}")
text = "".join(text_parts)
message = []
if text:
for search in KOOK_AT_SELECTOR_REGEX.finditer(text):
search_text = search.group(1).strip()
if search_text == "all":
message.append(AtAll())
continue
message.append(At(qq=search_text))
text = text.replace(f"(met){search_text}(met)", "")
message.append(Plain(text=text))
for img_url in images:
message.append(Image(file=img_url))
for file in files:
file_type = file[0]
file_name = file[1]
file_url = file[2]
if file_type == KookModuleType.FILE:
message.append(File(name=file_name, file=file_url))
elif file_type == KookModuleType.VIDEO:
message.append(Video(file=file_url))
elif file_type == KookModuleType.AUDIO:
message.append(Record(file=file_url))
else:
logger.warning(f"[KOOK] 跳过未知文件类型: {file_type.name}")
return message, text
async def convert_message(self, data: dict) -> AstrBotMessage:
def _handle_section_text(self, module: SectionModule) -> str:
"""专门处理 Section 里的文本提取"""
if isinstance(module.text, (KmarkdownElement, PlainTextElement)):
return module.text.content or ""
return ""
def _handle_image_group(
self, module: ContainerModule | ImageGroupModule
) -> list[str]:
"""专门处理图片组/容器里的合法 URL 提取"""
valid_urls = []
for el in module.elements:
image_src = el.src
if not el.src.startswith(("http://", "https://")):
logger.warning(f"[KOOK] 屏蔽非http图片url: {image_src}")
continue
valid_urls.append(el.src)
return valid_urls
async def convert_message(self, data: KookMessageEventData) -> AstrBotMessage:
abm = AstrBotMessage()
abm.raw_message = data
abm.raw_message = data.to_dict()
abm.self_id = self.client.bot_id
channel_type = data.get("channel_type")
author_id = data.get("author_id", "unknown")
channel_type = data.channel_type
author_id = data.author_id
# channel_type定义: https://developer.kookapp.cn/doc/event/event-introduction
match channel_type:
case "GROUP":
session_id = data.get("target_id") or "unknown"
case KookChannelType.GROUP:
session_id = data.target_id or "unknown"
abm.type = MessageType.GROUP_MESSAGE
abm.group_id = session_id
abm.session_id = session_id
case "PERSON":
case KookChannelType.PERSON:
abm.type = MessageType.FRIEND_MESSAGE
abm.group_id = ""
abm.session_id = data.get("author_id", "unknown")
case "BROADCAST":
session_id = data.get("target_id") or "unknown"
abm.session_id = data.author_id or "unknown"
case KookChannelType.BROADCAST:
session_id = data.target_id or "unknown"
abm.type = MessageType.OTHER_MESSAGE
abm.group_id = session_id
abm.session_id = session_id
@@ -333,28 +381,25 @@ class KookPlatformAdapter(Platform):
abm.sender = MessageMember(
user_id=author_id,
nickname=data.get("extra", {}).get("author", {}).get("username", ""),
nickname=data.extra.author.username if data.extra.author else "unknown",
)
abm.message_id = data.get("msg_id", "unknown")
abm.message_id = data.msg_id or "unknown"
# 普通文本消息
if data.get("type") == 9:
message, message_str = self._parse_kmarkdown_text_message(
data, str(abm.self_id)
)
if data.type == KookMessageType.KMARKDOWN:
message, message_str = self._parse_kmarkdown_text_message(data, abm.self_id)
abm.message = message
abm.message_str = message_str
# 卡片消息
elif data.get("type") == 10:
elif data.type == KookMessageType.CARD:
try:
abm.message, abm.message_str = self._parse_card_message(data)
except Exception as exp:
logger.error(f"[KOOK] 卡片消息解析失败: {exp}")
logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
abm.message_str = "[卡片消息解析失败]"
abm.message = [Plain(text="[卡片消息解析失败]")]
else:
logger.warning(f'[KOOK] 不支持的kook消息类型: "{data.get("type")}"')
logger.warning(f'[KOOK] 不支持的kook消息类型: "{data.type.name}"')
abm.message_str = "[不支持的消息类型]"
abm.message = [Plain(text="[不支持的消息类型]")]
astrbot/core/platform/sources/kook/kook_client.py
+103 -56
View File
@@ -1,6 +1,5 @@
import asyncio
import base64
import json
import os
import random
import time
@@ -9,13 +8,23 @@ from pathlib import Path
import aiofiles
import aiohttp
import pydantic
import websockets
from astrbot import logger
from astrbot.core.platform.message_type import MessageType
from .kook_config import KookConfig
from .kook_types import KookApiPaths, KookMessageType
from .kook_types import (
KookApiPaths,
KookGatewayIndexResponse,
KookHelloEventData,
KookMessageSignal,
KookMessageType,
KookResumeAckEventData,
KookUserMeResponse,
KookWebsocketEvent,
)
class KookClient:
@@ -23,7 +32,8 @@ class KookClient:
# 数据字段
self.config = config
self._bot_id = ""
self._bot_name = ""
self._bot_username = ""
self._bot_nickname = ""
# 资源字段
self._http_client = aiohttp.ClientSession(
@@ -48,37 +58,50 @@ class KookClient:
return self._bot_id
@property
def bot_name(self):
return self._bot_name
def bot_nickname(self):
return self._bot_nickname
async def get_bot_info(self) -> str:
"""获取机器人账号ID"""
@property
def bot_username(self):
return self._bot_username
async def get_bot_info(self) -> None:
"""获取机器人账号信息"""
url = KookApiPaths.USER_ME
try:
async with self._http_client.get(url) as resp:
if resp.status != 200:
logger.error(f"[KOOK] 获取机器人账号ID失败，状态码: {resp.status}")
return ""
logger.error(
f"[KOOK] 获取机器人账号信息失败，状态码: {resp.status} , {await resp.text()}"
)
return
try:
resp_content = KookUserMeResponse.from_dict(await resp.json())
except pydantic.ValidationError as e:
logger.error(
f"[KOOK] 获取机器人账号信息失败, 响应数据格式错误: \n{e}"
)
logger.error(f"[KOOK] 响应内容: {await resp.text()}")
return
data = await resp.json()
if data.get("code") != 0:
logger.error(f"[KOOK] 获取机器人账号ID失败: {data}")
return ""
if not resp_content.success():
logger.error(
f"[KOOK] 获取机器人账号信息失败: {resp_content.model_dump_json()}"
)
return
bot_id: str = data["data"]["id"]
bot_id: str = resp_content.data.id
self._bot_id = bot_id
logger.info(f"[KOOK] 获取机器人账号ID成功: {bot_id}")
bot_name: str = data["data"]["nickname"] or data["data"]["username"]
self._bot_name = bot_name
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_name}")
self._bot_nickname = resp_content.data.nickname
self._bot_username = resp_content.data.username
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
return bot_id
except Exception as e:
logger.error(f"[KOOK] 获取机器人账号ID异常: {e}")
return ""
logger.error(f"[KOOK] 获取机器人账号信息异常: {e}")
async def get_gateway_url(self, resume=False, sn=0, session_id=None):
async def get_gateway_url(self, resume=False, sn=0, session_id=None) -> str | None:
"""获取网关连接地址"""
url = KookApiPaths.GATEWAY_INDEX
@@ -96,14 +119,20 @@ class KookClient:
logger.error(f"[KOOK] 获取gateway失败，状态码: {resp.status}")
return None
data = await resp.json()
if data.get("code") != 0:
logger.error(f"[KOOK] 获取gateway失败: {data}")
resp_content = KookGatewayIndexResponse.from_dict(await resp.json())
if not resp_content.success():
logger.error(f"[KOOK] 获取gateway失败: {resp_content}")
return None
gateway_url: str = data["data"]["url"]
gateway_url: str = resp_content.data.url
logger.info(f"[KOOK] 获取gateway成功: {gateway_url.split('?')[0]}")
return gateway_url
except pydantic.ValidationError as e:
logger.error(f"[KOOK] 获取gateway失败, 响应数据格式错误: \n{e}")
logger.error(f"[KOOK] 原始响应内容: {await resp.text()}")
return None
except Exception as e:
logger.error(f"[KOOK] 获取gateway异常: {e}")
return None
@@ -156,7 +185,11 @@ class KookClient:
try:
while self.running:
try:
msg = await asyncio.wait_for(self.ws.recv(), timeout=10) # type: ignore
if self.ws is None:
logger.error("[KOOK] WebSocket 对象丢失，结束监听流程。")
break
msg = await asyncio.wait_for(self.ws.recv(), timeout=10)
if isinstance(msg, bytes):
try:
@@ -166,10 +199,15 @@ class KookClient:
continue
msg = msg.decode("utf-8")
data = json.loads(msg)
event = KookWebsocketEvent.from_json(msg)
# 处理不同类型的信令
await self._handle_signal(data)
await self._handle_signal(event)
except pydantic.ValidationError as e:
logger.error(f"[KOOK] 解析WebSocket事件数据格式失败: \n{e}")
logger.error(f"[KOOK] 原始响应内容: {msg}")
continue
except asyncio.TimeoutError:
# 超时检查，继续循环
@@ -187,38 +225,41 @@ class KookClient:
self.running = False
self._stop_event.set()
async def _handle_signal(self, data):
async def _handle_signal(self, event: KookWebsocketEvent):
"""处理不同类型的信令"""
signal_type = data.get("s")
data = event.data
if signal_type == 0: # 事件消息
# 更新消息序号
if "sn" in data:
self.last_sn = data["sn"]
await self.event_callback(data)
match event.signal:
case KookMessageSignal.MESSAGE:
if event.sn is not None:
self.last_sn = event.sn
await self.event_callback(data)
elif signal_type == 1: # HELLO握手
await self._handle_hello(data)
case KookMessageSignal.HELLO:
assert isinstance(data, KookHelloEventData)
await self._handle_hello(data)
elif signal_type == 3: # PONG心跳响应
await self._handle_pong(data)
case KookMessageSignal.RESUME_ACK:
assert isinstance(data, KookResumeAckEventData)
await self._handle_resume_ack(data)
elif signal_type == 5: # RECONNECT重连指令
await self._handle_reconnect(data)
case KookMessageSignal.PONG:
await self._handle_pong()
elif signal_type == 6: # RESUME ACK
await self._handle_resume_ack(data)
case KookMessageSignal.RECONNECT:
await self._handle_reconnect()
else:
logger.debug(f"[KOOK] 未处理的信令类型: {signal_type}")
case _:
logger.debug(
f"[KOOK] 未处理的信令类型: {event.signal.name}({event.signal.value})"
)
async def _handle_hello(self, data):
async def _handle_hello(self, data: KookHelloEventData):
"""处理HELLO握手"""
hello_data = data.get("d", {})
code = hello_data.get("code", 0)
code = data.code
if code == 0:
self.session_id = hello_data.get("session_id")
self.session_id = data.session_id
logger.info(f"[KOOK] 握手成功，session_id: {self.session_id}")
# TODO 重置重连延迟
# self.reconnect_delay = 1
@@ -228,12 +269,12 @@ class KookClient:
logger.error("[KOOK] Token已过期，需要重新获取")
self.running = False
async def _handle_pong(self, data):
async def _handle_pong(self):
"""处理PONG心跳响应"""
self.last_heartbeat_time = time.time()
self.heartbeat_failed_count = 0
async def _handle_reconnect(self, data):
async def _handle_reconnect(self):
"""处理重连指令"""
logger.warning("[KOOK] 收到重连指令")
# 清空本地状态
@@ -241,10 +282,9 @@ class KookClient:
self.session_id = None
self.running = False
async def _handle_resume_ack(self, data):
async def _handle_resume_ack(self, data: KookResumeAckEventData):
"""处理RESUME确认"""
resume_data = data.get("d", {})
self.session_id = resume_data.get("session_id")
self.session_id = data.session_id
logger.info(f"[KOOK] Resume成功，session_id: {self.session_id}")
async def _heartbeat_loop(self):
@@ -292,9 +332,16 @@ class KookClient:
async def _send_ping(self):
"""发送心跳PING"""
if self.ws is None:
logger.warning("[KOOK] 尚未连接kook WebSocket服务器, 跳过发送心跳包流程")
return
try:
ping_data = {"s": 2, "sn": self.last_sn}
await self.ws.send(json.dumps(ping_data)) # type: ignore
ping_data = KookWebsocketEvent(
signal=KookMessageSignal.PING,
data=None,
sn=self.last_sn,
)
await self.ws.send(ping_data.to_json())
except Exception as e:
logger.error(f"[KOOK] 发送心跳失败: {e}")
astrbot/core/platform/sources/kook/kook_config.py
-2
View File
@@ -9,7 +9,6 @@ class KookConfig:
# 基础配置
token: str
bot_nickname: str = ""
enable: bool = False
id: str = "kook"
@@ -41,7 +40,6 @@ class KookConfig:
# id=config_dict.get("id", "kook"),
enable=config_dict.get("enable", False),
token=config_dict.get("kook_bot_token", ""),
bot_nickname=config_dict.get("kook_bot_nickname", ""),
reconnect_delay=config_dict.get(
"kook_reconnect_delay",
KookConfig.reconnect_delay,
astrbot/core/platform/sources/kook/kook_event.py
+3 -2
View File
@@ -27,6 +27,7 @@ from .kook_types import (
KookCardMessage,
KookCardMessageContainer,
KookMessageType,
KookModuleType,
OrderMessage,
)
@@ -111,7 +112,7 @@ class KookEvent(AstrMessageEvent):
KookCardMessage(
modules=[
FileModule(
type="audio",
type=KookModuleType.AUDIO,
title=title,
src=url,
)
@@ -182,7 +183,7 @@ class KookEvent(AstrMessageEvent):
if item.reply_id:
reply_id = item.reply_id
if not item.text:
logger.debug(f'[Kook] 跳过空消息,类型为"{item.type}"')
logger.debug(f'[Kook] 跳过空消息,类型为"{item.type.name}"')
continue
try:
await self.client.send_text(
astrbot/core/platform/sources/kook/kook_types.py
+319 -55
View File
@@ -1,10 +1,8 @@
import json
from dataclasses import field
from enum import IntEnum
from typing import Literal
from enum import IntEnum, StrEnum
from typing import Annotated, Any, Literal
from pydantic import BaseModel, ConfigDict
from pydantic.dataclasses import dataclass
from pydantic import BaseModel, ConfigDict, Field, model_validator
class KookApiPaths:
@@ -25,8 +23,9 @@ class KookApiPaths:
DIRECT_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/direct-message/create"
# 定义参见kook事件结构文档: https://developer.kookapp.cn/doc/event/event-introduction
class KookMessageType(IntEnum):
"""定义参见kook事件结构文档: https://developer.kookapp.cn/doc/event/event-introduction"""
TEXT = 1
IMAGE = 2
VIDEO = 3
@@ -37,6 +36,26 @@ class KookMessageType(IntEnum):
SYSTEM = 255
class KookModuleType(StrEnum):
PLAIN_TEXT = "plain-text"
KMARKDOWN = "kmarkdown"
IMAGE = "image"
BUTTON = "button"
HEADER = "header"
SECTION = "section"
IMAGE_GROUP = "image-group"
CONTAINER = "container"
ACTION_GROUP = "action-group"
CONTEXT = "context"
DIVIDER = "divider"
FILE = "file"
AUDIO = "audio"
VIDEO = "video"
COUNTDOWN = "countdown"
INVITE = "invite"
CARD = "card"
ThemeType = Literal[
"primary", "success", "danger", "warning", "info", "secondary", "none", "invisible"
]
@@ -48,43 +67,81 @@ SectionMode = Literal["left", "right"]
CountdownMode = Literal["day", "hour", "second"]
class KookCardColor(str):
"""16 进制色值"""
class KookBaseDataClass(BaseModel):
model_config = ConfigDict(
extra="allow",
arbitrary_types_allowed=True,
populate_by_name=True,
)
@classmethod
def from_dict(cls, raw_data: dict):
return cls.model_validate(raw_data)
@classmethod
def from_json(cls, raw_data: str | bytes | bytearray):
return cls.model_validate_json(raw_data)
def to_dict(
self,
mode: Literal["json", "python"] | str = "python",
by_alias=True,
exclude_none=True,
exclude_unset=False,
) -> dict:
return self.model_dump(
by_alias=by_alias,
exclude_none=exclude_none,
mode=mode,
exclude_unset=exclude_unset,
)
def to_json(
self,
indent: int | None = None,
ensure_ascii=False,
by_alias=True,
exclude_none=True,
exclude_unset=False,
) -> str:
return self.model_dump_json(
indent=indent,
ensure_ascii=ensure_ascii,
by_alias=by_alias,
exclude_none=exclude_none,
exclude_unset=exclude_unset,
)
class KookCardModelBase:
class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
type: str
@dataclass
class PlainTextElement(KookCardModelBase):
content: str
type: str = "plain-text"
type: Literal[KookModuleType.PLAIN_TEXT] = KookModuleType.PLAIN_TEXT
emoji: bool = True
@dataclass
class KmarkdownElement(KookCardModelBase):
content: str
type: str = "kmarkdown"
type: Literal[KookModuleType.KMARKDOWN] = KookModuleType.KMARKDOWN
@dataclass
class ImageElement(KookCardModelBase):
src: str
type: str = "image"
type: Literal[KookModuleType.IMAGE] = KookModuleType.IMAGE
alt: str = ""
size: SizeType = "lg"
circle: bool = False
fallbackUrl: str | None = None
@dataclass
class ButtonElement(KookCardModelBase):
text: str
type: str = "button"
type: Literal[KookModuleType.BUTTON] = KookModuleType.BUTTON
theme: ThemeType = "primary"
value: str = ""
"""当为 link 时，会跳转到 value 代表的链接;
@@ -96,93 +153,88 @@ class ButtonElement(KookCardModelBase):
AnyElement = PlainTextElement | KmarkdownElement | ImageElement | ButtonElement | str
@dataclass
class ParagraphStructure(KookCardModelBase):
fields: list[PlainTextElement | KmarkdownElement]
type: str = "paragraph"
type: Literal["paragraph"] = "paragraph"
cols: int = 1
"""范围是 1-3 , 移动端忽略此参数"""
@dataclass
class HeaderModule(KookCardModelBase):
text: PlainTextElement
type: str = "header"
type: Literal[KookModuleType.HEADER] = KookModuleType.HEADER
@dataclass
class SectionModule(KookCardModelBase):
text: PlainTextElement | KmarkdownElement | ParagraphStructure
type: str = "section"
type: Literal[KookModuleType.SECTION] = KookModuleType.SECTION
mode: SectionMode = "left"
accessory: ImageElement | ButtonElement | None = None
@dataclass
class ImageGroupModule(KookCardModelBase):
"""1 到多张图片的组合"""
elements: list[ImageElement]
type: str = "image-group"
type: Literal[KookModuleType.IMAGE_GROUP] = KookModuleType.IMAGE_GROUP
@dataclass
class ContainerModule(KookCardModelBase):
"""1 到多张图片的组合，与图片组模块(ImageGroupModule)不同，图片并不会裁切为正方形。多张图片会纵向排列。"""
elements: list[ImageElement]
type: str = "container"
type: Literal[KookModuleType.CONTAINER] = KookModuleType.CONTAINER
@dataclass
class ActionGroupModule(KookCardModelBase):
"""用来放按钮的模块"""
elements: list[ButtonElement]
type: str = "action-group"
type: Literal[KookModuleType.ACTION_GROUP] = KookModuleType.ACTION_GROUP
@dataclass
class ContextModule(KookCardModelBase):
elements: list[PlainTextElement | KmarkdownElement | ImageElement]
"""最多包含10个元素"""
type: str = "context"
type: Literal[KookModuleType.CONTEXT] = KookModuleType.CONTEXT
@dataclass
class DividerModule(KookCardModelBase):
type: str = "divider"
"""展示分割线用的"""
type: Literal[KookModuleType.DIVIDER] = KookModuleType.DIVIDER
@dataclass
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
type: Literal[KookModuleType.FILE, KookModuleType.AUDIO, KookModuleType.VIDEO] = (
KookModuleType.FILE
)
cover: str | None = None
"""cover 仅音频有效, 是音频的封面图"""
@dataclass
class CountdownModule(KookCardModelBase):
"""startTime 和 endTime 为毫秒时间戳，startTime 和 endTime 不能小于服务器当前时间戳。"""
endTime: int
"""毫秒时间戳"""
type: str = "countdown"
type: Literal[KookModuleType.COUNTDOWN] = KookModuleType.COUNTDOWN
startTime: int | None = None
"""毫秒时间戳, 仅当mode为second才有这个字段"""
mode: CountdownMode = "day"
"""mode 主要是倒计时的样式"""
@dataclass
class InviteModule(KookCardModelBase):
code: str
"""邀请链接或者邀请码"""
type: str = "invite"
type: Literal[KookModuleType.INVITE] = KookModuleType.INVITE
# 所有模块的联合类型
AnyModule = (
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
@@ -192,34 +244,29 @@ AnyModule = (
| DividerModule
| FileModule
| CountdownModule
| InviteModule
)
| InviteModule,
Field(discriminator="type"),
]
class KookCardMessage(BaseModel):
class KookCardMessage(KookBaseDataClass):
"""卡片定义文档详见 : https://developer.kookapp.cn/doc/cardmessage
此类型不能直接to_json后发送,因为kook要求卡片容器json顶层必须是**列表**
若要发送卡片消息，请使用KookCardMessageContainer
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
type: Literal[KookModuleType.CARD] = KookModuleType.CARD
theme: ThemeType | None = None
size: SizeType | None = None
color: KookCardColor | None = None
modules: list[AnyModule] = field(default_factory=list)
color: str | None = None
"""16 进制色值"""
modules: list[AnyModule] = Field(default_factory=list)
"""单个 card 模块数量不限制，但是一条消息中所有卡片的模块数量之和最多是 50"""
def add_module(self, module: AnyModule):
self.modules.append(module)
def to_dict(self, exclude_none: bool = True):
"""exclude_none：去掉值为 None 字段，保留结构"""
return self.model_dump(exclude_none=exclude_none)
def to_json(self, indent: int | None = None, ensure_ascii: bool = True):
return json.dumps(self.to_dict(), indent=indent, ensure_ascii=ensure_ascii)
class KookCardMessageContainer(list[KookCardMessage]):
"""卡片消息容器(列表),此类型可以直接to_json后发送出去"""
@@ -232,10 +279,227 @@ class KookCardMessageContainer(list[KookCardMessage]):
[i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii
)
@classmethod
def from_dict(cls, raw_data: list[dict[str, Any]]):
return cls(KookCardMessage.from_dict(item) for item in raw_data)
@dataclass
class OrderMessage:
class OrderMessage(BaseModel):
index: int
text: str
type: KookMessageType
reply_id: str | int = ""
class KookMessageSignal(IntEnum):
"""KOOK WebSocket 信令类型
ws文档: https://developer.kookapp.cn/doc/websocket""" # noqa: W291
MESSAGE = 0
"""server->client 消息(s包含聊天和通知消息)"""
HELLO = 1
"""server->client 客户端连接 ws 时, 服务端返回握手结果"""
PING = 2
"""client->server 心跳，ping"""
PONG = 3
"""server->client 心跳，pong"""
RESUME = 4
"""client->server resume, 恢复会话"""
RECONNECT = 5
"""server->client reconnect, 要求客户端断开当前连接重新连接"""
RESUME_ACK = 6
"""server->client resume ack"""
class KookChannelType(StrEnum):
GROUP = "GROUP"
PERSON = "PERSON"
BROADCAST = "BROADCAST"
class KookAuthor(KookBaseDataClass):
id: str
username: str
identify_num: str
nickname: str
bot: bool
online: bool
avatar: str | None = None
vip_avatar: str | None = None
status: int
roles: list[int] = Field(default_factory=list)
class KookKMarkdown(KookBaseDataClass):
raw_content: str
mention_part: list[Any] = Field(default_factory=list)
mention_role_part: list[Any] = Field(default_factory=list)
class KookExtra(KookBaseDataClass):
type: int | str
code: str | None = None
body: dict[str, Any] | None = None
author: KookAuthor | None = None
kmarkdown: KookKMarkdown | None = None
last_msg_content: str | None = None
mention: list[str] = Field(default_factory=list)
mention_all: bool = False
mention_here: bool = False
class KookMessageEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.MESSAGE] = Field(
KookMessageSignal.MESSAGE, exclude=True
)
"""only for type hint"""
channel_type: KookChannelType
type: KookMessageType
target_id: str
author_id: str
content: str | dict[str, Any]
msg_id: str
msg_timestamp: int
nonce: str
from_type: int
extra: KookExtra
class KookHelloEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.HELLO] = Field(
KookMessageSignal.HELLO, exclude=True
)
"""only for type hint"""
code: int
session_id: str
class KookPingEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.PING] = Field(
KookMessageSignal.PING, exclude=True
)
"""only for type hint"""
class KookPongEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.PONG] = Field(
KookMessageSignal.PONG, exclude=True
)
"""only for type hint"""
class KookResumeEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.RESUME] = Field(
KookMessageSignal.RESUME, exclude=True
)
"""only for type hint"""
class KookReconnectEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.RECONNECT] = Field(
KookMessageSignal.RECONNECT, exclude=True
)
"""only for type hint"""
code: int
err: str
class KookResumeAckEventData(KookBaseDataClass):
signal: Literal[KookMessageSignal.RESUME_ACK] = Field(
KookMessageSignal.RESUME_ACK, exclude=True
)
"""only for type hint"""
session_id: str
class KookWebsocketEvent(KookBaseDataClass):
"""KOOK WebSocket 原始推送结构"""
signal: KookMessageSignal = Field(
..., validation_alias="s", serialization_alias="s"
)
"""信令类型"""
data: Annotated[
KookMessageEventData
| KookHelloEventData
| KookPingEventData
| KookPongEventData
| KookResumeEventData
| KookReconnectEventData
| KookResumeAckEventData
| None,
Field(discriminator="signal"),
] = Field(None, validation_alias="d", serialization_alias="d")
"""数据事件主体,对应原字段是'd'"""
sn: int | None = None
"""消息序号 , 用来确定消息顺序和ws重连时使用
详见ws连接流程文档: https://developer.kookapp.cn/doc/websocket#%E8%BF%9E%E6%8E%A5%E6%B5%81%E7%A8%8B""" # noqa: W291
@model_validator(mode="before")
@classmethod
def _inject_signal_into_data(cls, data: Any) -> Any:
"""在解析前，把外层的 s 同步到内层的 d 中，供 discriminator 使用"""
if isinstance(data, dict):
s_value = data.get("s")
d_value = data.get("d")
if s_value is not None and isinstance(d_value, dict):
d_value["signal"] = s_value
return data
class KookUserTag(KookBaseDataClass):
color: str
bg_color: str
text: str
class KookApiResponseBase(KookBaseDataClass):
code: int
message: str
data: Any
def success(self) -> bool:
return self.code == 0
class KookUserMeData(KookBaseDataClass):
"""USER_ME 接口返回的 'data' 字段主体"""
id: str
username: str
identify_num: str
nickname: str
bot: bool
online: bool
status: int
bot_status: int
avatar: str
vip_avatar: str | None = None
banner: str | None = None
roles: list[Any] = Field(default_factory=list)
is_vip: bool
vip_amp: bool
wealth_level: int
mobile_verified: bool
client_id: str
tag_info: KookUserTag | None = None
class KookUserMeResponse(KookApiResponseBase):
"""USER_ME 完整响应结构"""
data: KookUserMeData
class KookGatewayIndexData(KookBaseDataClass):
url: str
class KookGatewayIndexResponse(KookApiResponseBase):
"""USER_ME 完整响应结构"""
data: KookGatewayIndexData