能否让持续轮询的Telegram Bot进入休眠并通过/start命令唤醒？

我正在开发一款用于在Telegram发送格式化帖子的Bot，代码本身没问题。从运行日志能看到，Bot当前持续运行，每10秒发起一次getUpdates请求获取更新。但我只是偶尔使用这个Bot，想让它能进入休眠模式，只在收到/start命令时唤醒，请问该怎么实现？

运行日志 #

2024-05-04 12:22:05,706 - INFO - Scheduler started

2024-05-04 12:22:05,707 - INFO - Application started

2024-05-04 12:22:15,955 - INFO - HTTP Request: POST 
https://api.telegram.org/bot{TOKEN}/getUpdates "HTTP/1.1 200 OK"

2024-05-04 12:22:20,117 - INFO - HTTP Request: POST 
https://api.telegram.org/bot{TOKEN}/getUpdates "HTTP/1.1 200 OK"

2024-05-04 12:22:20,418 - INFO - HTTP Request: POST 
https://api.telegram.org/bot{TOKEN}/sendMessage "HTTP/1.1 200 OK"

2024-05-04 12:22:26,449 - INFO - HTTP Request: POST 
https://api.telegram.org/bot{TOKEN}/getUpdates "HTTP/1.1 200 OK"

主代码片段 #

def main():

app = ApplicationBuilder() \

    .read_timeout(100) \

    .write_timeout(100) \

    .token(os.environ["TG_BOT_TOKEN"]) \

    .build()



app.add_handler(CommandHandler("start", start))

app.add_handler(CommandHandler("stop", stop))

               .
               .
               .
               .
               .

app.run_polling()



if __name__ == '__main__':

    main()

实现方案 #

要实现休眠/唤醒逻辑，核心是控制轮询的启停，结合状态标记和线程管理来切换Bot状态：

1. 全局状态标记：用变量标记Bot是否处于活跃轮询状态
2. 线程托管轮询：把阻塞式的run_polling()放到子线程，避免主线程被卡死
3. 命令处理器控制启停：通过/start启动轮询，/stop终止轮询回到休眠

修改后的完整代码示例：

import os
import threading
import asyncio
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# 全局状态：标记Bot是否在运行轮询
is_running = False
polling_thread = None

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    global is_running, polling_thread
    if is_running:
        await update.message.reply_text("Bot已经在运行了")
        return
    
    # 定义轮询任务函数
    def run_polling_task():
        global is_running
        try:
            context.application.run_polling()
        finally:
            is_running = False
    
    # 启动轮询线程
    is_running = True
    polling_thread = threading.Thread(target=run_polling_task)
    polling_thread.start()
    await update.message.reply_text("Bot已唤醒，开始处理请求！")

async def stop(update: Update, context: ContextTypes.DEFAULT_TYPE):
    global is_running, polling_thread
    if not is_running:
        await update.message.reply_text("Bot已经休眠了")
        return
    
    # 停止轮询并等待线程结束
    await context.application.stop()
    polling_thread.join()
    is_running = False
    await update.message.reply_text("Bot已休眠，发送/start可唤醒")

def main():
    app = ApplicationBuilder() \
        .read_timeout(100) \
        .write_timeout(100) \
        .token(os.environ["TG_BOT_TOKEN"]) \
        .build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("stop", stop))

    # 初始启动仅维持事件循环，不主动轮询
    async def keep_alive():
        await app.start()
        while True:
            await asyncio.sleep(1)
    
    asyncio.run(keep_alive())

if __name__ == '__main__':
    main()

说明 #

• 初始启动后，Bot不会主动发起getUpdates请求，处于低功耗的休眠状态，仅通过事件循环维持命令监听
• 发送/start命令后，启动轮询线程，开始定期获取更新处理请求
• 发送/stop命令后，停止轮询并回收线程，Bot回到休眠状态

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