From fee19ff1aa8b7d8a2f8e5ca31b44e810e001eff7 Mon Sep 17 00:00:00 2001 From: Verum Date: Mon, 23 Feb 2026 14:38:02 +0700 Subject: [PATCH] =?UTF-8?q?=D0=90=D0=B2=D1=82=D0=BE=D0=BC=D0=B0=D1=82?= =?UTF-8?q?=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=BE=D0=B5=20=D1=83=D0=B4=D0=B0?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bot/utils/auto_delete.py | 636 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 636 insertions(+) create mode 100644 bot/utils/auto_delete.py diff --git a/bot/utils/auto_delete.py b/bot/utils/auto_delete.py new file mode 100644 index 0000000..300d060 --- /dev/null +++ b/bot/utils/auto_delete.py @@ -0,0 +1,636 @@ +""" +Утилиты для автоматического удаления сообщений +""" +from typing import Optional, Callable, Awaitable, Dict, Any +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from asyncio import sleep, create_task, Task, CancelledError + +from aiogram import Bot +from aiogram.types import Message +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError + +from middleware.loggers import logger +from .format_time import format_duration + +__all__ = ( + 'auto_delete_message', + 'schedule_delete', + 'cancel_delete', + 'delete_after', + 'auto_delete_manager', + 'AutoDeleteManager', + 'DeleteTask', + 'delete_both_after', + 'delete_messages_after', +) + + +@dataclass +class DeleteTask: + """ + Задача на удаление сообщения. + + Attributes: + chat_id: ID чата + message_id: ID сообщения + delete_at: Время удаления + task: Asyncio task + created_at: Время создания задачи + reason: Причина удаления + callback: Callback функция после удаления + """ + chat_id: int + message_id: int + delete_at: datetime + task: Optional[Task] = None + created_at: datetime = field(default_factory=datetime.now) + reason: Optional[str] = None + callback: Optional[Callable[[], Awaitable[None]]] = None + + @property + def delay(self) -> int: + """Задержка до удаления в секундах""" + delta = self.delete_at - datetime.now() + return max(0, int(delta.total_seconds())) + + @property + def is_expired(self) -> bool: + """Истекло ли время удаления""" + return datetime.now() >= self.delete_at + + def __repr__(self) -> str: + return ( + f"DeleteTask(chat={self.chat_id}, msg={self.message_id}, " + f"delay={self.delay}s, reason={self.reason})" + ) + + +class AutoDeleteManager: + """ + Менеджер автоматического удаления сообщений. + + Возможности: + - Планирование удаления с задержкой + - Отмена запланированного удаления + - Массовое удаление + - Callback функции + - История задач + - Автоматическая очистка завершенных задач + + Example: + ```python + from utils.auto_delete import auto_delete_manager + + # Планирование удаления + await auto_delete_manager.schedule( + bot=bot, + chat_id=message.chat.id, + message_id=message.message_id, + delay=60, + reason="Временное сообщение" + ) + + # Отмена удаления + auto_delete_manager.cancel(message.chat.id, message.message_id) + + # Получение статистики + stats = auto_delete_manager.get_stats() + ``` + """ + + def __init__(self): + # Активные задачи: {(chat_id, message_id): DeleteTask} + self.tasks: Dict[tuple[int, int], DeleteTask] = {} + + # Завершенные задачи (последние 100) + self.completed: list[DeleteTask] = [] + self.max_completed = 100 + + # Статистика + self.total_scheduled: int = 0 + self.total_deleted: int = 0 + self.total_failed: int = 0 + self.total_cancelled: int = 0 + + async def schedule( + self, + bot: Bot, + chat_id: int, + message_id: int, + delay: int, + reason: Optional[str] = None, + callback: Optional[Callable[[], Awaitable[None]]] = None, + log: bool = True + ) -> DeleteTask: + """ + Планирует удаление сообщения. + + Args: + bot: Экземпляр бота + chat_id: ID чата + message_id: ID сообщения + delay: Задержка в секундах + reason: Причина удаления + callback: Callback функция после удаления + log: Логировать планирование + + Returns: + DeleteTask: Созданная задача + + Example: + >> task = await auto_delete_manager.schedule( + ... bot=bot, + ... chat_id=message.chat.id, + ... message_id=message.message_id, + ... delay=60, + ... reason="Спам" + ... ) + """ + # Отменяем предыдущую задачу если есть + key = (chat_id, message_id) + if key in self.tasks: + self.cancel(chat_id, message_id) + + # Создаем задачу + delete_at = datetime.now() + timedelta(seconds=delay) + task_data = DeleteTask( + chat_id=chat_id, + message_id=message_id, + delete_at=delete_at, + reason=reason, + callback=callback + ) + + # Создаем asyncio task + task = create_task(self._delete_task(bot, task_data, log)) + task_data.task = task + + # Сохраняем + self.tasks[key] = task_data + self.total_scheduled += 1 + + if log: + delay_str = format_duration(delay) + logger.info( + f"Запланировано удаление сообщения через {delay_str}", + log_type='AUTO_DELETE' + ) + + return task_data + + async def _delete_task( + self, + bot: Bot, + task_data: DeleteTask, + log: bool + ) -> None: + """ + Внутренняя функция для выполнения задачи удаления. + + Args: + bot: Экземпляр бота + task_data: Данные задачи + log: Логировать выполнение + """ + key = (task_data.chat_id, task_data.message_id) + + try: + # Ждем + await sleep(task_data.delay) + + # Удаляем сообщение + await bot.delete_message( + chat_id=task_data.chat_id, + message_id=task_data.message_id + ) + + self.total_deleted += 1 + + if log: + reason_str = f" (причина: {task_data.reason})" if task_data.reason else "" + logger.info( + f"Сообщение удалено автоматически{reason_str}", + log_type='AUTO_DELETE' + ) + + # Вызываем callback если есть + if task_data.callback: + try: + await task_data.callback() + except Exception as e: + logger.error( + f"Ошибка в callback автоудаления: {e}", + log_type='AUTO_DELETE' + ) + + except CancelledError: + # Задача отменена + self.total_cancelled += 1 + + if log: + logger.debug( + f"Удаление сообщения отменено", + log_type='AUTO_DELETE' + ) + raise + + except (TelegramBadRequest, TelegramForbiddenError) as e: + # Ошибка удаления + self.total_failed += 1 + + if log: + logger.warning( + f"Не удалось автоматически удалить сообщение: {e}", + log_type='AUTO_DELETE' + ) + + finally: + # Удаляем из активных задач + if key in self.tasks: + completed_task = self.tasks.pop(key) + + # Сохраняем в завершенные + self.completed.append(completed_task) + if len(self.completed) > self.max_completed: + self.completed.pop(0) + + def cancel( + self, + chat_id: int, + message_id: int, + log: bool = True + ) -> bool: + """ + Отменяет запланированное удаление. + + Args: + chat_id: ID чата + message_id: ID сообщения + log: Логировать отмену + + Returns: + bool: True если задача была отменена + + Example: + >> cancelled = auto_delete_manager.cancel( + ... chat_id=message.chat.id, + ... message_id=message.message_id + ... ) + """ + key = (chat_id, message_id) + + if key in self.tasks: + task_data = self.tasks[key] + + # Отменяем asyncio task + if task_data.task and not task_data.task.done(): + task_data.task.cancel() + + # Удаляем из активных + self.tasks.pop(key) + + if log: + logger.debug( + f"Автоудаление отменено для сообщения {message_id}", + log_type='AUTO_DELETE' + ) + + return True + + return False + + def cancel_all(self, chat_id: Optional[int] = None) -> int: + """ + Отменяет все запланированные удаления. + + Args: + chat_id: ID чата (если None, отменяет для всех чатов) + + Returns: + int: Количество отмененных задач + + Example: + >> # Отменить для всех чатов + >> count = auto_delete_manager.cancel_all() + + >> # Отменить для конкретного чата + >> count = auto_delete_manager.cancel_all(chat_id=message.chat.id) + """ + cancelled_count = 0 + + # Собираем ключи для отмены + keys_to_cancel = [] + for key, task_data in self.tasks.items(): + if chat_id is None or task_data.chat_id == chat_id: + keys_to_cancel.append(key) + + # Отменяем + for key in keys_to_cancel: + if self.cancel(key[0], key[1], log=False): + cancelled_count += 1 + + if cancelled_count > 0: + logger.info( + f"Отменено {cancelled_count} задач автоудаления", + log_type='AUTO_DELETE' + ) + + return cancelled_count + + def get_task( + self, + chat_id: int, + message_id: int + ) -> Optional[DeleteTask]: + """ + Получает задачу по ID чата и сообщения. + + Args: + chat_id: ID чата + message_id: ID сообщения + + Returns: + Optional[DeleteTask]: Задача или None + """ + key = (chat_id, message_id) + return self.tasks.get(key) + + def get_chat_tasks(self, chat_id: int) -> list[DeleteTask]: + """ + Получает все задачи для чата. + + Args: + chat_id: ID чата + + Returns: + list[DeleteTask]: Список задач + """ + return [ + task for task in self.tasks.values() + if task.chat_id == chat_id + ] + + def get_stats(self) -> Dict[str, Any]: + """ + Возвращает статистику менеджера. + + Returns: + Dict: Словарь со статистикой + + Example: + >> stats = auto_delete_manager.get_stats() + >> print(f"Активных задач: {stats['active_tasks']}") + """ + return { + 'active_tasks': len(self.tasks), + 'completed_tasks': len(self.completed), + 'total_scheduled': self.total_scheduled, + 'total_deleted': self.total_deleted, + 'total_failed': self.total_failed, + 'total_cancelled': self.total_cancelled, + 'success_rate': ( + f"{(self.total_deleted / self.total_scheduled * 100):.1f}%" + if self.total_scheduled > 0 else "0%" + ) + } + + def cleanup_expired(self) -> int: + """ + Удаляет истекшие задачи (которые должны были выполниться, но не выполнились). + + Returns: + int: Количество удаленных задач + """ + expired_keys = [ + key for key, task in self.tasks.items() + if task.is_expired and (not task.task or task.task.done()) + ] + + for key in expired_keys: + self.tasks.pop(key) + + return len(expired_keys) + + +# Глобальный менеджер +auto_delete_manager = AutoDeleteManager() + + +# ================= УДОБНЫЕ ФУНКЦИИ ================= + +async def auto_delete_message( + bot: Bot, + chat_id: int, + message_id: int, + delay: int = 604800, + reason: Optional[str] = None +) -> DeleteTask: + """ + Автоматически удаляет сообщение через указанное время. + + Args: + bot: Экземпляр бота + chat_id: ID чата + message_id: ID сообщения + delay: Задержка в секундах (по умолчанию 7 дней) + reason: Причина удаления + + Returns: + DeleteTask: Созданная задача + + Example: + >> # Удалить через 1 минуту + >> await auto_delete_message(bot, chat_id, message_id, delay=60) + + >> # Удалить через 7 дней (по умолчанию) + >> await auto_delete_message(bot, chat_id, message_id) + """ + return await auto_delete_manager.schedule( + bot=bot, + chat_id=chat_id, + message_id=message_id, + delay=delay, + reason=reason + ) + + +async def schedule_delete( + message: Message, + delay: int, + reason: Optional[str] = None +) -> DeleteTask: + """ + Планирует удаление сообщения (упрощенная версия). + + Args: + message: Объект сообщения + delay: Задержка в секундах + reason: Причина удаления + + Returns: + DeleteTask: Созданная задача + + Example: + >> # Планируем удаление через 30 секунд + >> await schedule_delete(message, delay=30, reason="Временное") + """ + return await auto_delete_manager.schedule( + bot=message.bot, + chat_id=message.chat.id, + message_id=message.message_id, + delay=delay, + reason=reason + ) + + +def cancel_delete(message: Message) -> bool: + """ + Отменяет запланированное удаление сообщения. + + Args: + message: Объект сообщения + + Returns: + bool: True если удаление было отменено + + Example: + >> if cancel_delete(message): + ... await message.answer("Удаление отменено") + """ + return auto_delete_manager.cancel( + chat_id=message.chat.id, + message_id=message.message_id + ) + + +async def delete_after( + message: Message, + text: str, + delay: int = 10, + **kwargs +) -> Message: + """ + Отправляет сообщение и автоматически удаляет его через указанное время. + + Args: + message: Исходное сообщение + text: Текст нового сообщения + delay: Задержка до удаления в секундах + **kwargs: Дополнительные параметры для message.answer() + + Returns: + Message: Отправленное сообщение + + Example: + >> # Отправить и удалить через 10 секунд + >> await delete_after(message, "Это временное сообщение") + + >> # Отправить и удалить через 5 секунд + >> await delete_after( + ... message, + ... "⚠️ Ошибка!", + ... delay=5, + ... parse_mode="HTML" + ... ) + """ + sent_message = await message.answer(text, **kwargs) + + await auto_delete_manager.schedule( + bot=message.bot, + chat_id=sent_message.chat.id, + message_id=sent_message.message_id, + delay=delay, + reason="delete_after" + ) + + return sent_message + + +async def delete_both_after( + original: Message, + reply_text: str, + delay: int = 10, + **kwargs +) -> Message: + """ + Отправляет ответ и удаляет оба сообщения через указанное время. + + Args: + original: Исходное сообщение + reply_text: Текст ответа + delay: Задержка до удаления + **kwargs: Дополнительные параметры + + Returns: + Message: Отправленное сообщение + + Example: + >> # Удалить и команду, и ответ через 5 секунд + >> await delete_both_after( + ... message, + ... "✅ Команда выполнена", + ... delay=5 + ... ) + """ + # Отправляем ответ + sent = await delete_after(original, reply_text, delay, **kwargs) + + # Планируем удаление оригинала + await auto_delete_manager.schedule( + bot=original.bot, + chat_id=original.chat.id, + message_id=original.message_id, + delay=delay, + reason="delete_both" + ) + + return sent + + +async def delete_messages_after( + bot: Bot, + chat_id: int, + message_ids: list[int], + delay: int +) -> int: + """ + Планирует удаление нескольких сообщений. + + Args: + bot: Экземпляр бота + chat_id: ID чата + message_ids: Список ID сообщений + delay: Задержка до удаления + + Returns: + int: Количество запланированных удалений + + Example: + >> # Удалить все сообщения через 1 час + >> count = await delete_messages_after( + ... bot, + ... chat_id, + ... [123, 124, 125, 126], + ... delay=3600 + ... ) + """ + count = 0 + + for message_id in message_ids: + await auto_delete_manager.schedule( + bot=bot, + chat_id=chat_id, + message_id=message_id, + delay=delay, + reason="mass_delete", + log=False + ) + count += 1 + + logger.info( + f"Запланировано удаление {count} сообщений через {format_duration(delay)}", + log_type='AUTO_DELETE' + ) + + return count