Возможно опубликую раньше чем в четверг)

Если вы работали с Excel через Python, то могли столкнуться с ситуацией, что формат xls не поддерживается. 📊 Например, Openpyxl работает только с xlsx. В таком случае нам ничего не мешает написать собственный конвертер в пару строк кода. 🛠

Во-первых, нам понадобятся для этой задачи 2 легковесные библиотеки: pyexcel_xls и pyexcel_xlsx. Устанавливаем. 📚

Далее выполняем импорты:

from pyexcel_xls import get_data
from pyexcel_xlsx import save_data
import os

Теперь напишем простую функцию для конвертации из xls в xlsx:

def convert_xls_to_xlsx(xls_path):
    # Получаем имя файла без расширения
    file_name = os.path.splitext(xls_path)[0]
    xlsx_path = file_name + '.xlsx'

    # Читаем данные из .xls файла
    xls_data = get_data(xls_path)

    # Сохраняем данные в .xlsx формате
    save_data(xlsx_path, xls_data)

    return xlsx_path

Далее можно использовать ее в таком формате:

if file_path.endswith('.xls'):
    print("Конвертация .xls в .xlsx...")
    file_path = convert_xls_to_xlsx(file_path)
    print(f"Файл сконвертирован: {file_path}")

Например, как часть более сложной функции с тем же openpyxl.

P.S. Бота пишу сейчас. Это кусок с реальной практики. Решил, может кому полезно будет) 🤖💡

Осталось вычитать, так что ждите, скорее всего, в субботу утром.

💡 Как узнать информацию о Telegram-боте, имея только его токен? 🤖

Иногда в процессе работы с ботами бывает, что единственное, что у вас осталось от старого проекта, — это токен. И тогда возникает вопрос: как узнать информацию о боте, чтобы восстановить его данные?

📌 Есть простой способ:

1️⃣ Откройте в браузере следующую ссылку, заменив {BOT_TOKEN} на ваш токен:

https://api.telegram.org/botBOT_TOKEN/getMe  

2️⃣ Вы получите JSON-ответ с данными бота, который выглядит так:

{
  "ok": true,
  "result": {
    "id": 929198812,
    "is_bot": true,
    "first_name": "my_test",
    "username": "my_test_bot",
    "can_join_groups": true,
    "can_read_all_group_messages": false,
    "supports_inline_queries": false,
    "can_connect_to_business": false,
    "has_main_web_app": false
  }
}

📋 В ответе содержится:

- 🆔 ID бота
- 👤 Имя (first_name)
- 🔗 Логин (username)
- 🤝 Возможности бота: участие в группах, поддержка inline-запросов и т.д.

⚠️ Совет: всегда сохраняйте информацию о боте (токен, ID, логин) в безопасном месте, чтобы в будущем не сталкиваться с такими сложностями. 🔐

Теперь вы знаете, как легко восстановить данные о боте, имея только токен! 🚀

💡 Проверяем активность пользователей бота на 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️⃣ Открывает корзину и при получении задачи автоматически оформляет заказ.

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

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