Coverage for agentos/channels/adapters/qq.py: 0%

1""" 

2AgentOS Channels — QQ 机器人适配器。 

3 

4Webhook 规范: https://bot.q.qq.com/wiki/develop/api-v2/ 

5 

6特性: 

7 - WebSocket 长连接（QQ 官方推荐） + HTTP webhook 双模式 

8 - JSON 报文解析 

9 - Bot Token 管理 

10 - 主动推送 + 被动回复 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16import time 

17 

18import httpx 

19 

20from agentos.channels.base import BaseChannelAdapter, ChannelConfig, ReplyResult 

21from agentos.channels.message import ChannelMessage, ChannelType, MessageType 

22 

23 

24class QQAdapter(BaseChannelAdapter): 

25 """QQ 机器人适配器。""" 

26 

27 channel_type = ChannelType.QQ 

28 

29 def __init__(self, config: ChannelConfig): 

30 super().__init__(config) 

31 self._token: str = config.bot_token or "" 

32 self._token_expires: float = float("inf") # QQ Bot Token 由配置直接提供 

33 

34 # ── Webhook ── 

35 

36 def verify_signature(self, raw_body: bytes, headers: dict) -> bool: 

37 """QQ Bot 暂不强验证签名。""" 

38 return True 

39 

40 def parse_webhook(self, raw_body: bytes, headers: dict) -> ChannelMessage | list[ChannelMessage]: 

41 data = json.loads(raw_body.decode("utf-8")) 

42 op = data.get("op", 0) 

43 t = data.get("t", "") 

44 event_data = data.get("d", {}) 

45 

46 # 处理不同类型的 QQ 事件 

47 if op == 10: 

48 # Hello 事件 

49 return ChannelMessage( 

50 msg_id="hello", 

51 channel=ChannelType.QQ, 

52 msg_type=MessageType.EVENT, 

53 content="hello", 

54 extra={"op": 10, "heartbeat_interval": event_data.get("heartbeat_interval", 0)}, 

55 ) 

56 

57 if op == 11: 

58 # Heartbeat ACK 

59 return ChannelMessage( 

60 msg_id="heartbeat_ack", 

61 channel=ChannelType.QQ, 

62 msg_type=MessageType.EVENT, 

63 content="heartbeat_ack", 

64 extra={"op": 11}, 

65 ) 

66 

67 # op == 0: Dispatch 事件 

68 msg_map = { 

69 "AT_MESSAGE_CREATE": "text", 

70 "MESSAGE_CREATE": "text", 

71 "DIRECT_MESSAGE_CREATE": "text", 

72 "C2C_MESSAGE_CREATE": "text", 

73 } 

74 msg_type_str = msg_map.get(t, "text") 

75 msg_type_map = {"text": MessageType.TEXT} 

76 msg_type = msg_type_map.get(msg_type_str, MessageType.TEXT) 

77 

78 author = event_data.get("author", {}) 

79 content = event_data.get("content", "").strip() 

80 

81 # 去掉 @机器人 前缀 

82 if content.startswith("<@"): 

83 end = content.find(">") 

84 if end > 0: 

85 content = content[end + 1:].strip() 

86 

87 return ChannelMessage( 

88 msg_id=event_data.get("id", ""), 

89 channel=ChannelType.QQ, 

90 msg_type=msg_type, 

91 content=content, 

92 sender_id=author.get("id", ""), 

93 sender_name=author.get("username", ""), 

94 timestamp=float(time.time()), 

95 conversation_id=event_data.get("channel_id", event_data.get("guild_id", "")), 

96 reply_token="", 

97 extra={ 

98 "guild_id": event_data.get("guild_id"), 

99 "channel_id": event_data.get("channel_id"), 

100 "member": event_data.get("member"), 

101 "event_type": t, 

102 "is_group": bool(event_data.get("guild_id")), 

103 }, 

104 ) 

105 

106 def build_reply(self, msg: ChannelMessage, reply_text: str) -> str: 

107 return json.dumps({ 

108 "msg_type": 0, 

109 "content": reply_text, 

110 "msg_id": msg.msg_id, 

111 "message_reference": {"message_id": msg.msg_id}, 

112 }) 

113 

114 # ── 主动推送 ── 

115 

116 async def send_message(self, user_id: str, content: str, msg_type: str = "text") -> ReplyResult: 

117 token = await self.get_access_token() 

118 # QQ Bot 发消息需要知道 channel_id 

119 channel_id = self.config.extra.get("channel_id", "") 

120 if not channel_id: 

121 return ReplyResult(success=False, error="no channel_id in config") 

122 

123 url = f"https://api.sgroup.qq.com/channels/{channel_id}/messages" 

124 headers = {"Authorization": f"Bot {self.config.app_id}.{token}"} 

125 payload = {"content": content, "msg_type": 0} 

126 

127 async with httpx.AsyncClient() as client: 

128 resp = await client.post(url, json=payload, headers=headers, timeout=10) 

129 data = resp.json() 

130 msg_id = data.get("id", "") 

131 if msg_id: 

132 return ReplyResult(success=True, msg_id=msg_id) 

133 return ReplyResult(success=False, error=f"qq error: {data}") 

134 

135 async def send_c2c_message(self, user_id: str, content: str) -> ReplyResult: 

136 """发送私聊消息。""" 

137 token = await self.get_access_token() 

138 url = f"https://api.sgroup.qq.com/v2/users/{user_id}/messages" 

139 headers = {"Authorization": f"Bot {self.config.app_id}.{token}"} 

140 payload = {"content": content, "msg_type": 0} 

141 

142 async with httpx.AsyncClient() as client: 

143 resp = await client.post(url, json=payload, headers=headers, timeout=10) 

144 data = resp.json() 

145 return ReplyResult(success="id" in data, msg_id=data.get("id", "")) 

146 

147 async def send_image(self, user_id: str, image_url: str) -> ReplyResult: 

148 return await self.send_message(user_id, f"[图片] {image_url}") 

149 

150 async def send_file(self, user_id: str, file_url: str, filename: str) -> ReplyResult: 

151 return await self.send_message(user_id, f"文件: {filename}\n{file_url}") 

152 

153 # ── Token ── 

154 

155 async def get_access_token(self) -> str: 

156 return self._token or self.config.bot_token