Привет. Сегодня мы создадим полноценный инструмент повышения конверсии в звонок для 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 при переходе по ссылке.
Эта система — отличный пример того, как комбинация простых, но надежных технологий может дать мощный эффект, напрямую влияющий на бизнес-показатели. Вот код на гитхаб.