feat: integrates KOOK platform adapter#5658
Merged
Soulter merged 37 commits intoAstrBotDevs:masterfrom Mar 3, 2026
Merged
Conversation
dosubot
bot
added
the
size:XXL
Contributor
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! 此拉取请求旨在通过将 Kook 适配器集成到 AstrBot 的官方适配器目录中，显著扩展 AstrBot 的平台兼容性。这一变更使得 AstrBot 能够直接支持 Kook 平台的消息交互，从而为用户提供更广泛的部署和使用选择。核心改动包括添加 Kook 相关的配置、实现适配器的加载机制以及提供完整的 Kook 消息处理和发送功能。 Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
dosubot
bot
added
the
area:platform
|
Related Documentation Checked 1 published document(s) in 1 knowledge base(s). No updates required. How did I do? Any feedback? |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了 4 个问题，并给出了一些高层面的反馈：
- 新的
kook/config.py定义了大量配置（心跳、重连、日志等），但这些配置都没有真正被KookClient或适配器使用；建议要么通过将这些配置值接入KookClient（例如通过构造函数/关键字参数），要么删除该文件，以避免产生无用或重复的配置。 - 在
KookPlatformAdapter.handle_msg中，唤醒检测依赖于硬编码的bot_nickname = "astrbot"；更稳妥的做法是从 KOOK 或配置中推断机器人的名称/ID，而不是使用常量，这样在不同机器人名称的部署环境中也能正常工作。 - 存在两处不同的
MessageType导入（kook_client.py中从astrbot.core.platform.message_type导入，而kook_event.py中从astrbot.core.platform导入），这可能会造成困惑；建议在新的 KOOK 模块中统一MessageType的导入来源。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- 新的 `kook/config.py` 定义了大量配置（心跳、重连、日志等），但这些配置都没有真正被 `KookClient` 或适配器使用；建议要么通过将这些配置值接入 `KookClient`（例如通过构造函数/关键字参数），要么删除该文件，以避免产生无用或重复的配置。
- 在 `KookPlatformAdapter.handle_msg` 中，唤醒检测依赖于硬编码的 `bot_nickname = "astrbot"`；更稳妥的做法是从 KOOK 或配置中推断机器人的名称/ID，而不是使用常量，这样在不同机器人名称的部署环境中也能正常工作。
- 存在两处不同的 `MessageType` 导入（`kook_client.py` 中从 `astrbot.core.platform.message_type` 导入，而 `kook_event.py` 中从 `astrbot.core.platform` 导入），这可能会造成困惑；建议在新的 KOOK 模块中统一 `MessageType` 的导入来源。
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="275-284" />
<code_context>
+ async def reconnect(self):
</code_context>
<issue_to_address>
**issue (bug_risk):** 在关闭 HTTP 客户端后进行重连会复用一个已关闭的 `ClientSession`，这很可能会失败。
在 `reconnect` 中，`await self.close()` 会关闭 `self._http_client`，但 `connect()` 随后调用的 `get_gateway_url`/`get_bot_id` 会使用 `self._http_client.get(...)`，在已关闭的 `ClientSession` 上调用会抛出异常。可以考虑在 `reconnect` 中避免关闭 HTTP 客户端，或者在调用 `connect()` 之前重新创建一个新的 `aiohttp.ClientSession`（或以其他方式确保使用新的会话）。
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_event.py" line_range="47-56" />
<code_context>
+ self._file_message_counter = 0
</code_context>
<issue_to_address>
**suggestion (bug_risk):** `_file_message_counter` 没有在每次发送时重置，因此跨多条消息的文件上传日志可能会不准确。
由于 `_file_message_counter` 只在 `__init__` 中初始化，在 `send` 中从未重置，一旦你发送过带文件的消息，之后每一次 `send` 都会看到一个非零值，并按有文件上传的情况进行日志记录，即使当前并没有上传文件。应在 `send` 的开始或结束处重置 `_file_message_counter`，这样日志只反映当前消息所携带的文件。
Suggested implementation:
```python
self.astrbot_message_type: MessageType = message_obj.type
self._file_message_counter = 0
def _reset_file_message_counter(self) -> None:
"""Reset the per-send file upload counter.
This should be called at the start of each send() so that logging for
file uploads only reflects the files in the current outgoing message.
"""
self._file_message_counter = 0
def _warp_message(
self, index: int, message_component: BaseMessageComponent
) -> Coroutine[Any, Any, OrderMessage]:
```
为了完整实现你所描述的行为，确保 `send()`（负责发送消息的方法）在每次发送开始时调用 `self._reset_file_message_counter()`。例如：
```python
async def send(...):
self._reset_file_message_counter()
...
```
这可以确保 `_file_message_counter` 会在每一条要发送的消息前被重置，从而让关于文件上传的日志以每条消息为粒度，而不是在多条消息之间累积。
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="96" />
<code_context>
+ """click 代表用户点击的事件,默认为""，代表无任何事件。"""
+
+
+AnyElement = PlainTextElement | KmarkdownElement | ImageElement | ButtonElement | str
+
+
</code_context>
<issue_to_address>
**issue (complexity):** 建议用更简单的基于 protocol 的类型和 dataclass 封装替换大型的 `AnyModule`/`AnyElement` 联合类型以及 `list` 子类容器，在保持现有行为的前提下降低理解成本。
你可以保留完整的卡片模型，同时通过简化类型组合和容器模式来减少认知负担，而不会改变行为。
### 1. 用公共基类替换大型 `AnyModule`/`AnyElement` 联合类型
当前实现：
```python
class KookCardModelBase:
"""卡片模块基类"""
type: str
@dataclass
class HeaderModule(KookCardModelBase): ...
@dataclass
class SectionModule(KookCardModelBase): ...
# ...
AnyModule = (
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule
)
AnyElement = PlainTextElement | KmarkdownElement | ImageElement | ButtonElement | str
class KookCardMessage(BaseModel):
modules: list[AnyModule] = field(default_factory=list)
```
这些联合类型很大，而且由于模块本身是 dataclass 而不是 pydantic model，它们并没有被用于运行时的校验。你可以通过把基类改成合适的 protocol/model，并把 `modules` 声明为该基类类型，而不是显式的联合类型，从而降低复杂度：
```python
from typing import Protocol
class KookCardModule(Protocol):
type: str
# All your dataclasses implicitly satisfy this protocol:
# @dataclass
# class HeaderModule(KookCardModule): ...
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
theme: ThemeType | None = None
size: SizeType | None = None
color: KookCardColor | None = None
modules: list[KookCardModule] = field(default_factory=list)
def add_module(self, module: KookCardModule):
self.modules.append(module)
```
如果你实际上还没有在任何地方使用 `AnyElement`，也可以类似地定义一个 `Element` protocol（或者在不需要的地方直接删掉这个别名），以避免在类型注解中出现大型联合类型。
这样可以保留所有现有的类和行为，同时不再需要在每次新增模块/元素时手动保持 `AnyModule`/`AnyElement` 与之同步。
### 2. 避免让 `KookCardMessageContainer` 继承 `list`
继承 `list` 会带来额外的心智负担，但收益很有限。你可以改成一个带显式 `cards` 字段的简单封装类，同时保持现有的外部行为（`to_json` 和 `append`）：
当前实现：
```python
class KookCardMessageContainer(list[KookCardMessage]):
"""卡片消息容器(列表),此类型可以直接to_json后发送出去"""
def append(self, object: KookCardMessage) -> None:
return super().append(object)
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps(
[i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii
)
```
简化方案：
```python
from dataclasses import dataclass, field
@dataclass
class KookCardMessageContainer:
cards: list[KookCardMessage] = field(default_factory=list)
def append(self, card: KookCardMessage) -> None:
self.cards.append(card)
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps(
[card.to_dict() for card in self.cards],
indent=indent,
ensure_ascii=ensure_ascii,
)
```
现有调用代码：
```python
container = KookCardMessageContainer()
container.append(card)
payload = container.to_json()
```
仍然可以按原样工作，但这个类型现在是一个简单、语义清晰的容器，而不是一个特化的 `list` 子类。
</issue_to_address>
### Comment 4
<location path="astrbot/core/platform/sources/kook/config.py" line_range="6" />
<code_context>
+包含连接参数、重连策略、心跳设置等配置项
+"""
+
+# 连接配置
+CONNECTION_CONFIG = {
+ # 心跳配置
</code_context>
<issue_to_address>
**issue (complexity):** 建议将多个独立的配置字典和 getter 函数整合为一个带命名空间的 `CONFIG`，并通过一个统一的访问函数对外暴露，以简化配置的外部接口。
你可以在不删除任何配置项的前提下，通过把多个全局配置和 getter 合并为单一的、带命名空间的配置对象，来降低整体的配置复杂度。
与其暴露 5 个独立的字典和 3+ 个 getter 函数，不如保留所有键，但将它们归到一个 `CONFIG` 下，并使用单一访问 API：
```python
# config.py
CONFIG = {
"connection": {
"heartbeat_interval": 30,
"heartbeat_timeout": 6,
"max_heartbeat_failures": 3,
"initial_reconnect_delay": 1,
"max_reconnect_delay": 60,
"max_consecutive_failures": 5,
"websocket_timeout": 10,
"connection_timeout": 30,
"enable_compression": True,
"max_message_size": 1024 * 1024,
},
"logging": {
"level": "INFO",
"format": "[KOOK] %(message)s",
"enable_heartbeat_logs": False,
"enable_message_logs": False,
},
"error_handling": {
"retry_on_network_error": True,
"retry_on_token_expired": True,
"max_retry_attempts": 3,
"retry_delay_base": 2,
},
"performance": {
"enable_message_buffering": True,
"buffer_size": 100,
"enable_connection_pooling": True,
"max_concurrent_requests": 10,
},
"security": {
"verify_ssl": True,
"enable_rate_limiting": True,
"rate_limit_requests": 100,
"rate_limit_window": 60,
},
}
def get_config(section: str | None = None):
"""获取配置；section 为 None 时返回完整配置"""
if section is None:
return CONFIG
return CONFIG[section]
```
使用方式变为：
```python
connection_cfg = get_config("connection")
heartbeat_interval = connection_cfg["heartbeat_interval"]
```
这样可以在保留当前所有配置项和默认值的同时：
- 将对外的公共 API 从多个全局字典/getter 收缩为一个易于理解的入口。
- 降低读者的概念负担：只需要理解一个带分区的配置对象，而不是许多顶层符号。
- 让你更容易逐步将所需的配置节/键接入 `KookClient`/`KookPlatformAdapter`，而不会给人这些顶层字典已经全部被使用的错觉。
</issue_to_address>Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Original comment in English
Hey - I've found 4 issues, and left some high level feedback:
- The new
kook/config.pydefines extensive configuration (heartbeat, reconnect, logging, etc.) but none of it is actually used byKookClientor the adapter; consider either wiring these values intoKookClient(e.g., via constructor/kwargs) or removing the file to avoid dead/duplicated configuration. - In
KookPlatformAdapter.handle_msg, the wakeup detection relies on a hardcodedbot_nickname = "astrbot"; it would be more robust to infer the bot's name/ID from KOOK or config instead of a constant so deployments with different bot names still work correctly. - There are two different
MessageTypeimports (astrbot.core.platform.message_typeinkook_client.pyandastrbot.core.platforminkook_event.py), which can be confusing; consider unifying the import source forMessageTypeacross the new KOOK modules.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `kook/config.py` defines extensive configuration (heartbeat, reconnect, logging, etc.) but none of it is actually used by `KookClient` or the adapter; consider either wiring these values into `KookClient` (e.g., via constructor/kwargs) or removing the file to avoid dead/duplicated configuration.
- In `KookPlatformAdapter.handle_msg`, the wakeup detection relies on a hardcoded `bot_nickname = "astrbot"`; it would be more robust to infer the bot's name/ID from KOOK or config instead of a constant so deployments with different bot names still work correctly.
- There are two different `MessageType` imports (`astrbot.core.platform.message_type` in `kook_client.py` and `astrbot.core.platform` in `kook_event.py`), which can be confusing; consider unifying the import source for `MessageType` across the new KOOK modules.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="275-284" />
<code_context>
+ async def reconnect(self):
</code_context>
<issue_to_address>
**issue (bug_risk):** Reconnecting after closing the HTTP client will use a closed `ClientSession` and likely fail.
In `reconnect`, `await self.close()` shuts down `self._http_client`, but `connect()` then calls `get_gateway_url`/`get_bot_id`, which use `self._http_client.get(...)` and will raise on a closed `ClientSession`. Either avoid closing the HTTP client in `reconnect`, or recreate a fresh `aiohttp.ClientSession` (or otherwise ensure a new session) before calling `connect()`.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_event.py" line_range="47-56" />
<code_context>
+ self._file_message_counter = 0
</code_context>
<issue_to_address>
**suggestion (bug_risk):** `_file_message_counter` is never reset per send, so logging about file uploads can be incorrect across messages.
Because `_file_message_counter` is only initialized in `__init__` and never reset in `send`, once you’ve sent a message with files, every later `send` will see a non-zero value and log as if it’s uploading files even when it isn’t. Reset `_file_message_counter` at the start or end of `send` so the log reflects only the current message’s files.
Suggested implementation:
```python
self.astrbot_message_type: MessageType = message_obj.type
self._file_message_counter = 0
def _reset_file_message_counter(self) -> None:
"""Reset the per-send file upload counter.
This should be called at the start of each send() so that logging for
file uploads only reflects the files in the current outgoing message.
"""
self._file_message_counter = 0
def _warp_message(
self, index: int, message_component: BaseMessageComponent
) -> Coroutine[Any, Any, OrderMessage]:
```
To fully implement the behavior you described, ensure that `send()` (the method responsible for sending a message) calls `self._reset_file_message_counter()` at the start of each send. For example:
```python
async def send(...):
self._reset_file_message_counter()
...
```
This guarantees that `_file_message_counter` is reset for every outgoing message, so any logging about file uploads is accurate per message rather than accumulating across messages.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="96" />
<code_context>
+ """click 代表用户点击的事件,默认为""，代表无任何事件。"""
+
+
+AnyElement = PlainTextElement | KmarkdownElement | ImageElement | ButtonElement | str
+
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider replacing the large `AnyModule`/`AnyElement` unions and the `list` subclass container with simpler protocol-based types and a dataclass wrapper to lower cognitive overhead while preserving behavior.
You can keep the full card model but reduce the cognitive load by simplifying the type plumbing and container pattern without changing behavior.
### 1. Drop the big `AnyModule`/`AnyElement` unions in favor of a common base
Right now:
```python
class KookCardModelBase:
"""卡片模块基类"""
type: str
@dataclass
class HeaderModule(KookCardModelBase): ...
@dataclass
class SectionModule(KookCardModelBase): ...
# ...
AnyModule = (
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule
)
AnyElement = PlainTextElement | KmarkdownElement | ImageElement | ButtonElement | str
class KookCardMessage(BaseModel):
modules: list[AnyModule] = field(default_factory=list)
```
The union types are large, and they’re not leveraged for runtime validation (since the modules themselves are dataclasses, not pydantic models). You can reduce complexity by making the base class a proper protocol/model and typing `modules` as the base instead of an explicit union:
```python
from typing import Protocol
class KookCardModule(Protocol):
type: str
# All your dataclasses implicitly satisfy this protocol:
# @dataclass
# class HeaderModule(KookCardModule): ...
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
theme: ThemeType | None = None
size: SizeType | None = None
color: KookCardColor | None = None
modules: list[KookCardModule] = field(default_factory=list)
def add_module(self, module: KookCardModule):
self.modules.append(module)
```
If you don’t actually use `AnyElement` anywhere yet, you can similarly define an `Element` protocol (or just drop the alias where it isn’t needed) to avoid the large union in the type hints.
This preserves all existing classes and behavior, but removes the need to keep `AnyModule`/`AnyElement` synchronized with every new module/element you add.
### 2. Avoid subclassing `list` for `KookCardMessageContainer`
Subclassing `list` adds mental overhead for little benefit. You can make this a simple wrapper with an explicit `cards` field while keeping the same external behavior (`to_json` and `append`):
Current:
```python
class KookCardMessageContainer(list[KookCardMessage]):
"""卡片消息容器(列表),此类型可以直接to_json后发送出去"""
def append(self, object: KookCardMessage) -> None:
return super().append(object)
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps(
[i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii
)
```
Simplified:
```python
from dataclasses import dataclass, field
@dataclass
class KookCardMessageContainer:
cards: list[KookCardMessage] = field(default_factory=list)
def append(self, card: KookCardMessage) -> None:
self.cards.append(card)
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps(
[card.to_dict() for card in self.cards],
indent=indent,
ensure_ascii=ensure_ascii,
)
```
Existing call sites that did:
```python
container = KookCardMessageContainer()
container.append(card)
payload = container.to_json()
```
still work the same, but the type is now a simple, self-describing container instead of a specialized `list`.
</issue_to_address>
### Comment 4
<location path="astrbot/core/platform/sources/kook/config.py" line_range="6" />
<code_context>
+包含连接参数、重连策略、心跳设置等配置项
+"""
+
+# 连接配置
+CONNECTION_CONFIG = {
+ # 心跳配置
</code_context>
<issue_to_address>
**issue (complexity):** Consider consolidating the separate configuration dictionaries and getters into a single namespaced `CONFIG` with one access function to simplify the exposed configuration surface.
You can reduce the perceived/configuration surface complexity without removing any of the options by consolidating the multiple globals and getters into a single, namespaced config object.
Instead of exposing 5 separate dicts and 3+ getter functions, you can keep all keys but group them under one `CONFIG` with a single access API:
```python
# config.py
CONFIG = {
"connection": {
"heartbeat_interval": 30,
"heartbeat_timeout": 6,
"max_heartbeat_failures": 3,
"initial_reconnect_delay": 1,
"max_reconnect_delay": 60,
"max_consecutive_failures": 5,
"websocket_timeout": 10,
"connection_timeout": 30,
"enable_compression": True,
"max_message_size": 1024 * 1024,
},
"logging": {
"level": "INFO",
"format": "[KOOK] %(message)s",
"enable_heartbeat_logs": False,
"enable_message_logs": False,
},
"error_handling": {
"retry_on_network_error": True,
"retry_on_token_expired": True,
"max_retry_attempts": 3,
"retry_delay_base": 2,
},
"performance": {
"enable_message_buffering": True,
"buffer_size": 100,
"enable_connection_pooling": True,
"max_concurrent_requests": 10,
},
"security": {
"verify_ssl": True,
"enable_rate_limiting": True,
"rate_limit_requests": 100,
"rate_limit_window": 60,
},
}
def get_config(section: str | None = None):
"""获取配置；section 为 None 时返回完整配置"""
if section is None:
return CONFIG
return CONFIG[section]
```
Usage becomes:
```python
connection_cfg = get_config("connection")
heartbeat_interval = connection_cfg["heartbeat_interval"]
```
This keeps all current configuration options and values intact but:
- Shrinks the public API from multiple globals/getters to a single, easy-to-understand entry point.
- Reduces the conceptual overhead for readers: one config object with sections, instead of many top-level symbols.
- Makes it easier to gradually wire only the needed sections/keys into `KookClient`/`KookPlatformAdapter` without implying that each top-level dict is already fully honored.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This PR successfully integrates the KOOK adapter into AstrBot. However, a high-severity Path Traversal / Local File Inclusion (LFI) vulnerability was identified in the handling of card messages and asset uploads, where untrusted image sources from received messages could allow reading arbitrary files from the bot's host system. It is recommended to validate external URLs and restrict local file access. Furthermore, improvements are needed for hardcoded configurations, error handling, a serious base64 parsing bug, and some naming issues to enhance the code's robustness, maintainability, and configurability.
Contributor
Author
|
@Soulter 请问一下,如果我这边有关于适配器的单元测试,那这些单测应该放哪里呢,还是说单测先不放进来? |
Member
放到tests下就行了 |
Contributor
Author
那要不我开一个 |
Member
只包含 kook 相关测试吗？先放在 |
Contributor
Author
好的) |
Contributor
Author
|
@gemini-code-assist /gemini review |
Contributor
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些整体性的反馈：
- 在
KookClient.upload_asset中，用于记录target_path.as_posix()的except代码块在target_path被保证定义之前就会执行，这会额外抛出UnboundLocalError—— 建议单独定义一个原始路径字符串变量，并在异常处理程序中避免引用target_path。 KookConfig中定义并通过CONFIG_METADATA_2暴露的多个配置字段（例如reconnect_delay、max_reconnect_delay、max_retry_delay）实际上并没有在_main_loop或KookClient的重连逻辑中使用，目前的实现是硬编码的指数退避；建议要么把这些配置接入退避计算逻辑，要么移除它们，以避免产生令人困惑且无效的配置项。
给 AI Agents 的提示
Please address the comments from this code review:
## Overall Comments
- In `KookClient.upload_asset`, the `except` block that logs `target_path.as_posix()` executes before `target_path` is guaranteed to be defined, which will raise an additional `UnboundLocalError` — define a separate variable for the raw path string and avoid referencing `target_path` inside the exception handler.
- Several configuration fields defined in `KookConfig` and exposed via `CONFIG_METADATA_2` (e.g. `reconnect_delay`, `max_reconnect_delay`, `max_retry_delay`) are not actually used in the reconnect logic in `_main_loop` or `KookClient`, which currently hard-codes an exponential backoff; consider wiring these values into the backoff calculation or removing them to avoid confusing, ineffective settings.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="60-69" />
<code_context>
+ else:
+ file_url = file_url.removeprefix("file://")
+
+ try:
+ target_path = Path(file_url)
+ target_path = target_path.resolve()
+ except Exception as exp:
+ logger.error(
+ f'[KOOK] 获取文件 "{target_path.as_posix()}" 绝对路径失败: "{exp}"'
</code_context>
<issue_to_address>
**issue (bug_risk):** Exception handler in file:// branch uses target_path before it is guaranteed to be defined.
If `Path(file_url)` or `.resolve()` fails, `target_path` is never set, so the `except` block’s use of `target_path` will raise an `UnboundLocalError` and hide the original exception. Use `file_url` (or another variable defined before the `try`) in the log and user-facing message instead of `target_path` when path resolution fails.
</issue_to_address>
### Comment 2
<location path="tests/test_kook/shared.py" line_range="4" />
<code_context>
+from pathlib import Path
+
+
+TEST_DATA_DIR = Path("./tests/test_kook/data")
</code_context>
<issue_to_address>
**nitpick (testing):** Use a path relative to this file instead of the current working directory for test data
Using `Path("./tests/test_kook/data")` ties this to the repo root and can break when tests are run from other working directories. Prefer something like `TEST_DATA_DIR = Path(__file__).parent / "data"` so the path is resolved relative to this file instead.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="55" />
<code_context>
+ """16 进制色值"""
+
+
+class KookCardModelBase:
+ """卡片模块基类"""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider trimming the Kook card type definitions in this file down to only the currently used pieces (e.g., FileModule and a minimal KookCardMessage) and moving the full schema to a dedicated module for future use.
You can reduce complexity here by scoping the card model to only what’s currently needed, and deferring the full schema until it’s actually used.
Concretely:
1. **Keep only used module types in this file**
Right now, only `FileModule` is used. You can keep the rest of the structure the same, but trim unused module/element types and the broad union:
```python
# Keep
@dataclass
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
# Replace AnyModule with a minimal union for now
AnyModule = FileModule
```
This keeps the public surface compatible (`AnyModule` still exists) but avoids the cognitive load of unused modules.
2. **Move the full schema to a dedicated module when needed**
If you want to preserve the full type model for future work, move it out of this integration file into a separate module, and only import what’s needed:
```python
# kook_card_types.py
from pydantic.dataclasses import dataclass
class KookCardModelBase:
type: str
@dataclass
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
# keep other modules here but do not re-export them until they’re used
__all__ = ["FileModule"]
```
```python
# current file
from .kook_card_types import FileModule
AnyModule = FileModule # or re-import broader unions later when needed
```
3. **Optionally simplify `KookCardMessage` if only file/audio cards are used**
If the only current usage is sending a single audio/file module, you can simplify the model further:
```python
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
modules: list[FileModule] = field(default_factory=list)
def add_file(self, src: str, *, title: str = "", cover: str | None = None):
self.modules.append(FileModule(src=src, title=title, cover=cover))
```
This still supports the current behavior but expresses the actual constraints in use, and you can generalize back to `AnyModule` when more module types are added.
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续的评审。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- In
KookClient.upload_asset, theexceptblock that logstarget_path.as_posix()executes beforetarget_pathis guaranteed to be defined, which will raise an additionalUnboundLocalError— define a separate variable for the raw path string and avoid referencingtarget_pathinside the exception handler. - Several configuration fields defined in
KookConfigand exposed viaCONFIG_METADATA_2(e.g.reconnect_delay,max_reconnect_delay,max_retry_delay) are not actually used in the reconnect logic in_main_looporKookClient, which currently hard-codes an exponential backoff; consider wiring these values into the backoff calculation or removing them to avoid confusing, ineffective settings.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `KookClient.upload_asset`, the `except` block that logs `target_path.as_posix()` executes before `target_path` is guaranteed to be defined, which will raise an additional `UnboundLocalError` — define a separate variable for the raw path string and avoid referencing `target_path` inside the exception handler.
- Several configuration fields defined in `KookConfig` and exposed via `CONFIG_METADATA_2` (e.g. `reconnect_delay`, `max_reconnect_delay`, `max_retry_delay`) are not actually used in the reconnect logic in `_main_loop` or `KookClient`, which currently hard-codes an exponential backoff; consider wiring these values into the backoff calculation or removing them to avoid confusing, ineffective settings.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="60-69" />
<code_context>
+ else:
+ file_url = file_url.removeprefix("file://")
+
+ try:
+ target_path = Path(file_url)
+ target_path = target_path.resolve()
+ except Exception as exp:
+ logger.error(
+ f'[KOOK] 获取文件 "{target_path.as_posix()}" 绝对路径失败: "{exp}"'
</code_context>
<issue_to_address>
**issue (bug_risk):** Exception handler in file:// branch uses target_path before it is guaranteed to be defined.
If `Path(file_url)` or `.resolve()` fails, `target_path` is never set, so the `except` block’s use of `target_path` will raise an `UnboundLocalError` and hide the original exception. Use `file_url` (or another variable defined before the `try`) in the log and user-facing message instead of `target_path` when path resolution fails.
</issue_to_address>
### Comment 2
<location path="tests/test_kook/shared.py" line_range="4" />
<code_context>
+from pathlib import Path
+
+
+TEST_DATA_DIR = Path("./tests/test_kook/data")
</code_context>
<issue_to_address>
**nitpick (testing):** Use a path relative to this file instead of the current working directory for test data
Using `Path("./tests/test_kook/data")` ties this to the repo root and can break when tests are run from other working directories. Prefer something like `TEST_DATA_DIR = Path(__file__).parent / "data"` so the path is resolved relative to this file instead.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="55" />
<code_context>
+ """16 进制色值"""
+
+
+class KookCardModelBase:
+ """卡片模块基类"""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider trimming the Kook card type definitions in this file down to only the currently used pieces (e.g., FileModule and a minimal KookCardMessage) and moving the full schema to a dedicated module for future use.
You can reduce complexity here by scoping the card model to only what’s currently needed, and deferring the full schema until it’s actually used.
Concretely:
1. **Keep only used module types in this file**
Right now, only `FileModule` is used. You can keep the rest of the structure the same, but trim unused module/element types and the broad union:
```python
# Keep
@dataclass
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
# Replace AnyModule with a minimal union for now
AnyModule = FileModule
```
This keeps the public surface compatible (`AnyModule` still exists) but avoids the cognitive load of unused modules.
2. **Move the full schema to a dedicated module when needed**
If you want to preserve the full type model for future work, move it out of this integration file into a separate module, and only import what’s needed:
```python
# kook_card_types.py
from pydantic.dataclasses import dataclass
class KookCardModelBase:
type: str
@dataclass
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
# keep other modules here but do not re-export them until they’re used
__all__ = ["FileModule"]
```
```python
# current file
from .kook_card_types import FileModule
AnyModule = FileModule # or re-import broader unions later when needed
```
3. **Optionally simplify `KookCardMessage` if only file/audio cards are used**
If the only current usage is sending a single audio/file module, you can simplify the model further:
```python
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
modules: list[FileModule] = field(default_factory=list)
def add_file(self, src: str, *, title: str = "", cover: str | None = None):
self.modules.append(FileModule(src=src, title=title, cover=cover))
```
This still supports the current behavior but expresses the actual constraints in use, and you can generalize back to `AnyModule` when more module types are added.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Contributor
There was a problem hiding this comment.
Code Review
This PR successfully integrates the KOOK adapter into AstrBot's core code, demonstrating high implementation quality with complete connection, heartbeat, reconnection logic, and proper adaptation to KOOK message formats. The clear code structure and inclusion of test cases are excellent practices. A security audit identified sensitive information leakage where the bot's authentication token is logged at DEBUG level during initialization and at INFO level when logging the gateway URL. This credential exposure needs to be remediated by sanitizing the logged data. Additionally, I've identified a few areas for improvement: there's a risk of improper handling of falsy values in the configuration file parsing logic, duplicate metric reporting in the send_by_session method, and an unused attribute during adapter initialization. Specific suggestions for these points are provided in the code comments. All original comments remain valid as they do not contradict the provided rule regarding asyncio atomicity.
Contributor
Author
|
@gemini-code-assist /gemini review |
Contributor
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些总体反馈：
KookPlatformAdapter._main_loop中的重连逻辑目前是写死的指数退避（2**consecutive_failures，上限固定为 60 秒），并且忽略了KookConfig中的reconnect_delay/max_reconnect_delay，如果基于这些配置字段来计算退避时间，会更清晰也更便于控制。KookClient.connect暴露了resume参数并跟踪session_id/last_sn，但适配器从未调用过connect(resume=True)，也没有在重连流程中实现 resume 分支，这导致与 resume 相关的代码实际上处于死代码状态；建议要么在重连路径中接入 resume，要么移除未使用的参数/状态，以简化客户端。
面向 AI Agent 的提示
请根据这次代码评审中的评论进行修改：
## 总体评论
- `KookPlatformAdapter._main_loop` 中的重连逻辑目前是写死的指数退避（`2**consecutive_failures`，上限固定为 60 秒），并且忽略了 `KookConfig` 中的 `reconnect_delay` / `max_reconnect_delay`，如果基于这些配置字段来计算退避时间，会更清晰也更便于控制。
- `KookClient.connect` 暴露了 `resume` 参数并跟踪 `session_id` / `last_sn`，但适配器从未调用过 `connect(resume=True)`，也没有在重连流程中实现 resume 分支，这导致与 resume 相关的代码实际上处于死代码状态；建议要么在重连路径中接入 resume，要么移除未使用的参数/状态，以简化客户端。
## 具体评论
### 评论 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="258-259" />
<code_context>
+ """心跳循环"""
+ while self.running:
+ try:
+ # 随机化心跳间隔 (30±5秒)
+ interval = self.config.heartbeat_interval + random.randint(-5, 5)
+ await asyncio.sleep(interval)
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 随机化心跳间隔可能变为非正数，并且忽略了已配置的重连延迟。
随机化间隔 `config.heartbeat_interval + randint(-5, 5)` 在 `heartbeat_interval < 5` 时可能变为 0 或负数，从而导致接近忙循环的情况。可以通过将结果限制在一个最小值（例如 `interval = max(1, ...)`）来避免这个问题。另外，`KookConfig` 中的 `reconnect_delay` / `max_reconnect_delay` 从未被使用，即使 `_handle_hello` 的注释中提到了重连延迟；建议要么将它们整合进重连逻辑，要么如果不打算实现退避重连，就移除这些配置字段及相关注释，以避免困惑。
建议实现：
```python
async def _heartbeat_loop(self):
"""心跳循环"""
while self.running:
try:
# 随机化心跳间隔 (基于配置的 heartbeat_interval，加入 ±5 秒抖动)，并保证为正数
base_interval = getattr(self.config, "heartbeat_interval", 30)
jitter = random.randint(-5, 5)
interval = max(1, base_interval + jitter)
await asyncio.sleep(interval)
import asyncio
import random
import aiofiles
```
```python
import aiohttp
import websockets
```
1. 检查重连逻辑（很可能在实现 `_handle_hello` 或 websocket 重连的地方），然后：
- 使用 `KookConfig.reconnect_delay` / `KookConfig.max_reconnect_delay` 来实现重连尝试之间递增的退避时间，或者
- 如果不需要退避重连，则移除这些配置字段以及所有引用它们的注释。
2. 如果在每次循环迭代中需要发送心跳负载（例如通过 websocket 发送），请确保发送调用放在 `_heartbeat_loop` 中的 `await asyncio.sleep(interval)` 之后。
</issue_to_address>
### 评论 2
<location path="astrbot/core/platform/sources/kook/kook_event.py" line_range="134-146" />
<code_context>
+ return handle_plain(index, "(met)all(met)")
+ case Reply():
+ return handle_plain(index, "", reply_id=message_component.id)
+ case Json():
+ json_data = message_component.data
+ # kook卡片json外层得是一个列表
+ if isinstance(json_data, dict):
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 当 Json 组件中已经包含 JSON 字符串时，存在二次编码的风险。
如果某个插件传入的是已经序列化好的 JSON 字符串（例如他们事先构建好的 KOOK 卡片），当前逻辑会再次对其执行 `json.dumps`，从而发送一个被加了引号的 JSON 字符串，而不是预期的卡片数组。可以考虑在检测到输入是 `str` 时直接透传，仅在输入为 dict/list（或其他 Python 结构）时再做封装/序列化，以避免二次编码。
```suggestion
case Json():
json_data = message_component.data
# 如果插件已经传入预序列化的 JSON 字符串（例如完整的 KOOK 卡片数组），直接透传，避免二次编码
if isinstance(json_data, str):
encoded_json = json_data
else:
# kook卡片json外层得是一个列表
if isinstance(json_data, dict):
json_data = [json_data]
# 考虑到kook可能会更改消息结构,为了能让插件开发者
# 自行根据kook文档描述填卡片json内容,故不做模型校验
# KookCardMessage().model_validate(message_component.data).to_json(),
encoded_json = json.dumps(json_data)
return handle_plain(
index,
text=encoded_json,
type=KookMessageType.CARD,
)
```
</issue_to_address>
### 评论 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="10" />
<code_context>
+from pydantic.dataclasses import dataclass
+
+
+class KookApiPaths:
+ """Kook Api 路径"""
+
</code_context>
<issue_to_address>
**issue (complexity):** 建议将核心的卡片/消息 API 与详细的模块定义拆分开来，并使用一个简单的 Protocol 来替代大的 union，这样主 Kook 卡片模型可以更小、更不与每个模块类型强耦合。
你可以在不损失任何功能的前提下，通过收紧“核心”接口并将 `KookCardMessage` 与具体模块类型解耦，来降低这个抽象层看起来的“重量感”。
### 1. 收窄核心模块接口
目前这个文件在同一个地方定义了所有模块 + union + literal，即使实际上只用到了其中一部分。你可以在保留现有功能的情况下，通过拆分“核心定义”和“扩展定义”，让这个文件更容易浏览。
例如：
```python
# kook_card_core.py (only what is used today)
from enum import IntEnum
from typing import Protocol, Literal
from pydantic import BaseModel, ConfigDict
from pydantic.dataclasses import dataclass
from dataclasses import field
import json
class KookApiPaths:
BASE_URL = "https://www.kookapp.cn"
API_VERSION_PATH = "/api/v3"
USER_ME = f"{BASE_URL}{API_VERSION_PATH}/user/me"
GATEWAY_INDEX = f"{BASE_URL}{API_VERSION_PATH}/gateway/index"
ASSET_CREATE = f"{BASE_URL}{API_VERSION_PATH}/asset/create"
CHANNEL_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/message/create"
DIRECT_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/direct-message/create"
class KookMessageType(IntEnum):
TEXT = 1
IMAGE = 2
VIDEO = 3
FILE = 4
AUDIO = 8
KMARKDOWN = 9
CARD = 10
SYSTEM = 255
ThemeType = Literal["primary", "success", "danger", "warning", "info", "secondary", "none", "invisible"]
SizeType = Literal["xs", "sm", "md", "lg"]
class BaseCardModule(Protocol):
type: str # structural typing; all your dataclass modules already have this
@dataclass
class FileModule:
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
theme: ThemeType | None = None
size: SizeType | None = None
color: str | None = None
modules: list[BaseCardModule] = field(default_factory=list)
def add_module(self, module: BaseCardModule) -> None:
self.modules.append(module)
def to_dict(self, exclude_none: bool = True):
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]):
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps([i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii)
@dataclass
class OrderMessage:
index: int
text: str
type: KookMessageType
reply_id: str | int = ""
```
然后把所有详细的模块 dataclass（`HeaderModule`、`SectionModule` 等）移到一个“扩展 DSL” 文件中：
```python
# kook_card_modules.py
from pydantic.dataclasses import dataclass
from .kook_card_core import BaseCardModule, ThemeType, SizeType
# 将你现有的所有 @dataclass 模块放到这里，保持原样：
# PlainTextElement, KmarkdownElement, ImageElement, ButtonElement,
# HeaderModule, SectionModule, ImageGroupModule, ContainerModule, etc.
```
未来任何需要更丰富卡片结构的代码可以从 `kook_card_modules` 中导入，而主适配器/事件代码只需要面对精简的 `kook_card_core` API。
### 2. 用结构化类型替代 `AnyModule`
因为你目前并没有对照 KOOK 的 schema 做严格校验，同时你希望保持灵活性，那么像 `AnyModule = HeaderModule | ... | InviteModule` 这样的大 union 带来的价值其实有限，反而让核心消息模型与每个具体模块类型形成强耦合。
改用一个简单的 Protocol，可以在不引入联合类型爆炸的情况下，保留类型检查的好处：
```python
from typing import Protocol
class BaseCardModule(Protocol):
type: str
# 如果以后需要，也可以在这里补充一些公共字段
```
然后：
```python
class KookCardMessage(BaseModel):
...
modules: list[BaseCardModule] = field(default_factory=list)
def add_module(self, module: BaseCardModule) -> None:
self.modules.append(module)
```
你现有的 dataclass 模块已经符合这个 Protocol（它们都有 `type: str`，并且可以通过 `model_dump` 序列化为 JSON），所以行为上不会有变化，但 `KookCardMessage` 不再与每一个具体模块类型紧紧绑在一起。
这两个改动可以在保持现有功能完整的前提下，让这个文件更轻量、更易理解和维护，也能让未来对单个模块的改动变得不那么“高风险”。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- The reconnection logic in
KookPlatformAdapter._main_loopcurrently hardcodes an exponential backoff (2**consecutive_failureswith a fixed 60s cap) and ignoresreconnect_delay/max_reconnect_delayfromKookConfig, so it would be clearer and more controllable to base the backoff calculation on those config fields. KookClient.connectexposes aresumeparameter and trackssession_id/last_sn, but the adapter never callsconnect(resume=True)nor implements a resume branch in the reconnection flow, which makes the resume-related code effectively dead; consider either wiring resume into the reconnect path or removing the unused parameter/state to simplify the client.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The reconnection logic in `KookPlatformAdapter._main_loop` currently hardcodes an exponential backoff (`2**consecutive_failures` with a fixed 60s cap) and ignores `reconnect_delay`/`max_reconnect_delay` from `KookConfig`, so it would be clearer and more controllable to base the backoff calculation on those config fields.
- `KookClient.connect` exposes a `resume` parameter and tracks `session_id`/`last_sn`, but the adapter never calls `connect(resume=True)` nor implements a resume branch in the reconnection flow, which makes the resume-related code effectively dead; consider either wiring resume into the reconnect path or removing the unused parameter/state to simplify the client.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="258-259" />
<code_context>
+ """心跳循环"""
+ while self.running:
+ try:
+ # 随机化心跳间隔 (30±5秒)
+ interval = self.config.heartbeat_interval + random.randint(-5, 5)
+ await asyncio.sleep(interval)
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Randomized heartbeat interval can go non‑positive and ignores configured reconnect delays.
The randomized interval `config.heartbeat_interval + randint(-5, 5)` can become 0 or negative when `heartbeat_interval < 5`, leading to effectively busy loops. It would help to clamp the result to a minimum (e.g. `interval = max(1, ...)`). Also, `reconnect_delay`/`max_reconnect_delay` in `KookConfig` are never used despite the `_handle_hello` comment mentioning reconnect delay; consider either integrating them into the reconnection logic or removing them to avoid confusion.
Suggested implementation:
```python
async def _heartbeat_loop(self):
"""心跳循环"""
while self.running:
try:
# 随机化心跳间隔 (基于配置的 heartbeat_interval，加入 ±5 秒抖动)，并保证为正数
base_interval = getattr(self.config, "heartbeat_interval", 30)
jitter = random.randint(-5, 5)
interval = max(1, base_interval + jitter)
await asyncio.sleep(interval)
import asyncio
import random
import aiofiles
```
```python
import aiohttp
import websockets
```
1. Review the reconnection logic (likely where `_handle_hello` or websocket reconnect is implemented) and either:
- Use `KookConfig.reconnect_delay` / `KookConfig.max_reconnect_delay` to implement an increasing backoff between reconnect attempts, or
- Remove these configuration fields and any comments referencing them, if reconnection backoff is not desired.
2. If a heartbeat payload needs to be sent on each loop iteration (e.g. via websocket send), ensure that send call is placed after the `await asyncio.sleep(interval)` in `_heartbeat_loop`.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_event.py" line_range="134-146" />
<code_context>
+ return handle_plain(index, "(met)all(met)")
+ case Reply():
+ return handle_plain(index, "", reply_id=message_component.id)
+ case Json():
+ json_data = message_component.data
+ # kook卡片json外层得是一个列表
+ if isinstance(json_data, dict):
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Double-encoding risk when sending Json components that already contain JSON strings.
If a plugin passes a pre‑serialized JSON string (e.g., a KOOK card they've already built), this logic will `json.dumps` it again and send a quoted JSON string instead of the expected card array. Consider detecting `str` and passing it through unchanged, and only wrapping/dumping when the input is a dict/list (or other Python structures) to avoid double‑encoding.
```suggestion
case Json():
json_data = message_component.data
# 如果插件已经传入预序列化的 JSON 字符串（例如完整的 KOOK 卡片数组），直接透传，避免二次编码
if isinstance(json_data, str):
encoded_json = json_data
else:
# kook卡片json外层得是一个列表
if isinstance(json_data, dict):
json_data = [json_data]
# 考虑到kook可能会更改消息结构,为了能让插件开发者
# 自行根据kook文档描述填卡片json内容,故不做模型校验
# KookCardMessage().model_validate(message_component.data).to_json(),
encoded_json = json.dumps(json_data)
return handle_plain(
index,
text=encoded_json,
type=KookMessageType.CARD,
)
```
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="10" />
<code_context>
+from pydantic.dataclasses import dataclass
+
+
+class KookApiPaths:
+ """Kook Api 路径"""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider splitting the core card/message API from the detailed module definitions and using a simple protocol instead of a large union so the main Kook card model stays smaller and less coupled to every module type.
You can reduce the perceived heaviness of this abstraction without losing any functionality by tightening the “core” surface and decoupling `KookCardMessage` from knowing every module type.
### 1. Narrow the core module surface
Right now this file defines all modules + unions + literals in one place, even though only a subset is used. You can keep the current functionality but make this file easier to scan by splitting *core* vs *extended* definitions.
For example:
```python
# kook_card_core.py (only what is used today)
from enum import IntEnum
from typing import Protocol, Literal
from pydantic import BaseModel, ConfigDict
from pydantic.dataclasses import dataclass
from dataclasses import field
import json
class KookApiPaths:
BASE_URL = "https://www.kookapp.cn"
API_VERSION_PATH = "/api/v3"
USER_ME = f"{BASE_URL}{API_VERSION_PATH}/user/me"
GATEWAY_INDEX = f"{BASE_URL}{API_VERSION_PATH}/gateway/index"
ASSET_CREATE = f"{BASE_URL}{API_VERSION_PATH}/asset/create"
CHANNEL_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/message/create"
DIRECT_MESSAGE_CREATE = f"{BASE_URL}{API_VERSION_PATH}/direct-message/create"
class KookMessageType(IntEnum):
TEXT = 1
IMAGE = 2
VIDEO = 3
FILE = 4
AUDIO = 8
KMARKDOWN = 9
CARD = 10
SYSTEM = 255
ThemeType = Literal["primary", "success", "danger", "warning", "info", "secondary", "none", "invisible"]
SizeType = Literal["xs", "sm", "md", "lg"]
class BaseCardModule(Protocol):
type: str # structural typing; all your dataclass modules already have this
@dataclass
class FileModule:
src: str
title: str = ""
type: Literal["file", "audio", "video"] = "file"
cover: str | None = None
class KookCardMessage(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
type: str = "card"
theme: ThemeType | None = None
size: SizeType | None = None
color: str | None = None
modules: list[BaseCardModule] = field(default_factory=list)
def add_module(self, module: BaseCardModule) -> None:
self.modules.append(module)
def to_dict(self, exclude_none: bool = True):
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]):
def to_json(self, indent: int | None = None, ensure_ascii: bool = True) -> str:
return json.dumps([i.to_dict() for i in self], indent=indent, ensure_ascii=ensure_ascii)
@dataclass
class OrderMessage:
index: int
text: str
type: KookMessageType
reply_id: str | int = ""
```
Then move all the detailed module dataclasses (`HeaderModule`, `SectionModule`, etc.) into an “extended DSL” file:
```python
# kook_card_modules.py
from pydantic.dataclasses import dataclass
from .kook_card_core import BaseCardModule, ThemeType, SizeType
# all the existing @dataclass modules here exactly as you have them:
# PlainTextElement, KmarkdownElement, ImageElement, ButtonElement,
# HeaderModule, SectionModule, ImageGroupModule, ContainerModule, etc.
```
Any future code that needs richer card structures can import from `kook_card_modules`, but the main adapter/event code only sees the compact `kook_card_core` API.
### 2. Replace `AnyModule` with structural typing
Because you’re not validating against KOOK’s schema and you want flexibility, a broad union like `AnyModule = HeaderModule | ... | InviteModule` doesn’t buy much and tightly couples the core message model to every concrete module type.
Switching to a simple protocol keeps typing goodness without the explosion of unions:
```python
from typing import Protocol
class BaseCardModule(Protocol):
type: str
# you can optionally add common attributes if you ever need them
```
And then:
```python
class KookCardMessage(BaseModel):
...
modules: list[BaseCardModule] = field(default_factory=list)
def add_module(self, module: BaseCardModule) -> None:
self.modules.append(module)
```
Your existing dataclass modules already conform to this protocol (they all have `type: str` and are JSON-serializable via `model_dump`), so no behavior changes, but `KookCardMessage` is no longer entangled with every specific module type.
These two changes keep all current features intact, but make the file much lighter to understand and maintain, and make future changes to individual modules feel less risky.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
shuiping233
force-pushed
the
feat/astrbot_plugin_kook_adapter
branch
from
4f6c261 to
e27a3ad
Compare
Contributor
Author
|
ok,修完bug了,经过手动测试是能用了,至于原有kook适配器代码的写的比较不是很令人愉悦的部分,后面再说吧) @Soulter 如果有空的话,请帮忙review一下,虽然我应该吧pr拆开的,毕竟代码量太多确实不好人工审核( |
shuiping233
force-pushed
the
feat/astrbot_plugin_kook_adapter
branch
from
ce40c8c to
cb2d83a
Compare
shuiping233
added a commit
to shuiping233/astrbot_plugin_kook_adapter
that referenced
this pull request
Mar 2, 2026
shuiping233
force-pushed
the
feat/astrbot_plugin_kook_adapter
branch
from
96c179c to
dace3f0
Compare
Member
|
上传文件会提示 “资源不存在”，这个你有遇到吗 |
Contributor
Author
资源不存在我遇到过,我这里的情况是,发了一个外联http的svg到hook那边,他乎返回这个,似乎是因为他那边解析不了svg,但是我这边是image类型的,file类型的不存在我还没遇到过) 
|
Soulter
approved these changes
Mar 3, 2026
shuiping233
commented
Mar 3, 2026
| b64_str = file_url.removeprefix("base64://") | ||
| bytes_data = base64.b64decode(b64_str) | ||
|
|
||
| elif file_url.startswith("file://") or os.path.exists(file_url): |
Contributor
Author
There was a problem hiding this comment.
这里我测试debug了一下,我这边插件放了一个File对象发送去,然后file_url的值是没有file://开头的, File.get_file()去掉了了file://头,假设这里激进一点有可能会有file://开头的url进来,那是不是还要处理一下可能的file:///三个斜杠的情况? astrbot有没有utils函数处理这个协议头来着)
chain = []
chain.extend(
[
Plain("测试发送文件"),
File(
"unknown",
file="D:/game/SteamLibrary/steamapps/common/RunningWithRifles/media/packages/vanilla/sounds/dancing_shot5.wav",
),
]
)
yield event.chain_result(chain)
Member
There was a problem hiding this comment.
在file_url = file_url.removeprefix("file://") 前加一个 file_url = file_url.removeprefix("file:///") 这样可以解决？
Member
There was a problem hiding this comment.
加协议头 file:/// 是历史原因了。。屎山。。onebot。。
Soulter
changed the title
shuiping233
commented
Mar 3, 2026
| "type": "string", | ||
| "hint": "必填项。从 KOOK 开发者平台获取的机器人 Token。", | ||
| }, | ||
| "kook_bot_nickname": { |
Contributor
Author
There was a problem hiding this comment.
机器人的名字要修改的话应该是在kook机器人控制台里改的,KookClient里有个get_bot_info能通过kook接口拿到自身bot_id和bot_name,我不知道为什么明明我拿了bot_name却没拿来用上,但是有bot_name的话应该就不用用户手动填kook_bot_nickname进来了吧,也省的用户改了kook那边机器人名称,但是忘了改kook适配器配置)
|
No docs changes were generated in this run (docs repo had no updates). Docs repo: AstrBotDevs/AstrBot-docs AI change summary (not committed):
Experimental bot notice:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
将astrbot_plugin_kook_adapter引入astrbot官方适配器目录中,
相关链接:
Modifications / 改动点
astrbot/core/platform/sources/kook中CONFIG_METADATA_2里加了kook适配器的相关配置项astrbot/core/platform/manager.py里实现了kook适配器的正常导入功能Screenshots or Test Results / 运行截图或测试结果
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Summary by Sourcery
添加一方 KOOK 平台适配器，并将其集成到 AstrBot 的平台系统和配置元数据中。
新功能：
测试：
Original summary in English
Summary by Sourcery
Add a first-party KOOK platform adapter and integrate it into AstrBot's platform system and configuration metadata.
New Features:
Tests:
新功能：
Original summary in English
Summary by Sourcery
添加一方 KOOK 平台适配器，并将其集成到 AstrBot 的平台系统和配置元数据中。
新功能：
测试：
Original summary in English
Summary by Sourcery
Add a first-party KOOK platform adapter and integrate it into AstrBot's platform system and configuration metadata.
New Features:
Tests: