Привет, я Максим Королев из Петрович-ТЕХ, занимаюсь уровнем сервиса. Моя роль — в том числе про “как хороший сервис показать пользователю, чтобы тот не отказался от нового/сложного/страшного инструмента”.
Было так: не все пользователи ходили в техническую поддержку по “правильному” маршруту, через портал. Из-за этого мы теряли случаи, составляющие общую картину проблем, и озадачились: как упростить пользователям заход в поддержку?
Решили сделать семейство Telegram-ботов, которые сократили время на создание повторяющихся задач на 50–70%, подняли заполняемость до 100%, ускорили реакцию на инциденты.
Если в новом году хотите запилить много микроавтоматизаций через боты, наш кейс к вашему столу: вот как вышло у Петрович-ТЕХа.
Персонификация: заботимся о качестве бота правильной мотивацией разработки
Первого бота зовут Славик. Когда мы начали его создавать, написание кода, проверка скриптов и работа над ошибками были абстрактными. А потом мы придумали ему лицо — портрет осьминога, эдакого маскота функции. На обложке статьи как раз он.
Получилось, что теперь код мы пишем для Славика. Если допустить ошибку — страдает он, конкретный осьминог, а не просто «бот номер один». Это изменило нашу ответственность. Код стал качественнее, тесты серьёзнее, обработка ошибок внимательнее.
Решили дать имена и остальным ботам. Когда у каждого есть история и лицо — разработка становится созданием чего-то живого. Чуть алхимии реально помогает делать лучший код.
Архитектура и общие принципы
Все наши боты построены на общих принципах, которые дают быстро развить функционал под запрос и подде��живать код в актуальном состоянии.
Слоистая архитектура
Мы используем подход, близкий к Clean Architecture, с разделением на слои:
Transport Layer (Telegram Bot API)
Обработчики команд и сообщений
FSM для управления диалогами
Валидация входных данных
Application Layer (Use Cases)
Бизнес-логика создания задач
Обработка сценариев (регистрация, создание заявки)
Координация между слоями
Domain Layer (Бизнес-сущности)
Модели данных (User, Ticket, Failure)
Бизнес-правила и валидация
Value Objects (Priority, Status)
Infrastructure Layer (Внешние интеграции)
JIRA API клиент
Хранилища данных (JSON, PostgreSQL)
Службы уведомлений
Такое разделение позволяет легко менять детали реализации без влияния на бизнес-логику.
FSM для управления диалогами
Все многошаговые сценарии (регистрация, создание заявки) реализованы через Finite State Machine (FSM) из aiogram. Это чтобы четко контролировать последовательность шагов, сохранять промежуточные данные, легко добавлять новые шаги или изменять их порядок и обрабатывать отмену на любом этапе.
Пример структуры состояний для создания заявки:
class SearchFormStates(StatesGroup):
SELECT_PROBLEMATIC_SERVICE = State() # Шаг 1: Выбор сервиса
SELECT_REQUEST_TYPE = State() # Шаг 2: Тип запроса
ENTER_CITY = State() # Шаг 3: Город
ENTER_COMMENT = State() # Шаг 4: Комментарий
UPLOAD_ATTACHMENT = State() # Шаг 5: Файл
CONFIRM_AND_CREATE = State() # Шаг 6: ПодтверждениеХранение состояния
Для небольших ботов (до 100 активных пользователей) используем JSON-файлы с thread-safe блокировками (RLock). Это просто и достаточно быстро.
Для масштабирования переходим на:
In-memory кэш с TTL для часто запрашиваемых данных
PostgreSQL для персистентного хранения
Redis для распределенных блокировок (если нужно)
Обёртка над JIRA REST API
Все боты используют единый модуль для работы с JIRA. Это даёт:
Единообразную обработку ошибок
Retry-механизм с экспоненциальной задержкой
Кэширование метаданных (типы задач, поля, transitions)
Логирование всех запросов
Пример: метод создания задачи из общего модуля JIRA. Показывает, как обрабатываем ошибки, логируем запросы и делаем retry.
@handle_jira_errors
async def create_issue(self, issue_data: dict) -> JiraIssue:
"""Создает новую задачу в JIRA с retry и логированием"""
if not self.session:
raise JiraConnectionError("Сессия не инициализирована")
# Логируем запрос для отладки
logger.info(f"Создание задачи: {issue_data.get('fields', {}).get('summary')}")
max_retries = 3
for attempt in range(max_retries):
try:
async with self.session.post(
f"{self.base_url}/issue",
json=issue_data,
timeout=self.config.REQUEST_TIMEOUT
) as response:
# Проверяем статус ответа
if response.status == 200:
data = await response.json()
logger.info(f"Задача создана: {data.get('key')}")
return data
elif response.status == 429: # Rate limit
# Экспоненциальная задержка
wait_time = (2 ** attempt) * 2
logger.warning(f"Rate limit, ждём {wait_time} сек")
await asyncio.sleep(wait_time)
continue
else:
# Другие ошибки — пробрасываем дальше
text = await response.text()
raise JiraError(f"HTTP {response.status}: {text}")
except asyncio.TimeoutError:
if attempt < max_retries - 1:
logger.warning(f"Timeout, попытка {attempt + 1}/{max_retries}")
await asyncio.sleep(2 ** attempt)
continue
raise JiraConnectionError("Превышено время ожидания ответа от JIRA")
raise JiraError("Не удалось создать задачу после всех попыток")• Декоратор @handle_jira_errors централизует обработку ошибок
• Retry только для временных ошибок (429, timeout)
• Экспоненциальная задержка не перегружает JIRA
• Логирование помогает отлаживать проблемы в production
Безопасность
Все чувствительные данные в .env файлах
Токены никогда не коммитятся в Git
Используем python-dotenv для загрузки переменных окружения
Whitelist пользователей
Система регистрации с модерацией
Роли и права доступа
Rate limiting для защиты от спама (3 команды/сек)
Все входные данные проверяются перед обработкой
Великолепная шестерка ботов
Теперь сами боты. Опишу их по схеме: задача — функция бота — из чего бот собран — пример кода — лайфхаки реализации — результаты.
"Лупа" — репортинг ошибок поиска в JIRA
Задача: сотруднику нужно быстро зафиксировать баг в поиске на сайте или приложении — не отвлекаясь от работы.
Что делает:
Интерактивная форма создания задачи — выбор сервиса, тип проблемы, город
Обязательная загрузка файла (фото/видео, до 10 МБ)
Автоматическое заполнение всех полей в JIRA
Логирование в Excel для статистики
Технически:
aiogram 3.x с FSM (Finite State Machine)
Асинхронная архитектура
Кэширование подразделений из JIRA
Валидация номеров и защита от дублей
Пример: это обработчик шага выбора сервиса в "Лупе". Показывает, как валидируем ввод, сохраняем данные и переходим к следующему шагу.
@router.callback_query(F.data.startswith("service_"))
async def process_service_selection(
callback: CallbackQuery,
state: FSMContext
):
"""Обработка выбора проблемного сервиса"""
user_id = callback.from_user.id
service = callback.data.replace("service_", "")
# Сохраняем выбранный сервис в состоянии
await state.update_data(problematic_service=service)
# Переходим к следующему шагу
await state.set_state(SearchFormStates.SELECT_REQUEST_TYPE)
# Показываем клавиатуру с типами запросов
await callback.message.edit_text(
"Выберите тип запроса:",
reply_markup=create_request_type_keyboard()
)
await callback.answer()• Используем callback_query для кнопок — это надёжнее, чем текстовый ввод
• Сохраняем данные в state перед переходом — если пользователь отменит, данные не потеряются
• Используем edit_text вместо нового сообщения — меньше спама в чате
Результаты:
30+ задач за первый месяц
Время создания задачи: 5-10 мин → 2-3 мин (-60%)
100% корректное заполнение полей
"Рафтар" — сбор статусов проектов
Задача: собрать ежемесячный статус для контроля жизненного цикла 50+ проектов.
Что делает:
Пошаговый мастер заполнения
Блокировка повторного заполнения в течение месяца
Автоматическое создание RAF-задач в JIRA
Система напоминаний: ежемесячная рассылка + ежедневные пуш-напоминания + финальное уведомление
Excel-отчеты
Технически:
Планировщик задач (scheduler) для автоматических напоминаний
Умный поиск transitions в JIRA
Система отслеживания напоминаний (JSON)
Пример: это фрагмент планировщика из "Рафтара". Показывает, как формируется задание “напоминание” и как храним состояние.
class StatusReminderScheduler:
"""Планировщик для автоматических напоминаний"""
def __init__(self, bot: Bot):
self.bot = bot
self.is_running = False
async def send_monthly_reminders(self):
"""Отправка месячных напоминаний в первый рабочий день"""
period = get_current_period() # "Октябрь 2024"
# Загружаем базу напоминаний
db = load_reminders_db()
# Получаем всех пользователей с активными проектами
all_projects = get_all_projects()
users_with_projects = set()
for project in all_projects.values():
users_with_projects.update(project.get("users", []))
sent_count = 0
for user_id in users_with_projects:
# Проверяем, не заполнены ли уже статусы
if check_status_exists(user_id, period):
continue
# Отправляем напоминание
try:
await self.bot.send_message(
user_id,
f"📅 Напоминание: заполните статусы за {period}"
)
sent_count += 1
# Сохраняем информацию о напоминании
if str(user_id) not in db["reminders"]:
db["reminders"][str(user_id)] = {}
db["reminders"][str(user_id)][period] = {
"first_reminder_date": datetime.now().isoformat(),
"days_count": 0
}
except Exception as e:
logger.error(f"Ошибка отправки напоминания {user_id}: {e}")
# Сохраняем обновлённую базу
save_reminders_db(db)
logger.info(f"Отправлено {sent_count} месячных напоминаний")
async def start(self):
"""Запуск планировщика в фоновом режиме"""
self.is_running = True
while self.is_running:
now = datetime.now()
# Проверяем, нужно ли отправить месячные напоминания
if now.day == 1 and now.hour == 10:
await self.send_monthly_reminders()
# Ежедневная проверка напоминаний
await self.check_daily_reminders()
# Ждём до следующей проверки (каждый час)
await asyncio.sleep(3600)Почему так:
• Состояние храним в JSON — просто и достаточно для этой задачи
• Проверяем заполненность статусов перед отправкой — не спамим
• Логируем все действия — можно отследить, что происходило
• Фоновая задача не блокирует основной поток бота
Результаты:
Сбор по 50+ проектам ежемесячно
100% заполняемость благодаря напоминаниям
Время на сбор отчётности: неделя → 1 день
Полная прозрачность для руководства
"Гена" — помощник WMS
Задача: сотрудник склада создает тикет, видит свои задачи и получает уведомления об изменениях — всё в Telegram.
Что делает:
Создание тикетов с пошаговым мастером
Список "Мои задачи" с пагинацией (по 10)
Эмодзи-статусы для быстрого распознавания
Просмотр всех комментариев + добавление новых
Автоматические уведомления об изменениях статуса, исполнителя, комментариях
Технически:
Кэширование запросов к JIRA (in-memory с TTL)
aiohttp для асинхронных запросов
Retry-механизм с экспоненциальной задержкой
Rate limiting (3 команды/сек, 5 кнопок/сек)
Глобальный error handler
Пример: в "Гене" мы используем кэширование для снижения нагрузки на JIRA API и ускорения ответов. Это особенно важно, когда несколько пользователей одновременно запрашивают информацию об одной и той же задаче.
"""Jira клиент с кэшированием для повышения производительности"""
def __init__(self, config: dict):
super().__init__(config)
self._cache: Dict[str, Tuple[Any, float]] = {}
self._cache_ttl = {
'issue_details': 300, # 5 минут для деталей задач
'issue_comments': 180, # 3 минуты для комментариев
'search_results': 120, # 2 минуты для результатов поиска
'connection_check': 60 # 1 минута для проверки соединения
}
def _is_cache_valid(self, cache_key: str, cache_type: str) -> bool:
"""Проверка валидности кэша"""
if cache_key not in self._cache:
return False
_, timestamp = self._cache[cache_key]
ttl = self._cache_ttl.get(cache_type, 60)
return time.time() - timestamp < ttl
def get_issue_details(self, issue_key: str) -> Dict:
"""Получение деталей задачи с кэшированием"""
cache_key = self._get_cache_key("get_issue_details", issue_key)
# Проверяем кэш
if self._is_cache_valid(cache_key, 'issue_details'):
return self._get_from_cache(cache_key)
# Получаем данные из Jira
result = super().get_issue_details(issue_key)
if result:
self._save_to_cache(cache_key, result)
logger.info(f"[CACHE] Детали задачи {issue_key} закэшированы")
return result
def create_issue(self, ...) -> Tuple[bool, str]:
"""Создание задачи (инвалидирует кэш поиска)"""
success, result = super().create_issue(...)
if success:
# Инвалидируем кэш поиска
self._invalidate_cache("search_issues")
logger.info("[CACHE] Кэш поиска инвалидирован после создания задачи")
return success, resultПочему так:
• Разные TTL для разных типов данных — детали задач кэшируем дольше, чем
результаты поиска
• Автоматическая инвалидация при изменениях — после создания задачи
очищаем кэш поиска
• Простое in-memory кэширование — достаточно для бота с сотнями пользователей
• Логирование операций с кэшем — помогает отлаживать проблемы
Результаты:
100+ тикетов создано
Фиксация ошибок здесь и сейчас
Время создания: 5 мин → 1 мин
100% уведомляется о важных изменениях
Улучшена коммуникация между складом и IT
"Дежурный" — управление событиями в каналах
Задача: дежурный администратор фиксирует аварию, указывает время восстановления — система автоматически уведомляет каналы, создает задачу в JIRA и напоминает о статусе.
Что делает:
Создание аварий с уровнями критичности
Управление регламентными работами — время начала, окончания, недоступные сервисы
Отправка обычных сообщений в каналы
Просмотр активных событий с деталями
Автоматическое создание задач в JIRA
Система напоминаний о времени восстановления
Технически:
aiogram 3.x + aiohttp для асинхронных запросов
FSM для многошаговых диалогов
JSON-хранилище состояния с thread-safe операциями (RLock)
Система ролей (admin / superadmin)
Восстановление после перезапуска
Пример: "Дежурный" помогает администраторам быстро зарегистрировать технический сбой, создавая задачу в JIRA и отправляя уведомление в Telegram-канал.
@router.callback_query(F.data == "confirm_send")
async def confirm_send_callback(callback: CallbackQuery, state: FSMContext):
"""Подтверждение отправки аварии"""
data = await state.get_data()
msg_type = data["type"]
if msg_type == "alarm":
# Создаём задачу в JIRA
try:
jira_response = create_failure_issue(
summary=data["title"],
description=data["description"],
problem_level="Потенциальная недоступность сервиса",
problem_service=data["service"],
time_start_problem=dt.now().strftime("%Y-%m-%d %H:%M"),
influence="Клиенты"
)
if jira_response and 'key' in jira_response:
alarm_id = jira_response['key']
jira_url = f"https://jira.petrovich.tech/browse/{alarm_id}"
else:
raise Exception("Не удалось получить ID задачи из JIRA")
except Exception as jira_error:
logger.error(f"Ошибка создания задачи в JIRA: {jira_error}")
alarm_id = str(uuid.uuid4())[:4] # Резервный локальный ID
jira_url = None
# Сохраняем аварию в состоянии бота
bot_state.active_alarms[alarm_id] = {
"issue": data["title"],
"fix_time": dt.fromisoformat(data["fix_time"]),
"user_id": callback.from_user.id,
"created_at": dt.now().isoformat()
}
# Отправляем в публичный канал (упрощённое сообщение)
chat_message = (
f"🚨 <b>Технический сбой</b>\n"
f"• <b>Проблема:</b> {data['title']}\n"
f"• <b>Сервис:</b> {data['service']}\n"
f"• <b>Исправим до:</b> {fix_time.strftime('%d.%m.%Y %H:%M')}\n"
f"• <i>Мы уже работаем над устранением сбоя...</i>"
)
await callback.bot.send_message(
CONFIG["TELEGRAM"]["ALARM_CHANNEL_ID"],
chat_message,
parse_mode='HTML'
)Результаты:
100% автоматизация создания задач по авариям
Время оформления события: 10 мин → 1 мин
Полная история всех сбоев для анализа
Улучшена коммуникация с пользователями через каналы
"Ревизорро" — контроль повреждений транспортных средств
Задача: логист быстро фиксирует повреждение на автомобиле партнёра — автоматически считается срок устранения.
Что делает:
Система ролей (Логист → Старший логист → Ревизор → Админ)
Строгая валидация номеров ТС (нормализация к формату А111АА111)
Блокировка дублей на одно авто
Автоматический расчёт срока по типу дефекта
Отчеты в xls с подсветкой просрочек красным
Обязательное фото при продлении срока
Технически:
JSON-хранилища с thread-safe блокировками (RLock)
Автоматические бэкапы
Пример: в "Ревизорро" логисты вводят номера транспортных средств в разных форматах. Функция нормализации приводит их к единому виду.
@router.message(DefectCreationStates.entering_vehicle)
async def enter_vehicle_number(message: Message, state: FSMContext):
"""Приём номера ТС и создание замечания"""
vehicle = (message.text or '').strip()
try:
normalized_vehicle = normalize_vehicle_number(vehicle)
except ValueError as exc:
await message.answer(f"❌ {exc}\nВведите номер в формате А111АА111:")
return
# Проверяем, нет ли уже активного замечания для этого ТС
existing_defect = get_active_defect_by_vehicle(normalized_vehicle)
if existing_defect:
await message.answer(
f"⚠️ Замечание для этого ТС уже заведено.\n"
f"ID: #{existing_defect.get('id')}"
)
return
# Сохраняем нормализованный номер
await state.update_data(vehicle_number=normalized_vehicle)
await state.set_state(DefectCreationStates.waiting_for_photos)Почему так:
• Единый формат — все номера хранятся одинаково, независимо от ввода
• Валидация на этапе ввода — пользователь сразу видит ошибку
• Защита от дубликатов — проверяем активные замечания по нормализованному номеру
• Понятные сообщения об ошибках — пользователь знает, что исправить
• Регулярное выражение — быстрая проверка формата
Результаты:
40+ замечаний за квартал
Время на оформление: 10 мин → 1 мин
Полный контроль просрочек
Прозрачность для партнёров, логистов и КРО
"График" — управление заявками в контакт-центре
Задача: сотрудник быстро создаёт заявку о технической неполадке или больничном — данные автоматически попадают в JIRA и систему контакт-центра.
Что делает:
Заявки о технических неполадках
Заявки о больничных
Валидация всех данных с понятными сообщениями об ошибках
Автоматическое создание задач в JIRA
История всех заявок
Технически:
aiogram 3.x с FSM для пошаговых диалогов
aiohttp для асинхронных запросов
JIRA REST API для создания и управления задачами
JSON-хранилище заявок
Система логирования
Пример: бот "График" помогает руководителям контакт-центра создавать заявки о технических неполадках и больничных в JIRA. Особенность — строгая валидация дат и времени.
@router.message(TechnicalIssueStates.ENTER_DATE)
async def enter_date_tech(message: Message, state: FSMContext):
"""Валидация и парсинг даты"""
if message.text == "❌ Отмена":
await state.clear()
await message.answer("❌ Создание заявки отменено", reply_markup=create_main_keyboard())
return
try:
# Парсим дату в формате dd.mm.yyyy
parsed_date = datetime.strptime(message.text, "%d.%m.%Y")
# Преобразуем в формат YYYY-MM-DD для JIRA
jira_date = parsed_date.strftime("%Y-%m-%d")
except ValueError:
await message.answer(
"❌ Неверный формат даты. Используйте формат dd.mm.yyyy (например: 17.06.2025):",
reply_markup=create_cancel_keyboard()
)
return
await state.update_data(date=jira_date)
await state.set_state(TechnicalIssueStates.ENTER_START_TIME)
await message.answer(
"Введите время начала (формат: HH:MM):",
reply_markup=create_cancel_keyboard()
)
@router.message(TechnicalIssueStates.ENTER_START_TIME)
async def enter_start_time_tech(message: Message, state: FSMContext):
"""Валидация и парсинг времени"""
if message.text == "❌ Отмена":
await state.clear()
await message.answer("❌ Создание заявки отменено", reply_markup=create_main_keyboard())
return
try:
parsed_time = datetime.strptime(message.text, "%H:%M")
jira_time = parsed_time.strftime("%H:%M")
except ValueError:
await message.answer(
"❌ Неверный формат времени. Используйте формат HH:MM (например: 14:30):",
reply_markup=create_cancel_keyboard()
)
return
Результаты:
Централизованное создание заявок
Сокращение времени на оформление с 5 мин до 1 мин
100% синхронизация с JIRA
Полная история заявок для аудита
UX/UI и интерфейс
Что сделали мы для простоты, скорости и приличного качества UX в ботах.
Кнопки навигации
Везде одинаковые кнопки:
«❌ Отмена» — возврат в главное меню с очисткой состояния
«⬅️ Назад» — возврат на предыдущий шаг
«✅ Подтвердить» — финальное подтверждение действия
Правило «5 шагов»
Если сценарий требует больше 5 шагов, разбиваем на подзадачи или используем интерактивные кнопки вместо текстового ввода. Пример: в «Л��пе» выбор сервиса и типа запроса — через кнопки, а не текстовый ввод. Это сокращает количество шагов и уменьшает ошибки.
Формат сообщений об ошибках
Всегда понятные сообщения:
Что пошло не так
Что нужно исправить
Пример правильного формата
❌ Плохо: «Ошибка валидации»
✅ Хорошо: «❌ Неверный формат даты. Используйте формат dd.mm.yyyy (например: 17.06.2025)»
Визуальная обратная связь
Эмодзи-статусы для быстрого распознавания: 🆕 New, ⚙️ In Progress, ✅ Resolved, 🔒 Closed
Прогресс-индикаторы в многошаговых формах
Подтверждение успешных действий с ссылкой на созданную задачу
Тестирование и сопровождение
Как мы тестируем
Unit-тесты бизнес-логики
Тестируем валидацию, нормализацию данных, расчет сроков — всё, что не требует внешних API. Пример: тест нормализации номера ТС в "Ревизорро":
def test_normalize_vehicle_number():
assert normalize_vehicle("а123аа777") == "А123АА777"
assert normalize_vehicle("А 123 АА 777") == "А123АА777"
assert normalize_vehicle("а-123-аа-777") == "А123АА777"Тестирование FSM-сценариев
Проверяем, что состояния переключаются правильно, данные сохраняются между шагами, отмена работает на любом этапе.
Интеграционное тестирование
Для проверки интеграций с JIRA используем sandbox-проекты:
Создаём тестовые задачи
Проверяем заполнение полей
Удаляем после теста
Как мы мониторим
Метрики, которые смотрим ежедневно:
Количество ошибок в логах
Время отклика JIRA API
Количество активных пользователей
Успешность создания задач
Время обработки запросов
Логирование — структурированное, со всеми контекстными данными. Когда бот работает с 250+ пользователями, без логов вы потеряетесь.
Правила-шпаргалка
Если бы я снова собирал ботов с нуля, пользовался бы этой шпаргалкой.
Начать с малого
Не выйдет и не нужно создавать все фичи сразу. Первая версия «Лупы» была на 70% проще. Пользователи сами подсказали, что нужно добавить.
Персонификация — это не глупо
Когда у бота есть лицо и имя, разработка становится внимательнее. Славик, Лупа, Гена, Дежурный — персонажи, за которых хочется горой встать и оперативно “лечить”.
Дизайн UX важнее, чем функциональность
Интуитивный интерфейс, понятные сообщения об ошибках, визуальная обратная связь (эмодзи, кнопки) — это главное. UX при неудачном исполнении убивает даже хороший функционал.
Валидируем всё
Номер ТС, email, табельный номер, кириллица в ФИО — всё должно быть проверено, это спасает от 80% проблем в production.
Кэшируем и оптимизируем
JIRA API может быть медленной. Кэширование с TTL, асинхронные запросы, retry-механизм — это не какая-то оптимизация преждевременная, это необходимость.
Логируем и мониторим
Когда бот работает с большим количеством пользователей, без логов теряешься. Структурированное логирование, метрики использования, мониторинг ошибок — мастхэв.
Clean Architecture масштабируется!
Разделение слоев позволило быстро добавлять фичи и менять логику без перелома кода. Это особенно полезно, когда система растёт.
Итого
Славик и его потомки привели нас к общей методологии их создания, ею и поделился. Если вы работаете с ITSM и видите, где можно сократить путь от заказчика до исполнителя — создавайте ботов, этот тренд и правда несет много пользы. Начните с одного, послушайте пользователей, затем развивайте. И дайте ему лицо! :)
