Создание чат-бота с помощью AIOGRAM

Содержание

Введение

AIOGram(pip install aiogram) - это асинхронный фреймворк, предназначенный для создания ботов в Telegram на языке программирования Python. Он разработан для работы в асинхронной среде, что позволяет обрабатывать несколько запросов одновременно без блокировки исполнения. Это делает его идеальным выбором для создания высокопроизводительных и отзывчивых ботов.

Некоторые из ключевых особенностей AIOGram:

Асинхронность: Фреймворк полностью асинхронный, что означает, что он может эффективно обрабатывать множество запросов без блокировки потоков исполнения.
Простота использования: AIOGram предоставляет простой и интуитивно понятный API для взаимодействия с Telegram API. Это позволяет быстро создавать и настраивать ботов.
Расширяемость: Фреймворк предоставляет широкий спектр возможностей для расширения функциональности бота с помощью плагинов и middleware.
Поддержка множества типов сообщений: AIOGram поддерживает различные типы сообщений Telegram, такие как текстовые сообщения, изображения, аудио, видео, файлы и т. д.
Обработка команд: Фреймворк обеспечивает удобный способ обработки команд, отправленных пользователями боту, и реагирования на них соответствующим образом.
Работа с клавиатурами: AIOGram позволяет создавать и отправлять пользователю интерактивные клавиатуры для упрощения навигации и взаимодействия с ботом.

В целом, AIOGram представляет собой мощный инструмент для разработки ботов в Telegram на Python, обеспечивая высокую производительность, гибкость и удобство использования.

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

Давайте представим себе, что у вас есть две задачи: одна - загрузить файл из интернета, а вторая - выполнить какие-то вычисления. В синхронной программе, если одна из этих задач занимает много времени, вся программа будет стоять на паузе и ждать, пока эта задача не завершится. Но в асинхронной программе, вместо того, чтобы ждать, программа может начать выполнять другие задачи в то время, когда первая задача все еще выполняется.

Когда вы используете интернет, вы хотите, чтобы страницы загружались быстро и без задержек. Асинхронное программирование помогает сделать это, позволяя приложениям продолжать работу во время загрузки данных из интернета или выполнения других длительных операций.

Асинхронность также позволяет эффективно использовать ресурсы компьютера. Вместо того, чтобы простаивать и ждать, когда что-то происходит, компьютер может заниматься другими полезными делами.

Асинхронность - это мощный инструмент в современном программировании. Понимание этого позволяет создавать быстрые, отзывчивые и эффективные приложения, которые могут обрабатывать множество задач одновременно.

Создание бота

1. Перейде в бота в телеграме @BotFather

2. введите /newbot

3. Напишите боту название вашего бота

4. Придумайте username вашему боту(важно, чтобы в нем было слово bot)

5. Сохраните токен

Импортируем необходимые модули:

import asyncio
from aiogram import Bot, Dispatcher

asyncio - это модуль в стандартной библиотеке Python, предоставляющий инфраструктуру для написания асинхронного кода. В нашем случае, он нам нужен для асинхронной работы бота.

Dispatcher отвечает за обработку входящих сообщений от пользователей и вызов соответствующих обработчиков для их обработки. Он регистрирует обработчики для различных типов событий, таких как текстовые сообщения, команды, обновления клавиатуры и т. д. Dispatcher также позволяет создавать хендлеры (обработчики) для обработки асинхронных событий и взаимодействия с API Telegram.

Bot представляет собой интерфейс для взаимодействия с Telegram Bot API. Он обеспечивает доступ к различным методам API, таким как отправка сообщений, управление клавиатурами, работа с медиафайлами и т. д. Бот также связывается с Dispatcher, чтобы обрабатывать входящие сообщения и вызывать соответствующие методы обработки.

Вместе Dispatcher и Bot обеспечивают основную функциональность для разработки и управления ботами в Telegram с использованием библиотеки aiogram. Dispatcher регистрирует обработчики событий и вызывает их при поступлении новых сообщений, а Bot предоставляет интерфейс для отправки сообщений и взаимодействия с API Telegram.

Создадим переменные, в одну из них передадим в объект Bot токен, который был выдан при регистрации бота, а в другой создадим объект Dispatcher :

import asyncio
from aiogram import Bot, Dispatcher

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

Настала пора создать асинхронную функцию, которая будет отвечать за запуск бота, чтобы создать такую функцию, необходим перед ее объявлением написать async :

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

await - это оператор, который используется в асинхронном коде для ожидания завершения асинхронной операции. Когда вы используете await перед вызовом асинхронной функции или метода, выполнение текущей функции приостанавливается до тех пор, пока асинхронная операция не завершится, и затем продолжается.

Во второй строке вызывается метод start_polling() объекта Dispatcher (dp), чтобы начать получение обновлений от Telegram(например, ошибки) для вашего бота, так сказать запустить бота. Он передается объект Bot (bot), который представляет вашего бота в Telegram. Метод start_polling() запускает асинхронный цикл, который будет ожидать новых обновлений от Telegram и обрабатывать их с помощью зарегистрированных обработчиков событий. Этот метод будет ожидать, пока бот не будет остановлен, или пока его явно не остановят с помощью вызова dp.stop_polling().

Для запуска асинхронных функций используется блок if __name__ == '__main__' . Этот блок проверяет, запущен ли скрипт напрямую. Для вызова асинх. функции main(), используется метод run() из asyncio

import asyncio
from aiogram import Bot, Dispatcher

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

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

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

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

Для обработки сообщений нужно будет импортировать еще пару объектов:

from aiogram.types import Message
from aiogram.filters import CommandStart, Command

Как вы знаете, любое общение с ботом в телеграме начинается с команды /start, напишем функцию-обработчик для этой команды:

@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!')

1. @dp.message(CommandStart()): Это декоратор, который привязывает функцию к событию получения сообщения. В данном случае, функция будет вызвана, когда пользователь отправит команду /start. dp здесь, является экземпляром класса Dispatcher, который обрабатывает сообщения и команды в вашем боте.

2. async def start(message: Message):: Это определение функции с именем start. Она принимает один аргумент message, который представляет собой объект сообщения, полученного от пользователя. Message здесь, является типом данных из библиотеки aiogram.

3. await message.answer('Здравствуйте!'): Эта строка отправляет ответное сообщение пользователю. message.answer используется для отправки ответов в чат Telegram. В данном случае, бот отправляет сообщение "Здравствуйте!" в ответ на команду /start.

Вместо использования CommandStart() в декораторе можно было бы применить Command(), чтобы указать стартовую функцию /start. Для обработки других команд мы, действительно, будем использовать Command(), в котором будет указано название команды, подлежащей обработке.

Текущий код выглядит так:

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

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!')

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

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

Примечание 1.

Обработчики сообщений обязательно нужно написать до функции, которая запустит бота.

Чтобы обработать любую другую команду, используем следующий похожий код:

@dp.message(Command('help'))
async def help_command(message: Message): 
    await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь")

Используем функцию command() и вписываем туда обрабатываемую команду, давайте еще раз разберемся, как это работает.

Декоратор @dp.message(Command('help')):

Это строка использует декоратор, чтобы связать функцию help_command с командой /help.
dp (Dispatcher) — это объект, который обрабатывает все входящие сообщения и команды.
Command('help') — это условие, которое проверяет, является ли полученное сообщение командой /help.

Асинхронная функция async def help_command(message: Message):

Объявляет асинхронную функцию help_command, которая будет выполнена при получении команды /help.
message: Message — параметр, представляющий объект сообщения, полученного ботом.

await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь"):

Асинхронно отправляет ответное сообщение с помощью метода reply.
Сообщение содержит текст с описанием доступных команд бота.

Создание клавиатур

Для начала импортируем все необходимое:

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

Создадим клавиатуру в переменной, используя функцию ReplyKeyboardMarkup() и передадим в позиционный аргумент keyboard список, из списка кнопок, сами кнопки делаются при помощи функции KeyboardButton(), далее следует записать в кнопку текст(комманду), за которую и будет отвечать кнопка, имеем следующую клавиатуру:

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
])

Осталось эту клавиатуру куда-то прикрепить, сделаем так, чтобы при вводе команды старт, у пользователя открывалась клавиатура, для этого, в обработчике команды старт, в строке, которая отвечает за ответ пользователю, добавим аргумент reply_markup и передадим в нее переменную-клавиатуру, имеет такой код нашего бота:

import asyncio
from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters import CommandStart, Command
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardMarkup, KeyboardButton

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
])

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!', reply_markup=kb)

@dp.message(Command('help'))
async def help_command(message: Message):
    await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь")

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

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

Чтобы уменьшить размер клавиатуры, в переменной, которая отвечает за нее, припишем аргумент resize_keyboard со значением True:

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
], resize_keyboard=True)

Инлайн клавиатура (InlineKeyboard)- это специальный вид клавиатуры, который может быть встроен непосредственно в сообщение бота. Вместо того чтобы появляться внизу экрана, как обычная клавиатура, инлайн клавиатура отображается внутри текста сообщения.

Каждая кнопка в инлайн клавиатуре может содержать текст и опционально колбэк-данные (callback data, об этом в след. уроке), которые будут отправлены боту при нажатии на кнопку. Это позволяет боту определить, какая кнопка была нажата, и выполнить соответствующее действие.

Использование инлайн клавиатуры позволяет ботам предоставлять пользователям интерактивные возможности прямо в чате. Например, пользователь может выбирать из списка опций, выполнять действия или запускать определенные команды, просто нажимая на кнопки в инлайн клавиатуре, не покидая чата.

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

Записывается почти также и работает по той же аналогии:

help_kb = InlineKeyboardMarkup(inline_keyboard=[
    [InlineKeyboardButton(text='Курс', url='https://stepik.org/course/203657')]
])

Единственное отличие в том, что мы не можем просто оставить текст, мы должны что-то написать еще, например, url, добавим клавиатуру в обработчик комманды help и получим следующий код:

import asyncio
from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters import CommandStart, Command
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardMarkup, KeyboardButton

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
], resize_keyboard=True)

help_kb = InlineKeyboardMarkup(inline_keyboard=[
    [InlineKeyboardButton(text='Курс', url='https://stepik.org/course/203657')]
])

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!', reply_markup=kb)

@dp.message(Command('help'))
async def help_command(message: Message):
    await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь", reply_markup=help_kb)

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

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

Callback запросы

Callback - это способ, с помощью которого бот может понимать, когда пользователь нажимает на кнопку или ссылку в чате. Таким образом, бот может выполнять нужные действия в ответ на это действие пользователя, например, отправлять сообщение или выполнять определенную команду.

Когда вы создаете кнопку или ссылку в сообщении бота, вы можете привязать к ней callback-идентификатор. Когда пользователь нажимает на эту кнопку или ссылку, бот получает это сообщение и считывает callback-идентификатор, который вы привязали к действию. Затем бот может выполнять нужные действия на основе этого идентификатора, например, отправлять определенное сообщение или вызывать определенную команду. В aiogram для работы с callback используется специальный класс `CallbackQuery`, который содержит информацию о действии пользователя и методы для взаимодействия с ботом. Вы можете обработать callback-запрос и выполнить необходимые действия в обработчике callback-запросов.

Импортируем необходимое:

from aiogram.types import CallbackQuery

В нашей Инлайн-клавиатуре добавим кнопку, которая будет содержать callback, например, такую:

InlineKeyboardButton(text='Тест', callback_data='Course')

В данном случае, Course, здесь, выступает в роли названия callback'a, теперь нам нужно будет написать обработчик этого callback'a, создадим функцию с декоратором callback_query, который и будет основой нашего обработчика, в него передадим название callback'a, но чтобы отловить его, придется импортировать фильтр from aiogram import F, и использовать метод data, декоратор выглядит так:

@dp.callback_query(F.data == 'Course')

Теперь создадим асинхронную функцию, которая будет принимать сам callback , а после напишем ответ от него, весь обработчик будет выглядеть так:

@dp.callback_query(F.data == 'Course')
async def course_func(callback: CallbackQuery):
    await callback.message.answer('Вы сейчас проходите курс!')

Рассмотрим, как это будет работать в целом, когда пользователь нажимает на кнопку с заданным значением "Course" в телеграм-боте, будет вызвана функция "course_func". Функция "course_func" принимает объект "callback", который содержит информацию о запросе пользователя. В данном случае, при срабатывании события, бот отправляет сообщение пользователю с текстом "Вы сейчас проходите курс!".

Примечание 1.

При желании, сюда также можно добавить какую-нибудь клавиатуру.

Сейчас код выглядит так:

import asyncio
from aiogram import F
from aiogram import Bot, Dispatcher
from aiogram.types import Message, CallbackQuery
from aiogram.filters import CommandStart, Command
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardMarkup, KeyboardButton

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
], resize_keyboard=True)

help_kb = InlineKeyboardMarkup(inline_keyboard=[
    [InlineKeyboardButton(text='Курс', url='https://stepik.org/course/203657')],
    [InlineKeyboardButton(text='Тест', callback_data='Course')]
])

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!', reply_markup=kb)

@dp.message(Command('help'))
async def help_command(message: Message):
    await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь", reply_markup=help_kb)

@dp.callback_query(F.data == 'Course')
async def course_func(callback: CallbackQuery):
    await callback.message.answer('Вы сейчас проходите курс!')

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

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

Машина состояний

FSM (Finite State Machine) в Aiogram — это инструмент для управления состояниями бота. С его помощью можно организовать диалоги и другие процессы, которые требуют пошагового взаимодействия с пользователем.

Представьте, что бот — это магазин. FSM помогает боту помнить, на каком этапе "покупки" находится пользователь. Например:

Пользователь зашел в магазин (начальное состояние).
Выбрал товар (состояние выбора товара).
Оформляет заказ (состояние оформления заказа).
Подтвердил заказ (конечное состояние).

Вот как это работает в Aiogram:

Определение состояний: Вы заранее создаете набор состояний, через которые будет проходить пользователь.
Настройка переходов: Определяете, как пользователь будет переходить из одного состояния в другое (например, после выбора товара — переход к оформлению заказа).
Обработка каждого состояния: Для каждого состояния вы пишете обработчики, которые будут реагировать на действия пользователя.

Finite State Machine (FSM), или машина состояний, — это довольно сложная модель вычислений, которая описывает поведение системы, состоящее из конечного числа состояний, переходов между этими состояниями и действий, которые выполняются при переходах. В AIogram состояния дают возможность создавать логику, которая может меняться в зависимости от прошлых действий или внешних событий. Это позволяет создавать навигацию в боте, когда пользователю предлагается последовательность действий или беседа, проводимая в соответствии с состояниями. Работа с FSM в AIogram начинается с определения состояний и переходов между ними. При получении сообщения или другого события, бот перемещается из одного состояния в другое, выполняя при этом определенные действия в каждом состоянии. Импортируем все необходимое:

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

Сейчас нам нужно создать класс, который наследуется от StatesGroup, сейчас этот класс и есть машина состояний, далее создадим переменные, которые будет хранить FSM каждого пользователя, сделаем их объектами State():

class User(StatesGroup):
    name = State()
    amount = State()

Создадим первый обработчик, когда пользователь отправляет команду /fsm, срабатывает первый обработчик:

@dp.message(Command('fsm'))
async def fsm_first_step(message: Message, state: FSMContext):
    await state.set_state(User.name)
    await message.answer('Введите ваше имя')

@dp.message_handler(Command('fsm')) — декоратор, который указывает, что эта функция обрабатывает команду /fsm.
fsm_first_step — функция, которая выполняется при получении команды /fsm.
await state.set_state(User.name) — переводит пользователя в состояние User.name.
await message.answer('Введите ваше имя') — отправляет пользователю сообщение с просьбой ввести имя.

Когда пользователь вводит имя, срабатывает второй обработчик:

@dp.message_handler(state=User.name)
async def fsm_second_step(message: Message, state: FSMContext): 
    await state.update_data(name=message.text) await state.set_state(User.amount) await message.answer('Сколько уроков вы прошли?')

@dp.message_handler(state=User.name) — декоратор, который указывает, что эта функция обрабатывает сообщения в состоянии User.name.
fsm_second_step — функция, которая выполняется при получении имени пользователя.
await state.update_data(name=message.text) — сохраняет введенное имя в данные состояния.
await state.set_state(User.amount) — переводит пользователя в состояние User.amount.
await message.answer('Сколько уроков вы прошли?') — отправляет пользователю сообщение с просьбой ввести количество пройденных уроков.

Когда пользователь вводит количество пройденных уроков, срабатывает третий обработчик:

@dp.message_handler(state=User.amount)
async def fsm_third_step(message: Message, state: FSMContext): 
    await state.update_data(amount=message.text) data = await state.get_data() 
    await message.answer(f'Отлично!\nВас зовут: {data["name"]}\nВы прошли: {data["amount"]} уроков') 
    await state.clear()

@dp.message_handler(state=User.amount) — декоратор, который указывает, что эта функция обрабатывает сообщения в состоянии User.amount.
fsm_third_step — функция, которая выполняется при получении количества пройденных уроков.
await state.update_data(amount=message.text) — сохраняет введенное количество уроков в данные состояния.
data = await state.get_data() — получает все сохраненные данные состояния.
await message.answer(f'Отлично!\nВас зовут: {data["name"]}\nВы прошли: {data["amount"]} уроков') — отправляет пользователю итоговое сообщение с именем и количеством уроков.
await state.clear() — очищает состояние, возвращая пользователя в начальное состояние.

Как это работает в целом

Пользователь отправляет команду /fsm.
Бот запрашивает имя пользователя.
Пользователь вводит свое имя.
Бот запрашивает количество пройденных уроков.
Пользователь вводит количество уроков.
Бот отправляет итоговое сообщение с введенными данными и завершает процесс.

Каждый обработчик отвечает за определенное состояние и последовательно переводит пользователя от одного этапа к другому, пока все необходимые данные не будут собраны.

Примечание 1.

машина состояний обычно используется для создания регистрации в боте.

Примечание 2.

В конце написания логики FSM всегда очищайте state()

Примечание 3.

Если этого примера вам не хватило, то в конце курса будет разобран еще один.

Сейчас код выглядит так:

import asyncio
from aiogram import F
from aiogram import Bot, Dispatcher
from aiogram.types import Message, CallbackQuery
from aiogram.filters import CommandStart, Command
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardMarkup, KeyboardButton
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext

kb = ReplyKeyboardMarkup(keyboard=[
    [KeyboardButton(text='/start')],
    [KeyboardButton(text='/help')]
], resize_keyboard=True)

help_kb = InlineKeyboardMarkup(inline_keyboard=[
    [InlineKeyboardButton(text='Курс', url='https://stepik.org/course/203657')],
    [InlineKeyboardButton(text='Тест', callback_data='Course')]
])

bot = Bot('YOUR_TOKEN')
dp = Dispatcher()

class User(StatesGroup):
    name = State()
    amount = State()


@dp.message(CommandStart())
async def start(message: Message):
    await message.answer('Здравствуйте!', reply_markup=kb)

@dp.message(Command('help'))
async def help_command(message: Message):
    await message.reply("Команды бота:\n/start - Начать работу\n/help - Получить помощь", reply_markup=help_kb)

@dp.callback_query(F.data == 'Course')
async def course_func(callback: CallbackQuery):
    await callback.message.answer('Вы сейчас проходите курс!')

@dp.message(Command('fsm'))
async def fsm_first_step(message: Message, state: FSMContext):
    await state.set_state(User.name)
    await message.answer('Введите ваше имя')

@dp.message(User.name)
async def fsm_second_step(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(User.amount)
    await message.answer('Сколько уроков вы прошли?')

@dp.message(User.amount)
async def fsm_third_step(message: Message, state: FSMContext):
    await state.update_data(amount=message.text)
    data = await state.get_data()
    await message.answer(f'Отлично!\nВас зовут: {data["name"]}\nВы прошли: {data["amount"]} уроков')
    await state.clear()

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

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

Middleware

Данный урок является самым сложным из всего урока, для его прохождения понадобятся знания ООП, если вам что-то непонятно, можете пропустить урок, он необязателен.

Middleware в aiogram - это слой программного обеспечения, который находится между клиентом и сервером и обрабатывает запросы и ответы. В контексте aiogram, который является асинхронным фреймворком для создания Telegram-ботов на Python, middleware позволяет перехватывать и обрабатывать входящие обновления (updates) и сообщения до их обработки основным логическим кодом бота.

Основные функции middleware

Логирование: Отслеживание и запись входящих сообщений и обновлений.
Аутентификация: Проверка, имеет ли пользователь право на выполнение определенных действий.
Проверка данных: Валидация и предобработка данных перед их обработкой основным кодом.
Изменение данных: Изменение содержимого сообщений или других данных до их обработки.
Контроль доступа: Ограничение доступа к определенным командам или функциям на основе различных критериев.

Импортируем необходимое:

from aiogram.fsm.storage.memory import MemoryStorage
from aiogram import BaseMiddleware

MemoryStorage хоть к этому уроку и не относится, но рассказать про него стоит, это обычное хранилище.

Создадим middleware, который будет подсчитывать кол-во отправленных сообщений пользователем, для начала создадим класс, который наследуется от другого - BaseMiddleware, родительского класса, теперь у нас есть весь его функционал.

Напишем следующее:

class MessageCounterMiddleware(BaseMiddleware):
    def __init__(self):
        super().__init__()
        self.message_counts = {}

Определяется конструктор (__init__) класса:

super().__init__() вызывает конструктор родительского класса (BaseMiddleware), что позволяет инициализировать родительский класс правильно.
self.message_counts = {} создает пустой словарь для хранения количества сообщений каждого пользователя. Ключами словаря будут идентификаторы пользователей, а значениями — количество сообщений, отправленных этими пользователями.

Создадим метод __call__:

 async def __call__(self, handler, event, data):

Определяется асинхронный метод __call__, который делает этот объект вызываемым, подобно функции. Это ключевая часть middleware, которая будет выполняться при каждом событии:

handler — это функция, которая будет обработчиком события.
event — само событие, которое происходит (например, получение сообщения).
data — словарь с данными, который может быть изменен и передан обработчику

Далее напишем весь функционал и вот что у нас получится:

class MessageCounterMiddleware(BaseMiddleware):
    def __init__(self):
        super().__init__()
        self.message_counts = {}

    async def __call__(self, handler, event, data):
        if isinstance(event, Message):
            user_id = event.from_user.id
            if user_id not in self.message_counts:
                self.message_counts[user_id] = 0
            self.message_counts[user_id] += 1
            data['message_count'] = self.message_counts[user_id]
        return await handler(event, data)

Таким образом, этот middleware перехватывает каждое сообщение, отправленное пользователями, увеличивает счетчик сообщений для каждого пользователя и передает этот счетчик обработчику в виде дополнительного элемента данных.

Middleware написан, осталось его зарегистрировать(активировать), это будет выглядеть так:

dp = Dispatcher(storage=MemoryStorage())
dp.message.middleware(MessageCounterMiddleware())
dp.message.register(start_handler, Command(commands=['start']))
dp.message.register(message_handler)

Разберем каждую строчку детальнее:

dp = Dispatcher(storage=MemoryStorage())

Dispatcher — это объект, который управляет всеми обработчиками событий и middleware для вашего бота. Он распределяет события по зарегистрированным обработчикам.
storage=MemoryStorage() — передается параметр storage, который определяет, где будут храниться данные о пользователях и состояниях. В данном случае используется MemoryStorage, что означает хранение данных в оперативной памяти.

dp.message.middleware(MessageCounterMiddleware())

dp.message — это указатель на часть диспетчера, которая управляет сообщениями.
middleware(MessageCounterMiddleware()) — добавляет MessageCounterMiddleware в цепочку middleware для обработки сообщений. Это означает, что каждый раз, когда будет получено сообщение, оно сначала пройдет через MessageCounterMiddleware, который подсчитает количество сообщений от каждого пользователя, перед тем как сообщение будет обработано другими обработчиками.

dp.message.register(start_handler, Command(commands=['start']))

dp.message.register — регистрирует обработчик для сообщений.
start_handler — это функция-обработчик, которая будет вызвана, когда будет получена команда /start.
Command(commands=['start']) — указывает, что данный обработчик должен срабатывать на команду /start. Команды в Telegram начинаются с /, и Command — это фильтр, который проверяет, является ли сообщение командой и соответствует ли оно одной из указанных команд.

dp.message.register(message_handler)

dp.message.register — снова регистрирует обработчик для сообщений.
message_handler — это функция-обработчик, которая будет вызвана для всех остальных сообщений, которые не являются командами. Этот обработчик будет срабатывать на любые текстовые сообщения, отправленные пользователем.

Чтобы объяснить, как это все работает вместе:

Создается объект Dispatcher с использованием MemoryStorage для хранения данных.
Добавляется middleware MessageCounterMiddleware, который будет считать количество сообщений от каждого пользователя.
Регистрируется обработчик start_handler для команды /start.
Регистрируется общий обработчик message_handler для всех остальных сообщений.