В этой статье мы расскажем, как создать собственного Telegram-бота, который интегрируется с amoCRM и распределяет уведомления о событиях в CRM, в зависимости от ролей пользователей. Система позволяет сотрудникам получать уведомления в том формате, который соответствует их обязанностям: отдел продаж видит сумму сделки и стадию, администраторы получают подробную информацию, а обычные пользователи – краткое сообщение о новой сделке. Такой подход способствует сокращению времени реакции на новые лиды, упрощает контроль за сделками и снижает необходимость постоянного ручного мониторинга CRM-системы.
Наш пример будет состоять из двух приложений. Базы данных PostgreSQL и непосредственно скрипта бота.
Развертывание сервиса мы осуществим в облаке Amvera, которое даст нам
Управляемый PostgreSQL с бэкапами, мониторингом и другими полезными функциями.
Возможность обновлять код через три команды в IDE, используя git push (либо перетягивая файлы в интерфейсе), что экономит время.
Промобаланс в 111 руб. для бесплатного тестирования.
Первый шаг в реализации проекта – создание базы данных.
Зайдите на сайт Amvera, перейдите на вкладку PostgreSQL и создайте новую базу данных, указав название проекта и выбрав тарифный план.
После заполнения всех необходимых данных для взаимодействия с БД и нажатия кнопки «Завершить» начнется создание и запуск базы данных. По завершении сборки вы увидите сообщение о том, что приложение запущено.
После настройки базы данных можно переходить к созданию самого бота. Его основная задача – обрабатывать команды пользователей, получать данные из amoCRM и отправлять уведомления в Telegram.
Бот отслеживает входящие сообщения и, если пользователь отправляет команду /start, /sales или /admin, автоматически устанавливает его роль. Бот также опрашивает amoCRM через API, чтобы получить список новых сделок. После этого бот рассылает уведомления всем зарегистрированным пользователям, формируя текст сообщения в зависимости от их роли.
Начнём с получения необходимых токенов, без которых бот не сможет работать ни с amoCRM, ни с Telegram.
Во-первых, нужно получить Telegram Bot Token. Для этого необходимо:
Открыть в Telegram бота BotFather.
Отправить ему команду
/start, а затем/newbot.Следовать инструкциям: ввести название и «username» для вашего бота (username должен оканчиваться на
bot, напримерmycompany_sales_bot).После этого BotFather вышлет вам токен
Интеграция amoCRM и телеграм-бота
Во-вторых, нужен Access Token для amoCRM. Этот токен даёт боту право обращаться к API amoCRM от имени вашего аккаунта. Процесс получения может немного отличаться в зависимости от настроек, но общая схема такова:
Зарегистрируйте интеграцию: Перейдите в amoМАРКЕТ, кликните на три точки в правом верхнем углу и выберите «Создать интеграцию», «Внешняя интеграция». В появившемся окне укажите только название интеграции и нажмите «Сохранить».
Активируйте интеграцию: Откройте раздел «Установленные» и включите только что созданную интеграцию. После этого вы получите ключи доступа.
Получите долгосрочный токен (Access Token). Именно этот токен нужен, чтобы скрипт мог вызывать методы amoCRM, создавать и просматривать сделки. Скопируйте его и сохраните в переменную
ACCESS_TOKENв вашем коде.
В итоге у вас должно быть два токена:
TELEGRAM_BOT_TOKEN— строка, выданная BotFather, используемая для взаимодействия с Telegram API.ACCESS_TOKEN— строка, дающая право боту вызывать методы API amoCRM
Теперь приступим к написанию кода для бота.
Начнём с импортов и констант. Нам понадобятся библиотеки:
requests – для HTTP-запросов (Telegram, amoCRM).
psycopg2 – для подключения к PostgreSQL.
time, datetime – для расписания проверок и обработки дат. Кроме того, укажем основные переменные: домен amoCRM, токены amoCRM и Telegram, а также конфигурацию подключения к базе данных.
import requests
import time
import psycopg2
from datetime import datetime
# Домен вашей amoCRM
AMOCRM_DOMAIN = 'name.amocrm.ru'
# Токен доступа к API amoCRM
ACCESS_TOKEN = 'eyJ0eXAiOiJKV1QiLCJhbGc...' #Важно, токены лучше не хранить в коде и сохранять в секреты и переменные окружения. Здесь дан пример для наглядности
# Telegram Bot Token, полученный у BotFather
TELEGRAM_BOT_TOKEN = '8072282519:AA...'
# Параметры подключения к PostgreSQL (DB_CONFIG)
DB_CONFIG = {
'dbname': 'DB_name',
'user': 'User_Name',
'password': 'Password',
'host': 'HOST',
'port': 5432
}
Детально про подключение к базе данных вы можете посомтреть в докуменатции Amvera
Выделим две важные переменные — UPDATE_OFFSET и STAGE_MAPPING.
UPDATE_OFFSET хранит ID последнего обработанного сообщения в Telegram. Это нужно, чтобы бот не читал одни и те же сообщения по несколько раз.
STAGE_MAPPING задаёт соответствие между ID статусов в amoCRM и их описанием. Когда сделка имеет, например, status_id = 74340918, бот будет выводить «Переговоры», а не набор чисел.
UPDATE_OFFSET = None
STAGE_MAPPING = {
74340914: "Первичный конракт",
74340918: "Переговоры",
74340922: "Ожидает подписания",
74340926: "Заключена"
}
Теперь приступим к созданию схемы базы данных. В ней будут две таблицы:
deals – хранит сделки из amoCRM (идентификатор, название, сумма, дата создания, статус уведомления и стадия).
users – хранит информацию о пользователях бота: их chat_id в Telegram и роль (user, sales, admin).
def initialize_database():
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS deals (
id SERIAL PRIMARY KEY,
deal_id INTEGER UNIQUE,
name TEXT,
price NUMERIC,
created_at TIMESTAMP,
notified BOOLEAN DEFAULT false
);
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
chat_id BIGINT UNIQUE,
role TEXT DEFAULT 'user'
);
""")
conn.commit()
cur.close()
conn.close()
print("База данных и таблицы успешно инициализированы.")
except Exception as e:
print("Ошибка при инициализации базы данных:", e)
В процессе работы нам понадобится также хранить текущую стадию сделки (stage). Для этого добавим соответствующий столбец в таблицу deals, если он ещё не существует:
def update_database_schema():
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("ALTER TABLE deals ADD COLUMN IF NOT EXISTS stage TEXT;")
conn.commit()
cur.close()
conn.close()
print("Схема базы данных успешно обновлена.")
except Exception as e:
print("Ошибка при обновлении схемы базы данных:", e)
Бот будет обрабатывать входящие сообщения. Если пользователь отправит /start, /sales или /admin, мы меняем его роль в таблице users.
Роль пользователя позволяет боту понимать, какие данные нужно отправлять при уведомлениях.
def update_user_role(chat_id, role):
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
query = """
INSERT INTO users (chat_id, role)
VALUES (%s, %s)
ON CONFLICT (chat_id) DO UPDATE SET role = EXCLUDED.role;
"""
cur.execute(query, (chat_id, role))
conn.commit()
cur.close()
conn.close()
print(f"Роль пользователя {chat_id} обновлена на {role}.")
except Exception as e:
print("Ошибка при обновлении роли пользователя:", e)
def add_user(chat_id):
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
query = "INSERT INTO users (chat_id) VALUES (%s) ON CONFLICT (chat_id) DO NOTHING;"
cur.execute(query, (chat_id,))
conn.commit()
cur.close()
conn.close()
except Exception as e:
print("Ошибка при добавлении пользователя:", e)
def get_all_users():
users = []
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("SELECT chat_id, role FROM users;")
rows = cur.fetchall()
for row in rows:
users.append({'chat_id': row[0], 'role': row[1]})
cur.close()
conn.close()
except Exception as e:
print("Ошибка при получении пользователей:", e)
return users
update_user_role назначает новую роль существующему пользователю или добавляет, если его нет.
add_user добавляет пользователя с ролью «user» по умолчанию.
get_all_users возвращает список всех пользователей и их ролей.
Чтобы бот мог считывать новые сообщения, используем метод getUpdates у Telegram. Он возвращает список недавно отправленных сообщений. Мы обрабатываем их, чтобы обновлять роли пользователей или регистрировать новых участников бота.
def update_chat_ids():
global UPDATE_OFFSET
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getUpdates"
params = {}
if UPDATE_OFFSET is not None:
params['offset'] = UPDATE_OFFSET + 1
try:
response = requests.get(url, params=params, timeout=10)
data = response.json()
updates = data.get("result", [])
if updates:
for update in updates:
UPDATE_OFFSET = update.get("update_id", UPDATE_OFFSET)
message = update.get("message", {})
chat = message.get("chat", {})
text = message.get("text", "").strip()
chat_id = chat.get("id")
if chat_id:
if text in ["/start", "/sales", "/admin"]:
if text == "/start":
role = "user"
elif text == "/sales":
role = "sales"
elif text == "/admin":
role = "admin"
update_user_role(chat_id, role)
else:
add_user(chat_id)
except Exception as e:
print("Ошибка обновления chat_ids:", e)
Для отправки уведомлений разным категориям пользователей создадим три ключевые функции:
send_message_to_user – базовая отправка текста по chat_id;
send_role_based_notification – рассылка о новой сделке, с учётом роли.
send_status_change_notification – уведомляет sales и admin при смене статуса сделки.
def send_message_to_user(chat_id, message):
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
data = {'chat_id': chat_id, 'text': message}
try:
response = requests.post(url, data=data)
if response.status_code != 200:
print(f"Ошибка отправки сообщения для {chat_id}: {response.text}")
except Exception as e:
print(f"Ошибка отправки сообщения для {chat_id}: {e}")
def send_role_based_notification(deal):
users = get_all_users()
for user in users:
role = user['role']
chat_id = user['chat_id']
if role == 'admin':
message = (
f"[ADMIN] Новая сделка!\n"
f"ID: {deal['deal_id']}\n"
f"Название: {deal['name']}\n"
f"Сумма: {deal['price']} руб.\n"
f"Стадия: {deal['stage']}\n"
f"Создана: {deal['created_at'].strftime('%Y-%m-%d %H:%M:%S')}"
)
elif role == 'sales':
message = (
f"[SALES] Новая сделка!\n"
f"Название: {deal['name']}\n"
f"Сумма: {deal['price']} руб.\n"
f"Стадия: {deal['stage']}"
)
else:
message = f"Новая сделка: {deal['name']} (Стадия: {deal['stage']})"
send_message_to_user(chat_id, message)
def send_status_change_notification(change_info):
deal = change_info["deal"]
old_stage = deal["old_stage"]
new_stage = deal["stage"]
message = (
f"Изменение статуса сделки:\n"
f"Название: {deal['name']}\n"
f"ID сделки: {deal['deal_id']}\n"
f"Старый статус: {old_stage}\n"
f"Новый статус: {new_stage}"
)
users = get_all_users()
for user in users:
if user["role"] in ["sales", "admin"]:
send_message_to_user(user["chat_id"], message)
Администратор (admin) получает самые подробные данные о сделке.
Менеджер по продажам (sales) видит основные моменты.
Рядовой пользователь (user) получает краткое сообщение.
-
При смене стадии сделки (например, «Переговоры» → «Заключена») оповещаются только sales и admin.
Теперь наша цель — сохранить каждую сделку из amoCRM в таблице deals. Если сделка новая, мы фиксируем её; если уже существует, проверяем, не изменилась ли стадия.
def store_deal_in_db(lead):
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
deal_id = lead.get('id')
name = lead.get('name', 'Нет названия')
price = lead.get('price', 0)
created_at_ts = lead.get('created_at', 0)
created_at = datetime.fromtimestamp(created_at_ts)
stage_id = lead.get('status_id')
new_stage = STAGE_MAPPING.get(stage_id, f"Стадия {stage_id}") if stage_id else "Неизвестно"
cur.execute("SELECT stage FROM deals WHERE deal_id = %s;", (deal_id,))
result = cur.fetchone()
if result is None:
query = """
INSERT INTO deals (deal_id, name, price, created_at, stage)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (deal_id) DO NOTHING;
"""
cur.execute(query, (deal_id, name, price, created_at, new_stage))
conn.commit()
cur.close()
conn.close()
return {"action": "new",
"deal": {"deal_id": deal_id, "name": name, "price": price,
"created_at": created_at, "stage": new_stage}}
else:
old_stage = result[0]
if old_stage != new_stage:
cur.execute("UPDATE deals SET stage = %s WHERE deal_id = %s;", (new_stage, deal_id))
conn.commit()
cur.close()
conn.close()
return {"action": "changed",
"deal": {"deal_id": deal_id, "name": name, "price": price,
"created_at": created_at, "stage": new_stage, "old_stage": old_stage}}
else:
cur.close()
conn.close()
return {"action": "none"}
except Exception as e:
print("Ошибка при вставке сделки в БД:", e)
return {"action": "error"}
Возвращаем несколько вариантов: new (новая сделка), changed (стадия изменилась) или none (нет изменений).
Чтобы не рассылать уведомления по одной и той же сделке много раз, используем поле notified. Если notified = false, значит мы ещё не отправляли уведомление, и нужно его отправить. После отправки обновляем notified в true.
def get_unnotified_deals():
deals = []
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("SELECT deal_id, name, price, created_at, stage FROM deals WHERE notified = false;")
rows = cur.fetchall()
for row in rows:
deals.append({
'deal_id': row[0],
'name': row[1],
'price': row[2],
'created_at': row[3],
'stage': row[4]
})
cur.close()
conn.close()
except Exception as e:
print("Ошибка при получении неуведомленных сделок:", e)
return deals
def mark_deal_as_notified(deal_id):
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute("UPDATE deals SET notified = true WHERE deal_id = %s;", (deal_id,))
conn.commit()
cur.close()
conn.close()
except Exception as e:
print("Ошибка при обновлении статуса сделки:", e)
Таким образом, каждая новая сделка (или новая версия сделки) гарантированно получит только одно уведомление при создании.
Пришло время объединить всё: опрашивать amoCRM, сохранять сделки, рассылать уведомления и отслеживать изменения статуса. Всё это делается в функции check_new_deals():
def check_new_deals():
update_chat_ids()
url = f"https://{AMOCRM_DOMAIN}/api/v4/leads"
headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code != 200:
print(f"Ошибка получения данных из amoCRM: {response.status_code} {response.text}")
return
data = response.json()
leads = data.get("_embedded", {}).get("leads", [])
except Exception as e:
print("Ошибка запроса к amoCRM:", e)
return
status_changes = []
for lead in leads:
result = store_deal_in_db(lead)
if result.get("action") == "changed":
status_changes.append(result)
unnotified_deals = get_unnotified_deals()
for deal in unnotified_deals:
send_role_based_notification(deal)
mark_deal_as_notified(deal['deal_id'])
for change_info in status_changes:
send_status_change_notification(change_info)
Алгоритм:
update_chat_ids() – смотрим, не пришло ли новых команд в боте (для изменения ролей).
GET-запрос к amoCRM – получаем список сделок (leads).
store_deal_in_db(lead) – сохраняем каждую сделку, проверяем, не изменилась ли её стадия.
get_unnotified_deals() – выбираем все сделки, по которым ещё не рассылали уведомления.
send_role_based_notification(deal) – рассылаем уведомления разным ролям.
mark_deal_as_notified() – помечаем, что сделка уже «уведомлена».
Если у какой-то сделки изменилась стадия – добавляем инфо об этом в
status_changes, а после рассылаем отдельное уведомление об изменении.
Теперь финальный аккорд: «запустить» наше приложение. В блоке if name == "__main__": мы инициализируем базу, обновляем схему, а затем в бесконечном цикле каждые 5 секунд вызываем check_new_deals().
if __name__ == "__main__":
UPDATE_OFFSET = None
initialize_database()
update_database_schema()
print("Сервис запущен. Ожидание новых сделок и рассылка уведомлений с учетом ролей...")
while True:
check_new_deals()
time.sleep(5)
initialize_database() и update_database_schema() готовят нашу БД.
Каждые 5 секунд (можно изменить интервал) функция check_new_deals() проверяет amoCRM, таблицу пользователей, рассылку уведомлений и пр.
Теперь, когда наш код полностью написан, приступим к деплою на Amvera и тестированию бота. Для этого заходим на сайт Amvera и переходим во вкладку «Приложения». В разделе «Приложения» нажимаем «Создать приложение», указываем название и выбираем тарифный план. На начальном этапе рекомендуется использовать стандартный тариф, так как он обеспечивает стабильную работу, а при необходимости тариф можно будет изменить после сборки.
После этого загружаем файлы проекта — в нашем случае это:
main.py, в котором находится весь код бота.
requirements.txt, файл со списком зависимостей. Его содержимое будет выглядеть выглядеть так:
psycopg2~=2.9.10
requests~=2.32.3
schedule~=1.2.23. amvera.yaml файл конфигурации. Данный файл лучше сконфигурировать в интерфейсе Amvera при создании проекта.
version: null
meta:
environment: python
toolchain:
name: pip
version: 3.11
build:
requirementsPath: requirements.txt
run:
scriptName: main.py
persistenceMount: /data
containerPort: 80
servicePort: 80Загрузить файлы можно через графический интерфейс Amvera, выбрав нужные файлы на компьютере, либо при помощи системы контроля версий Git.
После загрузки всех файлов на Amvera необходимо указать конфигурацию запуска для создаваемого контейнера. Когда система предложит выбрать окружение, укажите Python, а затем, в пункте «Выберите инструмент», выберите pip. Нажмите «Завершить», после чего Amvera автоматически установит все зависимости из вашего файла requirements.txt и запустит контейнер с ботом.
Как только контейнер перейдёт в статус «Приложение запущено», бот будет готов к работе. Теперь можно приступить к тестированию.
Отправляем боту команду /start, чтобы он зарегистрировал нас как обычного пользователя. Затем переходим в amoCRM и создаём новую сделку.
В Telegram получаем краткую информацию о созданной сделке
На другом устройстве (или под другой учётной записью) отправляем боту команду /sales, чтобы бот назначил нам соответствующую роль. После этого снова создаём сделку в amoCRM. В Telegram сразу же приходит более подробное уведомление
Пробуем переместить нашу тестовую сделку на другую стадию в amoCRM.
В результате бот отправляет уведомление о смене статуса.
Создаём ещё одну тестовую сделку, будучи администратором (т.е. вводим команду /admin). Теперь уведомление в боте содержит всю доступную информацию о сделке.
Итого наш проект содержит три файла.
main.py - код бота
amvera.yaml - файл конфигурации
requirements.txt - файл со списком зависисмостей.
На этом наша работа над ботом завершена.
Мы написали бота, который интегрирован с amoCRM и развернули его на удаленном сервере в облаке Amvera.
Полный исходный код и все необходимые файлы вы можете найти в репозитории на GitHub.