Фильтр на проверку подписки на канал от пользователя

bot/filters/subscription.py

@@ -0,0 +1,246 @@
+"""
+Фильтр проверки подписки пользователя на каналы/группы
+"""
+from typing import Union, Optional
+from dataclasses import dataclass
+
+from aiogram import Bot
+from aiogram.enums import ChatMemberStatus
+from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
+from aiogram.filters import BaseFilter
+from aiogram.types import Message, CallbackQuery
+
+from middleware.loggers import logger
+
+__all__ = ('IsSubscribed', 'SubscriptionChecker')
+
+
+@dataclass
+class ChannelInfo:
+    """Информация о канале для проверки подписки"""
+    id: Union[str, int]
+    name: Optional[str] = None
+    invite_link: Optional[str] = None
+
+
+class SubscriptionChecker:
+    """
+    Вспомогательный класс для проверки подписок.
+    Может использоваться отдельно от фильтра.
+    """
+
+    # Статусы, считающиеся подпиской
+    SUBSCRIBED_STATUSES: set[str] = {
+        ChatMemberStatus.MEMBER,
+        ChatMemberStatus.ADMINISTRATOR,
+        ChatMemberStatus.CREATOR
+    }
+
+    # Статусы, означающие отсутствие подписки
+    NOT_SUBSCRIBED_STATUSES: set[str] = {
+        ChatMemberStatus.LEFT,
+        ChatMemberStatus.KICKED,
+        ChatMemberStatus.RESTRICTED  # Опционально
+    }
+
+    @classmethod
+    async def is_subscribed(
+            cls,
+            bot: Bot,
+            user_id: int,
+            channel_id: Union[str, int]
+    ) -> bool:
+        """
+        Проверяет подписку одного пользователя на один канал.
+
+        Args:
+            bot: Экземпляр бота
+            user_id: ID пользователя
+            channel_id: ID или username канала
+
+        Returns:
+            bool: True если подписан
+        """
+        try:
+            member = await bot.get_chat_member(
+                chat_id=channel_id,
+                user_id=user_id
+            )
+
+            is_sub = member.status in cls.SUBSCRIBED_STATUSES
+
+            logger.debug(
+                f"Проверка подписки user={user_id} на канал={channel_id}: {member.status} ({'✅' if is_sub else '❌'})",
+                log_type='SUBSCRIPTION'
+            )
+
+            return is_sub
+
+        except TelegramBadRequest as e:
+            logger.warning(
+                f"Канал {channel_id} недоступен или неверный ID: {e}",
+                log_type='SUBSCRIPTION'
+            )
+            return False
+
+        except TelegramForbiddenError as e:
+            logger.error(
+                f"Бот не имеет доступа к каналу {channel_id}: {e}",
+                log_type='SUBSCRIPTION'
+            )
+            return False
+
+        except Exception as e:
+            logger.error(
+                f"Непредвиденная ошибка проверки подписки на {channel_id}: {e}",
+                log_type='SUBSCRIPTION'
+            )
+            return False
+
+    @classmethod
+    async def check_all_channels(
+            cls,
+            bot: Bot,
+            user_id: int,
+            channels: list[Union[str, int]]
+    ) -> dict[Union[str, int], bool]:
+        """
+        Проверяет подписку на несколько каналов одновременно.
+
+        Args:
+            bot: Экземпляр бота
+            user_id: ID пользователя
+            channels: Список ID/username каналов
+
+        Returns:
+            dict: Словарь {channel_id: is_subscribed}
+        """
+        results = {}
+
+        for channel in channels:
+            results[channel] = await cls.is_subscribed(bot, user_id, channel)
+
+        return results
+
+    @classmethod
+    async def get_not_subscribed_channels(
+            cls,
+            bot: Bot,
+            user_id: int,
+            channels: list[Union[str, int]]
+    ) -> list[Union[str, int]]:
+        """
+        Возвращает список каналов, на которые пользователь НЕ подписан.
+
+        Args:
+            bot: Экземпляр бота
+            user_id: ID пользователя
+            channels: Список ID/username каналов
+
+        Returns:
+            list: Список каналов без подписки
+        """
+        not_subscribed = []
+
+        for channel in channels:
+            if not await cls.is_subscribed(bot, user_id, channel):
+                not_subscribed.append(channel)
+
+        return not_subscribed
+
+
+class IsSubscribed(BaseFilter):
+    """
+    Фильтр для проверки подписки пользователя на каналы/группы.
+
+    Поддерживает:
+    - Публичные каналы (username: "@channel_name")
+    - Приватные каналы/группы (ID: -1001234567890)
+    - Проверку всех или хотя бы одного канала
+    - Работу с Message и CallbackQuery
+
+    Attributes:
+        channels: Список ID или username каналов для проверки
+        require_all: Требовать подписку на все каналы (True) или хотя бы один (False)
+
+    Examples:
+        >> # Проверка подписки на один канал
+        >> @router.message(IsSubscribed(["@my_channel"]))
+        >> async def handler(message: Message):
+        ...     await message.answer("Ты подписан!")
+
+        >> # Проверка на несколько каналов (все обязательны)
+        >> @router.message(IsSubscribed(["@channel1", -1001234567890], require_all=True))
+        >> async def handler(message: Message):
+        ...     await message.answer("Ты подписан на все каналы!")
+
+        >> # Проверка на несколько каналов (хотя бы один)
+        >> @router.message(IsSubscribed(["@channel1", "@channel2"], require_all=False))
+        >> async def handler(message: Message):
+        ...     await message.answer("Ты подписан хотя бы на один канал!")
+    """
+
+    def __init__(
+            self,
+            channels: list[Union[str, int]],
+            require_all: bool = True
+    ) -> None:
+        """
+        Инициализация фильтра.
+
+        Args:
+            channels: Список ID или username каналов
+            require_all: True = все каналы, False = хотя бы один
+        """
+        if not channels:
+            raise ValueError("Список каналов не может быть пустым")
+
+        self.channels = channels
+        self.require_all = require_all
+
+    async def __call__(
+            self,
+            event: Union[Message, CallbackQuery],
+            bot: Bot
+    ) -> Union[bool, dict]:
+        """
+        Проверка подписки.
+
+        Args:
+            event: Message или CallbackQuery
+            bot: Экземпляр бота
+
+        Returns:
+            bool или dict: True/False для простой проверки,
+                          dict с деталями для сложной логики
+        """
+        user_id = event.from_user.id
+
+        # Проверка всех каналов
+        results = await SubscriptionChecker.check_all_channels(
+            bot, user_id, self.channels
+        )
+
+        # Логика проверки
+        if self.require_all:
+            # Все каналы обязательны
+            is_passed = all(results.values())
+        else:
+            # Хотя бы один канал
+            is_passed = any(results.values())
+
+        # Логирование
+        if not is_passed:
+            not_subscribed = [ch for ch, sub in results.items() if not sub]
+            logger.info(
+                f"Пользователь {user_id} не подписан на: {not_subscribed}",
+                log_type='SUBSCRIPTION',
+                message=event if isinstance(event, Message) else None
+            )
+
+        # Возвращаем результат + детали для handler
+        return {
+            'is_subscribed': is_passed,
+            'subscription_results': results,
+            'not_subscribed_channels': [ch for ch, sub in results.items() if not sub]
+        } if not is_passed else is_passed