Разработка асинхронного бота для Telegram с использованием aiogram

Раздел: Разработка на Python -> Библиотеки

Основное решение: базовая структура бота на aiogram 3.x

Для создания асинхронного Telegram бота с помощью библиотеки aiogram необходимо установить её через pip: pip install aiogram. Затем создаётся экземпляр Bot с токеном, полученным от BotFather, и Dispatcher, который управляет обработчиками. Простейший бот обрабатывает команду /start.


import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command

TOKEN = "ваш_токен"

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

@dp.message(Command("start"))
async def start_handler(message: types.Message):
    await message.answer("Привет! Я бот, работающий на aiogram.")

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

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

команды math python (команды модуля math в python)

После запуска бот отвечает на команду /start. Основные компоненты: Bot для взаимодействия с API Telegram, Dispatcher для регистрации обработчиков и маршрутизации обновлений, asyncio.run для запуска асинхронного цикла.

Ошибка: "ModuleNotFoundError: No module named 'aiogram'" – библиотека не установлена. Решение: выполнить pip install aiogram.

Ошибка: "Invalid token" – неверный токен. Решение: проверить токен в BotFather, убедиться, что он не содержит лишних пробелов.

Ошибка: "RuntimeError: asyncio.run() cannot be called from a running event loop" – в некоторых средах (Jupyter, IPython) нельзя использовать asyncio.run. Решение: использовать await dp.start_polling(bot) в существующем цикле или запускать через asyncio.create_task.

Как организовать обработку текстовых сообщений с фильтрами?

Для обработки не только команд, но и обычных сообщений используются фильтры, например, TextContains или StateFilter. В aiogram 3.x фильтры передаются в декоратор.


from aiogram.filters import Text

@dp.message(Text(text="Привет"))
async def hello_handler(message: types.Message):
    await message.answer("И тебе привет!")

Aiogram python (создание бота на aiogram (асинхронный telegram бот))

Также можно использовать F (магический фильтр) для создания сложных условий.


from aiogram import F

@dp.message(F.text.lower().contains("пока"))
async def bye_handler(message: types.Message):
    await message.answer("До свидания!")

библиотека имен python (библиотека имен в python)

Проблема: Фильтр не срабатывает, если сообщение содержит лишние пробелы. Решение: использовать F.text.lower().strip().contains(...) или регулярные выражения.

Как реализовать диалог с пользователем (FSM)?

Для многошаговых сценариев используется FSM (Finite State Machine). В aiogram есть встроенная поддержка состояний через State и StatesGroup.


from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext

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

@dp.message(Command("form"))
async def form_start(message: types.Message, state: FSMContext):
    await state.set_state(Form.name)
    await message.answer("Введите ваше имя:")

@dp.message(Form.name)
async def process_name(message: types.Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(Form.age)
    await message.answer("Введите ваш возраст:")

@dp.message(Form.age)
async def process_age(message: types.Message, state: FSMContext):
    data = await state.get_data()
    await message.answer(f"Спасибо, {data['name']}! Ваш возраст: {message.text}")
    await state.clear()

Python pandas типы данных (типы данных pandas в python)

Ошибка: Состояние не очищается, если пользователь прерывает диалог. Решение: добавить обработчик команды /cancel, который вызывает await state.clear().

Проблема: При перезапуске бота состояния теряются. Для продакшена стоит использовать хранилище (Redis, MemoryStorage по умолчанию).

Как добавлять клавиатуры (Reply и Inline)?

Клавиатуры делают взаимодействие удобнее. Reply-клавиатура отправляет кнопки в поле ввода, Inline – кнопки под сообщением.


from aiogram.types import ReplyKeyboardMarkup, KeyboardButton, InlineKeyboardMarkup, InlineKeyboardButton

# Reply клавиатура
reply_kb = ReplyKeyboardMarkup(
    keyboard=[
        [KeyboardButton(text="Кнопка 1"), KeyboardButton(text="Кнопка 2")],
        [KeyboardButton(text="Кнопка 3")]
    ],
    resize_keyboard=True
)

@dp.message(Command("reply_kb"))
async def show_reply_kb(message: types.Message):
    await message.answer("Выберите действие:", reply_markup=reply_kb)

# Inline клавиатура с callback_data
inline_kb = InlineKeyboardMarkup(
    inline_keyboard=[
        [InlineKeyboardButton(text="Нажми меня", callback_data="btn_pressed")]
    ]
)

@dp.message(Command("inline_kb"))
async def show_inline_kb(message: types.Message):
    await message.answer("Кнопка под сообщением:", reply_markup=inline_kb)

@dp.callback_query(lambda c: c.data == "btn_pressed")
async def process_callback(callback: types.CallbackQuery):
    await callback.answer("Вы нажали кнопку!")
    await callback.message.edit_text("Кнопка нажата")

Ошибка: Inline-кнопка не реагирует – неправильный формат callback_data. Решение: callback_data должна быть строкой до 64 байт, без пробелов и спецсимволов.

Проблема: Reply-клавиатура не исчезает после выбора. Для удаления нужно отправить новое сообщение с reply_markup=types.ReplyKeyboardRemove().

Как обрабатывать ошибки и логировать?

Можно использовать middleware или глобальный обработчик ошибок. В aiogram есть встроенный ErrorEvent.


from aiogram.types import ErrorEvent

@dp.error()
async def error_handler(event: ErrorEvent):
    print(f"Ошибка: {event.exception}")
    await event.update.message.answer("Произошла внутренняя ошибка.")

Также middleware позволяют перехватывать все обновления до обработчиков.


from aiogram import BaseMiddleware

class LoggingMiddleware(BaseMiddleware):
    async def __call__(self, handler, event, data):
        print(f"Получено обновление: {event}")
        return await handler(event, data)

dp.message.middleware(LoggingMiddleware())

Проблема: Middleware не применяется к командам – нужно регистрировать для каждого типа обновлений (dp.message.middleware, dp.callback_query.middleware и т.д.).

Продвинутые примеры кода для aiogram

Ниже представлены расширенные сценарии с полным кодом и описанием результата.

Пример

import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
from aiogram.fsm.storage.memory import MemoryStorage

TOKEN = "ваш_токен"
bot = Bot(token=TOKEN)
storage = MemoryStorage()
dp = Dispatcher(storage=storage)

class UserRegistration(StatesGroup):
    username = State()
    email = State()

users_db = {}

@dp.message(Command("register"))
async def register_start(message: types.Message, state: FSMContext):
    await state.set_state(UserRegistration.username)
    await message.answer("Введите ваш логин:")

@dp.message(UserRegistration.username)
async def process_username(message: types.Message, state: FSMContext):
    await state.update_data(username=message.text)
    await state.set_state(UserRegistration.email)
    await message.answer("Введите ваш email:")

@dp.message(UserRegistration.email)
async def process_email(message: types.Message, state: FSMContext):
    data = await state.get_data()
    user_id = message.from_user.id
    users_db[user_id] = {"username": data["username"], "email": message.text}
    await message.answer("Регистрация завершена!")
    print(users_db)  # Для демонстрации
    await state.clear()

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

if __name__ == "__main__":
    asyncio.run(main())
После запуска пользователь может ввести команду /register, затем имя и email. Данные сохраняются в словарь users_db. В консоли отобразится словарь с ID пользователя и его данными.

Пример

@dp.message(Command("photo"))
async def send_photo(message: types.Message):
    photo_url = "https://example.com/example.jpg"  # Замените на реальный URL
    caption = "Пример фотографии"
    await message.answer_photo(photo=photo_url, caption=caption)
Пользователь получит сообщение с фотографией по указанному URL и подписью "Пример фотографии".

Пример

import asyncio

user_ids = [123456789, 987654321]  # Пример списка

async def scheduled_broadcast(bot: Bot):
    while True:
        await asyncio.sleep(3600)  # каждый час
        for uid in user_ids:
            try:
                await bot.send_message(uid, "Ежечасное сообщение")
            except Exception as e:
                print(f"Не удалось отправить {uid}: {e}")

async def main():
    asyncio.create_task(scheduled_broadcast(bot))
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())
Бот будет отправлять заданное сообщение всем пользователям из списка каждый час. Ошибки при отправке (например, если пользователь заблокировал бота) будут выводиться в консоль.

Пример

from aiogram.fsm.storage.redis import RedisStorage
import redis.asyncio as aioredis

redis_client = aioredis.from_url("redis://localhost:6379/0")
storage = RedisStorage(redis_client)
dp = Dispatcher(storage=storage)
# Далее код FSM как обычно
Состояния пользователей будут сохраняться в Redis, что позволяет перезапускать бота без потери диалогов.

Пример

from aiogram import F

@dp.message(F.photo)
async def handle_photo(message: types.Message):
    photo = message.photo[-1]  # самое большое разрешение
    file_id = photo.file_id
    file = await bot.get_file(file_id)
    file_path = file.file_path
    await bot.download_file(file_path, destination=f"{file_id}.jpg")
    await message.reply("Фото сохранено!")
Бот получает фото, скачивает его на диск с именем file_id.jpg и отвечает пользователю.

Создание бота на aiogram (асинхронный Telegram бот) - comments

En
Aiogram python (python)