Как развернуть бесплатную мини-CRM в Telegram на Python и Yandex Cloud Functions

Зачем вам своя CRM, если вы не стартап

Представьте: вам пишут в Telegram. Просят цену, спрашивают про услугу, хотят договориться о встрече. Вы отвечаете, киваете, обещаете перезвонить. А через неделю забываете, кто это был и что он хотел. Лиды утекают сквозь пальцы, потому что вы не систематизируете контакты. Платные CRM-системы слишком тяжелы для одного человека или микробизнеса. А таблицы в Google — это как пытаться управлять Ferrari с помощью велосипедного руля. Нужно простое решение: прямо в мессенджере, бесплатно и без головной боли с серверами. Сегодня мы это соберем.

1 Рождение бота: берем у отца токен

Заходим в Telegram, ищем @BotFather. Пишем /newbot. Даем имя (например, Менеджер Лидов Вася) и username (должен заканчиваться на bot, типа vasya_leads_bot). Отец выдаст токен. Выглядит он так: 7102612345:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw. Это ваш ключ от царства. Скопируйте его в отдельный файл или блокнот. Пока что больше в Telegram делать ничего не нужно. Если хотите углубиться в создание ботов с нуля, есть хорошие курсы по созданию Telegram-бота.

2 Скелет CRM: что должен уметь бот

Наша мини-CRM — не монстр вроде Bitrix24. Ей нужно три функции:

• Добавить лид: Пользователь пишет команду /add, а потом вводит имя, телефон и комментарий. Бот сохраняет это.
• Показать всех лидов: Команда /list выводит табличку с сохраненными контактами.
• Удалить лида: Команда /delete 1 удаляет запись под номером 1.

Для хранения данных в serverless-архитектуре есть несколько путей. Самый простой и бесплатный в рамках Yandex Cloud — использовать Yandex Lockbox как хранилище секретов (для токена) и... обычный файл в памяти функции. Да-да, данные будут жить только во время выполнения функции. Это проблема? Нет, если вы готовы к небольшому хаку. Мы будем хранить данные в Yandex Object Storage (S3-совместимое), который тоже входит в бесплатный грант. Это даст нам постоянное хранилище за копейки (фактически 0 ₽ при небольших объемах).

3 Пишем код, который не сломается через месяц

Создаем на компьютере папку telegram_crm_bot. Внутри — файл bot.py. Устанавливаем виртуальное окружение и библиотеку: pip install python-telegram-bot==21.0 yandex-cloud.

Основная логика обработчика для Yandex Cloud Functions (используем актуальный на 2026 год фреймворк python-telegram-bot с поддержкой вебхуков):

import json
import logging
import os
from typing import Dict, Any

import boto3  # Для работы с Yandex Object Storage
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
    CallbackQueryHandler,
)

# Настройка логирования
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

# Константы (будут заполнены из переменных окружения)
BOT_TOKEN = os.environ.get("BOT_TOKEN")
AWS_ACCESS_KEY_ID = os.environ.get("AWS_ACCESS_KEY_ID")  # Ключ от Object Storage
AWS_SECRET_ACCESS_KEY = os.environ.get("AWS_SECRET_ACCESS_KEY")
BUCKET_NAME = os.environ.get("BUCKET_NAME", "telegram-crm-bucket")
DATA_FILE = "leads.json"

# Подключение к Yandex Object Storage (S3-совместимый)
def get_s3_client():
    session = boto3.session.Session()
    return session.client(
        service_name="s3",
        endpoint_url="https://storage.yandexcloud.net",
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    )

# Загрузка данных из Object Storage
def load_leads() -> list:
    try:
        s3 = get_s3_client()
        obj = s3.get_object(Bucket=BUCKET_NAME, Key=DATA_FILE)
        data = json.loads(obj["Body"].read().decode("utf-8"))
        return data.get("leads", [])
    except Exception as e:
        logger.warning(f"Failed to load leads: {e}. Starting with empty list.")
        return []

# Сохранение данных в Object Storage
def save_leads(leads: list):
    try:
        s3 = get_s3_client()
        data = json.dumps({"leads": leads}, ensure_ascii=False, indent=2)
        s3.put_object(Bucket=BUCKET_NAME, Key=DATA_FILE, Body=data.encode("utf-8"))
    except Exception as e:
        logger.error(f"Failed to save leads: {e}")

# Команда /start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    await update.message.reply_html(
        f"Привет, {user.first_name}! Я ваш менеджер лидов.\n"
        "Доступные команды:\n"
        "/add - добавить новый лид\n"
        "/list - показать все лиды\n"
        "/delete - удалить лид"
    )

# Команда /add - начинаем диалог
data_state = {}
async def add(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "Введите данные лида в формате:\n"
        "*Имя Фамилия*\n"
        "*Телефон*\n"
        "*Комментарий*\n"
        "Пример:\n"
        "Иван Петров\n"
        "+79161234567\n"
        "Хочет узнать цену на ремонт кухни",
        parse_mode="Markdown"
    )
    # Устанавливаем состояние ожидания данных
    data_state[update.effective_user.id] = "waiting_for_lead"

# Обработка текстовых сообщений (для добавления лида)
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    if user_id in data_state and data_state[user_id] == "waiting_for_lead":
        text = update.message.text
        lines = text.split("\n")
        if len(lines) >= 3:
            name = lines[0].strip()
            phone = lines[1].strip()
            comment = "\n".join(lines[2:]).strip()
            
            leads = load_leads()
            leads.append({
                "id": len(leads) + 1,
                "name": name,
                "phone": phone,
                "comment": comment,
                "date": update.message.date.isoformat()
            })
            save_leads(leads)
            del data_state[user_id]  # Сбрасываем состояние
            await update.message.reply_text(f"Лид '{name}' успешно добавлен! ID: {len(leads)}")
        else:
            await update.message.reply_text("Неверный формат. Попробуйте еще раз или отправьте /cancel.")
    else:
        await update.message.reply_text("Не понимаю. Используйте команды /start, /add, /list, /delete")

# Команда /list
async def list_leads(update: Update, context: ContextTypes.DEFAULT_TYPE):
    leads = load_leads()
    if not leads:
        await update.message.reply_text("Список лидов пуст.")
        return
    
    response = "*Ваши лиды:*\n\n"
    for lead in leads:
        response += f"ID: {lead['id']}\n"
        response += f"Имя: {lead['name']}\n"
        response += f"Телефон: {lead['phone']}\n"
        response += f"Комментарий: {lead['comment'][:50]}...\n" if len(lead['comment']) > 50 else f"Комментарий: {lead['comment']}\n"
        response += f"Дата: {lead['date'][:10]}\n"
        response += "---\n"
    
    await update.message.reply_text(response, parse_mode="Markdown")

# Команда /delete
async def delete_lead(update: Update, context: ContextTypes.DEFAULT_TYPE):
    args = context.args
    if not args or not args[0].isdigit():
        await update.message.reply_text("Использование: /delete \nПример: /delete 1")
        return
    
    lead_id = int(args[0])
    leads = load_leads()
    new_leads = [lead for lead in leads if lead["id"] != lead_id]
    
    if len(new_leads) == len(leads):
        await update.message.reply_text(f"Лид с ID {lead_id} не найден.")
    else:
        # Переиндексируем ID
        for i, lead in enumerate(new_leads, 1):
            lead["id"] = i
        save_leads(new_leads)
        await update.message.reply_text(f"Лид {lead_id} удален. Всего лидов: {len(new_leads)}")

# Главный обработчик для Yandex Cloud Functions
def handler(event: Dict[str, Any], context):
    """Точка входа для Yandex Cloud Functions."""
    # Инициализируем приложение
    application = Application.builder().token(BOT_TOKEN).build()
    
    # Регистрируем обработчики
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("add", add))
    application.add_handler(CommandHandler("list", list_leads))
    application.add_handler(CommandHandler("delete", delete_lead))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # Обработка вебхука от Telegram
    try:
        body = json.loads(event["body"])
        update = Update.de_json(body, application.bot)
        
        # Запускаем обработку синхронно (для serverless)
        application._initialize()
        result = application.process_update(update)
        
        return {
            "statusCode": 200,
            "body": "",
        }
    except Exception as e:
        logger.error(f"Handler error: {e}")
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)}),
        }

# Для локального тестирования
if __name__ == "__main__":
    # Локальный запуск с поллингом (не для продакшена в YCF)
    from telegram.ext import Updater
    
    updater = Updater(token=BOT_TOKEN, use_context=True)
    dp = updater.dispatcher
    
    dp.add_handler(CommandHandler("start", start))
    dp.add_handler(CommandHandler("add", add))
    dp.add_handler(CommandHandler("list", list_leads))
    dp.add_handler(CommandHandler("delete", delete_lead))
    dp.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    updater.start_polling()
    updater.idle()

4 Магия Yandex Cloud: настраиваем бессерверное чудо

Идем в консоль Yandex Cloud. Если аккаунта нет — создаете (потребуется банковская карта для верификации, но списаний не будет).

План действий:

1. Создаем Object Storage bucket: В разделе "Object Storage" создаем бакет с уникальным именем (например, telegram-crm-bucket-ваш-id). Записываем имя. В настройках бакета создаем статический ключ доступа (Service Account) — сохраняем AWS_ACCESS_KEY_ID и AWS_SECRET_ACCESS_KEY. Это наш аналог логина-пароля для S3 API. Бакет будет хранить файл leads.json.
2. Создаем функцию: В разделе "Cloud Functions" создаем новую функцию. Выбираем среду выполнения Python 3.12. Способ загрузки кода — архив ZIP. Упаковываем наш bot.py и файл зависимостей requirements.txt (с содержимым python-telegram-bot==21.0 boto3) в ZIP-архив.
3. Настраиваем переменные окружения функции: Это критически важный шаг. Добавляем четыре переменные:
   • BOT_TOKEN — токен вашего бота от BotFather.
   • AWS_ACCESS_KEY_ID и AWS_SECRET_ACCESS_KEY — ключи от Object Storage.
   • BUCKET_NAME — имя созданного бакета.
   Через переменные окружения мы избегаем хардкода чувствительных данных в коде. Это безопаснее и удобнее для обновлений.
4. Указываем точку входа: В настройках функции пишем bot.handler (имя файла bot, функция handler).
5. Создаем триггер: Выбираем тип триггера "HTTPS". Система выдаст вам публичный URL (например, https://functions.yandexcloud.net/abcdef123456/). Этот URL нужно сообщить Telegram для вебхуков.

Поздравляю, ваша функция живет в облаке. Но пока она не знает про Telegram.

5 Последний штрих: говорим Telegram, куда стучаться

Откройте терминал или используйте curl. Выполните команду (замените YOUR_BOT_TOKEN и YOUR_FUNCTION_URL на реальные значения):

curl -X POST \
  https://api.telegram.org/botYOUR_BOT_TOKEN/setWebhook \
  -H "Content-Type: application/json" \
  -d '{"url": "YOUR_FUNCTION_URL"}'

Ответ должен быть примерно таким: {"ok":true,"result":true,"description":"Webhook was set"}. Это значит, Telegram теперь будет отправлять все обновления от пользователей напрямую на адрес вашей функции в Yandex Cloud.

Почему это бесплатно? Считаем деньги

Давайте посчитаем на март 2026 года по тарифам Yandex Cloud:

Итого: 0 рублей в месяц при умеренном использовании. Если у вас резко вырастет количество лидов (например, 10 000 в день), то платить придется за превышение лимитов, но это будут копейки. Система масштабируется автоматически — вам не нужно думать о серверах. Это и есть сила serverless.

А что, если нужно больше? Куда расти

Этот бот — основа. Его можно превратить в мощного ИИ-агента. Например, подключить YandexGPT 3.0 (последняя версия на 2026 год) для автоматической классификации лидов или генерации ответов. Как это сделать, я подробно писал в статье про пайплайн из YandexGPT для Bitrix24. Принципы те же.

Можно добавить долгую память через векторную базу данных, чтобы бот помнил историю взаимодействий с клиентом. Об этом есть отдельный материал — "Настраиваем долгую память для OpenClaw".

А если вы хотите не просто хранить лиды, а чтобы бот сам звонил или отправлял SMS, можно интегрировать его с телефонией, как в гайде про генеративного ИИ-бота для заказов.

Типичные грабли, на которые наступают все

• Таймауты функции. Yandex Cloud Functions по умолчанию имеет лимит выполнения 10 секунд. Если ваш бот делает сложные операции (например, обращается к внешнему API), увеличьте лимит в настройках функции до 30 секунд.
• Холодный старт. Если к боту долго не обращались, функция "засыпает". Первый вызов после простоя может занять 2-3 секунды (функция инициализируется). Telegram ждет ответ 1-2 секунды. Если ответа нет, он может повторить запрос. Решение: настроить periodic trigger (раз в 5 минут), который будет "будить" функцию пустым вызовом, либо смириться с небольшой задержкой.
• Ошибки кодировки в Object Storage. Указывайте кодировку utf-8 при сохранении JSON-файла, иначе русский текст превратится в кракозябры.
• Забыли обновить вебхук после изменения URL функции. Если вы пересоздали функцию и получили новый URL, не забудьте выполнить setWebhook с новым адресом.

И последнее. Этот бот — ваш инструмент. Он не заменит человеческого общения, но спасет от бардака в личных сообщениях. А когда бизнес вырастет, вы сможете подключить его к полноценной CRM вроде Битрикс24, как описано в статье "OpenClaw в Битрикс24". Но это уже другая история. А пока — запускайте, тестируйте и ловите лидов. Бесплатно.

Подписаться на канал