Привет, Хабр!

Передача показаний счётчиков во многих домах всё ещё требует ручных действий. Жильцы откладывают эту задачу, пропускают сроки и получают пени. Чтобы повысить дисциплину, мы перенесли механику стриков, знакомую по одному приложению для изучения языков с совой  — последовательные своевременные сдачи показаний образуют серию и небольшую скидку как вознаграждение, которая мотивирует поддерживать регулярность.

В статье разберём Python-сервис, который отправляет персонализированные SMS, принимает ответы и подключает голосовой вызов для тех, кто не ответил.

Общая схема работы

Сервис запускается каждый месяц в определённые даты. Сначала оркестратор отправляет персонализированные SMS с учётом стрика жильца. Когда пользователь отвечает, получаем SMS с ответом и приложение на Flask извлекает из него показания, проверяет их корректность и обновляет запись в базе.

Тем, кто не ответил, уходит напоминание по SMS, а в последний день выполняется голосовой вызов с индивидуальным Text-to-Speech-сообщением для повышения конверсии. Такой каскад позволяет автоматически собрать показания, обновить стрики и убрать ручную обработку со стороны управляющей компании.

Архитектура решения

Сервис разделён на три компонента, которые обрабатывают разные этапы процесса: отправку уведомлений, приём ответов и хранение данных. Такой разнос упрощает масштабирование и позволяет независимо развивать каждую часть.

Планировщик кампаний

Работает по расписанию и управляет каскадом коммуникаций. Он:

• определяет текущий этап кампании;

• анализирует состояние жильцов и их стрики в базе;

• формирует и отправляет персонализированные SMS;

• инициирует голосовой TTS-вызов на финальном этапе для тех, кто не ответил.

Вебхук-сервер

HTTP-эндпоинт на Flask, который отвечает за сбор показаний. Когда жильцы отвечают на SMS, МТС Exolve пересылает сообщения на наш вебхук. Сервер извлекает показания из текста, проверяет корректность значений и обновляет запись жильца. В ответ отправляется либо подтверждение, либо понятное сообщение об ошибке.

Хранилище данных

SQLite-база с профилями жильцов, последними показаниями, текущим статусом месяца и значением стрика.

Шаг 1. Подготовка окружения и базы данных

Сначала настраиваем окружение: устанавливаем необходимые пакеты requests, flask, python-dotenv, schedule и подключаем переменные окружения для работы с API МТС Exolve. Здесь же включаем базовое логирование, чтобы фиксировать отправку сообщений и обработку ответов.

Код settings.py ▾

# settings.py
import logging
import os
from dotenv import load_dotenv


# Загружаем переменные из .env файла
load_dotenv()


# Настройки API Exolve
EXOLVE_API_KEY = os.getenv("EXOLVE_API_KEY")
SENDER_NUMBER = os.getenv("EXOLVE_PHONE_NUMBER") # Арендованный номер, с которого будут уходить SMS и звонки
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "change_me_in_prod") # Токен для защиты нашего вебхука


# Настройка логирования
logging.basicConfig(
   level=logging.INFO,
   format='%(asctime)s [%(levelname)s] %(message)s',
   handlers=[
       logging.FileHandler("housing_bot.log"), # Пишем в файл
       logging.StreamHandler()                 # И в консоль
   ]
)
logger = logging.getLogger("HousingBot")

Далее создаём минимальную схему хранилища на SQLite. Таблица жильцов содержит основные поля, необходимые для логики кампании:

• streak_count — число последовательных месяцев, в которые показания были сданы вовремя; центральная метрика геймификации.

• last_electricity, last_water — предыдущие показания; используются для валидации, чтобы исключить уменьшение значений.

• current_status — состояние на текущий месяц: pending — ожидаем, submitted — данные получены.

Такой схемы достаточно для запуска сервиса; позже её можно перенести на PostgreSQL без изменения остального кода.

Код db.py ▾

# db.py
import sqlite3
from settings import logger


DB_NAME = 'housing.db'


def init_db():
   """Инициализация структуры БД при первом запуске."""
   conn = sqlite3.connect(DB_NAME)
   cursor = conn.cursor()
   cursor.execute('''
       CREATE TABLE IF NOT EXISTS residents (
           phone TEXT PRIMARY KEY,
           name TEXT,
           flat_number TEXT,
           current_status TEXT DEFAULT 'pending',
           streak_count INTEGER DEFAULT 0,
           has_discount BOOLEAN DEFAULT 0,
           last_electricity INTEGER DEFAULT 0,
           last_water INTEGER DEFAULT 0
       )
   ''')
   conn.commit()
   conn.close()
   logger.info("База данных проверена и готова к работе.")

Шаг 2. Отправка SMS

Планировщик находит всех жильцов со статусом pending, определяет их сегмент и формирует персонализированные SMS для каждого жильца. Шаблон сообщения выбирается по значению стрика и текущему статусу: 

1. Сегмент «Чемпионы» (streak >= 3). Это самые ответственные жильцы. Мы подкрепляем их поведение наградой.
   Текст: "{Имя}, вы супер! Серия {N} мес. Сдайте показания (Свет+Вода) сегодня, и скидка 2% ваша!"

2. Сегмент «Начинающие» (streak > 0). Они в процессе формирования привычки. Используем психологию потери прогресса.
   Текст: "{Имя}, не сбавляйте темп! Серия {N} мес. Ждем показания (Свет+Вода), чтобы продолжить серию."

3. Сегмент «Молчуны» (streak = 0). Им нечего терять, поэтому нужна просто четкая инструкция.
   Текст: "{Имя}, прием показаний открыт. Чтобы начать серию побед и получить скидку, пришлите ответ: Свет+Вода (например: 320+15)."

Затем он отправляет сообщение через метод Messaging/v1/SendSMS МТС Exolve и сразу обновляет статус в базе, чтобы сообщение не ушло повторно при перезапуске сервиса.

# sender.py
import requests
import sqlite3
from settings import EXOLVE_API_KEY, SENDER_NUMBER, logger, DB_NAME




def send_sms_via_exolve(phone, text):
   """Обертка для отправки SMS через API."""
   url = "https://api.exolve.ru/messaging/v1/SendSMS"
   headers = {"Authorization": f"Bearer {EXOLVE_API_KEY}"}
   payload = {
       "number": SENDER_NUMBER,
       "destination": phone,
       "text": text
   }
   try:
       resp = requests.post(url, headers=headers, json=payload, timeout=10)
       resp.raise_for_status()
       logger.info(f"SMS успешно отправлено на {phone}")
       return True
   except Exception as e:
       logger.error(f"Ошибка отправки SMS на {phone}: {e}")
       return False




def run_day_1_campaign():
   """Основная рассылка первого дня."""
   conn = sqlite3.connect(DB_NAME)
   cursor = conn.cursor()
   # Выбираем всех, кто еще не сдал
   cursor.execute("SELECT phone, name, streak_count FROM residents WHERE current_status='pending'")


   count = 0
   for row in cursor.fetchall():
       phone, name, streak = row


       # Динамический контент
       if streak >= 3:
           msg = f"{name}, вы супер! Серия {streak} мес. Сдайте показания (Свет+Вода) сегодня, и скидка 2% ваша!"
       elif streak > 0:
           msg = f"{name}, не сбавляйте темп! Серия {streak} мес. Ждем показания (Свет+Вода), чтобы продолжить серию."
       else:
           msg = f"{name}, прием показаний открыт. Чтобы начать серию побед и получить скидку, пришлите ответ: Свет+Вода (например: 320+15)."


       if send_sms_via_exolve(phone, msg):
           # Меняем статус на 'waiting', чтобы не отправлять повторно при перезапуске
           cursor.execute("UPDATE residents SET current_status='waiting' WHERE phone=?", (phone,))
           count += 1


   conn.commit()
   conn.close()
   logger.info(f"Кампания День 1 завершена. Отправлено сообщений: {count}.")

Шаг 3. Приём данных и разбор ответов регулярками

Когда жилец отвечает на SMS, МТС Exolve принимает входящее сообщение и отправляет его на наш сервер по вебхуку. Приложение на Flask получает HTTP-запрос, извлекает текст ответа и передаёт его в функцию парсинга.

Мы используем регулярные выражения, чтобы корректно обрабатывать разные варианты ввода: от «320 15» до «свет 320, вода 15». Функция выделяет два числовых значения и приводит их к форме, удобной для последующей проверки.

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

Если данные не проходят проверку, сервер отправляет уточняющее SMS; если всё корректно, обновляет запись в базе, увеличивает стрик и отправляет подтверждение.

# parser.py
import re




def parse_readings(text):
   """
   Извлекает два целых числа из произвольного текста.
   Поддерживает форматы с пробелами, запятыми, плюсами и словами.
   """
   if not text: return None


   # 1. Оставляем только цифры и возможные разделители
   # Заменяем все буквы и спецсимволы на пробелы
   clean_text = re.sub(r'\D+', ' ', text)
   numbers = [int(n) for n in clean_text.split() if n.isdigit()]


   # Мы ожидаем ровно два показания (свет и вода)
   if len(numbers) == 2:
       return numbers[0], numbers[1]


   return None

Для обработки входящих сообщений используется вебхук на Flask. Платформа МТС Exolve принимает ответное SMS и отправляет его на наш сервер POST-запросом. В обработчике incoming_sms выполняется прикладная логика:

• Идентификация. По номеру отправителя ищем запись в базе. Если номер отсутствует, фиксируем событие и завершаем обработку.

• Парсинг. Передаём текст сообщения в функцию на регулярных выражениях, которая извлекает два числовых значения. Если извлечь данные не удалось, отправляем ответ с просьбой прислать показания в понятном формате.

• Проверка данных. Сравниваем новые значения с предыдущими. Показания не должны уменьшаться, поэтому при несоответствии отправляем сообщение с просьбой перепроверить цифры.

• Обновление записи. При корректных данных сохраняем новые показания, увеличиваем стрик и отправляем подтверждение.

Для обработки входящих сообщений используется вебхук на Flask. Платформа МТС Exolve принимает ответное SMS и отправляет его на наш сервер POST-запросом. В обработчике incoming_sms выполняется прикладная логика:

• Идентификация. По номеру отправителя sender ищем жильца в таблице residents по полю phone. Если номер отсутствует, логируем событие и завершаем обработку.

• Парсинг. Передаём текст сообщения text в функцию parse_readings, которая с помощью регулярных выражений пытается извлечь два целых числа. Если получить значения не удалось, отправляем ответ с просьбой прислать показания в формате «Свет+Вода».

• Проверка данных. Сравниваем новые показания el_new и wat_new с предыдущими значениями last_electricity и last_water. Показания не должны уменьшаться, поэтому при несоответствии отправляем сообщение с просьбой перепроверить цифры.

• Обновление записи. Если данные корректны, обновляем поля last_electricity и last_water, увеличиваем streak_count на один и переводим current_status в submitted, после чего отправляем подтверждение жильцу.

# server.py
from flask import Flask, request, jsonify
from parser import parse_readings
from sender import send_sms_via_exolve
from settings import logger, DB_NAME
import sqlite3


app = Flask(__name__)




@app.route('/sms-webhook', methods=['POST'])
def incoming_sms():
   data = request.json
   sender = data.get('sender')
   text = data.get('text', '')


   if not sender:
       return jsonify({"status": "ignored"}), 400


   logger.info(f"Входящее SMS от {sender}: {text}")


   # 1. Ищем жильца в базе
   conn = sqlite3.connect(DB_NAME)
   cursor = conn.cursor()
   cursor.execute("SELECT last_electricity, last_water, streak_count FROM residents WHERE phone=?", (sender,))
   resident = cursor.fetchone()  # Возвращает кортеж полей


   if not resident:
       logger.warning(f"Неизвестный номер: {sender}")
       conn.close()
       return jsonify({"status": "unknown_user"}), 200


   # 2. Парсим данные
   readings = parse_readings(text)


   if not readings:
       send_sms_via_exolve(sender, "Не удалось разобрать цифры. Пришлите в формате: Свет+Вода (например: 520+15)")
       conn.close()
       return jsonify({"status": "parse_error"}), 200


   el_new, wat_new = readings
   # resident[6] - last_electricity, resident[7] - last_water
   el_old, wat_old = resident[6], resident[7]


   # 3. Валидация (защита от опечаток "в меньшую сторону")
   # Примечание: в реальном проекте здесь стоит обработать кейс замены счетчика (когда отсчет с 0)
   if el_new < el_old or wat_new < wat_old:
       send_sms_via_exolve(sender, f"Ошибка! Новые показания меньше старых ({el_old} и {wat_old}). Проверьте цифры.")
   else:
       # 4. Успех: обновляем данные и начисляем награды
       streak = resident[4]
       new_streak = streak + 1
       
       # Скидка дается только если стрик 3 и более месяцев
       discount_val = 1 if new_streak >= 3 else 0


       cursor.execute("""
           UPDATE residents 
           SET streak_count = ?, current_status = 'submitted', has_discount = ?,
               last_electricity = ?, last_water = ?
           WHERE phone = ?
       """, (new_streak, discount_val, el_new, wat_new, sender))
       conn.commit()


       send_sms_via_exolve(sender,
                           f"Принято! Свет: {el_new}, Вода: {wat_new}. Ваш стрик: {new_streak} мес. Скидка начислена!")
       logger.info(f"Жилец {sender} успешно сдал показания.")


   conn.close()
   return jsonify({"status": "ok"}), 200

Шаг 4. Звонок роботом

На третий день планировщик формирует список жильцов, у которых current_status != 'submitted', и звонит им. Для каждого абонента собирается персонализированное TTS-сообщение: в текст подставляются имя name и номер квартиры flat_number.

Вызов выполняется через метод MakeVoiceMessage Voice API МТС Exolve. В запрос передаются параметры destination (номер жильца), source (арендованный номер) и блок tts с полями text, voice и emotion. Это позволяет динамически генерировать речь без использования заранее записанных аудио.

После отправки вызова сервис фиксирует событие в логах и обновляет запись жильца — в примере предусмотрен сброс streak_count для тех, кто не ответил ни на одно уведомление в течение кампании. Это завершает третий день и закрывает цикл месяца.

# voice.py
import requests
from settings import EXOLVE_API_KEY, SENDER_NUMBER, logger, DB_NAME
import sqlite3




def make_robot_call(phone, name, flat):
   url = "https://api.exolve.ru/call/v1/MakeVoiceMessage"
   headers = {"Authorization": f"Bearer {EXOLVE_API_KEY}"}


   # Динамическая персонализация: подставляем имя и квартиру
   tts_text = (
       f"Здравствуйте, {name}. Управляющая компания беспокоит. "
       f"По квартире номер {flat} не сданы показания счетчиков. "
       f"Прием закрывается сегодня. Пожалуйста, отправьте данные в СМС. Спасибо."
   )


   payload = {
       "source": SENDER_NUMBER,
       "destination": phone,
       "tts": {
           "text": tts_text,
           "voice": "alena",  # Женский голос
           "emotion": "strict"  # Строгая интонация для важности
       }
   }


   try:
       requests.post(url, headers=headers, json=payload, timeout=10)
       logger.info(f"Звонок заказан для {phone} (кв. {flat})")
   except Exception as e:
       logger.error(f"Ошибка звонка на {phone}: {e}")




def run_day_3_campaign():
   """Финальный обзвон должников."""
   conn = sqlite3.connect(DB_NAME)
   cursor = conn.cursor()


   # Выбираем всех, кто так и не сдал (статус не 'submitted')
   # Достаем номер квартиры для персонализации
   cursor.execute("SELECT phone, name, flat_number FROM residents WHERE current_status != 'submitted'")
   debtors = cursor.fetchall()


   logger.info(f"Запуск обзвона. Должников в списке: {len(debtors)}")


   for phone, name, flat in debtors:
       make_robot_call(phone, name, flat)
       # Штрафная санкция: сбрасываем стрик в ноль :(
       cursor.execute("UPDATE residents SET streak_count = 0 WHERE phone=?", (phone,))


   conn.commit()
   conn.close()

Сборка сервиса основной цикл

Чтобы не разворачивать отдельные процессы, веб-сервер и планировщик запускаем из одной точки входа main.py. Flask-приложение поднимается в отдельном потоке, а основной поток отвечает за выполнение задач по расписанию.

Функция run_web_server() стартует приложение app на Flask и принимает входящие вебхуки. Мы запускаем её в демоне threading.Thread, чтобы веб-сервер работал параллельно и не блокировал основной цикл. В функции run_scheduler() настраиваем расписание с помощью schedule: задаём время для рассылки SMS (run_day_1_campaign) и для обзвона (run_day_3_campaign), после чего в бесконечном цикле вызываем schedule.run_pending() с паузой в time.sleep(60).

В блоке if name == "__main__": инициализируем базу данных через init_db(), поднимаем поток с веб-сервером и запускаем планировщик в основном потоке. Такой вариант позволяет локально запустить весь сервис одной командой и при этом не смешивать обработку HTTP-запросов с задачами по расписанию.

# main.py
import threading
import time
import schedule
from server import app
from sender import run_day_1_campaign
from voice import run_day_3_campaign
from db import init_db
from settings import logger




def run_web_server():
   # Запускаем Flask на всех интерфейсах
   app.run(host='0.0.0.0', port=5000, use_reloader=False)




def run_scheduler():
   logger.info("Планировщик запущен. Ожидание задач...")


   # Настраиваем расписание
   # В реальной жизни здесь стоит добавить проверку current_date.day == 20
   schedule.every().day.at("10:00").do(run_day_1_campaign)
   schedule.every().day.at("18:00").do(run_day_3_campaign)


   while True:
       schedule.run_pending()
       time.sleep(60)




if __name__ == "__main__":
   # 1. Инициализируем БД при старте
   init_db()


   # 2. Запускаем веб-сервер для вебхуков в фоне
   server_thread = threading.Thread(target=run_web_server)
   server_thread.daemon = True  # Поток закроется вместе с основным скриптом
   server_thread.start()
   logger.info("Веб-сервер запущен в фоновом режиме.")


   # 3. Запускаем планировщик в главном потоке
   try:
       run_scheduler()
   except KeyboardInterrupt:
       logger.info("Остановка сервиса...")

Демонстрация

Ниже — два сценария, которые показывают, как сервис работает с точки зрения жильца. 

Помните, что в SMS один сегмент включает до 70 кириллических символов. Если превысить этот предел даже на один символ, сообщение будет стоить как два. Поэтому шаблоны стоит делать максимально компактными.

Сценарий 1. Ответственный Иван

1. День 1, 10:00: Иван получает SMS: "Иван, стрик 3 мес! Сдайте Свет+Воду сегодня — скидка 2% ваша!"

2. День 1, 10:05: Иван пишет в ответ коротко: "3450 120".

3. День 1, 10:06: Ему приходит мгновенный ответ: "Принято! Свет:3450, Вода:120. Стрик:4. Скидка есть!".
   Итог: УК получила данные автоматически за 6 минут, потратив минимум на SMS.

Сценарий 2. Забывчивый Петр

1. День 1: SMS от УК: "Петр, начните серию побед! Сдайте показания. Ответ: Свет+Вода". (Петр на совещании, смахнул уведомление).

2. День 3, 18:00: Звонок. Строгий голос робота Алены: "Здравствуйте, Петр. Управляющая компания беспокоит. По квартире номер 45 не сданы показания... Прием закрывается сегодня".

3. День 3, 18:02: Петр понимает, что тянуть некуда, и быстро отправляет SMS.
   Итог: Показания получены до закрытия реестра, кассового разрыва нет.

Заключение

По данным Tinkoff Data, 37% жителей периодически допускают просрочки по ЖКУ, и в 47% случаев причина проста — забычивость. Для УК это выливается в кассовые разрывы: платить ресурсоснабжающей организации нужно по графику, даже если часть квартир задерживает данные. При накоплении долга появляются риски претензий, корректировок задним числом и падения качества обслуживания.

Механика стриков и простые напоминания закрывают именно эту случайную часть просрочек. Это не борьба с хроническими неплательщиками, а способ стабилизировать поток данных и сократить разрывы, возникающие из-за забывания.

Стоимость SMS довольна высокая, поэтому подключать сервис разумно точечно — для тех, у кого в прошлом были задержки. Если механика сработала и пользователь начал сдавать показания регулярно, можно переводить его в более дешёвые каналы: Telegram-бот, короткая ссылка на веб-форму, почта.

Так УК получает предсказуемые даты сдачи показаний, меньшие кассовые разрывы и меньше ручной обработки, а жители — более удобный и понятный процесс.

Полный код проекта доступен в репозитории на GitHub.