Звонки в управляющую компанию различаются по срочности: один жилец хочет узнать про график капитального ремонта, другому нужен электрик — починить освещение на площадке, третий хочет сообщить о затоплении. Но когда все обращения попадают в общую очередь, повседневные вопросы начинают конкурировать со срочными заявками. Это ведет к повторным звонкам, жалобам и долгой реакции на инциденты.
Диспетчер должен быстро понимать, какую заявку поставить в очередь, а по какой — сразу соединить с аварийной службой. Но при ручной обработке линия может быть занята не срочным обращением, и в итоге аварийная заявка будет обработана с задержкой.
Привет, Хабр! В этом материале делимся подробным гайдом, как собрать голосового робота и автоматизировать первую линию диспетчерской с помощью сервиса API МТС Exolve. Так срочные звонки будут мгновенно попадать на линию, а несрочные — в очередь для обработки в рабочие часы.
Архитектура и общая схема работы
Решение состоит из четырех основных компонентов: Flask‑приложения, локальной базы данных и двух сервисов МТС Exolve — голосового робота и Telecom API. Бэкенд не управляет логикой диалога или распознаванием речи, а принимает готовый результат звонка.
Путь данных:
Сценарий начинается с голосового робота, который принимает звонок, фиксирует DTMF‑выбор и отправляют вебхук в бэкенд. В событии передаются номер телефона, категория срочности, тема и адрес. Робот передает данные через POST /webhook/call_input.
Flask‑сервис проверяет событие и создает запись в базе данных: он проверяет токен и обязательные поля в JSON — идентификатор звонка и телефон, выбирает сценарий обработки, сохраняет запись в SQLite и обновляет статус. SQLite хранит состояние сценария: карточку обращения, текущий статус, ответственный отдел и идентификатор звонка для защиты от дублей.
Срабатывает один из двух сценариев: несрочные обращения попадают в очередь диспетчера. Если заявка срочная, МТС Exolve через обратный звонок соединяет жильца с аварийной службой или отделом. Несрочные обращения попадают в отдельный поток. Сервис определяет ответственный отдел и сохраняет заявку в SQLite.
Диспетчер закрывает обработанную заявку: в панели /queue видно очередь заявок. Диспетчер видит обращение в панели управления и при необходимости сам запускает обратный звонок и закрывает задачу после обработки.
Пререквизит
Для локального запуска установите Python, зависимости из requirements.txt и.env с параметрами доступа.
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
Для локальной проверки нужны секрет вебхука, доступ к панели, параметры МТС Exolve и номера телефонов, на которые сервис будет звонить:
# Доступ к панели диспетчера
QUEUE_USERNAME=admin
QUEUE_PASSWORD=change_me
# Секрет вебхука и локальная база заявок
SECRET_KEY=change_me
WEBHOOK_SECRET=change_me
DB_NAME=uk_calls.db
# Параметры МТС Exolve для запуска колбэка
EXOLVE_API_KEY=your_api_key_here
EXOLVE_NUMBER=78005553535
EXOLVE_CALLBACK_RESOURCE_ID=123456
EXOLVE_CLIENT_AUDIO_RESOURCE_ID=
EXOLVE_MOCK=true
EXOLVE_TIMEOUT=10
# Номера, на которые сервис маршрутизирует обращения
EMERGENCY_SERVICE_PHONE=79991112233
DEFAULT_OPERATOR_PHONE=79991114455
Шаг 1. Принимаем событие от голосового робота
Входной контракт минимален. Сервису нужны call_id для связи действий в сценарии и tenant_phone для обратного звонка. Остальные поля определяют маршрут и детали обращения.
app.py @app.route("/webhook/call_input", methods=["POST"]) def handle_call_input(): """Эндпоинт приема вебхуков от голосового робота.""" token = request.args.get("token") if token != Config.WEBHOOK_SECRET: return "Unauthorized", 401 data = request.json or {} call_id = data.get("call_id") tenant_phone = data.get("tenant_phone") urgency_dtmf = data.get("urgency_dtmf") topic_dtmf = data.get("topic_dtmf") address = data.get("address", "Адрес не указан") description = data.get("description", "Описание не заполнено") if not call_id or not tenant_phone: return jsonify({"error": "Bad Request. Missing call_id or tenant_phone"}), 400
На вход приходит JSON и секрет. На выходе сервис отклоняет запрос или переходит к маршрутизации. В коде это обычные HTTP‑ответы для ошибки авторизации и неполных данных.
Шаг 2. Превращаем DTMF‑выбор в маршрут
После проверки входа сервис выбирает один из двух маршрутов: звонить сразу или положить заявку в очередь. Это снимает с диспетчера первичную сортировку, из‑за которой аварийный звонок может застрять среди обычных вопросов.
Если жилец выбирает первый пункт озвученного роботом голосового меню, запускается аварийный сценарий. Сервис сначала сохраняет аварийную заявку в SQLite с высоким приоритетом, а затем запускает колбэк на номер аварийной службы.
app.py if urgency_dtmf == "1": is_created = create_call_record( call_id=call_id, tenant_phone=tenant_phone, priority=1, status="callback_pending", topic="Авария", department="Аварийная служба", department_phone=Config.EMERGENCY_SERVICE_PHONE, address=address, description=description, ) if not is_created: return jsonify({"status": "already_processed"}), 200 success = initiate_callback(tenant_phone, Config.EMERGENCY_SERVICE_PHONE) if success: update_call_status(call_id, "callback_started") return jsonify({"status": "emergency_connected"}), 200 update_call_status(call_id, "failed") return jsonify({"status": "emergency_failed"}), 500
Для обычных обращений автоматический вызов не запускается. Сервис определяет тему по номеру нажатой кнопки. Если жилец нажал неверную кнопку или промолчал, система направит обращение в общем режиме.
app.py TOPICS = { "1": {"topic": "Подать данные счетчиков", "dept": "Бухгалтерия", "phone": "79991114455"}, "2": {"topic": "Оставить заявку на мастера", "dept": "Мастер участка", "phone": "79991114466"}, "3": {"topic": "Пожаловаться", "dept": "Отдел контроля качества", "phone": "79991114477"}, "4": {"topic": "Другое", "dept": "Общий отдел", "phone": "79991114488"}, } topic_info = TOPICS.get( str(topic_dtmf), { "topic": "Другое (не определено)", "dept": "Общий отдел", "phone": Config.DEFAULT_OPERATOR_PHONE, }, )
Выбор кнопки определяет тему и маршрут. Даже при ошибке ввода обращение попадет в очередь со статусом «Другое».
Шаг 3. Сохраняем состояние и защищаемся от дублей
Вебхуки могут приходить повторно из‑за сетевых сбоев или настроек провайдера. Если не фильтровать дубли, сервис создаст лишние заявки и запустит повторные звонки.
Идемпотентность обеспечивает call_id. Он связывает вебхук, запись в SQLite и действия диспетчера. В базе данных этот идентификатор — первичный ключ. В базе этот идентификатор сделан primary key, поэтому создать вторую запись с тем же звонком невозможно и обработчик вебхука вернет статус already_processed.
database.py Ниже показана логика вставки данных и обработки дублей. def create_call_record(call_id: str, tenant_phone: str, priority: int, status: str, **fields) -> bool: """Запись вызова. Возвращает False при повторном вебхуке с идентичным call_id.""" now = int(time.time()) conn = get_db_connection() try: with conn: conn.execute( """ INSERT INTO calls ( call_id, tenant_phone, priority, status, topic, department, department_phone, address, description, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( call_id, tenant_phone, priority, status, fields.get("topic"), fields.get("department"), fields.get("department_phone"), fields.get("address"), fields.get("description"), now, now, ), ) return True except sqlite3.IntegrityError: logger.warning(f"Повторный запрос для call_id: {call_id} отклонен базой данных.") return False finally: conn.close()
Функция сохраняет выбранный маршрут в таблицу заявок. Если звонок уже есть в базе, сервис фиксирует повторное событие.
Шаг 4. Запускаем аварийный обратный звонок
При аварии система соединяет жильца с дежурным мастером сразу после создания заявки. Срочный вызов не ждет, пока диспетчер проверит очередь вручную.
Сервис фиксирует обращение в SQLite и вызывает API МТС Exolve. Если API вернет ошибку или сеть будет недоступна, заявка останется в базе с отметкой о неудачном звонке.
Если запрос принят, сервис обновляет заявку. Если параметры не прошли проверку, API вернул ошибку или сеть упала, та же заявка помечается как неудачная.
app.py success = initiate_callback(tenant_phone, Config.EMERGENCY_SERVICE_PHONE) if success: update_call_status(call_id, "callback_started") return jsonify({"status": "emergency_connected"}), 200 update_call_status(call_id, "failed") return jsonify({"status": "emergency_failed"}), 500 Ответ Exolve сразу превращается в состояние заявки. Дальше диспетчер увидит именно этот статус в очереди. database.py def update_call_status(call_id: str, status: str) -> None: """Принудительно меняет статус звонка.""" now = int(time.time()) with get_db_connection() as conn: conn.execute( "UPDATE calls SET status = ?, updated_at = ? WHERE call_id = ?", (status, now, call_id), )
Шаг 5. Собираем запрос для МТС Exolve
Обратный звонок соединяет жильца с сотрудником управляющей компании. Платформа МТС Exolve звонит обоим абонентам и устанавливает соединение, когда каждый поднял трубку. Мы вынесли формирование запроса в отдельный модуль: это отделяет логику API от бизнес‑сценария.
exolve_api.py В запросе есть две стороны звонка: line_1 для жильца и line_2 для управляющей компании. В поле display_number указываем номер МТС Exolve, чтобы обе стороны видели его как входящий. def build_make_callback_payload( tenant_phone: str, destination_phone: str, ) -> Dict[str, Any]: """Формирует спецификацию Callback-вызова под стандарт API МТС Exolve.""" exolve_number = _normalize_phone(Config.EXOLVE_NUMBER, "EXOLVE_NUMBER") tenant_phone = _normalize_phone(tenant_phone, "tenant_phone") destination_phone = _normalize_phone(destination_phone, "destination_phone") payload: Dict[str, Any] = { "number_code": _to_int(exolve_number, "EXOLVE_NUMBER"), "callback_resource_id": _to_int(Config.EXOLVE_CALLBACK_RESOURCE_ID, "EXOLVE_CALLBACK_RESOURCE_ID"), "line_1": {"destinations": [{"number": tenant_phone}], "display_number": exolve_number}, "line_2": {"destinations": [{"number": destination_phone}], "display_number": exolve_number}, } return payload
После сборки запроса клиент отправляет HTTP‑запрос в МТС Exolve. Обработчик получает бинарный результат: запрос прошел или нет. Если параметры некорректны или сеть недоступна, сервис переводит заявку в статус failed, а причину ошибки сохраняет в логах.
Шаг 6. Сохраняем несрочные обращения в очередь
Несрочный вопрос не требует мгновенного соединения. Но его все равно нужно зафиксировать: кто звонил, по какому адресу, с какой темой и в какой отдел направить обращение.
После выбора темы сервис кладет обращение в рабочую очередь. Автоматический колбэк здесь не вызывается.
app.py is_created = create_call_record( call_id=call_id, tenant_phone=tenant_phone, priority=2, status="queued", topic=topic_info["topic"], department=topic_info["dept"], department_phone=topic_info["phone"], address=address, description=description, ) if is_created: logger.info(f"[{call_id}] Заявка '{topic_info['topic']}' сохранена в очередь.") return jsonify({"status": "added_to_queue"}), 200 return jsonify({"status": "already_processed"}), 200
Диспетчер видит активные заявки в панели управления. Приложение запрашивает данные напрямую из SQLite, которые еще требуют действий. В общем списке аварийные обращения стоят выше обычных.
database.py def get_active_calls() -> List[Dict[str, Any]]: """Возвращает список активных необработанных вызовов в очереди.""" conn = get_db_connection() try: rows = conn.execute( """ SELECT * FROM calls WHERE status IN ('callback_pending', 'queued', 'callback_started', 'failed') ORDER BY priority ASC, created_at DESC """ ).fetchall() return [dict(row) for row in rows] finally: conn.close()
Функция возвращает список для интерфейса на основе данных из базы.
Шаг 7. Даем диспетчеру ручной запуск колбэка
Для обычных обращений диспетчер смотрит карточку и сам решает, когда связаться с жильцом. Это позволяет сначала изучить детали заявки и описание проблемы.
Панель /queue защищена Basic Auth. Она показывает активные заявки и архив. Кнопка «Позвонить» остается только там, где повторный колбэк допустим: для заявок, которые ждут обработки или обработались с ошибкой при запуске колбэка.
app.py @app.route("/queue", methods=["GET"]) @require_queue_auth def view_queue(): """Считывает очереди из БД и рендерит интерфейс.""" active_calls = get_active_calls() archived_calls = get_archived_calls() return render_template_string( HTML_TEMPLATE, active_calls=active_calls, archived_calls=archived_calls, datetime=datetime, )
Главное действие диспетчера — запустить колбэк из уже сохраненной заявки. Когда диспетчер нажимает кнопку, сервис повторно читает данные заявки из базы. Это исключает запуск звонка по некорректным номерам или для уже обработанных записей.
app.py @app.route("/queue/call/<call_id>", methods=["POST"]) @require_queue_auth def trigger_manual_callback(call_id: str): """Маршрут кнопки 'Позвонить' — инициирует Callback и возвращает на страницу очереди.""" conn = get_db_connection() try: row = conn.execute("SELECT * FROM calls WHERE call_id = ?", (call_id,)).fetchone() if not row: return "Заявка не найдена", 404 call = dict(row) with conn: reserved = conn.execute( """ UPDATE calls SET status = 'callback_pending', updated_at = CAST(strftime('%s', 'now') AS INTEGER) WHERE call_id = ? AND status IN ('queued', 'failed') AND COALESCE(department_phone, '') <> '' """, (call_id,), ).rowcount if reserved != 1: logger.warning( f"Callback для заявки {call_id} не запущен: заявка уже обрабатывается, " "закрыта, не найдена или в ней не указан телефон отдела." ) return redirect(url_for("view_queue")) success = initiate_callback(call["tenant_phone"], call["department_phone"]) if success: update_call_status(call_id, "callback_started") logger.info(f"Диспетчер запустил Callback для: {call_id}") else: update_call_status(call_id, "failed") logger.error(f"Не удалось запустить Callback для: {call_id}") finally: conn.close() return redirect(url_for("view_queue"))
Обработчик проверяет четыре сценария:
заявка отсутствует в базе;
текущий статус не позволяет совершить вызов;
в профиле отдела не указан номер телефона;
API МТС Exolve вернул ошибку или запрос завершился по таймауту.
Шаг 8. Закрываем заявку
Диспетчер переводит обработанное обращение в архив. Это действие убирает заявку из списка активных задач и фиксирует завершение работы в базе данных.
app.py @app.route("/queue/complete/<call_id>", methods=["POST"]) @require_queue_auth def mark_as_completed(call_id: str): """Маршрут кнопки 'Выполнено' — закрывает заявку.""" update_call_status(call_id, "done") logger.info(f"Заявка {call_id} закрыта.") return redirect(url_for("view_queue"))
Сервис запрашивает выполненные заявки отдельным методом. В архив попадают последние 50 записей со статусом done, отсортированные по времени обновления.
database.py def get_archived_calls() -> List[Dict[str, Any]]: """Возвращает закрытые (выполненные) заявки для вывода в архив.""" conn = get_db_connection() try: rows = conn.execute( "SELECT * FROM calls WHERE status = 'done' ORDER BY updated_at DESC LIMIT 50" ).fetchall() return [dict(row) for row in rows] finally: conn.close()
Запуск и проверка
При запуске сервиса приложение само создает SQLite‑файл и таблицу заявок, если их еще нет.
python app.py
Для локальной проверки удобнее оставить режим эмуляции. Тогда аварийная ветка пройдет без реального звонка, а заявка получит отметку о запуске колбэка.
Отправим тестовый аварийный вебхук:
curl -X POST "http://localhost:5000/webhook/call\_input?token=change\_me" \
-H "Content-Type: application/json" \
-d '{
"call_id": "call_1001",
"tenant_phone": "+7 999 000-11-22",
"urgency_dtmf": "1",
"address": "Ленина, 10, кв. 15",
"description": "Прорвало трубу в ванной"
}'
Ожидаемый ответ в режиме эмуляции:
{"status":"emergency_connected"}
Для обычной заявки выберите неаварийный вариант и тему обращения. В JSON это поля urgency_dtmf и topic_dtmf. Сервис вернет {"status":"added_to_queue"}, а запись появится в /queue как заявка в ожидании обработки.
После этого откройте /queue и войдите с QUEUE_USERNAME/QUEUE_PASSWORD. Аварийная заявка после mock‑колбэка будет видна как callback_started. Если отправить тот же call_id повторно, сервис не создаст вторую заявку и вернет already_processed: запись с таким звонком уже есть в SQLite.
Что в итоге
Теперь звонок жильца автоматически становится электронной заявкой — и за счет автоматизации снижается нагрузка на диспетчера. Робот принимает вызов и определяет тему, а сервис управляет сценарием обработки. Аварийные заявки мгновенно запускают обратный звонок, а несрочные вопросы попадают в очередь для обработки позднее.
Как еще можно улучшить решение:
Настроить сбор адреса и описания через робота для еще большей автоматизации.
Вынести справочник маршрутов и отделов из кода в базу данных.
Добавить SLA и эскалацию для контроля времени реакции по авариям.
Интегрировать сервис с CRM или системой управления заявками УК.
Внедрить отчеты по типам звонков и нагрузке на подразделения.
Добавить очередь задач, подпись вебхуков и перейти на PostgreSQL для повышения надежности.
Пишите в комментариях, какой из сценариев выше вы хотели бы, чтобы мы разобрали в следующей части.