Как я уже писал в этом посте, со временем у многих людей меняются привычки и понятия комфорта. Так, до недавнего времени, ежедневные отчеты после формирования отправлялись руководителям по почте.
Возник запрос: хотим прямую доставку в мессенджер. Не проблема, начинаем разбираться.
Сразу добавляем условие: отчеты формируются долго, поэтому их расписание перенесено на ночь. Но кто хочет получать рабочие уведомления в полночь, а то и позже? Да и завязывать запуск сторонних скриптов в бота как-то не очень хочется.
Давайте напишем свой шедулер, он же - планировщик задач.
В этом нам поможет библиотека schedule. Разработчики библиотеки прямо пишут: schedule - не панацея для всех решений. Если вам нужна локализация для праздничных дней, мультипоточность, или слишком точное расписание (до миллисекунд), она вам не подойдет.
В нашем случае все намного проще, тайминги у нас с точностью до минут. Так что начнем.
- Добавляем библиотеки бота, планировщика и времени:
import telebot
import schedule
import time
- Пишем простенькую функцию, которая должна будет выполняться:
- Пропишем выполнение этой функции раз в минуту:
В поле "do" сначала указывается функция, которая должна выполниться, далее - ее аргументы.
- Запускаем в бесконечном цикле выполнение нашего планировщика с паузами в 1 секунду:
run_pending запускает проверку, нужно ли в конкретную секунду выполнить какую-либо задачу из прописанного расписания.
- Немного усложним нашу логику и расписание:
Обернем расписание и запуск проверки в отдельную функцию, добавим расписание на каждый рабочий день недели. В задании нам указали, что отчеты должны приходить в 8 утра.
def start_schedule():
schedule.every().monday.at("08:00").do(planned_job, )
schedule.every().tuesday.at("08:00").do(planned_job, )
schedule.every().wednesday.at("08:00").do(planned_job, )
schedule.every().thursday.at("08:00").do(planned_job, )
schedule.every().friday.at("08:00").do(planned_job, )
while True:
schedule.run_pending()
time.sleep(1)
Функцию planned_job переписываем под нашу конкретную задачу:
def planned_job():
users_ids = [123123123, 456456456, 789789789]
for telegram_id in users_ids:
with open('Report.xlsx', 'rb') as report:
bot.send_document(chat_id=telegram_id, document=report, caption='Ежедневный отчет')
Добавим немного магии. Настраиваем потоки.
Нам понадобится библиотека threading
from threading import Thread
В параметры запуска скрипта нашего бота добавляем запуск нового потока, который будет управлять нашим планировщиком.
И туда же мы пропишем параметры polling нашего бота, которые перезапустят его в случае падения.
if __name__ == '__main__':
Thread(target=start_schedule, args=()).start()
while True:
try:
bot.polling(none_stop=True)
except Exception as e:
time.sleep(3)
Многие советуют использовать метод infinity_polling для постоянной работы бота. По личному опыту могу сказать, что указанный мной метод работает всегда.
Также для запуска бота на сервере с linux, советую создать для него отдельного демона, как показано здесь.