如何通过Telegram API实时读取新到达的私信?
实时读取Telegram私信的实现方案
方法一:使用Telegram Bot API(推荐)
两种核心模式:轮询(Polling)和Webhook
轮询模式:定期调用
getUpdates接口拉取新消息,适合小型应用或测试场景
示例代码(Python,基于python-telegram-bot库):from telegram import Update from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes async def handle_private_msg(update: Update, context: ContextTypes.DEFAULT_TYPE): # 解析并处理私信内容 user_id = update.effective_user.id msg_text = update.effective_message.text print(f"收到用户 {user_id} 的消息: {msg_text}") if __name__ == '__main__': # 替换为你的机器人Token(从@BotFather获取) BOT_TOKEN = 'YOUR_BOT_TOKEN' app = ApplicationBuilder().token(BOT_TOKEN).build() # 仅监听私人聊天消息 private_msg_handler = MessageHandler(filters.ChatType.PRIVATE, handle_private_msg) app.add_handler(private_msg_handler) # 启动轮询 app.run_polling()注意:轮询频率可调整,但过于频繁可能触发API限流;机器人仅能接收主动给它发过消息的用户的私信。
Webhook模式:配置公网可访问的HTTPS地址,Telegram会主动推送新消息到该地址,适合生产环境
核心步骤:- 准备带SSL证书的公网域名/IP(Telegram要求Webhook地址必须为HTTPS)
- 调用接口配置Webhook:
https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook?url=https://your-public-domain/webhook - 服务端编写接口接收并解析POST请求
示例代码(Python Flask):
from flask import Flask, request import json app = Flask(__name__) @app.route('/webhook', methods=['POST']) def webhook(): update_data = json.loads(request.data) if 'message' in update_data: msg = update_data['message'] if msg['chat']['type'] == 'private': user_id = msg['from']['id'] msg_content = msg.get('text', '非文本消息') print(f"收到用户 {user_id} 的私信: {msg_content}") return 'ok' if __name__ == '__main__': # 生产环境建议用Gunicorn等WSGI服务器配合SSL证书 app.run(host='0.0.0.0', port=443, ssl_context=('cert.pem', 'key.pem'))
方法二:使用TDLib(Telegram Database Library)
如果需要直接操作个人Telegram账号(无需机器人),可使用TDLib实现全量消息监听
示例步骤:
- 下载或编译预编译的TDLib库
- 从
my.telegram.org获取个人账号的API ID和API Hash - 初始化客户端并监听新消息事件
示例代码片段(Python,基于pytdlib库):
from pytdlib import Client, types def on_update(update): if isinstance(update, types.UpdateNewMessage): msg = update.message if isinstance(msg.chat.type, types.ChatTypePrivate): user_id = msg.chat.id content = msg.content.text.text if isinstance(msg.content, types.MessageText) else '非文本消息' print(f"收到私信: 用户 {user_id} - {content}") client = Client( api_id=YOUR_API_ID, api_hash='YOUR_API_HASH', phone_number='YOUR_PHONE_NUMBER', database_encryption_key='your-encryption-key' ) client.add_update_listener(on_update) client.start()
注意:使用个人账号需遵守Telegram使用条款,避免违规操作导致封号。
关键注意事项
- 机器人无法接收未主动发起对话的用户的私信
- TDLib方式需要个人账号的API凭证,需提前在
my.telegram.org注册申请 - 生产环境优先选择Webhook模式,轮询仅适合开发测试场景
内容的提问来源于stack exchange,提问作者kaleab belayhun
相关产品推荐
相关产品推荐

