95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""Интерфейсы ботов.
|
|
|
|
Интерфейс - класс, определяющий поведение бота. Приложения наследуют Interface, размечают методы декораторами
|
|
command/text/state и регистрируют класс через register. В админке боту выбирается один из зарегистрированных интерфейсов.
|
|
|
|
@register
|
|
class Shop(Interface):
|
|
title = 'Магазин'
|
|
|
|
@command('start')
|
|
@text('главное меню')
|
|
def menu(self, ctx):
|
|
ctx.reply('Привет', keyboard=[['каталог']])
|
|
|
|
@state('await_address')
|
|
def address(self, ctx):
|
|
ctx.data['address'] = ctx.text
|
|
ctx.state = ''
|
|
|
|
Порядок разбора сообщения: команда, текст кнопки, состояние диалога, fallback.
|
|
"""
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from .context import Context
|
|
|
|
_registry: dict[str, type['Interface']] = {}
|
|
|
|
COMMAND = 'command'
|
|
TEXT = 'text'
|
|
STATE = 'state'
|
|
|
|
|
|
def _route(kind: str, values: tuple[str, ...]):
|
|
def decorator(func):
|
|
func.__dict__.setdefault('_telegram_routes', []).extend((kind, value) for value in values)
|
|
return func
|
|
return decorator
|
|
|
|
|
|
def command(*names: str):
|
|
"""Обрабатывает команды /name. Имена без слеша, регистр не важен."""
|
|
return _route(COMMAND, tuple(name.lstrip('/').lower() for name in names))
|
|
|
|
|
|
def text(*texts: str):
|
|
"""Обрабатывает сообщение с точно таким текстом, например нажатие кнопки клавиатуры."""
|
|
return _route(TEXT, texts)
|
|
|
|
|
|
def state(*states: str):
|
|
"""Обрабатывает любое сообщение, пока чат находится в этом состоянии."""
|
|
return _route(STATE, states)
|
|
|
|
|
|
class Interface:
|
|
title = ''
|
|
_routes: dict[tuple[str, str], str] = {}
|
|
|
|
def __init_subclass__(cls, **kwargs):
|
|
super().__init_subclass__(**kwargs)
|
|
routes = {}
|
|
for klass in reversed(cls.__mro__):
|
|
for name, attr in vars(klass).items():
|
|
for route in getattr(attr, '_telegram_routes', ()):
|
|
routes[route] = name
|
|
cls._routes = routes
|
|
|
|
def dispatch(self, ctx: 'Context'):
|
|
for route in ctx.routes():
|
|
name = self._routes.get(route)
|
|
if name:
|
|
return getattr(self, name)(ctx)
|
|
return self.fallback(ctx)
|
|
|
|
def fallback(self, ctx: 'Context'):
|
|
"""Вызывается, если сообщение не подошло ни под один обработчик."""
|
|
|
|
|
|
def key(cls: type[Interface]) -> str:
|
|
return f'{cls.__module__}.{cls.__qualname__}'
|
|
|
|
|
|
def register(cls: type[Interface]) -> type[Interface]:
|
|
_registry[key(cls)] = cls
|
|
return cls
|
|
|
|
|
|
def get(interface_key: str) -> type[Interface] | None:
|
|
return _registry.get(interface_key)
|
|
|
|
|
|
def choices() -> list[tuple[str, str]]:
|
|
return [(k, cls.title or k) for k, cls in sorted(_registry.items())]
|