Telegram bot to send auto message every n hours with python-telegram-bot

Question

I am quite new in building bots so I created a very simple Telegram bot and it works great but can't figure out how to make the bot send messages every n minutes or n hours when /start_auto command is initiated.

I made a workaround with while loop but it looks stupid and during the loop users won't be able to interact with the bot about other topics.

I want users to be able to start and stop this scheduled task by commands such as /start_auto and /stop_auto.

I know there are many other answered questions related to this topic but none of them seem to be working with my code.

import logging
import os
import time

from telegram.ext import Updater, CommandHandler, MessageHandler, Filters

logger = logging.getLogger(__name__)

PORT = int(os.environ.get('PORT', '8443'))


def start(update, context):
    """Sends a message when the command /start is issued."""
    update.message.reply_text('Hi!')


def help(update, context):
    """Sends a message when the command /help is issued."""
    update.message.reply_text('Help!')


def start_auto(update, context):
    """Sends a message when the command /start_auto is issued."""
    n = 0
    while n < 12:
        time.sleep(3600)
        update.message.reply_text('Auto message!')
        n += 1


def error(update, context):
    """Logs Errors caused by Updates."""
    logger.warning('Update "%s" caused error "%s"', update, context.error)


def main():
    TOKEN = 'TOKEN_GOES_HERE'
    APP_NAME = 'https://my-tele-test.herokuapp.com/' 

    updater = Updater(TOKEN, use_context=True)

    # Get the dispatcher to register handlers
    dp = updater.dispatcher

    dp.add_handler(CommandHandler("start", start))
    dp.add_handler(CommandHandler("help", help))
    dp.add_handler(CommandHandler("start_auto", start_auto))

    # log all errors
    dp.add_error_handler(error)
    updater.start_webhook(listen="0.0.0.0", port=PORT, url_path=TOKEN, webhook_url=APP_NAME + TOKEN)
    updater.idle()


if __name__ == '__main__':
    main()

Answer 1

from aiogram import Bot, Dispatcher, types
from aiogram.utils import executor
import random

TOKEN = "YOUR_BOT_TOKEN"  # اینجا توکن ربات خودت را قرار بده
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)

# امتیازات کاربران
user_scores = {}
user_avatars = {}  # آواتار کاربران

# لیست رنگ‌های گوی
colors = ["⚫ مشکی", "🔵 سرمه‌ای", "🟡 طلایی"]
correct_color = random.choice(colors)  # رنگ درست تصادفی انتخاب می‌شود

@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    user_scores[message.from_user.id] = 0  # تنظیم امتیاز اولیه
    user_avatars[message.from_user.id] = "👤"  # آواتار پیش‌فرض
    await message.reply("🎮 به بازی آتلانتیس خوش آمدید! دستور /play را بزنید تا بازی کنید.")

@dp.message_handler(commands=['play'])
async def play_game(message: types.Message):
    global correct_color
    correct_color = random.choice(colors)  # انتخاب رنگ جدید برای گوی درست
    keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True)
    buttons = [types.KeyboardButton(c) for c in colors]
    keyboard.add(*buttons)
    await message.reply("🎲 یک گوی انتخاب کن!", reply_markup=keyboard)

@dp.message_handler(lambda message: message.text in colors)
async def check_answer(message: types.Message):
    global correct_color
    user_id = message.from_user.id
    if message.text == correct_color:
        user_scores[user_id] += 10  # امتیاز اضافه شود
        await message.reply(f"✅ درست گفتی! {user_avatars[user_id]} امتیاز تو: {user_scores[user_id]} 🎉")
    else:
        await message.reply(f"❌ اشتباهه! رنگ درست {correct_color} بود. دوباره امتحان کن با /play 🎮")

@dp.message_handler(commands=['leaderboard'])
async def show_leaderboard(message: types.Message):
    leaderboard = sorted(user_scores.items(), key=lambda x: x[1], reverse=True)
    ranking = "🏆 رتبه‌بندی بازیکنان:\n"
    for i, (user_id, score) in enumerate(leaderboard[:10], start=1):
        ranking += f"{i}. {user_avatars[user_id]} کاربر {user_id}: {score} امتیاز\n"
    await message.reply(ranking)

@dp.message_handler(commands=['challenge'])
async def weekly_challenge(message: types.Message):
    challenge_message = "🔥 چالش این هفته: در 5 دقیقه 50 امتیاز بگیر! آیا قبول می‌کنی؟"
    await message.reply(challenge_message)

@dp.message_handler(commands=['set_avatar'])
async def set_avatar(message: types.Message):
    if len(message.text.split()) < 2:
        await message.reply("⚠️ لطفاً یک ایموجی برای آواتار خود ارسال کنید. مثال: /set_avatar 🦁")
        return
    user_avatars[message.from_user.id] = message.text.split()[1]
    await message.reply(f"✅ آواتار شما به {user_avatars[message.from_user.id]} تغییر کرد!")

@dp.message_handler(commands=['chat'])
async def chat_handler(message: types.Message):
    await message.reply("💬 چت کاربران فعال شد! هر چیزی که اینجا ارسال کنید، کاربران دیگر می‌توانند ببینند.")
    # در اینجا می‌توان یک گروه عمومی تلگرامی متصل کرد

@dp.message_handler(commands=['store'])
async def store_handler(message: types.Message):
    await message.reply("🛒 فروشگاه امتیازات: می‌توانید امتیاز خریداری کنید و گوی‌های ویژه بگیرید!")

@dp.message_handler(commands=['share'])
async def share_score(message: types.Message):
    await message.reply(f"📢 امتیاز خود را با دوستانت به اشتراک بگذار! \n 🌟 امتیاز شما: {user_scores.get(message.from_user.id, 0)}")

if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True)

Answer 2

You can just create task using asyncio. Remember, you don't need an 'await' here because you need just to append it to the event loop:

import asyncio
import logging
import os

from aiogram import Bot
from aiogram import Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import Chat

TELEGRAM_BOT_API_TOKEN = os.getenv('TELEGRAM_BOT_API_TOKEN')
bot = Bot(token=TELEGRAM_BOT_API_TOKEN,
          default=DefaultBotProperties(parse_mode=ParseMode.HTML)
          )

dp = Dispatcher(storage=MemoryStorage())


async def scheduler(delay: int):
    chat_ids = []  # ids

    while True:

        for chat_id in chat_ids:
            await bot.send_message(chat_id=chat_id, text='Time message')

        await asyncio.sleep(delay=delay)  # wait every 3600 seconds


async def main():
    await bot.delete_webhook(drop_pending_updates=True)
    task = asyncio.create_task(coro=scheduler(delay=3600)) # this task will be called every 3600 seconds
    await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

Answer 3

python-telegram-bot has a built-in feature for scheduling tasks, called JobQueue. Please have a look at this wiki page for versions <20 or the documentation for later versions for more info.

---

Disclaimer: I'm currently the maintainer of python-telegram-bot.

Answer 4

Everything is simple here, you only need the time library and one while loop to send an auto-message. time.sleep(30)) shows how often messages are sent in seconds.

import telebot
from telebot import custom_filters
import time


bot = telebot.TeleBot("TOKEN FOR BOT")



# Chat id can be private or supergroups.
@bot.message_handler(chat_id=[123456789], commands=['start']) # chat_id checks id corresponds to your list or not.
def admin_reply(message):
    bot.send_message(message.chat.id, "You are allowed to use this command.")
    while(True):
        bot.send_message(message.chat.id, "Allowed to receive an auto-message every 30 seconds.", time.sleep(30))

@bot.message_handler(commands=['start'])
def not_admin(message):
    bot.send_message(message.chat.id, "You are not allowed to use this command")
    while(True):
        bot.send_message(message.chat.id, "Allowed to receive an auto-message every minute.", time.sleep(60))

# Do not forget to register
bot.add_custom_filter(custom_filters.ChatFilter())
bot.polling(none_stop=True)

Answer 5

Why don't you use pyrogram. It has made everything easier.

from pyrogram import Client, filters 
import time

app = Client('my_account',          #Update this line with
               api_id=API_ID,       #your API_ID
               api_hash=API_HASH,   #your API_HASH
               bot_token=BOT_TOKEN) #your BOT_TOKEN

def main():
    while(True):
        app.send_message(chat_id=123456789, text="This message is sent every 5 mins")
        time.sleep(5000) app.run()

Answer 6

will post a solution which I found:

def callback_auto_message(context):
    context.bot.send_message(chat_id='12345678', text='Automatic message!')


def start_auto_messaging(update, context):
    chat_id = update.message.chat_id
    context.job_queue.run_repeating(callback_auto_message, 10, context=chat_id, name=str(chat_id))
    # context.job_queue.run_once(callback_auto_message, 3600, context=chat_id)
    # context.job_queue.run_daily(callback_auto_message, time=datetime.time(hour=9, minute=22), days=(0, 1, 2, 3, 4, 5, 6), context=chat_id)


def stop_notify(update, context):
    chat_id = update.message.chat_id
    context.bot.send_message(chat_id=chat_id, text='Stopping automatic messages!')
    job = context.job_queue.get_jobs_by_name(str(chat_id))
    job[0].schedule_removal()

And in the main function I created a commands:

dp.add_handler(CommandHandler("auto", start_auto_messaging))
dp.add_handler(CommandHandler("stop", stop_notify))