surfaces/docs/research/telegram-forum-topics.md

Research: aiogram 3.x Forum Topics API

Based on: Telegram Bot API 7.0+, aiogram 3.x docs, GitHub examples.

Создание супергруппы и тем

Ключевое ограничение: бот не может создать группу сам

Telegram Bot API не позволяет боту программно создавать группы. Пользователь должен:

1. Создать супергруппу вручную
2. Добавить бота как администратора
3. Включить Topics (ForumTopic feature)
4. Переслать сообщение из группы боту — бот получит chat_id

Это фундаментальное ограничение. Флоу адаптируется так: бот просит пользователя создать группу, пользователь пересылает любое сообщение оттуда.

Проверка прав бота в группе

from aiogram import Bot
from aiogram.exceptions import TelegramBadRequest

async def check_bot_admin_rights(bot: Bot, chat_id: int) -> bool:
    """Проверяет что бот является администратором с правом управления темами."""
    try:
        member = await bot.get_chat_member(chat_id, (await bot.get_me()).id)
        return (
            member.status in ("administrator", "creator")
            and getattr(member, "can_manage_topics", False)
        )
    except TelegramBadRequest:
        return False

---

Управление темами (Forum Topics)

Создание темы

from aiogram.types import ForumTopic

async def create_chat_topic(bot: Bot, group_id: int, name: str) -> int:
    """
    Создаёт тему в Forum-группе.
    Возвращает message_thread_id (используется для отправки сообщений в тему).
    """
    topic: ForumTopic = await bot.create_forum_topic(
        chat_id=group_id,
        name=name,  # до 128 символов
        # icon_color — опционально (7322096, 16766590, 13338331, 9367192, 16749490, 16478047)
    )
    return topic.message_thread_id


# Пример: создать "Чат 1"
# thread_id = await create_chat_topic(bot, group_id, "Чат 1")

Переименование, закрытие, удаление

async def rename_topic(bot: Bot, group_id: int, thread_id: int, new_name: str) -> None:
    await bot.edit_forum_topic(
        chat_id=group_id,
        message_thread_id=thread_id,
        name=new_name,
    )


async def archive_topic(bot: Bot, group_id: int, thread_id: int) -> None:
    """Закрывает тему (архивирует). Пользователи не могут писать в закрытую тему."""
    await bot.close_forum_topic(
        chat_id=group_id,
        message_thread_id=thread_id,
    )


async def reopen_topic(bot: Bot, group_id: int, thread_id: int) -> None:
    await bot.reopen_forum_topic(
        chat_id=group_id,
        message_thread_id=thread_id,
    )


async def delete_topic(bot: Bot, group_id: int, thread_id: int) -> None:
    """Удаляет тему и все её сообщения. Необратимо."""
    await bot.delete_forum_topic(
        chat_id=group_id,
        message_thread_id=thread_id,
    )

---

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

Отправить сообщение в конкретную тему

async def send_to_topic(
    bot: Bot,
    group_id: int,
    thread_id: int,
    text: str,
) -> None:
    await bot.send_message(
        chat_id=group_id,
        message_thread_id=thread_id,  # ключевой параметр
        text=text,
    )

Router фильтр для сообщений из конкретной темы

from aiogram import Router, F
from aiogram.types import Message

chat_router = Router()

# Слушать ВСЕ сообщения в темах (кроме General)
@chat_router.message(F.message_thread_id.is_not(None))
async def on_topic_message(message: Message) -> None:
    thread_id = message.message_thread_id
    group_id = message.chat.id
    # ... обработка

Подводный камень #1: message_thread_id может быть None

Сообщения в «General» теме (или если Topics выключены) имеют message_thread_id = None.

@router.message()
async def handler(message: Message):
    if message.message_thread_id is None:
        # Сообщение в General или не в Forum-группе
        return

    # Сообщение в конкретной теме
    thread_id = message.message_thread_id

---

FSM паттерны для онбординга

Многошаговый онбординг

from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message

class OnboardingState(StatesGroup):
    waiting_for_group = State()   # ждём пересылку из группы
    setting_up = State()          # настраиваем группу

onboarding_router = Router()


@onboarding_router.message(Command("start"))
async def cmd_start(message: Message, state: FSMContext) -> None:
    current_state = await state.get_state()
    if current_state is not None:
        # Пользователь уже в процессе — напоминаем
        await message.answer("Ты уже в процессе настройки. Перешли сообщение из группы.")
        return

    await state.set_state(OnboardingState.waiting_for_group)
    await message.answer(
        "Привет! Для начала создай группу в Telegram:\n"
        "1. Создай супергруппу\n"
        "2. Добавь меня как администратора\n"
        "3. Перешли мне любое сообщение из этой группы"
    )


@onboarding_router.message(
    OnboardingState.waiting_for_group,
    F.forward_from_chat.type == "supergroup",
)
async def handle_group_forward(message: Message, state: FSMContext, bot: Bot) -> None:
    group_id = message.forward_from_chat.id

    # Проверяем права
    if not await check_bot_admin_rights(bot, group_id):
        await message.answer(
            "Не могу управлять темами. Убедись что:\n"
            "- Я администратор группы\n"
            "- У меня есть право управлять темами"
        )
        return

    await state.update_data(group_id=group_id)
    await state.set_state(OnboardingState.setting_up)

    # Создаём первую тему
    thread_id = await create_chat_topic(bot, group_id, "Чат 1")
    await send_to_topic(bot, group_id, thread_id, "Привет! Я готов. Пиши здесь.")

    await state.update_data(chat_1_thread_id=thread_id)
    await state.clear()  # онбординг завершён

    await message.answer("✅ Всё готово! Пиши в Чат 1.")

Передача данных между шагами (StateData)

# Сохранять в StateData только то, что нужно в следующем шаге
# НЕ дублировать данные из БД

# Хорошо:
await state.update_data(group_id=group_id, pending_name="Анализ")

# Плохо: хранить большие объекты или данные которые уже в БД
await state.update_data(user_profile=big_dict)  # избыточно

# Читать:
data = await state.get_data()
group_id = data["group_id"]

FSM Storage — выбор хранилища

from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.storage.redis import RedisStorage

# MVP (разработка): данные теряются при перезапуске
storage = MemoryStorage()

# Продакшн: данные переживают перезапуски
from redis.asyncio import Redis
storage = RedisStorage(redis=Redis(host="localhost", port=6379))

# Альтернатива без Redis (простой персистентный вариант):
# pip install aiogram-sqlite-storage
from aiogram_sqlite_storage.sqlitestore import SQLStorage
storage = SQLStorage(db_path="fsm_states.db")

dp = Dispatcher(storage=storage)

Подводный камень #2: MemoryStorage теряет данные при перезапуске

Если бот упал в середине онбординга — пользователь застрянет. Используй SQLiteStorage для MVP.

Обработка timeout состояния

import asyncio
from datetime import datetime, timedelta

# Способ 1: пассивный — при следующем обращении проверяем дату
@router.message()
async def any_handler(message: Message, state: FSMContext) -> None:
    data = await state.get_data()
    started_at = data.get("started_at")

    if started_at:
        elapsed = datetime.now() - datetime.fromisoformat(started_at)
        if elapsed > timedelta(minutes=10):
            await state.clear()
            await message.answer("Сессия истекла. Начни заново: /start")
            return

    # ... продолжение обработки


# Способ 2: при старте шага сохранять timestamp
await state.update_data(started_at=datetime.now().isoformat())

---

Подводные камни (резюме)

Проблема	Решение
Бот не может создать группу	Пользователь создаёт, пересылает сообщение боту
message_thread_id is None	General тема или не Forum-группа — проверяй всегда
Бот не администратор	check_bot_admin_rights() перед операциями
FSM теряется при рестарте	Используй SQLiteStorage вместо MemoryStorage
Пользователь бросил онбординг	Сохраняй started_at, проверяй timeout
StateData разрастается	Хранить только параметры текущего шага

---

Конкурентный анализ UX

Бот 1: ChatGPT (@ChatGPT официальный)

Хорошо:

• Простой /start без лишних шагов
• Typing indicator во время генерации
• Кнопки «Regenerate» и «Edit» под ответом

Плохо:

• Нет разделения на чаты (одна сплошная лента)
• Нет управления историей

Паттерны для нас: typing indicator обязателен; кнопки под сообщением вместо команд

---

Бот 2: Notion AI бот

Хорошо:

• Онбординг в 3 шага с прогресс-баром
• Inline кнопки вместо команд
• Явный feedback: «✅ Готово», «❌ Ошибка»

Плохо:

• Много текста в одном сообщении

Паттерны для нас: пошаговый онбординг, явный feedback

---

Бот 3: TaskMaster AI

Хорошо:

• Отдельный «чат» на каждую задачу (аналог наших Topics)
• /list показывает кнопки по одной задаче на строку
• Подтверждение деструктивных действий: «Удалить чат? [Да] [Нет]»

Паттерны для нас: подтверждение перед удалением/архивацией

---

Паттерны для Lambda Surfaces

Паттерн	Источник	Применение
Typing indicator	ChatGPT	Всегда при запросе к платформе
Inline кнопки	Notion, TaskMaster	/new, /chats, настройки
Пошаговый онбординг	Notion	AuthPending → GroupSetup → Idle
Явный feedback	Notion	«✅ Чат создан», «❌ Ошибка»
Подтверждение удаления	TaskMaster	/archive и /delete
Прогресс в треде	общий паттерн	Долгие задачи агента

Анти-паттерны

• Молчаливая обработка — всегда показывай «обрабатываю...» если > 2 сек
• Команды вместо кнопок — пользователи не помнят команды, используй кнопки
• Все настройки в одном сообщении — разбивай на подменю
• Нет подтверждения — всегда спрашивай перед деструктивными действиями