Для работы с Telegram-ботами используется библиотека aiogram. Установка выполняется с помощью менеджера пакетов:
pip install aiogram
Для разработки рекомендуется использовать виртуальное окружение Python, как было показано в предыдущих уроках. Это позволяет изолировать зависимости проекта и избежать конфликтов между библиотеками.
Перед началом работы необходимо создать Telegram-бота и получить его токен. Создание бота и выдача токена выполняются через официальный бот BotFather(@BotFather) в Telegram.
Telegram-бот — это программа, которая:
Бот не «работает сам по себе», он реагирует на события. Основным событием является сообщение пользователя.
Минимальная программа Telegram-бота на aiogram строится вокруг трёх элементов:
Объект Bot отвечает за связь с серверами Telegram.
bot = Bot(token=TOKEN)
Через него отправляются сообщения, изображения и другие данные.
В учебных примерах он обычно используется не напрямую, а через методы объекта Message.
Dispatcher — это центральный объект программы.
dp = Dispatcher()
Его задача:
Dispatcher содержит список handlers, зарегистрированных в программе.
Handler — это асинхронная функция, которая вызывается, когда приходит подходящее сообщение.
@dp.message()
async def handler(message: Message):
await message.answer("Привет!")
Здесь:
@dp.message() — регистрация обработчика,handler — функция обработки,message — объект сообщения.Когда пользователь отправляет сообщение:
Один handler обрабатывает одно сообщение.
Message — это объект, который представляет одно сообщение пользователя.
Через него можно:
На начальном этапе используются следующие поля:
message.text # текст сообщения
message.chat.id # идентификатор чата
message.from_user # информация о пользователе
message.message_id # идентификатор сообщения
Наиболее часто используется message.text.
Методы Message упрощают отправку ответов.
await message.answer("Текст ответа")
Основные методы:
await message.answer(text) # отправить сообщение в чат
await message.reply(text) # ответить на конкретное сообщение
await message.delete() # удалить сообщение
Метод answer() автоматически отправляет сообщение в тот же чат, откуда пришёл запрос.
Пример простейшей обработки текста:
@dp.message()
async def echo(message: Message):
await message.answer(message.text)
Здесь:
message.text,message.answer().В упрощённом виде программа Telegram-бота состоит из:
1. Инициализация Bot
2. Инициализация Dispatcher
3. Регистрация handlers
4. Запуск обработки сообщений
Пример:
bot = Bot(token=TOKEN)
dp = Dispatcher()
@dp.message()
async def handler(message: Message):
await message.answer("Привет")
dp.start_polling(bot)
import asyncio
import logging
import argparse
from aiogram import Bot, Dispatcher, F
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart, Command
from aiogram.types import Message
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument(
"--token",
required=True,
help="Telegram bot token from BotFather"
)
return parser.parse_args()
async def main(token: str) -> None:
bot = Bot(
token=token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)
dp = Dispatcher()
@dp.message(CommandStart())
async def cmd_start(message: Message) -> None:
await message.answer(
"Привет! Я dummy-бот на aiogram.\n"
"Напиши любой текст и я его повторю."
)
@dp.message(Command("help"))
async def cmd_help(message: Message) -> None:
await message.answer("Просто напиши любой текст и я повторю его")
@dp.message(F.text)
async def echo(message: Message) -> None:
await message.answer(message.text)
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
args = parse_args()
asyncio.run(main(args.token))
FSM (конечный автомат состояний) используется, когда бот должен вести пошаговый диалог с пользователем.
FSM позволяет боту:
Без FSM каждое сообщение обрабатывается одинаково, независимо от контекста.
Работа с FSM строится вокруг четырёх основных элементов:
State — это логическая метка, обозначающая этап диалога.
Примеры:
Состояния объединяются в класс-наследник StatesGroup.
from aiogram.fsm.state import StatesGroup, State
class Form(StatesGroup):
waiting_name = State()
waiting_age = State()
Каждый атрибут класса — отдельное состояние.
FSM должна где-то хранить:
Для учебных целей используется хранилище в памяти:
from aiogram.fsm.storage.memory import MemoryStorage
dp = Dispatcher(storage=MemoryStorage())
Хранилище автоматически связывает состояние с пользователем и чатом.
FSMContext — объект, через который handler:
Он передаётся в handler автоматически.
await state.set_state(Form.waiting_name)
После этого все следующие сообщения пользователя будут обрабатываться handlers, привязанными к этому состоянию.
await state.clear()
Используется, когда диалог завершён или отменён.
@dp.message(Form.waiting_name)
async def process_name(message: Message, state: FSMContext):
...
Такой handler будет вызван только если пользователь находится в указанном состоянии.
@dp.message(CommandStart())
async def start(message: Message, state: FSMContext):
await state.set_state(Form.waiting_name)
@dp.message(Form.waiting_name)
async def name_step(message: Message, state: FSMContext):
await state.set_state(Form.waiting_age)
@dp.message(Form.waiting_age)
async def age_step(message: Message, state: FSMContext):
await state.clear()
Каждый handler:
await state.update_data(name=message.text)
Можно сохранять несколько значений:
await state.update_data(name="Иван", age=18)
data = await state.get_data()
name = data["name"]
Данные представляют собой обычный словарь.
В handlers с FSM обычно выполняются следующие действия:
message.text);Пример проверки без смены состояния:
if not message.text.isalpha():
await message.answer("Введите корректное значение")
return
Состояние остаётся прежним, пока данные не будут корректными.
После очистки состояния:
await state.clear()
пользователь:
Это позволяет разделять:
import asyncio
import logging
import argparse
import re
from aiogram import Bot, Dispatcher, F
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import Message
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.fsm.storage.memory import MemoryStorage
NAME_RE = re.compile(r"^[A-Za-zА-Яа-яЁё]+$")
def is_valid_name(value: str) -> bool:
value = value.strip()
return bool(value) and bool(NAME_RE.fullmatch(value))
class Profile(StatesGroup):
waiting_first_name = State()
waiting_last_name = State()
ready = State()
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--token", required=True, help="Telegram bot token")
return parser.parse_args()
async def main(token: str) -> None:
bot = Bot(
token=token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)
dp = Dispatcher(storage=MemoryStorage())
@dp.message(CommandStart())
async def cmd_start(message: Message, state: FSMContext) -> None:
await state.clear()
await message.answer("Привет! Я учебный бот")
await message.answer("Как твое имя?")
await state.set_state(Profile.waiting_first_name)
@dp.message(Profile.waiting_first_name, F.text)
async def first_name_step(message: Message, state: FSMContext) -> None:
text = (message.text or "").strip()
if not is_valid_name(text):
await message.answer(
"Хмммм либо у тебя дурацкое имя, либо ты что-то перепутал, попробуй еще раз"
)
return
await state.update_data(first_name=text)
await message.answer("Отлично! А теперь фамилия?")
await state.set_state(Profile.waiting_last_name)
@dp.message(Profile.waiting_last_name, F.text)
async def last_name_step(message: Message, state: FSMContext) -> None:
text = (message.text or "").strip()
if not is_valid_name(text):
await message.answer(
"Ух, если у тебя будет свадьба, советую взять фамилию супруга, а то это какой-то кошмар. Попробуй еще раз"
)
return
await state.update_data(last_name=text)
data = await state.get_data()
first_name = data["first_name"]
last_name = data["last_name"]
await message.answer(f"Рад познакомиться, {first_name} {last_name}!")
await state.set_state(Profile.ready)
@dp.message(Profile.ready)
async def after_ready(message: Message, state: FSMContext) -> None:
data = await state.get_data()
first_name = data.get("first_name", "друг")
last_name = data.get("last_name", "")
full_name = (first_name + " " + last_name).strip()
await message.answer(f"Извини, {full_name}, но я пока не умею общаться, в твоих руках это исправить!")
@dp.message(Profile.waiting_first_name)
@dp.message(Profile.waiting_last_name)
async def non_text_during_form(message: Message) -> None:
await message.answer("Общайся текстом, а то что ты как неандерталец")
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
args = parse_args()
asyncio.run(main(args.token))
Когда пользователь отправляет изображение, объект Message содержит поле:
message.photo — список объектов PhotoSize, представляющих одно изображение в разных размерах.Для работы используется последний элемент списка:
photo = message.photo[-1]
Объект PhotoSize содержит:
file_id — идентификатор файла в Telegram,width — ширина изображения,height — высота изображения,file_size — размер файла.При получении изображения поле message.text не используется.
Для обработки сообщений с изображениями используется фильтр F.photo:
@dp.message(F.photo)
async def on_photo(message: Message):
photo = message.photo[-1]
file_id = photo.file_id
await message.answer("Изображение получено")
Handler вызывается только для сообщений, содержащих изображение.
file_idЕсли имеется file_id, изображение отправляется следующим образом:
await message.answer_photo(photo=file_id, caption="Изображение")
Telegram использует файл, сохранённый на своих серверах.
Для отправки изображения, хранящегося на диске, используется объект FSInputFile:
from aiogram.types import FSInputFile
photo = FSInputFile("images/image.jpg")
await message.answer_photo(photo=photo, caption="Изображение с диска")
Путь к файлу указывается относительно рабочей директории программы или как абсолютный путь.
Для сохранения изображения выполняются следующие шаги:
file_id,Пример:
import os
@dp.message(F.photo)
async def save_photo(message: Message, bot: Bot):
photo = message.photo[-1]
file_id = photo.file_id
file = await bot.get_file(file_id)
os.makedirs("downloads", exist_ok=True)
filename = f"downloads/{file_id}.jpg"
await bot.download_file(file.file_path, filename)
await message.answer("Изображение сохранено")
Изображение сохраняется в указанную директорию.
Изображение может быть отправлено двумя способами:
message.photo;message.document.При отправке как файл используется объект Document, содержащий:
file_id,file_name,mime_type.Работа с файлами выполняется аналогично работе с изображениями.
Для работы с изображениями используются следующие элементы:
message.photoPhotoSize.file_id@dp.message(F.photo)message.answer_photo(...)Bot.get_file(...)Bot.download_file(...)FSInputFile