fix(qq): add configurable message format and onboard backfill

This commit is contained in:
Xubin Ren
2026-03-14 08:25:44 +00:00
parent 91d95f139e
commit af65145bc8
5 changed files with 110 additions and 12 deletions
+17 -11
View File
@@ -2,7 +2,7 @@
import asyncio
from collections import deque
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal
from loguru import logger
@@ -58,6 +58,7 @@ class QQConfig(Base):
app_id: str = ""
secret: str = ""
allow_from: list[str] = Field(default_factory=list)
msg_format: Literal["plain", "markdown"] = "plain"
class QQChannel(BaseChannel):
@@ -126,22 +127,27 @@ class QQChannel(BaseChannel):
try:
msg_id = msg.metadata.get("message_id")
self._msg_seq += 1
msg_type = self._chat_type_cache.get(msg.chat_id, "c2c")
if msg_type == "group":
use_markdown = self.config.msg_format == "markdown"
payload: dict[str, Any] = {
"msg_type": 2 if use_markdown else 0,
"msg_id": msg_id,
"msg_seq": self._msg_seq,
}
if use_markdown:
payload["markdown"] = {"content": msg.content}
else:
payload["content"] = msg.content
chat_type = self._chat_type_cache.get(msg.chat_id, "c2c")
if chat_type == "group":
await self._client.api.post_group_message(
group_openid=msg.chat_id,
msg_type=0,
content=msg.content,
msg_id=msg_id,
msg_seq=self._msg_seq,
**payload,
)
else:
await self._client.api.post_c2c_message(
openid=msg.chat_id,
msg_type=0,
content=msg.content,
msg_id=msg_id,
msg_seq=self._msg_seq,
**payload,
)
except Exception as e:
logger.error("Error sending QQ message: {}", e)
+17
View File
@@ -6,6 +6,7 @@ import select
import signal
import sys
from pathlib import Path
from typing import Any
# Force UTF-8 encoding for Windows console
if sys.platform == "win32":
@@ -259,6 +260,20 @@ def onboard():
console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
def _merge_missing_defaults(existing: Any, defaults: Any) -> Any:
"""Recursively fill in missing values from defaults without overwriting user config."""
if not isinstance(existing, dict) or not isinstance(defaults, dict):
return existing
merged = dict(existing)
for key, value in defaults.items():
if key not in merged:
merged[key] = value
else:
merged[key] = _merge_missing_defaults(merged[key], value)
return merged
def _onboard_plugins(config_path: Path) -> None:
"""Inject default config for all discovered channels (built-in + plugins)."""
import json
@@ -276,6 +291,8 @@ def _onboard_plugins(config_path: Path) -> None:
for name, cls in all_channels.items():
if name not in channels:
channels[name] = cls.default_config()
else:
channels[name] = _merge_missing_defaults(channels[name], cls.default_config())
with open(config_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)