如何调度两个Telegram机器人函数?schedule模块失效排查
问题分析与解决方案
你的代码存在几个关键问题导致schedule模块无法正常工作:
doo函数依赖message参数,但定时任务触发时没有这个参数传入,无法直接调用- 原函数的
bot.send_message依赖用户消息的chat.id,定时任务无法自动获取该ID - 缺少schedule模块的循环启动逻辑,且没处理Telegram机器人事件循环与schedule循环的兼容性
修正后的完整代码示例
我们需要调整逻辑,让定时任务能独立执行,同时兼容机器人的运行:
import telebot import random import schedule import time from threading import Thread from telebot import types # 替换成你的机器人Token bot = telebot.TeleBot("YOUR_BOT_TOKEN") # 存储需要推送消息的聊天ID(用户发/start后自动记录) target_chats = set() @bot.message_handler(commands=['start']) def foo(message): # 记录用户的聊天ID target_chats.add(message.chat.id) # 发送初始按钮消息 keyboard = types.InlineKeyboardMarkup() url_button = types.InlineKeyboardButton(text="Перейти на опрос", url="https://docs.google.com/forms/d/e/") keyboard.add(url_button) bot.send_message(message.chat.id, "Привет! Нажми на кнопку и перейди в опрос.", reply_markup=keyboard) # 发送随机消息 send_random_msg(message.chat.id) # 提取公共逻辑:读取txt发送随机消息 def send_random_msg(chat_id): try: with open("test.txt", encoding="utf8") as f: lines = f.read().splitlines() if lines: random_line = random.choice(lines) bot.send_message(chat_id, random_line) print(random_line) except Exception as e: print(f"发送随机消息出错: {e}") # 上午9点执行的定时任务 def morning_task(): for chat_id in target_chats: try: keyboard = types.InlineKeyboardMarkup() url_button = types.InlineKeyboardButton(text="Перейти на опрос", url="https://docs.google.com/forms/d/e/") keyboard.add(url_button) bot.send_message(chat_id, "Привет! Нажми на кнопку и перейди в опрос.", reply_markup=keyboard) send_random_msg(chat_id) except Exception as e: print(f"上午任务执行失败: {e}") # 晚上10点执行的定时任务 def night_task(): for chat_id in target_chats: try: keyboard = types.InlineKeyboardMarkup() url_button = types.InlineKeyboardButton(text="Перейти на опрос2", url="https://docs.google.com/forms/d/e/") keyboard.add(url_button) bot.send_message(chat_id, "Привет! Нажми на кнопку и перейди в опрос.2", reply_markup=keyboard) send_random_msg(chat_id) except Exception as e: print(f"晚上任务执行失败: {e}") # 设置定时任务 schedule.every().day.at("09:00").do(morning_task) schedule.every().day.at("22:00").do(night_task) # 单独开线程跑schedule循环,避免阻塞机器人 def run_schedule_loop(): while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务 # 启动schedule线程 Thread(target=run_schedule_loop, daemon=True).start() # 启动机器人 bot.infinity_polling()
关键修改说明
- 聊天ID存储:用户发送/start时自动记录聊天ID,定时任务遍历所有记录的ID推送消息
- 公共逻辑提取:把读取txt发随机消息的代码抽成独立函数,减少重复代码
- 定时任务重构:去掉对
message参数的依赖,直接用存储的聊天ID执行推送 - 线程分离:用单独线程运行schedule循环,避免阻塞Telegram机器人的事件循环
- 异常捕获:给任务添加异常处理,防止单个任务失败导致整个程序崩溃
内容的提问来源于stack exchange,提问作者Max
相关产品推荐
相关产品推荐

