💡 Проверяем активность пользователей бота на Python (Aiogram 3)

Привет, разработчики! Сегодня делюсь полезным кодом, который может вам пригодиться при работе с телеграм-ботами. 👨‍💻

Задача: проверить, сколько пользователей в вашей базе данных всё ещё получают сообщения от бота, а сколько его заблокировали.

Вот пример решения:

alive_users = 0  
all_count_user_in_db = 100  # Например, общее число пользователей  

for user_id in users:  
    try:  
        chat = await bot.get_chat(chat_id=user_id)  
        alive_users += 1  
    except TelegramBadRequest as e:  
        # Пользователь заблокировал бота или чат недоступен  
        print(f"Ошибка при проверке пользователя {user_id}: {e}")  
    except Exception as e:  
        # Обработка других непредвиденных ошибок  
        print(f"Неожиданная ошибка для пользователя {user_id}: {e}")  

await message.answer(  
    f"📊 В базе данных: {all_count_user_in_db} пользователей.\n"  
    f"✅ Доступны: {alive_users}.\n"  
    f"🚫 Недоступны: {all_count_user_in_db - alive_users}."  
)  

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

1️⃣ Считаем активных пользователей через bot.get_chat.
2️⃣ Ловим исключения, чтобы учесть тех, кто заблокировал бота или удалил чат.
3️⃣ Возвращаем статистику пользователю.

Такой подход помогает понимать актуальность вашей базы и анализировать вовлечённость аудитории. 🎯

🚀 Привет, друзья! 👋

Если вы читали мою предыдущую статью на Хабре о связке Celery + Redis + FastAPI, то могли заметить, что некоторые практики там выглядят избыточными, а местами не хватает асинхронности — особенно для фоновых задач.

Согласитесь, было бы удобно иметь один проект, где одновременно можно:

✅ запускать FastAPI для основного приложения,
✅ выполнять фоновые задачи,
✅ использовать планировщик задач.

И вот я рад представить вам моё авторское решение проблемы асинхронности и фоновых задач в формате "ВСЁ В ОДНОМ"! 🎉

💡 В реализации я использовал две потрясающие библиотеки:

🔹 APScheduler — проверенное временем решение для отложенных и периодических задач, включая асинхронные.
🔹 FastStream — современный "убийца Celery" для высокопроизводительных фоновых задач.

Не буду утомлять теорией — лучше покажу код. 💻

Весь исходник можно найти по ссылке: 👉 ССЫЛКА

Буду рад вашим комментариям, вопросам и предложениям по улучшению! Пишите! 📝

🔥 Обещаю в будущем подробнее рассказать о своём подходе и почему считаю эту связку "бронебойной"🚀

Немного прокачал подход по мидлвари в контексте взаимодействия с базой данных. Может кому будет полезно:

from typing import Callable, Dict, Any, Awaitable
from aiogram import BaseMiddleware
from aiogram.types import Message, CallbackQuery

from bot.dao.database import async_session_maker


class BaseDatabaseMiddleware(BaseMiddleware):
    async def __call__(
            self,
            handler: Callable[[Message | CallbackQuery, Dict[str, Any]], Awaitable[Any]],
            event: Message | CallbackQuery,
            data: Dict[str, Any]
    ) -> Any:
        async with async_session_maker() as session:
            self.set_session(data, session)
            try:
                result = await handler(event, data)
                await self.after_handler(session)
                return result
            except Exception as e:
                await session.rollback()
                raise e
            finally:
                await session.close()

    def set_session(self, data: Dict[str, Any], session) -> None:
        """Метод для установки сессии в словарь данных."""
        raise NotImplementedError("Этот метод должен быть реализован в подклассах.")

    async def after_handler(self, session) -> None:
        """Метод для выполнения действий после вызова хендлера (например, коммит)."""
        pass


class DatabaseMiddlewareWithoutCommit(BaseDatabaseMiddleware):
    def set_session(self, data: Dict[str, Any], session) -> None:
        data['session_without_commit'] = session


class DatabaseMiddlewareWithCommit(BaseDatabaseMiddleware):
    def set_session(self, data: Dict[str, Any], session) -> None:
        data['session_with_commit'] = session

    async def after_handler(self, session) -> None:
        await session.commit()

Если коротко, то тут генерируется 2 мидлвари: первый делает автоматические коммиты, а второй не делает.

Теперь мы регистрируем их в main-файле:

dp.update.middleware.register(DatabaseMiddlewareWithoutCommit())
dp.update.middleware.register(DatabaseMiddlewareWithCommit())

И теперь вызываем в зависимости от необходимости.

Пример без коммита:

@user_router.callback_query(F.data == "catalog")
async def page_catalog(call: CallbackQuery, session_without_commit: AsyncSession):
    await call.answer("Загрузка каталога...")
    catalog_data = await CategoryDao.find_all(session=session_without_commit)

    await call.message.edit_text(
        text="Выберите категорию товаров:",
        reply_markup=catalog_kb(catalog_data)
    )

Пример с коммитом:

@user_router.message(CommandStart())
async def cmd_start(message: Message, session_with_commit: AsyncSession):
    user_id = message.from_user.id
    user_info = await UserDAO.find_one_or_none(
        session=session_with_commit,
        filters=TelegramIDModel(telegram_id=user_id)
    )

    if user_info:
        return await message.answer(
            f"👋 Привет, {message.from_user.full_name}! Выберите необходимое действие",
            reply_markup=main_user_kb(user_id)
        )

    values = UserModel(
        telegram_id=user_id,
        username=message.from_user.username,
        first_name=message.from_user.first_name,
        last_name=message.from_user.last_name,
    )
    await UserDAO.add(session=session_with_commit, values=values)
    await message.answer(f"🎉 <b>Благодарим за регистрацию!</b>. Теперь выберите необходимое действие.",
                         reply_markup=main_user_kb(user_id))

💥 Друзья, добрый вечер! 💥

🎉 Новый проект под предстоящую статью готов! 🎉

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

🛠 Чтобы не усложнять, я намеренно не использовал веб-хуки, мини-апп и прочие технологии, сосредоточившись на самом главном: добавление товаров в магазин и продажи. 📦💵

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

📖 Подробное описание проекта будет в следующей статье на Хабре. 💻

🎥 В видео, прикрепленном к этому посту, вы можете увидеть, как работает бот: «мои покупки», каталог товаров, проведение платежей, добавление товаров и прочий функционал. 📽

🔗 Ссылка на исходник проекта: GitHub

🤖 Ссылка на работающего бота: @DigitalMarketAiogramBot

В этот раз, похоже, я превзошел себя по длине статьи 🤦‍♂️. Готовьтесь, будет много кода и немного теории.

Если успею все вычитать, то публикация будет завтра в первой половине дня.

Будете ждать?

💡 Интересный кейс из практики: автоматизация покупок в интернет-магазине 🛍

Друзья, решил поделиться увлекательным кейсом, который недавно реализовал!

📋 Задача:

Есть интернет-магазин, где периодически выставляют товары со скидкой 30-40%. Однако, таких товаров всего 1 экземпляр. Чтобы успеть купить:

1️⃣ Нажимаем «Добавить в корзину»
2️⃣ Переходим в корзину
3️⃣ Жмем «Оформить заказ»

Но есть нюансы:

- Корзина обновляется только после перезагрузки страницы.
- Сессии привязаны к браузеру: в другом браузере корзина не синхронизируется.

🛠 Цель клиента: автоматизировать процесс покупки, чтобы оформить заказ сразу после добавления товара в корзину. При этом клиенту нужно сохранить возможность обновлять страницы и делать покупки из разных мест.

💻 Задачу решил через Python двумя способами:

📌 Способ 1: "Поллинг" на чистом Playwright

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

1️⃣ Запускается браузер и выполняется автоматический вход.
2️⃣ Извлекаются куки, после чего клиент берет управление браузером.
3️⃣ Параллельно запускается второй браузер, где:

- Устанавливаются куки.
- Постоянно обновляется страница корзины.
- При обнаружении товара запускается скрипт, который мгновенно оформляет заказ.

📈 Результат: оформление заказа занимает менее секунды! Это возможно благодаря синхронизации корзин через куки (авторизация завязана на токене в куки).

📌 Способ 2: Веб-хуки + FastStream + Redis

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

Скрипт 1:

1️⃣ Запускает браузер, выполняет вход, извлекает куки и сохраняет их в Redis.
2️⃣ Клиент управляет браузером: кликает, обновляет страницы, добавляет товары в корзину.
3️⃣ При добавлении товара срабатывает триггер через FastStream, который передает задачу второму скрипту.

Скрипт 2:

1️⃣ Запускает второй браузер и извлекает куки из Redis.
2️⃣ Открывает корзину и при получении задачи автоматически оформляет заказ.

📈 Результат: быстрая и надежная синхронизация с использованием современных технологий.

Как вам такой подход?

Слишком жизненно...

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

Решил напомнить о существовании)

Процесс подготовки материала к новой статье идет полным ходом. Обратите внимание, что при работе со звездами можно легко и просто настроить возврат средств (как в боте @DurgerKingBot). Для этого достаточно вызвать команду:

await bot.refund_star_payment(user_id=message.from_user.id, telegram_payment_charge_id=payment_info.telegram_payment_charge_id)

Доброе утро, друзья! 🌅

Рад сообщить об обновлении бота для интернет-магазина цифровых товаров! Теперь он стал еще удобнее и функциональнее.

🔹 Главные нововведения:

— Три платежные системы вместо одной:

1️⃣ ЮKassa — интеграция через BotFather.
2️⃣ Telegram-Stars — удобная система для совершения оплат.
3️⃣ Robocassa — интеграция напрямую с использованием веб-хуков.

— Переход на веб-хуки: вместо лонг-поллинга теперь используется технология веб-хуков. Это обеспечило заметный прирост производительности.

— Использование Aiohttp: на этот раз я выбрал Aiohttp вместо FastApi для реализации хуков в проекте.

📌 В новой статье я расскажу о процессе перевода бота с лонг-поллинга на веб-хуки и подробно опишу интеграцию двух дополниельных платежных систем.

🎥 На видео (см. пост) вы можете увидеть, как работает оплата через все три платежные системы.

💻 Для тех, кто хочет изучить проект:

— Исходный код: GitHub
— Работающий бот: @DigitalMarketAiogramHookBot

Оставляйте вопросы и комментарии👇

Друзья, отличные новости!

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

Проект получил следующие глобальные обновления:

✅ Добавил две новые платежные системы — Telegram Stars и Robokassa (всего теперь три способа оплаты).

✅ Перевел бота на вебхуки для повышения производительности (веб-сервер построен на Aiohttp).

Пока ждете публикации, вы уже можете:

📂 Ознакомиться с исходным кодом проекта на GitHub.

🤖 Протестировать бота в действии: @DigitalMarketAiogramHookBot.

Следите за обновлениями 🙌