如何使用python-telegram-bot向任意用户主动发送消息？

你目前的代码只能给固定ID发消息，要实现向任意用户主动推送，核心是先收集用户的chat_id——因为Telegram Bot无法主动给从未发过消息的用户发消息，必须用户先发起对话，Bot才能获取并保存其chat_id。

实现步骤： #

1. 收集并保存用户chat_id：通过消息处理器，记录所有和Bot对话过的用户ID
2. 遍历ID列表批量发消息：在定时任务里遍历保存的ID，逐个发送消息

修改后的完整代码示例： #

import os
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from telegram.ext import ContextTypes, CallbackContext
from telegram import Update

# 存储用户chat_id的列表（如果需要持久化，建议用数据库）
chat_ids = []

async def handle_message(update: Update, context: CallbackContext):
    chat_id = update.effective_chat.id
    # 避免重复添加相同ID
    if chat_id not in chat_ids:
        chat_ids.append(chat_id)
        await update.message.reply_text("已成功记录你的ID，后续会收到推送消息")
    else:
        await update.message.reply_text("你的ID已存在，无需重复添加")

async def send_message(context: CallbackContext):  
    text = "这是定时推送的消息内容..."
    
    # 遍历所有已记录的用户ID发送消息
    for chat_id in chat_ids.copy():  # 用copy避免遍历中修改列表报错
        try:
            await context.bot.send_message(chat_id=chat_id, text=text)
        except Exception as e:
            print(f"发送消息到ID {chat_id} 失败: {str(e)}")
            # 移除无效ID（比如用户拉黑Bot）
            if chat_id in chat_ids:
                chat_ids.remove(chat_id)

def main() -> None:
    token = os.getenv('TELEGRAM_BOT_TOKEN')
    if not token:
        print("请设置TELEGRAM_BOT_TOKEN环境变量")
        return
    
    app = Application.builder().token(token).build()

    # 添加消息处理器，用于收集用户ID
    app.add_handler(MessageHandler(filters.TEXT, handle_message))

    # 定时任务，每20秒发送一次消息
    app.job_queue.run_repeating(send_message, interval=20, first=0)

    # 错误处理器（需自行实现error函数）
    # app.add_error_handler(error)

    app.run_polling(poll_interval=2, timeout=5)

if __name__ == '__main__':
    main()

关键说明： #

• chat_id存储：示例用列表存储，重启Bot后会丢失数据。如果需要持久化，推荐用SQLite、MySQL等数据库，每次启动时从数据库加载已有的chat_id。
• 异常处理：发送消息时可能遇到用户拉黑Bot、账号注销等情况，捕获异常并移除无效ID，避免后续重复报错。
• Telegram限制：Bot只能给主动发起过对话的用户发消息，这是Telegram的规则，无法绕过。

内容的提问来源于stack exchange，提问作者Gabriel Fernando Rey