Полное руководство по созданию 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)
Лучшие практики и советы
- Обработка ошибок Всегда обрабатывайте возможные исключения
- Валидация данных Проверяйте ввод пользователя
- Кэширование Используйте кэш для часто запрашиваемых данных
- База данных Для production-ботов используйте базу данных вместо MemoryStorage
- Безопасность Никогда не храните токен в коде, используйте переменные окружения
- Логирование Настройте подробное логирование для отладки
Заключение
Aiogram — это мощный и гибкий фреймворк для создания Telegram-ботов на Python. Он предоставляет все необходимые инструменты для создания как простых, так и сложных ботов с богатой функциональностью. Асинхронная природа библиотеки обеспечивает высокую производительность, а понятный API делает разработку приятной и эффективной.
Это руководство покрывает основные аспекты работы с aiogram, но библиотека предлагает еще больше возможностей работу с платежами, глубокую интеграцию с Telegram API, кастомные фильтры и многое другое. Официальная документация aiogram — отличный ресурс для дальнейшего изучения httpsdocs.aiogram.dev
Удачи в создании ваших Telegram-ботов!