Python and Telegram are an excellent pair for building fun, practical projects. Python’s simplicity and flexibility make it ideal for all kinds of programming, while Telegram, as a popular messaging app, allows you to easily create bots that interact with users in real-time. It’s a great combination for building smart, automated systems.
Ever wanted a bot that automates tasks for you? In this tutorial, you’ll discover how to create an automated Telegram bot using Python. From scheduling reminders to managing tasks, this article will walk you through every step so that you can build your own task manager and keep everything organized effortlessly.
Let’s get started!
Table of Contents
- Getting Your Telegram Bot Token
- Scheduler and Task Storage
- Task Management Functions
- Command Handlers for Your Automated Telegram Bot
- Send Reminder and Main Function
- Main Block
- Running the Code
- Full Code
Before we dive into the code, let’s make sure we’ve got everything we need:
To get this code running smoothly, you’ll need to install a few libraries. Just open your terminal or command prompt and run these commands:
$ pip install python-telegram-bot==13.14
$ pip install apscheduler
$ pip install pytz
Getting Your Telegram Bot Token
None of this will work without your Telegram token. So, here’s what you need to do:
- First, search for the official BotFather on Telegram.
- Next, enter the
/startcommand to see your options.
- Now, create a new bot by using the
/newbotcommand. You’ll need to enter a name and a username for your bot. The username must end with_bot, like I used “ThePycodes_bot”. After setting this up, you’ll receive your Telegram Token, which you’ll need for your code.
Imports
Let’s import the essential tools we’ll need:
- Tracking Bot Activities: To keep tabs on what’s happening with our bot, we’ll start by setting up logging.
import logging
- Handling Dates and Times: We’ll use
datetimeandpytzto manage and schedule tasks.
from datetime import datetime
import pytz
- Scheduling Tasks: To make sure our reminders go off at the right times, we’ll need
apschedulerfor scheduling.
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.base import JobLookupError
- Interacting with Telegram: Finally, to enable our bot to send and receive messages and handle commands, we’ll use the
telegramlibrary, which works with the Telegram API.
from telegram import Update, Bot
from telegram.ext import Updater, CommandHandler, CallbackContext
Now that we have our tools ready, the next step is to set up logging. This will help us keep track of everything our bot does. We’ll configure the logging to capture messages in a specific format, which includes the message content, the type of message, the time it occurred, and the bot’s name.
# Setup logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
Scheduler and Task Storage
First off, to manage all our tasks and ensure reminders are sent at the right time, we rely on apscheduler. This library is fantastic because it runs tasks in the background and schedules reminders without disrupting other processes. To get started, we initialize the BackgroundScheduler from apscheduler and start it up. This scheduler will handle executing tasks at the correct times.
# Initialize scheduler
scheduler = BackgroundScheduler()
scheduler.start()
Additionally, we use a straightforward dictionary named tasks_db to keep track of all the tasks for each user, ensuring everything is well-organized and ready to go.
# In-memory storage for tasks
tasks_db = {}
Task Management Functions
Since the main job of our bot is handling automated tasks, we need functions to manage them. As such, we made these functions:
Add Task
You can’t have tasks without adding them first. That’s where the add_task() function comes in. It checks if the user exists in the tasks_db dictionary; if not, it creates an empty list for them. Then, it adds a task by following these steps: assigning a unique ID, setting the scheduled time, adding the task description, and finally marking it as pending.
# Add a task to the in-memory storage
def add_task(user_id, task_time, task_description):
if user_id not in tasks_db:
tasks_db[user_id] = []
task_id = len(tasks_db[user_id]) + 1
tasks_db[user_id].append({
'id': task_id,
'task_time': task_time,
'task_description': task_description,
'status': 'Pending'
})
Get Tasks
If a user wants to see their tasks, the get_tasks() function is here to help. It checks if the user has any tasks saved in our tasks_db dictionary. If there are tasks, it fetches and returns them. If not, it simply returns an empty list. It’s like asking your assistant to check your to-do list: if it’s there, you’ll see it; if not, it’s blank.
# Fetch tasks from the in-memory storage
def get_tasks(user_id):
return tasks_db.get(user_id, [])
Command Handlers for Your Automated Telegram Bot
Let’s talk about how our bot handles commands. Just like having a chat with a friend who knows exactly what you need, our bot listens for specific commands and knows just how to respond. Here’s how it does that:
Start Command
Typing /start triggers the start() function. This function greets the user and provides some basic instructions on how to add a task. It’s like a friendly introduction and quick guide all in one!
# Function to handle the /start command
def start(update: Update, context: CallbackContext) -> None:
update.message.reply_text(
'Welcome to The Pycodes Task Manager Bot! Use /add <date> <time> <description> to add a task.')
Add Command
When you want to add a task, the add() function steps in to make sure everything is in order. First, it checks if all the necessary details were included in the /add command—date, time, and description. If any of these are missing, it sends a helpful usage guide so you know what to provide.
# Function to handle the /add command
def add(update: Update, context: CallbackContext) -> None:
logging.info(f"Arguments received: {context.args}")
if len(context.args) < 3:
update.message.reply_text("Usage: /add <date> <time> <description>")
return
Next, it extracts the date, time, and task description from your command. It combines the date and time into one datetime object and adjusts it to your specified timezone (don’t forget to set your timezone here).
try:
task_date_str = context.args[0] # Expecting date in YYYY-MM-DD format
task_time_str = context.args[1] # Expecting time in HH:MM:SS format
task_description = ' '.join(context.args[2:]) # Join the remaining arguments as the description
logging.info(f"Extracted date: {task_date_str}")
logging.info(f"Extracted time: {task_time_str}")
logging.info(f"Extracted description: {task_description}")
task_datetime_str = f"{task_date_str} {task_time_str}"
task_time = datetime.strptime(task_datetime_str, '%Y-%m-%d %H:%M:%S')
local_tz = pytz.timezone('Africa/Algiers')# Enter Your Timezone
task_time = local_tz.localize(task_time)
With the details set, it stores the task using add_task() and schedules a reminder with scheduler.add_job(). This way, your task will pop up exactly when it’s supposed to.
user_id = update.message.from_user.id
add_task(user_id, task_time, task_description)
scheduler.add_job(
send_reminder,
'date',
run_date=task_time,
args=[update.message.chat_id, task_description],
id=f"{user_id}_{len(tasks_db[user_id])}",
timezone='Africa/Algiers' # Enter Your Timezone
)
update.message.reply_text(f"Task added for {task_time.strftime('%Y-%m-%d %H:%M:%S')}.")
If anything goes wrong, such as missing or incorrectly formatted input, it catches the error and sends the usage guide again.
except (IndexError, ValueError) as e:
logging.error(f"Error in /add command: {e}")
update.message.reply_text("Usage: /add <date> <time> <description>")
This makes sure your task is added smoothly and scheduled just as you wanted!
Complete Command
Once a task is completed, we want to update its status from “pending” to “completed“. We do this using the /complete [task_id] command. This command triggers the complete() function, which retrieves and converts the task ID to an integer. The function then searches through the stored tasks to find the one with the matching ID. When it finds the task, it updates its status to “completed” and informs the user with a confirmation message.
# Function to handle the /complete command
def complete(update: Update, context: CallbackContext) -> None:
try:
task_id = int(context.args[0])
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
task_found = False
for task in tasks:
if task['id'] == task_id:
task['status'] = 'Completed'
task_found = True
break
if task_found:
update.message.reply_text(f"Task {task_id} marked as complete.")
else:
update.message.reply_text("Task not found.")
except (IndexError, ValueError):
update.message.reply_text("Usage: /complete <task_id>")
View Command
If you’re curious about your saved tasks, just type /view. This will activate the view() function, which fetches your tasks using get_tasks(). It then puts all your tasks into a neat, organized message and sends it to you. If you don’t have any tasks on your list, don’t worry—the bot will let you know that your task list is empty.
# Function to handle the /view command
def view(update: Update, context: CallbackContext) -> None:
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
if tasks:
response = '\n'.join(
[f"Task ID: {task['id']} - {task['task_time']} - {task['task_description']} (Status: {task['status']})" for
task in tasks])
update.message.reply_text(f"Your tasks:\n{response}")
else:
update.message.reply_text("No tasks found.")
Remove Command
Deleting a task is straightforward with the /remove [task_id] command. This command triggers the remove() function, which searches for the task ID in the task_db dictionary. If the task is found, it’s removed using scheduler.remove_job(). The bot will then inform you whether the task was successfully removed or if it wasn’t found. Any errors during this process will also be communicated to you.
# Function to handle the /remove command
def remove(update: Update, context: CallbackContext) -> None:
if update.message is None:
update.message.reply_text("Error: No message context available.")
return
try:
if len(context.args) != 1:
update.message.reply_text("Usage: /remove <task_id>")
return
task_id = int(context.args[0])
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
task_found = False
for task in tasks:
if task['id'] == task_id:
tasks_db[user_id] = [t for t in tasks if t['id'] != task_id]
job_id = f"{user_id}_{task_id}"
try:
scheduler.remove_job(job_id)
except JobLookupError:
logging.warning(f"Job {job_id} not found for removal.")
task_found = True
break
if task_found:
update.message.reply_text(f"Task {task_id} removed successfully.")
else:
update.message.reply_text("Task not found.")
except (IndexError, ValueError) as e:
logging.error(f"Error in /remove command: {e}")
update.message.reply_text("Usage: /remove <task_id>")
Help Command
Need help? Just type /help! It will give you a quick rundown of all the commands you can use with the bot.
# Function to handle the /help command
def help(update: Update, context: CallbackContext) -> None:
help_text = (
"Here are the commands you can use:\n"
"/start - Start running the bot and get a welcome message.\n"
"/add <date> <time> <description> - Add a new timestamped task.\n"
"/complete <task_id> - Mark a task as completed.\n"
"/view - View all your tasks.\n"
"/remove <task_id> - Remove a task.\n"
)
update.message.reply_text(help_text)
Send Reminder and Main Function
When it’s time for a task, the send_reminder() function sends a message to the user, reminding them about the task. It uses the bot to send a message with the task description to the specified chat ID.
# Function to send reminders
def send_reminder(chat_id, task_description):
bot = Bot(token="ENTER_YOUR_TELEGRAM_BOT_TOKEN")
bot.send_message(chat_id, f"Reminder: {task_description}")
To tie everything together, we use the main() function. This function sets up the bot using Updater with the Telegram API token, allowing it to listen for user commands and call the appropriate functions to handle them. The updater.start_polling() method keeps the bot actively checking for new messages and commands, while updater.idle() ensures the bot stays running until you manually stop it.
def main() -> None:
updater = Updater("ENTER_YOUR_TELEGRAM_BOT_TOKEN")
dp = updater.dispatcher
dp.add_handler(CommandHandler("start", start))
dp.add_handler(CommandHandler("add", add))
dp.add_handler(CommandHandler("complete", complete))
dp.add_handler(CommandHandler("view", view))
dp.add_handler(CommandHandler("remove", remove))
dp.add_handler(CommandHandler("help", help)) # Add the /help command handler
updater.start_polling()
updater.idle()
Main Block
This part makes sure the bot starts up only when you run this code directly, not when it’s imported into another script.
if __name__ == '__main__':
main()
Running the Code
Full Code
import logging
from datetime import datetime
import pytz
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.base import JobLookupError
from telegram import Update, Bot
from telegram.ext import Updater, CommandHandler, CallbackContext
# Setup logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
# Initialize scheduler
scheduler = BackgroundScheduler()
scheduler.start()
# In-memory storage for tasks
tasks_db = {}
# Add a task to the in-memory storage
def add_task(user_id, task_time, task_description):
if user_id not in tasks_db:
tasks_db[user_id] = []
task_id = len(tasks_db[user_id]) + 1
tasks_db[user_id].append({
'id': task_id,
'task_time': task_time,
'task_description': task_description,
'status': 'Pending'
})
# Fetch tasks from the in-memory storage
def get_tasks(user_id):
return tasks_db.get(user_id, [])
# Function to handle the /start command
def start(update: Update, context: CallbackContext) -> None:
update.message.reply_text(
'Welcome to The Pycodes Task Manager Bot! Use /add <date> <time> <description> to add a task.')
# Function to handle the /add command
def add(update: Update, context: CallbackContext) -> None:
logging.info(f"Arguments received: {context.args}")
if len(context.args) < 3:
update.message.reply_text("Usage: /add <date> <time> <description>")
return
try:
task_date_str = context.args[0] # Expecting date in YYYY-MM-DD format
task_time_str = context.args[1] # Expecting time in HH:MM:SS format
task_description = ' '.join(context.args[2:]) # Join the remaining arguments as the description
logging.info(f"Extracted date: {task_date_str}")
logging.info(f"Extracted time: {task_time_str}")
logging.info(f"Extracted description: {task_description}")
task_datetime_str = f"{task_date_str} {task_time_str}"
task_time = datetime.strptime(task_datetime_str, '%Y-%m-%d %H:%M:%S')
local_tz = pytz.timezone('Africa/Algiers')# Enter Your Timezone
task_time = local_tz.localize(task_time)
user_id = update.message.from_user.id
add_task(user_id, task_time, task_description)
scheduler.add_job(
send_reminder,
'date',
run_date=task_time,
args=[update.message.chat_id, task_description],
id=f"{user_id}_{len(tasks_db[user_id])}",
timezone='Africa/Algiers' # Enter Your Timezone
)
update.message.reply_text(f"Task added for {task_time.strftime('%Y-%m-%d %H:%M:%S')}.")
except (IndexError, ValueError) as e:
logging.error(f"Error in /add command: {e}")
update.message.reply_text("Usage: /add <date> <time> <description>")
# Function to handle the /complete command
def complete(update: Update, context: CallbackContext) -> None:
try:
task_id = int(context.args[0])
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
task_found = False
for task in tasks:
if task['id'] == task_id:
task['status'] = 'Completed'
task_found = True
break
if task_found:
update.message.reply_text(f"Task {task_id} marked as complete.")
else:
update.message.reply_text("Task not found.")
except (IndexError, ValueError):
update.message.reply_text("Usage: /complete <task_id>")
# Function to handle the /view command
def view(update: Update, context: CallbackContext) -> None:
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
if tasks:
response = '\n'.join(
[f"Task ID: {task['id']} - {task['task_time']} - {task['task_description']} (Status: {task['status']})" for
task in tasks])
update.message.reply_text(f"Your tasks:\n{response}")
else:
update.message.reply_text("No tasks found.")
# Function to handle the /remove command
def remove(update: Update, context: CallbackContext) -> None:
if update.message is None:
update.message.reply_text("Error: No message context available.")
return
try:
if len(context.args) != 1:
update.message.reply_text("Usage: /remove <task_id>")
return
task_id = int(context.args[0])
user_id = update.message.from_user.id
tasks = get_tasks(user_id)
task_found = False
for task in tasks:
if task['id'] == task_id:
tasks_db[user_id] = [t for t in tasks if t['id'] != task_id]
job_id = f"{user_id}_{task_id}"
try:
scheduler.remove_job(job_id)
except JobLookupError:
logging.warning(f"Job {job_id} not found for removal.")
task_found = True
break
if task_found:
update.message.reply_text(f"Task {task_id} removed successfully.")
else:
update.message.reply_text("Task not found.")
except (IndexError, ValueError) as e:
logging.error(f"Error in /remove command: {e}")
update.message.reply_text("Usage: /remove <task_id>")
# Function to handle the /help command
def help(update: Update, context: CallbackContext) -> None:
help_text = (
"Here are the commands you can use:\n"
"/start - Start running the bot and get a welcome message.\n"
"/add <date> <time> <description> - Add a new timestamped task.\n"
"/complete <task_id> - Mark a task as completed.\n"
"/view - View all your tasks.\n"
"/remove <task_id> - Remove a task.\n"
)
update.message.reply_text(help_text)
# Function to send reminders
def send_reminder(chat_id, task_description):
bot = Bot(token="ENTER_YOUR_TELEGRAM_BOT_TOKEN")
bot.send_message(chat_id, f"Reminder: {task_description}")
def main() -> None:
updater = Updater("ENTER_YOUR_TELEGRAM_BOT_TOKEN")
dp = updater.dispatcher
dp.add_handler(CommandHandler("start", start))
dp.add_handler(CommandHandler("add", add))
dp.add_handler(CommandHandler("complete", complete))
dp.add_handler(CommandHandler("view", view))
dp.add_handler(CommandHandler("remove", remove))
dp.add_handler(CommandHandler("help", help)) # Add the /help command handler
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
Happy Coding!
