Aiogram удаление и чистка базы пользователей

kirilljsx

Так, я столкнулся с такой проблемой, а точнее задачей.
Мне необходимо было почистить базу пользователей моего бота на Aiogram.

И так вот что я придумал (читайте коменты в коде):

# И так вот мой хендлер который отслеживает команду /clean
@main_router.message(Command("clean"))
async def handle_clean(message: types.Message):
    await clean_command(message, bot, user_db)

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

Также может возникнуть вопрос для чего функция handle_clean - она не несет ничего кроме разбивки логики:
handle_clean - это обработчик маршрута (handler). Его задача:
→ поймать команду /clean,
→ передать управление логике очистки,
→ ничего больше не делать.

Едем дальше, теперь о самой функции проверки и удаления в Aiogram:

import logging
from aiogram.exceptions import (
    TelegramAPIError,
    TelegramForbiddenError, # заменяет BotBlocked и ChatNotFound
    TelegramRetryAfter, # заменяет RetryAfter
)
logger = logging.getLogger(__name__)

# Посмотрите внимательно на импорт ошибок - это замена старый импортов из версии Aiogram 2

# Едем дальше вот наша функция которая проверят и удаляет пользователей
async def clean_command(message, bot, user_db):
    admin_id = int(os.getenv("ADMIN_ID"))

    # Проверяем что команда запущена от админа бота
    if message.from_user.id != admin_id:
        return await message.answer("Недостаточно прав для выполнения этой команды.")
    
    # Так как я довен и назвал по ошибке экземпляр БД то она у меня - user_db
    # Из экземпляра БД получаем все пользователей и записываем их кол-во в переменную
    users = await user_db.get_all_users()
    total = len(users)
    removed = 0

    # Просто уведомление что команда запустилась
    await message.answer(f"Начинаю очистку базы... Всего пользователей: {total}")
    
    # Тут самое интересное
    # Проходимся по каждому пользователю и проверяем
    for user_id, info in users.items():
        try:
            await bot.send_chat_action(chat_id=user_id, action="typing")
            await asyncio.sleep(0.05)  # уважаем лимиты Telegram

        except TelegramForbiddenError:
            # Проверка - пользователь заблокировал бота или чат не существует
            # Если пользователь удалил чат или заблочил бота то удаляет его
            await user_db.remove_user(user_id)
            removed += 1
            logger.info(f"Удалён пользователь {user_id}: бот заблокирован или чат не найден.")

        # Эта функция обрабатывает ограничение скорости
        # Что бы нас не заблокировали при отправке запросов
        except TelegramRetryAfter as e:
            logger.warning(f"Flood control: пауза {e.retry_after} сек.")
            await asyncio.sleep(e.retry_after)
            continue
        
        # Обрабатываем другие ошибки API
        except TelegramAPIError as e:
            logger.error(f"Ошибка Telegram API у пользователя {user_id}: {e}")
        
        # Логируем ошибку
        except Exception as e:
            logger.exception(f"Непредвиденная ошибка у пользователя {user_id}: {e}")

    # Сообщаем что очистка завершена сколько удалено из общего кол-во пользователей
    await message.answer(f"Очистка завершена. Удалено {removed} из {total} пользователей.")

В среднем такая проверка занимает около 5-7 минут (Моя база 10к +), топорно конечно. Но я просто в самом начале не предусмотрел отлавливание удаление чата/блокировки бота пользователями. По этому пришлось как-то выходить из ситуации

Надеюсь мое решение кому-то поможет

kirilljsx

Ах, да!
Чуть не забыл, как можно отлавливать пользователей если они заблокировали бота и сразу удалять их из БД:

# Для этого нам потребуется - ChatMemberUpdated
from aiogram.types import ChatMemberUpdated

# Если вы работали с aiogram то думаю роутер по умолчанию вы уже импортнули
# Но если нет то добавлю на всякий случай
from aiogram import Router


# Создаем основной роутер
main_router = Router()

@main_router.my_chat_member()
async def on_chat_member_update(event: ChatMemberUpdated):
    # Работаем только с личными чатами, если бот добавлен в канал то увы я пока не придумал
    if event.chat.type != "private":
        return

    # Проверяем, заблокировал ли пользователь бота
    if event.new_chat_member.status == "kicked":
        user_id = event.from_user.id
        await user_db.remove_user(user_id)
        logger.info(f"🗑️ Пользователь {user_id} заблокировал бота — удалён из БД.")

kirilljsx

Вообще я хочу сделать шаблон на гите сразу со всеми зависимостями, уже готовой структурой импортами и сразу с БД (НО ТОЛЬКО SQL lite через файл db) для будущих ботов - это может создавать ботов чуть быстрее так как сразу можно приступить к бизнес логике.

Если накидаете плюсиков +++ сделаю в близжайшее время.

Кстати вот сам мой бот по скачке коротких видео:

Скачать видео ВК Клипы | ТикТок

🤖 Умею скачивать видео из VK Clips, TikTok, Yappy. Интеграция: https://telega.in/tg_bots/vkclipssavebot/card

Telegram (t.me)

SkitRa

+++ Однозначно полезный шаблон. С нетерпением жду!!!

Matatabi

А если не получилось с помощью кода отлавливать тех пользователей которые заблокировали бота,значит их просто нет или что-то не сработало?