Полное руководство по созданию Telegram-бота на Python с aiogram

Полное руководство по созданию Telegram-бота на Python с aiogram


Полное руководство по созданию Telegram-бота на Python с aiogram

Введение в aiogram

Aiogram — это современный асинхронный фреймворк для создания Telegram-ботов на Python. В отличие от синхронных библиотек, aiogram построен на основе asyncio, что позволяет обрабатывать множество запросов одновременно без блокировок. Это делает ботов более отзывчивыми и производительными.

Основные преимущества aiogram

  • Полная асинхронность
  • Простота использования
  • Поддержка всех возможностей Telegram Bot API
  • Гибкая система middleware
  • Регулярные обновления

Установка и настройка

Для начала работы установите aiogram через pip

pip install aiogram

Создайте нового бота через BotFather и получите API-токен.

Базовая структура бота

Создайте файл bot.py со следующим содержимым

import logging
from aiogram import Bot, Dispatcher, executor, types

# Настройка логирования
logging.basicConfig(level=logging.INFO)

# Инициализация бота и диспетчера
API_TOKEN = 'YOUR_API_TOKEN'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

# Обработчик команды start
@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
    await message.reply(Привет! Я эхо-бот на aiogram.)

# Обработчик текстовых сообщений
@dp.message_handler()
async def echo(message types.Message)
    await message.answer(message.text)

# Запуск бота
if __name__ == '__main__'
    executor.start_polling(dp, skip_updates=True)

Этот код создает простого эхо-бота, который отвечает на любое сообщение тем же текстом.

Обработчики сообщений

Фильтры команд

@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message types.Message)
    await message.answer(Добро пожаловать! Доступные команды start, help, info)

Фильтры по содержанию

@dp.message_handler(content_types=types.ContentType.TEXT)
async def handle_text(message types.Message)
    await message.answer(Вы отправили текстовое сообщение!)

@dp.message_handler(content_types=types.ContentType.PHOTO)
async def handle_photo(message types.Message)
    await message.answer(Вы отправили фото!)

Кастомные фильтры

from aiogram import filters

@dp.message_handler(filters.Text(equals='привет', ignore_case=True))
async def hello_handler(message types.Message)
    await message.answer(И тебе привет!)

@dp.message_handler(filters.Text(contains='python', ignore_case=True))
async def python_handler(message types.Message)
    await message.answer(Вы упомянули Python!)

Клавиатуры и кнопки

Reply-клавиатура

from aiogram.types import ReplyKeyboardMarkup, KeyboardButton

# Создание клавиатуры
kb = ReplyKeyboardMarkup(
    resize_keyboard=True,  # автоматическое изменение размера
    one_time_keyboard=True  # скрыть после использования
)

# Добавление кнопок
btn1 = KeyboardButton('Кнопка 1')
btn2 = KeyboardButton('Кнопка 2')
btn3 = KeyboardButton('Кнопка 3')

kb.add(btn1, btn2).add(btn3)

@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
    await message.answer(Выберите действие, reply_markup=kb)

@dp.message_handler(filters.Text(equals='Кнопка 1'))
async def btn1_handler(message types.Message)
    await message.answer(Вы нажали кнопку 1!)

Inline-клавиатура

from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton

# Создание inline-клавиатуры
inline_kb = InlineKeyboardMarkup()
btn1 = InlineKeyboardButton('Кнопка 1', callback_data='btn1')
btn2 = InlineKeyboardButton('Кнопка 2', callback_data='btn2')
inline_kb.add(btn1, btn2)

@dp.message_handler(commands=['inline'])
async def show_inline(message types.Message)
    await message.answer(Выберите действие, reply_markup=inline_kb)

# Обработчик нажатий на inline-кнопки
@dp.callback_query_handler(lambda c c.data == 'btn1')
async def process_callback_btn1(callback_query types.CallbackQuery)
    await bot.answer_callback_query(callback_query.id)
    await bot.send_message(callback_query.from_user.id, 'Вы нажали кнопку 1!')

Finite State Machine (FSM)

FSM позволяет управлять состоянием диалога с пользователем.

from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup

# Инициализация хранилища состояний
storage = MemoryStorage()
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot, storage=storage)

# Определение состояний
class Form(StatesGroup)
    name = State()
    age = State()
    gender = State()

# Начало диалога
@dp.message_handler(commands='form')
async def cmd_start(message types.Message)
    await Form.name.set()
    await message.reply(Как вас зовут)

# Обработка имени
@dp.message_handler(state=Form.name)
async def process_name(message types.Message, state FSMContext)
    async with state.proxy() as data
        data['name'] = message.text
    
    await Form.next()
    await message.reply(Сколько вам лет)

# Обработка возраста
@dp.message_handler(state=Form.age)
async def process_age(message types.Message, state FSMContext)
    if not message.text.isdigit()
        await message.reply(Пожалуйста, введите число!)
        return
    
    async with state.proxy() as data
        data['age'] = int(message.text)
    
    await Form.next()
    await message.reply(Какой ваш пол)

# Обработка пола и завершение
@dp.message_handler(state=Form.gender)
async def process_gender(message types.Message, state FSMContext)
    async with state.proxy() as data
        data['gender'] = message.text
    
    # Завершаем состояние
    await state.finish()
    
    # Отправляем собранные данные
    await message.reply(
        fСпасибо! Ваши данныеn
        fИмя {data['name']}n
        fВозраст {data['age']}n
        fПол {data['gender']}
    )

Работа с медиафайлами

Отправка фото

@dp.message_handler(commands=['photo'])
async def send_photo(message types.Message)
    # Отправка фото по URL
    await bot.send_photo(
        chat_id=message.chat.id,
        photo=httpsexample.comimage.jpg,
        caption=Это пример фото
    )
    
    # Отправка фото из файла
    with open('local_image.jpg', 'rb') as photo
        await bot.send_photo(
            chat_id=message.chat.id,
            photo=photo,
            caption=Локальное фото
        )

Загрузка полученных файлов

@dp.message_handler(content_types=types.ContentType.PHOTO)
async def download_photo(message types.Message)
    # Получаем информацию о файле
    file_id = message.photo[-1].file_id
    file = await bot.get_file(file_id)
    file_path = file.file_path
    
    # Скачиваем файл
    await bot.download_file(file_path, 'downloaded_photo.jpg')
    await message.answer(Фото сохранено!)

Middleware и кастомизация

Создание middleware

class CounterMiddleware
    def __init__(self)
        self.counter = 0

    async def on_process_message(self, message types.Message, data dict)
        self.counter += 1
        logging.info(fСообщение #{self.counter})

# Регистрация middleware
dp.middleware.setup(CounterMiddleware())

Логирование

import logging
from aiogram import types
from aiogram.dispatcher.handler import CancelHandler
from aiogram.dispatcher.middlewares import BaseMiddleware

class LoggingMiddleware(BaseMiddleware)
    async def on_pre_process_update(self, update types.Update, data dict)
        logging.info(fПолучено обновление {update})
    
    async def on_pre_process_message(self, message types.Message, data dict)
        logging.info(fПолучено сообщение от {message.from_user.id} {message.text})
        
        # Пример фильтрации
        if спам in message.text.lower()
            logging.warning(Обнаружено спам-сообщение!)
            await message.answer(Сообщение содержит спам!)
            raise CancelHandler()  # Отменяем дальнейшую обработку

Webhook вместо Long Polling

Для production-среды лучше использовать webhook

from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiohttp import web

API_TOKEN = 'YOUR_API_TOKEN'
WEBHOOK_HOST = 'httpsyour.domain'
WEBHOOK_PATH = 'webhook'
WEBHOOK_URL = f{WEBHOOK_HOST}{WEBHOOK_PATH}

# Инициализация
bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)

# Обработчики
@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
    await message.answer(Привет! Я бот на webhook!)

# Настройка webhook
async def on_startup(app)
    await bot.set_webhook(WEBHOOK_URL)

async def on_shutdown(app)
    await bot.delete_webhook()

# Создание aiohttp приложения
app = web.Application()
app.router.add_post(WEBHOOK_PATH, dp._process_update)
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)

if __name__ == '__main__'
    web.run_app(app, host='0.0.0.0', port=3000)

Пример полноценного бота

Создаем бота-опросника

import logging
from aiogram import Bot, Dispatcher, executor, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup

logging.basicConfig(level=logging.INFO)

API_TOKEN = 'YOUR_API_TOKEN'

bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)

# Состояния для опроса
class PollStates(StatesGroup)
    question = State()
    options = State()

# Хранение опросов
polls = {}

# Команда начала создания опроса
@dp.message_handler(commands=['create_poll'])
async def cmd_create_poll(message types.Message)
    await PollStates.question.set()
    await message.answer(Введите вопрос для опроса)

# Получение вопроса
@dp.message_handler(state=PollStates.question)
async def process_question(message types.Message, state FSMContext)
    async with state.proxy() as data
        data['question'] = message.text
        data['creator'] = message.from_user.id
    
    await PollStates.next()
    await message.answer(Теперь введите варианты ответов через запятую)

# Получение вариантов ответов
@dp.message_handler(state=PollStates.options)
async def process_options(message types.Message, state FSMContext)
    options = [opt.strip() for opt in message.text.split(',') if opt.strip()]
    
    if len(options)  2
        await message.answer(Нужно как минимум два варианта ответа!)
        return
    
    async with state.proxy() as data
        poll_id = str(message.message_id)
        polls[poll_id] = {
            'question' data['question'],
            'options' options,
            'votes' {opt 0 for opt in options},
            'voted' set()
        }
    
    # Создание inline-клавиатуры с вариантами
    keyboard = types.InlineKeyboardMarkup()
    for option in options
        keyboard.add(types.InlineKeyboardButton(
            text=option, 
            callback_data=fvote_{poll_id}_{option}
        ))
    
    await state.finish()
    await message.answer(fОпрос {data['question']}, reply_markup=keyboard)

# Обработка голосов
@dp.callback_query_handler(lambda c c.data.startswith('vote_'))
async def process_vote(callback_query types.CallbackQuery)
    _, poll_id, option = callback_query.data.split('_', 2)
    
    if poll_id not in polls
        await callback_query.answer(Опрос не найден!)
        return
    
    poll = polls[poll_id]
    
    if callback_query.from_user.id in poll['voted']
        await callback_query.answer(Вы уже голосовали в этом опросе!)
        return
    
    # Обновление результатов
    poll['votes'][option] += 1
    poll['voted'].add(callback_query.from_user.id)
    
    # Обновление сообщения с результатами
    results = n.join([f{opt} {count} for opt, count in poll['votes'].items()])
    await callback_query.message.edit_text(
        f{poll['question']}nnРезультатыn{results}
    )
    await callback_query.answer(fВы проголосовали за {option})

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

Лучшие практики и советы

  1. Обработка ошибок Всегда обрабатывайте возможные исключения
  2. Валидация данных Проверяйте ввод пользователя
  3. Кэширование Используйте кэш для часто запрашиваемых данных
  4. База данных Для production-ботов используйте базу данных вместо MemoryStorage
  5. Безопасность Никогда не храните токен в коде, используйте переменные окружения
  6. Логирование Настройте подробное логирование для отладки

Заключение

Aiogram — это мощный и гибкий фреймворк для создания Telegram-ботов на Python. Он предоставляет все необходимые инструменты для создания как простых, так и сложных ботов с богатой функциональностью. Асинхронная природа библиотеки обеспечивает высокую производительность, а понятный API делает разработку приятной и эффективной.

Это руководство покрывает основные аспекты работы с aiogram, но библиотека предлагает еще больше возможностей работу с платежами, глубокую интеграцию с Telegram API, кастомные фильтры и многое другое. Официальная документация aiogram — отличный ресурс для дальнейшего изучения httpsdocs.aiogram.dev

Удачи в создании ваших Telegram-ботов!