Автоматическое удаление сообщений

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