PBModular Вики

Russian

English

Russian

English

Appearance

🚦 Finite State Machine (FSM) ​

Введение ​

В PBModular встроена система конечных автоматов состояний (FSM), которая позволяет создавать обработчики для многоэтапного взаимодействия с пользователем, например, формы регистрации, викторины или последовательного ввода команд. Она работает путем отслеживания "состояния" каждого пользователя в основной базе данных бота.

Шаг 1: Определение машины состояний и состояний ​

  1. Создайте класс, который наследуется от base.states.StateMachine.
  2. Внутри этого класса определите свои состояния, создав атрибуты класса, которые являются экземплярами base.states.State.

Пример:

python
# В файле main.py или отдельном файле states.py
from base.states import StateMachine, State

class RegistrationFSM(StateMachine):
    waiting_for_name = State()
    waiting_for_age = State()
    # Состояние по умолчанию - None (состояние не установлено)

Шаг 2: Регистрация машины состояний ​

В вашем основном классе BaseModule переопределите свойство state_machine, чтобы вернуть ваш класс FSM.

python
# В классе вашего BaseModule (например, main.py)
from .my_fsm import RegistrationFSM # Предполагается, что вы определили его в файле my_fsm.py

class MyModule(BaseModule):
    # ... другие свойства ...

    @property
    def state_machine(self) -> type[StateMachine]:
        """Регистрирует FSM для этого модуля."""
        return RegistrationFSM

Шаг 3: Создание обработчиков, специфичных для конкретного состояния ​

Используйте параметр fsm_state в декораторах обработчиков (@command, @message и т.д.), чтобы указать, на какое состояние (состояния) должен реагировать обработчик.

  • Для обработки команды, которая начинает процесс, вы можете опустить fsm_state или установить fsm_state=None, так как пользователи будут находиться в состоянии по умолчанию (None).
  • Для обработки последующих сообщений установите fsm_state в соответствующий объект State из вашего класса FSM.

Шаг 4: Управление состоянием в обработчиках ​

Когда вы регистрируете StateMachine, ваши обработчики могут принимать дополнительный четвертый аргумент, обычно называемый sm_controller. Этот объект представляет собой экземпляр вашего класса StateMachine, привязанный к текущему пользователю, и предоставляет методы для управления его состоянием и данными.

  • sm_controller.get_state() -> Optional[str]: Получение имени текущего состояния пользователя.
  • sm_controller.set_state(State, data: Optional[dict] = None): Устанавливает состояние пользователя. Вы также можете хранить словарь данных вместе с ним.
  • sm_controller.get_data() -> dict: Получение сохраненных данных пользователя.
  • sm_controller.update_data(**kwargs): Обновление словаря сохраненных данных пользователя.
  • sm_controller.clear(keep_data: bool = False): Сбрасывает состояние пользователя на значение по умолчанию (None). По умолчанию это также очищает данные пользователя.

Полный пример:

python
# main.py
from base.module import BaseModule, command, message
from base.states import StateMachine, State
from pyrogram.types import Message

# 1. Определите FSM
class RegistrationFSM(StateMachine):
    waiting_for_name = State()
    waiting_for_age = State()

class FSMModule(BaseModule):
    # 2. Зарегистрируйте FSM
    @property
    def state_machine(self) -> type[StateMachine]:
        return RegistrationFSM

    # 3. Создайте обработчики
    @command("register") # fsm_state=None по умолчанию
    async def start_registration(self, message: Message, sm_controller: RegistrationFSM):
        """Запускает процесс регистрации пользователя."""
        await message.reply("Отлично! Как тебя зовут?")
        # 4. Установите следующее состояние
        await sm_controller.set_state(sm_controller.waiting_for_name)

    @message(fsm_state=RegistrationFSM.waiting_for_name)
    async def handle_name(self, message: Message, sm_controller: RegistrationFSM):
        """Обрабатывает ввод имени пользователя."""
        name = message.text
        await message.reply(f"Спасибо, {имя}! Сколько тебе лет?")
        # Обновление данных и установка следующего состояния
        await sm_controller.update_data(name=name)
        await sm_controller.set_state(sm_controller.waiting_for_age)

    @message(fsm_state=RegistrationFSM.waiting_for_age)
    async def handle_age(self, message: Message, sm_controller: RegistrationFSM):
        """Обработка ввода возраста пользователя."""
        if not message.text.isdigit():
            await message.reply("Пожалуйста, введите действительное число для вашего возраста").
            return # Держите пользователя в том же состоянии

        age = int(message.text)
        # Получите все данные и завершите работу
        user_data = await sm_controller.get_data()
        name = user_data.get("name")
        
        await message.reply(
            f"Registration complete!\n"
            f"Name: {name}\n"
            f"Age: {age}"
        )
        # Очистите состояние и данные
        await sm_controller.clear()

    @command("cancel")
    async def cancel_process(self, message: Message, sm_controller: RegistrationFSM):
        """Отменяет любой текущий процесс FSM."""
        current_state = await sm_controller.get_state()
        if current_state:
            await sm_controller.clear()
            await message.reply("Процесс отменен").
        else:
            await message.reply("Нечего отменять").