Разработка асинхронного бота для Telegram с использованием aiogram
Основное решение: базовая структура бота на 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 и отвечает пользователю.