В Tirebase 5 торговых точек в Петербурге и продажи через маркетплейсы. Каждое утро — одна и та же рутина: открыть CRM, складской учёт, аналитику продаж, чат поддержки, таблицу с остатками. Пять вкладок, чтобы понять, что произошло за ночь. В какой-то момент я решил собрать Telegram-бота, который стягивает всё в одно место.
Проблема: информационный хаос
Типичная картина для малого бизнеса с онлайн-продажами:
- Заказы приходят через сайт, маркетплейсы и мессенджеры
- Остатки на складе обновляются в 1С, но никто не смотрит, пока товар не закончится
- Клиенты пишут в поддержку, и сообщения теряются между менеджерами
- Ежедневный отчёт собирается вручную в Excel
Владелец тратил кучу времени каждое утро только на то, чтобы собрать картину дня. А критичные уведомления (закончился товар-бестселлер, не обработан заказ за 2 часа) — просто терялись.
Архитектура решения
Я выбрал связку aiogram 3 + FastAPI с webhook-архитектурой. Не polling — потому что бот работает на том же сервере, где крутятся остальные сервисы, и webhook экономит ресурсы.
┌────────────────┐ ┌──────────────────┐ ┌─────────────┐
│ Telegram API │────▶│ FastAPI + aiogram│────▶│ PostgreSQL │
└────────────────┘ │ (webhook) │ └─────────────┘
└──────┬───────────┘
│
┌───────────┼───────────┐
▼ ▼ ▼
┌──────────┐ ┌────────┐ ┌──────────┐
│ 1С API │ │ CRM API│ │ Ozon API │
└──────────┘ └────────┘ └──────────┘
Точка входа: FastAPI + Webhook
from fastapi import FastAPI, Request
from aiogram import Bot, Dispatcher, Router
from aiogram.fsm.storage.redis import RedisStorage
app = FastAPI()
bot = Bot(token=config.BOT_TOKEN)
storage = RedisStorage.from_url(config.REDIS_URL)
dp = Dispatcher(storage=storage)
@app.post("/webhook/telegram")
async def telegram_webhook(request: Request):
update = await request.json()
await dp.feed_update(bot=bot, update=Update(**update))
return {"ok": True}
@app.on_event("startup")
async def on_startup():
webhook_url = f"{config.BASE_URL}/webhook/telegram"
await bot.set_webhook(webhook_url)
Команды: паттерн с роутерами
Каждый функциональный блок — отдельный роутер. Это позволяет изолировать логику и легко добавлять новые модули.
from aiogram import Router
from aiogram.filters import Command
orders_router = Router(name="orders")
stock_router = Router(name="stock")
reports_router = Router(name="reports")
support_router = Router(name="support")
@orders_router.message(Command("orders"))
async def cmd_orders(message: Message):
orders = await crm_client.get_today_orders()
total = sum(o.amount for o in orders)
pending = [o for o in orders if o.status == "pending"]
text = (
f"📦 Заказы за сегодня: {len(orders)}\n"
f"💰 Сумма: {total:,.0f} ₽\n"
f"⏳ Ожидают обработки: {len(pending)}"
)
await message.answer(text)
FSM для диалогов поддержки
Когда клиент пишет в поддержку, бот не просто пересылает сообщение — он ведёт диалог через конечный автомат. Сначала уточняет тему обращения, потом маршрутизирует к нужному менеджеру.
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
class SupportStates(StatesGroup):
choosing_topic = State()
describing_issue = State()
waiting_for_manager = State()
@support_router.message(Command("support"))
async def cmd_support(message: Message, state: FSMContext):
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Проблема с заказом", callback_data="topic:order")],
[InlineKeyboardButton(text="Возврат/обмен", callback_data="topic:return")],
[InlineKeyboardButton(text="Вопрос по товару", callback_data="topic:product")],
])
await message.answer("Выберите тему обращения:", reply_markup=kb)
await state.set_state(SupportStates.choosing_topic)
@support_router.callback_query(SupportStates.choosing_topic)
async def topic_chosen(callback: CallbackQuery, state: FSMContext):
topic = callback.data.split(":")[1]
manager = await get_available_manager(topic)
await state.update_data(topic=topic, manager_id=manager.telegram_id)
await callback.message.answer("Опишите вашу проблему:")
await state.set_state(SupportStates.describing_issue)
Проактивные уведомления: не ждать, пока спросят
Самая ценная часть бота — не команды, а уведомления. Я настроил периодические проверки через APScheduler:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()
@scheduler.scheduled_job("interval", minutes=30)
async def check_low_stock():
low_items = await stock_client.get_below_threshold()
if low_items:
text = "⚠️ Низкие остатки:\n"
for item in low_items:
text += f"• {item.name}: {item.qty} шт. (мин: {item.threshold})\n"
await bot.send_message(config.OWNER_ID, text)
@scheduler.scheduled_job("cron", hour=9, minute=0)
async def daily_report():
stats = await analytics.get_yesterday_summary()
text = (
f"📊 Отчёт за вчера:\n\n"
f"Заказов: {stats.orders_count}\n"
f"Выручка: {stats.revenue:,.0f} ₽\n"
f"Средний чек: {stats.avg_check:,.0f} ₽\n"
f"Новых клиентов: {stats.new_customers}\n"
f"Обращений в поддержку: {stats.support_tickets}\n"
f"Среднее время ответа: {stats.avg_response_time} мин"
)
await bot.send_message(config.OWNER_ID, text)
@scheduler.scheduled_job("interval", minutes=15)
async def check_stale_orders():
stale = await crm_client.get_unprocessed_orders(older_than_hours=2)
for order in stale:
await bot.send_message(
config.MANAGER_CHAT_ID,
f"🚨 Заказ #{order.id} не обработан уже {order.age_hours}ч!"
)
Контроль доступа
Бот не публичный. Доступ ограничен белым списком Telegram ID, а уровни прав разделены:
from functools import wraps
def require_role(role: str):
def decorator(handler):
@wraps(handler)
async def wrapper(message: Message, **kwargs):
user_role = await get_user_role(message.from_user.id)
if user_role not in ROLE_HIERARCHY.get(role, [role]):
await message.answer("Нет доступа.")
return
return await handler(message, **kwargs)
return wrapper
return decorator
@orders_router.message(Command("refund"))
@require_role("manager")
async def cmd_refund(message: Message):
# Только менеджер и выше может делать возврат
...
Что изменилось на практике
Точных замеров «до/после» у меня нет — я не мерил всё с секундомером. Но субъективно разница заметная. Утренний обзор свёлся к прочтению одного сообщения в Telegram вместо обхода пяти вкладок. Критичные остатки стали замечаться в пределах получаса, а не к вечеру. Необработанные заказы старше 2 часов — теперь редкость, потому что бот просто не даёт о них забыть.
Конечно, бот не решил все проблемы. Иногда уведомления бывают шумными — нужно тюнить пороги. И пару раз APScheduler молча падал, и я не сразу замечал. Но в целом жить стало ощутимо проще.
Чему я научился
FSM в aiogram 3 — мощная штука. Для любого многошагового диалога используйте конечные автоматы, а не цепочку if/else.
Webhook > Polling для продакшена. Polling проще для разработки, но webhook надёжнее и экономичнее. Тут, правда, есть нюанс: с webhook нужно следить за HTTPS-сертификатом и вовремя обновлять URL при деплое. Пару раз я на этом обжигался.
Бот — это не замена CRM. Это интерфейс к данным. Вся логика остаётся в существующих системах, бот только агрегирует и уведомляет.
Начинайте с уведомлений. Именно проактивные алерты дали максимальную пользу — команды используются реже, чем я ожидал.
На разработку у меня ушло примерно 2-3 недели, но я бы заложил запас на доработки — первая версия точно не закроет все сценарии.