Кінцевий автомат (FSM)#
Кінцевий автомат, кінцевий автоматон, або машина станів (FSM, FSA, finite automaton, state machine) - це математична модель обчислень.
Це абстрактна машина, яка може перебувати в одному зі скінченної кількості станів у будь-який момент часу. Кінцевий автомат може переходити з одного стану в інший у відповідь на деякі вхідні дані; перехід з одного стану в інший називається переходом.
Кінцевий автомат визначається списком його станів, початковим станом і вхідними даними, які запускають кожен перехід.
Джерело: WikiPedia
Приклад використання#
Не всі функції бота можна реалізувати як єдиний обробник (handler), наприклад, якщо Вам потрібно буде збирати деякі дані від користувача в окремих кроках, вам потрібно буде використовувати FSM.
Гайда, подивимось як реалізувати це крок за кроком
Крок за кроком#
Перед обробкою будь-яких станів Вам потрібно буде вказати тип станів, які Ви хочете обробляти
class Form(StatesGroup):
name = State()
like_bots = State()
language = State()
А потім напишіть обробник (handler) для кожного стану окремо від початку діалогу
Тут діалог можна почати лише за допомогою команди /start
, тому давайте обробимо її та зробимо перехід користувача до стану Form.name
@form_router.message(CommandStart())
async def command_start(message: Message, state: FSMContext) -> None:
await state.set_state(Form.name)
await message.answer(
"Hi there! What's your name?",
reply_markup=ReplyKeyboardRemove(),
)
Після цього Вам потрібно буде зберегти деякі дані в пам’яті та перейти до наступного кроку.
@form_router.message(Form.name)
async def process_name(message: Message, state: FSMContext) -> None:
await state.update_data(name=message.text)
await state.set_state(Form.like_bots)
await message.answer(
f"Nice to meet you, {html.quote(message.text)}!\nDid you like to write bots?",
reply_markup=ReplyKeyboardMarkup(
keyboard=[
[
KeyboardButton(text="Yes"),
KeyboardButton(text="No"),
]
],
resize_keyboard=True,
),
)
На наступних кроках користувач може дати різні відповіді, це може бути «так», «ні» або будь-що інше
Обробка yes
і скоро нам потрібно буде обробити стан Form.language
@form_router.message(Form.like_bots, F.text.casefold() == "yes")
async def process_like_write_bots(message: Message, state: FSMContext) -> None:
await state.set_state(Form.language)
await message.reply(
"Cool! I'm too!\nWhat programming language did you use for it?",
reply_markup=ReplyKeyboardRemove(),
)
Обробка no
@form_router.message(Form.like_bots, F.text.casefold() == "no")
async def process_dont_like_write_bots(message: Message, state: FSMContext) -> None:
data = await state.get_data()
await state.clear()
await message.answer(
"Not bad not terrible.\nSee you soon.",
reply_markup=ReplyKeyboardRemove(),
)
await show_summary(message=message, data=data, positive=False)
І обробка будь-яких інших відповідей
@form_router.message(Form.like_bots)
async def process_unknown_write_bots(message: Message) -> None:
await message.reply("I don't understand you :(")
Всі можливі випадки кроку like_bots було розглянуто, нумо реалізуємо останній крок
@form_router.message(Form.language)
async def process_language(message: Message, state: FSMContext) -> None:
data = await state.update_data(language=message.text)
await state.clear()
if message.text.casefold() == "python":
await message.reply(
"Python, you say? That's the language that makes my circuits light up! 😉"
)
await show_summary(message=message, data=data)
async def show_summary(message: Message, data: Dict[str, Any], positive: bool = True) -> None:
name = data["name"]
language = data.get("language", "<something unexpected>")
text = f"I'll keep in mind that, {html.quote(name)}, "
text += (
f"you like to write bots with {html.quote(language)}."
if positive
else "you don't like to write bots, so sad..."
)
await message.answer(text=text, reply_markup=ReplyKeyboardRemove())
І тепер Ви виконали всі кроки на зображенні, але ви можете зробити можливість скасувати діалог, давайте зробимо це за допомогою команди або тексту
@form_router.message(Command("cancel"))
@form_router.message(F.text.casefold() == "cancel")
async def cancel_handler(message: Message, state: FSMContext) -> None:
"""
Allow user to cancel any action
"""
current_state = await state.get_state()
if current_state is None:
return
logging.info("Cancelling state %r", current_state)
await state.clear()
await message.answer(
"Cancelled.",
reply_markup=ReplyKeyboardRemove(),
)
Повний приклад#
1import asyncio
2import logging
3import sys
4from os import getenv
5from typing import Any, Dict
6
7from aiogram import Bot, Dispatcher, F, Router, html
8from aiogram.enums import ParseMode
9from aiogram.filters import Command, CommandStart
10from aiogram.fsm.context import FSMContext
11from aiogram.fsm.state import State, StatesGroup
12from aiogram.types import (
13 KeyboardButton,
14 Message,
15 ReplyKeyboardMarkup,
16 ReplyKeyboardRemove,
17)
18
19TOKEN = getenv("BOT_TOKEN")
20
21form_router = Router()
22
23
24class Form(StatesGroup):
25 name = State()
26 like_bots = State()
27 language = State()
28
29
30@form_router.message(CommandStart())
31async def command_start(message: Message, state: FSMContext) -> None:
32 await state.set_state(Form.name)
33 await message.answer(
34 "Hi there! What's your name?",
35 reply_markup=ReplyKeyboardRemove(),
36 )
37
38
39@form_router.message(Command("cancel"))
40@form_router.message(F.text.casefold() == "cancel")
41async def cancel_handler(message: Message, state: FSMContext) -> None:
42 """
43 Allow user to cancel any action
44 """
45 current_state = await state.get_state()
46 if current_state is None:
47 return
48
49 logging.info("Cancelling state %r", current_state)
50 await state.clear()
51 await message.answer(
52 "Cancelled.",
53 reply_markup=ReplyKeyboardRemove(),
54 )
55
56
57@form_router.message(Form.name)
58async def process_name(message: Message, state: FSMContext) -> None:
59 await state.update_data(name=message.text)
60 await state.set_state(Form.like_bots)
61 await message.answer(
62 f"Nice to meet you, {html.quote(message.text)}!\nDid you like to write bots?",
63 reply_markup=ReplyKeyboardMarkup(
64 keyboard=[
65 [
66 KeyboardButton(text="Yes"),
67 KeyboardButton(text="No"),
68 ]
69 ],
70 resize_keyboard=True,
71 ),
72 )
73
74
75@form_router.message(Form.like_bots, F.text.casefold() == "no")
76async def process_dont_like_write_bots(message: Message, state: FSMContext) -> None:
77 data = await state.get_data()
78 await state.clear()
79 await message.answer(
80 "Not bad not terrible.\nSee you soon.",
81 reply_markup=ReplyKeyboardRemove(),
82 )
83 await show_summary(message=message, data=data, positive=False)
84
85
86@form_router.message(Form.like_bots, F.text.casefold() == "yes")
87async def process_like_write_bots(message: Message, state: FSMContext) -> None:
88 await state.set_state(Form.language)
89
90 await message.reply(
91 "Cool! I'm too!\nWhat programming language did you use for it?",
92 reply_markup=ReplyKeyboardRemove(),
93 )
94
95
96@form_router.message(Form.like_bots)
97async def process_unknown_write_bots(message: Message) -> None:
98 await message.reply("I don't understand you :(")
99
100
101@form_router.message(Form.language)
102async def process_language(message: Message, state: FSMContext) -> None:
103 data = await state.update_data(language=message.text)
104 await state.clear()
105
106 if message.text.casefold() == "python":
107 await message.reply(
108 "Python, you say? That's the language that makes my circuits light up! 😉"
109 )
110
111 await show_summary(message=message, data=data)
112
113
114async def show_summary(message: Message, data: Dict[str, Any], positive: bool = True) -> None:
115 name = data["name"]
116 language = data.get("language", "<something unexpected>")
117 text = f"I'll keep in mind that, {html.quote(name)}, "
118 text += (
119 f"you like to write bots with {html.quote(language)}."
120 if positive
121 else "you don't like to write bots, so sad..."
122 )
123 await message.answer(text=text, reply_markup=ReplyKeyboardRemove())
124
125
126async def main():
127 bot = Bot(token=TOKEN, parse_mode=ParseMode.HTML)
128 dp = Dispatcher()
129 dp.include_router(form_router)
130
131 await dp.start_polling(bot)
132
133
134if __name__ == "__main__":
135 logging.basicConfig(level=logging.INFO, stream=sys.stdout)
136 asyncio.run(main())