Python+schedule实现Telegram Bot定时推送问题排查

• bot.polling()为阻塞方法，代码运行到该行会持续监听用户消息，后续schedule初始化、定时任务循环的代码永远不会被执行
• 天气、BTC价格的请求逻辑全写在全局作用域，脚本启动时仅会请求一次，后续推送永远使用首次拉取的旧数据，不会更新
• 绑定/start、/help的处理函数依赖入参message获取聊天ID，schedule触发任务时不会传入该参数，直接调用会抛参数缺失错误；且定向推送频道不需要依赖用户消息触发的message对象
• 现有定时规则为每1秒执行一次，和“每日早间固定时间推送”的需求不匹配
• bot监听和schedule任务循环都是阻塞逻辑，放在同一主线程运行必然会阻塞其中一个

1. 将数据拉取逻辑封装为独立函数，每次触发推送/命令时重新请求最新数据，避免全局缓存旧内容
2. 单独编写定时推送函数，直接指定目标频道ID，不依赖用户消息的入参
3. 通过多线程分离bot监听和schedule定时循环逻辑，两个进程互不阻塞
4. 替换schedule定时规则为每日固定时间触发，匹配早间推送需求
5. 增加基础异常捕获和请求超时配置，避免网络错误导致整个脚本崩溃

import config
import telebot
import requests
import schedule
import time
import threading
from my_parser import parse
from bs4 import BeautifulSoup as BS

bot = telebot.TeleBot(config.token)
# 替换为实际要推送的频道ID，格式为-100开头的数字串
TARGET_CHAT_ID = -1001234567890

def fetch_latest_data():
    """每次调用重新拉取最新的天气、BTC价格数据，拼接为待发送文本"""
    # 拉取天气数据
    r = requests.get('https://example', timeout=10)
    html = BS(r.content, 'html.parser')
    weather_block = html.select_one('#content')
    t_min = weather_block.select_one('.temperature .min').text
    t_max = weather_block.select_one('.temperature .max').text
    min_text = weather_block.select_one('.wDescription .description').text

    # 拉取BTC价格数据
    response = requests.get(url='https://example', timeout=10)
    data = response.json()
    btc_price = f"B: {round(data.get('btc_usd').get('last'), 2)}$"

    return f"{t_min}, {t_max}\n{min_text}\n{parse()}\n{btc_price}"

def scheduled_push():
    """定时任务执行的推送逻辑"""
    try:
        msg_text = fetch_latest_data()
        bot.send_message(TARGET_CHAT_ID, msg_text)
    except Exception as e:
        print(f"定时推送出错: {str(e)}")

@bot.message_handler(commands=['start', 'help'])
def handle_start_help(message):
    """响应手动触发的/start、/help命令，返回最新数据"""
    msg_text = fetch_latest_data()
    bot.send_message(message.chat.id, msg_text)

def run_schedule_loop():
    """子线程运行定时任务循环"""
    # 配置每日推送时间，修改at内的24小时制时间字符串即可调整推送时点，比如"07:30"
    schedule.every().day.at("08:00").do(scheduled_push)
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == '__main__':
    # 启动守护线程运行定时任务，主进程退出时子线程同步终止
    schedule_thread = threading.Thread(target=run_schedule_loop, daemon=True)
    schedule_thread.start()
    # 主线程运行bot消息监听
    bot.polling(none_stop=True, interval=2)

• 运行前需将TARGET_CHAT_ID替换为实际目标频道ID，提前将bot加入频道并授予发消息权限
• 若需要测试定时逻辑，可以临时将定时规则改为schedule.every(30).seconds.do(scheduled_push)，验证推送正常后再改回每日固定时间规则
• 所有网络请求增加了10秒超时限制，避免异常请求卡住整个脚本进程
• 分线程运行后，手动命令响应和自动定时推送互不干扰，不会出现逻辑阻塞问题

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