aiogram/examples/finite_state_machine.py
Kostiantyn Kriuchkov c516ea9d6a
Refactor and improve bot examples (#1256)
* Refactor and improve bot messages

Refactored bot code to use aiogram enumerations and enhanced chat messages with markdown beautifications for a more user-friendly display.

CommandStart() is now used instead of Command('start') for readability.

Furthermore, the bot's 'stop' command was improved, ensuring it executes appropriately during KeyboardInterrupt or SystemExit.

Additionally, the bot's logging was adjusted to output to sys.stdout for better logs' readability.

* Added Changelog

* Add guidance comments on obtaining bot tokens from environment variables

* Remove hardcoded tokens, opt for environment variable

* Remove unnecessary spaces and reorganize imports

* Fix error, switch default storage from Redis to Memory, and add logging to multibot example
2023-08-10 22:10:30 +03:00

139 lines
4 KiB
Python

import asyncio
import logging
import sys
from os import getenv
from typing import Any, Dict
from aiogram import Bot, Dispatcher, F, Router, html
from aiogram.enums import ParseMode
from aiogram.filters import Command, CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import (
KeyboardButton,
Message,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
)
TOKEN = getenv("BOT_TOKEN")
form_router = Router()
class Form(StatesGroup):
name = State()
like_bots = State()
language = State()
@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(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(),
)
@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,
),
)
@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, 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(),
)
@form_router.message(Form.like_bots)
async def process_unknown_write_bots(message: Message) -> None:
await message.reply("I don't understand you :(")
@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())
async def main():
bot = Bot(token=TOKEN, parse_mode=ParseMode.HTML)
dp = Dispatcher()
dp.include_router(form_router)
await dp.start_polling(bot)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
try:
asyncio.run(main())
except (KeyboardInterrupt, SystemExit):
logging.info("Bot stopped!")