Делаем динамический кол-трекинг: Flask + SQLite + офлайн-конверсии Яндекс.Метрики / Habr

Привет, Хабр. В перформанс-маркетинге обычно обучают кампании по начальным событиям в воронке вроде заявки. Для алгоритма это считается конверсией, но для бизнеса важнее сделка. В итоге до сделки и оплаты доходит только часть лидов, но для алгоритма они одинаковые, и автостратегия продолжает искать и тех, кто не конвертируется в оплату, и тех, кто оплачивает.

Чтобы алгоритм работал лучше и искал только тех, кто вероятнее готов к сделке, между собой связываются рекламное объявление, звонок и итоговая сделка. Для этого в Яндекс через офлайн-события возвращается звонок или уже факт сделки.

В этом гайде разберём MVP на Python: он добавляет номер на лендинге под yclid, хранит выдачу в SQLite, принимает вебхук звонка от МТС Exolve, создаёт конверсию и формирует CSV под импорт в Яндекс.Метрику. Получается повторяемый поток данных от рекламного клика до офлайн-цели без ручной склейки.

В конце статьи у вас будет рабочий сценарий запуска, тестовые запросы и список технических доработок для боевого контура.

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

Внутри проекта обычный Flask-сервис, таблицы allocations и conversions, небольшой фронтовый скрипт для добавления номера и endpoint для экспорта. За счёт этого код легко проверить локально, а затем поэтапно довести до прод-уровня.

Стек: Python 3, Flask, SQLite, JavaScript, МТС Exolve webhook, CSV-экспорт.

Пайплайн начинается в браузере, когда пользователь приходит на страницу с yclid в URL. Фронт отправляет этот идентификатор в POST /api/assign_number, а бэкенд возвращает номер с учётом окна атрибуции и текущей загрузки пула. После звонка МТС Exolve присылает вебхук с номером назначения, сервис ищет последнюю релевантную выдачу и фиксирует конверсию. Финальный шаг — выгрузка CSV и импорт офлайн-конверсий в Яндекс.Метрику.

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

В проекте всего несколько файлов. config.py хранит ключевые параметры окружения, db_init.py отвечает за создание таблиц, а app.py содержит весь HTTP-контур: выдачу номера, приём вебхуков и экспорт конверсий. На стороне интерфейса templates/index.html даёт базовую страницу, а static/script.js добавляет номера и кеширует его в sessionStorage.

Фронт отправляет yclid и получает JSON с номером. Вебхук приходит с номером назначения и токеном в query-параметре. Экспорт возвращает CSV с колонками Yclid, Target, DateTime, Price, который можно забирать вручную или автоматизировать через отдельный job.

Пререквизит

Для запуска достаточно Python 3.10+ и зависимостей из requirements.txt в изолированном окружении.

Для локального теста вебхуку нужен публичный туннель к вашему серверу: облако МТС Exolve не достучится до локалхоста напрямую. Проще всего поднять такой туннель через ngrok и указать полученный URL вида https://....ngrok-free.app в кабинете МТС Exolve.

Сейчас параметры лежат в config.py. Для MVP это допустимо, но в рабочем окружении лучше сразу перейти на переменные окружения и не хранить секреты в репозитории.

Шаг 1. Готовим конфиг и таблицы

Если начать с эндпоинта и отложить схему хранения на потом, сервис быстро начинает расходиться по логике. Один обработчик пишет как получилось, второй ищет данные по другим правилам, а выгрузка не понимает, что считать новой конверсией. Здесь мы фиксируем единый контракт данных, чтобы остальные шаги опирались на стабильную структуру.

Сервис поднимает Config, где уже есть параметры для БД, вебхук и окна атрибуции номера. При старте вызывается init_db(), и в SQLite создаются две таблицы. В allocations храним связь между phone_number и yclid с временем выдачи. В conversions храним подтверждённые звонки и флаг exported, который нужен для управляемой выгрузки.

config.py
class Config:
    EXOLVE_API_KEY = "your_api_key_here"
    WEBHOOK_SECRET = "my-super-secret-token"
    ALLOCATION_MINUTES = 15
    DB_NAME = 'calltracker.db'

db_init.py
def init_db():
    conn = sqlite3.connect(Config.DB_NAME)
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS allocations (
            phone_number TEXT,
            yclid TEXT,
            allocated_at INTEGER
        )
    ''')

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS conversions (
            yclid TEXT,
            call_time INTEGER,
            exported INTEGER DEFAULT 0
        )
    ''')
   conn.close()

На входе здесь только конфигурация и путь к файлу БД, на выходе — рабочая схема хранения для выдач и конверсий. Идемпотентность обеспечивается CREATE TABLE IF NOT EXISTS: повторный старт не ломает структуру. MVP ограничен тем, что пока нет индексов и миграций, поэтому под рост нагрузки лучше добавить индексы по yclid, phone_number, времени выдачи и перейти на управляемые миграции.

Шаг 2. Выдаём номер

В кол-трекинге номер должен быть стабильным для конкретного пользователя, но общий пул должен использоваться экономно. Если всегда выдавать новый номер, пул быстро закончится. Нужен контролируемый компромисс.

/api/assign_number принимает yclid и сначала ищет недавнюю выдачу в пределах ALLOCATION_MINUTES. Если запись есть, сервис сразу возвращает тот же номер. Если нет, вызывается get_available_number(), где проверяется занятость номеров за текущее окно. При полном пуле срабатывает buy_number_from_exolve(), и сейчас это заглушка с недействующим номером. После выбора номер сохраняется в allocations, чтобы следующий запрос уже работал через готовую связь.

app.py
@app.route('/api/assign_number', methods=['POST'])
def assign_number():
    data = request.json
    yclid = data.get('yclid')
    if not yclid:
        return jsonify({"error": "Missing yclid"}), 400

    conn = sqlite3.connect(app.config['DB_NAME'])
    cursor = conn.cursor()
    threshold = int(time.time()) - (app.config['ALLOCATION_MINUTES'] * 60)

    cursor.execute("""
        SELECT phone_number FROM allocations
        WHERE yclid = ? AND allocated_at > ?
    """, (yclid, threshold))
    existing = cursor.fetchone()

    if existing:
        conn.close()
        return jsonify({"phone": existing[0]})

    phone = get_available_number()
    if not phone:
        phone = buy_number_from_exolve()

    if phone:
        cursor.execute(
            "INSERT INTO allocations (phone_number, yclid, allocated_at) VALUES (?, ?, ?)",
            (phone, yclid, int(time.time()))
        )
        conn.commit()
    conn.close()  
    return jsonify({"phone": phone}) if phone else ("No numbers", 503)

Ключевая логика ротации и автодокупки номеров вынесена в отдельные функции:

get_exolve_numbers() отвечает за пул
buy_number_from_exolve() — за fallback, когда свободных номеров нет
get_available_number() — за правило ротации

app.py
def get_exolve_numbers():
   return ["79110001122", "79110001123", "79110001124"]


def buy_number_from_exolve():
   print("⚠️ Пул исчерпан! Инициируем покупку номера (Auto-scaling)...")
   return "79998887766"


def get_available_number():
   pool = get_exolve_numbers()

   conn = sqlite3.connect(app.config['DB_NAME'])
   cursor = conn.cursor()

   threshold = int(time.time()) - (app.config['ALLOCATION_MINUTES'] * 60)
   cursor.execute("SELECT phone_number FROM allocations WHERE allocated_at > ?", (threshold,))
   busy_numbers = {row[0] for row in cursor.fetchall()}
   conn.close()

   for phone in pool:
       if phone not in busy_numbers:
           return phone

   return None

Контракт метода простой: на входе JSON с yclid, на выходе номер или ошибка 400. Главная ценность в липкости внутри сессионного окна и экономия пула через ротацию. Риск в текущем варианте — гонки при параллельных запросах, когда два запроса могут выбрать один и тот же свободный номер. Для продакшена обычно добавляют транзакционную блокировку и ограничение уникальности активной выдачи.

Шаг 3. Связываем звонок с рекламным кликом

Эндпоинт /webhook/call получает запрос от МТС Exolve и сначала проверяет токен, чтобы отсечь произвольные вызовы. Затем из JSON берётся destination и ищется последняя запись в allocations по этому номеру внутри часового окна search_window. Если связка найдена, в conversions добавляется строка с yclid и временем звонка. Если связки нет, сервис не падает, а пишет диагностическое сообщение.

app.py
@app.route('/webhook/call', methods=['POST'])
def incoming_call():
    token = request.args.get('token')
    if token != app.config['WEBHOOK_SECRET']:
        return "Forbidden", 403

    data = request.json
    target_number = data.get('destination')
    call_time = int(time.time())

    if not target_number:
        return "No destination", 400

    conn = sqlite3.connect(app.config['DB_NAME'])
    cursor = conn.cursor()

    search_window = 3600
    cursor.execute("""
        SELECT yclid FROM allocations
        WHERE phone_number = ? AND allocated_at > ?
        ORDER BY allocated_at DESC LIMIT 1
    """, (target_number, call_time - search_window))

    row = cursor.fetchone()
    if row:
        yclid = row[0]
        cursor.execute("INSERT INTO conversions (yclid, call_time) VALUES (?, ?)", (yclid, call_time))
        conn.commit()
    conn.close()  
    return "OK", 200

На входе вебхук JSON и токен, на выходе понятные HTTP-статусы: 403, 400 или 200. Метод закрывает базовую атрибуцию, но пока не защищён от дублей при повторной доставке одного и того же события. Для прода стоит хранить call_id и делать уникальный индекс по нему. Ещё один важный шаг — заменить токен в query string на проверку подписи запроса.

Шаг 4. Выгружаем офлайн-конверсии

/admin/export_csv выбирает строки из conversions, где exported = 0, формирует CSV в памяти через StringIO и отдаёт файл в HTTP-ответе. Для каждой строки сервис пишет yclid, тип цели CALL, время и стоимость. В текущей версии обновление флага выгрузки закомментировано, поэтому повторный экспорт вернёт те же данные.

app.py
@app.route('/admin/export_csv')
def export_csv():
    conn = sqlite3.connect(app.config['DB_NAME'])
    cursor = conn.cursor()
    cursor.execute("SELECT yclid, call_time FROM conversions WHERE exported = 0")
    rows = cursor.fetchall()
    conn.close() 


    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(['Yclid', 'Target', 'DateTime', 'Price'])

    for row in rows:
        yclid, timestamp = row
        writer.writerow([yclid, 'CALL', timestamp, 0])

    return output.getvalue(), 200, {
        'Content-Type': 'text/csv',
        'Content-Disposition': 'attachment; filename=calls_export.csv'
    }

Этот блок переводит внутреннюю БД в внешний контракт импорта. Потенциальная ошибка здесь чаще всего в формате DateTime: в коде используется Unix timestamp, и перед загрузкой в Яндекс.Метрику формат нужно проверить для вашего кабинета.

Яндекс.Метрика строга к формату CSV. Если файл не валидируется, проверьте три момента:

Формат даты должен быть в Unix Timestamp
Заголовки обязательны и чувствительны к регистру. Строка должна выглядеть точно так: Yclid, Target, DateTime, Price
Название цели в поле Target (в коде это CALL) должно буква в букву совпадать с идентификатором цели, которую вы создали в интерфейсе Метрики в разделе «Загрузка офлайн конверсий»

Для продакшена включите обновление exported = 1, закрыть ендпоинт авторизацией и добавить журнал выгрузок.

Шаг 5. Добавляем номер на лендинге

После загрузки страницы script.js достаёт yclid из query-параметров. Если метка есть, скрипт сначала пытается взять номер из sessionStorage, чтобы не ходить в API повторно. Если кеша нет, отправляет POST /api/assign_number, сохраняет номер и меняет в DOM текст и ссылку tel:+... у #header-phone. Если yclid отсутствует, номер остаётся дефолтным.

script.js
document.addEventListener("DOMContentLoaded", async function() {
 const urlParams = new URLSearchParams(window.location.search);
 const yclid = urlParams.get('yclid');

 function formatPhone(phoneString) {
     if (!phoneString) return phoneString;
     const clean = phoneString.replace(/\D/g, '');
     const match = clean.match(/^(\d{1})(\d{3})(\d{3})(\d{2})(\d{2})$/);
     if (match) {
         return `+${match[1]} (${match[2]}) ${match[3]}-${match[4]}-${match[5]}`;
     }
     return phoneString;
 }

 if (yclid) {
     console.log("CallTracker: Обнаружен yclid=", yclid);

     let phone = sessionStorage.getItem('tracker_phone');

     if (!phone) {
         try {
             const response = await fetch('/api/assign_number', {
                 method: 'POST',
                 headers: {'Content-Type': 'application/json'},
                 body: JSON.stringify({ yclid: yclid })
             });
             const data = await response.json();
             if (data.phone) {
                 phone = data.phone;
                 sessionStorage.setItem('tracker_phone', phone);
                 console.log("CallTracker: Выдан номер", phone);
             }
         } catch (e) {
             console.error("Calltracker error", e);
         }
     } else {
         console.log("CallTracker: Номер взят из кэша", phone);
     }

     if (phone) {
         const el = document.getElementById('header-phone');
         if (el) {
             el.innerText = formatPhone(phone);
             el.href = 'tel:+' + phone;
             el.style.color = 'green';
         }
     }
 } else {
     console.log("CallTracker: Органический трафик (нет yclid). Показываем дефолтный номер.");
 }
});

На входе фронт получает yclid и ответ API, на выходе — корректный номер в интерфейсе и кликабельный tel-линк. Идемпотентность в пределах вкладки достигается за счёт sessionStorage, поэтому лишних вызовов меньше.

В коде стоит обернуть вызов fetch в блок try/catch. Если API вернул ошибку, скрипт должен показать резервный номер компании. Атрибуции в этом случае не будет, но клиент хотя бы сможет дозвониться.

В итоге

Мы собрали минимальный рабочий контур динамического кол-трекинга: номер добавляется, звонок атрибутируется к yclid, а конверсии выгружаются в формате для Яндекс.Метрики. Такой MVP уже закрывает главный разрыв между рекламным кликом и фактом звонка. Архитектура остаётся простой, поэтому её легко дорабатывать под рост трафика и требования безопасности.

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

Возможности для развития

Подключить мониторинг по ключевым метрикам: доля неатрибутированных звонков, ошибки webhook, размер пула номеров
Сделать автоматическую покупку номеров через API МТС Exolve для автоматического расширения вместо заглушек
Сейчас номер закрепляется за пользователем на ALLOCATION_MINUTES на 15 минут, даже если вкладку закрыли через несколько секунд, из-за чего пул быстро забивается пустыми аллокациями. Подчищать такие номера фоновым cron-джобом раз в минуту
Добавить очистку allocations, определяя, когда пользователь закрыл вкладку с сайтом
Локально у пользователя хранить данные о его визите. Например, уникальный идентификатор посещения, по которому на сервере уже есть yclid, и если пользователь повторно пришел на сайт и позвонил по другому номеру телефона, все равно склеить данные

Код на гитхабе