You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在pyTelegramBotAPI中运行两个任务?后续任务无法执行求助

问题分析与解决方案

你的代码存在几个核心问题,导致后续任务无法执行:

  1. 阻塞调用:bot.polling()是同步阻塞方法,一旦执行就会占据主线程,后面的代码完全不会被运行。
  2. 异步使用错误:task1和task2都不是异步函数,直接用asyncio.run和asyncio.create_task调用不符合异步逻辑。
  3. 定时任务逻辑缺陷:task2只执行一次时间检查,即使触发条件满足也只会发送一次,且无法持续监控时间。
  4. 类名大小写错误:原代码中telebot.Telebot应为telebot.TeleBot,大小写错误会导致机器人初始化失败。

方案一:用线程分离同步任务

适合习惯使用同步版pyTelegramBotAPI的场景,通过线程把机器人轮询和定时任务分开执行:

import telebot
import threading
import datetime
import time

# 替换为你的机器人token和管理员ID
bot = telebot.TeleBot("your_bot_token")
admin_id = 123456789

# 示例消息处理函数
@bot.message_handler(commands=['start'])
def handle_start(message):
    bot.send_message(message.chat.id, "机器人已启动!")

def run_bot():
    # none_stop=True保证机器人持续运行
    bot.polling(none_stop=True)

def timed_task():
    while True:
        now = datetime.datetime.now()
        if now.hour == 12 and now.minute == 30:
            try:
                bot.send_document(admin_id, open("work_db.db", "rb"))
                # 休眠60秒避免同一分钟内重复发送
                time.sleep(60)
            except Exception as e:
                print(f"发送文件失败:{str(e)}")
        # 每隔10秒检查一次时间,减少资源占用
        time.sleep(10)

if __name__ == '__main__':
    # 把机器人轮询放到后台线程运行
    threading.Thread(target=run_bot, daemon=True).start()
    # 主线程运行定时任务
    timed_task()

方案二:使用异步版pyTelegramBotAPI

如果想基于asyncio实现异步任务,需要使用官方提供的异步分支:

import asyncio
import datetime
from telebot.async_telebot import AsyncTeleBot

# 替换为你的机器人token和管理员ID
bot = AsyncTeleBot("your_bot_token")
admin_id = 123456789

# 异步消息处理函数
@bot.message_handler(commands=['start'])
async def handle_start(message):
    await bot.send_message(message.chat.id, "异步机器人已启动!")

async def run_bot():
    await bot.polling(none_stop=True)

async def timed_task():
    while True:
        now = datetime.datetime.now()
        if now.hour == 12 and now.minute == 30:
            try:
                await bot.send_document(admin_id, open("work_db.db", "rb"))
                await asyncio.sleep(60)
            except Exception as e:
                print(f"发送文件失败:{str(e)}")
        await asyncio.sleep(10)

async def main():
    # 同时运行两个异步任务
    await asyncio.gather(run_bot(), timed_task())

if __name__ == '__main__':
    asyncio.run(main())

内容的提问来源于stack exchange,提问作者Emir

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.28 09:53:34