Почему Telegram как платформа для CRM?
В современном бизнесе важно быть там, где есть клиенты. А клиенты все чаще — в мессенджерах. Telegram, с его мощным API, кросс-платформенностью и популярностью, представляет собой идеальную площадку для развертывания легкой, эффективной и доступной CRM-системы.
Такая CRM не требует от ваших менеджеров по продажам или сотрудников поддержки постоянно сидеть в веб-интерфейсе. Уведомления, задачи и сообщения от клиентов приходят прямо в привычный мессенджер. Это значительно ускоряет реакцию и повышает удобство работы.
В этой статье мы поэтапно разберем, как создать свою собственную CRM для Telegram, используя Python. Мы рассмотрим архитектуру, ключевые библиотеки и напишем конкретные примеры кода для реализации основного функционала.
Что мы будем разрабатывать?
Наша CRM будет иметь следующий базовый функционал:
1. Телеграм-бот как интерфейс для сотрудников.
2. Регистрация и аутентификация сотрудников.
3. Управление клиентами (добавление, просмотр, редактирование).
4. Управление заявками/сделками (создание, изменение статуса, назначение ответственного).
5. Уведомления о новых заявках и изменениях.
6. Простая база данных для хранения информации.
Технологический стек
- Python 3.8+
- python-telegram-bot (`python-telegram-bot` v20.x) — современная и асинхронная библиотека для работы с Telegram Bot API.
- SQLAlchemy — ORM (Object-Relational Mapping) для работы с базой данных на высоком уровне.
- SQLite — простая файловая база данных для начала разработки (легко заменяется на PostgreSQL или MySQL).
- Alembic (опционально) — для управления миграциями базы данных.
Часть 1: Настройка проекта и окружения
1.1. Создание бота в Telegram
Первым делом нужно создать самого бота через `@BotFather`.
- Напишите `/start` BotFather-у.
- Затем команду `/newbot`.
- Следуйте инструкциям: задайте имя бота (например, `MyCompany CRM`) и уникальный username (например, `mycompany_crm_bot`).
- В завершение вы получите **API-токен** вида `1234567890:ABCDEFGhIjKlMnOpQRsTuvWxYZ-abcdefghi`. Сохраните его, он понадобится для подключения к вашему коду.
1.2. Структура проекта
Создайте следующую структуру папок:
/my_telegram_crm
..../crm_bot.py # Главный файл, запуск бота
..../config.py # Конфигурация: токен, настройки БД
..../models.py # Модели SQLAlchemy (таблицы БД)
..../handlers.py # Обработчики сообщений и команд
..../database.py # Инициализация и работа с БД
..../utils.py # Вспомогательные функции
..../requirements.txt # Зависимости проекта
1.3. Установка зависимостей
Создайте файл `requirements.txt`:
python-telegram-bot==20.7
sqlalchemy==2.0.23
alembic==1.12.1
Установите их через pip:
pip install -r requirements.txt
1.4. Настройка конфигурации (`config.py`)
Вынесем критически важные данные в конфиг.
import os
class Config:
....# Токен вашего бота, полученный от @BotFather
....BOT_TOKEN = os.getenv('BOT_TOKEN', '1234567890:ABCDEFGhIjKlMnOpQRsTuvWxYZ-abcdefghi')
....# Настройки базы данных (для начала используем SQLite)
....DB_PATH = os.getenv('DB_PATH', 'sqlite:///crm_database.db')
....# ID администратора/ов (узнать свой ID можно через @userinfobot)
....# Уведомления будут отправляться сюда
....ADMIN_IDS = [123456789, 987654321] # Замените на реальные ID
Часть 2: Проектирование и настройка базы данных (`models.py`)
Опишем основные сущности нашей CRM.
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, Enum
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime
import enum
# Базовый класс для моделей
Base = declarative_base()
# Статусы для заявки
class StatusEnum(enum.Enum):
....NEW = "Новая"
....IN_PROGRESS = "В работе"
....WAITING = "Ожидание"
....DONE = "Завершена"
....CANCELLED = "Отменена"
# Модель сотрудника
class Employee(Base):
....__tablename__ = 'employees'
....id = Column(Integer, primary_key=True)
....telegram_id = Column(Integer, unique=True, nullable=False) # ID пользователя ....в Telegram
....username = Column(String(100)) # @username
....first_name = Column(String(100))
....last_name = Column(String(100))
....role = Column(String(50), default='manager') # manager, admin, etc.
....is_active = Column(Integer, default=1) # 1 - активен, 0 - неактивен
....created_at = Column(DateTime, default=datetime.utcnow)
# Связь "один ко многим" с заявками (один сотрудник - много заявок)
leads = relationship("Lead", back_populates="assignee")
# Модель клиента
class Customer(Base):
....__tablename__ = 'customers'
....id = Column(Integer, primary_key=True)
....telegram_id = Column(Integer, unique=True) # Если клиент тоже из Telegram
....username = Column(String(100))
....first_name = Column(String(100))
....last_name = Column(String(100))
....phone_number = Column(String(20))
....email = Column(String(100))
....created_at = Column(DateTime, default=datetime.utcnow)
# Связь с заявками
....leads = relationship("Lead", back_populates="customer")
# Модель заявки/сделки
class Lead(Base):
....__tablename__ = 'leads'
....id = Column(Integer, primary_key=True)
....title = Column(String(200), nullable=False) # Краткое название заявки
....description = Column(Text) # Подробное описание
....status = Column(Enum(StatusEnum), default=StatusEnum.NEW)
....created_at = Column(DateTime, default=datetime.utcnow)
....updated_at = Column(DateTime, default=datetime.utcnow, ........onupdate=datetime.utcnow)
# Внешние ключи
....customer_id = Column(Integer, ForeignKey('customers.id'))
....assignee_id = Column(Integer, ForeignKey('employees.id')) # Ответственный ....сотрудник
# Связи
customer = relationship("Customer", back_populates="leads")
assignee = relationship("Employee", back_populates="leads")
Часть 3: Инициализация базы данных и сессии (`database.py`)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base
from config import Config
# Создаем движок для подключения к БД
engine = create_engine(Config.DB_PATH, echo=True) # echo=True для отладки SQL запросов
# Создаем таблицы в базе данных
Base.metadata.create_all(engine)
# Создаем фабрику сессий
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
# Функция для получения сессии (будет использоваться в обработчиках)
def get_db_session():
....session = SessionLocal()
....try:
........yield session
....finally:
........session.close()
Часть 4: Регистрация сотрудников и система прав
Прежде чем работать с CRM, сотрудник должен быть в нее добавлен. Реализуем простую команду `/start`, которая будет проверять, есть ли пользователь в базе.
В `handlers.py`:
from telegram import Update, ReplyKeyboardMarkup
from telegram.ext import ContextTypes, CommandHandler, MessageHandler, filters, ConversationHandler
from models import Employee, Customer, Lead, StatusEnum
from database import get_db_session
from config import Config
# States для ConversationHandler
AWAITING_PHONE = 1
# Команда /start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
....user = update.effective_user
....session = next(get_db_session())
....# Проверяем, есть ли сотрудник в базе
....employee = session.query(Employee).filter(Employee.telegram_id == user.id).first()
....if employee:
........# Сотрудник найден
........welcome_text = f"Добро пожаловать в CRM, {employee.first_name}!"
........keyboard = [['📊 Мои заявки', '➕ Новая заявка'], ['👥 Клиенты', '❓ ........Помощь']]
........reply_markup = ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
........await update.message.reply_text(welcome_text, reply_markup=reply_markup)
....else:
........# Сотрудник не найден, проверяем, может это клиент?
........customer = session.query(Customer).filter(Customer.telegram_id == ............user.id).first()
........if not customer:
............# Это новый клиент, предлагаем зарегистрироваться
............new_customer = Customer(
................telegram_id=user.id,
................username=user.username,
................first_name=user.first_name,
................last_name=user.last_name
............)
............session.add(new_customer)
............session.commit()
............await update.message.reply_text(
................"Спасибо за обращение! Ваша заявка будет обработана в ................ближайшее время."
............)
............# Отправляем уведомление администратору о новом клиенте
............for admin_id in Config.ADMIN_IDS:
................await context.bot.send_message(
....................chat_id=admin_id,
....................text=f"🎉 Новый потенциальный клиент: {user.first_name} @{user.username}"
................)
........else:
............await update.message.reply_text("Вы уже зарегистрированы как клиент. С вами свяжется менеджер.")
....session.close()
# ... другие обработчики ...
Часть 5: Ядро функционала — обработчики команд
5.1. Создание новой заявки (используем ConversationHandler)
Это более сложный, но мощный инструмент для многошаговых диалогов.
# States для процесса создания заявки
TITLE, DESCRIPTION = range(2)
async def new_lead_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
...."""Начинает процесс создания заявки."""
....await update.message.reply_text("Отлично! Давайте создадим новую заявку. Введите краткое название:")
....return TITLE
async def new_lead_title(update: Update, context: ContextTypes.DEFAULT_TYPE):
...."""Получает название заявки."""
....user_title = update.message.text
....context.user_data['lead_title'] = user_title
....await update.message.reply_text("Хорошо. Теперь введите подробное описание:")
....return DESCRIPTION
async def new_lead_description(update: Update, context: ContextTypes.DEFAULT_TYPE):
...."""Получает описание и сохраняет заявку в БД."""
....user_description = update.message.text
....user = update.effective_user
....session = next(get_db_session())
....# Находим сотрудника, который создает заявку
....employee = session.query(Employee).filter(Employee.telegram_id == user.id).first()
....if not employee:
........await update.message.reply_text("Ошибка: вы не зарегистрированы как сотрудник.")
........session.close()
........return ConversationHandler.END
....# Находим или создаем клиента "по умолчанию" для примера.
....# В реальной системе клиент должен выбираться из списка или привязываться иначе.
....default_customer = session.query(Customer).first()
....# Создаем заявку
....new_lead = Lead(
........title=context.user_data['lead_title'],
........description=user_description,
........customer_id=default_customer.id,
........assignee_id=employee.id # Назначаем на себя
....)
....session.add(new_lead)
....session.commit()
....# Очищаем временные данные
....context.user_data.clear()
....await update.message.reply_text(f"✅ Заявка '#{new_lead.id} {new_lead.title}' успешно создана и назначена на вас!")
....# Отправляем уведомление админам (если нужно)
....for admin_id in Config.ADMIN_IDS:
........if admin_id != user.id: # Чтобы не дублировать себе
............await context.bot.send_message(
................chat_id=admin_id,
................text=f"🆕 Создана новая заявка: #{new_lead.id} {new_lead.title}\n"
f"Ответственный: {employee.first_name}"
............)
....session.close()
....return ConversationHandler.END
async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
...."""Отменяет текущий диалог."""
....await update.message.reply_text('Создание заявки отменено.')
....context.user_data.clear()
....return ConversationHandler.END
# Регистрируем ConversationHandler в главном файле (crm_bot.py)
lead_conv_handler = ConversationHandler(
entry_points=[MessageHandler(filters.Regex('^➕ Новая заявка$'), new_lead_start)],
states={
TITLE: [MessageHandler(filters.TEXT & ~filters.COMMAND, new_lead_title)],
DESCRIPTION: [MessageHandler(filters.TEXT & ~filters.COMMAND, new_lead_description)],
},
fallbacks=[CommandHandler('cancel', cancel)],
)
5.2. Просмотр списка заявок
async def show_my_leads(update: Update, context: ContextTypes.DEFAULT_TYPE):
...."""Показывает заявки, назначенные на текущего сотрудника."""
....user = update.effective_user
....session = next(get_db_session())
....employee = session.query(Employee).filter(Employee.telegram_id == user.id).first()
....if not employee:
........await update.message.reply_text("Ошибка доступа.")
........session.close()
........return
....leads = session.query(Lead).filter(Lead.assignee_id == employee.id).order_by(Lead.status, Lead.updated_at.desc()).all()
....if not leads:
........await update.message.reply_text("У вас нет активных заявок.")
........session.close()
........return
....response_text = "📋 *Ваши заявки:*\n\n"
....for lead in leads:
........response_text += f"*#{lead.id}* - {lead.title}\n"
........response_text += f"Статус: `{lead.status.value}`\n"
........response_text += f"Клиент: {lead.customer.first_name}\n"
........response_text += f"Обновлена: {lead.updated_at.strftime('%d.%m.%Y %H:%M')}\n\n"
....await update.message.reply_text(response_text, parse_mode='Markdown')
....session.close()
Часть 6: Запуск бота (`crm_bot.py`)
Здесь мы соберем все компоненты вместе.
from config import Config
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from handlers import start, show_my_leads, lead_conv_handler
def main():
....# Создаем Application и передаем ему токен бота
....application = Application.builder().token(Config.BOT_TOKEN).build()
....# Регистрируем обработчики
....application.add_handler(CommandHandler("start", start))
....application.add_handler(MessageHandler(filters.Regex('^📊 Мои заявки$'), show_my_leads))
....application.add_handler(lead_conv_handler) # Добавляем ConversationHandler
....# Запускаем бота
....print("Бот запущен...")
....application.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == '__main__':
....main()
Часть 7: Дальнейшее развитие и улучшения
Наша базовая CRM готова. Но это только фундамент. Вот куда можно двигаться дальше:
1. Административная панель: Веб-интерфейс на Flask/Django для админов, где можно добавлять сотрудников, смотреть общую статистику, править любые данные.
2. Инлайн-кнопки и меню: Замена текстовых кнопок на `InlineKeyboardButton` для действий "Взять в работу", "Изменить статус", "Назначить другому менеджеру" прямо в сообщении с заявкой.
3. Интеграция с внешними системами: Отправка данных в Google Sheets, 1С или AmoCRM через их API.
4. Финальная авторизация: Реализация более строгой системы ролей (админ, менеджер, гость).
5. Каналы и группы: Настройка получения заявок не только в ЛС бота, но и из Telegram-каналов или чатов поддержки.
6. Ведение диалога с клиентом: Функционал, при котором менеджер может ответить клиенту прямо из интерфейса бота, и сообщение перешлется клиенту (используя `bot.copy_message`).
7. Миграция на PostgreSQL: Для большей надежности и производительности в продакшене.
Заключение
Разработка CRM в Telegram на Python — это мощный и относительно быстрый способ автоматизировать процессы вашего бизнеса, находясь в одном из самых популярных мессенджеров. Мы рассмотрели ключевые этапы: от настройки бота и проектирования базы данных до реализации базового функционала по управлению заявками с помощью асинхронной библиотеки `python-telegram-bot` и ORM SQLAlchemy.
Этот код является отправной точкой. Вы можете и должны адаптировать его под свои конкретные бизнес-процессы, расширять функционал и улучшать интерфейс. Удачи в разработке!
Подписывайтесь:
Телеграм https://t.me/lets_go_code
Канал "Просто о программировании" https://dzen.ru/lets_go_code