Создание бота для Telegram с использованием Python: базовые и продвинутые подходы

Расширенные примеры разработки Telegram ботов

1. Полноценный бот с регистрацией, FSM, inline клавиатурой и базой данных (python-telegram-bot)

import logging
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ConversationHandler, MessageHandler, filters
import sqlite3

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

conn = sqlite3.connect('users.db', check_same_thread=False)
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
    user_id INTEGER PRIMARY KEY,
    username TEXT,
    bio TEXT,
    age INTEGER
)''')
conn.commit()

# Состояния
BIO, AGE = range(2)

async def start(update: Update, context):
    await update.message.reply_text(
        'Привет! Я бот для сбора информации.\n'
        'Используйте /register для регистрации.'
    )

async def register(update: Update, context):
    await update.message.reply_text('Напишите коротко о себе (биография):')
    return BIO

async def bio(update: Update, context):
    context.user_data['bio'] = update.message.text
    await update.message.reply_text('Укажите ваш возраст:')
    return AGE

async def age(update: Update, context):
    try:
        age = int(update.message.text)
    except ValueError:
        await update.message.reply_text('Пожалуйста, введите число.')
        return AGE
    user = update.effective_user
    cursor.execute('INSERT OR REPLACE INTO users (user_id, username, bio, age) VALUES (?, ?, ?, ?)',
                   (user.id, user.username, context.user_data['bio'], age))
    conn.commit()
    await update.message.reply_text('Регистрация завершена!')
    return ConversationHandler.END

async def cancel(update: Update, context):
    await update.message.reply_text('Регистрация отменена.')
    return ConversationHandler.END

async def show_profile(update: Update, context):
    user = update.effective_user
    cursor.execute('SELECT * FROM users WHERE user_id = ?', (user.id,))
    row = cursor.fetchone()
    if row:
        text = f"ID: {row[0]}\nНик: {row[1]}\nБио: {row[2]}\nВозраст: {row[3]}"
    else:
        text = 'Вы не зарегистрированы. Используйте /register.'
    await update.message.reply_text(text)

async def inline_menu(update: Update, context):
    keyboard = [
        [InlineKeyboardButton('Мой профиль', callback_data='profile'),
         InlineKeyboardButton('Помощь', callback_data='help')],
        [InlineKeyboardButton('О боте', callback_data='about')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await update.message.reply_text('Меню:', reply_markup=reply_markup)

async def button_handler(update: Update, context):
    query = update.callback_query
    await query.answer()
    if query.data == 'profile':
        user = query.from_user
        cursor.execute('SELECT * FROM users WHERE user_id = ?', (user.id,))
        row = cursor.fetchone()
        text = f"ID: {row[0]}\nНик: {row[1]}\nБио: {row[2]}\nВозраст: {row[3]}" if row else 'Не зарегистрирован.'
        await query.edit_message_text(text)
    elif query.data == 'help':
        await query.edit_message_text('Доступные команды: /start, /register, /profile, /menu')
    elif query.data == 'about':
        await query.edit_message_text('Бот для тестирования написан на python-telegram-bot.')

def main():
    application = Application.builder().token('TOKEN').build()
    application.add_handler(CommandHandler('start', start))
    application.add_handler(CommandHandler('menu', inline_menu))
    application.add_handler(CommandHandler('profile', show_profile))
    conv_handler = ConversationHandler(
        entry_points=[CommandHandler('register', register)],
        states={
            BIO: [MessageHandler(filters.TEXT & ~filters.COMMAND, bio)],
            AGE: [MessageHandler(filters.TEXT & ~filters.COMMAND, age)]
        },
        fallbacks=[CommandHandler('cancel', cancel)]
    )
    application.add_handler(conv_handler)
    application.add_handler(CallbackQueryHandler(button_handler))
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == '__main__':
    main()

Результат работы (логирование):

2024-01-01 12:00:00,000 - telegram.ext.Application - INFO - Application started
2024-01-01 12:00:05,123 - telegram.ext.Dispatcher - INFO - Received update 1 from user 123456
...

2. Асинхронный бот на aiogram с FSM и middleware для логирования

from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command, CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import Message
from aiogram import F
import asyncio

BOT_TOKEN = 'TOKEN'
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

class Form(StatesGroup):
    name = State()
    age = State()

@dp.message(CommandStart())
async def cmd_start(message: Message):
    await message.answer('Привет! Введите /fill для заполнения анкеты.')

@dp.message(Command('fill'))
async def cmd_fill(message: Message, state: FSMContext):
    await state.set_state(Form.name)
    await message.answer('Как вас зовут?')

@dp.message(Form.name, F.text)
async def process_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(Form.age)
    await message.answer('Сколько вам лет?')

@dp.message(Form.age, F.text)
async def process_age(message: Message, state: FSMContext):
    if not message.text.isdigit():
        await message.answer('Введите число.')
        return
    data = await state.get_data()
    name = data['name']
    age = message.text
    await message.answer(f'Спасибо, {name}, вам {age} лет.')
    await state.clear()

@dp.message()
async def handle_unknown(message: Message):
    await message.answer('Я не понимаю. Используйте /start или /fill.')

async def main():
    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())

Результат: после команды /fill диалог в несколько шагов.

3. Вебхук на Flask с обработкой команд и отправкой сообщений

from flask import Flask, request, jsonify
import telegram
import logging

logging.basicConfig(level=logging.INFO)

TOKEN = 'TOKEN'
bot = telegram.Bot(token=TOKEN)
app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    update = telegram.Update.de_json(request.get_json(force=True), bot)
    # Обработка сообщений
    if update.message:
        chat_id = update.message.chat.id
        text = update.message.text
        if text == '/start':
            bot.send_message(chat_id=chat_id, text='Привет от Flask!')
        else:
            bot.send_message(chat_id=chat_id, text=f'Вы написали: {text}')
    return 'OK', 200

@app.route('/set_webhook', methods=['GET', 'POST'])
def set_webhook():
    url = 'https://yourdomain.com/webhook'
    s = bot.set_webhook(url)
    if s:
        return 'Webhook set!', 200
    else:
        return 'Error', 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8443, ssl_context=('cert.pem', 'key.pem'))

После запуска сервера вызвать /set_webhook или напрямую через API: https://api.telegram.org/botTOKEN/setWebhook?url=https://yourdomain.com/webhook. Убедитесь, что сертификат действителен.

4. Использование inline режима для поиска (например, погода)

import requests
from telegram import InlineQueryResultArticle, InputTextMessageContent
from telegram.ext import InlineQueryHandler

async def inline_weather(update, context):
    query = update.inline_query.query
    if not query:
        return
    # Простой запрос к API погоды (пример)
    api_key = 'YOUR_OPENWEATHER_API'
    url = f'http://api.openweathermap.org/data/2.5/weather?q={query}&appid={api_key}&units=metric'
    try:
        r = requests.get(url)
        data = r.json()
        if data['cod'] == 200:
            temp = data['main']['temp']
            desc = data['weather'][0]['description']
            text = f'Погода в {query}: {temp}°C, {desc}'
        else:
            text = 'Город не найден.'
    except:
        text = 'Ошибка получения данных.'
    results = [InlineQueryResultArticle(
        id='1',
        title=text,
        input_message_content=InputTextMessageContent(text)
    )]
    await update.inline_query.answer(results, cache_time=10)

application.add_handler(InlineQueryHandler(inline_weather))

Пользователь в любом чате вводит @bot_name <город> и получает погоду.

5. Отправка медиа и документов с обработкой ошибок

async def send_photo(update, context):
    try:
        with open('image.png', 'rb') as f:
            await update.message.reply_photo(f, caption='Пример фото')
    except FileNotFoundError:
        await update.message.reply_text('Файл не найден.')

async def receive_photo(update, context):
    # Сохраняем фото
    file = await update.message.photo[-1].get_file()
    await file.download_to_drive('received_photo.jpg')
    await update.message.reply_text(f'Фото сохранено как received_photo.jpg')

application.add_handler(CommandHandler('sendphoto', send_photo))
application.add_handler(MessageHandler(filters.PHOTO, receive_photo))

6. Платежи: тестовый счёт

from telegram import LabeledPrice, Update
from telegram.ext import PreCheckoutQueryHandler

async def start_payment(update, context):
    chat_id = update.effective_chat.id
    title = 'Подписка на месяц'
    description = 'Доступ к функциям на 30 дней'
    payload = 'sub-monthly'
    provider_token = '284685063:TEST:YOUR_TOKEN'  # от BotFather для тестов
    currency = 'RUB'
    prices = [LabeledPrice('Подписка', 14900)]  # 149 рублей
    await context.bot.send_invoice(
        chat_id, title, description, payload,
        provider_token, currency, prices,
        need_name=True,
        need_email=True,
        need_phone_number=False
    )

async def pre_checkout(update: Update, context):
    query = update.pre_checkout_query
    if query.invoice_payload != 'sub-monthly':
        await query.answer(ok=False, error_message='Неверный payload')
    else:
        await query.answer(ok=True)

async def successful_payment(update, context):
    await update.message.reply_text('Оплата прошла успешно! Спасибо.')

application.add_handler(CommandHandler('buy', start_payment))
application.add_handler(PreCheckoutQueryHandler(pre_checkout))
application.add_handler(MessageHandler(filters.SUCCESSFUL_PAYMENT, successful_payment))

Для тестов используйте токен из BotFather в разделе Payments.