Building Telegram Bots with Python: A Complete Guide

pip install python-telegram-bot

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

# Replace with your token
TOKEN = 'YOUR_BOT_TOKEN_HERE'

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Send a message when the command /start is issued."""
    await update.message.reply_text(
        'Hi! I am your first bot. Send me any message and I will echo it back!'
    )

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Echo the user message."""
    await update.message.reply_text(update.message.text)

def main():
    # Create the Application
    application = Application.builder().token(TOKEN).build()

    # Register handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

    # Start the bot
    print("Bot is running...")
    application.run_polling()

if __name__ == '__main__':
    main()

python my_first_bot.py

await context.bot.send_message(chat_id=chat_id, text="Hello!")

# With formatting
await context.bot.send_message(
    chat_id=chat_id,
    text="*Bold* and _italic_ text",
    parse_mode='Markdown'
)

await context.bot.send_photo(
    chat_id=chat_id,
    photo='https://example.com/image.jpg',
    caption='Check out this photo!'
)

# From local file
with open('photo.jpg', 'rb') as photo:
    await context.bot.send_photo(chat_id=chat_id, photo=photo)

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "Available commands:\n"
        "/start - Start the bot\n"
        "/help - Show this help message\n"
        "/weather - Get weather info"
    )

application.add_handler(CommandHandler("help", help_command))

async def weather(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if context.args:
        city = ' '.join(context.args)
        await update.message.reply_text(f"Getting weather for {city}...")
    else:
        await update.message.reply_text("Please specify a city: /weather London")

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("Option 1", callback_data='opt1'),
            InlineKeyboardButton("Option 2", callback_data='opt2')
        ],
        [InlineKeyboardButton("Visit Website", url='https://example.com')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    
    await update.message.reply_text(
        'Choose an option:',
        reply_markup=reply_markup
    )

async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # Acknowledge the button press
    
    if query.data == 'opt1':
        await query.edit_message_text('You chose Option 1!')
    elif query.data == 'opt2':
        await query.edit_message_text('You chose Option 2!')

# Register handlers
application.add_handler(CommandHandler('menu', menu))
application.add_handler(CallbackQueryHandler(button_callback))

from telegram import ReplyKeyboardMarkup, KeyboardButton

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [KeyboardButton("Weather"), KeyboardButton("News")],
        [KeyboardButton("Help")]
    ]
    reply_markup = ReplyKeyboardMarkup(
        keyboard,
        resize_keyboard=True,
        one_time_keyboard=False
    )

    await update.message.reply_text(
        'Welcome! Choose an option:',
        reply_markup=reply_markup
    )

from telegram.ext import ConversationHandler

# Define states
NAME, AGE = range(2)

async def start_registration(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("What's your name?")
    return NAME

async def get_name(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['name'] = update.message.text
    await update.message.reply_text("How old are you?")
    return AGE

async def get_age(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['age'] = update.message.text
    name = context.user_data['name']
    age = context.user_data['age']
    
    await update.message.reply_text(
        f"Registration complete!\nName: {name}\nAge: {age}"
    )
    return ConversationHandler.END

async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Registration cancelled.")
    return ConversationHandler.END

# Create conversation handler
conv_handler = ConversationHandler(
    entry_points=[CommandHandler('register', start_registration)],
    states={
        NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_name)],
        AGE: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_age)]
    },
    fallbacks=[CommandHandler('cancel', cancel)]
)

application.add_handler(conv_handler)

from telegram import InlineQueryResultArticle, InputTextMessageContent

async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.inline_query.query

    if not query:
        return

    results = [
        InlineQueryResultArticle(
            id='1',
            title='Result 1',
            input_message_content=InputTextMessageContent(
                f"You searched for: {query}"
            ),
            description='Click to send this result'
        )
    ]

    await update.inline_query.answer(results)

application.add_handler(InlineQueryHandler(inline_query))

async def handle_document(update: Update, context: ContextTypes.DEFAULT_TYPE):
    document = update.message.document
    file = await document.get_file()
    await file.download_to_drive(f'{document.file_name}')

    await update.message.reply_text(
        f"Downloaded: {document.file_name}\n"
        f"Size: {document.file_size} bytes"
    )

application.add_handler(MessageHandler(filters.Document.ALL, handle_document))

async def new_member(update: Update, context: ContextTypes.DEFAULT_TYPE):
    for member in update.message.new_chat_members:
        await update.message.reply_text(
            f"Welcome {member.first_name}!"
        )

application.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, new_member))

async def admin_only_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    chat = update.effective_chat

    member = await context.bot.get_chat_member(chat.id, user.id)

    if member.status in ['creator', 'administrator']:
        await update.message.reply_text("Admin command executed!")
    else:
        await update.message.reply_text("This command is for admins only!")

await update.message.reply_text(
    "*Bold text*\n"
    "_Italic text_\n"
    "[Link](https://example.com)\n"
    "`Code`\n"
    "```python\n"
    "def hello():\n"
    "    print('Hello')\n"
    "```",
    parse_mode='Markdown'
)

await update.message.reply_text(
    "<b>Bold text</b>\n"
    "<i>Italic text</i>\n"
    "<a href='https://example.com'>Link</a>\n"
    "<code>Code</code>\n"
    "<pre>Preformatted</pre>",
    parse_mode='HTML'
)

async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    print(f'Update {update} caused error {context.error}')

    if update and update.effective_message:
        await update.effective_message.reply_text(
            "Sorry, an error occurred. Please try again later."
        )

application.add_error_handler(error_handler)

# Python example with rate limiting
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_calls=20, period=60):
        self.calls = defaultdict(list)
        self.max_calls = max_calls
        self.period = period
    
    async def wait_if_needed(self, user_id):
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.period)
        
        self.calls[user_id] = [
            call_time for call_time in self.calls[user_id]
            if call_time > cutoff
        ]
        
        if len(self.calls[user_id]) >= self.max_calls:
            sleep_time = (self.calls[user_id][0] - cutoff).total_seconds()
            await asyncio.sleep(sleep_time)
        
        self.calls[user_id].append(now)

rate_limiter = RateLimiter()

async def rate_limited_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await rate_limiter.wait_if_needed(update.effective_user.id)
    await update.message.reply_text("Command executed!")

# Python with SQLite
import sqlite3
from telegram import Update
from telegram.ext import ContextTypes

class Database:
    def __init__(self, db_file='bot_data.db'):
        self.conn = sqlite3.connect(db_file, check_same_thread=False)
        self.create_tables()
    
    def create_tables(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                user_id INTEGER PRIMARY KEY,
                username TEXT,
                first_name TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS user_scores (
                user_id INTEGER,
                score INTEGER DEFAULT 0,
                FOREIGN KEY (user_id) REFERENCES users (user_id)
            )
        ''')
        self.conn.commit()
    
    def add_user(self, user_id, username, first_name):
        cursor = self.conn.cursor()
        cursor.execute(
            'INSERT OR IGNORE INTO users (user_id, username, first_name) VALUES (?, ?, ?)',
            (user_id, username, first_name)
        )
        cursor.execute(
            'INSERT OR IGNORE INTO user_scores (user_id) VALUES (?)',
            (user_id,)
        )
        self.conn.commit()
    
    def update_score(self, user_id, points):
        cursor = self.conn.cursor()
        cursor.execute(
            'UPDATE user_scores SET score = score + ? WHERE user_id = ?',
            (points, user_id)
        )
        self.conn.commit()
    
    def get_score(self, user_id):
        cursor = self.conn.cursor()
        cursor.execute('SELECT score FROM user_scores WHERE user_id = ?', (user_id,))
        result = cursor.fetchone()
        return result[0] if result else 0
    
    def get_leaderboard(self, limit=10):
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT u.first_name, s.score 
            FROM users u 
            JOIN user_scores s ON u.user_id = s.user_id 
            ORDER BY s.score DESC 
            LIMIT ?
        ''', (limit,))
        return cursor.fetchall()

# Initialize database
db = Database()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    db.add_user(user.id, user.username, user.first_name)
    await update.message.reply_text(f"Welcome {user.first_name}!")

async def add_points(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    db.update_score(user_id, 10)
    score = db.get_score(user_id)
    await update.message.reply_text(f"You earned 10 points! Total: {score}")

async def leaderboard(update: Update, context: ContextTypes.DEFAULT_TYPE):
    leaders = db.get_leaderboard()
    text = "🏆 Leaderboard:\n\n"
    for i, (name, score) in enumerate(leaders, 1):
        text += f"{i}. {name}: {score} points\n"
    await update.message.reply_text(text)

# Python with Redis
import redis
import json

class RedisStorage:
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis = redis.Redis(host=host, port=port, db=db, decode_responses=True)
    
    def set_user_state(self, user_id, state):
        self.redis.set(f'user:{user_id}:state', state, ex=3600)  # 1 hour expiry
    
    def get_user_state(self, user_id):
        return self.redis.get(f'user:{user_id}:state')
    
    def delete_user_state(self, user_id):
        self.redis.delete(f'user:{user_id}:state')
    
    def cache_data(self, key, data, expiry=300):
        self.redis.setex(key, expiry, json.dumps(data))
    
    def get_cached_data(self, key):
        data = self.redis.get(key)
        return json.loads(data) if data else None

# Usage
redis_store = RedisStorage()

async def start_conversation(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    redis_store.set_user_state(user_id, 'awaiting_name')
    await update.message.reply_text("Please enter your name:")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    state = redis_store.get_user_state(user_id)
    
    if state == 'awaiting_name':
        name = update.message.text
        redis_store.set_user_state(user_id, 'awaiting_age')
        # Store name temporarily
        redis_store.cache_data(f'user:{user_id}:name', name)
        await update.message.reply_text(f"Hello {name}! Now enter your age:")

    elif state == 'awaiting_age':
        age = update.message.text
        name = redis_store.get_cached_data(f'user:{user_id}:name')
        redis_store.delete_user_state(user_id)
        await update.message.reply_text(f"Thanks {name}! Age {age} saved.")

from telegram.ext import ApplicationBuilder
import ssl

async def post_init(application):
    await application.bot.set_webhook(
        url='https://yourdomain.com/webhook',
        certificate=open('/path/to/cert.pem', 'rb')
    )

def main():
    application = ApplicationBuilder() \
        .token(TOKEN) \
        .post_init(post_init) \
        .build()
    
    # Add your handlers here
    application.add_handler(CommandHandler("start", start))
    
    # For production with webhook
    application.run_webhook(
        listen='0.0.0.0',
        port=8443,
        url_path='webhook',
        key='private.key',
        cert='cert.pem',
        webhook_url='https://yourdomain.com/webhook'
    )

if __name__ == '__main__':
    main()

web: python bot.py
worker: python worker.py

python-telegram-bot==20.7
pymongo==4.5.0
redis==5.0.1
requests==2.31.0

# lambda_function.py
import json
import os
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

TOKEN = os.environ['BOT_TOKEN']

application = Application.builder().token(TOKEN).build()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Hello from Lambda!")

application.add_handler(CommandHandler("start", start))

def lambda_handler(event, context):
    try:
        application.initialize()
        update = Update.de_json(json.loads(event['body']), application.bot)
        application.process_update(update)
        
        return {
            'statusCode': 200,
            'body': json.dumps('Success')
        }
    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps('Error')
        }

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
CMD ["python", "bot.py"]

version: '3.8'
services:
  bot:
    build: .
    environment:
      - BOT_TOKEN=${BOT_TOKEN}
      - DATABASE_URL=${DATABASE_URL}
    restart: unless-stopped
  redis:
    image: redis:alpine
    restart: unless-stopped

BOT_TOKEN=your_bot_token_here
DATABASE_URL=mongodb://localhost:27017
REDIS_URL=redis://localhost:6379
API_KEY=your_external_api_key

import os
from dotenv import load_dotenv

load_dotenv()

TOKEN = os.getenv('BOT_TOKEN')
DATABASE_URL = os.getenv('DATABASE_URL')

import re

def sanitize_input(text):
    # Remove potentially dangerous characters
    cleaned = re.sub(r'[<>&\"\']', '', text)
    return cleaned.strip()

async def safe_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_input = update.message.text
    safe_input = sanitize_input(user_input)

    # Process safe_input...

# Simple whitelist approach
ALLOWED_USERS = [123456789, 987654321]  # User IDs

async def restricted_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    
    if user_id not in ALLOWED_USERS:
        await update.message.reply_text("Unauthorized access.")
        return

    # Proceed with command...

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_requests=10, window=60):
        self.requests = defaultdict(list)
        self.max_requests = max_requests
        self.window = window
    
    def is_allowed(self, user_id):
        now = time.time()
        user_requests = self.requests[user_id]
        
        # Remove old requests
        user_requests = [req_time for req_time in user_requests 
                        if now - req_time < self.window]
        self.requests[user_id] = user_requests
        
        if len(user_requests) < self.max_requests:
            user_requests.append(now)
            return True
        return False

rate_limiter = RateLimiter()

async def rate_limited_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    
    if not rate_limiter.is_allowed(user_id):
        await update.message.reply_text("Rate limit exceeded. Please wait.")
        return

    # Process the request...

my_telegram_bot/
├── bot.py              # Main bot file
├── handlers/           # Command handlers
│   ├── __init__.py
│   ├── start.py
│   ├── admin.py
│   └── user.py
├── models/             # Database models
│   ├── __init__.py
│   └── user.py
├── utils/              # Utilities
│   ├── __init__.py
│   └── helpers.py
├── config.py           # Configuration
└── requirements.txt

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    BOT_TOKEN = os.getenv('BOT_TOKEN')
    DATABASE_URL = os.getenv('DATABASE_URL')
    REDIS_URL = os.getenv('REDIS_URL')
    ADMIN_IDS = [int(x) for x in os.getenv('ADMIN_IDS', '').split(',')]
    
    # Bot settings
    MAX_MESSAGE_LENGTH = 4096
    RATE_LIMIT = 30  # messages per minute

from telegram import Update
from telegram.ext import ContextTypes

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    welcome_text = f"""
Hello {user.first_name}!

I'm your friendly Telegram bot. Here's what I can do:

/help - Show all commands
/search - Search for information
/settings - Configure your preferences

Feel free to explore!
    """
    await update.message.reply_text(welcome_text)

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    help_text = """
Available Commands:

Basic Commands:
/start - Start the bot
/help - Show this help message

Utility Commands:
/weather <city> - Get weather information
/calc <expression> - Calculate math expressions

Admin Commands:
/stats - Bot statistics (admin only)
/broadcast - Send message to all users
    """
    await update.message.reply_text(help_text)

from telegram.ext import Application, CommandHandler, MessageHandler, filters
from config import Config
from handlers.start import start, help_command
from handlers.admin import stats, broadcast
from handlers.user import weather, calculator

def main():
    application = Application.builder().token(Config.BOT_TOKEN).build()
    
    # Add handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("stats", stats))
    application.add_handler(CommandHandler("weather", weather))
    application.add_handler(CommandHandler("calc", calculator))
    
    # Error handling
    application.add_error_handler(error_handler)
    
    # Start the bot
    application.run_polling()

async def error_handler(update, context):
    print(f"Exception while handling an update: {context.error}")

if __name__ == '__main__':
    main()

import logging
from datetime import datetime

# Setup logging
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO,
    handlers=[
        logging.FileHandler('bot.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class BotMetrics:
    def __init__(self):
        self.commands_processed = 0
        self.messages_received = 0
        self.errors_occurred = 0
        self.start_time = datetime.now()
    
    def log_command(self, command, user_id):
        self.commands_processed += 1
        logger.info(f"Command {command} from user {user_id}")
    
    def log_message(self, user_id):
        self.messages_received += 1
    
    def log_error(self, error):
        self.errors_occurred += 1
        logger.error(f"Bot error: {error}")

metrics = BotMetrics()

async def monitored_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    metrics.log_command('start', update.effective_user.id)
    await start(update, context)

# task_bot.py
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    Application, CommandHandler, CallbackQueryHandler, 
    ConversationHandler, MessageHandler, filters, ContextTypes
)
from datetime import datetime, timedelta
import sqlite3
import json

# Conversation states
TASK_TITLE, TASK_DESCRIPTION, TASK_DUE_DATE = range(3)

class TaskManager:
    def __init__(self, db_file='tasks.db'):
        self.conn = sqlite3.connect(db_file, check_same_thread=False)
        self.create_tables()
    
    def create_tables(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER,
                title TEXT NOT NULL,
                description TEXT,
                due_date TIMESTAMP,
                status TEXT DEFAULT 'pending',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    
    def add_task(self, user_id, title, description=None, due_date=None):
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT INTO tasks (user_id, title, description, due_date)
            VALUES (?, ?, ?, ?)
        ''', (user_id, title, description, due_date))
        self.conn.commit()
        return cursor.lastrowid
    
    def get_user_tasks(self, user_id, status=None):
        cursor = self.conn.cursor()
        if status:
            cursor.execute(
                'SELECT * FROM tasks WHERE user_id = ? AND status = ? ORDER BY created_at DESC',
                (user_id, status)
            )
        else:
            cursor.execute(
                'SELECT * FROM tasks WHERE user_id = ? ORDER BY created_at DESC',
                (user_id,)
            )
        return cursor.fetchall()
    
    def update_task_status(self, task_id, status):
        cursor = self.conn.cursor()
        cursor.execute(
            'UPDATE tasks SET status = ? WHERE id = ?',
            (status, task_id)
        )
        self.conn.commit()
    
    def delete_task(self, task_id):
        cursor = self.conn.cursor()
        cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
        self.conn.commit()

task_manager = TaskManager()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [InlineKeyboardButton("Add Task", callback_data='add_task')],
        [InlineKeyboardButton("My Tasks", callback_data='list_tasks')],
        [InlineKeyboardButton("Help", callback_data='help')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)

    await update.message.reply_text(
        "Task Manager Bot\n\n"
        "Manage your tasks efficiently with this bot!",
        reply_markup=reply_markup
    )

async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()
    
    if query.data == 'add_task':
        await query.edit_message_text("Please enter the task title:")
        return TASK_TITLE
    
    elif query.data == 'list_tasks':
        await show_tasks(query, context)
    
    elif query.data == 'help':
        await show_help(query)

async def show_help(query):
    help_text = """
Task Manager Bot Help

Commands:
/start - Start the bot
/tasks - List your tasks
/add - Add a new task
/stats - Your task statistics

Features:
- Add tasks with titles and descriptions
- Set due dates for tasks
- View pending and completed tasks
- Track your productivity

Use the inline keyboard to navigate easily!
    """
    await query.edit_message_text(help_text)

async def show_tasks(query, context):
    user_id = query.from_user.id
    tasks = task_manager.get_user_tasks(user_id)

    if not tasks:
        await query.edit_message_text("You don't have any tasks yet!")
        return

    pending_tasks = [t for t in tasks if t[5] == 'pending']
    completed_tasks = [t for t in tasks if t[5] == 'completed']

    text = f"Your Tasks\n\n"
    text += f"Pending: {len(pending_tasks)}\n"
    text += f"Completed: {len(completed_tasks)}\n\n"

    for task in pending_tasks[:5]:  # Show only 5 recent tasks
        text += f"Task: {task[2]}\n"
        if task[3]:
            text += f"   Description: {task[3][:50]}...\n"
        if task[4]:
            due_date = datetime.fromisoformat(task[4])
            text += f"   Due: {due_date.strftime('%Y-%m-%d')}\n"
        text += f"   [ID: {task[0]}]"
        text += "\n\n"

    keyboard = [
        [InlineKeyboardButton("Add New Task", callback_data='add_task')],
        [InlineKeyboardButton("Mark Complete", callback_data='complete_task')],
        [InlineKeyboardButton("Delete Task", callback_data='delete_task')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)

    await query.edit_message_text(text, reply_markup=reply_markup)

async def add_task_title(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['task_title'] = update.message.text
    await update.message.reply_text("Great! Now enter the task description (or send /skip to skip):")
    return TASK_DESCRIPTION

async def add_task_description(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['task_description'] = update.message.text
    await update.message.reply_text("When is this task due? (Send date as YYYY-MM-DD or /skip):")
    return TASK_DUE_DATE

async def skip_description(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['task_description'] = None
    await update.message.reply_text("When is this task due? (Send date as YYYY-MM-DD or /skip):")
    return TASK_DUE_DATE

async def add_task_due_date(update: Update, context: ContextTypes.DEFAULT_TYPE):
    try:
        due_date = datetime.strptime(update.message.text, '%Y-%m-%d')
        context.user_data['due_date'] = due_date.isoformat()
    except ValueError:
        await update.message.reply_text("Invalid date format. Please use YYYY-MM-DD:")
        return TASK_DUE_DATE
    
    await save_task(update, context)
    return ConversationHandler.END

async def skip_due_date(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data['due_date'] = None
    await save_task(update, context)
    return ConversationHandler.END

async def save_task(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    title = context.user_data['task_title']
    description = context.user_data.get('task_description')
    due_date = context.user_data.get('due_date')

    task_id = task_manager.add_task(user_id, title, description, due_date)

    # Clear user data
    context.user_data.clear()

    await update.message.reply_text(
        f"Task added successfully!\n"
        f"Title: {title}\n"
        f"ID: {task_id}"
    )

async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data.clear()
    await update.message.reply_text("Task creation cancelled.")
    return ConversationHandler.END

async def tasks_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    tasks = task_manager.get_user_tasks(user_id, status='pending')

    if not tasks:
        await update.message.reply_text("You don't have any pending tasks!")
        return

    text = "Your Pending Tasks:\n\n"
    for task in tasks:
        text += f"Task: {task[2]}\n"
        if task[3]:
            text += f"   Description: {task[3]}\n"
        text += f"   [ID: {task[0]}]\n\n"

    await update.message.reply_text(text)

def main():
    application = Application.builder().token(TOKEN).build()
    
    # Conversation handler for adding tasks
    conv_handler = ConversationHandler(
        entry_points=[CallbackQueryHandler(button_handler, pattern='^add_task$')],
        states={
            TASK_TITLE: [MessageHandler(filters.TEXT & ~filters.COMMAND, add_task_title)],
            TASK_DESCRIPTION: [
                MessageHandler(filters.TEXT & ~filters.COMMAND, add_task_description),
                CommandHandler('skip', skip_description)
            ],
            TASK_DUE_DATE: [
                MessageHandler(filters.TEXT & ~filters.COMMAND, add_task_due_date),
                CommandHandler('skip', skip_due_date)
            ]
        },
        fallbacks=[CommandHandler('cancel', cancel)]
    )
    
    # Add handlers
    application.add_handler(CommandHandler('start', start))
    application.add_handler(CommandHandler('tasks', tasks_command))
    application.add_handler(conv_handler)
    application.add_handler(CallbackQueryHandler(button_handler))
    
    print("Task Manager Bot is running...")
    application.run_polling()

if __name__ == '__main__':
    main()

# Telegram Payments (Python example)
from telegram import LabeledPrice

async def create_invoice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    title = "Premium Subscription"
    description = "Get access to premium features for 1 month"
    payload = "unique-payload-for-verification"
    provider_token = "YOUR_STRIPE_TOKEN"  # From BotFather
    currency = "USD"
    prices = [LabeledPrice("Premium Subscription", 999)]  # $9.99 in cents
    
    await context.bot.send_invoice(
        chat_id, title, description, payload,
        provider_token, currency, prices
    )

# Simple dice game
async def dice_game(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = await update.message.reply_dice()
    dice_value = message.dice.value

    if dice_value == 6:
        await message.reply_text("You rolled a 6! You win!")
    else:
        await message.reply_text(f"You rolled a {dice_value}. Try again!")

# Web app button
from telegram import WebAppInfo

async def web_app_demo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [[InlineKeyboardButton(
        "Open Web App",
        web_app=WebAppInfo(url="https://your-web-app.com")
    )]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    
    await update.message.reply_text(
        "Try our web app:",
        reply_markup=reply_markup
    )

# Unit testing example
import unittest
from unittest.mock import Mock
from telegram import Update, Message, Chat, User

class TestBot(unittest.TestCase):
    def setUp(self):
        self.user = User(123, 'test_user', False)
        self.chat = Chat(123, 'private')
        self.message = Message(1, None, self.chat, from_user=self.user, text='/start')
        self.update = Update(1, message=self.message)
    
    def test_start_command(self):
        # Mock context
        context = Mock()
        
        # Test the function
        import asyncio
        asyncio.run(start(self.update, context))
        
        # Verify the response
        context.bot.send_message.assert_called_once()

if __name__ == '__main__':
    unittest.main()

# Simple analytics
class BotAnalytics:
    def __init__(self):
        self.user_actions = {}
    
    def track_event(self, user_id, event_type, metadata=None):
        if user_id not in self.user_actions:
            self.user_actions[user_id] = []
        
        self.user_actions[user_id].append({
            'timestamp': datetime.now(),
            'event_type': event_type,
            'metadata': metadata
        })
    
    def get_user_activity(self, user_id):
        return self.user_actions.get(user_id, [])
    
    def get_popular_commands(self):
        commands = {}
        for user_actions in self.user_actions.values():
            for action in user_actions:
                if action['event_type'] == 'command':
                    cmd = action['metadata']['command']
                    commands[cmd] = commands.get(cmd, 0) + 1
        return commands

analytics = BotAnalytics()