A flexible task scheduling system for Telegram bots that allows you to run scripts, commands, and functions at specified intervals.
Clone this repository:
git clone https://github.com/yourusername/telegram-bot-scheduler.git
cd telegram-bot-scheduler
Install dependencies:
pip install python-telegram-bot python-dotenv schedule
Create a .env file with your Telegram bot token:
TELEGRAM_BOT_TOKEN=your_bot_token_here
All scheduled tasks are configured in the config.json file under the scheduled_tasks section:
{
"admin_users": [123456789],
"notification_users": [123456789],
"scheduled_tasks": [
{
"name": "daily_report",
"schedule": "every day at 9:00",
"task_type": "script",
"script_path": "scripts/daily_report.py",
"enabled": true,
"notify_on_completion": true,
"notify_on_error": true,
"timeout": 300
}
]
}
name: A unique identifier for the taskschedule: When the task should run (e.g., "every day at 9:00")task_type: Type of task ("script" or "command")script_path: Path to a Python script (for "script" type)command: Shell command to execute (for "command" type)enabled: Whether the task is activenotify_on_completion: Whether to send a notification when the task completes successfullynotify_on_error: Whether to send a notification when the task failstimeout: Maximum execution time in secondsevery minute - Run every minuteevery hour - Run every hourevery day at 9:00 - Run daily at 9:00 AMevery monday at 3:00 - Run weekly on Mondays at 3:00 AMevery month at 9:00 - Run on the first day of each month at 9:00 AMRun the main script to start the bot and scheduler:
python main.py
You can manage tasks using the manage_tasks.py script:
# List all tasks
python manage_tasks.py list
# Show details of a specific task
python manage_tasks.py show daily_report
# Run a task immediately
python manage_tasks.py run daily_report
# Enable a task
python manage_tasks.py enable weekly_backup
# Disable a task
python manage_tasks.py disable hourly_check
# Reload tasks from config.json after making changes
python manage_tasks.py reload
If you're an admin user, you can manage tasks directly through the Telegram bot:
/tasks - List all scheduled tasks
/task_status daily_report - Show details of a specific task
/enable_task weekly_backup - Enable a task
/disable_task hourly_check - Disable a task
{
"name": "daily_report",
"schedule": "every day at 9:00",
"task_type": "script",
"script_path": "scripts/daily_report.py",
"enabled": true,
"notify_on_completion": true,
"notify_on_error": true,
"timeout": 300
}
{
"name": "hourly_check",
"schedule": "every hour",
"task_type": "command",
"command": "python scripts/hourly_check.py",
"enabled": true,
"notify_on_completion": false,
"notify_on_error": true,
"timeout": 60
}
{
"name": "weekly_backup",
"schedule": "every monday at 3:00",
"task_type": "command",
"command": "bash scripts/backup.sh",
"enabled": true,
"notify_on_completion": true,
"notify_on_error": true,
"timeout": 1800
}
scripts/ directory:#!/usr/bin/env python3
import os
import sys
from notify import NotificationClient
# Add parent directory to path so we can import from the main project
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
def main():
# Your task logic here
result = "Task completed successfully!"
# Send notification
client = NotificationClient()
client.broadcast_notification(result)
return 0
if __name__ == "__main__":
sys.exit(main())
chmod +x scripts/your_script.py
config.json:{
"name": "your_custom_task",
"schedule": "every day at 12:00",
"task_type": "script",
"script_path": "scripts/your_script.py",
"enabled": true,
"notify_on_completion": true,
"notify_on_error": true,
"timeout": 300
}
Add a command task to config.json:
{
"name": "disk_cleanup",
"schedule": "every sunday at 2:00",
"task_type": "command",
"command": "find /tmp -type f -atime +7 -delete",
"enabled": true,
"notify_on_completion": true,
"notify_on_error": true,
"timeout": 600
}
Let's create a task that checks disk space and sends an alert if it's low:
scripts/check_disk_space.py:#!/usr/bin/env python3
import os
import sys
import subprocess
# Add parent directory to path so we can import from the main project
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from notify import NotificationClient
def main():
# Get disk usage
df = subprocess.check_output(['df', '-h', '/']).decode('utf-8')
lines = df.strip().split('\n')
usage_line = lines[1]
parts = usage_line.split()
usage_percent = int(parts[4].strip('%'))
# Send alert if disk usage is above 80%
if usage_percent > 80:
client = NotificationClient()
message = f"⚠️ WARNING: Disk usage is at {usage_percent}%!\n{usage_line}"
client.broadcast_notification(message)
print(f"Alert sent: {message}")
else:
print(f"Disk usage is normal: {usage_percent}%")
return 0
if __name__ == "__main__":
sys.exit(main())
chmod +x scripts/check_disk_space.py
config.json:{
"name": "disk_space_check",
"schedule": "every day at 8:00",
"task_type": "script",
"script_path": "scripts/check_disk_space.py",
"enabled": true,
"notify_on_completion": false,
"notify_on_error": true,
"timeout": 60
}
python manage_tasks.py reload
config.jsontelegram_bot.log) for errorspython manage_tasks.py run task_namechmod +x scripts/your_script.py)MIT
Contributions are welcome! Please feel free to submit a Pull Request.