Привет, я Максим Королев из Петрович-ТЕХа. В предыдущей статье рассказывал, как мы сделали семейство Telegram-ботов для ITSM. Один из ключевых - «Дежурный»: с ним администраторы фиксируют аварии, регламентные работы и прочее, создают задачи в Jira, публикуют уведомления в каналы и в интранет компании.
И тут Telegram ограничивают. Когда у дежурного в разгар аварии «не грузится телега» - это риск для бизнеса. Нужно было срочно решать проблему.
Единственной стабильной альтернативой оказался - да-да - MAX. У него есть Bot API, он не попадает под ограничения РКН и уверенно работает на территории РФ.
Задача не «сделать второго бота с нуля», а сохранить одного логического Дежурного и дать ему второй вход из MAX. В статье расскажу, как я изменил архитектуру: вынес всю бизнес-логику в отдельный слой CORE, а Telegram и MAX сделал вызовами команд этого ядра.
Как было: все в обработчиках
Изначально «Дежурный» был классическим aiogram-ботом:
Обработчики команд и callback'ов жили в handlers/
Внутри них же вызывались Jira, SimpleOne (Petlocal), отправка в каналы, обновление состояния
FSM (конечный автомат диалога) и данные - в состоянии aiogram и в общем bot_state (активные сбои/работы)
Схема:
Пользователь (Telegram) → aiogram handlers → Jira / каналы / Petlocal / bot_state
Когда Telegram начало штормить, мне нужно было подключить MAX максимально быстро. Но при такой архитектуре единственный путь - скопировать всю логику в новый слой под maxapi: те же создания сбоев, остановки, продления, те же проверки и форматы. Два места для поддержки и источника багов - не вариант, особенно когда делаешь все в авральном режиме.
Что сделал: CORE + два адаптера
Я перешел к схеме «одно ядро - два канала»:
CORE - это пакет core/ в репозитории. Он не знает про Telegram и MAX. Ему передают:
Данные (описание сбоя, ID, минуты продления и т.д.)
reply_fn - функцию «отправить ответ пользователю» (в нужном канале)
При необходимости экземпляр Telegram-бота для отправки в общие каналы и вызова SCM
Обработчики в Telegram и в MAX только:
Разбирают ввод пользователя (команда, текст, callback)
Проверяют права (админ TG / админ MAX)
Вызывают функции из CORE и передают им reply_fn, сформированную под свой мессенджер
Так я получил одну точку правды для бизнес-логики и два тонких транспортных слоя.
Что внутри CORE
В CORE три группы функций.
Создание событий (core/creation.py)
create_alarm - создание сбоя: Jira (опционально), SCM, пост в канал Telegram (и при настройке - в MAX), пост на Petlocal, запись в bot_state.active_alarms
create_maintenance - регламентная работа: каналы, Petlocal, bot_state.active_maintenances
send_regular_message - обычное сообщение в канал (и при необходимости в MAX/Petlocal)
Все три принимают reply_fn - функцию, которая знает, куда отправить ответ пользователю («Сбой создан», «Ошибка: …»). Для Telegram это message.answer / edit_text, для MAX - event.message.answer или отправка через MaxService. Самому CORE без разницы - он просто вызывает reply_fn и не думает о транспорте.
Действия над событиями (core/actions.py)
stop_alarm(alarm_id, telegram_bot, reply_fn) - закрытие сбоя: SCM, пост «Сбой устранен» на Petlocal, удаление из состояния, уведомление в каналы, ответ пользователю через reply_fn
stop_maintenance(work_id, …) - завершение работы, аналогично
extend_alarm(alarm_id, minutes, …) / extend_maintenance(…) - продление времени с обновлением состояния и каналов
Здесь же живет общая отправка в каналы (send_to_alarm_channels) и работа с SimpleOne/SCM. Вызывается одинаково - и из Telegram-обработчиков, и из MAX. Когда я переписывал бота под MAX, мне не пришлось трогать ни строчки в actions.py - достаточно было передать правильный reply_fn из нового адаптера.
Справочные функции (core/events.py, core/help_text.py)
get_active_events_text(view, page, html=False) - текст списка активных сбоев или работ (для «Текущих событий»)
get_help_text(html=False) - текст справки
Параметр html - единственная уступка различиям между мессенджерами: Telegram понимает HTML-разметку, MAX - нет, ему отдается обычный текст. Одна реализация, два канала - ровно то, чего я добивался при рефакторинге.
Как Telegram и MAX вызывают CORE
Telegram
В handlers/alarm/confirmation.py по нажатию «Подтвердить» вызывается create_alarm, create_maintenance или send_regular_message из core.creation. В reply_fn передается callback.message.edit_text и answer с главным меню
В handlers/manage/ при остановке и продлении - та же схема: вызов stop_alarm / extend_alarm из core.actions. Часть обработчиков я еще перевожу на этот паттерн, но архитектурно цель ясная - везде один вызов в CORE, без локальной логики
MAX
В adapters/max/handlers.py зарегистрированы команды и callback'и maxapi:
/start, /help → get_help_text(html=False), ответ уходит в чат MAX
«Текущие события», «Список», кнопки сбоев/работ → get_active_events_text(..., html=False)
«Остановить <id>» / «Продлить <id> <минуты>» и кнопки с payload → вызовы stop_alarm, stop_maintenance, extend_alarm, extend_maintenance из core.actions. Все через reply_fn, который шлет ответ обратно в MAX
Сценарий «Сообщить» (сбой / работа / обычное сообщение) живет в adapters/max/create_flow.py: пошаговый диалог с сессиями по user_id, в конце - вызов create_alarm / create_maintenance / send_regular_message из core.creation
MAX-адаптер не содержит бизнес-логики - только разбор ввода, проверку MAX_ADMIN_IDS и вызов CORE. Именно поэтому его удалось написать быстро: когда ядро уже готово, новый транспорт - это по сути маппинг команд на функции.
Как выглядит reply_fn на практике
До сих пор я описывал reply_fn абстрактно - «функция, которая знает, куда ответить». Покажу, как это выглядит в коде.
Сигнатура одна: async def reply_fn(text: str) -> None. В MAX дополнительно передаются attachments для клавиатуры, но CORE об этом не знает.
Пример: формирование reply_fn в Telegram-адаптере. После нажатия «Подтвердить» оборачиваем edit_text и answer с главным меню в одну функцию и передаем в CORE.
@router.callback_query(F.data == "confirm_send") async def confirm_send_callback(callback: CallbackQuery, state: FSMContext): data = await state.get_data() msg_type = data.get("type") async def reply_fn(text: str): await callback.message.edit_text(text, parse_mode="HTML", reply_markup=None) await callback.message.answer( "Выберите действие:", reply_markup=create_main_keyboard(user_id), ) if msg_type == "alarm": await create_alarm(data, callback.bot, reply_fn, user_id) elif msg_type == "maintenance": await create_maintenance(data, callback.bot, reply_fn, user_id) # ...
Пример: то же самое в MAX-адаптере. При нажатии кнопки «Остановить» извлекаем item_id из payload и вызываем тот же stop_alarm - только reply_fn шлет ответ в MAX.
if payload.startswith("action_stop_a_"): item_id = payload.replace("action_stop_a_", "", 1) if not telegrambot: await event.message.answer("❌ Сервис недоступен.", attachments=main_menu()) return async def reply_fn(t: str): await event.message.answer(t, attachments=main_menu()) await stop_alarm(item_id, telegrambot, reply_fn) clear_manage_session(uid) return
Почему так:
В Telegram reply_fn дергает callback.message.edit_text и answer - редактирует сообщение и показывает меню
В MAX - только event.message.answer с attachments (клавиатура)
CORE вызывается одинаково: stop_alarm(..., reply_fn) - ему без разницы, откуда пришел запрос
Добавление третьего мессенджера - это еще один reply_fn, без изменений в CORE
От кнопки до ответа: путь через MAX-адаптер
Типичный сценарий: дежурный в MAX нажимает «Остановить» у сбоя → приходит MessageCallback с payload="action_stop_a_FA-1234" → обработчик парсит ID, проверяет права, формирует reply_fn и передает управление в CORE.
Пример: полный обработчик callback в MAX. Показывает проверку прав, разбор payload и вызов CORE.
# adapters/max/handlers.py (фрагмент) @dp.message_callback() async def on_callback(event): payload = getattr(event.callback, "payload", None) or "" uid = getattr(event.callback.user, "user_id", None) if not is_max_admin(uid): await event.message.answer("❌ У вас нет прав...") return if payload.startswith("action_stop_a_"): item_id = payload.replace("action_stop_a_", "", 1) async def reply_fn(t: str): await event.message.answer(t, attachments=main_menu()) await stop_alarm(item_id, telegrambot, reply_fn) clear_manage_session(uid) return
Почему так:
is_max_admin(uid) проверяет user_id по списку MAX_ADMIN_IDS из .env - аналог ADMIN_IDS в Telegram
Внутри stop_alarm в CORE выполняется SCM, Petlocal, удаление из bot_state, отправка в каналы через telegram_bot
Пользователю в MAX уходит ответ только через переданный reply_fn - CORE не знает, что работает с MAX
Что внутри CORE: фрагмент create_alarm
Пример: упрощенная функция создания сбоя из CORE. Обратите внимание - нет импортов aiogram или maxapi, только бизнес-логика и вызов reply_fn.
# core/creation.py (упрощенно) ReplyFn = Callable[[str], Awaitable[Any]] async def create_alarm( data: dict, telegram_bot: Any, # для отправки в общие каналы (TG/MAX) reply_fn: ReplyFn, # ответ именно этому пользователю (TG или MAX) user_id: int, ) -> bool: description_text = data.get("description", "") issue = description_text[:100] if description_text else "Проблема не указана" fix_time = dt.fromisoformat(data["fix_time"]) if isinstance(data["fix_time"], str) else data["fix_time"] create_jira = data.get("create_jira", True) if create_jira: try: jira_response = await create_failure_issue(...) alarm_id = jira_response["key"] except Exception as jira_error: alarm_id = str(uuid.uuid4())[:8] await reply_fn("⚠️ Не удалось создать задачу в Jira. Авария создана с локальным ID.") else: alarm_id = str(uuid.uuid4())[:8] bot_state.active_alarms[alarm_id] = { "issue": issue, "fix_time": fix_time.isoformat(), "user_id": user_id, "service": data["service"], } await send_to_alarm_channels(telegram_bot, ...) await reply_fn(f"🚨 Сбой {alarm_id} создан.") return True
Почему так:
ReplyFn - единственный «мессенджерный» контракт. Кто вызвал CORE (Telegram или MAX), тот и задал, куда текст уйдет
telegram_bot передается для отправки в общие каналы - это единственная зависимость от конкретного транспорта
При ошибке Jira бот не падает - создает локальный ID и предупреждает пользователя
Весь код тестируется без мессенджеров: подставляем мок reply_fn и проверяем логику
Telegram Bot API vs MAX Bot API: что нужно знать
Для тех, кто задумает аналогичный двухканальный бот - вот практические различия, на которые я наткнулся при переходе.
Оба API - REST. Telegram работает через api.telegram.org, MAX - через botapi.max.ru. У MAX есть официальная OpenAPI-спецификация в репозитории max-messenger/max-bot-api-schema. Получение событий - long polling в обоих случаях (webhook для MAX я не использовал).
Кнопки и callback
Telegram | MAX | |
|---|---|---|
Тип кнопки | InlineKeyboardButton с callback_data | CallbackButton с payload |
Ограничение данных | Строка до 64 байт | Строка (payload) |
Событие нажатия | CallbackQuery | MessageCallback |
Отправка клавиатуры | reply_markup=InlineKeyboardMarkup(...) | attachments=[ButtonsPayload(buttons=rows).pack()] |
URL-кнопки | Есть | Нет в том же виде |
Концептуально одно и то же: «кнопка → payload → обработчик». Но API и типы разные, поэтому клавиатуры для MAX собираются отдельно в adapters/max/keyboards.py.
FSM
В aiogram есть встроенный FSM: StatesGroup, FSMContext, storage (memory, Redis). В maxapi готового FSM нет. Поэтому для сценария «Сообщить» в MAX я реализовал свой пошаговый механизм с сессиями - подробности ниже.
Разметка и форматирование
В Telegram я активно использую HTML (parse_mode="HTML"). В MAX передаю обычный текст - отсюда параметр html=False в get_help_text и get_active_events_text. Это та самая «единственная уступка различиям», о которой говорил выше.
Что такое maxapi
maxapi - Python-клиент к MAX Bot API. Репозиторий - max-messenger/max-botapi-python, в README указано, что форк проверен командой MAX. Установка: pip install maxapi (на Windows для декодирования Brotli часто нужен еще brotlicffi). По духу похож на aiogram: Dispatcher, обработчики по типу события (message_created, message_callback), команды через Command('start').
Сессии в MAX: свой FSM за час
В maxapi нет аналога aiogram FSM. Мне нужен был пошаговый диалог для сценария «Сообщить» (сбой / работа / обычное сообщение), и нужен был быстро - Telegram уже штормило.
Два слоя сессий
Я завел два словаря в памяти (adapters/max/sessions.py):
# Сценарий «Сообщить»: user_id → {"step": str, "data": dict} _sessions: Dict[int, Dict[str, Any]] = {} # Сценарий «Управлять»: user_id → {"step": str, "item_id": str|None, "item_type": str|None} managesessions: Dict[int, Dict[str, Any]] = {}
Функции доступа: get_session, set_session, update_session_data, clear_session; для «Управлять» - аналогично с префиксом manage_.
Переход по шагам и валидация ввода сосредоточены в create_flow.py и в обработчиках callback в handlers.py. CORE вызывается только в конце цепочки - при создании сбоя, работы или сообщения.
Почему в памяти
Один инстанс бота, небольшое число дежурных - персистентное хранилище здесь избыточно. Перезапуск очищает сессии, и это нормально: незавершенный диалог «Сообщить» - не критичные данные. Если нагрузка вырастет - переехать будет делом одного дня.
Таймауты и брошенные диалоги
TTL нет - сессия живет до явного выхода («Назад», «Отмена») или до перезапуска. При желании можно добавить фоновую задачу, которая раз в N минут удаляет записи старше K минут
Конфликты (один пользователь открыл «Сообщить» в двух чатах) не обрабатываю: ключ - user_id, «второй» вход перезапишет первый. На практике дежурные работают из одного клиента MAX
Отмена работает на любом шаге: ввод «отмена» / «отменить» / «cancel» вызывает clear_session(user_id) и ответ «Действие отменено»
Этого хватает для полноценного пошагового сценария без встроенного FSM и без персистентного хранилища.
Один процесс, два поллинга
Запуск в main.py устроен просто:
Поднимается aiogram-бот (Telegram polling)
Если в конфиге включен MAX и задан MAX_BOT_TOKEN, в той же event loop создается задача run_max_polling(bot) - в ней крутится maxapi polling
Telegram-бот передается в MAX-адаптер, потому что CORE при создании событий, остановке и продлении шлет уведомления в общие каналы (Telegram и при необходимости MAX) - для этого используется один и тот же экземпляр aiogram Bot
Один процесс, общее состояние (bot_state), общая логика (CORE) - и два параллельных поллинга. Никаких отдельных сервисов или межпроцессной синхронизации.
Права
Telegram: как раньше - ADMIN_IDS, SUPERADMIN_IDS в .env
MAX: в .env добавлены MAX_BOT_TOKEN, MAX_ADMIN_IDS (список user_id через запятую), MAX_MANAGEMENT_ENABLED. Проверка прав - по user_id из события в списке MAX_ADMIN_IDS
Админы MAX получают тот же функционал, что и в Telegram: создание сбоев/работ/сообщений, список событий, остановка, продление. Разницы в возможностях нет - отличается только транспорт.
Итог
Telegram и MAX стали тонкими адаптерами, которые парсят команды и коллбэки, а потом вызывают CORE, передавая ему reply_fn - способ ответить пользователю в нужном мессенджере.
Результат: работа с ботом возможна и из Telegram, и из MAX, данные и сценарии единые. Один «Дежурный», два входа, ноль дублирования бизнес-логики.
Если вы тоже задумываетесь о втором мессенджере или хотите вытащить логику из «толстых» обработчиков в ядро — надеюсь, этот опыт с CORE и двумя адаптерами окажется полезным. А с учетом того, что происходит с Telegram в России прямо сейчас, задача может стать актуальной быстрее, чем кажется.
