qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
"""
|
|
企业微信平台适配器
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from ..platform_adapter import register_platform_adapter, BasePlatformAdapter, PlatformMetadata
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@register_platform_adapter(
|
|
name="wecom",
|
|
display_name="企业微信",
|
|
description="企业微信消息平台适配器，支持文本、图片、文件、语音消息",
|
|
support_streaming=False,
|
|
support_image=True,
|
|
support_file=True,
|
|
support_voice=True,
|
|
)
|
|
class WecomPlatformAdapter(BasePlatformAdapter):
|
|
"""企业微信平台适配器"""
|
|
|
|
def get_metadata(self) -> PlatformMetadata:
|
|
from ..platform_adapter import platform_registry
|
|
return platform_registry["wecom"]
|
|
|
|
def verify_signature(self, payload: dict, signature: str, timestamp: str) -> bool:
|
|
"""企业微信签名验证: SHA1(token + timestamp + nonce)"""
|
|
if not signature or not timestamp:
|
|
return True # 无签名要求默认通过
|
|
token = self.config.get("token", "")
|
|
encoding_aes_key = self.config.get("encoding_aes_key", "")
|
|
nonce = (encoding_aes_key or token or "defaultnonce")[:16]
|
|
tmp = "".join(sorted([token, timestamp, nonce]))
|
|
expected = hashlib.sha1(tmp.encode()).hexdigest()
|
|
return signature == expected
|
|
|
|
def parse_webhook(self, payload: dict) -> dict:
|
|
"""解析企业微信webhook payload为统一消息格式"""
|
|
msg_type = payload.get("MsgType", "text")
|
|
content = ""
|
|
content_type = "text"
|
|
|
|
if msg_type == "text":
|
|
content = payload.get("Text", {}).get("Content", "")
|
|
content_type = "text"
|
|
elif msg_type == "image":
|
|
content = payload.get("Image", {}).get("MediaId", "")
|
|
content_type = "image"
|
|
elif msg_type == "voice":
|
|
content = payload.get("Voice", {}).get("MediaId", "")
|
|
content_type = "voice"
|
|
elif msg_type == "video":
|
|
content = payload.get("Video", {}).get("MediaId", "")
|
|
content_type = "video"
|
|
elif msg_type == "file":
|
|
content = payload.get("FileName", "")
|
|
content_type = "file"
|
|
elif msg_type == "location":
|
|
loc = payload.get("Location", {})
|
|
content = json.dumps(loc, ensure_ascii=False)
|
|
content_type = "location"
|
|
elif msg_type == "event":
|
|
content = json.dumps({"event": payload.get("Event", ""), "event_key": payload.get("EventKey", "")})
|
|
content_type = "event"
|
|
else:
|
|
content = json.dumps(payload, ensure_ascii=False)
|
|
content_type = msg_type
|
|
|
|
return {
|
|
"content_type": content_type,
|
|
"content": content,
|
|
"sender_id": payload.get("FromUserName", ""),
|
|
"sender_name": payload.get("FromUserName", ""),
|
|
"message_id": str(payload.get("MsgId", "")),
|
|
"timestamp": int(payload.get("CreateTime", 0)),
|
|
"group_id": None,
|
|
"group_name": None,
|
|
"raw": payload,
|
|
}
|
|
|
|
def build_reply(self, channel_config: dict, content: str, reply_to: str = None) -> dict:
|
|
"""构建企业微信回复消息"""
|
|
webhook_key = channel_config.get("webhook_key", "")
|
|
return {
|
|
"api_url": f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={webhook_key}",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": {"msgtype": "text", "text": {"content": content}},
|
|
}
|