Привет. Сегодня мы создадим полноценный инструмент повышения конверсии в звонок для email- и CRM-маркетологов. Речь пойдет о системе заказа обратного звонка прямо из письма.
Пользователь получает письмо с кликабельным календарем для выбора даты. Нажав на подходящий день, мы самостоятельно определяем время звонка и автоматически ставим задачу менеджеру в amoCRM.
Архитектура решения
Система состоит из следующих компонентов:
Персонализированный email-шаблон с интерактивным календарем, где каждая дата — это уникальная ссылка для заказа звонка.
Бэкенд-сервис (API-endpoint), который принимает запросы из письма, обрабатывает их и является ядром всей логики. Пишем его на Python с Flask.
Идентификация клиента из письма через GET-параметры в ссылке на дате.
Сервис «Умная проверка номера» МТС Exolve, который по телефону абонента проверяет его доступность и возвращает оптимальное время для звонка.
Взаимодействие с amoCRM — конечная точка, где мы ищем контакт по почтовому адресу клиента и ставим задачу менеджеру на звонок в рекомендованное время.
Теперь разберем по шагам.
Шаг 1. Точка входа: интерактивный календарь в письме
Все начинается с письма. Нам нужно, чтобы ссылка не просто вела на сайт, а несла в себе информацию о клиенте: его email и выборе даты звонка. Идеальный способ — передать данные через GET-параметры. Чтобы пользователь мог выбрать дату, мы сверстаем простой календарь HTML-таблицей. Это самый надежный способ для корректного отображения в 99% почтовых клиентов.
Переменная {{EMAIL}} в ссылке — это специальный маркер, который большинство сервисов email-рассылок (ESP) автоматически заменят на реальный адрес подписчика.
from datetime import datetime, timedelta def generate_callback_calendar_html(days_ahead=7): """Генерирует HTML-код календаря для вставки в email.""" base_url = "https://api.yourdomain.com/v1/callback-request" # Замените на ваш URL html = '<table border="1" cellpadding="10" style="border-collapse: collapse; text-align: center;"><tr>' today = datetime.today() for i in range(days_ahead): current_date = today + timedelta(days=i) date_str = current_date.strftime('%Y-%m-%d') day_str = current_date.strftime('%d.%m') link = f"{base_url}?email={{EMAIL}}&date={date_str}" html += f'<td><a href="{link}" style="text-decoration: none; color: #007bff;">{day_str}</a></td>' html += '</tr></table>' return html
Полученный HTML-код вы вставляете в свой email-шаблон. Пользователь увидит в письме аккуратную таблицу с датами, каждая из которых является уникальной ссылкой:
Шаг 2. Бэкенд: интеграция с amoCRM
Когда пользователь нажимает на дату, его запрос летит на наш сервер. Специальный класс-клиент управляет всей логикой работы с API amoCRM, включая процесс OAuth 2.0 аутентификации. Критически важный момент для production-систем - обработка токенов должна быть потокобезопасной. Чтобы избежать "гонки состояний", когда два одновременных запроса пытаются обновить истекший токен, мы используем механизм блокировки файла (.lock), который гарантирует, что только один процесс в один момент времени работает с файлом токенов.
import requests import json import time import os from dotenv import load_dotenv load_dotenv() # --- Конфигурация --- TOKEN_FILE = 'tokens.json' LOCK_FILE = 'tokens.json.lock' # Файл для механизма блокировки CLIENT_ID = os.getenv("AMOCRM_CLIENT_ID") CLIENT_SECRET = os.getenv("AMOCRM_CLIENT_SECRET") REDIRECT_URI = os.getenv("AMOCRM_REDIRECT_URI") BASE_URL = os.getenv("AMOCRM_BASE_URL") AUTH_CODE = os.getenv("AMOCRM_AUTH_CODE") # --- Механизм блокировки файла --- def _acquire_lock(): while os.path.exists(LOCK_FILE): time.sleep(0.1) with open(LOCK_FILE, 'w') as f: pass def _release_lock(): if os.path.exists(LOCK_FILE): os.remove(LOCK_FILE) # --- Функции для работы с токенами --- def _save_tokens(tokens): tokens['created_at'] = int(time.time()) with open(TOKEN_FILE, "w") as f: json.dump(tokens, f) return tokens def _get_initial_tokens(): url = f"{BASE_URL}/oauth2/access_token" payload = {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "authorization_code", "code": AUTH_CODE, "redirect_uri": REDIRECT_URI} try: response = requests.post(url, json=payload) response.raise_for_status() return _save_tokens(response.json()) except requests.exceptions.RequestException as e: print(f"Ошибка при получении первоначальных токенов: {e.response.text}") return None def _refresh_tokens(existing_tokens): url = f"{BASE_URL}/oauth2/access_token" payload = {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "refresh_token", "refresh_token": existing_tokens['refresh_token'], "redirect_uri": REDIRECT_URI} try: response = requests.post(url, json=payload) response.raise_for_status() return _save_tokens(response.json()) except requests.exceptions.RequestException as e: print(f"Ошибка при обновлении токенов: {e.response.text}") return None def get_actual_tokens(): _acquire_lock() try: if not os.path.exists(TOKEN_FILE): return _get_initial_tokens() with open(TOKEN_FILE, "r") as f: tokens = json.load(f) if int(time.time()) - tokens.get('created_at', 0) > 82800: # 23 часа with open(TOKEN_FILE, "r") as f: refreshed_tokens = json.load(f) if int(time.time()) - refreshed_tokens.get('created_at', 0) > 82800: return _refresh_tokens(refreshed_tokens) return refreshed_tokens return tokens finally: _release_lock() class AmoCRMClient: def __init__(self): self.base_url = BASE_URL self.tokens = get_actual_tokens() if not self.tokens: raise Exception("Не удалось получить или обновить токены.") self.access_token = self.tokens['access_token'] def _make_request(self, method, endpoint, **kwargs): self.tokens = get_actual_tokens() if not self.tokens: raise Exception("Аутентификация провалена прямо перед запросом.") self.access_token = self.tokens['access_token'] url = f"{self.base_url}{endpoint}" headers = {"Authorization": f"Bearer {self.access_token}"} response = requests.request(method, url, headers=headers, **kwargs) return response def find_contact_by_email(self, email): endpoint = "/api/v4/contacts" params = {"query": email} response = self._make_request('GET', endpoint, params=params) if response.status_code == 200: contacts = response.json() if contacts and '_embedded' in contacts and contacts['_embedded']['contacts']: return contacts['_embedded']['contacts'] return None def create_task_for_entity(self, entity_id, entity_type, task_text, due_timestamp): endpoint = "/api/v4/tasks" payload = [{"task_type_id": 1, "text": task_text, "complete_till": due_timestamp, "entity_id": entity_id, "entity_type": entity_type}] response = self._make_request('POST', endpoint, json=payload) return response.status_code == 200
Шаг 3. Определяем лучшее время для звонка с Number Lookup API
Получив номер телефона клиента из CRM, мы отправляем его в МТС Exolve для анализа через метод GetBestCallTime.
В отличие от других подобных сервисов, МТС Exolve работает синхронно: мы отправляем запрос с номером и сразу получаем в ответ. В коде предусмотрели сценарий, если сервис вернет ошибку или не даст рекомендацию, то система назначит звонок на стандартное время.
import requests import os from dotenv import load_dotenv load_dotenv() EXOLVE_API_KEY = os.getenv("EXOLVE_API_KEY") def get_optimal_call_time(phone_number: str, on_date: str): if not EXOLVE_API_KEY: print("[WARNING] Запрос пропущен из-за отсутствия API ключа Exolve.") return "12:00" cleaned_phone = "".join(filter(str.isdigit, phone_number)) print(f"[*] Запрашиваю лучшее время для звонка через Exolve API для номера {cleaned_phone}...") api_url = "https://api.exolve.ru/hlr/v1/GetBestCallTime" headers = {"Authorization": f"Bearer {EXOLVE_API_KEY}"} payload = {"number": cleaned_phone} try: response = requests.post(api_url, json=payload, headers=headers, timeout=10) if response.status_code == 200: data = response.json() result_period = data.get("result") if result_period: # Из "17:00:00,21:00:00" берем "17:00" recommended_time = result_period.split(',')[0][:5] print(f"[+] Рекомендованное время от Exolve: {recommended_time}") return recommended_time else: print(f"[!] Ошибка от API Exolve: {response.status_code}, {response.text}") except requests.exceptions.RequestException as e: print(f"[CRITICAL ERROR] Ошибка ��ри обращении к API Exolve: {e}") print("[WARNING] Не удалось получить рекомендацию. Возвращено стандартное время.") return "12:00"
Шаг 4. Финальный этап: собираем все вместе
Теперь объединим все модули в главном файле app.py. Flask-приложение будет выполнять следующую последовательность действий:
Принять запрос и извлечь email и date.
Найти контакт в amoCRM по email и извлечь номер телефона.
Отправить номер в наш сервис и получить рекомендованное время.
Создать в amoCRM задачу с текстом и точным временем выполнения.
Ответить пользователю, что его запрос принят.
from flask import Flask, request, jsonify from datetime import datetime from amo_client import AmoCRMClient from hlr_service import get_optimal_call_time app = Flask(__name__) @app.route('/v1/callback-request', methods=['GET']) def handle_callback_request(): email = request.args.get('email') request_date_str = request.args.get('date') if not email or not request_date_str: return jsonify({"error": "Email and date parameters are required"}), 400 try: crm = AmoCRMClient() contact = crm.find_contact_by_email(email)[0] # Берем первый найденный контакт except Exception as e: return jsonify({"error": f"CRM service error: {e}"}), 503 if not contact: return jsonify({"error": "Client with this email not found in CRM"}), 404 contact_id = contact.get('id') contact_name = contact.get('name', '') phone = None if contact.get('custom_fields_values'): for field in contact.get('custom_fields_values', []): if field.get('field_code') == 'PHONE': phone = field['values'][0]['value'] break # Важная логика: если телефона нет, HLR невозможен if phone: optimal_time_str = get_optimal_call_time(phone, request_date_str) task_datetime = datetime.strptime(f"{request_date_str} {optimal_time_str}", '%Y-%m-%d %H:%M') task_text = f"Умный callback: перезвонить клиенту {contact_name}" else: optimal_time_str = None # Времени нет task_datetime = datetime.strptime(f"{request_date_str} 23:59", '%Y-%m-%d %H:%M') task_text = f"Обратный звонок (нет номера): связаться с клиентом {contact_name}" due_timestamp = int(task_datetime.timestamp()) task_created = crm.create_task_for_entity(contact_id, 'contacts', task_text, due_timestamp) if not task_created: return jsonify({"error": "Failed to create CRM task"}), 500 # Формируем ответ пользователю в зависимости от того, было ли найдено время response_time_str = f"ориентировочно в {optimal_time_str}" if optimal_time_str else "" return (f"<h1>Спасибо, {contact_name or 'уважаемый клиент'}!</h1>" f"<p>Мы получили ваш запрос. Менеджер свяжется с вами {request_date_str} {response_time_str}.</p>") if __name__ == '__main__': app.run(debug=True, port=5001)
Шаг 5. Постановка задачи в amoCRM
После нажатия клиентом на ссылку в письме, бэкенд обрабатывает запрос, определяет лучшее время для звонка и создает в amoCRM задачу, привязанную к карточке этого клиента.
Финальный результат — задача с текстом «Умный callback...» и ��онкретным рекомендованным временем. Вот как выглядит результат работы скрипта — автоматически созданная задача в карточке клиента:
Мы разработали полнофункциональную систему обратного звонка. Она решает поставленную задачу: делает заказ звонка для клиента бесшовным, а саму заявку — сразу ставит в работу менеджеру.
Как улучшить это решение
Добавить логирование, обработку ошибок API и механизм повторных попыток на случай, если CRM недоступна.
Отправлять клиенту автоматическое email-подтверждение с датой запланированного звонка и возможностью изменить время.
При переходе по ссылке из письма выводить записанное время и предлагать выбор другого времени.
Учитывать загруженность менеджеров при постановке задач в CRM.
Определять и учитывать часовой пояс по IP при переходе по ссылке.
Эта система — отличный пример того, как комбинация простых, но надежных технологий может дать мощный эффект, напрямую влияющий на бизнес-показатели. Вот код на гитхаб.
