От чатов к процессам: как бот склеивает TG и MAX и помогает жить по ITIL 4
От чатов к процессам: как бот склеивает TG и MAX и помогает жить по ITIL 4

Привет, я Максим Королев из Петрович-Теха. В прошлой статье про «Дежурного» я рассказывал, как мы ушли от «толстых» обработчиков aiogram к архитектуре CORE + два адаптера — Telegram и MAX.

С замедлением Telegram стало понятно, что нам нужен стабильный резервный канал. Часть команды уже перешла на MAX, часть осталась в Telegram. И вот мы столкнулись с классической проблемой — один коллега пишет в TG, другой отвечает из MAX, диалог разрывается, контекст теряется, инцидент ведется «вслепую». Нужен был мост. И мы ввели параллельную работу в Telegram и в MAX, соединили эти два мира между собой и глубже встроили бота в процессы ITIL 4/ITSM. 

Главное сделали так, что:

  • коллеги могут общаться между собой «TG ↔ MAX», не теряя диалог

  • при закрытии инцидента бот автоматически архивирует историю общения в Jira — для дальнейшего RCA

В этой статье расскажу:

  • как развивался «Дежурный»

  • как устроен мост между чатами Telegram и MAX (с примером кода)

  • как реализована автоархивация диалога в Jira (с примером кода)

  • и почему всё это хорошо ложится на ITIL 4 / ITSM

Один CORE, два мессенджера и «единый диалог»

Архитектурная база осталась той же.

CORE: ничего не знает про Telegram и MAX; оперирует инцидентами, регламентными работами, приоритетами, SLA; принимает reply_fn и при необходимости telegram_bot.

Адаптеры Telegram/MAX: разбирают вход (команды, payload, сообщения); проверяют права, вызывают функции CORE и/или пересылают сообщения.

Когда мы решили сделать единый диалог поверх двух мессенджеров, базовая идея была простой. Есть один логический «чат по инциденту» и два его «физических отображения» — чат в Telegram и чат в MAX. И бот знает, какие чаты связаны, и ретранслирует сообщения между ними.

Так как CORE ничего не знает о конкретных каналах связи, просто пишешь адаптер к любому мессенджеру, не трогая логику управления инцидентами. Это сильно экономит время на тестирование и снижает риск сломать что-то в бою при добавлении нового канала.

Пример: мост между Telegram и MAX

Хранилище связок чатов

Сначала нам нужен простой маппинг «инцидент → TG-чат и MAX-чат»:

# bridges.py
from dataclasses import dataclass
from typing import Dict, Optional
@dataclass
class ChatBridge:
    incident_id: str
    telegram_chat_id: int
    max_chat_id: str  # формат зависит от maxapi
# incident_id -> ChatBridge
_bridges: Dict[str, ChatBridge] = {}
def register_bridge(incident_id: str, telegram_chat_id: int, max_chat_id: str) -> None:
    _bridges[incident_id] = ChatBridge(
        incident_id=incident_id,
        telegram_chat_id=telegram_chat_id,
        max_chat_id=max_chat_id,
    )
def get_bridge_by_incident(incident_id: str) -> Optional[ChatBridge]:
    return _bridges.get(incident_id)
def get_bridge_by_telegram_chat(chat_id: int) -> Optional[ChatBridge]:
    for bridge in _bridges.values():
        if bridge.telegram_chat_id == chat_id:
            return bridge
    return None
def get_bridge_by_max_chat(chat_id: str) -> Optional[ChatBridge]:
    for bridge in _bridges.values():
        if bridge.max_chat_id == chat_id:
            return bridge
    return None

Регистрация связки может происходить:

  • при создании инцидента (когда бот создаёт/привязывает рабочие чаты);

  • или при явной команде «связать этот TG-чат и этот MAX-чат с инцидентом X».

Telegram: прием сообщения и пересылка в MAX

Пример адаптера на aiogram:

# adapters/telegram/bridge_handlers.py
from aiogram import Router, F
from aiogram.types import Message
from .max_client import send_message_to_max_chat
from .bridges import get_bridge_by_telegram_chat
router = Router()
@router.message(F.text)
async def on_telegram_message(message: Message):
    # Системные команды обрабатываются своими хендлерами
    if message.text.startswith("/"):
        return
    bridge = get_bridge_by_telegram_chat(message.chat.id)
    if not bridge:
        # Этот чат ни с чем не связан — обычный диалог, ничего не шлём в MAX
        return
    text = message.text
    # Добавляем лёгкий маркер, кто и откуда пишет
    prefix = f"[TG] {message.from_user.full_name}: "
    relay_text = prefix + text
    await send_message_to_max_chat(
        chat_id=bridge.max_chat_id,
        text=relay_text,
    )
send_message_to_max_chat — это тонкая обёртка над maxapi:
# adapters/telegram/max_client.py
from maxapi import MaxBotClient  # пример, зависит от вашей обёртки
max_client = MaxBotClient(token="...")
async def send_message_to_max_chat(chat_id: str, text: str):
    await max_client.send_text(
        chat_id=chat_id,
        text=text,
    )

MAX: прием сообщения и пересылка в Telegram

Со стороны maxapi идея та же: 

# adapters/max/bridge_handlers.py
from maxapi.dispatcher import Dispatcher
from maxapi.events import MessageCreated
from .telegram_client import send_message_to_telegram_chat
from .bridges import get_bridge_by_max_chat
dp = Dispatcher()
@dp.message_created()
async def on_max_message(event: MessageCreated):
    message = event.message
    text = getattr(message, "text", "") or ""
    # Команды, callback-и и служебные вещи обрабатываются отдельно
    if text.startswith("/"):
        return
    chat_id = getattr(message.chat, "chat_id", None)
    if not chat_id:
        return
    bridge = get_bridge_by_max_chat(chat_id)
    if not bridge:
        return
    user = getattr(message, "from_", None)
    user_name = getattr(user, "display_name", "MAX user")
    prefix = f"[MAX] {user_name}: "
    relay_text = prefix + text
    await send_message_to_telegram_chat(
        chat_id=bridge.telegram_chat_id,
        text=relay_text,
    )

И небольшой Telegram-клиент: 

# adapters/max/telegram_client.py
from aiogram import Bot
telegram_bot = Bot(token="...")
async def send_message_to_telegram_chat(chat_id: int, text: str):
    await telegram_bot.send_message(
        chat_id=chat_id,
        text=text,
        disable_web_page_preview=True,
    )

В итоге:

  • Коллега А пишет в TG — сообщение уходит в связанный MAX-чат.

  • Коллега Б отвечает из MAX — сообщение уходит в TG.

  • Все видят единый диалог, просто в своём любимом клиенте.

Про медиа-контент. Сейчас бот пересылает только текст. Фото, видео, документы в планах, но пока не реализовано (нет острой нужды). Если у вас в команде активно шлют скриншоты в инцидент-чаты, это важно учесть заранее.

В коде мы отсекаем сообщения, начинающиеся с / (служебные команды). Они обрабатываются собственными хендлерами и в MAX не уходят. Это сделано намеренно, чтобы не засорять чат командным мусором.

Пример: автоархивация диалога в Jira при закрытии инцидента

Теперь к второй важной части — автоархивации диалога.

Сформулировали идею — пока инцидент «живой», в связанных чатах идёт обсуждение. При закрытии инцидента бот собирает историю диалога (по связанным чатам и ID инцидента), красиво форматирует, сохраняет в Jira как комментарий (наш вариант) или как вложение.

Хранилище событий чата по инциденту

Лог завязан не на конкретный чат, а на incident_id: бот пишет все сообщения из связанных TG- и MAX-чатов в одну ленту, получается единый диалог поверх всех каналов. Пример простого in‑memory-хранилища (у вас ваша БД):

# conversation_log.py
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List
@dataclass
class ChatMessage:
    timestamp: datetime
    source: str       # "TG" или "MAX"
    user: str
    text: str
# incident_id -> список сообщений
_logs: Dict[str, List[ChatMessage]] = {}
def append_message(incident_id: str, source: str, user: str, text: str) -> None:
    _logs.setdefault(incident_id, []).append(
        ChatMessage(
            timestamp=datetime.utcnow(),
            source=source,
            user=user,
            text=text,
        )
    )
def get_messages(incident_id: str) -> List[ChatMessage]:
    return _logs.get(incident_id, [])
def clear_log(incident_id: str) -> None:
    _logs.pop(incident_id, None)
Теперь доработаем мосты, чтобы они писали в лог:
# adapters/telegram/bridge_handlers.py (фрагмент)
from conversation_log import append_message
from .bridges import get_bridge_by_telegram_chat
@router.message(F.text)
async def on_telegram_message(message: Message):
    if message.text.startswith("/"):
        return
    bridge = get_bridge_by_telegram_chat(message.chat.id)
    if not bridge:
        return
    user_name = message.from_user.full_name
    text = message.text
    # 1. Пишем в лог общения по инциденту
    append_message(
        incident_id=bridge.incident_id,
        source="TG",
        user=user_name,
        text=text,
    )
    # 2. Ретранслируем в MAX, как раньше
    prefix = f"[TG] {user_name}: "
    relay_text = prefix + text
    await send_message_to_max_chat(
        chat_id=bridge.max_chat_id,
        text=relay_text,
    )
И аналогично для MAX:
# adapters/max/bridge_handlers.py (фрагмент)
from conversation_log import append_message
from .bridges import get_bridge_by_max_chat
@dp.message_created()
async def on_max_message(event: MessageCreated):
    message = event.message
    text = getattr(message, "text", "") or ""
    if text.startswith("/"):
        return
    chat_id = getattr(message.chat, "chat_id", None)
    if not chat_id:
        return
    bridge = get_bridge_by_max_chat(chat_id)
    if not bridge:
        return
    user = getattr(message, "from_", None)
    user_name = getattr(user, "display_name", "MAX user")
    # 1. Пишем в лог
    append_message(
        incident_id=bridge.incident_id,
        source="MAX",
        user=user_name,
        text=text,
    )
    # 2. Ретранслируем в Telegram
    prefix = f"[MAX] {user_name}: "
    relay_text = prefix + text
    await send_message_to_telegram_chat(
        chat_id=bridge.telegram_chat_id,
        text=relay_text,
    )

Формирование текста архива

При закрытии инцидента нам нужен легко читаемый лог:

# conversation_log_formatter.py
from typing import List
from conversation_log import ChatMessage
def format_log_markdown(messages: List[ChatMessage]) -> str:
    if not messages:
        return "Диалог по инциденту отсутствует."
    lines = ["h2. Диалог по инциденту (Telegram + MAX)", ""]
    for msg in sorted(messages, key=lambda m: m.timestamp):
        ts = msg.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")
        prefix = f"*{ts}* [{msg.source}] {msg.user}:"
        # Простейший эскейп спецсимволов Jira/Markdown можно добавить по вкусу
        lines.append(f"{prefix}\n  {msg.text}")
    return "\n".join(lines)

Здесь пример для wiki-разметки Jira (h2., жирный и т.п.), можно адаптировать под ваш формат.

Встраивание в CORE: закрытие инцидента

Теперь в функции CORE, которая отвечает за остановку/закрытие инцидента, мы добавляем шаг «собрать и сохранить лог». Упрощенный пример:

# core/actions.py
from typing import Awaitable, Callable, Any
from conversation_log import get_messages, clear_log
from conversation_log_formatter import format_log_markdown
from integrations.jira_client import add_comment_to_issue
ReplyFn = Callable[[str], Awaitable[Any]]
# Функция stop_alarm здесь играет роль шага Resolve/Close в процессе Incident Management: CORE завершает инцидент, а лог из Telegram и MAX превращается в официальные рабочие материалы для разбора инцидента (evidence) прямо в задаче Jira
async def stop_alarm(
    alarm_id: str,
    telegram_bot: Any,
    reply_fn: ReplyFn,
) -> bool:
    # 1. Бизнес-логика закрытия: SCM, каналы, обновление состояния и т.д.
    # ...
    # alarm = bot_state.active_alarms.pop(alarm_id, None)
    # ...
    # 2. Автоархивация диалога в Jira
    jira_key = get_jira_key_by_alarm_id(alarm_id)  # ваша реализация маппинга
    if jira_key:
        messages = get_messages(alarm_id)
        log_text = format_log_markdown(messages)
        try:
            await add_comment_to_issue(
                issue_key=jira_key,
                comment_text=log_text,
            )
        except Exception as e:
            # Не ломаем остановку инцидента из-за проблемы с Jira
            await reply_fn(
                "⚠️ Инцидент остановлен, но не удалось прикрепить диалог в Jira. "
                "Пожалуйста, проверьте интеграцию."
            )
        else:
            # После успешной архивации чистим лог
            clear_log(alarm_id)
    # 3. Ответ пользователю
    await reply_fn(f"✅ Сбой {alarm_id} остановлен.")
    return True
add_comment_to_issue — минимальная обёртка над Jira API:
# integrations/jira_client.py
from jira import JIRA  # пример: python-jira
_jira = JIRA(
    server="https://jira.example.com",
    basic_auth=("bot_user", "bot_password"),
)
async def add_comment_to_issue(issue_key: str, comment_text: str) -> None:
    # Если нет async-клиента, можно вынести в executor / отдельный сервис
    _jira.add_comment(issue_key, comment_text)

В результате, как только инцидент закрывается через CORE, диалог по нему (из TG и MAX) упаковывается в один лог, лог автоматически прикрепляется к задаче Jira. Дежурному не нужно выискивать чаты, копировать куски переписок, вспоминать, кто что предлагал. Весь контекст уже лежит в задаче, остается структурировать выводы.

Почему это хорошо по ITIL 4 / ITSM

Incident Management:

  • единый вход и единый процесс закрытия инцидента;

  • меньше потерь на коммуникацию за счёт моста «TG ↔ MAX»;

  • прозрачное завершение с зафиксированным итогом в Jira.

Major Incident & коммуникации:

  • все участники видят одну и ту же картину, независимо от клиента;

  • история реагирования не живёт «в чатах», а попадает в инцидентную задачу.

Problem Management / RCA:

  • разбор полётов опирается не на воспоминания, а на точную хронологию действий;

  • анализируются реальные шаги: какие гипотезы поднимали, в каком порядке, что сработало.

Continual Improvement:

  • логи по инцидентам становятся источником улучшений: паттерны ошибок, узкие места в коммуникации, задержки в реакции.

Итог

Мы не отключали Telegram. Вместо этого добавили MAX и связали их мостом, чтобы коллеги могли продолжать общение там, где им удобно, не теряя единого диалога по инциденту.​ Мост между TG и MAX и автоархивация диалога в Jira превратили «Дежурного» из простого бота в интерфейс к процессам ITIL 4/ITSM:

  • быстрый и стандартный старт инцидента;

  • согласованное общение при реагировании;

  • готовый материал для RCA, пост‑мортемов и улучшения процессов.

 В планах — научить бота прикреплять к задаче в Jira не только текст, но и файлы из переписки, а также добавить возможность «вынуть» конкретное сообщение из архива обратно в чат.

А как у вас организовано хранение переписок по инцидентам? Храните всё в чатах и потом героически ищете? Или тоже автоматизируете?

Буду рад пообщаться в комментариях.