aiogram/examples/finite_state_machine.py

128 lines
3.8 KiB
Python
Raw Normal View History

2021-05-25 01:00:58 +03:00
import asyncio
import logging
import sys
from os import getenv
2021-10-12 01:11:53 +03:00
from typing import Any, Dict
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
from aiogram import Bot, Dispatcher, F, Router, html
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import KeyboardButton, Message, ReplyKeyboardMarkup, ReplyKeyboardRemove
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
form_router = Router()
2021-05-25 01:00:58 +03:00
class Form(StatesGroup):
2021-10-12 01:11:53 +03:00
name = State()
like_bots = State()
language = State()
2021-05-25 01:00:58 +03:00
@form_router.message(commands=["start"])
2021-10-12 01:11:53 +03:00
async def command_start(message: Message, state: FSMContext) -> None:
2021-05-25 01:00:58 +03:00
await state.set_state(Form.name)
2021-10-12 01:11:53 +03:00
await message.answer(
"Hi there! What's your name?",
reply_markup=ReplyKeyboardRemove(),
)
2021-05-25 01:00:58 +03:00
@form_router.message(commands=["cancel"])
2021-10-12 01:11:53 +03:00
@form_router.message(F.text.casefold() == "cancel")
async def cancel_handler(message: Message, state: FSMContext) -> None:
2021-05-25 01:00:58 +03:00
"""
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()
2021-10-12 01:11:53 +03:00
await message.answer(
"Cancelled.",
reply_markup=ReplyKeyboardRemove(),
)
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
@form_router.message(Form.name)
async def process_name(message: Message, state: FSMContext) -> None:
2021-05-25 01:00:58 +03:00
await state.update_data(name=message.text)
2021-10-12 01:11:53 +03:00
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,
),
)
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
@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)
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
@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)
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
await message.reply(
"Cool! I'm too!\nWhat programming language did you use for it?",
reply_markup=ReplyKeyboardRemove(),
2021-05-25 01:00:58 +03:00
)
2021-10-12 01:11:53 +03:00
@form_router.message(Form.like_bots)
async def process_unknown_write_bots(message: Message, state: FSMContext) -> None:
await message.reply("I don't understand you :(")
2021-05-25 01:00:58 +03:00
2021-10-12 01:11:53 +03:00
@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()
text = (
"Thank for all! Python is in my hearth!\nSee you soon."
if message.text.casefold() == "python"
else "Thank for information!\nSee you soon."
)
await message.answer(text)
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..."
2021-05-25 01:00:58 +03:00
)
2021-10-12 01:11:53 +03:00
await message.answer(text=text, reply_markup=ReplyKeyboardRemove())
2021-05-25 01:00:58 +03:00
async def main():
bot = Bot(token=getenv("TELEGRAM_TOKEN"), parse_mode="HTML")
2021-10-12 01:11:53 +03:00
dp = Dispatcher()
dp.include_router(form_router)
2021-05-25 01:00:58 +03:00
await dp.start_polling(bot)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
asyncio.run(main())