smart-bot-factory 0.3.2

Библиотека для создания умных чат-ботов

Smart Bot Factory

Библиотека для создания умных Telegram ботов с AI, аналитикой и гибкой архитектурой

📋 Содержание

• Установка
• Быстрый старт
• CLI Команды
• Декораторы
  • event_handler
  • schedule_task
  • global_handler
• Хуки для кастомизации
• Telegram роутеры
• Расширенные возможности

---

🚀 Установка

pip install smart_bot_factory

⚡ Быстрый старт

1. Создание бота через CLI

# Создать структуру нового бота
sbf create my-bot

# Настроить .env файл
sbf config my-bot

# Запустить бота
sbf run my-bot

2. Минимальный код бота

"""my-bot.py"""
import asyncio
from smart_bot_factory.router import EventRouter
from smart_bot_factory.message import send_message_by_human
from smart_bot_factory.creation import BotBuilder

# Инициализация
event_router = EventRouter("my-bot")
bot_builder = BotBuilder("my-bot")

@event_router.event_handler("collect_phone", once_only=True)
async def handle_phone(user_id: int, phone: str):
    """ИИ создает: {"тип": "collect_phone", "инфо": "[phone]"}"""
    await send_message_by_human(
        user_id=user_id,
        message_text=f"✅ Телефон {phone} сохранен"
    )
    return {"status": "success"}

async def main():
    bot_builder.register_routers(event_router)
    await bot_builder.build()
    await bot_builder.start()

if __name__ == "__main__":
    asyncio.run(main())

---

🎮 CLI Команды

Создание бота

# Создать нового бота из базового шаблона
sbf create my-bot

# Скопировать существующего бота
sbf copy best-valera new-valera

💡 Команда copy - создает нового бота на основе существующего:

• ✅ Копирует код бота с автозаменой bot_id
• ✅ Копирует все промпты
• ✅ Копирует тесты и файлы
• ✅ Создает новый .env (не копирует токены)

Управление ботами

# Показать список ботов
sbf list

# Запустить бота
sbf run my-bot

# Удалить бота
sbf rm my-bot

# Удалить без подтверждения
sbf rm my-bot --force

Настройка

# Открыть .env файл в редакторе
sbf config my-bot

# Управление промптами
sbf prompts my-bot                    # Список промптов
sbf prompts my-bot --edit system      # Редактировать промпт
sbf prompts my-bot --add custom       # Добавить новый промпт

Тестирование

# Запустить все тесты
sbf test my-bot

# Запустить конкретный файл
sbf test my-bot --file test_booking.yaml

# Подробный вывод
sbf test my-bot -v

Утилиты

# Показать путь к проекту
sbf path

# Создать UTM ссылку
sbf link

---

📦 Декораторы

event_handler - Обработчики событий

Назначение: Обрабатывают события от ИИ немедленно (как только ИИ создает событие).

Сигнатура:

@event_router.event_handler(
    event_type: str,      # Тип события
    notify: bool = False, # Уведомлять админов
    once_only: bool = True # Выполнять только 1 раз
)
async def handler(user_id: int, event_data: str):
    # Ваш код
    return {"status": "success"}

Параметры:

• event_type (обязательный) - Уникальное имя события
• notify (по умолчанию False) - Отправлять уведомление админам после выполнения
• once_only (по умолчанию True) - Если True, событие выполнится только 1 раз для пользователя

Как работает:

1. ИИ создает событие в JSON: {"тип": "collect_phone", "инфо": "[phone]"}
2. Обработчик вызывается немедленно
3. Событие сохраняется в БД со статусом completed
4. Если once_only=True - повторные события блокируются

Примеры:

# Базовый пример
@event_router.event_handler("collect_phone")
async def save_phone(user_id: int, phone_number: str):
    """Сохраняет телефон клиента"""
    await send_message_by_human(
        user_id=user_id,
        message_text=f"✅ Телефон {phone_number} сохранен"
    )
    return {"status": "success", "phone": phone_number}

# С уведомлением админов
@event_router.event_handler("new_lead", notify=True, once_only=True)
async def process_lead(user_id: int, lead_info: str):
    """Обрабатывает нового лида"""
    # Админы получат уведомление автоматически
    return {"status": "lead_created", "info": lead_info}

# Может выполняться многократно
@event_router.event_handler("ask_question", once_only=False)
async def handle_question(user_id: int, question: str):
    """Обрабатывает вопросы (может быть много)"""
    # Логика обработки
    return {"status": "answered"}

---

schedule_task - Запланированные задачи

Назначение: Выполняются через заданное время после создания события.

Сигнатура:

@event_router.schedule_task(
    task_name: str,              # Название задачи
    delay: Union[str, int],      # Задержка: "1h 30m" или секунды
    notify: bool = False,        # Уведомлять админов
    smart_check: bool = True,    # Умная проверка активности
    once_only: bool = True,      # Выполнять только 1 раз
    event_type: Union[str, Callable] = None  # Источник времени события
)
async def handler(user_id: int, user_data: str):
    # Ваш код
    return {"status": "sent"}

Параметры:

• task_name (обязательный) - Уникальное имя задачи
• delay (обязательный) - Задержка выполнения:
  • Строка: "1h 30m", "2h", "45m", "30s"
  • Число: 3600 (секунды)
• notify (по умолчанию False) - Уведомлять админов
• smart_check (по умолчанию True) - Умная проверка:
  • Отменяет задачу если пользователь перешел на новый этап
  • Переносит задачу если пользователь был активен
• once_only (по умолчанию True) - Выполнять только 1 раз для пользователя
• event_type (опционально) - Источник времени события:
  • Строка: "appointment_booking" - ищет событие в БД и вычисляет время
  • Функция: async def(user_id, user_data) -> datetime - кастомная логика

Формула времени с event_type:

reminder_time = event_datetime - delay

Примеры:

# Простое напоминание через 24 часа
@event_router.schedule_task("follow_up", delay="24h")
async def send_follow_up(user_id: int, reminder_text: str):
    """
    ИИ создает: {"тип": "follow_up", "инфо": "Не забудьте про запись"}
    Выполнится через 24 часа
    """
    await send_message_by_human(
        user_id=user_id,
        message_text=f"👋 {reminder_text}"
    )
    return {"status": "sent"}

# Напоминание относительно события из БД
@event_router.schedule_task(
    "booking_reminder",
    delay="2h",  # За 2 часа до записи
    event_type="appointment_booking"  # Ищет в БД событие типа "appointment_booking"
)
async def remind_booking(user_id: int, user_data: str):
    """
    ИИ создает событие: {"тип": "appointment_booking", "инфо": "дата: 2025-10-15, время: 19:00"}
    Затем создает: {"тип": "booking_reminder", "инфо": ""}
    
    Логика:
    1. Находит в БД последнее событие "appointment_booking" для user_id
    2. Парсит из него datetime: 2025-10-15 19:00
    3. Вычисляет: reminder_time = 19:00 - 2h = 17:00
    4. Отправляет напоминание в 17:00
    """
    await send_message_by_human(
        user_id=user_id,
        message_text="⏰ Напоминаю о записи через 2 часа!"
    )
    return {"status": "sent"}

# Напоминание с кастомной функцией получения времени
async def get_booking_from_api(user_id: int, user_data: str) -> datetime:
    """Получает время записи из внешнего API"""
    from yclients_api import get_next_booking
    booking = await get_next_booking(user_id)
    return booking['datetime']  # datetime объект

@event_router.schedule_task(
    "api_reminder",
    delay="1h",
    event_type=get_booking_from_api  # Функция вместо строки
)
async def send_api_reminder(user_id: int, user_data: str):
    """
    ИИ создает: {"тип": "api_reminder", "инфо": ""}
    
    Логика:
    1. Вызывается get_booking_from_api(user_id, "")
    2. Функция возвращает datetime из API
    3. Вычисляется: reminder_time = api_datetime - 1h
    4. Отправляется в вычисленное время
    """
    await send_message_by_human(user_id, "⏰ Напоминание из API!")
    return {"status": "sent"}

# Без smart_check (отправить в любом случае)
@event_router.schedule_task("important_reminder", delay="12h", smart_check=False)
async def important_reminder(user_id: int, text: str):
    """Отправится в любом случае, даже если пользователь активен"""
    await send_message_by_human(user_id, f"🔔 {text}")
    return {"status": "sent"}

---

global_handler - Глобальные обработчики

Назначение: Выполняются для всех пользователей одновременно.

Сигнатура:

@event_router.global_handler(
    handler_type: str,           # Тип обработчика
    delay: Union[str, int],      # Задержка
    notify: bool = False,        # Уведомлять админов
    once_only: bool = True,      # Выполнять только 1 раз
    event_type: Union[str, Callable] = None  # Источник времени
)
async def handler(handler_data: str):
    # Ваш код
    return {"status": "sent"}

Отличия от schedule_task:

• Нет user_id - работает глобально
• Нет smart_check - не привязан к активности пользователя
• Одно выполнение = одна рассылка всем

Примеры:

# Рассылка всем через 2 часа
@event_router.global_handler("promo_announcement", delay="2h", notify=True)
async def send_promo(announcement_text: str):
    """
    ИИ создает: {"тип": "promo_announcement", "инфо": "Скидка 20%!"}
    Отправится всем через 2 часа
    """
    await send_message_to_users_by_stage(
        stage="all",
        message_text=f"🎉 {announcement_text}",
        bot_id="my-bot"
    )
    return {"status": "sent", "recipients": "all"}

# С кастомной функцией времени
async def get_promo_end_time(handler_data: str) -> datetime:
    """Получает время окончания акции из CRM"""
    from crm_api import get_active_promo
    promo = await get_active_promo()
    return promo['end_datetime']

@event_router.global_handler(
    "promo_ending_notification",
    delay="2h",
    event_type=get_promo_end_time
)
async def notify_promo_ending(handler_data: str):
    """Уведомление всем за 2 часа до окончания акции"""
    await send_message_to_users_by_stage(
        stage="all",
        message_text="⏰ Акция заканчивается через 2 часа!",
        bot_id="my-bot"
    )
    return {"status": "sent"}

---

🎣 Хуки для кастомизации

Хуки позволяют внедрять свою логику в стандартную обработку сообщений без переписывания всей функции.

Доступные хуки

bot_builder = BotBuilder("my-bot")

# 1. Валидация сообщения (ДО обработки AI)
@bot_builder.validate_message
async def check_spam(message_text: str, message_obj):
    if "спам" in message_text.lower():
        await message_obj.answer("⛔ Спам запрещен")
        return False  # Блокировать обработку
    return True  # Продолжить

# 2. Обогащение системного промпта
@bot_builder.enrich_prompt
async def add_client_info(system_prompt: str, user_id: int):
    session = await supabase_client.get_active_session(user_id)
    phone = session.get('metadata', {}).get('phone')
    
    if phone:
        return f"{system_prompt}\n\nТелефон клиента: {phone}"
    return system_prompt

# 3. Обогащение контекста для AI
@bot_builder.enrich_context
async def add_external_data(messages: list):
    # Добавляем данные из внешнего API
    messages.append({
        "role": "system",
        "content": "Дополнительная информация из CRM..."
    })
    return messages

# 4. Обработка ответа AI
@bot_builder.process_response
async def modify_response(response_text: str, ai_metadata: dict, user_id: int):
    # Модифицируем ответ
    if "цена" in response_text.lower():
        response_text += "\n\n💰 Актуальные цены на сайте"
    return response_text, ai_metadata

# 5. Фильтры отправки
@bot_builder.filter_send
async def block_during_booking(user_id: int):
    if is_processing_booking(user_id):
        return True  # Блокировать отправку
    return False  # Разрешить

# 6. Кастомная логика после /start
@bot_builder.on_start
async def custom_start(user_id: int, session_id: str, message, state):
    """Вызывается ПОСЛЕ стандартного /start"""
    keyboard = InlineKeyboardMarkup(...)
    await message.answer("Выберите действие:", reply_markup=keyboard)

---

📱 Telegram роутеры

Подключайте чистые aiogram.Router для кастомных команд, callback'ов и фильтров.

Создание роутера

from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton

# Создаем aiogram Router
telegram_router = Router(name="my_commands")

@telegram_router.message(Command("price", "цена"))
async def price_handler(message: Message):
    """Команда /price"""
    await message.answer(
        "💰 Наши цены:\n"
        "• Услуга 1 - 1000₽\n"
        "• Услуга 2 - 2000₽"
    )

@telegram_router.message(Command("catalog"))
async def catalog_handler(message: Message):
    """Команда /catalog с кнопками"""
    keyboard = InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text="🔥 Акции", callback_data="promo")],
        [InlineKeyboardButton(text="📅 Записаться", callback_data="book")]
    ])
    await message.answer("Выберите:", reply_markup=keyboard)

@telegram_router.callback_query(F.data == "book")
async def handle_booking(callback: CallbackQuery):
    """Обработка кнопки"""
    await callback.answer("Записываю...")
    await callback.message.answer("Напишите желаемую дату")

@telegram_router.message(F.text.lower().contains("помощь"))
async def help_handler(message: Message):
    """Реагирует на слово 'помощь'"""
    await message.answer("Чем могу помочь?")

# Регистрация в боте
bot_builder.register_telegram_router(telegram_router)

Множественная регистрация

commands_router = Router(name="commands")
callbacks_router = Router(name="callbacks")
filters_router = Router(name="filters")

# Регистрируем все сразу
bot_builder.register_telegram_routers(
    commands_router,
    callbacks_router,
    filters_router
)

⚠️ Важно: Пользовательские роутеры подключаются ПЕРВЫМИ (высший приоритет), поэтому ваши команды обрабатываются раньше стандартных.

---

🔧 Расширенные возможности

Кастомный PromptLoader

Создайте свой загрузчик промптов с автоматическим определением пути:

from smart_bot_factory.utils import UserPromptLoader

# Автоматически найдет bots/my-bot/prompts
custom_loader = UserPromptLoader("my-bot")

# Или наследуйтесь для кастомизации
class MyPromptLoader(UserPromptLoader):
    def __init__(self, bot_id):
        super().__init__(bot_id)
        self.extra_file = self.prompts_dir / 'extra.txt'

my_loader = MyPromptLoader("my-bot")

# Установите ДО build()
bot_builder.set_prompt_loader(my_loader)

Полная замена обработки событий

Замените стандартную функцию process_events:

from smart_bot_factory.message import get_bot
from smart_bot_factory.core.decorators import execute_event_handler

async def my_process_events(session_id, events, user_id):
    """Моя кастомная обработка событий"""
    bot = get_bot()
    
    for event in events:
        event_type = event.get('тип')
        
        if event_type == 'booking':
            # Ваша кастомная логика
            telegram_user = await bot.get_chat(user_id)
            name = telegram_user.first_name
            # ... обработка
        else:
            # Стандартная обработка остальных
            await execute_event_handler(event_type, user_id, event.get('инфо'))

# Установите ДО build()
bot_builder.set_event_processor(my_process_events)

Доступ к aiogram Bot

Получите прямой доступ к aiogram.Bot:

from smart_bot_factory.message import get_bot

@event_router.event_handler("check_user")
async def get_user_info(user_id: int, event_data: str):
    """Получает информацию из Telegram"""
    bot = get_bot()
    
    # Используем любые методы aiogram Bot
    telegram_user = await bot.get_chat(user_id)
    name = telegram_user.first_name
    username = telegram_user.username
    
    await bot.send_message(user_id, f"Привет, {name}!")
    return {"name": name, "username": username}

Отправка сообщений с файлами

from smart_bot_factory.message import send_message

@event_router.event_handler("send_catalog")
async def send_catalog(user_id: int, event_data: str):
    """Отправляет каталог с файлами"""
    from smart_bot_factory.message import get_bot
from smart_bot_factory.supabase import SupabaseClient

    bot = get_bot()
    supabase_client = SupabaseClient("my-bot")
    
    # Получаем message объект (для ответа)
    # В реальности используйте message из контекста
    
    await send_message(
        message=message,  # aiogram Message объект
        text="📁 Вот наш каталог:",
        supabase_client=supabase_client,
        files_list=["catalog.pdf", "price_list.pdf"],
        parse_mode="Markdown"
    )
    
    return {"status": "sent"}

---

📚 Полный пример

"""advanced-bot.py - Продвинутый пример"""

import asyncio
from datetime import datetime, timedelta

from smart_bot_factory.router import EventRouter
from smart_bot_factory.message import send_message_by_human, get_bot
from smart_bot_factory.creation import BotBuilder
from smart_bot_factory.supabase import SupabaseClient

from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton

# Инициализация
event_router = EventRouter("advanced-bot")
telegram_router = Router(name="commands")
bot_builder = BotBuilder("advanced-bot")
supabase_client = SupabaseClient("advanced-bot")

# ========== СОБЫТИЯ ==========

@event_router.event_handler("collect_phone", notify=True, once_only=True)
async def save_phone(user_id: int, phone: str):
    session = await supabase_client.get_active_session(user_id)
    if session:
        metadata = session.get('metadata', {})
        metadata['phone'] = phone
        await supabase_client.update_session_metadata(session['id'], metadata)
    
    await send_message_by_human(user_id, f"✅ Телефон {phone} сохранен")
    return {"status": "success"}

# ========== ЗАДАЧИ ==========

async def get_appointment_time(user_id: int, user_data: str) -> datetime:
    """Получает время из YClients API"""
    # Ваша интеграция с YClients
    return datetime.now() + timedelta(hours=24)

@event_router.schedule_task(
    "appointment_reminder",
    delay="2h",
    event_type=get_appointment_time,
    smart_check=False
)
async def remind_appointment(user_id: int, user_data: str):
    await send_message_by_human(user_id, "⏰ Запись через 2 часа!")
    return {"status": "sent"}

# ========== ГЛОБАЛЬНЫЕ ==========

@event_router.global_handler("daily_promo", delay="24h", once_only=False)
async def daily_promo(text: str):
    await send_message_to_users_by_stage(
        stage="all",
        message_text=f"🎉 {text}",
        bot_id="advanced-bot"
    )
    return {"status": "sent"}

# ========== TELEGRAM КОМАНДЫ ==========

@telegram_router.message(Command("price"))
async def price_cmd(message: Message):
    await message.answer("💰 Цены: ...")

@telegram_router.callback_query(F.data == "book")
async def booking_callback(callback):
    await callback.answer("Записываю...")

# ========== ХУКИ ==========

@bot_builder.validate_message
async def check_business_hours(message_text: str, message_obj):
    """Проверка рабочих часов"""
    hour = datetime.now().hour
    if hour < 9 or hour > 21:
        await message_obj.answer("Мы работаем с 9:00 до 21:00")
        return False
    return True

@bot_builder.enrich_prompt
async def add_client_data(system_prompt: str, user_id: int):
    """Добавляет данные клиента в промпт"""
    session = await supabase_client.get_active_session(user_id)
    phone = session.get('metadata', {}).get('phone')
    
    if phone:
        return f"{system_prompt}\n\nТелефон клиента: {phone}"
    return system_prompt

# ========== ЗАПУСК ==========

async def main():
    # Регистрация
    bot_builder.register_routers(event_router)
    bot_builder.register_telegram_router(telegram_router)
    
    # Кастомизация (опционально)
    # from smart_bot_factory.utils import UserPromptLoader
    # bot_builder.set_prompt_loader(UserPromptLoader("advanced-bot"))
    
    # Сборка и запуск
    await bot_builder.build()
    await bot_builder.start()

if __name__ == "__main__":
    asyncio.run(main())

---

📖 Структура проекта

project/
├── bots/
│   └── my-bot/
│       ├── prompts/              # Промпты для AI
│       │   ├── system_prompt.txt
│       │   ├── welcome_message.txt
│       │   └── final_instructions.txt
│       ├── tests/                # YAML тесты
│       ├── welcome_files/        # Файлы приветствия
│       ├── files/                # Файлы для отправки
│       └── .env                  # Конфигурация
├── my-bot.py                     # Код бота
└── .env                          # Глобальная конфигурация (опционально)

---

⚙️ Конфигурация (.env)

# Telegram
TELEGRAM_BOT_TOKEN=your_token_here

# Supabase
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_KEY=your_key_here

# OpenAI
OPENAI_API_KEY=sk-your-key
OPENAI_MODEL=gpt-4o-mini
OPENAI_MAX_TOKENS=1500

# Администраторы (Telegram ID через запятую)
ADMIN_TELEGRAM_IDS=123456789,987654321

# Режим отладки (показывать JSON)
DEBUG_MODE=false

---

🎯 Сравнение декораторов

Декоратор	Когда выполняется	Для кого	Параметры
@event_handler	Немедленно	1 пользователь	event_type, notify, once_only
@schedule_task	Через время	1 пользователь	task_name, delay, event_type, smart_check, once_only, notify
@global_handler	Через время	Все пользователи	handler_type, delay, event_type, once_only, notify

---

🔑 Ключевые концепции

once_only=True

Гарантирует выполнение события только 1 раз для пользователя:

• При сохранении: Проверяет БД, если есть - не сохраняет
• При выполнении: Проверяет БД, если есть completed - отменяет

@event_router.event_handler("welcome_bonus", once_only=True)
async def give_bonus(user_id: int, bonus_info: str):
    # Выполнится только 1 раз, даже если пользователь сделает /start заново
    return {"status": "bonus_given"}

smart_check=True

Умная проверка для запланированных задач:

• Отменяет задачу если пользователь перешел на новый этап
• Переносит задачу если пользователь был недавно активен

@event_router.schedule_task("follow_up", delay="24h", smart_check=True)
async def follow_up(user_id: int, text: str):
    # Не отправится если пользователь уже был активен
    return {"status": "sent"}

event_type - Привязка ко времени события

Планирует задачу относительно времени события:

Строка - ищет в БД:

@event_router.schedule_task("reminder", delay="2h", event_type="appointment")
async def remind(user_id: int, text: str):
    # 1. ИИ создает событие: {"тип": "appointment", "инфо": "дата: 2025-10-15, время: 19:00"}
    # 2. ИИ создает задачу: {"тип": "reminder", "инфо": ""}
    # 3. Ищется в БД событие "appointment" для user_id
    # 4. Парсится datetime: 2025-10-15 19:00
    # 5. Вычисляется: 19:00 - 2h = 17:00
    # 6. Задача выполняется в 17:00
    pass

Функция - кастомная логика:

async def get_time_from_api(user_id: int, user_data: str) -> datetime:
    booking = await external_api.get_booking(user_id)
    return booking['datetime']

@event_router.schedule_task("api_reminder", delay="1h", event_type=get_time_from_api)
async def remind(user_id: int, text: str):
    # 1. ИИ создает: {"тип": "api_reminder", "инфо": ""}
    # 2. Вызывается get_time_from_api(user_id, "")
    # 3. Функция возвращает datetime из API
    # 4. Вычисляется: api_datetime - 1h
    # 5. Задача выполняется в вычисленное время
    pass

---

🚀 Публикация изменений

Если вы разработчик библиотеки:

# Автоматически увеличивает версию и публикует в PyPI
uv run publish.py

# Требует PYPI_API_TOKEN в .env файле

---

📞 Поддержка

• Документация: GitHub
• Issues: GitHub Issues

---

📄 Лицензия

MIT