qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
202 lines
8.1 KiB
Python
202 lines
8.1 KiB
Python
"""
|
|
WhatsApp平台适配器
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import time
|
|
from typing import Any
|
|
|
|
from ..platform_adapter import register_platform_adapter, BasePlatformAdapter, PlatformMetadata
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@register_platform_adapter(
|
|
name="whatsapp",
|
|
display_name="WhatsApp",
|
|
description="WhatsApp Cloud API消息适配器，支持文本、图片、文件、音频、视频、位置等消息类型",
|
|
support_streaming=False,
|
|
support_image=True,
|
|
support_file=True,
|
|
support_voice=True,
|
|
)
|
|
class WhatsAppPlatformAdapter(BasePlatformAdapter):
|
|
"""WhatsApp平台适配器"""
|
|
|
|
def get_metadata(self) -> PlatformMetadata:
|
|
from ..platform_adapter import platform_registry
|
|
return platform_registry["whatsapp"]
|
|
|
|
def verify_signature(self, payload: dict, signature: str, timestamp: str) -> bool:
|
|
"""WhatsApp签名验证: HMAC-SHA256(AppSecret, payload)"""
|
|
if not signature:
|
|
# WhatsApp可使用 verify_token 进行webhook验证
|
|
app_secret = self.config.get("app_secret", "")
|
|
if app_secret:
|
|
return False
|
|
return True # 无签名默认通过
|
|
|
|
app_secret = self.config.get("app_secret", "")
|
|
if not app_secret:
|
|
logger.warning("[WhatsApp] 未配置app_secret，跳过签名验证")
|
|
return True
|
|
|
|
try:
|
|
# WhatsApp使用 SHA256 HMAC 验证payload
|
|
body_str = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
|
|
expected = "sha256=" + hmac.new(
|
|
app_secret.encode("utf-8"),
|
|
body_str.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
return hmac.compare_digest(expected, signature)
|
|
except Exception as e:
|
|
logger.error(f"[WhatsApp] 签名验证异常: {e}")
|
|
return False
|
|
|
|
def parse_webhook(self, payload: dict) -> dict:
|
|
"""解析WhatsApp Cloud API webhook payload为统一消息格式
|
|
|
|
WhatsApp Cloud API格式:
|
|
{
|
|
"entry": [{
|
|
"changes": [{
|
|
"value": {
|
|
"messaging_product": "whatsapp",
|
|
"metadata": {"phone_number_id": "..."},
|
|
"messages": [{"from": "...", "type": "text", "text": {"body": "..."}}]
|
|
}
|
|
}]
|
|
}]
|
|
}
|
|
"""
|
|
entries = payload.get("entry", [])
|
|
|
|
content = ""
|
|
content_type = "text"
|
|
sender_id = ""
|
|
sender_name = ""
|
|
message_id = ""
|
|
timestamp = int(time.time())
|
|
|
|
for entry in entries:
|
|
changes = entry.get("changes", [])
|
|
for change in changes:
|
|
value = change.get("value", {})
|
|
messages = value.get("messages", [])
|
|
metadata = value.get("metadata", {})
|
|
phone_number_id = metadata.get("phone_number_id", "")
|
|
|
|
for msg in messages:
|
|
msg_from = msg.get("from", "")
|
|
msg_type = msg.get("type", "text")
|
|
msg_id = msg.get("id", "")
|
|
|
|
sender_id = msg_from
|
|
sender_name = msg_from
|
|
message_id = msg_id
|
|
timestamp = int(msg.get("timestamp", time.time()))
|
|
|
|
if msg_type == "text":
|
|
body = msg.get("text", {}).get("body", "")
|
|
content = body
|
|
content_type = "text"
|
|
elif msg_type == "image":
|
|
image = msg.get("image", {})
|
|
content = f"[图片] id={image.get('id', '')} caption={image.get('caption', '')}"
|
|
content_type = "image"
|
|
elif msg_type == "document":
|
|
doc = msg.get("document", {})
|
|
content = f"[文件] {doc.get('filename', 'unknown')} id={doc.get('id', '')}"
|
|
content_type = "file"
|
|
elif msg_type == "audio":
|
|
audio = msg.get("audio", {})
|
|
content = f"[语音] id={audio.get('id', '')}"
|
|
content_type = "voice"
|
|
elif msg_type == "video":
|
|
video = msg.get("video", {})
|
|
content = f"[视频] id={video.get('id', '')}"
|
|
content_type = "video"
|
|
elif msg_type == "location":
|
|
loc = msg.get("location", {})
|
|
content = f"[位置] lat={loc.get('latitude')} lng={loc.get('longitude')}"
|
|
content_type = "location"
|
|
elif msg_type == "contacts":
|
|
contacts = msg.get("contacts", [])
|
|
names = [c.get("name", {}).get("formatted_name", "") for c in contacts]
|
|
content = f"[联系人] {', '.join(names)}"
|
|
content_type = "contacts"
|
|
elif msg_type == "button":
|
|
button = msg.get("button", {})
|
|
content = f"[按钮] text={button.get('text', '')} payload={button.get('payload', '')}"
|
|
content_type = "button"
|
|
elif msg_type == "interactive":
|
|
interactive = msg.get("interactive", {})
|
|
list_reply = interactive.get("list_reply", {})
|
|
button_reply = interactive.get("button_reply", {})
|
|
if list_reply:
|
|
content = f"[列表回复] {list_reply.get('title', '')}: {list_reply.get('id', '')}"
|
|
elif button_reply:
|
|
content = f"[按钮回复] {button_reply.get('title', '')}: {button_reply.get('id', '')}"
|
|
else:
|
|
content = f"[交互] {json.dumps(interactive)}"
|
|
content_type = "interactive"
|
|
elif msg_type == "sticker":
|
|
sticker = msg.get("sticker", {})
|
|
content = f"[贴纸] id={sticker.get('id', '')}"
|
|
content_type = "sticker"
|
|
elif msg_type == "reaction":
|
|
reaction = msg.get("reaction", {})
|
|
content = f"[反应] emoji={reaction.get('emoji', '')}"
|
|
content_type = "reaction"
|
|
else:
|
|
content = json.dumps(msg, ensure_ascii=False)
|
|
content_type = msg_type
|
|
|
|
if not content:
|
|
# 可能是状态更新（无消息内容）
|
|
content = json.dumps(payload, ensure_ascii=False)
|
|
content_type = "status"
|
|
|
|
return {
|
|
"content_type": content_type,
|
|
"content": content,
|
|
"sender_id": sender_id,
|
|
"sender_name": sender_name,
|
|
"message_id": message_id,
|
|
"timestamp": timestamp,
|
|
"group_id": None, # WhatsApp主要是私聊
|
|
"group_name": None,
|
|
"raw": payload,
|
|
}
|
|
|
|
def build_reply(self, channel_config: dict, content: str, reply_to: str = None) -> dict:
|
|
"""构建WhatsApp回复消息"""
|
|
phone_number_id = channel_config.get("phone_number_id", "")
|
|
access_token = channel_config.get("access_token", "")
|
|
recipient = channel_config.get("recipient_id", "")
|
|
|
|
if not phone_number_id or not access_token:
|
|
logger.warning("[WhatsApp] 未配置phone_number_id或access_token，无法构建回复")
|
|
return {}
|
|
|
|
body = {
|
|
"messaging_product": "whatsapp",
|
|
"recipient_type": "individual",
|
|
"to": recipient,
|
|
"type": "text",
|
|
"text": {"preview_url": False, "body": content},
|
|
}
|
|
|
|
return {
|
|
"api_url": f"https://graph.facebook.com/v18.0/{phone_number_id}/messages",
|
|
"headers": {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {access_token}",
|
|
},
|
|
"body": body,
|
|
}
|