Создание приложений на FastAPI. Часть четвертая: Интеграция с PTB 21.6

Telegram-боты представляют собой инструмент для автоматизации взаимодействия с пользователями в рамках одного из самых популярных мессенджеров. В основе их работы лежит получение данных (обновлений) от пользователей, которые могут включать сообщения, команды, документы и другие типы данных. Telegram-бот может получать эти обновления двумя основными способами: через polling и webhook.

Polling vs Webhook: В чем разница?

Polling (опрашивание) — это механизм, при котором бот периодически отправляет запросы к серверам Telegram для получения новых данных. В этом случае бот как бы «подключается» к серверу и проверяет, не появилось ли новых сообщений, которые ему необходимо обработать. Хотя такой метод достаточно прост в реализации, он имеет ряд недостатков:

• Частые запросы к API Telegram могут привести к задержкам при получении сообщений и увеличению нагрузки на сервер.
• Неэффективное использование ресурсов: бот должен постоянно запрашивать сервер, даже если нет новых данных, что делает его менее производительным.
• Ограниченная масштабируемость: при большом числе пользователей опрашивание может значительно увеличить нагрузку на сервер, что потребует больше ресурсов для его поддержки.

В противоположность этому, webhook — это механизм, при котором сервер Telegram отправляет данные боту сразу, как только появляются новые обновления. Бот при этом не должен запрашивать данные — он ожидает «входящего» запроса. Важными характеристиками webhook являются:

• Мгновенная доставка: обновления отправляются моментально, что улучшает отзывчивость бота.
• Экономия ресурсов: так как бот получает обновления только в случае их появления, серверу бота не требуется поддерживать постоянные запросы к API Telegram.
• Асинхронность: webhook идеально подходит для приложений, работающих на асинхронных фреймворках, таких как FastAPI, благодаря своей модели событийного взаимодействия. Это делает его более производительным и отзывчивым при большом количестве пользователей.

Почему webhook более подходящий для FastAPI?

FastAPI — это асинхронный веб-фреймворк, созданный для высокопроизводительных приложений. Он работает на базе ASGI (Asynchronous Server Gateway Interface), что позволяет ему обрабатывать запросы и выполнять операции без блокировки основного потока. Асинхронные фреймворки, такие как FastAPI, идеально сочетаются с моделью webhook по следующим причинам:

1. Эффективное использование асинхронности. Webhook, в отличие от polling, не требует постоянного запроса к серверу. FastAPI может обрабатывать обновления в асинхронном режиме, что позволяет выполнять параллельные задачи без блокировки.
2. Масштабируемость и производительность. FastAPI поддерживает асинхронную обработку множества подключений, что особенно важно для высоконагруженных систем, таких как Telegram-боты, обрабатывающие множество сообщений в реальном времени. В сочетании с webhook, который мгновенно отправляет обновления на сервер, это обеспечивает быстрое реагирование и высокую производительность приложения.
3. Меньшая задержка в обработке данных. В то время как polling подразумевает периодическое обращение к API Telegram с возможными задержками между запросами, webhook позволяет серверу FastAPI мгновенно получать новые сообщения, что значительно ускоряет процесс обработки событий.
4. Низкие эксплуатационные затраты. Так как при использовании webhook сервер не должен постоянно поддерживать запросы к API Telegram, это снижает количество необходимых вычислительных ресурсов и упрощает обслуживание. Это особенно важно для облачных решений и при работе с ограниченными ресурсами.

Работа Telegram-бота с webhook в FastAPI состоит из нескольких основных шагов:

1. Регистрация вебхука. При запуске бота он должен зарегистрировать свой вебхук на сервере Telegram, передав ему URL, по которому будут поступать обновления. Этот URL должен указывать на эндпоинт в FastAPI, который будет обрабатывать входящие запросы.
2. Обработка обновлений. Когда пользователь отправляет сообщение или команду боту, Telegram направляет запрос на зарегистрированный URL вебхука, содержащий все необходимые данные. FastAPI принимает этот запрос и асинхронно обрабатывает его, направляя команду на соответствующий обработчик.
3. Ответ на обновление. После обработки запроса бот может отправить ответ пользователю (например, сообщение с подтверждением выполнения команды), используя методы API Telegram. Все это происходит асинхронно, обеспечивая высокую производительность и минимальную задержку.

Использование webhook для Telegram-ботов также требует соблюдения ряда мер безопасности:

• HTTPS-соединение. Для передачи обновлений через webhook требуется защищенное HTTPS-соединение. Это необходимо для предотвращения перехвата данных между сервером Telegram и ботом.
• Валидация запросов. Сервер должен проверять, что запросы поступают именно от Telegram, а не от сторонних источников. Это можно сделать с помощью валидации данных и проверки исходных IP-адресов.
• Секретные URL. Один из лучших способов защиты webhook заключается в использовании секретных URL для эндпоинтов вебхука, что затрудняет доступ к ним неавторизованным пользователям.

Использование webhook для взаимодействия Telegram-бота с FastAPI — это оптимальный выбор для создания высокопроизводительных и масштабируемых приложений. Благодаря асинхронной природе FastAPI и мгновенной доставке обновлений через webhook, такие системы могут эффективно обрабатывать большие объемы данных, предоставляя пользователям быстрый отклик и высокую надежность.

Когда Telegram-бот работает на вебхуках, сервер FastAPI должен быть готов к обработке входящих запросов от Telegram. Этот процесс можно разбить на несколько шагов, начиная с инициализации бота и установки вебхука и заканчивая асинхронной обработкой событий.

Ниже представлена реализация основного кода для инициализации FastAPI-приложения с жизненным циклом приложения (lifespan), который управляет интеграцией с такими внешними сервисами, как Redis и Taskiq, а также обработкой запросов от Telegram.

В жизненном цикле приложения с использованием lifespan FastAPI инициализирует подключение к Redis, кеширование и брокер задач Taskiq, а также настраивает Telegram-бота и обрабатывает его старт и остановку. Всё это в рамках продолжения цикла статей по шаблону FastAPI, для инициации бота Redis и Taskiq не нужен.

main.py:

from fastapi import FastAPI
from contextlib import asynccontextmanager
from typing import AsyncIterator
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from src.tgbot.loader import application
from src.conf import taskiq_broker, set_async_redis_client

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Инициализация Redis клиента для кеширования
    redis_client = await set_async_redis_client()
    FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")
    await FastAPICache.clear()

    # Запуск Taskiq брокера задач, если это не рабочий процесс
    if not taskiq_broker.is_worker_process:
        await taskiq_broker.startup()

    # Настройка обработчиков Telegram-бота
    await setup_handlers(application)
    
    # Запуск Telegram-бота в асинхронном режиме
    async with application:
        await application.start()  # Запуск бота
        yield
        await application.stop()   # Остановка бота

    # Остановка Taskiq брокера
    if not taskiq_broker.is_worker_process:
        await taskiq_broker.shutdown()

# Инициализация FastAPI с указанием lifespan
app = FastAPI(lifespan=lifespan)

1. Асинхронный жизненный цикл (lifespan): FastAPI поддерживает асинхронную инициализацию и завершение компонентов. В данном случае мы инициализируем Redis и Taskiq перед началом работы приложения и останавливаем их в конце.
2. Redis для кеширования: Redis используется для кеширования данных приложения. Кеширование помогает сократить нагрузку на базу данных и ускорить доступ к часто используемым данным.
3. Taskiq: Этот компонент отвечает за асинхронное выполнение задач. Он запускается только тогда, когда FastAPI не является рабочим процессом Taskiq. Это предотвращает конфликты при одновременной работе с задачами.
4. Настройка Telegram-бота: Бот запускается и останавливается в асинхронном контексте. Обработчики сообщений и команд, описанные далее, настраиваются через функцию setup_handlers.
5. Запуск и остановка бота: При старте приложения FastAPI Telegram-бот запускается вместе с ним. Когда приложение закрывается, бот также корректно останавливается, освобождая все ресурсы.

Теперь рассмотрим часть, касающуюся инициализации самого Telegram-бота и его подключения к FastAPI через вебхуки. Важно, чтобы бот был правильно настроен для работы с webhook и мог обрабатывать обновления от пользователей через асинхронные запросы.

tgbot/loader.py:

from warnings import filterwarnings
from telegram import Bot
from telegram.ext import Application
from telegram.warnings import PTBUserWarning
from src.conf import settings, logger

# Игнорирование некоторых предупреждений от библиотеки Telegram
filterwarnings(action="ignore", message=r".*CallbackQueryHandler", category=PTBUserWarning)

# Инициализация Telegram-бота с использованием токена
bot = Bot(token=settings.TELEGRAM_TOKEN)

# Создание асинхронного приложения Telegram
application = (
    Application.builder()
    .updater(None)  # Используем Webhook вместо Updater
    .token(settings.TELEGRAM_TOKEN)
    .read_timeout(7)  # Таймаут для чтения данных
    .get_updates_read_timeout(42)  # Таймаут для получения обновлений
    .build()
)

async def set_webhook():
    """Устанавливаем Webhook для бота на сервере Telegram"""
    url = f"https://{settings.DOMAIN}/bot/{settings.SECRET_BOT_URL}/webhooks/"
    try:
        response = await bot.set_webhook(url=url)
        logger.info(f"Webhook установлен: {url}, ответ сервера: {response}")
        return True
    except Exception as e:
        logger.error(f"Ошибка при установке вебхука: {e}")
        return False

1. Инициализация бота: В начале создается объект Bot, используя предоставленный токен бота из конфигурации. Этот токен обеспечивает доступ боту к Telegram API.
2. Создание приложения Telegram: Через Application.builder() инициализируется асинхронное приложение, которое будет работать на вебхуках. Опция updater(None) указывает на использование вебхуков вместо традиционного метода polling для получения обновлений.
3. Настройка webhook: Метод set_webhook() настраивает URL вебхука для обработки обновлений от Telegram. Этот URL должен указывать на публичный адрес FastAPI-приложения. Таким образом, каждый раз, когда пользователь отправляет сообщение или команду, Telegram направляет это событие на указанный URL.

Современные боты, особенно при работе с большими аудиториями и сложной логикой, требуют быстрого и отзывчивого подхода к обработке запросов. Одним из ключевых преимуществ FastAPI является его асинхронная природа, которая идеально подходит для интеграции с Telegram-ботами. Асинхронные задачи позволяют распределять нагрузку на сервер, снижать задержки в обработке запросов и обеспечивать высокую производительность. Это особенно важно для долгосрочных операций или тех, которые требуют взаимодействия с внешними API, базами данных или другими сервисами. Основными преимуществами асинхронных задач в контексте Telegram-ботов являются:

1. Параллелизация задач без блокировки основного потока. Асинхронная модель FastAPI позволяет эффективно управлять задачами, не блокируя основной поток, что делает возможной параллельную обработку множества сообщений. Это особенно важно в сценариях, где бот работает с тысячами пользователей одновременно.
2. Снижение задержек. В асинхронных задачах мы можем выполнять несколько операций одновременно, например, отправлять запросы в базу данных, удалять старые сообщения, работать с Redis и выполнять другие задачи в фоновом режиме, что значительно ускоряет ответ пользователю.
3. Долгосрочные операции. Для Telegram-ботов асинхронность особенно полезна в задачах, которые требуют длительного времени на выполнение, таких как взаимодействие с внешними сервисами, выполнение операций с большими данными или планирование задач на будущее.

Рассмотрим следующий пример функции для удаления команд из чата. Благодаря асинхронности эта операция выполняется быстро и эффективно, освобождая основной поток для обработки других запросов.

tgbot/services/cleaner.py:

import asyncio

from src.utils.re_compile import COMMAND_PATTERN
from telegram import Update
from telegram.error import TelegramError

from ..loader import bot, logger
from .patterns import COMMANDS

async def clear_commands(update: Update) -> None:
    """Удаление команд бота из чата."""
    try:
        chat_id = update.message.chat.id
        message_id = update.message.message_id
        message = update.message
        if not message:
            return

        if message.location or message.contact:
            await bot.delete_message(chat_id, message_id)
            return

        command = COMMAND_PATTERN.findall(message.text)
        if command and command[0] in COMMANDS:
            await bot.delete_message(chat_id, message_id)

    except TelegramError:
        text = (
            'Для корректной работы, я должен быть администратором группы! '
            'Иначе я не смогу удалять технические сообщения.'
        )
        await bot.send_message(chat_id, text)

• Асинхронное удаление сообщений: Функция clear_commands асинхронно удаляет команды из чата, если они соответствуют определённому паттерну. Это полезно в групповых чатах, где бот должен очищать технические сообщения, чтобы не загромождать чат.
• Асинхронная обработка ошибок: Если бот не имеет достаточных прав, он асинхронно отправляет сообщение об ошибке, не блокируя основной поток.

В более сложных случаях асинхронная обработка позволяет запускать несколько задач одновременно. Это может быть полезно для удаления сообщений и управления данными в Redis параллельно. Рассмотрим пример функции, которая удаляет клавиатуры и сообщения одновременно.

import asyncio
from telegram import Update
from telegram.ext import ContextTypes
from .redis_crud import RedisCRUDManager
from ..loader import bot, logger
from .patterns import KEYBOARD

async def remove_incoming_message(update: Update, context: ContextTypes.DEFAULT_TYPE, message_id: str = None) -> None:
    """
    Удаление клавиатуры после нажатия.
    ### Args:
    - update (:obj:`Update`)
    - context (:obj:`ContextTypes.DEFAULT_TYPE`)
    - message_id (:obj:`str`): optionals
    """
    chat_id = update.effective_chat.id
    del_menu_id = message_id if message_id else update.effective_message.message_id
    
    try:
        # Удаление сообщения и параллельное удаление данных из Redis
        await context.bot.delete_message(chat_id, del_menu_id)
        keyboard_manager = RedisCRUDManager(pattern=KEYBOARD)
        asyncio.gather(
            keyboard_manager.delete_messages_for_chat_id(chat_id),
            delete_scheduled_job(context, f'{KEYBOARD}:{chat_id}:{del_menu_id}')
        )
    except Exception as error:
        logger.error(error)

• Параллельная обработка задач: С помощью asyncio.gather мы запускаем удаление сообщений в чате и очистку Redis от соответствующих данных одновременно, что существенно ускоряет выполнение обеих операций.
• Асинхронная работа с Redis: Redis используется для хранения временных данных, таких как идентификаторы сообщений и клавиатур. Благодаря асинхронности работа с Redis не блокирует другие задачи.

Асинхронные задачи также важны для планирования долгосрочных операций. Например, бот может удалять сообщения через определённое время после их отправки. В этом случае asyncio.gather помогает запускать несколько таких задач параллельно.

async def delete_inline_keyboards_or_messages_by_ids(chat_id, message_ids, context):
    for message_id in message_ids:
        try:
            # Параллельное удаление сообщений и задач
            asyncio.gather(
                context.bot.delete_message(chat_id=chat_id, message_id=message_id),
                delete_scheduled_job(context, f'{KEYBOARD}:{message_id}')
            )
        except Exception as e:
            logger.error(f"Не удалось удалить сообщение: {e}")

• Удаление нескольких сообщений: Эта функция удаляет несколько сообщений одновременно, используя асинхронные задачи для каждого сообщения.
• Планирование задач: delete_scheduled_job используется для удаления запланированных задач из очереди, что может быть полезно для операций с долгосрочным ожиданием.

Асинхронные задачи в Telegram-ботах, работающих на FastAPI, значительно повышают их производительность и отзывчивость. Возможность параллельной обработки запросов и долгосрочных операций позволяет улучшить работу бота с большими группами пользователей и сложными взаимодействиями, делая приложение более масштабируемым и эффективным.

Создание масштабируемого и гибкого проекта для работы с Telegram-ботом требует тщательно продуманной архитектуры файлов и папок. Структура проекта не только помогает организовать код, но и упрощает его поддержку, тестирование и дальнейшее расширение. Рассмотрим ключевые директории и файлы, используемые в проекте FastAPI, и поясним их предназначение.

Проект организован по модульному принципу, где каждая часть кода (такие как обработчики запросов, сервисы и работа с данными) выделена в отдельные папки. Такая организация облегчает поддержку кода, поскольку каждый модуль выполняет свою специфическую задачу. Вот пример общей структуры:

.
├── backend
│   ├── alembic
│   ├── src
│   │   ├── admin
│   │   ├── auth
│   │   ├── conf
│   │   ├── crud
│   │   ├── db
│   │   ├── media
│   │   ├── models
│   │   ├── schemas
│   │   ├── static
│   │   ├── tasks
│   │   ├── tgbot
│   │   │   ├── handlers
│   │   │   ├── services
│   │   │   ├── dispatcher.py
│   │   │   ├── loader.py
│   │   │   └── routers.py
│   │   ├── users
│   │   ├── utils
│   │   ├── __init__.py
│   │   ├── main.py
│   │   └── routers.py
├── infra
│   ├── nginx
│   ├── debug
│   ├── stage
├── .env
├── Dockerfile
├── requirements.txt
└── README.md

1. backend/src

Это основная папка с кодом приложения. Все ключевые модули и бизнес-логика бота находятся здесь.

• admin/ — модуль для административных функций и управления системой. Включает роуты и обработчики для административной панели или мониторинга.
• auth/ — отвечает за аутентификацию и авторизацию пользователей. Здесь находятся обработчики для входа в систему, регистрации и проверки прав доступа.
• conf/ — конфигурационные файлы проекта. Этот модуль управляет настройками окружения, такими как токен Telegram-бота, параметры базы данных, Redis, настройки брокеров задач и другие важные параметры.
• crud/ — модуль, содержащий CRUD-операции (Create, Read, Update, Delete) для работы с базой данных. Сюда входит взаимодействие с моделями данных через SQLAlchemy или другой ORM.
• db/ — содержит файлы для работы с базой данных, такие как настройки подключения, сессии для асинхронных транзакций и миграции. Это важная часть для управления данными бота.
• media/ — отвечает за работу с медиафайлами, например, за загрузку изображений, аудио и других типов данных. Здесь может находиться логика обработки файлов, сохранения в хранилище (например, S3), а также их выдачи пользователям.
• models/ — модели базы данных, определяющие структуру таблиц. Обычно эти файлы содержат схемы баз данных для ORM, такие как SQLAlchemy или Pydantic-модели.
• schemas/ — описывают схемы данных и структуры запросов и ответов. Используются для валидации данных, которые отправляются или принимаются ботом. Это помогает поддерживать согласованность и точность структуры данных.
• tasks/ — модуль для управления асинхронными задачами, которые могут выполняться в фоне, такими как удаление сообщений, выполнение долгосрочных задач и другие фоновые операции. Здесь обычно подключаются Taskiq и Celery.
• tgbot/ — папка, которая включает все, что связано непосредственно с Telegram-ботом. Она состоит из нескольких подпапок и файлов:

  • handlers/ — модули, отвечающие за обработку различных типов событий и сообщений от пользователей. Здесь содержатся файлы, которые обрабатывают команды, ответы на сообщения и взаимодействия с клавиатурой.

    Пример файла chat_distributor.py:

    from telegram import Update
    from telegram.ext import ContextTypes
    
    async def check_text(update: Update, context: ContextTypes.DEFAULT_TYPE):
        # Обработка текстового запроса пользователя. Его классификация и запуск функции в зависимости от результата классификации.
        pass

  • services/ — это сервисный слой приложения, содержащий бизнес-логику, которая не связана непосредственно с обработкой команд. Например, управление клавиатурами, отправка сообщений, взаимодействие с внешними API.
  • dispatcher.py — отвечает за регистрацию обработчиков сообщений и команд. Это центральный файл, где происходит подключение всех обработчиков, которые затем обрабатывают входящие запросы.
  • loader.py — модуль, где инициализируются ключевые компоненты бота, такие как объект бота, приложение Telegram, а также регистрация вебхуков. Этот файл используется для загрузки всех основных компонентов бота.
  • routers.py — модуль, в котором описываются маршруты вебхуков и точки входа для обработки запросов от Telegram.

2. infra/

Папка, содержащая инфраструктурные файлы, такие как конфигурации для Docker и Nginx, а также скрипты для развертывания проекта в разных средах.

• nginx/ — настройки для обратного прокси-сервера Nginx. Nginx обычно используется для перенаправления запросов на FastAPI и работы с SSL-сертификатами для защиты вебхуков.
• debug/ — файлы, необходимые для запуска проекта в режиме разработки. Сюда могут входить файлы для Docker Compose и другие конфигурационные файлы для локальной среды.
• stage/ — файлы для развертывания проекта на промежуточном или продакшен-сервере.

3. .env

Файл с переменными окружения, содержащий конфиденциальные данные, такие как токен бота, URL базы данных, API-ключи и другие настройки.

4. Dockerfile — Файл для создания Docker-образа приложения. Он описывает, как установить зависимости и запустить приложение в контейнере.

5. requirements.txt — Файл с зависимостями Python, необходимыми для работы проекта. Включает библиотеки FastAPI, Telegram Bot API, Redis и другие.

6. README.md — Документация проекта, в которой описывается, как запустить и развернуть приложение, а также краткое описание функциональности бота.

Такая модульная структура проекта позволяет легко управлять масштабируемостью и поддерживаемостью приложения. Разделение на отдельные модули для каждой задачи — обработка команд, бизнес-логика, взаимодействие с базами данных — делает код читаемым и легким для изменения или расширения.

Telegram-бот — это комплексное приложение, где каждая часть системы отвечает за выполнение конкретной задачи. В рамках работы на FastAPI бот получает обновления через вебхуки, обрабатывает сообщения, взаимодействует с Redis для хранения временных данных, а также отправляет сообщения пользователям. В этом разделе мы рассмотрим ключевые компоненты Telegram-бота, включая маршрутизацию вебхуков, обработку сообщений, работу с Redis и реализацию команд.

Маршрутизация вебхуков — это точка входа для Telegram-бота, где FastAPI получает обновления от серверов Telegram в формате JSON. Вебхук автоматически вызывается Telegram, когда пользователи взаимодействуют с ботом, и передает событие на наш сервер для дальнейшей обработки. Этот подход помогает ботам мгновенно реагировать на запросы без необходимости постоянно опрашивать серверы (как в случае с polling).

tgbot/routers.py:

import asyncio

from fastapi import APIRouter, Request, status
from fastapi.responses import JSONResponse
from src.conf import settings
from src.tgbot.loader import application
from src.tgbot.services.cleaner import clear_commands
from telegram import Update

tg_router = APIRouter()

@tg_router.post(f"/{settings.SECRET_BOT_URL}/webhooks/", status_code=status.HTTP_200_OK)
async def receive_update(request: Request):
    """
    Получает обновления от серверов Telegram через вебхук.
    Преобразует входящий JSON в объект Update и отправляет его в очередь на обработку.
    """
    update = Update.de_json(data=await request.json(), bot=application.bot)
    if update.message:
        asyncio.create_task(clear_commands(update))
    await application.update_queue.put(update)
    return JSONResponse(content={"ok": True})

1. Маршрут для вебхуков: Вебхук определен как POST-запрос к URL, который содержит секретный путь. Это предотвращает случайные обращения к эндпоинту от сторонних источников.
2. Обработка обновлений: Когда Telegram отправляет обновления в формате JSON, они преобразуются в объект Update и помещаются в очередь update_queue для дальнейшей асинхронной обработки.
3. Удаление технических сообщений: в  clear_commands, описанной ранее, мы удаляем команды бота в параллельном потоке.
4. Асинхронная обработка: Благодаря асинхронности FastAPI, бот может мгновенно отвечать на запросы, не задерживая другие операции.

Когда обновление попадает в очередь, оно обрабатывается системой команд, которая реагирует на команды и сообщения, отправленные пользователями. Важной частью этого процесса является обработка сервисных сообщений, отправка ответов пользователю и выполнение дополнительных действий (например, удаление сообщений через некоторое время).

tgbot/services/message_handler.py:

import traceback
import telegram
from src.utils.re_compile import NUMBER_BYTE_OFFSET
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
from ..loader import bot
from .cleaner import delete_message

async def send_service_message(context: ContextTypes.DEFAULT_TYPE, chat_id: int, reply_text: str, parse_mode: str = None, message_thread_id: int = None) -> None:
    """
    Отправляет сервисное сообщение в чат и планирует его удаление через 20 секунд.
    """
    message = await bot.send_message(
        chat_id,
        reply_text,
        parse_mode,
        message_thread_id=message_thread_id
    )
    context.job_queue.run_once(
        delete_message,
        20,  # Удалить через 20 секунд
        chat_id=chat_id,
        name=f'send_service_message: {chat_id}',
        data={'chat_id': chat_id, 'message_id': message.message_id}
    )

1. Асинхронная отправка сообщения: Бот отправляет сервисные сообщения пользователям, например, подтверждение выполнения команды. Этот процесс полностью асинхронен, что позволяет боту продолжать работу без ожидания завершения отправки.
2. Запланированное удаление: После отправки сообщения, бот может запланировать его удаление через заданный интервал (в данном случае через 20 секунд) с помощью очереди задач. Это помогает избежать загромождения чатов техническими сообщениями.

Redis используется для хранения временных данных, таких как идентификаторы сообщений, которые нужно удалить, или данные пользователей. Благодаря асинхронной модели Redis, бот может быстро взаимодействовать с данными, не блокируя основные операции.

Пример кода — tgbot/services/redis_crud.py:

import json
from src.conf.redis import AsyncRedisClient

class RedisCRUDManager:
    def __init__(self, pattern):
        self.redis_client = AsyncRedisClient.get_client()
        self.pattern = pattern

    def _get_key(self, chat_id):
        """Возвращает ключ для Redis с заданным chat_id."""
        return f"telbot_chat:{chat_id}:{self.pattern}"

    async def save_message_id(self, chat_id, message_id):
        """Сохраняем message_id в Redis, используя chat_id как ключ"""
        await self.redis_client.sadd(self._get_key(chat_id), message_id)

    async def get_message_ids(self, chat_id):
        """Получаем все message_id для chat_id"""
        return await self.redis_client.smembers(self._get_key(chat_id))

    async def delete_messages_for_chat_id(self, chat_id):
        """Удаляем все сохранённые message_id для chat_id, если они существуют"""
        key = self._get_key(chat_id)
        if await self.redis_client.exists(key):
            await self.redis_client.delete(key)

1. Хранение данных: Redis используется для быстрого сохранения временных данных, таких как идентификаторы сообщений, которые могут потребоваться для последующего удаления. Эти данные можно получить по ключу, основанному на chat_id.
2. Асинхронная работа: Все операции с Redis выполняются асинхронно, что ускоряет взаимодействие с данными и позволяет боту не блокировать другие операции.

Функция отправки сообщений send_message_to_chat предназначена для обработки ситуаций, когда может возникнуть ошибка при отправке сообщения, например, если сообщение слишком длинное или содержит неверные символы. Если сообщение слишком большое для отправки, функция автоматически делит его на части и отправляет по частям. Если ошибка заключается в некорректных символах (например, неверный знак), функция пытается исправить сообщение путем удаления или замены проблемных символов или отправляет оригинальное сообщение без parse_mode.

def find_byte_offset(message):
    """
    Ищет в сообщении числовое значение после 'byte offset'.
    Возвращает найденное значение как целое число или None, если совпадение не найдено.
    """
    match = NUMBER_BYTE_OFFSET.search(message)
    if match:
        return int(match.group(1))
    return None


def delete_at_byte_offset(text, offset):
    """
    Безопасно удаляет символ по указанному смещению байтов из текста в кодировке UTF-8.
    """
    byte_text = text.encode('utf-8')
    if offset < 0 or offset >= len(byte_text):
        return text
    while offset > 0 and (byte_text[offset] & 0xC0) == 0x80:
        offset -= 1
    end_offset = offset + 1
    while end_offset < len(byte_text) and (byte_text[end_offset] & 0xC0) == 0x80:
        end_offset += 1
    modified_byte_text = byte_text[:offset] + byte_text[end_offset:]
    try:
        return modified_byte_text.decode('utf-8')
    except UnicodeDecodeError:
        return text


def split_message(text, max_length=4000):
    """
    Разбивает текст на части, не превышающие max_length символов.
    """
    return [text[i:i + max_length] for i in range(0, len(text), max_length)]


async def send_message_to_chat(chat_id: int, message: str, reply_to_message_id: int = None, parse_mode: ParseMode = None, retry_count: int = 3) -> None:
    """
    Отправляет сообщение через Telegram бота с возможностью исправления и повторной отправки при ошибке.

    ### Args:
    - chat_id (`int`): Идентификатор чата в Telegram.
    - message (`str`): Текст сообщения.
    - reply_to_message_id (`int`, optional): Идентификатор сообщения, на которое нужно ответить.
    - parse_mode (`ParseMode`, optional): Режим парсинга сообщения.
    - retry_count (`int`, optional): Количество попыток отправки при ошибке.

    ### Return:
    - send_message (`telegram.Message`): В случае успеха возвращается отправленное сообщение.

    """
    origin_message = message
    for attempt in range(retry_count, -1, -1):
        try:
            send_message = await bot.send_message(
                chat_id=chat_id,
                text=message,
                parse_mode=parse_mode,
                reply_to_message_id=reply_to_message_id,
            )
            return send_message
        except telegram.error.BadRequest as err:
            if "Message is too long" in str(err):
                messages = split_message(message)
                for msg in messages:
                    await bot.send_message(chat_id=chat_id, text=msg, reply_to_message_id=reply_to_message_id)
            if attempt <= 1:
                message = origin_message
                parse_mode = None
                continue
            error_message = str(err)
            offset = find_byte_offset(error_message)
            if offset is not None:
                message = delete_at_byte_offset(message, offset)

        except Exception as err:
            if attempt == 0:
                traceback_str = traceback.format_exc()
                await bot.send_message(
                    chat_id=settings.TELEGRAM_ADMIN_ID,
                    text=f'Необработанная ошибка в `send_message_to_chat`: {str(err)}\n\nТрассировка:\n{traceback_str[-1024:]}'
                )
                return None

1. Асинхронная отправка сообщения: Функция отправляет сообщение пользователю в чат. Если сообщение слишком длинное, бот автоматически разбивает его на несколько частей и отправляет по частям.
2. Обработка ошибок: В случае возникновения ошибки (например, если сообщение слишком длинное или содержит неверные символы), бот пытается повторить отправку с корректировкой сообщения.

Одной из ключевых задач в работе Telegram-бота является управление пользовательскими данными. Обычно для бота важно знать, зарегистрирован ли пользователь, с какими группами он взаимодействует, и какие данные связаны с его аккаунтом. В FastAPI это достигается через асинхронные запросы к базе данных и кешированию этих данных в Redis.

Для реализации гибкости и производительности используется класс UserRedisManager, который асинхронно сохраняет и извлекает данные пользователей из Redis, а также взаимодействует с базой данных для получения более сложных структур данных.

services/registrar.py:

import json
from typing import List, Optional, Tuple
from sqlalchemy.ext.asyncio import AsyncSession
from src.conf.redis import AsyncRedisClient
from src.crud import tg_group_dao, tg_user_dao, user_ai_model_dao
from src.db.deps import get_async_session
from src.models import TgGroup, TgUser
from src.schemas.tg_user_schema import TgUserCreate

class UserRedisManager:
    def __init__(self):
        self.redis_client = AsyncRedisClient.get_client()

    async def set_user_in_redis(self, tg_user, user: TgUser) -> dict:
        """Сохраняет или обновляет запись пользователя в Redis."""
        redis_user = {
            'user_id': user.user.id if user.user else None,
            'tg_user_id': tg_user.id,
            'groups_connections': [group.id for group in user.groups],
            'is_blocked_bot': user.is_blocked_bot,
        }
        await self.redis_client.set(f"user:{tg_user.id}", json.dumps(redis_user))
        return redis_user

    async def get_or_create_user(self, tg_user, prefetch_related: List[str], session: AsyncSession) -> Tuple[dict, Optional[TgUser]]:
        """
        Возвращает данные пользователя из Redis или создает нового в базе данных.
        Если пользователя нет в Redis, данные запрашиваются из базы данных и кешируются.
        """
        redis_key = f"user:{tg_user.id}"
        redis_user = await self.redis_client.get(redis_key)
        redis_user = json.loads(redis_user) if redis_user else None

        # Если данных в Redis нет, загружаем пользователя из базы данных
        prefetch_related.extend(["groups", "user", "active_model"])
        user = await tg_user_dao.get_with_prefetch_related(id=tg_user.id, prefetch_related=prefetch_related, db_session=session)

        # Если пользователь не найден в базе данных, создаем новую запись
        if not user:
            obj_in = TgUserCreate(
                id=tg_user.id,
                username=tg_user.username or f'n-{str(1010101 + tg_user.id)[::-1]}',
                first_name=tg_user.first_name or tg_user.username,
                last_name=tg_user.last_name
            )
            user = await tg_user_dao.create(obj_in=obj_in, db_session=session, relationship_refresh=prefetch_related)
            await user_ai_model_dao.create_default(user=user, db_session=session)

        # Сохраняем или обновляем данные пользователя в Redis
        redis_user = await self.set_user_in_redis(tg_user, user)
        return redis_user, user

1. Асинхронное взаимодействие с Redis: UserRedisManager используется для сохранения и извлечения данных о пользователях из Redis. Это ускоряет доступ к данным, так как бот не всегда обращается к базе данных напрямую, а может использовать кешированные данные.
2. Гибкое извлечение данных: Функция get_or_create_user проверяет, есть ли пользователь в Redis, и если его нет, делает запрос к базе данных. Если пользователь отсутствует в базе данных, создается новая запись. Это гарантирует, что у бота всегда есть актуальная информация о пользователе.
3. Предзагрузка связанных данных: С помощью параметра prefetch_related загружаются связанные данные пользователя, такие как группы, с которыми он взаимодействует, или связанные модели (например, AI-модели, если они используются в проекте).

Асинхронная проверка регистрации пользователя и взаимодействие с Redis помогает эффективно управлять состоянием пользователей в системе. Команда бота может включать в себя проверку регистрации пользователя, загрузку связанных данных и выполнение дальнейших действий, основываясь на этих данных.

get_user():

async def get_user(tg_user, chat, allow_unregistered: bool = False, prefetch_related: list = []) -> TgUser | None:
    """Проверяет регистрацию пользователя или создает его в базе данных, если необходимо."""
    user_manager = UserRedisManager()

    # Открытие сессии для взаимодействия с базой данных
    async for session in get_async_session():

        # Получаем или создаем пользователя в Redis и базе данных
        red_user, user = await user_manager.get_or_create_user(tg_user, prefetch_related, session)

        # Проверяем, не заблокирован ли пользователь (необходимо в функциях где требуется пройти определенную авторизацию)
        if red_user.get('is_blocked_bot') and not allow_unregistered:
            return None

        # Работа с группами в случае, если чат не является приватным
        if chat.type != 'private':
            group = await tg_group_dao.get_by_chat_id(chat_id=chat['id'], db_session=session)
            if not group:
                obj_in = TgGroup(chat_id=chat.id, title=chat.title, link=chat.link)
                group = await tg_group_dao.create(obj_in=obj_in, db_session=session)

            if user and (group.id not in red_user.get('groups_connections')):
                user = await tg_user_dao.add_group(user=user, group=group, db_session=session)
                await user_manager.set_user_in_redis(tg_user, user)

        return user

1. Проверка регистрации пользователя: При вызове функции бот проверяет, зарегистрирован ли пользователь в базе данных или кеширован в Redis. Если данных нет, создается новая запись.
2. Гибкая работа с группами: Если пользователь взаимодействует с ботом в групповом чате, функция также проверяет, добавлен ли пользователь в группу, и, если нет, добавляет его.
3. Работа с сессией базы данных: Использование get_async_session() позволяет выполнять все запросы к базе данных асинхронно, что важно для поддержания высокой производительности при взаимодействии с большими объемами данных.

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

check_registration():

async def check_registration(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """
    Проверяет регистрацию пользователя в базе данных или Redis перед выполнением команды.
    """
    allow_unregistered = True
    prefetch_related = ['active_model']  # Предзагрузка связанных данных
    return await get_user(
        tg_user=update.effective_user,
        chat=update.effective_chat,
        allow_unregistered=allow_unregistered,
        prefetch_related=prefetch_related,
    )

1. Проверка регистрации: Эта функция вызывается перед выполнением любой команды, чтобы убедиться, что пользователь зарегистрирован в системе. Это гибкий механизм, который позволяет работать как с зарегистрированными, так и с незарегистрированными пользователями, в зависимости от бизнес-логики.
2. Предзагрузка связанных данных: Использование параметра prefetch_related позволяет сразу загружать связанные данные, что снижает количество дополнительных запросов к базе данных.

Асинхронная обработка команд и сообщений в Telegram-боте значительно улучшает производительность и отзывчивость приложения. Возможность параллельного выполнения запросов к базе данных и работы с Redis позволяет эффективно управлять пользовательскими данными и быстрее реагировать на команды. 

Основой функциональности Telegram-бота является обработка входящих сообщений и команд. Чтобы добавить новые команды и обработчики, нужно определить, какие именно события должен обрабатывать бот, и как на них реагировать.

Обработчики в Telegram-ботах могут реагировать на текстовые сообщения, команды, кнопки клавиатуры, мультимедийные файлы и другие события. Для обработки этих событий можно использовать мощные фильтры, предоставляемые библиотекой python-telegram-bot, а также продвинутые паттерны обработки сообщений.

tgbot/dispatcher.py:

from telegram.ext import MessageHandler, filters
from telegram.ext import CommandHandler
from src.tgbot.handlers.chat_distributor import check_request
from src.tgbot.handlers.command_handler import start, help_command, info

async def setup_handlers(dp):
    # Обработка новых команд
    dp.add_handler(CommandHandler("start", start))
    dp.add_handler(CommandHandler("help", help_command))
    dp.add_handler(CommandHandler("info", info))

    # Обработка текстовых сообщений
    dp.add_handler(MessageHandler(filters.TEXT, check_request))
    
    return dp

1. Добавление команды /start: Команда /start — это типичная точка входа для большинства ботов. Она используется для приветствия пользователя и предоставления базовой информации о боте.
2. Команда /help: Команда /help предоставляет список доступных команд и инструкций, помогая пользователям взаимодействовать с ботом.
3. Фильтры для текстовых сообщений: С помощью MessageHandler бот может обрабатывать любые текстовые сообщения, поступающие от пользователей, что позволяет реализовать более сложную логику обработки текстов, включая поиск по ключевым словам или выполнение определенных действий в зависимости от содержимого сообщения.

Добавление новых команд и обработчиков — это базовый уровень расширения функциональности бота. Однако, для создания более сложных и гибких сценариев взаимодействия с пользователями, можно использовать продвинутые паттерны, такие как последовательные сценарии, логические ветвления или даже интеграция с диалоговыми системами на базе AI.

Пример: Обработка последовательных сценариев

В некоторых случаях необходимо построить диалог с пользователем, состоящий из нескольких шагов. Например, бот может последовательно запрашивать информацию у пользователя, ожидая ответов на каждый запрос.

Пример кода — Сценарий регистрации пользователя:

from telegram.ext import ConversationHandler, CommandHandler, MessageHandler, filters

# Состояния сценария
REGISTRATION_NAME, REGISTRATION_AGE = range(2)

async def start_registration(update, context):
    await update.message.reply_text("Как вас зовут?")
    return REGISTRATION_NAME

async def get_name(update, context):
    context.user_data['name'] = update.message.text
    await update.message.reply_text(f"Привет, {context.user_data['name']}! Сколько вам лет?")
    return REGISTRATION_AGE

async def get_age(update, context):
    context.user_data['age'] = update.message.text
    await update.message.reply_text(f"Отлично! Вы {context.user_data['name']}, и вам {context.user_data['age']} лет.")
    return ConversationHandler.END

async def cancel_registration(update, context):
    await update.message.reply_text("Регистрация отменена.")
    return ConversationHandler.END

def setup_conversation_handlers(dp):
    registration_handler = ConversationHandler(
        entry_points=[CommandHandler('register', start_registration)],
        states={
            REGISTRATION_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_name)],
            REGISTRATION_AGE: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_age)]
        },
        fallbacks=[CommandHandler('cancel', cancel_registration)]
    )
    
    dp.add_handler(registration_handler)

1. Последовательные шаги: Сценарий регистрации последовательно запрашивает имя и возраст у пользователя. Используется ConversationHandler, который управляет состояниями и переходами между ними.
2. Логические ветвления: В зависимости от ответа пользователя, бот переходит на следующий шаг сценария. Это позволяет создавать сложные диалоги, где каждое сообщение пользователя может влиять на дальнейший ход разговора.
3. Команда отмены: С помощью команды /cancel пользователь может в любой момент выйти из сценария, что обеспечивает гибкость взаимодействия.

Для расширения функциональности бота можно интегрировать его с внешними API, например, для получения информации о погоде, курсе валют или данных из других сервисов. Это позволяет не только улучшить пользовательский опыт, но и добавляет боту новые возможности.

Пример: Интеграция с API погоды

import aiohttp

async def get_weather(city):
    api_key = 'your_api_key_here'
    url = f'http://api.weatherapi.com/v1/current.json?key={api_key}&q={city}&aqi=no'
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            return data

async def weather_command(update, context):
    city = update.message.text.split(' ', 1)[1]  # Извлекаем название города из сообщения
    weather_data = await get_weather(city)
    
    if 'error' not in weather_data:
        temp_c = weather_data['current']['temp_c']
        condition = weather_data['current']['condition']['text']
        await update.message.reply_text(f"В городе {city} сейчас {temp_c}°C, {condition}.")
    else:
        await update.message.reply_text("Город не найден. Попробуйте еще раз.")

# Добавление обработчика команды /weather
dp.add_handler(CommandHandler("weather", weather_command))

1. Асинхронный запрос к API: Использование aiohttp для асинхронного запроса к API погоды позволяет боту быстро получать и обрабатывать данные, не блокируя другие операции.
2. Обработка ответа: Полученные данные о погоде форматируются и отправляются пользователю в виде сообщения.
3. Расширение функциональности: Добавление команды /weather дает пользователям возможность запросить актуальную информацию о погоде для любого города.

Кроме базовых команд, таких как /start или /help, Telegram-боты могут обрабатывать сложные команды, содержащие параметры. Например, команда может содержать различные опции или настройки, которые пользователь может передать.

Пример: Обработка команды с параметрами

async def search_command(update, context):
    query = update.message.text.split(' ', 1)[1]  # Получаем текст после команды
    search_results = await perform_search(query)
    
    if search_results:
        result_message = "\n".join([f"{result['title']} - {result['link']}" for result in search_results])
        await update.message.reply_text(result_message)
    else:
        await update.message.reply_text("Ничего не найдено.")

# Добавление обработчика команды /search
dp.add_handler(CommandHandler("search", search_command))

1. Команда с параметром: Пользователь вводит команду /search с параметром для поиска (например, текст для поиска). Бот извлекает этот параметр и выполняет поисковый запрос.
2. Асинхронная обработка: Поиск выполняется асинхронно, что позволяет боту обрабатывать запросы даже при большом объеме данных.

Расширение функциональности Telegram-бота за счет добавления новых команд и обработчиков позволяет значительно улучшить взаимодействие с пользователем. Использование сложных сценариев обработки сообщений, асинхронных запросов к внешним API и гибких паттернов обработки команд делает бота мощным инструментом для решения самых разных задач. Возможность интеграции с другими системами и сервисами открывает неограниченные возможности для автоматизации и улучшения пользовательского опыта.

Одной из особенностей Telegram-ботов является их возможность работать как с приватными чатами, так и с групповыми чатами или супергруппами. Для бота это открывает ряд дополнительных возможностей, но также требует продуманного подхода к обработке сообщений, управлению правами и поддержке нескольких активных чатов. В этом разделе мы рассмотрим, как реализовать поддержку многопользовательских сценариев, эффективно работать с группами и управлять правами пользователей.

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

Пример: Проверка типа чата и выполнение действий

from telegram import Chat

async def handle_message(update, context):
    chat = update.effective_chat
    user = update.effective_user

    if chat.type == Chat.PRIVATE:
        await update.message.reply_text(f"Привет, {user.first_name}! Это приватный чат.")
    elif chat.type in [Chat.GROUP, Chat.SUPERGROUP]:
        await update.message.reply_text(f"Вы в группе {chat.title}. Приветствуем всех участников!")

1. Различение типов чатов: Telegram поддерживает разные типы чатов, и бот должен знать, в каком типе чата он работает, чтобы корректно взаимодействовать с пользователями. Например, в приватных чатах он может вести себя по одному сценарию, а в группах — по другому.
2. Контекстные ответы: В зависимости от типа чата бот может отправлять разные ответы или выполнять различные действия. Это полезно для того, чтобы обеспечить корректное взаимодействие в группах или индивидуально.

Группы и супергруппы в Telegram открывают возможность взаимодействия сразу с несколькими пользователями, и это требует от бота способности обрабатывать массовые взаимодействия, управлять сообщениями и выполнять специальные команды для управления группой.

Пример: Реализация команды для администратора группы

from telegram import ChatMemberAdministrator, ChatMemberOwner
from telegram.ext import CommandHandler

async def admin_command(update, context):
    user = update.effective_user
    chat = update.effective_chat
    member = await chat.get_member(user.id)

    if isinstance(member, (ChatMemberAdministrator, ChatMemberOwner)):
        await update.message.reply_text("Вы администратор этой группы.")
        # Выполняем команду для администратора
    else:
        await update.message.reply_text("Эта команда доступна только администраторам.")
        
dp.add_handler(CommandHandler("admin", admin_command))

1. Проверка прав администратора: В этом примере бот проверяет, является ли пользователь администратором группы перед выполнением команды. Это важно для управления доступом к критически важным функциям бота в группе.
2. Управление группой: Бот может выполнять специальные действия, такие как модерация сообщений или управление участниками группы, только если пользователь, отправивший команду, имеет права администратора.

Работа в групповых чатах требует более сложного управления сообщениями, так как бот может получать сразу много сообщений от разных пользователей. Важно корректно обрабатывать потоки сообщений и, при необходимости, взаимодействовать с конкретными пользователями или группой в целом.

Пример: Ответ на упоминание бота в группе

from telegram.ext import MessageHandler, filters

async def handle_mention(update, context):
    message = update.message
    if context.bot.username in message.text:
        await message.reply_text(f"Кто-то упомянул меня в чате {message.chat.title}.")

dp.add_handler(MessageHandler(filters.TEXT & filters.Entity("mention"), handle_mention))

1. Обработка упоминаний: В группах бот может быть упомянут пользователями. В этом примере бот реагирует на упоминание своего имени в сообщении, отправляя ответ. Это может быть полезно для случаев, когда бот должен принимать участие только в тех сообщениях, где его упоминают напрямую.
2. Фильтрация сообщений: С помощью фильтров библиотека telegram.ext позволяет обрабатывать только те сообщения, которые соответствуют определенным критериям (например, содержащие упоминание бота).

Telegram предоставляет боту доступ к информации о правах пользователей в чатах, включая возможность отличать обычных пользователей от администраторов. Это позволяет создавать более сложные сценарии работы с правами доступа, когда определенные команды доступны только определенным пользователям или ролям.

Пример: Модерация сообщений только для администраторов

from telegram.ext import MessageHandler, filters

async def delete_message(update, context):
    user = update.effective_user
    chat = update.effective_chat
    member = await chat.get_member(user.id)

    if isinstance(member, (ChatMemberAdministrator, ChatMemberOwner)):
        await context.bot.delete_message(chat.id, update.message.message_id)
        await update.message.reply_text("Сообщение удалено.")
    else:
        await update.message.reply_text("Только администраторы могут удалять сообщения.")

dp.add_handler(MessageHandler(filters.TEXT, delete_message))

1. Модерация администратором: Этот пример показывает, как можно предоставить права на удаление сообщений только администраторам группы. Это предотвращает злоупотребления ботом со стороны обычных пользователей и позволяет обеспечить контроль за чатами.
2. Управление доступом: Бот может проверять права пользователя и выполнять действия в зависимости от роли пользователя в группе. Это особенно полезно для управления большими группами или супергруппами, где есть необходимость в строгом управлении доступом.

Когда бот работает с несколькими чатами одновременно, важно иметь возможность отслеживать активность в каждом чате и сохранять данные о взаимодействиях для анализа или дальнейшего использования.

Пример: Логирование активности в группах

import logging

async def log_group_activity(update, context):
    chat = update.effective_chat
    user = update.effective_user
    message = update.message.text

    logging.info(f"Пользователь {user.id} ({user.first_name}) отправил сообщение в чате {chat.title}: {message}")

dp.add_handler(MessageHandler(filters.TEXT, log_group_activity))

1. Логирование сообщений: Этот пример демонстрирует, как бот может записывать активность в группах, включая ID пользователя, его имя и отправленное сообщение. Это может быть полезно для анализа поведения пользователей, управления активностью или мониторинга чатов.
2. Масштабируемость: Логирование позволяет боту отслеживать активность в нескольких группах одновременно, что полезно при работе с большим количеством чатов.

Работа с несколькими чатами и группами — это важная часть расширенных возможностей Telegram-ботов. Поддержка многопользовательских сценариев требует от бота гибкости в управлении правами, обработке сообщений и взаимодействии с группами. Возможность различать типы чатов, управлять доступом и правами пользователей, а также эффективно обрабатывать потоки сообщений позволяет создавать мощные и сложные системы, которые могут взаимодействовать с большими группами пользователей одновременно.

PS: Код проекта доступен на GitHub по следующей ссылке: https://github.com/exp-ext/fastapi_template. Обратите внимание, что он может не полностью соответствовать описанному здесь, так как проект находится в процессе развития. В дальнейшем шаблон будет дополняться новыми методами и проходить рефакторинг.