如何在Python Telegram Bot的job_queue run_repeating回调中获取chat_id

场景1：针对单个用户触发的重复任务 #

当用户通过命令（如/start）主动开启重复任务时，可将用户的chat_id绑定到任务上下文里，直接传递给重复任务函数：

from telegram.ext import CommandHandler

def start(update, context):
    # 获取当前触发命令的用户chat_id
    chat_id = update.effective_chat.id
    # 启动重复任务时，将chat_id存入任务上下文
    context.job_queue.run_repeating(
        callback=repeating,
        interval=5,
        context=chat_id
    )
    update.message.reply_text("重复任务已启动")

def repeating(context):
    # 从任务上下文中取出预先传入的chat_id
    chat_id = context.job.context
    # 发送个性化消息
    context.bot.send_message(chat_id=chat_id, text=res["message"])

def main():
    updater = Updater(constants.BOT_API_KEY, use_context=True)
    dp = updater.dispatcher
    
    # 注册命令处理器，让用户能触发任务
    dp.add_handler(CommandHandler("start", start))
    
    updater.start_polling()
    updater.idle()

main()

注意：无需手动创建JobQueue，Updater实例已自带job_queue属性，直接调用即可。

场景2：给多个用户批量发送个性化消息 #

如果需要给一批用户推送消息，得先把用户的chat_id存储起来（比如用JSON文件、SQLite数据库），再在重复任务中遍历发送：

第一步：实现用户注册逻辑（保存chat_id） #

import json

def register(update, context):
    chat_id = update.effective_chat.id
    # 读取已存储的用户ID列表
    try:
        with open("user_ids.json", "r") as f:
            user_ids = json.load(f)
    except FileNotFoundError:
        user_ids = []
    
    if chat_id not in user_ids:
        user_ids.append(chat_id)
        with open("user_ids.json", "w") as f:
            json.dump(user_ids, f)
        update.message.reply_text("已完成注册，将收到推送消息")
    else:
        update.message.reply_text("你已在推送列表中")

第二步：修改重复任务函数，批量发送消息 #

def repeating(context):
    # 读取存储的用户chat_id列表
    try:
        with open("user_ids.json", "r") as f:
            user_ids = json.load(f)
    except FileNotFoundError:
        user_ids = []
    
    # 遍历每个用户，发送个性化内容
    for chat_id in user_ids:
        # 可根据chat_id生成专属消息内容
        personalized_msg = f"用户{chat_id}，这是你的专属重复消息：{res['message']}"
        context.bot.send_message(chat_id=chat_id, text=personalized_msg)

def main():
    updater = Updater(constants.BOT_API_KEY, use_context=True)
    dp = updater.dispatcher
    
    dp.add_handler(CommandHandler("register", register))
    # 启动全局重复任务
    updater.job_queue.run_repeating(repeating, interval=5)
    
    updater.start_polling()
    updater.idle()

main()

内容的提问来源于stack exchange，提问作者Muhammed Didin