qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
137 lines
5.2 KiB
Python
137 lines
5.2 KiB
Python
"""
|
|
飞书平台适配器
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
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="feishu",
|
|
display_name="飞书",
|
|
description="飞书开放平台消息适配器，支持文本、图片、文件、富文本消息",
|
|
support_streaming=True,
|
|
support_image=True,
|
|
support_file=True,
|
|
support_voice=False,
|
|
)
|
|
class FeishuPlatformAdapter(BasePlatformAdapter):
|
|
"""飞书平台适配器"""
|
|
|
|
def get_metadata(self) -> PlatformMetadata:
|
|
from ..platform_adapter import platform_registry
|
|
return platform_registry["feishu"]
|
|
|
|
def verify_signature(self, payload: dict, signature: str, timestamp: str) -> bool:
|
|
"""飞书签名验证: HMAC-SHA256"""
|
|
if not signature or not timestamp:
|
|
# 飞书首次URL验证(challenge)不需要签名验证
|
|
if payload.get("challenge"):
|
|
return True
|
|
return True # 宽松模式
|
|
|
|
app_secret = self.config.get("app_secret", "")
|
|
if not app_secret:
|
|
logger.warning("[Feishu] 未配置app_secret，跳过签名验证")
|
|
return True
|
|
|
|
# 飞书签名格式: timestamp + nonce + body
|
|
try:
|
|
body_str = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
|
|
nonce = timestamp # 飞书用timestamp作为nonce的一部分
|
|
sign_str = f"{timestamp}\n{nonce}\n{body_str}"
|
|
expected = hmac.new(
|
|
app_secret.encode("utf-8"),
|
|
sign_str.encode("utf-8"),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
return hmac.compare_digest(signature, expected)
|
|
except Exception as e:
|
|
logger.error(f"[Feishu] 签名验证异常: {e}")
|
|
return False
|
|
|
|
def parse_webhook(self, payload: dict) -> dict:
|
|
"""解析飞书webhook payload为统一消息格式"""
|
|
# 飞书URL验证challenge已在外部处理，这里处理正常消息
|
|
|
|
event = payload.get("event", {})
|
|
message = event.get("message", {})
|
|
msg_type = message.get("message_type", "text")
|
|
raw_content = message.get("content", "{}")
|
|
|
|
content = ""
|
|
content_type = "text"
|
|
sender_id = event.get("sender", {}).get("sender_id", {}).get("open_id", "")
|
|
sender_name = event.get("sender", {}).get("sender_id", {}).get("open_id", "")
|
|
message_id = event.get("header", {}).get("message_id", "")
|
|
|
|
try:
|
|
if isinstance(raw_content, str):
|
|
content_obj = json.loads(raw_content)
|
|
else:
|
|
content_obj = raw_content if isinstance(raw_content, dict) else {}
|
|
|
|
if msg_type == "text":
|
|
content = content_obj.get("text", "")
|
|
content_type = "text"
|
|
elif msg_type == "image":
|
|
image_key = message.get("image_key", "")
|
|
content = f"[图片] image_key={image_key}"
|
|
content_type = "image"
|
|
elif msg_type == "file":
|
|
file_key = message.get("file_key", "")
|
|
file_name = content_obj.get("file_name", "unknown")
|
|
content = f"[文件] {file_name} file_key={file_key}"
|
|
content_type = "file"
|
|
elif msg_type == "post":
|
|
# 富文本消息，提取text内容
|
|
post_content = content_obj.get("content", [])
|
|
text_parts = []
|
|
for paragraph in post_content:
|
|
if isinstance(paragraph, list):
|
|
for element in paragraph:
|
|
if isinstance(element, dict) and element.get("tag") == "text":
|
|
text_parts.append(element.get("text", ""))
|
|
content = "".join(text_parts)
|
|
content_type = "post"
|
|
else:
|
|
content = str(content_obj)
|
|
content_type = msg_type
|
|
except (json.JSONDecodeError, TypeError) as e:
|
|
logger.warning(f"[Feishu] 解析content失败: {e}, raw={str(raw_content)[:200]}")
|
|
content = str(raw_content)
|
|
|
|
return {
|
|
"content_type": content_type,
|
|
"content": content,
|
|
"sender_id": sender_id,
|
|
"sender_name": sender_name,
|
|
"message_id": message_id,
|
|
"timestamp": event.get("header", {}).get("create_time", 0),
|
|
"group_id": event.get("chat_id", None),
|
|
"group_name": None,
|
|
"raw": payload,
|
|
}
|
|
|
|
def build_reply(self, channel_config: dict, content: str, reply_to: str = None) -> dict:
|
|
"""构建飞书回复消息"""
|
|
app_id = channel_config.get("app_id", "")
|
|
app_secret = channel_config.get("app_secret", "")
|
|
return {
|
|
"api_url": "https://open.feishu.cn/open-apis/im/v1/messages",
|
|
"headers": {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {channel_config.get('tenant_access_token', '')}",
|
|
},
|
|
"body": {
|
|
"msg_type": "text",
|
|
"content": json.dumps({"text": content}),
|
|
},
|
|
}
|