Python-telegram-bot如何检测实时位置共享的开始与结束事件

Telegram Bot API本身不会主动推送实时位置共享自动到期的回调事件，你可以结合本地超时任务的方案补全自动结束的检测逻辑，完整实现步骤如下：

1. 利用python-telegram-bot自带的job_queue实现延迟任务调度，通过context.user_data存储每个用户对应的超时任务实例，避免全局变量的并发冲突问题。
2. 检测到实时位置共享开始事件时，读取消息中live_period字段的值，注册对应时长的延迟任务，到期后自动触发结束逻辑。
3. 每次收到实时位置更新事件时，先取消之前注册的旧超时任务，再根据当前最新的live_period重新注册新的延迟任务。
4. 检测到用户手动结束实时位置共享事件时，直接取消当前用户绑定的超时任务，执行结束逻辑即可。

完整实现代码示例：

from telegram.ext import Updater, MessageHandler, Filters, CallbackContext
from telegram.update import Update

API_KEY = "你的Bot API密钥"

# 实时位置到期回调函数
def live_location_timeout(context: CallbackContext):
    user_id = context.job.context
    context.bot.send_message(user_id, "实时位置共享已自动到期结束")
    # 清理存储的任务实例
    if "live_location_job" in context.user_data:
        del context.user_data["live_location_job"]

def location(update: Update, context: CallbackContext):
    user = update.effective_user
    msg_type = 0

    if update.edited_message:
        message = update.edited_message
    else:
        message = update.message

    # 先判断edit_date存在性
    has_edit_date = message.edit_date is not None
    # 再判断live_period存在性
    has_live_period = message.location.live_period is not None

    if has_edit_date:
        msg_type += 1
    if has_live_period:
        msg_type += 1 << 1

    if msg_type == 0:
        context.bot.send_message(user.id, "单次（非实时）位置更新")
    elif msg_type == 1:
        # 手动结束实时位置
        context.bot.send_message(user.id, "实时位置共享已手动结束")
        # 取消已注册的超时任务
        if "live_location_job" in context.user_data:
            context.user_data["live_location_job"].schedule_removal()
            del context.user_data["live_location_job"]
    elif msg_type == 2:
        # 实时位置开始
        context.bot.send_message(user.id, "实时位置共享已开始")
        # 注册超时任务
        job = context.job_queue.run_once(
            live_location_timeout,
            when=message.location.live_period,
            context=user.id
        )
        context.user_data["live_location_job"] = job
    elif msg_type == 3:
        # 实时位置更新
        context.bot.send_message(user.id, "实时位置更新中")
        # 取消旧的超时任务，重新注册新的
        if "live_location_job" in context.user_data:
            context.user_data["live_location_job"].schedule_removal()
        job = context.job_queue.run_once(
            live_location_timeout,
            when=message.location.live_period,
            context=user.id
        )
        context.user_data["live_location_job"] = job

def main() -> None:
    updater = Updater(API_KEY)
    dispatcher = updater.dispatcher

    location_handler = MessageHandler(Filters.location, location)
    dispatcher.add_handler(location_handler)

    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

注意事项 #

• 如果用户在共享过程中调整了实时位置的持续时长，Telegram会自动更新消息中的live_period字段值，上述重新注册任务的逻辑会自动适配新的时长，无需额外处理。
• 如果服务重启，所有内存中未执行的超时任务会丢失，若需要高可靠的结束检测，可以将用户的实时位置到期时间存储在持久化存储（如SQLite、Redis）中，服务启动时扫描未到期的记录恢复对应超时任务即可。

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