Привет, Хабр. В перформанс-маркетинге обычно обучают кампании по начальным событиям в воронке вроде заявки. Для алгоритма это считается конверсией, но для бизнеса важнее сделка. В итоге до сделки и оплаты доходит только часть лидов, но для алгоритма они одинаковые, и автостратегия продолжает искать и тех, кто не конвертируется в оплату, и тех, кто оплачивает.
Чтобы алгоритм работал лучше и искал только тех, кто вероятнее готов к сделке, между собой связываются рекламное объявление, звонок и итоговая сделка. Для этого в Яндекс через офлайн-события возвращается звонок или уже факт сделки.
В этом г��йде разберём 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, и если пользователь повторно пришел на сайт и позвонил по другому номеру телефона, все равно склеить данные
Код на гитхабе
