qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
155 lines
5.8 KiB
Python
155 lines
5.8 KiB
Python
"""
|
|
Telegram平台适配器
|
|
"""
|
|
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="telegram",
|
|
display_name="Telegram",
|
|
description="Telegram Bot API消息适配器，支持文本、图片、文件、语音、视频、贴纸等消息类型",
|
|
support_streaming=True,
|
|
support_image=True,
|
|
support_file=True,
|
|
support_voice=True,
|
|
)
|
|
class TelegramPlatformAdapter(BasePlatformAdapter):
|
|
"""Telegram平台适配器"""
|
|
|
|
def get_metadata(self) -> PlatformMetadata:
|
|
from ..platform_adapter import platform_registry
|
|
return platform_registry["telegram"]
|
|
|
|
def verify_signature(self, payload: dict, signature: str, timestamp: str) -> bool:
|
|
"""Telegram签名验证: HMAC-SHA256(BotToken, 可选)"""
|
|
if not signature:
|
|
return True # Telegram不使用签名头验证，通过setWebhook时设置secret_token
|
|
|
|
bot_token = self.config.get("bot_token", "")
|
|
if not bot_token:
|
|
logger.warning("[Telegram] 未配置bot_token，跳过签名验证")
|
|
return True
|
|
|
|
try:
|
|
# Telegram支持通过 X-Telegram-Bot-Api-Secret-Token 头验证
|
|
# 新式验证：secret_token = SHA256(bot_token)
|
|
expected = hashlib.sha256(bot_token.encode("utf-8")).hexdigest()
|
|
return signature == expected
|
|
except Exception as e:
|
|
logger.error(f"[Telegram] 签名验证异常: {e}")
|
|
return False
|
|
|
|
def parse_webhook(self, payload: dict) -> dict:
|
|
"""解析Telegram webhook payload为统一消息格式"""
|
|
message = payload.get("message", payload.get("edited_message", {}))
|
|
if not message:
|
|
# 可能是callback_query等其他更新
|
|
callback = payload.get("callback_query", {})
|
|
if callback:
|
|
message = callback.get("message", {})
|
|
message["_callback_data"] = callback.get("data", "")
|
|
|
|
chat = message.get("chat", {})
|
|
from_user = message.get("from", {})
|
|
chat_id = str(chat.get("id", ""))
|
|
chat_type = chat.get("type", "private")
|
|
chat_title = chat.get("title", "")
|
|
|
|
sender_id = str(from_user.get("id", ""))
|
|
sender_name = from_user.get("first_name", "") or from_user.get("username", "")
|
|
|
|
content = ""
|
|
content_type = "text"
|
|
|
|
if "text" in message:
|
|
content = message.get("text", "")
|
|
content_type = "text"
|
|
elif "photo" in message:
|
|
photos = message.get("photo", [])
|
|
largest = photos[-1] if photos else {}
|
|
content = f"[图片] file_id={largest.get('file_id', '')}"
|
|
content_type = "image"
|
|
elif "document" in message:
|
|
doc = message.get("document", {})
|
|
content = f"[文件] {doc.get('file_name', 'unknown')} file_id={doc.get('file_id', '')}"
|
|
content_type = "file"
|
|
elif "voice" in message:
|
|
voice = message.get("voice", {})
|
|
content = f"[语音] file_id={voice.get('file_id', '')} duration={voice.get('duration', 0)}s"
|
|
content_type = "voice"
|
|
elif "video" in message:
|
|
video = message.get("video", {})
|
|
content = f"[视频] file_id={video.get('file_id', '')}"
|
|
content_type = "video"
|
|
elif "sticker" in message:
|
|
sticker = message.get("sticker", {})
|
|
content = f"[贴纸] emoji={sticker.get('emoji', '')} set={sticker.get('set_name', '')}"
|
|
content_type = "sticker"
|
|
elif "location" in message:
|
|
loc = message.get("location", {})
|
|
content = f"[位置] lat={loc.get('latitude')} lng={loc.get('longitude')}"
|
|
content_type = "location"
|
|
elif "contact" in message:
|
|
contact = message.get("contact", {})
|
|
content = f"[联系人] {contact.get('first_name', '')} phone={contact.get('phone_number', '')}"
|
|
content_type = "contact"
|
|
elif "animation" in message:
|
|
anim = message.get("animation", {})
|
|
content = f"[动画] file_id={anim.get('file_id', '')}"
|
|
content_type = "animation"
|
|
else:
|
|
content = json.dumps(message, ensure_ascii=False)
|
|
content_type = "unknown"
|
|
|
|
# 群聊/频道标识
|
|
group_id = None
|
|
group_name = None
|
|
if chat_type in ("group", "supergroup"):
|
|
group_id = chat_id
|
|
group_name = chat_title
|
|
elif chat_type == "channel":
|
|
group_id = chat_id
|
|
group_name = chat_title
|
|
|
|
return {
|
|
"content_type": content_type,
|
|
"content": content,
|
|
"sender_id": sender_id,
|
|
"sender_name": sender_name,
|
|
"message_id": str(message.get("message_id", "")),
|
|
"timestamp": message.get("date", 0),
|
|
"group_id": group_id,
|
|
"group_name": group_name,
|
|
"raw": payload,
|
|
}
|
|
|
|
def build_reply(self, channel_config: dict, content: str, reply_to: str = None) -> dict:
|
|
"""构建Telegram回复消息"""
|
|
bot_token = channel_config.get("bot_token", "")
|
|
chat_id = channel_config.get("chat_id", "")
|
|
if not bot_token:
|
|
logger.warning("[Telegram] 未配置bot_token，无法构建回复")
|
|
return {}
|
|
|
|
body = {
|
|
"chat_id": chat_id,
|
|
"text": content,
|
|
"parse_mode": "HTML",
|
|
}
|
|
if reply_to:
|
|
body["reply_to_message_id"] = reply_to
|
|
|
|
return {
|
|
"api_url": f"https://api.telegram.org/bot{bot_token}/sendMessage",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": body,
|
|
}
|