fix: add support for webhook qq official

libs/qq_official_api/api.py

@@ -0,0 +1,274 @@
+import time
+from quart import request
+import base64
+import binascii
+import httpx
+from quart import Quart
+import xml.etree.ElementTree as ET
+from typing import Callable, Dict, Any
+from pkg.platform.types import events as platform_events, message as platform_message
+import aiofiles
+from .qqofficialevent import QQOfficialEvent
+import json
+import hmac
+import base64
+import hashlib
+import traceback
+from cryptography.hazmat.primitives.asymmetric import ed25519
+from .qqofficialevent import QQOfficialEvent
+
+def handle_validation(body: dict, bot_secret: str):
+    
+    # bot正确的secert是32位的，此处仅为了适配演示demo
+    while len(bot_secret) < 32:
+        bot_secret = bot_secret * 2
+    bot_secret = bot_secret[:32]
+    # 实际使用场景中以上三行内容可清除
+
+    seed_bytes = bot_secret.encode()
+
+    signing_key = ed25519.Ed25519PrivateKey.from_private_bytes(seed_bytes)
+
+    msg = body['d']['event_ts'] + body['d']['plain_token']
+    msg_bytes = msg.encode()
+
+    signature = signing_key.sign(msg_bytes)
+
+    signature_hex = signature.hex()
+
+    response = {
+        "plain_token": body['d']['plain_token'],
+        "signature": signature_hex
+    }
+
+    return response
+
+class QQOfficialClient:
+    def __init__(self, secret: str, token: str, app_id: str):
+        self.app = Quart(__name__)
+        self.app.add_url_rule(
+            "/callback/command",
+            "handle_callback",
+            self.handle_callback_request,
+            methods=["GET", "POST"],
+        )
+        self.secret = secret
+        self.token = token
+        self.app_id = app_id
+        self._message_handlers = {
+        }
+        self.base_url = "https://api.sgroup.qq.com"
+        self.access_token = ""
+        self.access_token_expiry_time = None
+
+    async def check_access_token(self):
+        """检查access_token是否存在"""
+        if not self.access_token or await self.is_token_expired():
+            return False
+        return bool(self.access_token and self.access_token.strip())
+    
+    async def get_access_token(self):
+        """获取access_token"""
+        url = "https://bots.qq.com/app/getAppAccessToken"
+        async with httpx.AsyncClient() as client:
+            params = {
+                "appId":self.app_id,
+                "clientSecret":self.secret,
+            }
+            headers = {
+                "content-type":"application/json",
+            }
+            try:
+                response = await client.post(url,json=params,headers=headers)
+                if response.status_code == 200:
+                    response_data = response.json()
+                access_token = response_data.get("access_token")
+                expires_in = int(response_data.get("expires_in",7200))
+                self.access_token_expiry_time = time.time() + expires_in - 60
+                if access_token:
+                    self.access_token = access_token
+            except Exception as e:
+                raise Exception(f"获取access_token失败: {e}")
+
+
+    async def handle_callback_request(self):
+        """处理回调请求"""
+        try:
+            # 读取请求数据
+            body = await request.get_data()
+            payload = json.loads(body)
+
+            
+            # 验证是否为回调验证请求
+            if payload.get("op") == 13:
+                # 生成签名
+                response = handle_validation(payload, self.secret)
+
+                return response
+
+            if payload.get("op") == 0:
+                    message_data = await self.get_message(payload)
+                    if message_data:
+                        event = QQOfficialEvent.from_payload(message_data)
+                        await self._handle_message(event)
+            
+            return {"code": 0, "message": "success"}
+
+        except Exception as e:
+            traceback.print_exc()
+            return {"error": str(e)}, 400
+
+    
+
+    async def run_task(self, host: str, port: int, *args, **kwargs):
+        """启动 Quart 应用"""
+        await self.app.run_task(host=host, port=port, *args, **kwargs)
+
+    def on_message(self, msg_type: str):
+        """注册消息类型处理器"""
+
+        def decorator(func: Callable[[platform_events.Event], None]):
+            if msg_type not in self._message_handlers:
+                self._message_handlers[msg_type] = []
+            self._message_handlers[msg_type].append(func)
+            return func
+
+        return decorator
+
+    async def _handle_message(self, event:QQOfficialEvent):
+        """处理消息事件"""
+        msg_type = event.t
+        if msg_type in self._message_handlers:
+            for handler in self._message_handlers[msg_type]:
+                await handler(event)
+
+
+    async def get_message(self,msg:dict) -> Dict[str,Any]:
+        """获取消息"""
+        message_data = {
+            "t":  msg.get("t",{}),
+            "user_openid": msg.get("d",{}).get("author",{}).get("user_openid",{}),
+            "timestamp": msg.get("d",{}).get("timestamp",{}),
+            "d_author_id": msg.get("d",{}).get("author",{}).get("id",{}),
+            "content": msg.get("d",{}).get("content",{}),
+            "d_id": msg.get("d",{}).get("id",{}),
+            "id": msg.get("id",{}),
+            "channel_id": msg.get("d",{}).get("channel_id",{}),
+            "username": msg.get("d",{}).get("author",{}).get("username",{}),
+            "guild_id": msg.get("d",{}).get("guild_id",{}),
+            "member_openid": msg.get("d",{}).get("author",{}).get("openid",{}),
+            "group_openid": msg.get("d",{}).get("group_openid",{})
+        }
+        attachments = msg.get("d", {}).get("attachments", [])
+        image_attachments = [attachment['url'] for attachment in attachments if await self.is_image(attachment)]
+        image_attachments_type = [attachment['content_type'] for attachment in attachments if await self.is_image(attachment)]
+        if image_attachments:
+            message_data["image_attachments"] = image_attachments[0]
+            message_data["content_type"] = image_attachments_type[0]
+        else:
+            
+            message_data["image_attachments"] = None
+        
+        return message_data
+    
+
+    async def is_image(self,attachment:dict) -> bool:
+        """判断是否为图片附件"""
+        content_type = attachment.get("content_type","")
+        return content_type.startswith("image/")
+    
+    
+    async def send_private_text_msg(self,user_openid:str,content:str,msg_id:str):
+        """发送私聊消息"""
+        if not await self.check_access_token():
+             await self.get_access_token()
+
+        url = self.base_url + "/v2/users/" + user_openid + "/messages"
+        async with httpx.AsyncClient() as client:
+            headers = {
+                "Authorization": f"QQBot {self.access_token}",
+                "Content-Type": "application/json",
+            }
+            data = {
+                "content": content,
+                "msg_type": 0,
+                "msg_id": msg_id,
+            }
+            response = await client.post(url,headers=headers,json=data)
+            if response.status_code == 200:
+                return
+            else:
+                raise ValueError(response)
+    
+
+    async def send_group_text_msg(self,group_openid:str,content:str,msg_id:str):
+        """发送群聊消息"""
+        if not await self.check_access_token():
+            await self.get_access_token()
+
+        url = self.base_url + "/v2/groups/" + group_openid + "/messages"
+        async with httpx.AsyncClient() as client:
+            headers = {
+                "Authorization": f"QQBot {self.access_token}",
+                "Content-Type": "application/json",
+            }
+            data = {
+                "content": content,
+                "msg_type": 0,
+                "msg_id": msg_id,
+            }
+            response = await client.post(url,headers=headers,json=data)
+            if response.status_code == 200:
+                return
+            else:
+                raise Exception(response.read().decode())
+
+    async def send_channle_group_text_msg(self,channel_id:str,content:str,msg_id:str):
+        """发送频道群聊消息"""
+        if not await self.check_access_token():
+             await self.get_access_token()
+
+        url = self.base_url + "/channels/" + channel_id + "/messages"
+        async with httpx.AsyncClient() as client:
+            headers = {
+                "Authorization": f"QQBot {self.access_token}",
+                "Content-Type": "application/json",
+            }
+            params = {
+                "content": content,
+                "msg_type": 0,
+                "msg_id": msg_id,
+            }
+            response = await client.post(url,headers=headers,json=params)
+            if response.status_code == 200:
+                return True
+            else:
+                raise Exception(response)
+
+    async def send_channle_private_text_msg(self,guild_id:str,content:str,msg_id:str):
+        """发送频道私聊消息"""
+        if not await self.check_access_token():
+             await self.get_access_token()
+
+        url = self.base_url + "/dms/" + guild_id + "/messages"
+        async with httpx.AsyncClient() as client:
+            headers = {
+                "Authorization": f"QQBot {self.access_token}",
+                "Content-Type": "application/json",
+            }
+            params = {
+                "content": content,
+                "msg_type": 0,
+                "msg_id": msg_id,
+            }
+            response = await client.post(url,headers=headers,json=params)
+            if response.status_code == 200:
+                return True
+            else:
+                raise Exception(response)
+
+    async def is_token_expired(self):
+        """检查token是否过期"""
+        if self.access_token_expiry_time is None:
+            return True
+        return time.time() > self.access_token_expiry_time

libs/qq_official_api/qqofficialevent.py

@@ -0,0 +1,114 @@
+from typing import Dict, Any, Optional
+
+class QQOfficialEvent(dict):
+    @staticmethod
+    def from_payload(payload: Dict[str, Any]) -> Optional["QQOfficialEvent"]:
+        try:
+            event = QQOfficialEvent(payload)
+            return event
+        except KeyError:
+            return None
+        
+
+    @property
+    def t(self) -> str:
+        """
+        事件类型
+        """
+        return self.get("t", "")
+    
+    @property
+    def user_openid(self) -> str:
+        """
+        用户openid
+        """
+        return self.get("user_openid",{})
+    
+    @property
+    def timestamp(self) -> str:
+        """
+        时间戳
+        """
+        return self.get("timestamp",{})
+    
+    
+    @property
+    def d_author_id(self) -> str:
+        """
+        作者id
+        """
+        return self.get("id",{})
+    
+    @property
+    def content(self) -> str:
+        """
+        内容
+        """
+        return self.get("content",'')
+    
+    @property
+    def d_id(self) -> str:
+        """
+        d_id
+        """
+        return self.get("d_id",{})
+    
+    @property
+    def id(self) -> str:
+        """
+        消息id，msg_id
+        """
+        return self.get("id",{})
+    
+    @property
+    def channel_id(self) -> str:
+        """
+        频道id
+        """
+        return self.get("channel_id",{})
+    
+    @property
+    def username(self) -> str:
+        """
+        用户名
+        """
+        return self.get("username",{})
+    
+    @property
+    def guild_id(self) -> str:
+        """
+        频道id
+        """
+        return self.get("guild_id",{})
+    
+    @property
+    def member_openid(self) -> str:
+        """
+        成员openid
+        """
+        return self.get("openid",{})
+    
+    @property
+    def attachments(self) -> str:
+        """
+        附件url
+        """
+        url = self.get("image_attachments", "")
+        if url and not url.startswith("https://"):
+            url = "https://" + url
+        return url
+    
+    @property
+    def group_openid(self) -> str:
+        """
+        群组id
+        """
+        return self.get("group_openid",{})
+    
+    @property
+    def content_type(self) -> str:
+        """
+        文件类型
+        """
+        return self.get("content_type","")
+