Как собрать очередь заявок в Telegram-боте: от FSM до inline-кнопок

Очередь модерации в Telegram-боте — это связка из трёх вещей: FSM (пошаговый сбор заявки от пользователя), inline-кнопок под сообщением админа (принять / отклонить / на доработку) и статусов в базе данных (pending, approved, rejected, revision). Пользователь жмёт «Старт», бот ведёт его по шагам, в конце отправляет готовую заявку админу одним сообщением с кнопками. Админ нажимает решение прямо в чате — бот обновляет сообщение и шлёт ответ пользователю.

Это самый частый паттерн в коммерческих ботах: курсы, услуги, вакансии, модерация объявлений, заявки в сообщество, приём контента в канал. Везде, где есть «пришла заявка — человек проверил — пользователь получил ответ», работает одна и та же механика.

**Ключевое правило: **заявки в базе никогда не удаляются. Только меняется статус. Это даёт полную историю и возможность вернуться к любому решению через месяц.

Корневая идея простая: пользователь не должен писать «принято?» в личку модератору. Модератор не должен открывать базу, искать заявку, проставлять галочки. Решение принимается в один клик, прямо в Telegram.

Что это

Система состоит из трёх участников и трёх зон ответственности. Пользователь заполняет форму и ждёт ответа. База данных хранит заявки и их статусы. Администратор видит заявки и принимает решения кнопками под сообщением.

Технически это один пайплайн: пользовательский сценарий на FSM, административный сценарий на callback_query, и прослойка в виде БД, где у каждой заявки есть уникальный id и текущий статус. Никакой магии — только согласованная работа трёх слоёв.

Зачем нужно

• Заявки на курс, услугу, вакансию, размещение в каталоге, вступление в сообщество.
• Модерация пользовательского контента перед публикацией в канале или чате.
• Приём заявок в закрытый канал или платную группу с проверкой оплаты.
• Сбор откликов от пользователей с последующей передачей менеджеру.
• Любой процесс, где решение принимает человек, а пользователь ждёт ответа.

Как устроено

Фундамент системы — статусы. Каждая заявка всегда находится ровно в одном из них:

Минимальная структура заявки в базе данных:

**Совет: **поле admin\_message\_id критически важно. Когда модератор нажимает кнопку, нужно обновить сообщение в его чате (убрать кнопки, показать финальный статус). Без этого id кнопки зависнут, и любой повторный клик даст ошибку.

Пример

Полный рабочий скелет на aiogram 3. Сокращения нет — все блоки из источника, только переписаны под структуру базы знаний.

Конфигурация и инициализация бота:

from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, CallbackQuery
import json
import aiosqlite

BOT_TOKEN = "<token>"
ADMIN_CHAT_ID = 123456789  # личка админа или id группы

bot = Bot(token=BOT_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(storage=storage)

class Form(StatesGroup):
    name = State()
    description = State()
    contact = State()

DB_SCHEMA = """
CREATE TABLE IF NOT EXISTS applications (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    admin_message_id INTEGER,
    reason TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""

async def init_db():
    async with aiosqlite.connect("bot.db") as db:
        await db.execute(DB_SCHEMA)
        await db.commit()

Пользовательская часть: старт FSM-сценария и сбор ответов.

@dp.message(Command("start"))
async def cmd_start(message: types.Message, state: FSMContext):
    await state.set_state(Form.name)
    await message.answer("Привет! Как тебя зовут?")

@dp.message(Form.name)
async def process_name(message: types.Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(Form.description)
    await message.answer("Расскажи кратко о себе (одним сообщением).")

@dp.message(Form.description)
async def process_description(message: types.Message, state: FSMContext):
    await state.update_data(description=message.text)
    await state.set_state(Form.contact)
    await message.answer("Оставь контакт для связи (телефон, telegram, почта).")

@dp.message(Form.contact)
async def process_contact(message: types.Message, state: FSMContext):
    data = await state.get_data()
    data["contact"] = message.text
    payload = json.dumps(data, ensure_ascii=False)

    kb = InlineKeyboardMarkup(inline_keyboard=[
        [
            InlineKeyboardButton(text="Принять", callback_data="approve"),
            InlineKeyboardButton(text="Отклонить", callback_data="reject")
        ],
        [
            InlineKeyboardButton(text="На доработку", callback_data="revision")
        ]
    ])

    text = (
        f"Новая заявка от @{message.from_user.username or message.from_user.id}:\n\n"
        f"Имя: {data['name']}\n"
        f"О себе: {data['description']}\n"
        f"Контакт: {data['contact']}"
    )

    sent = await bot.send_message(ADMIN_CHAT_ID, text, reply_markup=kb)

    async with aiosqlite.connect("bot.db") as db:
        await db.execute(
            "INSERT INTO applications (user_id, payload, admin_message_id) VALUES (?, ?, ?)",
            (message.from_user.id, payload, sent.message_id)
        )
        await db.commit()

    await state.clear()
    await message.answer("Заявка отправлена на модерацию. Мы свяжемся с тобой.")

Обработчики кнопок администратора. Здесь же — обновление сообщения, чтобы кнопки не висели вечно.

def is_admin(user_id: int) -> bool:
    return user_id == ADMIN_CHAT_ID

async def update_admin_message(callback: CallbackQuery, status_text: str):
    await callback.message.edit_text(
        callback.message.text + f"\n\nРешение: {status_text}",
        reply_markup=None
    )

async def notify_user(user_id: int, text: str):
    try:
        await bot.send_message(user_id, text)
    except Exception as e:
        # пользователь мог заблокировать бота — это нормальная ситуация
        print(f"Не удалось уведомить пользователя {user_id}: {e}")

@dp.callback_query(lambda c: c.data in ("approve", "reject", "revision"))
async def process_decision(callback: CallbackQuery):
    if not is_admin(callback.from_user.id):
        await callback.answer("Недостаточно прав.", show_alert=True)
        return

    action = callback.data
    status_map = {"approve": ("approved", "Принято"),
                  "reject": ("rejected", "Отклонено"),
                  "revision": ("revision", "На доработку")}
    new_status, label = status_map[action]

    async with aiosqlite.connect("bot.db") as db:
        await db.execute(
            "UPDATE applications SET status = ? WHERE admin_message_id = ?",
            (new_status, callback.message.message_id)
        )
        await db.commit()
        cursor = await db.execute(
            "SELECT user_id FROM applications WHERE admin_message_id = ?",
            (callback.message.message_id,)
        )
        row = await cursor.fetchone()

    await update_admin_message(callback, label)
    if row:
        user_text = {
            "approved": "Твоя заявка принята. Скоро свяжемся.",
            "rejected": "Заявка отклонена. Если остались вопросы — напиши нам.",
            "revision": "Заявка отправлена на доработку."
        }[new_status]
        await notify_user(row[0], user_text)
    await callback.answer()

@dp.message(Command("queue"))
async def cmd_queue(message: types.Message):
    if not is_admin(message.from_user.id):
        return
    async with aiosqlite.connect("bot.db") as db:
        cursor = await db.execute(
            "SELECT id, user_id, created_at FROM applications "
            "WHERE status = 'pending' ORDER BY id"
        )
        rows = await cursor.fetchall()
    if not rows:
        await message.answer("Очередь пуста.")
        return
    lines = [f"#{r[0]} — user {r[1]} — {r[2]}" for r in rows]
    await message.answer("Очередь:\n" + "\n".join(lines))

@dp.message(Command("stats"))
async def cmd_stats(message: types.Message):
    if not is_admin(message.from_user.id):
        return
    async with aiosqlite.connect("bot.db") as db:
        cursor = await db.execute(
            "SELECT status, COUNT(*) FROM applications GROUP BY status"
        )
        rows = await cursor.fetchall()
    lines = [f"{r[0]}: {r[1]}" for r in rows]
    await message.answer("Статистика:\n" + "\n".join(lines) if lines else "Заявок пока нет.")

async def main():
    await init_db()
    await dp.start_polling(bot)

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

Несколько администраторов

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

# Вместо одного ADMIN_CHAT_ID — id группы, куда бот уже добавлен
ADMIN_CHAT_ID = -1001234567890  # id группы (с минусом и 100 в начале)

# В обработчике решения — проверяем, что кликнул именно участник группы.
# Участники группы определяются по chat.type == 'supergroup' у самого сообщения.
async def is_group_member(user_id: int) -> bool:
    try:
        member = await bot.get_chat_member(ADMIN_CHAT_ID, user_id)
        return member.status in ("member", "administrator", "creator")
    except Exception:
        return False

@dp.callback_query(lambda c: c.data in ("approve", "reject", "revision"))
async def process_decision(callback: CallbackQuery):
    # callback.message.chat.id == ADMIN_CHAT_ID — клик пришёл из правильного места
    if callback.message.chat.id != ADMIN_CHAT_ID or not await is_group_member(callback.from_user.id):
        await callback.answer("Недостаточно прав.", show_alert=True)
        return
    # ... остальная логика та же

**Важно: **сделайте группу приватной и добавьте бот администратором с правом удалять сообщения. Иначе случайные участники увидят персональные данные заявок.

Вариант на Node.js (Grammy)

Если стек на JavaScript — аналогичная архитектура собирается на Grammy. Логика та же: FSM-сценарий на сессии, callback_query для админа, статусы в БД.

import { Bot, InlineKeyboard, session } from "grammy";
import { SQLiteAdapter } from "@grammyjs/storage-sqlite";
import Database from "better-sqlite3";

const bot = new Bot(process.env.BOT_TOKEN);
const db = new Database("bot.db");
const ADMIN_CHAT_ID = Number(process.env.ADMIN_CHAT_ID);

db.exec(`
  CREATE TABLE IF NOT EXISTS applications (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    admin_message_id INTEGER,
    reason TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  );
`);

const storage = new SQLiteAdapter(db);
bot.use(session({ initial: () => ({}), storage }));

bot.command("start", async (ctx) => {
  ctx.session.step = "name";
  await ctx.reply("Привет! Как тебя зовут?");
});

bot.on("message:text", async (ctx) => {
  if (!ctx.session.step) return;
  if (ctx.session.step === "name") {
    ctx.session.name = ctx.message.text;
    ctx.session.step = "description";
    await ctx.reply("Расскажи кратко о себе.");
  } else if (ctx.session.step === "description") {
    ctx.session.description = ctx.message.text;
    ctx.session.step = "contact";
    await ctx.reply("Оставь контакт.");
  } else if (ctx.session.step === "contact") {
    ctx.session.contact = ctx.message.text;
    const payload = JSON.stringify(ctx.session);
    const kb = new InlineKeyboard()
      .text("Принять", "approve").text("Отклонить", "reject")
      .row().text("На доработку", "revision");
    const text = `Новая заявка от @${ctx.from.username || ctx.from.id}:\n\n`
      + `Имя: ${ctx.session.name}\nО себе: ${ctx.session.description}\nКонтакт: ${ctx.session.contact}`;
    const sent = await ctx.api.sendMessage(ADMIN_CHAT_ID, text, { reply_markup: kb });
    db.prepare("INSERT INTO applications (user_id, payload, admin_message_id) VALUES (?, ?, ?)")
      .run(ctx.from.id, payload, sent.message_id);
    ctx.session = {};
    await ctx.reply("Заявка отправлена на модерацию.");
  }
});

bot.on("callback_query:data", async (ctx) => {
  if (ctx.from.id !== ADMIN_CHAT_ID) {
    await ctx.answerCallbackQuery({ text: "Недостаточно прав.", show_alert: true });
    return;
  }
  const map = { approve: ["approved", "Принято"], reject: ["rejected", "Отклонено"], revision: ["revision", "На доработку"] };
  const [newStatus, label] = map[ctx.callbackQuery.data];
  db.prepare("UPDATE applications SET status = ? WHERE admin_message_id = ?")
    .run(newStatus, ctx.callbackQuery.message.message_id);
  const row = db.prepare("SELECT user_id FROM applications WHERE admin_message_id = ?")
    .get(ctx.callbackQuery.message.message_id);
  await ctx.editMessageText(ctx.callbackQuery.message.text + `\n\nРешение: ${label}`);
  if (row) {
    try {
      const t = { approved: "Заявка принята.", rejected: "Заявка отклонена.", revision: "Заявка на доработке." }[newStatus];
      await ctx.api.sendMessage(row.user_id, t);
    } catch (e) { /* пользователь мог заблокировать бота */ }
  }
  await ctx.answerCallbackQuery();
});

bot.start();

Публикация после одобрения

После смены статуса на approved запускается ваша бизнес-логика. Именно тут бот перестаёт быть модератором и становится исполнителем. Три типовых сценария:

Добавление в каталог — публикация в канале или общем чате. Самый простой путь: одно сообщение в канал с кнопкой «Открыть».

CATALOG_CHANNEL_ID = -1002000000000  # канал для публикаций

async def publish_to_catalog(payload: dict):
    text = f"Новая публикация: {payload['name']}\n\n{payload['description']}"
    await bot.send_message(CATALOG_CHANNEL_ID, text)

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

PRIVATE_CHANNEL_ID = -1002000000001

async def grant_access(user_id: int):
    invite = await bot.create_chat_invite_link(
        PRIVATE_CHANNEL_ID,
        member_limit=1,
        expire_date=None  # или timestamp для ограничения по времени
    )
    await bot.send_message(user_id, f"Доступ открыт. Ссылка: {invite.invite_link}")

Запись во внешнюю систему — webhook в CRM, Google Sheets, Airtable или собственный API. Хорошо ложится в очередь задач, чтобы не блокировать обработчик кнопки.

import aiohttp

async def push_to_crm(payload: dict, app_id: int):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://crm.example.com/api/leads",
            json={"name": payload["name"], "contact": payload["contact"], "source": "telegram-bot", "external_id": app_id},
            headers={"Authorization": "Bearer <crm-token>"}
        ) as resp:
            return await resp.json()

# В обработчике process_decision после смены статуса:
# if new_status == "approved":
#     asyncio.create_task(push_to_crm(json.loads(payload), app_id))

Промпты для вайбкодинга

Готовые задачи для агента-кодера (Claude Code, Cursor, Codex). Архитектура стандартная — паттерн собирается за один хороший промпт.

Базовая система с нуля

Собери Telegram-бота на aiogram 3 с очередью модерации. Требования: (1) FSM-сценарий из 3 шагов — имя, описание, контакт. (2) SQLite-таблица applications с полями id, user_id, payload, status (pending/approved/rejected/revision), admin_message_id, reason, created_at. Заявки никогда не удалять, только менять статус. (3) Inline-кнопки под сообщением админа: Принять / Отклонить / На доработку. После нажатия — edit_text с новым статусом и reply_markup=None. (4) Проверка прав админа в каждом callback_query, не только в UI. (5) try/except на bot.send_message пользователю — он мог заблокировать бота. (6) Команды /queue и /stats для админа. Дай рабочий код, без воды.

Добавить команды администратора

В существующий бот добавь админские команды с проверкой прав. /queue — список всех заявок со статусом pending, отсортированных по id. /stats — количество заявок по каждому статусу. /get \<id\> — полный текст заявки с payload. /reject \<id\> \<причина\> — принудительно отклонить с указанием причины в reason. Все команды должны быть недоступны не-админу. Сделай миграцию, если меняется схема БД.

Добавить публикацию в канал

После одобрения заявки (status → approved) опубликуй её в Telegram-канале @catalog_channel. Сообщение: имя автора, описание, контакт. Под сообщением — inline-кнопка «Связаться» с url на пользователя (tg://user?id={user_id}). Публикацию делай через asyncio.create_task, чтобы не блокировать callback_query. Логируй результат публикации (message_id канала) в новую колонку posted_message_id.