Разработка CRM-системы в Telegram на Python: Полное руководство с примерами

Разработка CRM-системы в Telegram на Python: Полное руководство с примерами

Отличная идея! Вот подробная статья о создании CRM-приложения для Telegram на Python, с фокусом на понятность и практические примеры.

Разработка CRM-системы в Telegram на Python: Полное руководство с примерами

Введение: Почему 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.

Этот код является отправной точкой. Вы можете и должны адаптировать его под свои конкретные бизнес-процессы, расширять функционал и улучшать интерфейс. Удачи в разработке!