2

Есть Telegram-бот (pyTelegramBotAPI), запущенный на PythonAnywhere (Flask), он умеет реагировать на определенные команды. Возможно ли дополнительно к этому реализовать периодическую отправку сообщений всем пользователям в определенное время?

У меня так и не получилось найти инструкций по этому вопросу. Встречал иногда упоминания библиотеки schedule, но встроить ее в PythonAnywhere не вышло, так как там работает только 1 поток.

2
  • 1
    Если с английским всё норм, то можно здесь почитать pythoncircle.com/post/50/…
    – svstnv
    Commented 27 янв. 2020 в 5:23
  • Если я правильно понял, надо создать ещё один скрипт, который shedule как сказал @svstnv Commented 6 нояб. 2022 в 3:38

3 ответа 3

0

Используйте модуль datetime. Через него получаете время, проверяете то ли это время, отсылает, если все ок. Что отослать всем используйте цикл по пользователям.

2
  • Насколько я знаю, я не могу просто взять и начать проверять то ли сейчас время посреди работы веб-приложения. Commented 28 янв. 2020 в 9:48
  • Извините, я невнимательно прочитал, получается.
    – n1tr0xs
    Commented 28 янв. 2020 в 9:52
0

Можно использовать библитеку aiocron и сделать hendler.Пример для повторение каждую минуту @aiocron.crontab('*/1 * * * *')

0

Можно использовать celery для выполнения периодических заданий. Для этого надо настроить celery (самый короткий и ясный гайд), и использовать "агента" celery - beat_schedule, который будет периодически запускать задачи.

Например:

#celery_settings.py
import os
import celery

os.environ.setdefault("FASTAPI_SETTINGS_MODULE", "service/settings.env")
app = celery.Celery("fastapi_celery", broker="redis://localhost:6379")
app.config_from_object("settings.env", namespace="CELERY")
app.autodiscover_tasks()

#настраиваем beat_schedule
app.conf.beat_schedule = {
    "parsing_steam":
        {
            "task": "tasks.table_update",
            "schedule": 60.0,
        },
    "updating_ml":
        {
            "task": "tasks.update_weights",
            "schedule": 3000.0,
        }
}

Перед этим рекомендуется переместить функции-задачи celery в отдельную директорию, и использовать декоратор @*твой_app*.task или создать новое определение с декоратором @*твой_app*.task, в котором будет вызываться старая функция (всё это для чистоты кода).

Например:

from celery_settings import app
from data.parser.parser import table_update
from service.ml.ml import update_weights


@app.task
def table_update():
    table_update()


@app.task
def update_weights():
    update_weights()

Это должно работать. POV celery - широко используемая асинхронная очередь задач, которая используется вместе с различными API, от FastApi, Django до aiogram.

1
  • То самое чувство, когда ответил на вопрос четырёхгодичной давности)
    – Роман
    Commented 7 мая в 18:20

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.