From 49e9c56bd1c763084c7eb8ea75b8d26bf5d487c6 Mon Sep 17 00:00:00 2001 From: Verum Date: Mon, 23 Feb 2026 14:24:11 +0700 Subject: [PATCH] =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D1=8C=D1=82=D1=80=20=D0=BD?= =?UTF-8?q?=D0=B0=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D0=BA=D1=83=20?= =?UTF-8?q?=D0=BF=D0=BE=D0=B4=D0=BF=D0=B8=D1=81=D0=BA=D0=B8=20=D0=BD=D0=B0?= =?UTF-8?q?=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=20=D0=BE=D1=82=20=D0=BF=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2=D0=B0=D1=82=D0=B5=D0=BB=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bot/filters/subscription.py | 246 ++++++++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 bot/filters/subscription.py diff --git a/bot/filters/subscription.py b/bot/filters/subscription.py new file mode 100644 index 0000000..3158efe --- /dev/null +++ b/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