feat: implements support for KOOK role mentions (#7626)

* feat: 实现kook适配器响应`@`角色(role)的能力

* refactor: kook适配器处理role时,`At`组件保留`@`角色的名称而不是id

* fix: kook适配器处理role时,role_id的判断问题

* refactor: 移除kook适配器中的一个# type: ignore

* fix: 修复kook适配器 role mention转换成`At`组件时保留不是角色名称的bug;

* unittest: 给kook适配器添加带有role mention的事件消息的单测,并添加消息组件转换判断单测

* unittest: 部分重构test_kook_event.py和test_kook_types.py 单测

* unittest: 添加kook适配器的 `user/me` `user/view` 接口响应数据验证单测

* fix: 修复kook适配器接收频道权限更新消息会报错的bug

* fix:  不额外处理kook的道具消息

* fix: 使用async with self._http_client.get

* refactor: kook适配器转换文本内容为消息组件时,只strip mention之间的空格

* fix: 修复 role_mention_counter 计数不正确的问题

* fix: 修复kook适配器发送卡片失败的问题;区分两类kook 数据类的to_dict to_json行为

* chore: 添加注释

* refactor: 重构kook适配器的角色缓存功能,使其无锁,性能更好且具备良好的重试机制

* refactor: kook适配器的channel_id 改为 guild_id

* feat: kook适配器响应频道角色更新事件时不再清空整个角色id缓存,而是只清理特定频道的角色id缓存

* unittest: 添加kook适配器的update_role事件的数据类验证单测

* refactor: 补上了一些打印的日志消息文本

refactor: 补上了一些打印的日志消息文本

refactor: 补上了一些打印的日志消息文本

* refactor: 修复kook适配器潜在可能的类型问题

* refactor: `clean_roles_cache`重命名为`clear_guild_roles_cache`

astrbot/core/platform/sources/kook/kook_adapter.py

@@ -13,12 +13,13 @@ from astrbot.api.platform import (
     PlatformMetadata,
     register_platform_adapter,
 )
-from astrbot.core.message.components import File, Record, Video
+from astrbot.core.message.components import BaseMessageComponent, 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_roles_record import KookRolesRecord
 from .kook_types import (
     ContainerModule,
     FileModule,
@@ -27,14 +28,18 @@ from .kook_types import (
     KmarkdownElement,
     KookCardMessageContainer,
     KookChannelType,
+    KookMarkdownMentionRolePart,
+    KookMentionTagName,
     KookMessageEventData,
     KookMessageType,
     KookModuleType,
+    KookRoleExtraType,
     PlainTextElement,
     SectionModule,
 )
 
-KOOK_AT_SELECTOR_REGEX = re.compile(r"\(met\)([^()]+)\(met\)")
+KOOK_AT_SELECTOR_REGEX = re.compile(r"\((met|rol)\)([^()]+)\(\1\)")
+AT_MENTION_PREFIX_REGEX = re.compile(r"^@[^\s]+(\s*-\s*[^\s]+)?\s*")
 
 
 @register_platform_adapter(
@@ -53,6 +58,7 @@ class KookPlatformAdapter(Platform):
         self._reconnect_task = None
         self.running = False
         self._main_task = None
+        self._roles_cache = KookRolesRecord("", self.client.http_client)
 
     async def send_by_session(
         self, session: MessageSesion, message_chain: MessageChain
@@ -84,7 +90,7 @@ class KookPlatformAdapter(Platform):
         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] 收到来自机器人自身的消息, 忽略此消息")
+                logger.debug("[KOOK] 判断此消息为来自机器人自身的消息, 忽略此消息")
                 return
             try:
                 abm = await self.convert_message(event)
@@ -92,8 +98,18 @@ class KookPlatformAdapter(Platform):
             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()}")
+            match event.extra.type:
+                case KookRoleExtraType():
+                    # 此时 target_id 就是频道id(guild_id)
+                    guild_id = event.target_id
+                    logger.info(
+                        f'[KOOK] 收到频道"{guild_id}"的角色更新通知, 类型为"{event.extra.type.value}", 刷新角色id缓存'
+                    )
+                    self._roles_cache.clear_guild_roles_cache(int(guild_id))
+                case _:
+                    logger.debug(
+                        f'[KOOK] 判断此消息为"{event.extra.type}"类型的系统通知, 因未实现此消息的处理流程而忽略此消息, 原始消息数据: {event.to_json()}'
+                    )
 
     async def run(self):
         """主运行循环"""
@@ -124,6 +140,8 @@ class KookPlatformAdapter(Platform):
                 logger.info("[KOOK] 尝试连接KOOK服务器...")
 
                 # 尝试连接
+                await self.client.get_bot_info()
+                self._roles_cache.set_bot_id(self.client.bot_id)
                 success = await self.client.connect()
 
                 if success:
@@ -191,47 +209,86 @@ class KookPlatformAdapter(Platform):
 
         logger.info("[KOOK] 资源清理完成")
 
-    def _parse_kmarkdown_text_message(
-        self, data: KookMessageEventData, self_id: str
-    ) -> tuple[list, str]:
-        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 [], ""
+    async def _convert_text_message_to_component(
+        self,
+        content: str,
+        raw_content: str,
+        mention_role_part: list[KookMarkdownMentionRolePart] | None = None,
+        guild_id: str | None = None,
+        mention_name_map: dict[str, str] | None = None,
+    ) -> tuple[list[BaseMessageComponent], str]:
+        # kook平台有一个角色(role)的概念,他表示拥有某一类权限的许多用户
+        # 且角色本身也有一个自己的id,与正常用户id不同
+        # 而在频道中是可以`@`角色的,而想要知道bot是否属于某个角色
+        # 需要通过 `/user/view` 接口获取当前bot账号的某个频道下所属角色的id
+        # 为了解决 https://github.com/AstrBotDevs/AstrBot/issues/7539
+        # 在确定机器人需要响应某个`(rol)xxx(rol)`时,需要将角色id替换装当前的bot id
+        # 包装成`At`机器人自己,而`At`的name就保留角色名称
+        # 如果没有查询到角色id或者bot不属于某类角色, 则不处理此`(rol)xxx(rol)`
+        # 暂时想不到能在不修改原有消息内容的情况下处理这个角色mention的方案
 
-        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.mention_part
-        if isinstance(mention_part, list):
-            for item in mention_part:
-                if not isinstance(item, dict):
-                    continue
-                mention_id = item.get("id")
-                if mention_id is None:
-                    continue
-                mention_name_map[str(mention_id)] = str(item.get("username", ""))
-
-        components = []
+        message_str = raw_content
+        bot_id = self.client.bot_id
+        bot_nickname = self.client.bot_nickname
+        bot_username = self.client.bot_username
+        components: list[BaseMessageComponent] = []
+        if mention_name_map is None:
+            mention_name_map = {}
         cursor = 0
+
+        role_mention_counter = -1
+
         for match in KOOK_AT_SELECTOR_REGEX.finditer(content):
             if match.start() > cursor:
-                plain_text = content[cursor : match.start()]
+                plain_text = content[cursor : match.start()].strip(" ")
                 if plain_text:
                     components.append(Plain(text=plain_text))
 
-            mention_target = match.group(1).strip()
-            if mention_target == "all":
+            tag_name = match.group(1)
+            mention_target = match.group(2).strip()
+            if tag_name == KookMentionTagName.MENTION and mention_target == "all":
                 components.append(AtAll())
+            elif tag_name == KookMentionTagName.ROLE:
+                role_mention_counter += 1
+                role_id = 0
+                role_mention_name = mention_target
+                if mention_role_part is not None:
+                    if len(mention_role_part) > role_mention_counter:
+                        role_mention_name = mention_role_part[role_mention_counter].name
+                        role_id = mention_role_part[role_mention_counter].role_id
+                        if (
+                            bot_nickname == role_mention_name
+                            or bot_username == role_mention_name
+                        ):
+                            components.append(
+                                At(
+                                    qq=bot_id,
+                                    name=role_mention_name,  # 保留角色名称
+                                )
+                            )
+                            continue
+                if not mention_target.isdigit() and role_id == 0:
+                    continue
+
+                role_id = role_id or int(mention_target)
+                if not guild_id:
+                    continue
+
+                if not guild_id.isdigit():
+                    continue
+
+                if not await self._roles_cache.has_role_in_channel(
+                    role_id, int(guild_id)
+                ):
+                    continue
+
+                components.append(
+                    At(
+                        qq=bot_id,
+                        name=role_mention_name,  # 保留角色名称
+                    )
+                )
+
             elif mention_target:
                 components.append(
                     At(
@@ -242,11 +299,11 @@ class KookPlatformAdapter(Platform):
             cursor = match.end()
 
         if cursor < len(content):
-            tail_text = content[cursor:]
+            tail_text = content[cursor:].strip(" ")
             if tail_text:
                 components.append(Plain(text=tail_text))
 
-        message_str = raw_content
+        message_str = raw_content.strip()
         if components:
             for comp in components:
                 if isinstance(comp, Plain):
@@ -254,9 +311,8 @@ class KookPlatformAdapter(Platform):
                         continue
                     break
                 if isinstance(comp, At):
-                    if str(comp.qq) == str(self_id):
-                        message_str = re.sub(
-                            r"^@[^\s]+(\s*-\s*[^\s]+)?\s*",
+                    if str(comp.qq) == str(self.client.bot_id):
+                        message_str = AT_MENTION_PREFIX_REGEX.sub(
                             "",
                             message_str,
                             count=1,
@@ -270,10 +326,44 @@ class KookPlatformAdapter(Platform):
 
         return components, message_str
 
-    def _parse_card_message(self, data: KookMessageEventData) -> tuple[list, str]:
+    async def _parse_kmarkdown_message(
+        self, data: KookMessageEventData
+    ) -> tuple[list[BaseMessageComponent], str]:
+        kmarkdown = data.extra.kmarkdown
+        guild_id = data.extra.guild_id
+        mention_role_part = None
+        if kmarkdown:
+            mention_role_part = kmarkdown.mention_role_part
+        # 无法处理可能会收到的道具消息content,只能保留原样
+        content = str(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
+
+        mention_name_map: dict[str, str] = {}
+        mention_part = kmarkdown.mention_part
+        for item in mention_part:
+            mention_id = item.id
+            if mention_id is None:
+                continue
+            mention_name_map[str(mention_id)] = str(item.username)
+
+        return await self._convert_text_message_to_component(
+            content, raw_content, mention_role_part, guild_id, mention_name_map
+        )
+
+    async def _parse_card_message(
+        self, data: KookMessageEventData
+    ) -> tuple[list[BaseMessageComponent], str]:
         content = data.content
         if not isinstance(content, str):
             content = str(content)
+        guild_id = data.extra.guild_id
 
         card_list = KookCardMessageContainer.from_dict(json.loads(content))
 
@@ -304,18 +394,13 @@ class KookPlatformAdapter(Platform):
                         logger.debug(f"[KOOK] 跳过或未处理模块: {module.type}")
 
         text = "".join(text_parts)
-        message = []
+        message: list[BaseMessageComponent] = []
 
         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))
+            component_parts, text = await self._convert_text_message_to_component(
+                text, text, guild_id=guild_id
+            )
+            message.extend(component_parts)
 
         for img_url in images:
             message.append(Image(file=img_url))
@@ -387,12 +472,10 @@ class KookPlatformAdapter(Platform):
         abm.message_id = data.msg_id or "unknown"
 
         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
+            abm.message, abm.message_str = await self._parse_kmarkdown_message(data)
         elif data.type == KookMessageType.CARD:
             try:
-                abm.message, abm.message_str = self._parse_card_message(data)
+                abm.message, abm.message_str = await self._parse_card_message(data)
             except Exception as exp:
                 logger.error(f"[KOOK] 卡片消息解析失败: {exp}")
                 logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")

astrbot/core/platform/sources/kook/kook_client.py

@@ -3,6 +3,7 @@ import base64
 import os
 import random
 import time
+import traceback
 import zlib
 from pathlib import Path
 
@@ -59,12 +60,18 @@ class KookClient:
 
     @property
     def bot_nickname(self):
+        """机器人昵称"""
         return self._bot_nickname
 
     @property
     def bot_username(self):
+        """机器人名称"""
         return self._bot_username
 
+    @property
+    def http_client(self):
+        return self._http_client
+
     async def get_bot_info(self) -> None:
         """获取机器人账号信息"""
         url = KookApiPaths.USER_ME
@@ -151,7 +158,6 @@ class KookClient:
             gateway_url = await self.get_gateway_url(
                 resume=resume, sn=self.last_sn, session_id=self.session_id
             )
-            await self.get_bot_info()
 
             if not gateway_url:
                 return False
@@ -215,8 +221,8 @@ class KookClient:
                 except websockets.exceptions.ConnectionClosed:
                     logger.warning("[KOOK] WebSocket连接已关闭")
                     break
-                except Exception as e:
-                    logger.error(f"[KOOK] 消息处理异常: {e}")
+                except Exception:
+                    logger.error(f"[KOOK] 消息处理异常: {traceback.format_exc()}")
                     break
 
         except Exception as e:
@@ -236,11 +242,15 @@ class KookClient:
                 await self.event_callback(data)
 
             case KookMessageSignal.HELLO:
-                assert isinstance(data, KookHelloEventData)
+                assert isinstance(data, KookHelloEventData), (
+                    f"期望 data 为 {KookHelloEventData.__name__}, 实际为 {type(data).__name__}，"
+                )
                 await self._handle_hello(data)
 
             case KookMessageSignal.RESUME_ACK:
-                assert isinstance(data, KookResumeAckEventData)
+                assert isinstance(data, KookResumeAckEventData), (
+                    f"期望 data 为 {KookResumeAckEventData.__name__}, 实际为 {type(data).__name__}，"
+                )
                 await self._handle_resume_ack(data)
 
             case KookMessageSignal.PONG:
@@ -367,8 +377,8 @@ class KookClient:
             "type": kook_message_type,
         }
         if reply_message_id:
-            payload["quote"] = reply_message_id
-            payload["reply_msg_id"] = reply_message_id
+            payload["quote"] = str(reply_message_id)
+            payload["reply_msg_id"] = str(reply_message_id)
 
         try:
             async with self._http_client.post(url, json=payload) as resp:

astrbot/core/platform/sources/kook/kook_roles_record.py

@@ -0,0 +1,164 @@
+import asyncio
+import time
+from collections import OrderedDict
+from dataclasses import dataclass
+
+import aiohttp
+import pydantic
+
+from astrbot import logger
+
+from .kook_types import KookApiPaths, KookUserViewResponse
+
+USER_VIEW_REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=3)
+ROLES_CACHE_MAX_SIZE = 2000
+MAX_RETRY_TIMES = 3
+RETRY_INTERVAL_SECOND = 1 * 60
+
+
+@dataclass
+class RolesCache:
+    value: set[int] | None = None
+    failed_count: int = 0
+    latest_update_time: float = 0
+
+    def update(self, roles: set[int] | None) -> None:
+        if roles is not None:
+            self.failed_count = 0
+        self.value = roles
+        self.latest_update_time = time.time()
+
+    def add_failed(self):
+        self.failed_count += 1
+
+    def reset(self, without_value=False):
+        if not without_value:
+            self.value = None
+        self.failed_count = 0
+        self.latest_update_time = 0
+
+
+class KookRolesRecord:
+    """自动和缓存获取机器人所需响应的消息频道的role信息"""
+
+    def __init__(self, bot_id: str, http_client: aiohttp.ClientSession):
+        # self._locks: dict[int, asyncio.Lock] = defaultdict(asyncio.Lock)
+        self._lock = asyncio.Lock()
+        self._bot_id = bot_id
+        self._http_client = http_client
+        # TODO 这个些配置后续加到适配器配置项里
+        self._cache_max_size = ROLES_CACHE_MAX_SIZE
+        self._max_retry_times = MAX_RETRY_TIMES
+        self._retry_interval = RETRY_INTERVAL_SECOND
+        self._roles_cache: OrderedDict[int, RolesCache] = OrderedDict()
+        self._pending_tasks: dict[int, asyncio.Future] = {}
+
+    def set_bot_id(self, bot_id: str):
+        self._bot_id = bot_id
+
+    def clear_guild_roles_cache(self, guild_id: int):
+        self._roles_cache.pop(guild_id, None)
+        self._pending_tasks.pop(guild_id, None)
+
+    async def _fetch_roles_by_guild_id(self, guild_id: int) -> set[int] | None:
+        # 由于需要判断bot账号是属于某个角色(role)才会回复消息,
+        # 而后续来自同一个频道的消息,在第一次查这个role的时候,
+        # 会一直阻塞消息接收直到请求完成或者报错,
+        # 所以,这里特意调低了timeout时间,避免阻塞太久
+        url = KookApiPaths.USER_VIEW
+        try:
+            async with self._http_client.get(
+                url,
+                params={
+                    "guild_id": guild_id,
+                    "user_id": self._bot_id,
+                },
+                # TODO 这个超时时间后续加到适配器配置项里
+                timeout=USER_VIEW_REQUEST_TIMEOUT,
+            ) as resp:
+                if resp.status != 200:
+                    logger.error(
+                        f'[KOOK] 获取机器人在频道"{guild_id}"的角色id信息失败，状态码: {resp.status} , {await resp.text()}'
+                    )
+                    return
+                try:
+                    resp_content = KookUserViewResponse.from_dict(await resp.json())
+                except pydantic.ValidationError as e:
+                    logger.error(
+                        f'[KOOK] 获取机器人在频道"{guild_id}"的角色id信息失败, 响应数据格式错误: \n{e}'
+                    )
+                    logger.error(f"[KOOK] 响应内容: {await resp.text()}")
+                    return
+
+                if not resp_content.success():
+                    logger.error(
+                        f'[KOOK] 获取机器人在频道"{guild_id}"的角色id信息失败: {resp_content.model_dump_json()}'
+                    )
+                    return
+
+                logger.info(f'[KOOK] 获取机器人在频道"{guild_id}"的角色id成功')
+                return set(resp_content.data.roles)
+
+        except Exception as e:
+            logger.error(
+                f'[KOOK] 获取机器人在频道"{guild_id}"的角色id信息时请求异常: {e}'
+            )
+            return
+
+    async def has_role_in_channel(self, role_id: int, guild_id: int) -> bool:
+        if (cache := self._roles_cache.get(guild_id)) is not None:
+            self._roles_cache.move_to_end(guild_id)
+            roles = cache.value
+            if roles is not None:
+                return role_id in roles
+
+        new_future: asyncio.Future[set[int] | None] = asyncio.Future()
+        actual_future: asyncio.Future[set[int] | None] = self._pending_tasks.setdefault(
+            guild_id, new_future
+        )
+
+        if actual_future is not new_future:
+            roles = await actual_future
+            if roles is None:
+                return False
+            return role_id in roles
+
+        try:
+            if (cache := self._roles_cache.get(guild_id)) is not None:
+                if (
+                    cache.failed_count > self._max_retry_times
+                    and time.time() - cache.latest_update_time < self._retry_interval
+                ):
+                    new_future.set_result(None)
+                    return False
+
+            # 简单的容量控制 (LRU)
+            if len(self._roles_cache) + 1 > self._cache_max_size:
+                self._roles_cache.popitem(last=False)
+
+            roles_set = await self._fetch_roles_by_guild_id(guild_id)
+
+            cache = self._roles_cache.get(guild_id)
+            if cache is not None:
+                cache.update(roles_set)
+                self._roles_cache.move_to_end(guild_id)
+            else:
+                cache = RolesCache(roles_set, latest_update_time=time.time())
+                self._roles_cache[guild_id] = cache
+
+            result = False
+            if roles_set is None:
+                cache.add_failed()
+            else:
+                result = role_id in roles_set
+
+            new_future.set_result(roles_set)
+            return result
+        except Exception as e:
+            new_future.set_result(None)
+            logger.error(
+                f'[KOOK] 获取机器人在频道"{guild_id}"的角色id信息时发生异常: {e}'
+            )
+            return False
+        finally:
+            self._pending_tasks.pop(guild_id, None)

astrbot/core/platform/sources/kook/kook_types.py

@@ -2,7 +2,7 @@ import json
 from enum import Enum, IntEnum
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 
 class KookApiPaths:
@@ -13,6 +13,7 @@ class KookApiPaths:
 
     # 初始化相关
     USER_ME = f"{BASE_URL}{API_VERSION_PATH}/user/me"
+    USER_VIEW = f"{BASE_URL}{API_VERSION_PATH}/user/view"
     GATEWAY_INDEX = f"{BASE_URL}{API_VERSION_PATH}/gateway/index"
 
     # 消息相关
@@ -23,6 +24,14 @@ class KookApiPaths:
     DIRECT_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/direct-message/create"
 
 
+class KookMentionTagName(str, Enum):
+    """用来匹配 `(tagName)value(tagName)` 格式里的tagName , 例如: `(met)all(met)`
+    定义参见KMarkdown语法文档: https://developer.kookapp.cn/doc/kmarkdown"""
+
+    MENTION = "met"
+    ROLE = "rol"
+
+
 class KookMessageType(IntEnum):
     """定义参见kook事件结构文档: https://developer.kookapp.cn/doc/event/event-introduction"""
 
@@ -56,6 +65,14 @@ class KookModuleType(str, Enum):
     CARD = "card"
 
 
+class KookRoleExtraType(str, Enum):
+    """定义参见kook事件结构文档: https://developer.kookapp.cn/doc/event/event-introduction"""
+
+    ADDED_ROLE = "added_role"
+    DELETED_ROLE = "deleted_role"
+    UPDATED_ROLE = "updated_role"
+
+
 ThemeType = Literal[
     "primary", "success", "danger", "warning", "info", "secondary", "none", "invisible"
 ]
@@ -67,7 +84,9 @@ SectionMode = Literal["left", "right"]
 CountdownMode = Literal["day", "hour", "second"]
 
 
-class KookBaseDataClass(BaseModel):
+class KookBaseReceiveDataClass(BaseModel):
+    """接收数据基类,`to_dict`/`to_json`默认保证尽量json原样输出"""
+
     model_config = ConfigDict(
         extra="allow",
         arbitrary_types_allowed=True,
@@ -84,11 +103,51 @@ class KookBaseDataClass(BaseModel):
 
     def to_dict(
         self,
-        mode: Literal["json", "python"] | str = "python",
+        mode: Literal["json", "python"] | str = "json",
+        by_alias=True,
+        exclude_none=False,
+        exclude_unset=True,
+    ) -> dict:
+        """默认配置预期场景为尽量原样输出,若需要使用此数据类发送json数据,
+        请`exclude_none=True, exclude_unset=False`"""
+        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=False,
+        exclude_unset=True,
+    ) -> str:
+        """默认配置预期场景为尽量原样输出,若需要使用此数据类发送json数据,
+        请`exclude_none=True, exclude_unset=False`"""
+        return self.model_dump_json(
+            indent=indent,
+            ensure_ascii=ensure_ascii,
+            by_alias=by_alias,
+            exclude_none=exclude_none,
+            exclude_unset=exclude_unset,
+        )
+
+
+class KookBaseSendDataClass(KookBaseReceiveDataClass):
+    """发送数据基类,`to_dict`/`to_json`保证默认输出内容格式包含接口格式所需最简格式内容"""
+
+    def to_dict(
+        self,
+        mode: Literal["json", "python"] | str = "json",
         by_alias=True,
         exclude_none=True,
         exclude_unset=False,
     ) -> dict:
+        """默认配置预期场景为发送数据,若需要使用此数据类接收数据并尽量原样json输出,
+        请`exclude_none=False, exclude_unset=True`"""
         return self.model_dump(
             by_alias=by_alias,
             exclude_none=exclude_none,
@@ -104,6 +163,8 @@ class KookBaseDataClass(BaseModel):
         exclude_none=True,
         exclude_unset=False,
     ) -> str:
+        """默认配置预期场景为发送数据,若需要使用此数据类接收数据并尽量原样json输出,
+        请`exclude_none=False, exclude_unset=True`"""
         return self.model_dump_json(
             indent=indent,
             ensure_ascii=ensure_ascii,
@@ -113,7 +174,7 @@ class KookBaseDataClass(BaseModel):
         )
 
 
-class KookCardModelBase(KookBaseDataClass):
+class KookCardModelBase(KookBaseSendDataClass):
     """卡片模块基类"""
 
     type: str
@@ -249,10 +310,28 @@ AnyModule = Annotated[
 ]
 
 
-class KookCardMessage(KookBaseDataClass):
+class KookCardMessage(KookBaseSendDataClass):
     """卡片定义文档详见 : https://developer.kookapp.cn/doc/cardmessage
-    此类型不能直接to_json后发送,因为kook要求卡片容器json顶层必须是**列表**
-    若要发送卡片消息，请使用KookCardMessageContainer
+    适用于发送单个卡片消息
+    将此消息类型放入`Json`的data字段进行卡片消息发送,适配器会自动添加顶层的列表
+    若要发送多个卡片消息，推荐使用KookCardMessageContainer进行卡片消息组装
+
+    使用方法：
+    ```python
+    chain = []
+    chain.append(
+            Json(
+                data=KookCardMessage(
+                    theme="info",
+                    size="lg",
+                    modules=[
+                        HeaderModule(text=PlainTextElement(content="test1")),
+                    ],
+                ).to_dict()
+            )
+        )
+    yield event.chain_result(chain)
+    ```
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -269,14 +348,71 @@ class KookCardMessage(KookBaseDataClass):
 
 
 class KookCardMessageContainer(list[KookCardMessage]):
-    """卡片消息容器(列表),此类型可以直接to_json后发送出去"""
+    """卡片消息容器(列表),可放入多个卡片消息(KookCardMessage)
+
+    使用方法:
+    ```python
+    chain = []
+        chain.append(
+            Json(
+                data=KookCardMessageContainer(
+                    [
+                        KookCardMessage(
+                            theme="info",
+                            size="lg",
+                            modules=[
+                                HeaderModule(text=PlainTextElement(content="test1")),
+                            ],
+                        )
+                    ]
+                ).to_dict()
+            )
+        )
+        yield event.chain_result(chain)
+    ```
+    """
 
     def append(self, object: KookCardMessage) -> None:
         return super().append(object)
 
-    def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
+    def to_dict(
+        self,
+        by_alias=True,
+        exclude_none=True,
+        exclude_unset=False,
+    ) -> list[dict]:
+        """默认配置预期场景为发送数据,若需要使用此数据类接收数据并尽量原样json输出,
+        请`exclude_none=False, exclude_unset=True`"""
+        return [
+            i.to_dict(
+                by_alias=by_alias,
+                exclude_none=exclude_none,
+                exclude_unset=exclude_unset,
+            )
+            for i in self
+        ]
+
+    def to_json(
+        self,
+        indent: int | None = None,
+        ensure_ascii: bool = True,
+        by_alias=True,
+        exclude_none=True,
+        exclude_unset=False,
+    ) -> str:
+        """默认配置预期场景为发送数据,若需要使用此数据类接收数据并尽量原样json输出,
+        请`exclude_none=False, exclude_unset=True`"""
         return json.dumps(
-            [i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii
+            [
+                i.to_dict(
+                    by_alias=by_alias,
+                    exclude_none=exclude_none,
+                    exclude_unset=exclude_unset,
+                )
+                for i in self
+            ],
+            indent=indent,
+            ensure_ascii=ensure_ascii,
         )
 
     @classmethod
@@ -293,7 +429,7 @@ class OrderMessage(BaseModel):
 
 class KookMessageSignal(IntEnum):
     """KOOK WebSocket 信令类型
-    ws文档: https://developer.kookapp.cn/doc/websocket"""  # noqa: W291
+    ws文档: https://developer.kookapp.cn/doc/websocket"""
 
     MESSAGE = 0
     """server->client  消息(s包含聊天和通知消息)"""
@@ -317,7 +453,7 @@ class KookChannelType(str, Enum):
     BROADCAST = "BROADCAST"
 
 
-class KookAuthor(KookBaseDataClass):
+class KookAuthor(KookBaseReceiveDataClass):
     id: str
     username: str
     identify_num: str
@@ -330,25 +466,110 @@ class KookAuthor(KookBaseDataClass):
     roles: list[int] = Field(default_factory=list)
 
 
-class KookKMarkdown(KookBaseDataClass):
+class KookMarkdownMentionPart(KookBaseReceiveDataClass):
+    """
+    文档参考: https://developer.kookapp.cn/doc/event/message
+    """
+
+    id: str
+    username: str
+    full_name: str
+    avatar: str
+
+
+class KookMarkdownMentionRolePart(KookBaseReceiveDataClass):
+    """
+    文档参考: https://developer.kookapp.cn/doc/event/message
+    """
+
+    role_id: int
+    name: str
+    color: int
+    color_type: int
+    color_map: list[Any]
+    position: int | None = None
+    hoist: int | None = None
+    mentionable: int | None = None
+    permissions: int | None = None
+
+
+class KookKMarkdown(KookBaseReceiveDataClass):
     raw_content: str
-    mention_part: list[Any] = Field(default_factory=list)
-    mention_role_part: list[Any] = Field(default_factory=list)
+    mention_part: list[KookMarkdownMentionPart] = Field(default_factory=list)
+    mention_role_part: list[KookMarkdownMentionRolePart] = Field(default_factory=list)
 
 
-class KookExtra(KookBaseDataClass):
-    type: int | str
+class KookRole(KookBaseReceiveDataClass):
+    """服务器角色对象数据结构"""
+
+    role_id: int = Field(alias="role_id")
+    name: str | None = None
+    color: int | None = None
+    position: int | None = None
+    hoist: int | None = 0  # 是否在成员列表中单独展示
+    mentionable: int | None = 0  # 是否允许所有人提到该角色
+    permissions: int | None = None
+
+
+class KookRoleEventBody(KookBaseReceiveDataClass):
+    """
+    服务器角色相关事件 (added_role, updated_role, deleted_role) 的 Body 部分
+    文档参考: https://developer.kookapp.cn/doc/event/guild-role
+    """
+
+    role_id: int | None = None  # 在 deleted_role 中通常只给 ID
+    name: str | None = None
+    color: int | None = None
+    position: int | None = None
+    hoist: int | None = None
+    mentionable: int | None = None
+    permissions: int | None = None
+    # 有些事件会将完整的 role 对象包裹在 body 里
+    # 如果是 added_role 且需要处理更完整的结构，可以扩展
+
+
+class KookExtra(KookBaseReceiveDataClass):
+    """事件结构定义
+    文档参考 : https://developer.kookapp.cn/doc/event/event-introduction"""
+
+    type: KookRoleExtraType | str | int
+    """当 type 非系统消息(255)时, type为int
+
+    当 type 为系统消息(255)时, type为str
+    """
+
     code: str | None = None
-    body: dict[str, Any] | None = None
+    body: KookRole | 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
+    guild_id: str | None = None
+    guild_type: int | None = None
+    channel_name: str | None = None
+    visible_only: str | None = None
+    mention_no_at: list | None = None
+    mention_roles: list[int] | None = None
+    nav_channels: list | None = None
+    emoji: list | None = None
+    preview_content: str | None = None
+    channel_type: int | None = None
+    send_msg_device: int | None = None
+
+    @field_validator("type", mode="before")
+    @classmethod
+    def parse_type(cls, value):
+        """优先尝试匹配枚举，失败则保留原值"""
+        if isinstance(value, str):
+            if value in {e.value for e in KookRoleExtraType}:
+                return KookRoleExtraType(value)
+
+        return value
 
 
-class KookMessageEventData(KookBaseDataClass):
+class KookMessageEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.MESSAGE] = Field(
         KookMessageSignal.MESSAGE, exclude=True
     )
@@ -358,7 +579,7 @@ class KookMessageEventData(KookBaseDataClass):
     type: KookMessageType
     target_id: str
     author_id: str
-    content: str | dict[str, Any]
+    content: str | dict[str, Any]  # 道具消息时这里是dict
     msg_id: str
     msg_timestamp: int
     nonce: str
@@ -366,7 +587,7 @@ class KookMessageEventData(KookBaseDataClass):
     extra: KookExtra
 
 
-class KookHelloEventData(KookBaseDataClass):
+class KookHelloEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.HELLO] = Field(
         KookMessageSignal.HELLO, exclude=True
     )
@@ -376,28 +597,28 @@ class KookHelloEventData(KookBaseDataClass):
     session_id: str
 
 
-class KookPingEventData(KookBaseDataClass):
+class KookPingEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.PING] = Field(
         KookMessageSignal.PING, exclude=True
     )
     """only for type hint"""
 
 
-class KookPongEventData(KookBaseDataClass):
+class KookPongEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.PONG] = Field(
         KookMessageSignal.PONG, exclude=True
     )
     """only for type hint"""
 
 
-class KookResumeEventData(KookBaseDataClass):
+class KookResumeEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.RESUME] = Field(
         KookMessageSignal.RESUME, exclude=True
     )
     """only for type hint"""
 
 
-class KookReconnectEventData(KookBaseDataClass):
+class KookReconnectEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.RECONNECT] = Field(
         KookMessageSignal.RECONNECT, exclude=True
     )
@@ -407,7 +628,7 @@ class KookReconnectEventData(KookBaseDataClass):
     err: str
 
 
-class KookResumeAckEventData(KookBaseDataClass):
+class KookResumeAckEventData(KookBaseReceiveDataClass):
     signal: Literal[KookMessageSignal.RESUME_ACK] = Field(
         KookMessageSignal.RESUME_ACK, exclude=True
     )
@@ -416,7 +637,7 @@ class KookResumeAckEventData(KookBaseDataClass):
     session_id: str
 
 
-class KookWebsocketEvent(KookBaseDataClass):
+class KookWebsocketEvent(KookBaseReceiveDataClass):
     """KOOK WebSocket 原始推送结构"""
 
     signal: KookMessageSignal = Field(
@@ -451,22 +672,22 @@ class KookWebsocketEvent(KookBaseDataClass):
         return data
 
 
-class KookUserTag(KookBaseDataClass):
+class KookUserTag(KookBaseReceiveDataClass):
     color: str
     bg_color: str
     text: str
 
 
-class KookApiResponseBase(KookBaseDataClass):
+class KookApiResponseBase(KookBaseReceiveDataClass):
     code: int
     message: str
-    data: Any
+    data: dict  # 就算请求失败了也是空dict
 
     def success(self) -> bool:
         return self.code == 0
 
 
-class KookUserMeData(KookBaseDataClass):
+class KookUserMeData(KookBaseReceiveDataClass):
     """USER_ME 接口返回的 'data' 字段主体"""
 
     id: str
@@ -495,7 +716,47 @@ class KookUserMeResponse(KookApiResponseBase):
     data: KookUserMeData
 
 
-class KookGatewayIndexData(KookBaseDataClass):
+class KookUserMeViewData(KookBaseReceiveDataClass):
+    """USER_ME 接口返回的 'data' 字段主体"""
+
+    class KookTagInfo(KookBaseReceiveDataClass):
+        color: str
+        bg_color: str
+        text: str
+
+    id: str
+    username: str
+    identify_num: str
+    online: bool
+    os: str
+    status: int
+    avatar: str
+    vip_avatar: str
+    banner: str
+    nickname: str
+    roles: list[int]
+    is_vip: bool
+    vip_amp: bool
+    bot: bool
+    kpm_vip: str | None = None
+    wealth_level: int
+    bot_status: int
+    tag_info: KookTagInfo
+    mobile_verified: bool
+    is_sys: bool
+    client_id: str
+    verified: bool
+    joined_at: int
+    active_time: int
+
+
+class KookUserViewResponse(KookApiResponseBase):
+    """USER_VIEW 完整响应结构"""
+
+    data: KookUserMeViewData
+
+
+class KookGatewayIndexData(KookBaseReceiveDataClass):
     url: str
 
 

tests/test_kook/data/kook_api_response_user_me.json

@@ -0,0 +1,36 @@
+{
+    "code": 0,
+    "message": "操作成功",
+    "data": {
+        "id": "573092175",
+        "username": "bot_username",
+        "identify_num": "9561",
+        "online": false,
+        "os": "Websocket",
+        "status": 0,
+        "avatar": "https://example.com",
+        "vip_avatar": "https://example.com",
+        "banner": "",
+        "nickname": "bot_nickname",
+        "roles": [],
+        "is_vip": false,
+        "vip_amp": false,
+        "bot": true,
+        "kpm_vip": null,
+        "wealth_level": 0,
+        "bot_status": 0,
+        "tag_info": {
+            "color": "#0096FF",
+            "bg_color": "#0096FF33",
+            "text": "机器人"
+        },
+        "mobile_verified": true,
+        "is_sys": false,
+        "client_id": "g3nsxNQhNMZFKatU",
+        "verified": false,
+        "mobile_prefix": "86",
+        "mobile": "****",
+        "invited_count": 0,
+        "intent": 255
+    }
+}

tests/test_kook/data/kook_api_response_user_view.json

@@ -0,0 +1,36 @@
+{
+    "code": 0,
+    "message": "操作成功",
+    "data": {
+        "id": "573092175",
+        "username": "bot_username",
+        "identify_num": "9561",
+        "online": false,
+        "os": "Websocket",
+        "status": 0,
+        "avatar": "https://img.kookapp.cn/assets/bot.png?x-oss-process=style/icon",
+        "vip_avatar": "https://img.kookapp.cn/assets/bot.png?x-oss-process=style/icon",
+        "banner": "",
+        "nickname": "bot_nickname",
+        "roles": [
+            13726212
+        ],
+        "is_vip": false,
+        "vip_amp": false,
+        "bot": true,
+        "kpm_vip": null,
+        "wealth_level": 0,
+        "bot_status": 0,
+        "tag_info": {
+            "color": "#0096FF",
+            "bg_color": "#0096FF33",
+            "text": "机器人"
+        },
+        "mobile_verified": true,
+        "is_sys": false,
+        "client_id": "g3nsxNQhNMZFKatU",
+        "verified": false,
+        "joined_at": 1772260532000,
+        "active_time": 1776418003694
+    }
+}

tests/test_kook/data/kook_ws_event_group_message.json

@@ -26,7 +26,7 @@
                     "banner": "",
                     "nickname": "some_username",
                     "roles": [
-                        63724577
+                        63423577
                     ],
                     "is_vip": false,
                     "vip_amp": false,

tests/test_kook/data/kook_ws_event_group_message_with_mention.json

@@ -0,0 +1,88 @@
+{
+    "s": 0,
+    "d": {
+        "channel_type": "GROUP",
+        "type": 9,
+        "target_id": "2732467349811313213",
+        "author_id": "7324688132731983",
+        "content": "(rol)25555643(rol) /help (met)3351526782(met)  (met)all(met) ",
+        "msg_id": "9b047d81-40fe-41af-ad39-916ae77e6b20",
+        "msg_timestamp": 1776405840600,
+        "nonce": "r2oQkO7kRpNSgmsv7TNl2zOA",
+        "from_type": 1,
+        "extra": {
+            "type": 9,
+            "code": "",
+            "author": {
+                "id": "3351526782",
+                "username": "some_username",
+                "identify_num": "4198",
+                "nickname": "some_username",
+                "bot": false,
+                "online": true,
+                "avatar": "https://example.com",
+                "vip_avatar": "https://example.com",
+                "status": 1,
+                "roles": [
+                    63423577
+                ],
+                "os": "Websocket",
+                "banner": "",
+                "is_vip": false,
+                "vip_amp": false,
+                "nameplate": [],
+                "wealth_level": 0,
+                "is_sys": false
+            },
+            "kmarkdown": {
+                "raw_content": "@some_role /help @some_username  @全体成员",
+                "mention_part": [
+                    {
+                        "id": "3351526782",
+                        "username": "some_username",
+                        "full_name": "some_username#4198",
+                        "avatar": "https://example.com",
+                        "wealth_level": 0
+                    }
+                ],
+                "mention_role_part": [
+                    {
+                        "role_id": 25555643,
+                        "name": "some_role",
+                        "color": 0,
+                        "color_type": 1,
+                        "color_map": []
+                    }
+                ],
+                "channel_part": [],
+                "spl": []
+            },
+            "last_msg_content": "some_username：@some_role /help @some_username  @全体成员",
+            "mention": [
+                "3351526782"
+            ],
+            "mention_all": true,
+            "mention_here": false,
+            "guild_id": "1239678456780469",
+            "guild_type": 0,
+            "channel_name": "聊天大厅",
+            "visible_only": "",
+            "mention_no_at": [],
+            "mention_roles": [
+                25555643
+            ],
+            "nav_channels": [],
+            "emoji": [],
+            "preview_content": "",
+            "channel_type": 1,
+            "send_msg_device": 0
+        }
+    },
+    "extra": {
+        "verifyToken": "kW4FH_ASHio1hosd",
+        "encryptKey": "",
+        "callbackUrl": "",
+        "intent": 255
+    },
+    "sn": 1
+}

tests/test_kook/data/kook_ws_event_group_system_message_update_role.json

@@ -0,0 +1,38 @@
+{
+    "s": 0,
+    "d": {
+        "channel_type": "GROUP",
+        "type": 255,
+        "target_id": "6059671921720469",
+        "author_id": "1",
+        "content": "[系统消息]",
+        "msg_id": "35de353d-aa69-40a8-b329-e4498ed8d30d",
+        "msg_timestamp": 1776498088031,
+        "nonce": "",
+        "from_type": 1,
+        "extra": {
+            "type": "updated_role",
+            "body": {
+                "role_id": 16752184,
+                "name": "some_username",
+                "color": 0,
+                "position": 5,
+                "hoist": 0,
+                "mentionable": 0,
+                "permissions": 5571747823,
+                "desc": "",
+                "color_type": 1,
+                "color_map": {},
+                "type": 1,
+                "op_permissions": 0
+            }
+        }
+    },
+    "sn": 4,
+    "extra": {
+        "verifyToken": "kW4FH_ASHio1hosd",
+        "encryptKey": "",
+        "callbackUrl": "",
+        "intent": 255
+    }
+}

tests/test_kook/shared.py

@@ -1,5 +1,111 @@
-from pathlib import Path
+import json
+from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock
 
 
+import aiohttp
+from astrbot.api.platform import AstrBotMessage, MessageType
+from astrbot.core.message.components import (
+    File,
+    Record,
+)
+
+
+from pathlib import Path
+
 CURRENT_DIR = Path(__file__).parent
 TEST_DATA_DIR = CURRENT_DIR / "data"
+
+
+class KookEventDataPath:
+    GROUP_MESSAGE_WITH_MENTION = (
+        TEST_DATA_DIR / "kook_ws_event_group_message_with_mention.json"
+    )
+    GROUP_MESSAGE = TEST_DATA_DIR / "kook_ws_event_group_message.json"
+    HELLO = TEST_DATA_DIR / "kook_ws_event_hello.json"
+    MESSAGE_WITH_CARD_1 = TEST_DATA_DIR / "kook_ws_event_message_with_card_1.json"
+    MESSAGE_WITH_CARD_2 = TEST_DATA_DIR / "kook_ws_event_message_with_card_2.json"
+    PING = TEST_DATA_DIR / "kook_ws_event_ping.json"
+    PONG = TEST_DATA_DIR / "kook_ws_event_pong.json"
+    PRIVATE_MESSAGE = TEST_DATA_DIR / "kook_ws_event_private_message.json"
+    PRIVATE_SYSTEM_MESSAGE = TEST_DATA_DIR / "kook_ws_event_private_system_message.json"
+    RECONNECT_ERR = TEST_DATA_DIR / "kook_ws_event_reconnect_err.json"
+    RESUME_ACK = TEST_DATA_DIR / "kook_ws_event_resume_ack.json"
+    RESUME = TEST_DATA_DIR / "kook_ws_event_resume.json"
+    GROUP_SYSTEM_MESSAGE_UPDATE_ROLE = TEST_DATA_DIR / "kook_ws_event_group_system_message_update_role.json"
+
+
+class KookApiDataPath:
+    USER_ME = TEST_DATA_DIR / "kook_api_response_user_me.json"
+    USER_VIEW = TEST_DATA_DIR / "kook_api_response_user_view.json"
+
+
+def mock_kook_client(upload_asset_return: str, send_text_return: str):
+    client = MagicMock()
+
+    client.upload_asset = AsyncMock(return_value=upload_asset_return)
+    client.send_text = AsyncMock(return_value=send_text_return)
+    return client
+
+
+def mock_http_client(
+    http_method: str = "get",
+    return_value: str | dict | list | None = None,
+    status: int = 200,
+):
+    """Mock aiohttp ClientSession"""
+
+    if isinstance(return_value, (dict, list)):
+        response_text = json.dumps(return_value)
+    else:
+        response_text = return_value or "{}"
+
+    mock_response = MagicMock()
+    mock_response.status = status
+    mock_response.text = AsyncMock(return_value=response_text)
+    mock_response.json = AsyncMock(
+        return_value=json.loads(response_text) if response_text else {}
+    )
+    mock_response.read = AsyncMock(return_value=response_text.encode())
+    mock_response.__aenter__ = AsyncMock(return_value=mock_response)
+    mock_response.__aexit__ = AsyncMock(return_value=None)
+    mock_session = MagicMock()
+
+    async def mock_method(*args, **kwargs):
+        return mock_response
+
+    setattr(mock_session, http_method.lower(), mock_method)
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=None)
+
+    mock_session.close = AsyncMock()
+
+    return mock_session
+
+
+def mock_file_message(input: str):
+    message = MagicMock(spec=File)
+    message.get_file = AsyncMock(return_value=input)
+    return message
+
+
+def mock_record_message(input: str):
+    message = MagicMock(spec=Record)
+    message.text = input
+    message.convert_to_file_path = AsyncMock(return_value=input)
+    return message
+
+
+def mock_astrbot_message():
+    message = AstrBotMessage()
+    message.type = MessageType.OTHER_MESSAGE
+    message.group_id = "test"
+    message.session_id = "test"
+    message.message_id = "test"
+    return message
+
+
+def mock_kook_roles_record(bot_id: str, http_client: aiohttp.ClientSession):
+    instance = AsyncMock()
+    instance.has_role_in_channel = AsyncMock(return_value=True)
+    return instance

tests/test_kook/test_kook_client.py

@@ -0,0 +1,156 @@
+import asyncio
+from dataclasses import dataclass, field
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from astrbot.core.message.components import (
+    At,
+    AtAll,
+    BaseMessageComponent,
+    Plain,
+    Record,
+)
+from astrbot.core.platform.sources.kook.kook_client import KookClient
+from astrbot.core.platform.sources.kook.kook_config import KookConfig
+from astrbot.core.platform.sources.kook.kook_types import (
+    KookMessageEventData,
+    KookWebsocketEvent,
+)
+from tests.test_kook.shared import (
+    KookEventDataPath,
+    mock_http_client,
+    mock_kook_roles_record,
+)
+
+TEST_BOT_ID = 1234567891
+TEST_BOT_USERNAME = "test_username"
+TEST_BOT_NICKNAME = "test_nickname"
+
+
+def mock_kook_client(config: KookConfig, event_callback):
+    class MockKookClient:
+        def __init__(self, config, callback):
+            self.bot_id = TEST_BOT_ID
+            self.bot_nickname = TEST_BOT_NICKNAME
+            self.bot_username = TEST_BOT_USERNAME
+            self.http_client = mock_http_client()
+            self.connect = AsyncMock()
+            self.close = AsyncMock()
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            pass
+
+    return MockKookClient(config, event_callback)
+
+
+def get_json_field(content: dict, json_field_path: list[str | int]) -> Any:
+    expend_value = content
+    for key in json_field_path:
+        expend_value = expend_value[key]
+    return expend_value
+
+
+@dataclass
+class JsonFieldPaths:
+    message_str: list[int | str] = field(default_factory=list)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "expected_json_data_path, expected_message_str, expected_message_components",
+    [
+        (
+            KookEventDataPath.GROUP_MESSAGE_WITH_MENTION,
+            ["d", "extra", "kmarkdown", "raw_content"],
+            [
+                # 这里默认机器人一定属于某个角色id
+                At(qq=TEST_BOT_ID, name="some_role"),
+                Plain(text="/help"),
+                At(qq=3351526782, name="some_username"),
+                AtAll(qq="all", name=""),
+            ],
+        ),
+        (
+            KookEventDataPath.GROUP_MESSAGE,
+            ["d", "extra", "kmarkdown", "raw_content"],
+            [Plain(text="done!")],
+        ),
+        (
+            KookEventDataPath.MESSAGE_WITH_CARD_1,
+            "[audio]",
+            [
+                Plain(text="[audio]"),
+                Record(
+                    file="https://img.kookapp.cn/attachments/2026-03/03/69a6841c3125d.wav",
+                    url="",
+                    text=None,
+                    path=None,
+                ),
+            ],
+        ),
+        (
+            KookEventDataPath.MESSAGE_WITH_CARD_2,
+            ["d", "extra", "kmarkdown", "raw_content"],
+            [
+                Plain(text="(met)"),
+                Plain(text="all(met) #hello \\*\\*world\\*\\*  [audio]\n😆"),
+                Record(
+                    file="https://img.kookapp.cn/attachments/2026-03/03/69a6841c3125d.wav",
+                    url="",
+                    text=None,
+                    path=None,
+                ),
+            ],
+        ),
+        (
+            KookEventDataPath.PRIVATE_MESSAGE,
+            ["d", "extra", "kmarkdown", "raw_content"],
+            [Plain(text="/help")],
+        ),
+    ],
+)
+async def test_kook_event_warp_message(
+    expected_json_data_path: Path,
+    expected_message_str: list[int | str] | str,
+    expected_message_components: list[BaseMessageComponent],
+):
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr(
+        "astrbot.core.platform.sources.kook.kook_adapter.KookClient", mock_kook_client
+    )
+    monkeypatch.setattr(
+        "astrbot.core.platform.sources.kook.kook_adapter.KookRolesRecord",
+        mock_kook_roles_record,
+    )
+
+    from astrbot.core.platform.sources.kook.kook_adapter import KookPlatformAdapter
+
+    adapter = KookPlatformAdapter({}, {}, asyncio.Queue())
+
+    raw_event_str = expected_json_data_path.read_text(encoding="utf-8")
+    raw_event = json.loads(raw_event_str)
+    event = KookWebsocketEvent.from_json(
+        raw_event_str,
+    )
+    assert isinstance(event.data, KookMessageEventData)
+
+    astrbotMessage = await adapter.convert_message(event.data)
+    assert astrbotMessage.self_id == TEST_BOT_ID
+    assert astrbotMessage.sender.user_id == raw_event["d"]["author_id"]
+    assert (
+        astrbotMessage.sender.nickname == raw_event["d"]["extra"]["author"]["username"]
+    )
+    assert astrbotMessage.raw_message == raw_event["d"]
+    assert astrbotMessage.message_id == raw_event["d"]["msg_id"]
+    assert astrbotMessage.message == expected_message_components
+    if isinstance(expected_message_str, str):
+        assert astrbotMessage.message_str == expected_message_str
+    else:
+        assert get_json_field(raw_event, expected_message_str)

tests/test_kook/test_kook_event.py

@@ -1,10 +1,8 @@
-from unittest.mock import AsyncMock, MagicMock
+import json
 
 import pytest
-from astrbot.api.platform import AstrBotMessage, MessageType, PlatformMetadata, Unknown
-from astrbot.api.event import MessageChain
+from astrbot.api.platform import PlatformMetadata, Unknown
 from astrbot.core.message.components import (
-    File,
     Image,
     Plain,
     Video,
@@ -12,44 +10,18 @@ from astrbot.core.message.components import (
     AtAll,
     BaseMessageComponent,
     Json,
-    Record,
     Reply,
 )
 
 
 from astrbot.core.platform.sources.kook.kook_event import KookEvent
 from astrbot.core.platform.sources.kook.kook_types import KookMessageType, OrderMessage
-
-
-async def mock_kook_client(upload_asset_return: str, send_text_return: str):
-    # 1. Mock 掉整个 KookClient 类
-    client = MagicMock()
-
-    client.upload_asset = AsyncMock(return_value=upload_asset_return)
-    client.send_text = AsyncMock(return_value=send_text_return)
-    return client
-
-
-def mock_file_message(input: str):
-    message = MagicMock(spec=File)
-    message.get_file = AsyncMock(return_value=input)
-    return message
-
-
-def mock_record_message(input: str):
-    message = MagicMock(spec=Record)
-    message.text = input
-    message.convert_to_file_path = AsyncMock(return_value=input)
-    return message
-
-
-def mock_astrbot_message():
-    message = AstrBotMessage()
-    message.type = MessageType.OTHER_MESSAGE
-    message.group_id = "test"
-    message.session_id = "test"
-    message.message_id = "test"
-    return message
+from tests.test_kook.shared import (
+    mock_astrbot_message,
+    mock_file_message,
+    mock_kook_client,
+    mock_record_message,
+)
 
 
 @pytest.mark.asyncio
@@ -161,7 +133,7 @@ async def test_kook_event_warp_message(
     expected_output: OrderMessage,
     expected_error: type[BaseException] | None,
 ):
-    client = await mock_kook_client(
+    client = mock_kook_client(
         upload_asset_return,
         "",
     )
@@ -184,5 +156,20 @@ async def test_kook_event_warp_message(
         return
 
     result = await event._wrap_message(1, input_message)
-    assert result == expected_output
-    
+
+    expected_output_text: str | list | dict = expected_output.text
+    is_json_text = False
+    try:
+        expected_output_text = json.loads(expected_output_text)
+        is_json_text = True
+    except:
+        pass
+
+    if is_json_text:
+        assert json.loads(result.text) == expected_output_text
+    else:
+        assert result.text == expected_output_text
+
+    assert result.index == expected_output.index
+    assert result.type == expected_output.type
+    assert result.reply_id == expected_output.reply_id

tests/test_kook/test_kook_types.py

@@ -15,16 +15,19 @@ from astrbot.core.platform.sources.kook.kook_types import (
     ImageGroupModule,
     InviteModule,
     KmarkdownElement,
+    KookApiResponseBase,
     KookCardMessage,
     KookMessageSignal,
     KookModuleType,
+    KookUserMeResponse,
+    KookUserViewResponse,
     KookWebsocketEvent,
     ParagraphStructure,
     PlainTextElement,
     SectionModule,
     KookCardMessageContainer,
 )
-from tests.test_kook.shared import TEST_DATA_DIR
+from tests.test_kook.shared import TEST_DATA_DIR, KookApiDataPath, KookEventDataPath
 
 
 def test_kook_card_message_container_append():
@@ -109,29 +112,31 @@ def test_all_kook_card_type():
     ).to_json(indent=4, ensure_ascii=False)
     assert json_output == expect_json_data
 
+
 @pytest.mark.parametrize(
-    "expected_json_data_filename",
+    "expected_json_data_path",
     [
-        ("kook_ws_event_group_message.json"),
-        ("kook_ws_event_hello.json"),
-        ("kook_ws_event_message_with_card_1.json"),
-        ("kook_ws_event_message_with_card_2.json"),
-        ("kook_ws_event_ping.json"),
-        ("kook_ws_event_pong.json"),
-        ("kook_ws_event_private_message.json"),
-        ("kook_ws_event_private_system_message.json"),
-        ("kook_ws_event_reconnect_err.json"),
-        ("kook_ws_event_resume_ack.json"),
-        ("kook_ws_event_resume.json"),
-        
+        (KookEventDataPath.GROUP_MESSAGE_WITH_MENTION),
+        (KookEventDataPath.GROUP_MESSAGE),
+        (KookEventDataPath.HELLO),
+        (KookEventDataPath.MESSAGE_WITH_CARD_1),
+        (KookEventDataPath.MESSAGE_WITH_CARD_2),
+        (KookEventDataPath.PING),
+        (KookEventDataPath.PONG),
+        (KookEventDataPath.PRIVATE_MESSAGE),
+        (KookEventDataPath.PRIVATE_SYSTEM_MESSAGE),
+        (KookEventDataPath.RECONNECT_ERR),
+        (KookEventDataPath.RESUME_ACK),
+        (KookEventDataPath.RESUME),
+        (KookEventDataPath.GROUP_SYSTEM_MESSAGE_UPDATE_ROLE),
     ],
 )
-def test_websocket_event_type_parse(expected_json_data_filename:str):
-    expected_json_data_str =(TEST_DATA_DIR / expected_json_data_filename).read_text(encoding="utf-8")
+def test_websocket_event_type_parse(expected_json_data_path: Path):
+    expected_json_data_str = (expected_json_data_path).read_text(encoding="utf-8")
     event = KookWebsocketEvent.from_json(
         expected_json_data_str,
     )
-    event_dict = event.to_dict(mode="json",exclude_unset=True,exclude_none=False)
+    event_dict = event.to_dict()
     assert event_dict == json.loads(expected_json_data_str)
 
 
@@ -141,8 +146,27 @@ def test_websocket_event_create():
         data=None,
         sn=0,
     )
-    assert ping_data.to_dict(mode="json")== {
+    assert ping_data.to_dict(exclude_none=True, exclude_unset=False) == {
         "s": KookMessageSignal.PING.value,
         "sn": 0,
     }
-    
+
+
+@pytest.mark.parametrize(
+    "expected_json_data_path, expected_dataclass",
+    [
+        (KookApiDataPath.USER_ME, KookUserMeResponse),
+        (KookApiDataPath.USER_VIEW, KookUserViewResponse),
+    ],
+)
+def test_api_response_type_parse(
+    expected_json_data_path: Path, expected_dataclass: type[KookApiResponseBase]
+):
+    expected_json_data_str = (expected_json_data_path).read_text(encoding="utf-8")
+
+    response_body = expected_dataclass.from_json(
+        expected_json_data_str,
+    )
+
+    body_dict = response_body.to_dict()
+    assert body_dict == json.loads(expected_json_data_str)