Telegram-бот как CRM: ведём клиентов прямо в чате

CRM-система не обязана быть громоздким веб-приложением. Если вы фрилансер или маленькая команда — Telegram-бот может стать полноценной CRM: ведение сделок, статусы клиентов, напоминания, ежедневные отчёты. Всё прямо в чате.

Что будет уметь бот

Добавлять клиентов и сделки
Двигать сделки по воронке через inline-кнопки
Напоминать о задачах (APScheduler)
Показывать дневной/недельный отчёт
Хранить всё в PostgreSQL (или SQLite для простоты)

Схема базы данных

# models.py
from datetime import datetime
from sqlalchemy import (
    Column, Integer, BigInteger, String, Text,
    DateTime, ForeignKey, Enum as SAEnum
)
from sqlalchemy.orm import DeclarativeBase, relationship
import enum


class Base(DeclarativeBase):
    pass


class DealStatus(str, enum.Enum):
    NEW = "new"
    IN_PROGRESS = "in_progress"
    PROPOSAL = "proposal"
    WON = "won"
    LOST = "lost"


class Client(Base):
    __tablename__ = "clients"

    id = Column(Integer, primary_key=True)
    name = Column(String(200), nullable=False)
    phone = Column(String(20))
    email = Column(String(200))
    company = Column(String(200))
    notes = Column(Text)
    telegram_id = Column(BigInteger, index=True)
    created_by = Column(BigInteger, nullable=False)  # admin telegram_id
    created_at = Column(DateTime, default=datetime.utcnow)

    deals = relationship("Deal", back_populates="client")


class Deal(Base):
    __tablename__ = "deals"

    id = Column(Integer, primary_key=True)
    title = Column(String(300), nullable=False)
    amount = Column(Integer, default=0)  # в рублях
    status = Column(
        SAEnum(DealStatus), default=DealStatus.NEW, index=True
    )
    client_id = Column(Integer, ForeignKey("clients.id"))
    created_by = Column(BigInteger, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    notes = Column(Text)

    client = relationship("Client", back_populates="deals")
    reminders = relationship("Reminder", back_populates="deal")


class Reminder(Base):
    __tablename__ = "reminders"

    id = Column(Integer, primary_key=True)
    deal_id = Column(Integer, ForeignKey("deals.id"))
    text = Column(Text, nullable=False)
    remind_at = Column(DateTime, nullable=False, index=True)
    is_sent = Column(Integer, default=0)
    created_by = Column(BigInteger, nullable=False)

    deal = relationship("Deal", back_populates="reminders")

Добавление клиента (FSM)

# states.py
from aiogram.fsm.state import State, StatesGroup


class AddClientState(StatesGroup):
    name = State()
    phone = State()
    company = State()


class AddDealState(StatesGroup):
    client = State()
    title = State()
    amount = State()


class AddReminderState(StatesGroup):
    text = State()
    datetime = State()

# handlers/clients.py
from aiogram import Router, F
from aiogram.types import Message, CallbackQuery
from aiogram.fsm.context import FSMContext
from sqlalchemy.ext.asyncio import AsyncSession

from models import Client
from states import AddClientState

router = Router()


@router.callback_query(F.data == "crm:add_client")
async def start_add_client(callback: CallbackQuery, state: FSMContext):
    await state.set_state(AddClientState.name)
    await callback.message.answer("Введите имя клиента:")
    await callback.answer()


@router.message(AddClientState.name)
async def process_client_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(AddClientState.phone)
    await message.answer("Телефон (или «—» чтобы пропустить):")


@router.message(AddClientState.phone)
async def process_client_phone(message: Message, state: FSMContext):
    phone = message.text if message.text != "—" else None
    await state.update_data(phone=phone)
    await state.set_state(AddClientState.company)
    await message.answer("Компания (или «—» чтобы пропустить):")


@router.message(AddClientState.company)
async def process_client_company(
    message: Message, state: FSMContext, session: AsyncSession
):
    company = message.text if message.text != "—" else None
    data = await state.get_data()

    client = Client(
        name=data["name"],
        phone=data["phone"],
        company=company,
        created_by=message.from_user.id,
    )
    session.add(client)
    await session.commit()

    await message.answer(
        f"Клиент {client.name} добавлен (ID: {client.id})"
    )
    await state.clear()

Воронка сделок через inline-кнопки

Каждая сделка имеет статус. Менеджер двигает её по воронке одним нажатием:

# handlers/deals.py
from aiogram import Router, F
from aiogram.types import CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from models import Deal, DealStatus, Client

router = Router()

STATUS_LABELS = {
    DealStatus.NEW: "Новая",
    DealStatus.IN_PROGRESS: "В работе",
    DealStatus.PROPOSAL: "КП отправлено",
    DealStatus.WON: "Выиграна",
    DealStatus.LOST: "Проиграна",
}

STATUS_FLOW = {
    DealStatus.NEW: [DealStatus.IN_PROGRESS, DealStatus.LOST],
    DealStatus.IN_PROGRESS: [DealStatus.PROPOSAL, DealStatus.LOST],
    DealStatus.PROPOSAL: [DealStatus.WON, DealStatus.LOST],
}


def deal_card(deal: Deal) -> tuple[str, InlineKeyboardMarkup]:
    """Формирует текст и клавиатуру для карточки сделки."""
    client_name = deal.client.name if deal.client else "—"
    text = (
        f"{deal.title}\n\n"
        f"Клиент: {client_name}\n"
        f"Сумма: {deal.amount:,} ₽\n"
        f"Статус: {STATUS_LABELS[deal.status]}\n"
        f"Обновлено: {deal.updated_at.strftime('%d.%m.%Y %H:%M')}"
    )

    buttons = []
    next_statuses = STATUS_FLOW.get(deal.status, [])
    for s in next_statuses:
        buttons.append(InlineKeyboardButton(
            text=f"→ {STATUS_LABELS[s]}",
            callback_data=f"deal:move:{deal.id}:{s.value}",
        ))

    kb_rows = [buttons] if buttons else []
    kb_rows.append([
        InlineKeyboardButton(
            text="Напоминание", callback_data=f"deal:remind:{deal.id}"
        ),
        InlineKeyboardButton(
            text="Заметка", callback_data=f"deal:note:{deal.id}"
        ),
    ])

    return text, InlineKeyboardMarkup(inline_keyboard=kb_rows)


@router.callback_query(F.data == "crm:deals")
async def list_deals(callback: CallbackQuery, session: AsyncSession):
    result = await session.execute(
        select(Deal)
        .options(selectinload(Deal.client))
        .where(
            Deal.created_by == callback.from_user.id,
            Deal.status.notin_([DealStatus.WON, DealStatus.LOST]),
        )
        .order_by(Deal.updated_at.desc())
        .limit(10)
    )
    deals = result.scalars().all()

    if not deals:
        await callback.message.edit_text("Активных сделок нет.")
        return

    # Показываем первую сделку, остальные — кнопками
    text, kb = deal_card(deals[0])
    await callback.message.edit_text(text, reply_markup=kb)


@router.callback_query(F.data.startswith("deal:move:"))
async def move_deal(callback: CallbackQuery, session: AsyncSession):
    parts = callback.data.split(":")
    deal_id = int(parts[2])
    new_status = DealStatus(parts[3])

    deal = await session.get(
        Deal, deal_id, options=[selectinload(Deal.client)]
    )
    if not deal:
        await callback.answer("Сделка не найдена", show_alert=True)
        return

    old_status = deal.status
    deal.status = new_status
    await session.commit()

    text, kb = deal_card(deal)
    text += f"\n\n{STATUS_LABELS[old_status]} → {STATUS_LABELS[new_status]}"
    await callback.message.edit_text(text, reply_markup=kb)

Напоминания (APScheduler)

APScheduler запускает задачи по расписанию. Каждую минуту проверяем — есть ли напоминания, которые пора отправить:

# services/scheduler.py
from datetime import datetime

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import async_sessionmaker

from models import Reminder, Deal


async def check_reminders(bot, session_factory: async_sessionmaker):
    async with session_factory() as session:
        now = datetime.utcnow()
        result = await session.execute(
            select(Reminder)
            .join(Deal)
            .where(Reminder.is_sent == 0, Reminder.remind_at <= now)
        )
        reminders = result.scalars().all()

        for reminder in reminders:
            deal = await session.get(Deal, reminder.deal_id)
            text = (
                f"<b>Напоминание</b>\n\n"
                f"Сделка: {deal.title}\n"
                f"{reminder.text}"
            )
            try:
                await bot.send_message(reminder.created_by, text)
                reminder.is_sent = 1
            except Exception:
                pass

        await session.commit()


def setup_scheduler(bot, session_factory: async_sessionmaker):
    scheduler = AsyncIOScheduler()
    scheduler.add_job(
        check_reminders,
        "interval",
        minutes=1,
        args=[bot, session_factory],
    )
    scheduler.start()
    return scheduler

Ежедневный отчёт

Каждый день в 09:00 бот отправляет сводку: сколько новых сделок, какие задачи на сегодня, общая сумма в воронке.

# services/reports.py
from datetime import datetime, timedelta

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import async_sessionmaker

from models import Deal, DealStatus, Reminder


async def daily_report(bot, session_factory: async_sessionmaker, admin_ids: list[int]):
    async with session_factory() as session:
        today = datetime.utcnow().date()
        yesterday = today - timedelta(days=1)

        # Новые сделки за вчера
        new_count = await session.scalar(
            select(func.count(Deal.id)).where(
                func.date(Deal.created_at) == yesterday
            )
        )

        # Активные сделки
        active = await session.execute(
            select(
                Deal.status,
                func.count(Deal.id),
                func.sum(Deal.amount),
            )
            .where(Deal.status.notin_([DealStatus.WON, DealStatus.LOST]))
            .group_by(Deal.status)
        )

        # Напоминания на сегодня
        reminders_count = await session.scalar(
            select(func.count(Reminder.id)).where(
                func.date(Reminder.remind_at) == today,
                Reminder.is_sent == 0,
            )
        )

        lines = [f"<b>Отчёт за {yesterday.strftime('%d.%m.%Y')}</b>\n"]
        lines.append(f"Новых сделок: {new_count}")
        lines.append(f"Напоминаний на сегодня: {reminders_count}\n")

        total_amount = 0
        lines.append("<b>Воронка:</b>")
        for status, count, amount in active:
            label = {
                DealStatus.NEW: "Новые",
                DealStatus.IN_PROGRESS: "В работе",
                DealStatus.PROPOSAL: "КП отправлено",
            }.get(status, status.value)
            amt = amount or 0
            total_amount += amt
            lines.append(f"  {label}: {count} ({amt:,} ₽)")

        lines.append(f"\n<b>Итого в воронке: {total_amount:,} ₽</b>")

        text = "\n".join(lines)

        for admin_id in admin_ids:
            try:
                await bot.send_message(admin_id, text)
            except Exception:
                pass

Добавляем в scheduler:

scheduler.add_job(
    daily_report,
    "cron",
    hour=9,
    minute=0,
    args=[bot, session_factory, settings.admin_ids],
)

Быстрый поиск клиентов

Inline-режим — менеджер вводит @yourbot Иванов в любом чате и видит карточки клиентов:

from aiogram import Router
from aiogram.types import InlineQuery, InlineQueryResultArticle, InputTextMessageContent
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from models import Client

router = Router()


@router.inline_query()
async def search_clients(query: InlineQuery, session: AsyncSession):
    search = query.query.strip()
    if len(search) < 2:
        return

    result = await session.execute(
        select(Client)
        .where(Client.name.ilike(f"%{search}%"))
        .limit(10)
    )
    clients = result.scalars().all()

    results = []
    for client in clients:
        results.append(InlineQueryResultArticle(
            id=str(client.id),
            title=client.name,
            description=f"{client.company or '—'} | {client.phone or '—'}",
            input_message_content=InputTextMessageContent(
                message_text=f"Клиент: {client.name}\n"
                f"Телефон: {client.phone or '—'}\n"
                f"Компания: {client.company or '—'}"
            ),
        ))

    await query.answer(results, cache_time=10)

Зависимости

[project]
dependencies = [
    "aiogram>=3.13",
    "sqlalchemy[asyncio]>=2.0",
    "asyncpg>=0.29",
    "apscheduler>=3.10",
    "pydantic-settings>=2.0",
    "alembic>=1.13",
]