refactor: 给kook适配器尽可能覆盖pydantic数据类,使其更容易获取数据和校验消息结构#5719
Open
shuiping233 wants to merge 2 commits intoAstrBotDevs:masterfrom
Open
refactor: 给kook适配器尽可能覆盖pydantic数据类,使其更容易获取数据和校验消息结构#5719shuiping233 wants to merge 2 commits intoAstrBotDevs:masterfrom
shuiping233 wants to merge 2 commits intoAstrBotDevs:masterfrom
Conversation
dosubot
bot
added
the
size:XL
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适配器进行了重大重构，通过全面引入Pydantic数据类来处理KOOK平台的WebSocket事件和API响应。这一改进旨在为适配器提供更强大的类型安全、自动数据验证和更清晰的数据结构定义，从而显著提升代码的健壮性、可维护性和开发效率。它简化了消息的解析和构造过程，使得适配器在处理复杂数据时更加可靠和易于理解。 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
|
|
Related Documentation Checked 1 published document(s) in 1 knowledge base(s). No updates required. How did I do? Any feedback? |
dosubot
bot
added
the
area:platform
![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 个问题，并给出了一些整体反馈：
- 在
_parse_card_message中，当KookCardMessageContainer.from_dict抛出ValidationError时，你目前只是记录了错误然后继续执行，这会导致card_list未定义并在运行时出错；建议在记录日志后立刻返回一个安全的兜底值（例如return [], ""）。 KookCardMessageContainer.from_dict当前返回的是一个裸的list[KookCardMessage]，而不是KookCardMessageContainer实例，这对于一个定义在该类型上的 classmethod 来说有些出乎意料；建议返回cls(card_list)，以便调用方始终获得容器类型。kookGatewayIndexData的命名（首字母小写）与其他 Pydantic 模型（如KookUserMeData等）不一致；建议重命名为KookGatewayIndexData以保持清晰和一致性。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_parse_card_message`, when `KookCardMessageContainer.from_dict` raises a `ValidationError` you only log the error and continue, which will leave `card_list` undefined and lead to a runtime error; consider returning a safe fallback (e.g., `return [], ""`) immediately after logging.
- `KookCardMessageContainer.from_dict` currently returns a bare `list[KookCardMessage]` rather than an instance of `KookCardMessageContainer`, which is surprising for a classmethod on that type; consider returning `cls(card_list)` so callers consistently get the container type.
- The naming of `kookGatewayIndexData` (lowercase leading letter) is inconsistent with the rest of the Pydantic models (`KookUserMeData`, etc.); consider renaming it to `KookGatewayIndexData` for clarity and consistency.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_adapter.py" line_range="276-277" />
<code_context>
+ try:
+ card_list = KookCardMessageContainer.from_dict(json.loads(content))
+
+ except pydantic.ValidationError as e:
+ logger.error(f"[KOOK] 解析卡片消息错误: {e}")
+ logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
</code_context>
<issue_to_address>
**issue (bug_risk):** Catching pydantic.ValidationError here will currently raise NameError and also leaves card_list undefined.
Consider importing `pydantic` in this module and returning early from `_parse_card_message` on validation failure (e.g. `return [], ""`) so later code doesn’t iterate over an undefined `card_list`. Alternatively, ensure the exception handling path always initializes `card_list` before it is used.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="253-254" />
<code_context>
- return json.dumps(self.to_dict(), indent=indent, ensure_ascii=ensure_ascii)
-
class KookCardMessageContainer(list[KookCardMessage]):
"""卡片消息容器(列表),此类型可以直接to_json后发送出去"""
</code_context>
<issue_to_address>
**suggestion (bug_risk):** from_dict on KookCardMessageContainer returns a plain list instead of an instance of cls, which can be surprising.
`from_dict` currently returns a plain `list[KookCardMessage]` (`[KookCardMessage.from_dict(item) for item in raw_data]`), which conflicts with being defined on `KookCardMessageContainer` and with the type hint/`to_json` behavior. This inconsistency can cause subtle type/behavior issues. Consider returning a `KookCardMessageContainer` instance instead, e.g. `cls(KookCardMessage.from_dict(item) for item in raw_data)`.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="220" />
<code_context>
# 所有模块的联合类型
-AnyModule = (
+AnyModule = Annotated[
HeaderModule
| SectionModule
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new KOOK models by removing unnecessary discriminated unions and base-class abstractions while keeping the same typing and behavior.
You can reduce the new abstraction/indirection without losing any of the new functionality.
### 1. Drop discriminated `AnyModule` while keeping enums/literals
You already have `KookModuleType` and per‑module `type: Literal[... ]`, so you don’t actually need a discriminated union here. Pydantic will still serialize/deserialize correctly as long as the field names/types match KOOK’s payloads.
Current:
```python
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]
```
This can be simplified to:
```python
AnyModule = (
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule
)
```
`KookCardMessage.modules: list[AnyModule]` will still validate/serialize fine because each concrete module has a fixed `type` literal; the only thing you lose is automatic discriminated union resolution, which you’re not relying on anywhere else.
### 2. Simplify `KookWebsocketEvent` by moving parsing to the client
The discriminated union + pre‑validator on `KookWebsocketEvent` introduces quite a bit of protocol‑specific cleverness into the type layer:
```python
class KookWebsocketEvent(KookBaseDataClass):
signal: KookMessageSignal = Field(..., alias="s")
data: Annotated[
KookMessageEvent | KookHelloEvent | KookResumeAckEvent | None,
Field(discriminator="signal"),
] = Field(None, alias="d")
@model_validator(mode="before")
@classmethod
def _inject_signal_into_data(cls, data: Any) -> Any:
...
```
Since you still match on `event.signal` in the client, you can simplify `KookWebsocketEvent` to treat `data` as an untyped payload and do the per‑signal parsing where you already branch:
```python
class KookWebsocketEvent(KookBaseDataClass):
signal: KookMessageSignal = Field(..., alias="s")
d: dict[str, Any] | None = Field(None, alias="d")
sn: int | None = None
```
Then, in your client code:
```python
async def _handle_signal(self, raw_event: dict[str, Any]) -> None:
event = KookWebsocketEvent.from_dict(raw_event)
match event.signal:
case KookMessageSignal.MESSAGE:
msg = KookMessageEvent.from_dict(event.d or {})
# handle message
case KookMessageSignal.HELLO:
hello = KookHelloEvent.from_dict(event.d or {})
# handle hello
case KookMessageSignal.RESUME_ACK:
resume = KookResumeAckEvent.from_dict(event.d or {})
# handle resume-ack
# ...
```
You retain the same strong typing for `KookMessageEvent`, `KookHelloEvent`, `KookResumeAckEvent`, but the WebSocket envelope becomes a straightforward representation of the raw protocol, without discriminator plumbing or mutation in a pre‑validator.
### 3. Narrow the use of `KookBaseDataClass` to where it adds value
Right now `KookBaseDataClass` mostly renames `model_validate`/`model_dump`/`model_dump_json` and turns on `extra="allow"` globally:
```python
class KookBaseDataClass(BaseModel):
model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
@classmethod
def from_dict(cls, raw_data: dict):
return cls.model_validate(raw_data)
@classmethod
def from_json(cls, raw_data: str | bytes | bytearray):
return cls.model_validate_json(raw_data)
def to_dict(self, by_alias=False, exclude_none=True) -> dict:
return self.model_dump(by_alias=by_alias, exclude_none=exclude_none)
def to_json(...):
return self.model_dump_json(...)
```
If you want to keep the friendlier `from_*`/`to_*` names (they’re convenient for HTTP/WS payloads), consider:
* Using `KookBaseDataClass` only for **API/WS payload models** (`KookMessageEvent`, `KookUserMeResponse`, etc.).
* Letting card models (`KookCardMessage`, `KookCardModelBase` and friends) inherit directly from `BaseModel` since they already have simple usage and don’t need `extra="allow"`:
```python
class KookCardModelBase(BaseModel):
type: str
class KookCardMessage(BaseModel):
type: Literal[KookModuleType.CARD] = KookModuleType.CARD
# ...
```
This reduces one layer of inheritance/indirection in the hot path for card handling, while keeping the convenience helpers where they actually simplify call‑sites (HTTP and WebSocket parsing).
</issue_to_address>
### Comment 4
<location path="astrbot/core/platform/sources/kook/kook_adapter.py" line_range="268" />
<code_context>
- def _parse_card_message(self, data: dict) -> tuple[list, str]:
- content = data.get("content", "[]")
+ def _parse_card_message(self, data: KookMessageEvent) -> tuple[list, str]:
+ content = data.content
if not isinstance(content, str):
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying `_parse_card_message` to a single linear traversal with a generic image-extraction helper so the hot path depends only on a few core fields instead of many module-specific helpers and pattern matches.
You can keep the typed models and still reduce complexity/coupling a bit in the hot path with small, local changes.
### 1. Make `_parse_card_message` robust and more linear
Right now, if `KookCardMessageContainer.from_dict` fails, `card_list` is undefined and the function continues. Also, the pattern-matching + helpers cause a lot of jumping for a very narrow extraction (text + image URLs).
You can keep the typed models but make the traversal more linear and self‑contained:
```python
def _parse_card_message(self, data: KookMessageEvent) -> tuple[list, str]:
content = data.content
if not isinstance(content, str):
content = str(content)
try:
card_list = KookCardMessageContainer.from_dict(json.loads(content))
except pydantic.ValidationError as e:
logger.error(f"[KOOK] 解析卡片消息错误: {e}")
logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
# 明确失败返回，避免 card_list 未定义
return [Plain(text="[卡片消息解析失败]")], "[卡片消息解析失败]"
text_parts: list[str] = []
images: list[str] = []
for card in card_list:
for module in card.modules:
module_type = module.type
if module_type == "section":
# inline 原来的 _handle_section_text
text = getattr(module, "text", None)
if isinstance(text, (KmarkdownElement, PlainTextElement)):
if text.content:
text_parts.append(text.content)
elif module_type in ("container", "image-group"):
# 复用一个简洁的图片提取 helper（见下）
images.extend(self._extract_image_urls(module.elements))
elif module_type == "header":
header_text = getattr(getattr(module, "text", None), "content", None)
if header_text:
text_parts.append(header_text)
else:
logger.debug(f"[KOOK] 跳过或未处理模块: {module_type}")
text = "".join(text_parts)
message: list = []
if text:
message.append(Plain(text=text))
for img_url in images:
message.append(Image(file=img_url))
return message, text
```
### 2. Replace two module‑specific helpers with one focused image extractor
Instead of `_handle_section_text` and `_handle_image_group(module: ContainerModule | ImageGroupModule)`, you can keep a single helper that only depends on the *elements*, not on the specific module classes. This keeps the adapter less tightly coupled to the full card schema:
```python
def _extract_image_urls(self, elements) -> list[str]:
"""从 elements 中提取合法图片 URL，不关心具体 module 类型。"""
valid_urls: list[str] = []
for el in elements or []:
image_src = getattr(el, "src", None)
if not isinstance(image_src, str):
logger.warning(
f'[KOOK] 处理卡片中的图片时发生错误,图片url "{image_src}" 应该为str类型, 而不是 "{type(image_src)}" '
)
continue
if not image_src.startswith(("http://", "https://")):
logger.warning(f"[KOOK] 屏蔽非http图片url: {image_src}")
continue
valid_urls.append(image_src)
return valid_urls
```
This way:
- The hot path only cares about `module.type`, `module.text.content`, and `elements[].src`, similar to the original dict-based version.
- You still benefit from typed models and validation, but the adapter logic is less tightly coupled to every specific module class.
- The number of helper methods shrinks and the control flow becomes easier to read without losing functionality.
</issue_to_address>帮我变得更有用！请在每条评论上点击 👍 或 👎，我会根据你的反馈改进后续的代码审查。
Original comment in English
Hey - I've found 4 issues, and left some high level feedback:
- In
_parse_card_message, whenKookCardMessageContainer.from_dictraises aValidationErroryou only log the error and continue, which will leavecard_listundefined and lead to a runtime error; consider returning a safe fallback (e.g.,return [], "") immediately after logging. KookCardMessageContainer.from_dictcurrently returns a barelist[KookCardMessage]rather than an instance ofKookCardMessageContainer, which is surprising for a classmethod on that type; consider returningcls(card_list)so callers consistently get the container type.- The naming of
kookGatewayIndexData(lowercase leading letter) is inconsistent with the rest of the Pydantic models (KookUserMeData, etc.); consider renaming it toKookGatewayIndexDatafor clarity and consistency.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_parse_card_message`, when `KookCardMessageContainer.from_dict` raises a `ValidationError` you only log the error and continue, which will leave `card_list` undefined and lead to a runtime error; consider returning a safe fallback (e.g., `return [], ""`) immediately after logging.
- `KookCardMessageContainer.from_dict` currently returns a bare `list[KookCardMessage]` rather than an instance of `KookCardMessageContainer`, which is surprising for a classmethod on that type; consider returning `cls(card_list)` so callers consistently get the container type.
- The naming of `kookGatewayIndexData` (lowercase leading letter) is inconsistent with the rest of the Pydantic models (`KookUserMeData`, etc.); consider renaming it to `KookGatewayIndexData` for clarity and consistency.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_adapter.py" line_range="276-277" />
<code_context>
+ try:
+ card_list = KookCardMessageContainer.from_dict(json.loads(content))
+
+ except pydantic.ValidationError as e:
+ logger.error(f"[KOOK] 解析卡片消息错误: {e}")
+ logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
</code_context>
<issue_to_address>
**issue (bug_risk):** Catching pydantic.ValidationError here will currently raise NameError and also leaves card_list undefined.
Consider importing `pydantic` in this module and returning early from `_parse_card_message` on validation failure (e.g. `return [], ""`) so later code doesn’t iterate over an undefined `card_list`. Alternatively, ensure the exception handling path always initializes `card_list` before it is used.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="253-254" />
<code_context>
- return json.dumps(self.to_dict(), indent=indent, ensure_ascii=ensure_ascii)
-
class KookCardMessageContainer(list[KookCardMessage]):
"""卡片消息容器(列表),此类型可以直接to_json后发送出去"""
</code_context>
<issue_to_address>
**suggestion (bug_risk):** from_dict on KookCardMessageContainer returns a plain list instead of an instance of cls, which can be surprising.
`from_dict` currently returns a plain `list[KookCardMessage]` (`[KookCardMessage.from_dict(item) for item in raw_data]`), which conflicts with being defined on `KookCardMessageContainer` and with the type hint/`to_json` behavior. This inconsistency can cause subtle type/behavior issues. Consider returning a `KookCardMessageContainer` instance instead, e.g. `cls(KookCardMessage.from_dict(item) for item in raw_data)`.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="220" />
<code_context>
# 所有模块的联合类型
-AnyModule = (
+AnyModule = Annotated[
HeaderModule
| SectionModule
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new KOOK models by removing unnecessary discriminated unions and base-class abstractions while keeping the same typing and behavior.
You can reduce the new abstraction/indirection without losing any of the new functionality.
### 1. Drop discriminated `AnyModule` while keeping enums/literals
You already have `KookModuleType` and per‑module `type: Literal[... ]`, so you don’t actually need a discriminated union here. Pydantic will still serialize/deserialize correctly as long as the field names/types match KOOK’s payloads.
Current:
```python
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]
```
This can be simplified to:
```python
AnyModule = (
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule
)
```
`KookCardMessage.modules: list[AnyModule]` will still validate/serialize fine because each concrete module has a fixed `type` literal; the only thing you lose is automatic discriminated union resolution, which you’re not relying on anywhere else.
### 2. Simplify `KookWebsocketEvent` by moving parsing to the client
The discriminated union + pre‑validator on `KookWebsocketEvent` introduces quite a bit of protocol‑specific cleverness into the type layer:
```python
class KookWebsocketEvent(KookBaseDataClass):
signal: KookMessageSignal = Field(..., alias="s")
data: Annotated[
KookMessageEvent | KookHelloEvent | KookResumeAckEvent | None,
Field(discriminator="signal"),
] = Field(None, alias="d")
@model_validator(mode="before")
@classmethod
def _inject_signal_into_data(cls, data: Any) -> Any:
...
```
Since you still match on `event.signal` in the client, you can simplify `KookWebsocketEvent` to treat `data` as an untyped payload and do the per‑signal parsing where you already branch:
```python
class KookWebsocketEvent(KookBaseDataClass):
signal: KookMessageSignal = Field(..., alias="s")
d: dict[str, Any] | None = Field(None, alias="d")
sn: int | None = None
```
Then, in your client code:
```python
async def _handle_signal(self, raw_event: dict[str, Any]) -> None:
event = KookWebsocketEvent.from_dict(raw_event)
match event.signal:
case KookMessageSignal.MESSAGE:
msg = KookMessageEvent.from_dict(event.d or {})
# handle message
case KookMessageSignal.HELLO:
hello = KookHelloEvent.from_dict(event.d or {})
# handle hello
case KookMessageSignal.RESUME_ACK:
resume = KookResumeAckEvent.from_dict(event.d or {})
# handle resume-ack
# ...
```
You retain the same strong typing for `KookMessageEvent`, `KookHelloEvent`, `KookResumeAckEvent`, but the WebSocket envelope becomes a straightforward representation of the raw protocol, without discriminator plumbing or mutation in a pre‑validator.
### 3. Narrow the use of `KookBaseDataClass` to where it adds value
Right now `KookBaseDataClass` mostly renames `model_validate`/`model_dump`/`model_dump_json` and turns on `extra="allow"` globally:
```python
class KookBaseDataClass(BaseModel):
model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
@classmethod
def from_dict(cls, raw_data: dict):
return cls.model_validate(raw_data)
@classmethod
def from_json(cls, raw_data: str | bytes | bytearray):
return cls.model_validate_json(raw_data)
def to_dict(self, by_alias=False, exclude_none=True) -> dict:
return self.model_dump(by_alias=by_alias, exclude_none=exclude_none)
def to_json(...):
return self.model_dump_json(...)
```
If you want to keep the friendlier `from_*`/`to_*` names (they’re convenient for HTTP/WS payloads), consider:
* Using `KookBaseDataClass` only for **API/WS payload models** (`KookMessageEvent`, `KookUserMeResponse`, etc.).
* Letting card models (`KookCardMessage`, `KookCardModelBase` and friends) inherit directly from `BaseModel` since they already have simple usage and don’t need `extra="allow"`:
```python
class KookCardModelBase(BaseModel):
type: str
class KookCardMessage(BaseModel):
type: Literal[KookModuleType.CARD] = KookModuleType.CARD
# ...
```
This reduces one layer of inheritance/indirection in the hot path for card handling, while keeping the convenience helpers where they actually simplify call‑sites (HTTP and WebSocket parsing).
</issue_to_address>
### Comment 4
<location path="astrbot/core/platform/sources/kook/kook_adapter.py" line_range="268" />
<code_context>
- def _parse_card_message(self, data: dict) -> tuple[list, str]:
- content = data.get("content", "[]")
+ def _parse_card_message(self, data: KookMessageEvent) -> tuple[list, str]:
+ content = data.content
if not isinstance(content, str):
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying `_parse_card_message` to a single linear traversal with a generic image-extraction helper so the hot path depends only on a few core fields instead of many module-specific helpers and pattern matches.
You can keep the typed models and still reduce complexity/coupling a bit in the hot path with small, local changes.
### 1. Make `_parse_card_message` robust and more linear
Right now, if `KookCardMessageContainer.from_dict` fails, `card_list` is undefined and the function continues. Also, the pattern-matching + helpers cause a lot of jumping for a very narrow extraction (text + image URLs).
You can keep the typed models but make the traversal more linear and self‑contained:
```python
def _parse_card_message(self, data: KookMessageEvent) -> tuple[list, str]:
content = data.content
if not isinstance(content, str):
content = str(content)
try:
card_list = KookCardMessageContainer.from_dict(json.loads(content))
except pydantic.ValidationError as e:
logger.error(f"[KOOK] 解析卡片消息错误: {e}")
logger.error(f"[KOOK] 原始消息内容: {data.to_json()}")
# 明确失败返回，避免 card_list 未定义
return [Plain(text="[卡片消息解析失败]")], "[卡片消息解析失败]"
text_parts: list[str] = []
images: list[str] = []
for card in card_list:
for module in card.modules:
module_type = module.type
if module_type == "section":
# inline 原来的 _handle_section_text
text = getattr(module, "text", None)
if isinstance(text, (KmarkdownElement, PlainTextElement)):
if text.content:
text_parts.append(text.content)
elif module_type in ("container", "image-group"):
# 复用一个简洁的图片提取 helper（见下）
images.extend(self._extract_image_urls(module.elements))
elif module_type == "header":
header_text = getattr(getattr(module, "text", None), "content", None)
if header_text:
text_parts.append(header_text)
else:
logger.debug(f"[KOOK] 跳过或未处理模块: {module_type}")
text = "".join(text_parts)
message: list = []
if text:
message.append(Plain(text=text))
for img_url in images:
message.append(Image(file=img_url))
return message, text
```
### 2. Replace two module‑specific helpers with one focused image extractor
Instead of `_handle_section_text` and `_handle_image_group(module: ContainerModule | ImageGroupModule)`, you can keep a single helper that only depends on the *elements*, not on the specific module classes. This keeps the adapter less tightly coupled to the full card schema:
```python
def _extract_image_urls(self, elements) -> list[str]:
"""从 elements 中提取合法图片 URL，不关心具体 module 类型。"""
valid_urls: list[str] = []
for el in elements or []:
image_src = getattr(el, "src", None)
if not isinstance(image_src, str):
logger.warning(
f'[KOOK] 处理卡片中的图片时发生错误,图片url "{image_src}" 应该为str类型, 而不是 "{type(image_src)}" '
)
continue
if not image_src.startswith(("http://", "https://")):
logger.warning(f"[KOOK] 屏蔽非http图片url: {image_src}")
continue
valid_urls.append(image_src)
return valid_urls
```
This way:
- The hot path only cares about `module.type`, `module.text.content`, and `elements[].src`, similar to the original dict-based version.
- You still benefit from typed models and validation, but the adapter logic is less tightly coupled to every specific module class.
- The number of helper methods shrinks and the control flow becomes easier to read without losing functionality.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
dosubot
bot
added
size:XXL
shuiping233
force-pushed
the
refactor/add_kook_event_data_type
branch
from
2eae5a7 to
e803d27
Compare
Contributor
Author
|
@sourcery-ai review |
Contributor
Author
|
@gemini-code-assist /gemini review |
Contributor
There was a problem hiding this comment.
Hey，我发现了 3 个问题，并给出了一些整体性的反馈：
- 在
_send_ping中，你使用s=KookMessageSignal.PING来构造KookWebsocketEvent；为避免依赖基于别名的初始化行为，建议改为传入signal=KookMessageSignal.PING，这样可以让代码在 Pydantic 配置变化时更加稳健。 - 在
get_bot_info和get_gateway_url的pydantic.ValidationError分支里，你在上下文管理器内部已经调用了await resp.json()消费了响应体，又在之后调用await resp.text()；这可能会失败或返回空数据。更安全的做法是先捕获一次原始 body，然后同时用于解析和错误日志。
供 AI Agent 使用的提示词
Please address the comments from this code review:
## Overall Comments
- 在 `_send_ping` 中，你使用 `s=KookMessageSignal.PING` 来构造 `KookWebsocketEvent`；为避免依赖基于别名的初始化行为，建议改为传入 `signal=KookMessageSignal.PING`，这样可以让代码在 Pydantic 配置变化时更加稳健。
- 在 `get_bot_info` 和 `get_gateway_url` 的 `pydantic.ValidationError` 分支里，你在上下文管理器内部已经调用了 `await resp.json()` 消费了响应体，又在之后调用 `await resp.text()`；这可能会失败或返回空数据。更安全的做法是先捕获一次原始 body，然后同时用于解析和错误日志。
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="97-98" />
<code_context>
+ logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
return bot_id
+ except pydantic.ValidationError as e:
+ logger.error(f"[KOOK] 获取机器人账号信息失败, 响应数据格式错误: \n{e}")
+ logger.error(f"[KOOK] 响应内容: {await resp.text()}")
+
</code_context>
<issue_to_address>
**issue (bug_risk):** 在 `ValidationError` 处理器中使用 `resp` 可能会抛出 `UnboundLocalError`，并且可能会尝试从已关闭的响应中读取数据。
在 `except pydantic.ValidationError` 块中你调用了 `await resp.text()`，但 `resp` 是在上面的 `async with` 块中创建的。如果在 `from_dict(await resp.json())` 之前或执行过程中发生了校验失败，`async with` 可能已经退出，此时 `resp` 要么根本没有绑定，要么已经被关闭，从而导致 `UnboundLocalError`，或者尝试从已关闭的流中读取数据。
更安全的模式是：在 `async with` 块内部读取并保存响应 body，然后再进行校验，这样原始文本在 `except` 处理中仍然可用，例如：
```python
raw_text = await resp.text()
try:
resp_content = KookUserMeResponse.from_json(raw_text)
except pydantic.ValidationError as e:
logger.error(...)
logger.error(f"[KOOK] 响应内容: {raw_text}")
return ""
```
同时应完全避免在 `async with` 作用域之外使用 `resp`。
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="337-342" />
<code_context>
try:
- ping_data = {"s": 2, "sn": self.last_sn}
- await self.ws.send(json.dumps(ping_data)) # type: ignore
+ ping_data = KookWebsocketEvent(
+ s=KookMessageSignal.PING,
+ d=None,
+ sn=self.last_sn,
+ )
+ await self.ws.send(ping_data.to_json())
except Exception as e:
logger.error(f"[KOOK] 发送心跳失败: {e}")
</code_context>
<issue_to_address>
**issue (bug_risk):** `KookWebsocketEvent` 构造函数是用 `s=` 调用的，而不是使用模型字段名，这很可能导致 `TypeError`。
在 `KookWebsocketEvent` 中，该字段定义如下：
```python
signal: KookMessageSignal = Field(..., alias="s")
```
在 Pydantic v2 中，构造模型时必须使用字段名（`signal`），而不是别名（`s`）。使用 `s=` 会抛出 `TypeError: __init__() got an unexpected keyword argument 's'`。
应将构造更新为：
```python
ping_data = KookWebsocketEvent(
signal=KookMessageSignal.PING,
data=None,
sn=self.last_sn,
)
```
在调用 `to_json()` 时，别名仍然会被正确应用。
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="190-194" />
<code_context>
while self.running:
try:
- msg = await asyncio.wait_for(self.ws.recv(), timeout=10) # type: ignore
+ if self.ws is None:
+ logger.warning(
+ "[KOOK] 尚未连接kook WebSocket服务器, 跳过消息监听流程"
+ )
+ continue
+ msg = await asyncio.wait_for(self.ws.recv(), timeout=10)
</code_context>
<issue_to_address>
**issue (performance):** 当 `self.ws` 为 `None` 时，监听循环会在没有任何延迟的情况下持续运行，导致频繁日志输出和不必要的 CPU 占用。
在这个分支中，`self.ws is None` 时会记录一条 warning 然后立即 `continue`，而此时 `running` 仍为 `True`，从而形成一个忙等循环，疯狂刷日志并浪费 CPU。可以在 `continue` 之前添加一个短暂的 `await asyncio.sleep(...)`，或者在这里直接跳出循环，让重连机制来处理。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进评审质量。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- In
_send_ping, you constructKookWebsocketEventwiths=KookMessageSignal.PING; to avoid relying on alias-based initialization behavior, consider passingsignal=KookMessageSignal.PINGinstead so the code is robust against Pydantic config changes. - In the
pydantic.ValidationErrorbranches ofget_bot_infoandget_gateway_url, you callawait resp.text()after having already consumedawait resp.json()inside the context manager; this can fail or return empty data, so it would be safer to capture the raw body once and reuse it for both parsing and error logging.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_send_ping`, you construct `KookWebsocketEvent` with `s=KookMessageSignal.PING`; to avoid relying on alias-based initialization behavior, consider passing `signal=KookMessageSignal.PING` instead so the code is robust against Pydantic config changes.
- In the `pydantic.ValidationError` branches of `get_bot_info` and `get_gateway_url`, you call `await resp.text()` after having already consumed `await resp.json()` inside the context manager; this can fail or return empty data, so it would be safer to capture the raw body once and reuse it for both parsing and error logging.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="97-98" />
<code_context>
+ logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
return bot_id
+ except pydantic.ValidationError as e:
+ logger.error(f"[KOOK] 获取机器人账号信息失败, 响应数据格式错误: \n{e}")
+ logger.error(f"[KOOK] 响应内容: {await resp.text()}")
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Using `resp` inside the `ValidationError` handler can raise an `UnboundLocalError` and may try to read from a closed response.
In the `except pydantic.ValidationError` block you call `await resp.text()`, but `resp` is created inside the `async with` block above. If validation fails before or during `from_dict(await resp.json())`, the `async with` may already have exited and `resp` may either not be bound at all or already be closed, causing an `UnboundLocalError` or attempts to read from a closed stream.
A safer pattern is to read and store the response body inside the `async with` block and then validate it, so the raw text is still available in the `except` handler, for example:
```python
raw_text = await resp.text()
try:
resp_content = KookUserMeResponse.from_json(raw_text)
except pydantic.ValidationError as e:
logger.error(...)
logger.error(f"[KOOK] 响应内容: {raw_text}")
return ""
```
Also avoid using `resp` outside the `async with` scope entirely.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="337-342" />
<code_context>
try:
- ping_data = {"s": 2, "sn": self.last_sn}
- await self.ws.send(json.dumps(ping_data)) # type: ignore
+ ping_data = KookWebsocketEvent(
+ s=KookMessageSignal.PING,
+ d=None,
+ sn=self.last_sn,
+ )
+ await self.ws.send(ping_data.to_json())
except Exception as e:
logger.error(f"[KOOK] 发送心跳失败: {e}")
</code_context>
<issue_to_address>
**issue (bug_risk):** The `KookWebsocketEvent` constructor is called with `s=` instead of the model field name, which is likely to raise a `TypeError`.
In `KookWebsocketEvent`, the field is defined as:
```python
signal: KookMessageSignal = Field(..., alias="s")
```
With Pydantic v2 you must use the field name (`signal`), not the alias (`s`), when constructing the model. Using `s=` will raise `TypeError: __init__() got an unexpected keyword argument 's'`.
Update the construction to:
```python
ping_data = KookWebsocketEvent(
signal=KookMessageSignal.PING,
data=None,
sn=self.last_sn,
)
```
The alias will still be applied when calling `to_json()`.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="190-194" />
<code_context>
while self.running:
try:
- msg = await asyncio.wait_for(self.ws.recv(), timeout=10) # type: ignore
+ if self.ws is None:
+ logger.warning(
+ "[KOOK] 尚未连接kook WebSocket服务器, 跳过消息监听流程"
+ )
+ continue
+ msg = await asyncio.wait_for(self.ws.recv(), timeout=10)
</code_context>
<issue_to_address>
**issue (performance):** When `self.ws` is `None`, the listen loop continues without any delay, causing a tight log loop and unnecessary CPU usage.
In this branch, `self.ws is None` logs a warning and immediately `continue`s while `running` stays `True`, creating a busy loop that spams logs and wastes CPU. Add a short `await asyncio.sleep(...)` before continuing, or exit the loop here and let a reconnect mechanism handle it.
</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
本次 PR 对 KOOK 适配器进行了大规模的重构，全面采用 Pydantic 数据类来处理 WebSocket 事件和 API 响应。这是一个非常出色的改进，极大地增强了代码的健壮性、可读性和可维护性。通过使用 Pydantic 的类型校验和序列化功能，代码变得更加清晰和可靠。
主要改动包括：
- 将所有 KOOK 的数据结构（如消息事件、卡片消息、API 响应等）都定义为 Pydantic 模型。
- 利用 Pydantic 的可辨识联合（discriminated unions）来优雅地处理不同类型的 WebSocket 事件。
- 简化了消息处理逻辑，直接访问模型属性而不是使用
get()方法，并增加了更健壮的错误处理。 - 相应地更新了测试用例，并增加了新的测试数据来覆盖新的数据模型。
整体来看，这是一次高质量的重构。我只发现了一些可以微调的小细节，已在代码中提出。感谢你的贡献！
Contributor
Author
|
@sourcery-ai review |
Contributor
Author
|
@gemini-code-assist /gemini review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些总体反馈：
- 忽略机器人自身发送消息的逻辑，从对比配置中的
bot_nickname改成了对比author_id与self.client.bot_id；如果你仍然需要或依赖基于昵称的过滤逻辑，建议要么保留之前的行为，要么在配置里明确废弃该选项，以避免出现静默的行为变更。 - 若干
BaseModel字段使用了可变默认值，比如roles: list[int] = []、mention: list[str] = []等；为了清晰起见并遵循 pydantic 的最佳实践，建议改用Field(default_factory=list)（例如KookAuthor、KookKMarkdown、KookExtra、KookUserMeData中的相关字段）。
给 AI Agent 的提示词
Please address the comments from this code review:
## Overall Comments
- The logic for ignoring self-sent messages has changed from comparing `bot_nickname` in config to comparing `author_id` with `self.client.bot_id`; if nickname-based filtering is still desired or relied upon, consider either preserving that behavior or explicitly deprecating the config option to avoid silent behavior changes.
- Several `BaseModel` fields use mutable defaults like `roles: list[int] = []`, `mention: list[str] = []`, etc.; for clarity and to follow pydantic best practices, consider switching these to `Field(default_factory=list)` (e.g. in `KookAuthor`, `KookKMarkdown`, `KookExtra`, `KookUserMeData`).
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="319-328" />
<code_context>
+ BROADCAST = "BROADCAST"
+
+
+class KookAuthor(KookBaseDataClass):
+ id: str
+ username: str
+ identify_num: str
+ nickname: str
+ bot: bool
+ online: bool
+ avatar: str | None = None
+ vip_avatar: str | None = None
+ status: int
+ roles: list[int] = []
+
+
</code_context>
<issue_to_address>
**suggestion:** Using mutable list literals as defaults in Pydantic models can be brittle; `Field(default_factory=list)` is safer.
For the fields using `[]` (e.g. `KookAuthor.roles`, `KookKMarkdown.mention_part`, `KookExtra.mention`), switch to `Field(default_factory=list)`. This avoids relying on Pydantic’s current deep-copy behavior and keeps the models safer and more portable (e.g. if later converted to dataclasses or used outside Pydantic).
Suggested implementation:
```python
class KookAuthor(KookBaseDataClass):
id: str
username: str
identify_num: str
nickname: str
bot: bool
online: bool
avatar: str | None = None
vip_avatar: str | None = None
status: int
roles: list[int] = Field(default_factory=list)
```
In the same file (`astrbot/core/platform/sources/kook/kook_types.py`), you should apply the same pattern to other list-typed fields that currently use `[]` as a default:
1. Change `KookKMarkdown.mention_part: list[...] = []` to `KookKMarkdown.mention_part: list[...] = Field(default_factory=list)`.
2. Change `KookExtra.mention: list[...] = []` to `KookExtra.mention: list[...] = Field(default_factory=list)`.
Make sure `Field` is imported from `pydantic` at the top (it already appears in the snippet you provided).
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="120" />
<code_context>
-class KookCardModelBase:
+class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the card module `type` definitions by using the `KookModuleType` enum directly as the discriminator instead of repeating `Enum + Literal[...] + default` for each model.
You can reduce a fair bit of boilerplate in the card models by removing the `Enum + Literal + default` duplication while keeping the discriminated union and all current behavior.
Pydantic v2 supports `Enum` values as discriminators directly, so you can:
* type `type` as `KookModuleType` instead of `Literal[...]`
* set the enum default as the value
* keep `Field(discriminator="type")` unchanged
This removes the triple‑specification (Enum member, `Literal[...]`, default) and simplifies `KookCardModelBase`.
### 1. Simplify `KookCardModelBase`
```python
class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
type: KookModuleType
```
### 2. Simplify each module’s `type` field
Example: `PlainTextElement`, `SectionModule`, `FileModule`:
```python
class PlainTextElement(KookCardModelBase):
content: str
type: KookModuleType = KookModuleType.PLAIN_TEXT
emoji: bool = True
class SectionModule(KookCardModelBase):
text: PlainTextElement | KmarkdownElement | ParagraphStructure
type: KookModuleType = KookModuleType.SECTION
mode: SectionMode = "left"
accessory: ImageElement | ButtonElement | None = None
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: KookModuleType = KookModuleType.FILE
cover: str | None = None
"""cover 仅音频有效, 是音频的封面图"""
```
For `FileModule`, if you still want to accept `"file" | "audio" | "video"` on input, you can keep that in the enum (`FILE`, `AUDIO`, `VIDEO`) and let Pydantic coerce incoming strings to the enum; the discriminator will still work.
### 3. Keep the discriminated union unchanged
`AnyModule` can stay as is; it will work with `KookModuleType` directly:
```python
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]
```
This change:
* preserves the discriminated union behavior and serialization format,
* keeps `KookModuleType` (so you still have enum‑backed safety),
* but removes all the repeated `Literal[...] = EnumMember` boilerplate and the extra cognitive load of three sources of truth for each module type.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="69" />
<code_context>
+ def bot_username(self):
+ return self._bot_username
+
+ async def get_bot_info(self):
+ """获取机器人账号信息"""
url = KookApiPaths.USER_ME
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting a shared helper to parse HTTP responses into Pydantic models so both methods reuse the same validation and logging logic and avoid double-reading the body.
You can keep the stronger typing while reducing duplication and complexity around HTTP parsing by centralizing the Pydantic handling (and also avoid double‑reading the response body).
### 1. Factor out a generic HTTP → Pydantic helper
Right now:
- `get_bot_info`:
- does `await resp.json()`
- calls `KookUserMeResponse.from_dict(...)` in a nested try/except
- logs and then calls `await resp.text()` again in the exception path
- `get_gateway_url`:
- does `await resp.json()`
- calls `KookGatewayIndexResponse.from_dict(...)`
- has a separate `except pydantic.ValidationError` that also tries to log `await resp.text()`
This is a lot of ceremony per call, and double‑reading the body can be fragile.
You can extract a small helper that:
- reads the raw response body once,
- parses it into the given Pydantic model,
- does consistent logging for status/validation errors.
For example:
```python
# inside KookClient
async def _parse_response(
self,
resp: aiohttp.ClientResponse,
model: type[pydantic.BaseModel],
context: str,
):
raw = await resp.text()
if resp.status != 200:
logger.error(f"[KOOK] {context}失败，状态码: {resp.status} , {raw}")
return None
try:
return model.model_validate_json(raw)
except pydantic.ValidationError as e:
logger.error(f"[KOOK] {context}失败, 响应数据格式错误:\n{e}")
logger.error(f"[KOOK] 响应内容: {raw}")
return None
```
Then your call sites become much simpler and symmetric:
```python
async def get_bot_info(self):
url = KookApiPaths.USER_ME
try:
async with self._http_client.get(url) as resp:
resp_content = await self._parse_response(
resp, KookUserMeResponse, "获取机器人账号信息"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(
f"[KOOK] 获取机器人账号信息失败: {resp_content.model_dump_json()}"
)
return
self._bot_id = resp_content.data.id
self._bot_nickname = resp_content.data.nickname
self._bot_username = resp_content.data.username
logger.info(f"[KOOK] 获取机器人账号ID成功: {self._bot_id}")
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
except Exception as e:
logger.error(f"[KOOK] 获取机器人账号信息异常: {e}")
```
```python
async def get_gateway_url(self, resume=False, sn=0, session_id=None):
url = KookApiPaths.GATEWAY_INDEX
params = {}
if resume:
params["resume"] = 1
params["sn"] = sn
if session_id:
params["session_id"] = session_id
try:
async with self._http_client.get(url, params=params) as resp:
resp_content = await self._parse_response(
resp, KookGatewayIndexResponse, "获取gateway"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(f"[KOOK] 获取gateway失败: {resp_content}")
return None
gateway_url: str = resp_content.data.url
logger.info(f"[KOOK] 获取gateway成功: {gateway_url.split('?')[0]}")
return gateway_url
except Exception as e:
logger.error(f"[KOOK] 获取gateway异常: {e}")
return None
```
This keeps all the Pydantic validation and structured logging, but removes repetitive try/except blocks and error‑logging boilerplate from each API method, making the code easier to follow and change.帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据反馈改进之后的代码审查。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- The logic for ignoring self-sent messages has changed from comparing
bot_nicknamein config to comparingauthor_idwithself.client.bot_id; if nickname-based filtering is still desired or relied upon, consider either preserving that behavior or explicitly deprecating the config option to avoid silent behavior changes. - Several
BaseModelfields use mutable defaults likeroles: list[int] = [],mention: list[str] = [], etc.; for clarity and to follow pydantic best practices, consider switching these toField(default_factory=list)(e.g. inKookAuthor,KookKMarkdown,KookExtra,KookUserMeData).
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The logic for ignoring self-sent messages has changed from comparing `bot_nickname` in config to comparing `author_id` with `self.client.bot_id`; if nickname-based filtering is still desired or relied upon, consider either preserving that behavior or explicitly deprecating the config option to avoid silent behavior changes.
- Several `BaseModel` fields use mutable defaults like `roles: list[int] = []`, `mention: list[str] = []`, etc.; for clarity and to follow pydantic best practices, consider switching these to `Field(default_factory=list)` (e.g. in `KookAuthor`, `KookKMarkdown`, `KookExtra`, `KookUserMeData`).
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="319-328" />
<code_context>
+ BROADCAST = "BROADCAST"
+
+
+class KookAuthor(KookBaseDataClass):
+ id: str
+ username: str
+ identify_num: str
+ nickname: str
+ bot: bool
+ online: bool
+ avatar: str | None = None
+ vip_avatar: str | None = None
+ status: int
+ roles: list[int] = []
+
+
</code_context>
<issue_to_address>
**suggestion:** Using mutable list literals as defaults in Pydantic models can be brittle; `Field(default_factory=list)` is safer.
For the fields using `[]` (e.g. `KookAuthor.roles`, `KookKMarkdown.mention_part`, `KookExtra.mention`), switch to `Field(default_factory=list)`. This avoids relying on Pydantic’s current deep-copy behavior and keeps the models safer and more portable (e.g. if later converted to dataclasses or used outside Pydantic).
Suggested implementation:
```python
class KookAuthor(KookBaseDataClass):
id: str
username: str
identify_num: str
nickname: str
bot: bool
online: bool
avatar: str | None = None
vip_avatar: str | None = None
status: int
roles: list[int] = Field(default_factory=list)
```
In the same file (`astrbot/core/platform/sources/kook/kook_types.py`), you should apply the same pattern to other list-typed fields that currently use `[]` as a default:
1. Change `KookKMarkdown.mention_part: list[...] = []` to `KookKMarkdown.mention_part: list[...] = Field(default_factory=list)`.
2. Change `KookExtra.mention: list[...] = []` to `KookExtra.mention: list[...] = Field(default_factory=list)`.
Make sure `Field` is imported from `pydantic` at the top (it already appears in the snippet you provided).
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/kook/kook_types.py" line_range="120" />
<code_context>
-class KookCardModelBase:
+class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the card module `type` definitions by using the `KookModuleType` enum directly as the discriminator instead of repeating `Enum + Literal[...] + default` for each model.
You can reduce a fair bit of boilerplate in the card models by removing the `Enum + Literal + default` duplication while keeping the discriminated union and all current behavior.
Pydantic v2 supports `Enum` values as discriminators directly, so you can:
* type `type` as `KookModuleType` instead of `Literal[...]`
* set the enum default as the value
* keep `Field(discriminator="type")` unchanged
This removes the triple‑specification (Enum member, `Literal[...]`, default) and simplifies `KookCardModelBase`.
### 1. Simplify `KookCardModelBase`
```python
class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
type: KookModuleType
```
### 2. Simplify each module’s `type` field
Example: `PlainTextElement`, `SectionModule`, `FileModule`:
```python
class PlainTextElement(KookCardModelBase):
content: str
type: KookModuleType = KookModuleType.PLAIN_TEXT
emoji: bool = True
class SectionModule(KookCardModelBase):
text: PlainTextElement | KmarkdownElement | ParagraphStructure
type: KookModuleType = KookModuleType.SECTION
mode: SectionMode = "left"
accessory: ImageElement | ButtonElement | None = None
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: KookModuleType = KookModuleType.FILE
cover: str | None = None
"""cover 仅音频有效, 是音频的封面图"""
```
For `FileModule`, if you still want to accept `"file" | "audio" | "video"` on input, you can keep that in the enum (`FILE`, `AUDIO`, `VIDEO`) and let Pydantic coerce incoming strings to the enum; the discriminator will still work.
### 3. Keep the discriminated union unchanged
`AnyModule` can stay as is; it will work with `KookModuleType` directly:
```python
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]
```
This change:
* preserves the discriminated union behavior and serialization format,
* keeps `KookModuleType` (so you still have enum‑backed safety),
* but removes all the repeated `Literal[...] = EnumMember` boilerplate and the extra cognitive load of three sources of truth for each module type.
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/kook/kook_client.py" line_range="69" />
<code_context>
+ def bot_username(self):
+ return self._bot_username
+
+ async def get_bot_info(self):
+ """获取机器人账号信息"""
url = KookApiPaths.USER_ME
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting a shared helper to parse HTTP responses into Pydantic models so both methods reuse the same validation and logging logic and avoid double-reading the body.
You can keep the stronger typing while reducing duplication and complexity around HTTP parsing by centralizing the Pydantic handling (and also avoid double‑reading the response body).
### 1. Factor out a generic HTTP → Pydantic helper
Right now:
- `get_bot_info`:
- does `await resp.json()`
- calls `KookUserMeResponse.from_dict(...)` in a nested try/except
- logs and then calls `await resp.text()` again in the exception path
- `get_gateway_url`:
- does `await resp.json()`
- calls `KookGatewayIndexResponse.from_dict(...)`
- has a separate `except pydantic.ValidationError` that also tries to log `await resp.text()`
This is a lot of ceremony per call, and double‑reading the body can be fragile.
You can extract a small helper that:
- reads the raw response body once,
- parses it into the given Pydantic model,
- does consistent logging for status/validation errors.
For example:
```python
# inside KookClient
async def _parse_response(
self,
resp: aiohttp.ClientResponse,
model: type[pydantic.BaseModel],
context: str,
):
raw = await resp.text()
if resp.status != 200:
logger.error(f"[KOOK] {context}失败，状态码: {resp.status} , {raw}")
return None
try:
return model.model_validate_json(raw)
except pydantic.ValidationError as e:
logger.error(f"[KOOK] {context}失败, 响应数据格式错误:\n{e}")
logger.error(f"[KOOK] 响应内容: {raw}")
return None
```
Then your call sites become much simpler and symmetric:
```python
async def get_bot_info(self):
url = KookApiPaths.USER_ME
try:
async with self._http_client.get(url) as resp:
resp_content = await self._parse_response(
resp, KookUserMeResponse, "获取机器人账号信息"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(
f"[KOOK] 获取机器人账号信息失败: {resp_content.model_dump_json()}"
)
return
self._bot_id = resp_content.data.id
self._bot_nickname = resp_content.data.nickname
self._bot_username = resp_content.data.username
logger.info(f"[KOOK] 获取机器人账号ID成功: {self._bot_id}")
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
except Exception as e:
logger.error(f"[KOOK] 获取机器人账号信息异常: {e}")
```
```python
async def get_gateway_url(self, resume=False, sn=0, session_id=None):
url = KookApiPaths.GATEWAY_INDEX
params = {}
if resume:
params["resume"] = 1
params["sn"] = sn
if session_id:
params["session_id"] = session_id
try:
async with self._http_client.get(url, params=params) as resp:
resp_content = await self._parse_response(
resp, KookGatewayIndexResponse, "获取gateway"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(f"[KOOK] 获取gateway失败: {resp_content}")
return None
gateway_url: str = resp_content.data.url
logger.info(f"[KOOK] 获取gateway成功: {gateway_url.split('?')[0]}")
return gateway_url
except Exception as e:
logger.error(f"[KOOK] 获取gateway异常: {e}")
return None
```
This keeps all the Pydantic validation and structured logging, but removes repetitive try/except blocks and error‑logging boilerplate from each API method, making the code easier to follow and change.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
|
|
||
| class KookCardModelBase: | ||
| class KookCardModelBase(KookBaseDataClass): |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议直接使用 KookModuleType 枚举作为区分字段（discriminator），从而简化卡片模块中 type 字段的定义，避免在每个模型上重复维护 Enum + Literal[...] + default 三套信息。
通过去掉 Enum + Literal + default 这层重复，同时保持区分联合（discriminated union）和现有行为，可以显著减少卡片模型里的样板代码。
Pydantic v2 支持直接使用 Enum 值作为 discriminator，因此可以：
- 将
type的类型从Literal[...]改为KookModuleType - 用枚举值作为默认值
- 保持
Field(discriminator="type")不变
这样就删除了“枚举成员 + Literal[...] + 默认值”这三重声明，只保留更简单的 KookCardModelBase 定义。
1. 简化 KookCardModelBase
class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
type: KookModuleType2. 简化各模块的 type 字段
示例：PlainTextElement、SectionModule、FileModule：
class PlainTextElement(KookCardModelBase):
content: str
type: KookModuleType = KookModuleType.PLAIN_TEXT
emoji: bool = True
class SectionModule(KookCardModelBase):
text: PlainTextElement | KmarkdownElement | ParagraphStructure
type: KookModuleType = KookModuleType.SECTION
mode: SectionMode = "left"
accessory: ImageElement | ButtonElement | None = None
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: KookModuleType = KookModuleType.FILE
cover: str | None = None
"""cover 仅音频有效, 是音频的封面图"""对于 FileModule，如果你仍希望在输入上接受 "file" | "audio" | "video" 这样的字符串，可以把这些值保留在枚举中（FILE、AUDIO、VIDEO），然后让 Pydantic 来把传入字符串转换为枚举值；discriminator 仍然能正常工作。
3. 保持区分联合定义不变
AnyModule 可以保持原样，它可以直接与 KookModuleType 一起工作：
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]这一改动：
- 保留了区分联合的行为和序列化格式；
- 保留了
KookModuleType（仍然有枚举带来的类型安全）； - 但去掉了所有
Literal[...] = EnumMember形式的重复样板，减少了为每个模块维护三份“单一事实源”的负担。
Original comment in English
issue (complexity): Consider simplifying the card module type definitions by using the KookModuleType enum directly as the discriminator instead of repeating Enum + Literal[...] + default for each model.
You can reduce a fair bit of boilerplate in the card models by removing the Enum + Literal + default duplication while keeping the discriminated union and all current behavior.
Pydantic v2 supports Enum values as discriminators directly, so you can:
- type
typeasKookModuleTypeinstead ofLiteral[...] - set the enum default as the value
- keep
Field(discriminator="type")unchanged
This removes the triple‑specification (Enum member, Literal[...], default) and simplifies KookCardModelBase.
1. Simplify KookCardModelBase
class KookCardModelBase(KookBaseDataClass):
"""卡片模块基类"""
type: KookModuleType2. Simplify each module’s type field
Example: PlainTextElement, SectionModule, FileModule:
class PlainTextElement(KookCardModelBase):
content: str
type: KookModuleType = KookModuleType.PLAIN_TEXT
emoji: bool = True
class SectionModule(KookCardModelBase):
text: PlainTextElement | KmarkdownElement | ParagraphStructure
type: KookModuleType = KookModuleType.SECTION
mode: SectionMode = "left"
accessory: ImageElement | ButtonElement | None = None
class FileModule(KookCardModelBase):
src: str
title: str = ""
type: KookModuleType = KookModuleType.FILE
cover: str | None = None
"""cover 仅音频有效, 是音频的封面图"""For FileModule, if you still want to accept "file" | "audio" | "video" on input, you can keep that in the enum (FILE, AUDIO, VIDEO) and let Pydantic coerce incoming strings to the enum; the discriminator will still work.
3. Keep the discriminated union unchanged
AnyModule can stay as is; it will work with KookModuleType directly:
AnyModule = Annotated[
HeaderModule
| SectionModule
| ImageGroupModule
| ContainerModule
| ActionGroupModule
| ContextModule
| DividerModule
| FileModule
| CountdownModule
| InviteModule,
Field(discriminator="type"),
]This change:
- preserves the discriminated union behavior and serialization format,
- keeps
KookModuleType(so you still have enum‑backed safety), - but removes all the repeated
Literal[...] = EnumMemberboilerplate and the extra cognitive load of three sources of truth for each module type.
| def bot_username(self): | ||
| return self._bot_username | ||
|
|
||
| async def get_bot_info(self): |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议抽取一个通用的辅助方法，将 HTTP 响应解析为 Pydantic 模型，这样这两个方法可以复用相同的校验与日志逻辑，并避免对响应 body 的重复读取。
通过集中处理 Pydantic 相关逻辑，你可以在保持强类型的同时，降低 HTTP 解析逻辑的重复和复杂度（也顺便避免对响应 body 的二次读取）。
1. 抽取通用的 HTTP → Pydantic 辅助函数
当前情况：
get_bot_info：- 调用
await resp.json() - 在嵌套的 try/except 中调用
KookUserMeResponse.from_dict(...) - 出错时记录日志，然后在异常分支里再次调用
await resp.text()
- 调用
get_gateway_url：- 调用
await resp.json() - 调用
KookGatewayIndexResponse.from_dict(...) - 单独的
except pydantic.ValidationError中同样尝试await resp.text()来记录日志
- 调用
这样每个调用都需要写不少样板代码，而且对响应 body 的双重读取也比较脆弱。
你可以抽出一个小工具函数：
- 只读取一次原始响应 body；
- 使用给定的 Pydantic 模型进行解析；
- 对状态码或校验错误做一致的日志记录。
示例：
# inside KookClient
async def _parse_response(
self,
resp: aiohttp.ClientResponse,
model: type[pydantic.BaseModel],
context: str,
):
raw = await resp.text()
if resp.status != 200:
logger.error(f"[KOOK] {context}失败，状态码: {resp.status} , {raw}")
return None
try:
return model.model_validate_json(raw)
except pydantic.ValidationError as e:
logger.error(f"[KOOK] {context}失败, 响应数据格式错误:\n{e}")
logger.error(f"[KOOK] 响应内容: {raw}")
return None然后调用处就可以简化并保持对称：
async def get_bot_info(self):
url = KookApiPaths.USER_ME
try:
async with self._http_client.get(url) as resp:
resp_content = await self._parse_response(
resp, KookUserMeResponse, "获取机器人账号信息"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(
f"[KOOK] 获取机器人账号信息失败: {resp_content.model_dump_json()}"
)
return
self._bot_id = resp_content.data.id
self._bot_nickname = resp_content.data.nickname
self._bot_username = resp_content.data.username
logger.info(f"[KOOK] 获取机器人账号ID成功: {self._bot_id}")
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
except Exception as e:
logger.error(f"[KOOK] 获取机器人账号信息异常: {e}")async def get_gateway_url(self, resume=False, sn=0, session_id=None):
url = KookApiPaths.GATEWAY_INDEX
params = {}
if resume:
params["resume"] = 1
params["sn"] = sn
if session_id:
params["session_id"] = session_id
try:
async with self._http_client.get(url, params=params) as resp:
resp_content = await self._parse_response(
resp, KookGatewayIndexResponse, "获取gateway"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(f"[KOOK] 获取gateway失败: {resp_content}")
return None
gateway_url: str = resp_content.data.url
logger.info(f"[KOOK] 获取gateway成功: {gateway_url.split('?')[0]}")
return gateway_url
except Exception as e:
logger.error(f"[KOOK] 获取gateway异常: {e}")
return None这样既保留了 Pydantic 的校验和结构化日志，也去掉了每个 API 方法里重复的 try/except 和错误日志样板，使代码更易于阅读和修改。
Original comment in English
issue (complexity): Consider extracting a shared helper to parse HTTP responses into Pydantic models so both methods reuse the same validation and logging logic and avoid double-reading the body.
You can keep the stronger typing while reducing duplication and complexity around HTTP parsing by centralizing the Pydantic handling (and also avoid double‑reading the response body).
1. Factor out a generic HTTP → Pydantic helper
Right now:
get_bot_info:- does
await resp.json() - calls
KookUserMeResponse.from_dict(...)in a nested try/except - logs and then calls
await resp.text()again in the exception path
- does
get_gateway_url:- does
await resp.json() - calls
KookGatewayIndexResponse.from_dict(...) - has a separate
except pydantic.ValidationErrorthat also tries to logawait resp.text()
- does
This is a lot of ceremony per call, and double‑reading the body can be fragile.
You can extract a small helper that:
- reads the raw response body once,
- parses it into the given Pydantic model,
- does consistent logging for status/validation errors.
For example:
# inside KookClient
async def _parse_response(
self,
resp: aiohttp.ClientResponse,
model: type[pydantic.BaseModel],
context: str,
):
raw = await resp.text()
if resp.status != 200:
logger.error(f"[KOOK] {context}失败，状态码: {resp.status} , {raw}")
return None
try:
return model.model_validate_json(raw)
except pydantic.ValidationError as e:
logger.error(f"[KOOK] {context}失败, 响应数据格式错误:\n{e}")
logger.error(f"[KOOK] 响应内容: {raw}")
return NoneThen your call sites become much simpler and symmetric:
async def get_bot_info(self):
url = KookApiPaths.USER_ME
try:
async with self._http_client.get(url) as resp:
resp_content = await self._parse_response(
resp, KookUserMeResponse, "获取机器人账号信息"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(
f"[KOOK] 获取机器人账号信息失败: {resp_content.model_dump_json()}"
)
return
self._bot_id = resp_content.data.id
self._bot_nickname = resp_content.data.nickname
self._bot_username = resp_content.data.username
logger.info(f"[KOOK] 获取机器人账号ID成功: {self._bot_id}")
logger.info(f"[KOOK] 获取机器人名称成功: {self._bot_nickname}")
except Exception as e:
logger.error(f"[KOOK] 获取机器人账号信息异常: {e}")async def get_gateway_url(self, resume=False, sn=0, session_id=None):
url = KookApiPaths.GATEWAY_INDEX
params = {}
if resume:
params["resume"] = 1
params["sn"] = sn
if session_id:
params["session_id"] = session_id
try:
async with self._http_client.get(url, params=params) as resp:
resp_content = await self._parse_response(
resp, KookGatewayIndexResponse, "获取gateway"
)
if resp_content is None or not resp_content.success():
if resp_content is not None:
logger.error(f"[KOOK] 获取gateway失败: {resp_content}")
return None
gateway_url: str = resp_content.data.url
logger.info(f"[KOOK] 获取gateway成功: {gateway_url.split('?')[0]}")
return gateway_url
except Exception as e:
logger.error(f"[KOOK] 获取gateway异常: {e}")
return NoneThis keeps all the Pydantic validation and structured logging, but removes repetitive try/except blocks and error‑logging boilerplate from each API method, making the code easier to follow and change.
Contributor
There was a problem hiding this comment.
Code Review
This PR extensively refactors the KOOK adapter, fully introducing Pydantic data classes to handle WebSocket events and API responses. This is an excellent improvement, significantly enhancing code type safety, readability, and robustness, while reducing redundant code and improving error handling. A critical security issue was identified where sensitive information, such as verifyToken, could be leaked into logs during a validation error. Additionally, some minor issues were found that could make the code more concise and clear, with specific suggestions provided in the review comments. The rest of the changes appear to follow security best practices, including improved bot identification and robust card message parsing.
|
|
||
| except pydantic.ValidationError as e: | ||
| logger.error(f"[KOOK] 解析WebSocket事件数据格式失败: \n{e}") | ||
| logger.error(f"[KOOK] 原始响应内容: {msg}") |
Contributor
There was a problem hiding this comment.
The listen method logs the raw WebSocket message msg when a pydantic.ValidationError occurs. This raw message contains sensitive information such as the verifyToken and potentially the encryptKey in the top-level extra field. Logging these secrets can lead to unauthorized access if logs are compromised. It is recommended to sanitize the message by masking or removing sensitive fields before logging.
Contributor
Author
There was a problem hiding this comment.
有点屏蔽敏感内容太麻烦,而且适配器的其他消息在debug下也会打印原始响应内容,要改的话那要改的可多了.jpg
Contributor
Author
|
ok,这个pr我觉得应该是弄完了) |
shuiping233
force-pushed
the
refactor/add_kook_event_data_type
branch
2 times, most recently
from
bb10fb0 to
351ffb1
Compare
shuiping233
force-pushed
the
refactor/add_kook_event_data_type
branch
from
351ffb1 to
6633cf6
Compare
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
1 participant
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.
Modifications / 改动点
astrbot/core/platform/sources/kook, 将大部分ws消息事件和请求接口定义成pydantic数据类,并将大量字段get方法改成更好使用的类型判断tests/test_kook, pydantic的to_json行为和之前有点不同,所以要跟着修改一下测试用例(met)xx(met)格式字符串,导致机器人不响应携带了任何文件的的用户消息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.由 Sourcery 提供的总结
重构 KOOK 适配器和客户端，使用强类型的 Pydantic 模型来处理 WebSocket 事件、卡片结构和 REST 响应，从而改进消息解析、校验和日志记录，并更新测试以匹配新的序列化行为。
增强内容：
测试：
Original summary in English
Summary by Sourcery
Refactor the KOOK adapter and client to use strongly-typed Pydantic models for WebSocket events, card structures, and REST responses, improving message parsing, validation, and logging while updating tests to match the new serialization behavior.
Enhancements:
Tests:
Original summary in English
由 Sourcery 提供的总结
重构 KOOK 适配器和客户端，使用强类型的 Pydantic 模型来处理 WebSocket 事件、卡片结构和 REST 响应，从而改进消息解析、校验和日志记录，并更新测试以匹配新的序列化行为。
增强内容：
测试：
Original summary in English
Summary by Sourcery
Refactor the KOOK adapter and client to use strongly-typed Pydantic models for WebSocket events, card structures, and REST responses, improving message parsing, validation, and logging while updating tests to match the new serialization behavior.
Enhancements:
Tests:
Original summary in English
由 Sourcery 提供的总结
重构 KOOK 适配器和客户端，使用强类型的 Pydantic 模型来处理 WebSocket 事件、卡片结构和 REST 响应，从而改进消息解析、校验和日志记录，并更新测试以匹配新的序列化行为。
增强内容：
测试：
Original summary in English
Summary by Sourcery
Refactor the KOOK adapter and client to use strongly-typed Pydantic models for WebSocket events, card structures, and REST responses, improving message parsing, validation, and logging while updating tests to match the new serialization behavior.
Enhancements:
Tests:
Original summary in English
由 Sourcery 提供的总结
重构 KOOK 适配器和客户端，使用强类型的 Pydantic 模型来处理 WebSocket 事件、卡片结构和 REST 响应，从而改进消息解析、校验和日志记录，并更新测试以匹配新的序列化行为。
增强内容：
测试：
Original summary in English
Summary by Sourcery
Refactor the KOOK adapter and client to use strongly-typed Pydantic models for WebSocket events, card structures, and REST responses, improving message parsing, validation, and logging while updating tests to match the new serialization behavior.
Enhancements:
Tests: