如何在python-telegram-bot中使用send_message()主动给用户发消息
实现Telegram机器人主动发送消息
要让机器人无需用户触发就主动发消息,核心要搞定两个关键点:目标用户的Chat ID,以及能调用Telegram API的Bot实例。
步骤1:获取用户Chat ID
Telegram不允许机器人给从未互动过的用户发消息,所以先让用户给机器人发一条消息(比如触发/start),然后在代码里获取并记录用户的Chat ID:
修改你的start_commmand函数,添加打印/返回Chat ID的逻辑:
async def start_commmand(update, context): umsg = update.message chat_id = umsg.chat.id print(f"用户Chat ID:{chat_id}") # 控制台会输出用户ID,记下来 await umsg.reply_text(f"Welcome, i'll help u with ur schedule\n你的Chat ID是:{chat_id}")
步骤2:主动发送消息的两种实现方式
方式一:直接使用Bot类实例发送
直接创建Bot对象,调用send_message方法,注意这是异步操作,需要用asyncio.run来执行:
from dotenv import load_dotenv from telegram.ext import * from telegram import InputFile, Bot import os import re import asyncio command = ['start'] load_dotenv() tokenize = os.getenv("TELEGRAM_BOT_TOKEN") target_chat_id = 123456789 # 替换成你刚才拿到的用户Chat ID async def start_commmand(update, context): umsg = update.message chat_id = umsg.chat.id print(f"用户Chat ID:{chat_id}") await umsg.reply_text(f"Welcome, i'll help u with ur schedule\n你的Chat ID是:{chat_id}") async def send_active_message(chat_id, bot_token): bot = Bot(token=bot_token) await bot.send_message(chat_id=chat_id, text="这是机器人主动给你发的消息!") if __name__ == "__main__": # 先主动发送消息 asyncio.run(send_active_message(target_chat_id, tokenize)) # 启动机器人 application = Application.builder().token(tokenize).build() application.add_handler(CommandHandler(command[0], start_commmand)) application.run_polling(1.0)
方式二:通过Application的bot属性发送
如果已经创建了Application实例,可以直接用它的bot属性来发送消息,避免重复创建Bot对象:
# 替换main函数里的主动发送逻辑 if __name__ == "__main__": application = Application.builder().token(tokenize).build() # 用application的bot属性发送消息 async def send_via_app(): await application.bot.send_message(chat_id=target_chat_id, text="通过Application主动发消息!") asyncio.run(send_via_app()) application.add_handler(CommandHandler(command[0], start_commmand)) application.run_polling(1.0)
扩展:定时主动发送消息
如果需要定时发送(比如每天提醒),可以用JobQueue实现:
# 添加定时任务函数 async def scheduled_message(context): await context.bot.send_message(chat_id=target_chat_id, text="这是定时推送的消息!") if __name__ == "__main__": application = Application.builder().token(tokenize).build() # 设置定时任务:每30秒发送一次,立即执行第一次 job_queue = application.job_queue job_queue.run_repeating(scheduled_message, interval=30, first=0) application.add_handler(CommandHandler(command[0], start_commmand)) application.run_polling(1.0)
注意事项
- 必须确保用户已经和机器人互动过(发过消息或触发过命令),否则Telegram API会拒绝机器人的发送请求。
send_message是异步函数,所有调用它的地方必须用await,且要在异步上下文里执行(比如用asyncio.run包裹)。
内容的提问来源于stack exchange,提问作者Dzadafa
相关产品推荐
相关产品推荐

