From 8375d7ecc5e54ea935a00175dd26f667eab95346 Mon Sep 17 00:00:00 2001 From: Fringg Date: Tue, 24 Feb 2026 02:50:18 +0300 Subject: [PATCH] feat: add multi-channel mandatory subscription system - Multi-channel subscription enforcement via middleware, events, and cabinet API - 3-layer cache architecture: Redis -> PostgreSQL -> rate-limited Telegram API - ChatMemberUpdated event-driven tracking with automatic VPN access control - Admin management via bot FSM handler and REST API with full CRUD - Channel ID normalization: @username resolved to numeric ID at creation time - Fail-closed error handling: API errors deny access (security-first) - Background reconciliation with keyset pagination (100 per batch) - Per-user rate limiting on subscription check button (5s cooldown) - Redis connection pooling via cache singleton (no per-request connections) - Database: channel_id index, multi-row upsert optimization - Localization: en, ru, zh, fa, ua translations for all new strings - Frontend blocking UI with channel list and subscription status - Admin channel management page with toggle, delete, and create --- .env.example | 4 +- app/bot.py | 16 +- app/cabinet/dependencies.py | 61 +-- app/cabinet/routes/__init__.py | 2 + app/cabinet/routes/admin_channels.py | 95 +++++ app/cabinet/schemas/channel.py | 78 ++++ app/config.py | 2 - app/database/crud/required_channel.py | 225 +++++++++++ app/database/models.py | 45 +++ app/handlers/admin/__init__.py | 1 + app/handlers/admin/bot_configuration.py | 7 +- app/handlers/admin/monitoring.py | 29 +- app/handlers/admin/required_channels.py | 285 +++++++++++++ app/handlers/channel_member.py | 184 +++++++++ app/handlers/start.py | 20 +- app/keyboards/inline.py | 37 +- app/localization/locales/en.json | 6 +- app/localization/locales/fa.json | 6 +- app/localization/locales/ru.json | 6 +- app/localization/locales/ua.json | 6 +- app/localization/locales/zh.json | 8 +- app/middlewares/channel_checker.py | 372 ++++++++--------- app/services/channel_subscription_service.py | 272 +++++++++++++ app/services/contest_rotation_service.py | 12 +- app/services/monitoring_service.py | 380 ++++++++++-------- app/services/payment/yookassa.py | 7 +- app/services/referral_contest_service.py | 15 +- app/services/system_settings_service.py | 2 - app/services/yookassa_service.py | 115 +----- app/utils/cache.py | 94 +++++ app/webapi/routes/miniapp.py | 44 +- main.py | 4 + .../versions/0008_add_required_channels.py | 61 +++ .../versions/0009_add_channel_id_index.py | 28 ++ tests/conftest.py | 11 + .../services/test_payment_service_yookassa.py | 2 +- 36 files changed, 1906 insertions(+), 636 deletions(-) create mode 100644 app/cabinet/routes/admin_channels.py create mode 100644 app/cabinet/schemas/channel.py create mode 100644 app/database/crud/required_channel.py create mode 100644 app/handlers/admin/required_channels.py create mode 100644 app/handlers/channel_member.py create mode 100644 app/services/channel_subscription_service.py create mode 100644 migrations/alembic/versions/0008_add_required_channels.py create mode 100644 migrations/alembic/versions/0009_add_channel_id_index.py diff --git a/.env.example b/.env.example index e0ab9878..17dd707b 100644 --- a/.env.example +++ b/.env.example @@ -116,10 +116,8 @@ BLACKLIST_UPDATE_INTERVAL_HOURS=24 # Интервал обновле BLACKLIST_IGNORE_ADMINS=true # Игнорировать администраторов (из ADMIN_IDS) при проверке черного списка SUBSCRIPTION_RENEWAL_BALANCE_THRESHOLD_KOPEKS=20000 # Порог баланса (в копейках) для фильтра «готовы к продлению» -# Обязательная подписка на канал -CHANNEL_SUB_ID= # Опционально ID твоего канала (-100) +# Channel subscription settings (channels are managed via admin panel) CHANNEL_IS_REQUIRED_SUB=false # Обязательна ли подписка на канал -CHANNEL_LINK= # Опционально ссылка на канал CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE=true # Отключать триальные подписки при отписке от канала CHANNEL_REQUIRED_FOR_ALL=false # Требовать подписку на канал для ВСЕХ пользователей (платных и триальных) diff --git a/app/bot.py b/app/bot.py index dd657a2b..92221314 100644 --- a/app/bot.py +++ b/app/bot.py @@ -45,6 +45,7 @@ from app.handlers.admin import ( referrals as admin_referrals, remnawave as admin_remnawave, reports as admin_reports, + required_channels as admin_required_channels, rules as admin_rules, servers as admin_servers, statistics as admin_statistics, @@ -58,6 +59,7 @@ from app.handlers.admin import ( users as admin_users, welcome_text as admin_welcome_text, ) +from app.handlers.channel_member import register_handlers as register_channel_member_handlers from app.handlers.stars_payments import register_stars_handlers from app.middlewares.auth import AuthMiddleware from app.middlewares.blacklist import BlacklistMiddleware @@ -135,15 +137,11 @@ async def setup_bot() -> tuple[Bot, Dispatcher]: dp.callback_query.middleware(button_stats_middleware) logger.info('📊 ButtonStatsMiddleware активирован') - if settings.CHANNEL_IS_REQUIRED_SUB: - from app.middlewares.channel_checker import ChannelCheckerMiddleware + from app.middlewares.channel_checker import ChannelCheckerMiddleware - channel_checker_middleware = ChannelCheckerMiddleware() - dp.message.middleware(channel_checker_middleware) - dp.callback_query.middleware(channel_checker_middleware) - logger.info('🔒 Обязательная подписка включена - ChannelCheckerMiddleware активирован') - else: - logger.info('🔓 Обязательная подписка отключена - ChannelCheckerMiddleware не зарегистрирован') + channel_checker = ChannelCheckerMiddleware() + dp.message.middleware(channel_checker) + dp.callback_query.middleware(channel_checker) dp.message.middleware(AuthMiddleware()) dp.callback_query.middleware(AuthMiddleware()) dp.pre_checkout_query.middleware(AuthMiddleware()) @@ -194,6 +192,8 @@ async def setup_bot() -> tuple[Bot, Dispatcher]: admin_bulk_ban.register_bulk_ban_handlers(dp) admin_blacklist.register_blacklist_handlers(dp) admin_blocked_users.register_handlers(dp) + admin_required_channels.register_handlers(dp) + register_channel_member_handlers(dp) common.register_handlers(dp) register_stars_handlers(dp) user_contests.register_handlers(dp) diff --git a/app/cabinet/dependencies.py b/app/cabinet/dependencies.py index 10b1d7a0..17018721 100644 --- a/app/cabinet/dependencies.py +++ b/app/cabinet/dependencies.py @@ -1,9 +1,6 @@ """FastAPI dependencies for cabinet module.""" -import asyncio - import structlog -from aiogram import Bot from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.ext.asyncio import AsyncSession @@ -23,17 +20,6 @@ logger = structlog.get_logger(__name__) security = HTTPBearer(auto_error=False) -# Кешированный Bot для проверки подписки на канал -_channel_check_bot: Bot | None = None - - -def _get_channel_check_bot() -> Bot: - """Получить или создать Bot для проверки подписки на канал.""" - global _channel_check_bot - if _channel_check_bot is None: - _channel_check_bot = Bot(token=settings.BOT_TOKEN) - return _channel_check_bot - async def get_cabinet_db() -> AsyncSession: """Get database session for cabinet operations.""" @@ -163,42 +149,31 @@ async def get_current_cabinet_user( }, ) - # Check required channel subscription - ТОЛЬКО для Telegram юзеров - if settings.CHANNEL_IS_REQUIRED_SUB and settings.CHANNEL_SUB_ID: - # Пропускаем проверку для email-only юзеров (нет telegram_id) + # Check required channel subscription - Telegram users only + if settings.CHANNEL_IS_REQUIRED_SUB: + # Skip for email-only users (no telegram_id) if user.telegram_id is not None: - # Проверяем админа по telegram_id ИЛИ email + # Skip admin check is_admin = settings.is_admin( telegram_id=user.telegram_id, email=user.email if user.email_verified else None ) if not is_admin: - try: - bot = _get_channel_check_bot() - chat_member = await asyncio.wait_for( - bot.get_chat_member(chat_id=settings.CHANNEL_SUB_ID, user_id=user.telegram_id), - timeout=10.0, - ) - # Не закрываем сессию - бот переиспользуется + from app.services.channel_subscription_service import channel_subscription_service - if chat_member.status not in ['member', 'administrator', 'creator']: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail={ - 'code': 'channel_subscription_required', - 'message': 'Please subscribe to our channel to continue', - 'channel_link': settings.CHANNEL_LINK, - }, - ) - except HTTPException: - raise - except TimeoutError: - logger.warning('Timeout checking channel subscription for user', telegram_id=user.telegram_id) - # Don't block user if check times out - except Exception as e: - logger.warning( - 'Failed to check channel subscription for user', telegram_id=user.telegram_id, error=e + channels_with_status = await channel_subscription_service.get_channels_with_status(user.telegram_id) + is_subscribed = ( + all(ch['is_subscribed'] for ch in channels_with_status) if channels_with_status else True + ) + + if not is_subscribed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + 'code': 'channel_subscription_required', + 'message': 'Please subscribe to the required channels to continue', + 'channels': channels_with_status, + }, ) - # Don't block user if check fails return user diff --git a/app/cabinet/routes/__init__.py b/app/cabinet/routes/__init__.py index a0d51c4d..7a285b9f 100644 --- a/app/cabinet/routes/__init__.py +++ b/app/cabinet/routes/__init__.py @@ -7,6 +7,7 @@ from .admin_ban_system import router as admin_ban_system_router from .admin_broadcasts import router as admin_broadcasts_router from .admin_button_styles import router as admin_button_styles_router from .admin_campaigns import router as admin_campaigns_router +from .admin_channels import router as admin_channels_router from .admin_email_templates import router as admin_email_templates_router from .admin_partners import router as admin_partners_router from .admin_payment_methods import router as admin_payment_methods_router @@ -101,6 +102,7 @@ router.include_router(admin_updates_router) router.include_router(admin_traffic_router) router.include_router(admin_pinned_messages_router) router.include_router(admin_button_styles_router) +router.include_router(admin_channels_router) # WebSocket route router.include_router(websocket_router) diff --git a/app/cabinet/routes/admin_channels.py b/app/cabinet/routes/admin_channels.py new file mode 100644 index 00000000..6ba26c9c --- /dev/null +++ b/app/cabinet/routes/admin_channels.py @@ -0,0 +1,95 @@ +"""Admin API for managing required channels.""" + +import structlog +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database.crud.required_channel import ( + add_channel, + delete_channel, + get_all_channels, + toggle_channel, + update_channel, +) +from app.database.models import User +from app.services.channel_subscription_service import channel_subscription_service + +from ..dependencies import get_cabinet_db, get_current_admin_user +from ..schemas.channel import ( + ChannelCreateRequest, + ChannelListResponse, + ChannelResponse, + ChannelUpdateRequest, +) + + +logger = structlog.get_logger(__name__) + +router = APIRouter(prefix='/admin/channel-subscriptions', tags=['Cabinet Admin Channels']) + + +@router.get('', response_model=ChannelListResponse) +async def list_channels( + db: AsyncSession = Depends(get_cabinet_db), + _admin: User = Depends(get_current_admin_user), +) -> ChannelListResponse: + channels = await get_all_channels(db) + return ChannelListResponse( + items=[ChannelResponse.model_validate(ch) for ch in channels], + total=len(channels), + ) + + +@router.post('', response_model=ChannelResponse, status_code=201) +async def create_channel( + data: ChannelCreateRequest, + db: AsyncSession = Depends(get_cabinet_db), + _admin: User = Depends(get_current_admin_user), +) -> ChannelResponse: + # NOTE: REST API does not resolve @username to numeric channel ID (no Bot instance available). + # The bot FSM handler is the primary creation path and resolves automatically. + # When using REST API, callers should provide numeric channel IDs (e.g. -1001234567890) + # to ensure ChatMemberUpdated event matching works correctly. + ch = await add_channel(db, channel_id=data.channel_id, channel_link=data.channel_link, title=data.title) + await channel_subscription_service.invalidate_channels_cache() + return ChannelResponse.model_validate(ch) + + +@router.patch('/{channel_db_id}', response_model=ChannelResponse) +async def update_channel_endpoint( + channel_db_id: int, + data: ChannelUpdateRequest, + db: AsyncSession = Depends(get_cabinet_db), + _admin: User = Depends(get_current_admin_user), +) -> ChannelResponse: + update_data = data.model_dump(exclude_unset=True) + ch = await update_channel(db, channel_db_id, **update_data) + if not ch: + raise HTTPException(status_code=404, detail='Channel not found') + await channel_subscription_service.invalidate_channels_cache() + return ChannelResponse.model_validate(ch) + + +@router.post('/{channel_db_id}/toggle', response_model=ChannelResponse) +async def toggle_channel_endpoint( + channel_db_id: int, + db: AsyncSession = Depends(get_cabinet_db), + _admin: User = Depends(get_current_admin_user), +) -> ChannelResponse: + ch = await toggle_channel(db, channel_db_id) + if not ch: + raise HTTPException(status_code=404, detail='Channel not found') + await channel_subscription_service.invalidate_channels_cache() + return ChannelResponse.model_validate(ch) + + +@router.delete('/{channel_db_id}', status_code=204) +async def delete_channel_endpoint( + channel_db_id: int, + db: AsyncSession = Depends(get_cabinet_db), + _admin: User = Depends(get_current_admin_user), +) -> None: + ok = await delete_channel(db, channel_db_id) + if not ok: + raise HTTPException(status_code=404, detail='Channel not found') + await channel_subscription_service.invalidate_channels_cache() diff --git a/app/cabinet/schemas/channel.py b/app/cabinet/schemas/channel.py new file mode 100644 index 00000000..0fce290f --- /dev/null +++ b/app/cabinet/schemas/channel.py @@ -0,0 +1,78 @@ +"""Pydantic v2 schemas for channel subscription management.""" + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from app.database.crud.required_channel import validate_channel_id as _validate_channel_id_format + + +def _validate_channel_link_value(v: str | None) -> str | None: + """Shared channel_link validation: t.me URL, @username auto-convert, http->https upgrade.""" + if v is None: + return v + v = v.strip() + if v.startswith('http://t.me/'): + v = v.replace('http://', 'https://', 1) + if v.startswith('https://t.me/'): + return v + if v.startswith('@'): + return f'https://t.me/{v[1:]}' + raise ValueError('channel_link must be a t.me URL or @username') + + +class ChannelResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: int + channel_id: str + channel_link: str | None + title: str | None + is_active: bool + sort_order: int + + +class ChannelListResponse(BaseModel): + items: list[ChannelResponse] + total: int + + +class ChannelCreateRequest(BaseModel): + channel_id: str + channel_link: str | None = None + title: str | None = Field(None, max_length=255) + + @field_validator('channel_id') + @classmethod + def validate_channel_id(cls, v: str) -> str: + return _validate_channel_id_format(v) + + @field_validator('channel_link') + @classmethod + def validate_channel_link(cls, v: str | None) -> str | None: + return _validate_channel_link_value(v) + + +class ChannelUpdateRequest(BaseModel): + channel_id: str | None = None + channel_link: str | None = None + title: str | None = Field(None, max_length=255) + is_active: bool | None = None + sort_order: int | None = None + + @field_validator('channel_id') + @classmethod + def validate_channel_id(cls, v: str | None) -> str | None: + if v is None: + return v + return _validate_channel_id_format(v) + + @field_validator('channel_link') + @classmethod + def validate_channel_link(cls, v: str | None) -> str | None: + return _validate_channel_link_value(v) + + +class ChannelSubscriptionStatus(BaseModel): + channel_id: str + channel_link: str | None + title: str | None + is_subscribed: bool diff --git a/app/config.py b/app/config.py index 335051a9..34d4a9d5 100644 --- a/app/config.py +++ b/app/config.py @@ -66,8 +66,6 @@ class Settings(BaseSettings): ADMIN_REPORTS_TOPIC_ID: int | None = None ADMIN_REPORTS_SEND_TIME: str | None = None - CHANNEL_SUB_ID: str | None = None - CHANNEL_LINK: str | None = None CHANNEL_IS_REQUIRED_SUB: bool = False CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE: bool = True CHANNEL_REQUIRED_FOR_ALL: bool = False diff --git a/app/database/crud/required_channel.py b/app/database/crud/required_channel.py new file mode 100644 index 00000000..766a21f4 --- /dev/null +++ b/app/database/crud/required_channel.py @@ -0,0 +1,225 @@ +import re +from datetime import UTC, datetime + +import structlog +from aiogram import Bot +from sqlalchemy import delete, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database.models import RequiredChannel, UserChannelSubscription + + +logger = structlog.get_logger(__name__) + +# Explicit allowlist of fields that can be updated via update_channel() +_UPDATABLE_FIELDS = frozenset({'channel_id', 'channel_link', 'title', 'is_active', 'sort_order'}) + +# Validation patterns for channel_id +_CHANNEL_ID_USERNAME = re.compile(r'^@[a-zA-Z][a-zA-Z0-9_]{3,30}$') +_CHANNEL_ID_NUMERIC = re.compile(r'^-100\d{10,13}$') + + +def validate_channel_id(channel_id: str) -> str: + """Validate and normalize channel_id. Raises ValueError on invalid input.""" + channel_id = channel_id.strip() + if _CHANNEL_ID_USERNAME.match(channel_id) or _CHANNEL_ID_NUMERIC.match(channel_id): + return channel_id + raise ValueError( + f'Invalid channel_id format: {channel_id!r}. Must be @username (4-31 chars) or numeric ID like -100XXXXXXXXXX' + ) + + +async def resolve_channel_id(bot: Bot, channel_id: str) -> str: + """Resolve @username to numeric channel ID via Telegram API. + + Required because ChatMemberUpdated events use numeric IDs. + If channel_id is already numeric, returns as-is. + """ + if _CHANNEL_ID_NUMERIC.match(channel_id): + return channel_id + try: + chat = await bot.get_chat(channel_id) + resolved = str(chat.id) + logger.info('Resolved channel username to numeric ID', username=channel_id, numeric_id=resolved) + return resolved + except Exception as e: + logger.error('Failed to resolve channel ID', channel_id=channel_id, error=e) + raise ValueError(f'Cannot resolve channel {channel_id}: {e}') from e + + +# -- RequiredChannel CRUD -------------------------------------------------------- + + +async def get_active_channels(db: AsyncSession) -> list[RequiredChannel]: + """Get all active required channels (sorted by sort_order).""" + result = await db.execute( + select(RequiredChannel) + .where(RequiredChannel.is_active.is_(True)) + .order_by(RequiredChannel.sort_order, RequiredChannel.id) + ) + return list(result.scalars().all()) + + +async def get_all_channels(db: AsyncSession) -> list[RequiredChannel]: + """Get all required channels (including inactive).""" + result = await db.execute(select(RequiredChannel).order_by(RequiredChannel.sort_order, RequiredChannel.id)) + return list(result.scalars().all()) + + +async def get_channel_by_id(db: AsyncSession, channel_db_id: int) -> RequiredChannel | None: + result = await db.execute(select(RequiredChannel).where(RequiredChannel.id == channel_db_id)) + return result.scalar_one_or_none() + + +async def get_channel_by_channel_id(db: AsyncSession, channel_id: str) -> RequiredChannel | None: + result = await db.execute(select(RequiredChannel).where(RequiredChannel.channel_id == channel_id)) + return result.scalar_one_or_none() + + +async def add_channel( + db: AsyncSession, + channel_id: str, + channel_link: str | None = None, + title: str | None = None, +) -> RequiredChannel: + channel_id = validate_channel_id(channel_id) + channel = RequiredChannel( + channel_id=channel_id, + channel_link=channel_link, + title=title, + ) + db.add(channel) + await db.commit() + await db.refresh(channel) + return channel + + +async def update_channel( + db: AsyncSession, + channel_db_id: int, + **kwargs, +) -> RequiredChannel | None: + """Update channel fields. Only fields in _UPDATABLE_FIELDS are accepted.""" + channel = await get_channel_by_id(db, channel_db_id) + if not channel: + return None + + for key, value in kwargs.items(): + if key not in _UPDATABLE_FIELDS: + logger.warning('Rejected update of non-updatable field', field=key) + continue + if key == 'channel_id' and value is not None: + value = validate_channel_id(value) + setattr(channel, key, value) + + channel.updated_at = datetime.now(UTC) + await db.commit() + await db.refresh(channel) + return channel + + +async def delete_channel(db: AsyncSession, channel_db_id: int) -> bool: + channel = await get_channel_by_id(db, channel_db_id) + if not channel: + return False + # Also clean up user subscriptions for this channel + await db.execute(delete(UserChannelSubscription).where(UserChannelSubscription.channel_id == channel.channel_id)) + await db.delete(channel) + await db.commit() + return True + + +async def toggle_channel(db: AsyncSession, channel_db_id: int) -> RequiredChannel | None: + channel = await get_channel_by_id(db, channel_db_id) + if not channel: + return None + channel.is_active = not channel.is_active + channel.updated_at = datetime.now(UTC) + await db.commit() + await db.refresh(channel) + return channel + + +# -- UserChannelSubscription CRUD ------------------------------------------------ + + +async def upsert_user_channel_sub( + db: AsyncSession, + telegram_id: int, + channel_id: str, + is_member: bool, +) -> None: + """Upsert user subscription status (PostgreSQL ON CONFLICT).""" + now = datetime.now(UTC) # Single timestamp for both INSERT and UPDATE + stmt = ( + pg_insert(UserChannelSubscription) + .values( + telegram_id=telegram_id, + channel_id=channel_id, + is_member=is_member, + checked_at=now, + ) + .on_conflict_do_update( + constraint='uq_user_channel_sub', + set_={ + 'is_member': is_member, + 'checked_at': now, + }, + ) + ) + await db.execute(stmt) + # NOTE: caller is responsible for commit (allows batching) + + +async def get_user_channel_subs( + db: AsyncSession, + telegram_id: int, +) -> list[UserChannelSubscription]: + """Get all channel subscriptions for a user.""" + result = await db.execute(select(UserChannelSubscription).where(UserChannelSubscription.telegram_id == telegram_id)) + return list(result.scalars().all()) + + +async def get_user_channel_sub( + db: AsyncSession, + telegram_id: int, + channel_id: str, +) -> UserChannelSubscription | None: + result = await db.execute( + select(UserChannelSubscription).where( + UserChannelSubscription.telegram_id == telegram_id, + UserChannelSubscription.channel_id == channel_id, + ) + ) + return result.scalar_one_or_none() + + +async def bulk_upsert_user_subs( + db: AsyncSession, + telegram_id: int, + subs: dict[str, bool], # {channel_id: is_member} +) -> None: + """Batch upsert user subscriptions with single multi-row INSERT.""" + if not subs: + return + now = datetime.now(UTC) + values = [ + { + 'telegram_id': telegram_id, + 'channel_id': channel_id, + 'is_member': is_member, + 'checked_at': now, + } + for channel_id, is_member in subs.items() + ] + stmt = pg_insert(UserChannelSubscription).values(values) + stmt = stmt.on_conflict_do_update( + constraint='uq_user_channel_sub', + set_={ + 'is_member': stmt.excluded.is_member, + 'checked_at': stmt.excluded.checked_at, + }, + ) + await db.execute(stmt) + await db.commit() diff --git a/app/database/models.py b/app/database/models.py index 730c902f..dc4907d6 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -2810,3 +2810,48 @@ class PaymentMethodConfig(Base): def __repr__(self) -> str: return f"" + + +class RequiredChannel(Base): + """Channels that users must subscribe to in order to use the bot.""" + + __tablename__ = 'required_channels' + + id = Column(Integer, primary_key=True, autoincrement=True) + channel_id = Column(String(100), unique=True, nullable=False) # @username or -100xxx (always string) + channel_link = Column(String(500), nullable=True) # https://t.me/xxx + title = Column(String(255), nullable=True) # Display name + is_active = Column(Boolean, nullable=False, server_default='true') + sort_order = Column(Integer, nullable=False, server_default='0') + created_at = Column(AwareDateTime(), nullable=False, server_default=func.now()) + updated_at = Column(AwareDateTime(), nullable=True, onupdate=func.now()) + + def __repr__(self) -> str: + return f'' + + +class UserChannelSubscription(Base): + """Cache of user subscription status per required channel.""" + + __tablename__ = 'user_channel_subscriptions' + + id = Column(Integer, primary_key=True, autoincrement=True) + telegram_id = Column(BigInteger, nullable=False) + channel_id = Column(String(100), nullable=False) # matches RequiredChannel.channel_id + is_member = Column(Boolean, nullable=False, server_default='false') + checked_at = Column(AwareDateTime(), nullable=False, server_default=func.now()) + + __table_args__ = ( + UniqueConstraint('telegram_id', 'channel_id', name='uq_user_channel_sub'), + # UniqueConstraint creates its own index; only add telegram_id index for + # "get all subs for user" queries + Index('ix_user_channel_sub_telegram_id', 'telegram_id'), + # Standalone channel_id index for delete_channel() bulk DELETE + Index('ix_user_channel_sub_channel_id', 'channel_id'), + ) + + def __repr__(self) -> str: + return ( + f'' + ) diff --git a/app/handlers/admin/__init__.py b/app/handlers/admin/__init__.py index 88db9e56..4640cf94 100644 --- a/app/handlers/admin/__init__.py +++ b/app/handlers/admin/__init__.py @@ -24,6 +24,7 @@ from . import ( referrals, remnawave, reports, + required_channels, rules, servers, statistics, diff --git a/app/handlers/admin/bot_configuration.py b/app/handlers/admin/bot_configuration.py index 6705a3d0..65610535 100644 --- a/app/handlers/admin/bot_configuration.py +++ b/app/handlers/admin/bot_configuration.py @@ -318,10 +318,11 @@ def _get_group_status(group_key: str) -> tuple[str, str]: if key == 'core': token_ok = bool(getattr(settings, 'BOT_TOKEN', '')) - channel_ok = bool(settings.CHANNEL_LINK or not settings.CHANNEL_IS_REQUIRED_SUB) - if token_ok and channel_ok: + # Channel subscription channels are now managed via DB (admin panel), + # not a single CHANNEL_LINK setting. Dashboard cannot async-query DB here. + if token_ok: return '🟢', 'Бот готов к работе' - return '🟡', 'Проверьте токен и обязательную подписку' + return '🟡', 'Проверьте токен бота' if key == 'subscriptions': price_ready = settings.PRICE_30_DAYS > 0 and settings.AVAILABLE_SUBSCRIPTION_PERIODS diff --git a/app/handlers/admin/monitoring.py b/app/handlers/admin/monitoring.py index df9aa3d9..7940e1a6 100644 --- a/app/handlers/admin/monitoring.py +++ b/app/handlers/admin/monitoring.py @@ -137,13 +137,16 @@ def _build_notification_settings_view(language: str): return summary_text, keyboard -def _build_notification_preview_message(language: str, notification_type: str): +async def _build_notification_preview_message(language: str, notification_type: str): texts = get_texts(language) now = datetime.now(UTC) price_30_days = settings.format_price(settings.PRICE_30_DAYS) from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup + from app.keyboards.inline import get_channel_sub_keyboard + from app.services.channel_subscription_service import channel_subscription_service + header = '🧪 Тестовое уведомление мониторинга\n\n' if notification_type == 'trial_channel_unsubscribed': @@ -157,25 +160,9 @@ def _build_notification_preview_message(language: str, notification_type: str): ) check_button = texts.t('CHANNEL_CHECK_BUTTON', '✅ Я подписался') message = template.format(check_button=check_button) - buttons: list[list[InlineKeyboardButton]] = [] - if settings.CHANNEL_LINK: - buttons.append( - [ - InlineKeyboardButton( - text=texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться'), - url=settings.CHANNEL_LINK, - ) - ] - ) - buttons.append( - [ - InlineKeyboardButton( - text=check_button, - callback_data='sub_channel_check', - ) - ] - ) - keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) + # Use all required channels for the preview keyboard + required_channels = await channel_subscription_service.get_required_channels() + keyboard = get_channel_sub_keyboard(required_channels, language=language) elif notification_type == 'expired_1d': template = texts.get( 'SUBSCRIPTION_EXPIRED_1D', @@ -307,7 +294,7 @@ def _build_notification_preview_message(language: str, notification_type: str): async def _send_notification_preview(bot, chat_id: int, language: str, notification_type: str) -> None: - message, keyboard = _build_notification_preview_message(language, notification_type) + message, keyboard = await _build_notification_preview_message(language, notification_type) await bot.send_message( chat_id, message, diff --git a/app/handlers/admin/required_channels.py b/app/handlers/admin/required_channels.py new file mode 100644 index 00000000..75e84b81 --- /dev/null +++ b/app/handlers/admin/required_channels.py @@ -0,0 +1,285 @@ +"""Admin handler for managing required channel subscriptions.""" + +import structlog +from aiogram import Bot, F, Router +from aiogram.fsm.context import FSMContext +from aiogram.fsm.state import State, StatesGroup +from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message + +from app.database.crud.required_channel import ( + add_channel, + delete_channel, + get_all_channels, + get_channel_by_id, + resolve_channel_id, + toggle_channel, + validate_channel_id, +) +from app.database.database import AsyncSessionLocal +from app.services.channel_subscription_service import channel_subscription_service +from app.utils.decorators import admin_required + + +logger = structlog.get_logger(__name__) + +router = Router(name='admin_required_channels') + + +class AddChannelStates(StatesGroup): + waiting_channel_id = State() + waiting_channel_link = State() + waiting_channel_title = State() + + +# -- List channels ---------------------------------------------------------------- + + +def _channels_keyboard(channels: list) -> InlineKeyboardMarkup: + buttons = [] + for ch in channels: + status = 'ON' if ch.is_active else 'OFF' + title = ch.title or ch.channel_id + buttons.append( + [ + InlineKeyboardButton( + text=f'{status} {title}', + callback_data=f'reqch:view:{ch.id}', + ) + ] + ) + buttons.append([InlineKeyboardButton(text='+ Add channel', callback_data='reqch:add')]) + buttons.append([InlineKeyboardButton(text='< Back', callback_data='admin:back')]) + return InlineKeyboardMarkup(inline_keyboard=buttons) + + +def _channel_detail_keyboard(channel_id: int, is_active: bool) -> InlineKeyboardMarkup: + toggle_text = 'Disable' if is_active else 'Enable' + return InlineKeyboardMarkup( + inline_keyboard=[ + [InlineKeyboardButton(text=toggle_text, callback_data=f'reqch:toggle:{channel_id}')], + [InlineKeyboardButton(text='Delete', callback_data=f'reqch:delete:{channel_id}')], + [InlineKeyboardButton(text='< Back to list', callback_data='reqch:list')], + ] + ) + + +@router.callback_query(F.data == 'reqch:list') +@admin_required +async def show_channels_list(callback: CallbackQuery, **kwargs) -> None: + async with AsyncSessionLocal() as db: + channels = await get_all_channels(db) + + if not channels: + text = 'Required Channels\n\nNo channels configured. Click "Add" to create one.' + else: + lines = ['Required Channels\n'] + for ch in channels: + status = 'ON' if ch.is_active else 'OFF' + title = ch.title or ch.channel_id + lines.append(f'{status} {ch.channel_id} -- {title}') + text = '\n'.join(lines) + + await callback.message.edit_text(text, reply_markup=_channels_keyboard(channels)) + await callback.answer() + + +@router.callback_query(F.data.startswith('reqch:view:')) +@admin_required +async def view_channel(callback: CallbackQuery, **kwargs) -> None: + try: + channel_db_id = int(callback.data.split(':')[2]) + except (ValueError, IndexError): + await callback.answer('Invalid channel ID', show_alert=True) + return + async with AsyncSessionLocal() as db: + ch = await get_channel_by_id(db, channel_db_id) + + if not ch: + await callback.answer('Channel not found', show_alert=True) + return + + status = 'Active' if ch.is_active else 'Disabled' + text = ( + f'{ch.title or "Untitled"}\n\n' + f'ID: {ch.channel_id}\n' + f'Link: {ch.channel_link or "--"}\n' + f'Status: {status}\n' + f'Sort order: {ch.sort_order}' + ) + + await callback.message.edit_text(text, reply_markup=_channel_detail_keyboard(ch.id, ch.is_active)) + await callback.answer() + + +# -- Toggle / Delete --------------------------------------------------------------- + + +@router.callback_query(F.data.startswith('reqch:toggle:')) +@admin_required +async def toggle_channel_handler(callback: CallbackQuery, **kwargs) -> None: + try: + channel_db_id = int(callback.data.split(':')[2]) + except (ValueError, IndexError): + await callback.answer('Invalid channel ID', show_alert=True) + return + async with AsyncSessionLocal() as db: + ch = await toggle_channel(db, channel_db_id) + + if ch: + await channel_subscription_service.invalidate_channels_cache() + status = 'enabled' if ch.is_active else 'disabled' + await callback.answer(f'Channel {status}', show_alert=True) + + # Refresh list + async with AsyncSessionLocal() as db: + channels = await get_all_channels(db) + await callback.message.edit_text( + 'Required Channels', + reply_markup=_channels_keyboard(channels), + ) + + +@router.callback_query(F.data.startswith('reqch:delete:')) +@admin_required +async def delete_channel_handler(callback: CallbackQuery, **kwargs) -> None: + try: + channel_db_id = int(callback.data.split(':')[2]) + except (ValueError, IndexError): + await callback.answer('Invalid channel ID', show_alert=True) + return + async with AsyncSessionLocal() as db: + ok = await delete_channel(db, channel_db_id) + + if ok: + await channel_subscription_service.invalidate_channels_cache() + await callback.answer('Channel deleted', show_alert=True) + else: + await callback.answer('Delete failed', show_alert=True) + + async with AsyncSessionLocal() as db: + channels = await get_all_channels(db) + await callback.message.edit_text( + 'Required Channels', + reply_markup=_channels_keyboard(channels), + ) + + +# -- Add channel flow -------------------------------------------------------------- + + +@router.callback_query(F.data == 'reqch:add') +@admin_required +async def start_add_channel(callback: CallbackQuery, state: FSMContext, **kwargs) -> None: + await state.set_state(AddChannelStates.waiting_channel_id) + await callback.message.edit_text( + 'Add Channel\n\nSend channel ID (e.g. @mychannel or -1001234567890):' + ) + await callback.answer() + + +@router.message(AddChannelStates.waiting_channel_id) +@admin_required +async def process_channel_id(message: Message, state: FSMContext, **kwargs) -> None: + if not message.text: + await message.answer('Please send a text message.') + return + channel_id = message.text.strip() + + # Validate channel_id format + try: + channel_id = validate_channel_id(channel_id) + except ValueError as e: + await message.answer(f'Invalid format. {e}\n\nTry again:') + return + + # Resolve @username to numeric ID (ChatMemberUpdated events use numeric IDs) + original_channel_id = channel_id + bot: Bot = message.bot + try: + channel_id = await resolve_channel_id(bot, channel_id) + except ValueError as e: + await message.answer(f'Cannot verify channel: {e}\n\nMake sure the bot is admin in this channel. Try again:') + return + + await state.update_data(channel_id=channel_id, original_channel_id=original_channel_id) + await state.set_state(AddChannelStates.waiting_channel_link) + await message.answer( + f'Channel: {channel_id}\n\n' + 'Now send the channel link (e.g. https://t.me/mychannel)\n' + 'Or send - to skip:' + ) + + +@router.message(AddChannelStates.waiting_channel_link) +@admin_required +async def process_channel_link(message: Message, state: FSMContext, **kwargs) -> None: + if not message.text: + await message.answer('Please send a text message.') + return + link = message.text.strip() + if link == '-': + link = None + + if link is not None: + # Validate and normalize channel link + if not link.startswith(('https://t.me/', 'http://t.me/', '@')): + await message.answer('Link must be a t.me URL or @username. Try again:') + return + if link.startswith('@'): + link = f'https://t.me/{link[1:]}' + if link.startswith('http://'): + link = link.replace('http://', 'https://', 1) + + await state.update_data(channel_link=link) + await state.set_state(AddChannelStates.waiting_channel_title) + await message.answer( + 'Send display name for the channel (e.g. Project News)\nOr send - to skip:' + ) + + +@router.message(AddChannelStates.waiting_channel_title) +@admin_required +async def process_channel_title(message: Message, state: FSMContext, **kwargs) -> None: + if not message.text: + await message.answer('Please send a text message.') + return + title = message.text.strip() + if title == '-': + title = None + + data = await state.get_data() + await state.clear() + + # Use original @username as title fallback when channel was resolved to numeric + original_id = data.get('original_channel_id') + if not title and original_id and original_id != data['channel_id']: + title = original_id + + async with AsyncSessionLocal() as db: + try: + ch = await add_channel( + db, + channel_id=data['channel_id'], + channel_link=data.get('channel_link'), + title=title, + ) + await channel_subscription_service.invalidate_channels_cache() + + text = ( + 'Channel added!\n\n' + f'ID: {ch.channel_id}\n' + f'Link: {ch.channel_link or "--"}\n' + f'Title: {ch.title or "--"}' + ) + except Exception as e: + text = 'Error adding channel. Please try again.' + logger.error('Error adding channel', error=e) + + async with AsyncSessionLocal() as db: + channels = await get_all_channels(db) + + await message.answer(text, reply_markup=_channels_keyboard(channels)) + + +def register_handlers(dp_router: Router) -> None: + dp_router.include_router(router) diff --git a/app/handlers/channel_member.py b/app/handlers/channel_member.py new file mode 100644 index 00000000..5345d421 --- /dev/null +++ b/app/handlers/channel_member.py @@ -0,0 +1,184 @@ +"""ChatMemberUpdated event handler for real-time channel subscription tracking. + +KEY COMPONENT for scalability: the bot receives push notifications from Telegram +when users join/leave channels, instead of polling via getChatMember. +Requirement: bot must be admin in each required channel. + +IMPORTANT: Events are FILTERED to only process required channels. +Without filtering, the bot would process events from ALL channels it admins. +""" + +from datetime import UTC, datetime + +import structlog +from aiogram import Bot, Router +from aiogram.filters import IS_MEMBER, IS_NOT_MEMBER, ChatMemberUpdatedFilter +from aiogram.types import ChatMemberUpdated + +from app.config import settings +from app.database.crud.subscription import deactivate_subscription, is_active_paid_subscription, reactivate_subscription +from app.database.crud.user import get_user_by_telegram_id +from app.database.database import AsyncSessionLocal +from app.database.models import SubscriptionStatus, UserStatus +from app.keyboards.inline import get_channel_sub_keyboard +from app.localization.loader import DEFAULT_LANGUAGE +from app.localization.texts import get_texts +from app.services.channel_subscription_service import channel_subscription_service +from app.services.subscription_service import SubscriptionService + + +logger = structlog.get_logger(__name__) + +router = Router(name='channel_member') + + +async def _is_required_channel(channel_id: str) -> bool: + """Check if the channel_id is one of our required channels.""" + required_ids = await channel_subscription_service.get_required_channel_ids() + return channel_id in required_ids + + +@router.chat_member(ChatMemberUpdatedFilter(member_status_changed=IS_NOT_MEMBER >> IS_MEMBER)) +async def on_user_joined_channel(event: ChatMemberUpdated, bot: Bot) -> None: + """User subscribed to a channel -- update cache and reactivate VPN if applicable.""" + user = event.new_chat_member.user + channel_id = str(event.chat.id) # Normalize int to str (DB stores string) + + # FILTER: Only process events for required channels + if not await _is_required_channel(channel_id): + return + + await channel_subscription_service.on_user_joined(user.id, channel_id) + + # Check if user is now subscribed to ALL required channels + if not settings.CHANNEL_IS_REQUIRED_SUB: + return + + is_all_subscribed = await channel_subscription_service.is_user_subscribed_to_all(user.id) + if not is_all_subscribed: + return # Still missing some channels + + # Reactivate subscription if it was disabled due to channel unsubscribe + if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE and not settings.CHANNEL_REQUIRED_FOR_ALL: + return + + async with AsyncSessionLocal() as db: + try: + db_user = await get_user_by_telegram_id(db, user.id) + if not db_user or not db_user.subscription: + return + if db_user.status == UserStatus.BLOCKED.value: + return + + subscription = db_user.subscription + if subscription.status != SubscriptionStatus.DISABLED.value: + return + + # Don't reactivate expired subscriptions + if subscription.end_date and subscription.end_date <= datetime.now(UTC): + return + + await reactivate_subscription(db, subscription) + logger.info('Subscription reactivated via channel event', telegram_id=user.id) + + # Re-enable in RemnaWave panel + if db_user.remnawave_uuid: + service = SubscriptionService() + try: + await service.enable_remnawave_user(db_user.remnawave_uuid) + except Exception as api_error: + logger.error('Failed to enable RemnaWave user', error=api_error) + + # Notify the user + try: + texts = get_texts(db_user.language or DEFAULT_LANGUAGE) + notification_text = texts.t( + 'SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE', + 'Your subscription has been restored! Thank you for subscribing to the channels.', + ) + await bot.send_message(user.id, notification_text) + except Exception as notify_error: + logger.warning('Failed to send notification', telegram_id=user.id, error=notify_error) + + await db.commit() + except Exception as e: + logger.error('Error reactivating subscription on channel join', error=e) + await db.rollback() + + +@router.chat_member(ChatMemberUpdatedFilter(member_status_changed=IS_MEMBER >> IS_NOT_MEMBER)) +async def on_user_left_channel(event: ChatMemberUpdated, bot: Bot) -> None: + """User unsubscribed from a channel -- update cache and deactivate VPN if applicable.""" + user = event.old_chat_member.user + channel_id = str(event.chat.id) # Normalize int to str (DB stores string) + + # FILTER: Only process events for required channels + if not await _is_required_channel(channel_id): + return + + await channel_subscription_service.on_user_left(user.id, channel_id) + + if not settings.CHANNEL_IS_REQUIRED_SUB: + return + + # Skip admins -- never deactivate admin subscriptions + if settings.is_admin(user.id): + return + + if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE and not settings.CHANNEL_REQUIRED_FOR_ALL: + return + + async with AsyncSessionLocal() as db: + try: + db_user = await get_user_by_telegram_id(db, user.id) + if not db_user or not db_user.subscription: + return + + subscription = db_user.subscription + if subscription.status != SubscriptionStatus.ACTIVE.value: + return + + # CHANNEL_REQUIRED_FOR_ALL: deactivate regardless of trial status + # CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE: only deactivate trial subscriptions + if settings.CHANNEL_REQUIRED_FOR_ALL: + pass # Deactivate any active subscription + elif not subscription.is_trial: + return # Not a trial -- skip + + # Guard against paid subscriptions (user paid money, don't punish) + if is_active_paid_subscription(subscription): + return + + await deactivate_subscription(db, subscription) + logger.info('Subscription deactivated via channel event', telegram_id=user.id) + + # Disable in RemnaWave panel + if db_user.remnawave_uuid: + service = SubscriptionService() + try: + await service.disable_remnawave_user(db_user.remnawave_uuid) + except Exception as api_error: + logger.error('Failed to disable RemnaWave user', error=api_error) + + # Notify the user with channel subscription keyboard + try: + texts = get_texts(db_user.language or DEFAULT_LANGUAGE) + unsub_channels = await channel_subscription_service.get_unsubscribed_channels(user.id) + notification_text = texts.t( + 'SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE', + 'Your subscription has been paused because you left a required channel.', + ) + channel_kb = get_channel_sub_keyboard(unsub_channels, language=db_user.language or DEFAULT_LANGUAGE) + await bot.send_message(user.id, notification_text, reply_markup=channel_kb) + except Exception as notify_error: + logger.warning('Failed to send notification', telegram_id=user.id, error=notify_error) + + await db.commit() + except Exception as e: + logger.error('Error deactivating subscription on channel leave', error=e) + await db.rollback() + + +def register_handlers(dp_router: Router) -> None: + """Register channel member event handlers on the dispatcher/router.""" + dp_router.include_router(router) diff --git a/app/handlers/start.py b/app/handlers/start.py index 56888b12..721815c7 100644 --- a/app/handlers/start.py +++ b/app/handlers/start.py @@ -2,7 +2,6 @@ from datetime import UTC, datetime import structlog from aiogram import Bot, Dispatcher, F, types -from aiogram.enums import ChatMemberStatus from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError from aiogram.filters import Command, StateFilter from aiogram.fsm.context import FSMContext @@ -37,6 +36,7 @@ from app.middlewares.channel_checker import ( ) from app.services.admin_notification_service import AdminNotificationService from app.services.campaign_service import AdvertisingCampaignService +from app.services.channel_subscription_service import channel_subscription_service from app.services.main_menu_button_service import MainMenuButtonService from app.services.pinned_message_service import ( deliver_pinned_message_to_user, @@ -1866,20 +1866,22 @@ async def required_sub_channel_check( texts = get_texts(language) - chat_member = await bot.get_chat_member(chat_id=settings.CHANNEL_SUB_ID, user_id=query.from_user.id) + # Ensure bot is set on service + if not channel_subscription_service.bot: + channel_subscription_service.bot = bot - if chat_member.status not in [ - ChatMemberStatus.MEMBER, - ChatMemberStatus.ADMINISTRATOR, - ChatMemberStatus.CREATOR, - ]: + # Invalidate cache for fresh check (user just clicked "I subscribed") + await channel_subscription_service.invalidate_user_cache(query.from_user.id) + + is_subscribed = await channel_subscription_service.is_user_subscribed_to_all(query.from_user.id) + if not is_subscribed: # НЕ удаляем payload - пользователь может попробовать снова после подписки logger.info( - "📦 CHANNEL CHECK: Подписка не подтверждена, payload '' сохранён для следующей попытки", + 'CHANNEL CHECK: Подписка не подтверждена, payload сохранён для следующей попытки', pending_start_payload=pending_start_payload, ) return await query.answer( - texts.t('CHANNEL_SUBSCRIBE_REQUIRED_ALERT', '❌ Вы не подписались на канал!'), + texts.t('CHANNEL_SUBSCRIBE_REQUIRED_ALERT', 'Please subscribe to all required channels first!'), show_alert=True, ) diff --git a/app/keyboards/inline.py b/app/keyboards/inline.py index 6a581b42..b70e1975 100644 --- a/app/keyboards/inline.py +++ b/app/keyboards/inline.py @@ -275,22 +275,37 @@ def get_privacy_policy_keyboard(language: str = DEFAULT_LANGUAGE) -> InlineKeybo def get_channel_sub_keyboard( - channel_link: str | None, + channels: list[dict] | str | None = None, language: str = DEFAULT_LANGUAGE, ) -> InlineKeyboardMarkup: - texts = get_texts(language) + """Subscription keyboard for required channels. + Args: + channels: List of dicts with 'channel_link' and 'title' keys, + OR a string (legacy single channel_link for backwards compat). + language: Locale code for button text. + """ + texts = get_texts(language) buttons: list[list[InlineKeyboardButton]] = [] - if channel_link: - buttons.append( - [ - InlineKeyboardButton( - text=texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться'), - url=channel_link, - ) - ] - ) + if isinstance(channels, str): + # Legacy: single channel link string + if channels: + buttons.append( + [ + InlineKeyboardButton( + text=texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться'), + url=channels, + ) + ] + ) + elif isinstance(channels, list): + for ch in channels: + link = ch.get('channel_link') + title = ch.get('title') + if link: + label = title or texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться') + buttons.append([InlineKeyboardButton(text=label, url=link)]) buttons.append( [ diff --git a/app/localization/locales/en.json b/app/localization/locales/en.json index 6b4c159d..5f262c69 100644 --- a/app/localization/locales/en.json +++ b/app/localization/locales/en.json @@ -930,9 +930,9 @@ "CHANGE_DEVICES_SUCCESS_INCREASE": "\n✅ Device limit increased!\n\n📱 Was: {old_count} → Now: {new_count}\n💰 Charged: {amount}\n", "CHANGE_DEVICES_TITLE": "📱 Change device limit", "CHANNEL_CHECK_BUTTON": "✅ I have joined", - "CHANNEL_REQUIRED_TEXT": "🔒 Please join the announcement channel to access the bot, then press the button below.", + "CHANNEL_REQUIRED_TEXT": "Please subscribe to the required channels and then press the button below.", "CHANNEL_SUBSCRIBE_BUTTON": "🔗 Subscribe", - "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "❌ You haven't joined the channel!", + "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "Please subscribe to all required channels first!", "CHANNEL_SUBSCRIBE_THANKS": "✅ Thanks for subscribing", "CHECK_STATUS_BUTTON": "📊 Check status", "CHECK_STATUS_NO_CHANGES": "Status has not changed", @@ -1431,10 +1431,12 @@ "SUBSCRIPTION_NO_SERVERS": "No servers", "SUBSCRIPTION_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Balance: {balance}\n📱 Subscription: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Subscription details\n🎭 Type: {subscription_type}\n📅 Valid until: {end_date}\n⏰ Time left: {time_left}\n📈 Traffic: {traffic}\n🌍 Servers: {servers}\n📱 Devices: {devices_used} / {device_limit}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Balance: {balance}\n📱 Subscription: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Subscription details\n🎭 Type: {subscription_type}\n📈 Traffic: {traffic}\n🌍 Servers: {servers}\n📱 Devices: {devices_used} / {device_limit}", + "SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE": "Your subscription has been paused because you left a required channel.\n\nSubscribe to all channels to restore VPN access.", "SUBSCRIPTION_PROMO_DISCOUNT_HINT": "⚡ Extra {percent}% discount is active and will apply automatically. It stacks with other discounts.", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE": "⚡ Extra discount {percent}%: -{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER": "⏳ Discount active for {time_left}\n{bar}", "SUBSCRIPTION_PURCHASED": "🎉 Subscription purchased successfully!", + "SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE": "Your subscription has been restored!\n\nThank you for subscribing to the channels. VPN is active again.", "SUBSCRIPTION_SETTINGS_BUTTON": "⚙️ Settings", "SUBSCRIPTION_SETTINGS_OVERVIEW": "⚙️ Subscription settings\n\n📊 Current parameters:\n🌐 Countries: {countries_count}\n📈 Traffic: {traffic_used} / {traffic_limit}\n📱 Devices: {devices_used} / {devices_limit}\n\nChoose what you want to change:", "SUBSCRIPTION_SETTINGS_PAID_ONLY": "⚠️ Settings are available only for paid subscriptions", diff --git a/app/localization/locales/fa.json b/app/localization/locales/fa.json index b147ae14..debb9020 100644 --- a/app/localization/locales/fa.json +++ b/app/localization/locales/fa.json @@ -951,9 +951,9 @@ "CHANGE_DEVICES_TITLE": "📱 تغییر تعداد دستگاه", "CHANGE_TARIFF_BUTTON": "📦 تعرفه", "CHANNEL_CHECK_BUTTON": "✅ عضو شدم", - "CHANNEL_REQUIRED_TEXT": "🔒 برای استفاده از ربات در کانال خبری عضو شوید، سپس دکمه زیر بزنید.", + "CHANNEL_REQUIRED_TEXT": "لطفاً در کانال‌های اجباری عضو شوید و سپس دکمه زیر را فشار دهید.", "CHANNEL_SUBSCRIBE_BUTTON": "📢 عضویت در کانال", - "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "❌ هنوز عضو کانال نشده‌اید!", + "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "لطفاً ابتدا در همه کانال‌های اجباری عضو شوید!", "CHANNEL_SUBSCRIBE_THANKS": "✅ ممنون از عضویت", "CHECK_STATUS_BUTTON": "📊 بررسی وضعیت", "CHECK_STATUS_NO_CHANGES": "وضعیت تغییر نکرده", @@ -1452,10 +1452,12 @@ "SUBSCRIPTION_NO_SERVERS": "❌ سروری در دسترس نیست", "SUBSCRIPTION_OVERVIEW_TEMPLATE": "📱 اشتراک شما\n\n📊 وضعیت: {status}\n📅 اعتبار: {expiry}\n📈 ترافیک: {traffic}\n📱 دستگاه‌ها: {devices}\n🌍 سرورها: {servers}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 موجودی: {balance}\n📱 اشتراک: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 اطلاعات اشتراک\n🎭 نوع: {subscription_type}\n📈 ترافیک: {traffic}\n🌍 سرورها: {servers}\n📱 دستگاه‌ها: {devices}\n🌍 کشورها: {countries}", + "SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE": "اشتراک شما متوقف شده است زیرا از کانال اجباری خارج شدید.\n\nبرای بازیابی دسترسی VPN در همه کانال‌ها عضو شوید.", "SUBSCRIPTION_PROMO_DISCOUNT_HINT": "⚡ تخفیف اضافی {percent}% فعال شد.\n\nبا سایر تخفیف‌ها جمع می‌شود!", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE": "⚡ تخفیف اضافی {percent}%: -{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER": "\n⏳ اعتبار تخفیف: {time_left} {bar}", "SUBSCRIPTION_PURCHASED": "🎉 اشتراک خریداری شد!", + "SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE": "اشتراک شما بازیابی شد!\n\nبا تشکر از عضویت در کانال‌ها. VPN دوباره فعال است.", "SUBSCRIPTION_SETTINGS_BUTTON": "⚙️ تنظیمات", "SUBSCRIPTION_SETTINGS_OVERVIEW": "⚙️ تنظیمات اشتراک", "SUBSCRIPTION_SETTINGS_PAID_ONLY": "⚠️ تنظیمات فقط برای اشتراک پولی", diff --git a/app/localization/locales/ru.json b/app/localization/locales/ru.json index 8c875193..5ecc0a87 100644 --- a/app/localization/locales/ru.json +++ b/app/localization/locales/ru.json @@ -951,9 +951,9 @@ "CHANGE_DEVICES_TITLE": "📱 Изменение количества устройств", "CHANGE_TARIFF_BUTTON": "📦 Тариф", "CHANNEL_CHECK_BUTTON": "✅ Я подписался", - "CHANNEL_REQUIRED_TEXT": "🔒 Для использования бота подпишитесь на новостной канал, а затем нажмите кнопку ниже.", + "CHANNEL_REQUIRED_TEXT": "Пожалуйста, подпишитесь на обязательные каналы и нажмите кнопку ниже.", "CHANNEL_SUBSCRIBE_BUTTON": "🔗 Подписаться", - "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "❌ Вы не подписались на канал!", + "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "Пожалуйста, подпишитесь на все обязательные каналы!", "CHANNEL_SUBSCRIBE_THANKS": "✅ Спасибо за подписку", "CHECK_STATUS_BUTTON": "📊 Проверить статус", "CHECK_STATUS_NO_CHANGES": "Статус не изменился", @@ -1452,10 +1452,12 @@ "SUBSCRIPTION_NO_SERVERS": "Нет серверов", "SUBSCRIPTION_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Баланс: {balance}\n📱 Подписка: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Информация о подписке\n🎭 Тип: {subscription_type}\n📅 Действует до: {end_date}\n⏰ Осталось: {time_left}\n📈 Трафик: {traffic}\n🌍 Серверы: {servers}\n📱 Устройства: {devices_used} / {device_limit}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Баланс: {balance}\n📱 Подписка: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Информация о подписке\n🎭 Тип: {subscription_type}\n📈 Трафик: {traffic}\n🌍 Серверы: {servers}\n📱 Устройства: {devices_used} / {device_limit}", + "SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE": "Ваша подписка приостановлена, так как вы отписались от обязательного канала.\n\nПодпишитесь на все каналы для восстановления доступа к VPN.", "SUBSCRIPTION_PROMO_DISCOUNT_HINT": "⚡ Активирована доп. скидка {percent}%. \n\nСуммируется с другими скидками!", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE": "⚡ Доп. скидка {percent}%: -{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER": "⏳ Скидка действует ещё: {time_left}\n{bar}", "SUBSCRIPTION_PURCHASED": "🎉 Подписка успешно приобретена!", + "SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE": "Ваша подписка восстановлена!\n\nСпасибо за подписку на каналы. VPN снова активен.", "SUBSCRIPTION_SETTINGS_BUTTON": "⚙️ Настройки", "SUBSCRIPTION_SETTINGS_OVERVIEW": "⚙️ Настройки подписки\n\n📊 Текущие параметры:\n🌐 Стран: {countries_count}\n📈 Трафик: {traffic_used} / {traffic_limit}\n📱 Устройства: {devices_used} / {devices_limit}\n\nВыберите что хотите изменить:", "SUBSCRIPTION_SETTINGS_PAID_ONLY": "⚠️ Настройки доступны только для платных подписок", diff --git a/app/localization/locales/ua.json b/app/localization/locales/ua.json index f533e09d..e2bcdf48 100644 --- a/app/localization/locales/ua.json +++ b/app/localization/locales/ua.json @@ -871,9 +871,9 @@ "CHANGE_DEVICES_SUCCESS_INCREASE": "\n  ✅ Кількість пристроїв збільшено!\n\n  📱 Було: {old_count} → Стало: {new_count}\n  💰 Списано: {amount}\n  ", "CHANGE_DEVICES_TITLE": "📱 Зміна кількості пристроїв", "CHANNEL_CHECK_BUTTON": "✅ Я підписався", - "CHANNEL_REQUIRED_TEXT": "🔒 Для використання бота підпишіться на канал новин, а потім натисніть кнопку нижче.", + "CHANNEL_REQUIRED_TEXT": "Будь ласка, підпишіться на обов'язкові канали та натисніть кнопку нижче.", "CHANNEL_SUBSCRIBE_BUTTON": "🔗 Підписатися", - "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "❌ Ви не підписалися на канал!", + "CHANNEL_SUBSCRIBE_REQUIRED_ALERT": "Будь ласка, підпишіться на всі обов'язкові канали!", "CHANNEL_SUBSCRIBE_THANKS": "✅ Дякуємо за підписку", "CHECK_STATUS_BUTTON": "📊 Перевірити статус", "CHECK_STATUS_NO_CHANGES": "Статус не змінився", @@ -1362,10 +1362,12 @@ "SUBSCRIPTION_NO_SERVERS": "Немає серверів", "SUBSCRIPTION_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Баланс: {balance}\n📱 Підписка: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Інформація про підписку\n🎭 Тип: {subscription_type}\n📅 Діє до: {end_date}\n⏰ Залишилося: {time_left}\n📈 Трафік: {traffic}\n🌍 Сервери: {servers}\n📱 Пристрої: {devices_used} / {device_limit}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE": "👤 {full_name}\n💰 Баланс: {balance}\n📱 Підписка: {status_emoji} {status_display}{warning}{tariff_info_block}\n\n📱 Інформація про підписку\n🎭 Тип: {subscription_type}\n📈 Трафік: {traffic}\n🌍 Сервери: {servers}\n📱 Пристрої: {devices_used} / {device_limit}", + "SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE": "Вашу підписку призупинено, оскільки ви відписались від обов'язкового каналу.\n\nПідпишіться на всі канали для відновлення доступу до VPN.", "SUBSCRIPTION_PROMO_DISCOUNT_HINT": "⚡ Активовано дод. знижку {percent}%. \n\nСумується з іншими знижками!", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE": "⚡ Дод. знижка {percent}%: -{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER": "⏳ Знижка діє ще: {time_left}\n{bar}", "SUBSCRIPTION_PURCHASED": "🎉 Підписку успішно придбано!", + "SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE": "Вашу підписку відновлено!\n\nДякуємо за підписку на канали. VPN знову активний.", "SUBSCRIPTION_SETTINGS_BUTTON": "⚙️ Налаштування підписки", "SUBSCRIPTION_SETTINGS_OVERVIEW": "⚙️ Налаштування підписки\n\n📊 Поточні параметри:\n🌐 Країн: {countries_count}\n📈 Трафік: {traffic_used} / {traffic_limit}\n📱 Пристрої: {devices_used} / {devices_limit}\n\nОберіть що хочете змінити:", "SUBSCRIPTION_SETTINGS_PAID_ONLY": "⚠️ Налаштування доступні лише для платних підписок", diff --git a/app/localization/locales/zh.json b/app/localization/locales/zh.json index e1a9289d..3b82ad22 100644 --- a/app/localization/locales/zh.json +++ b/app/localization/locales/zh.json @@ -869,9 +869,9 @@ "CHANGE_DEVICES_SUCCESS_INCREASE":"\n  ✅设备数量已增加！\n\n  📱之前：{old_count}→现在：{new_count}\n  💰已扣除：{amount}\n  ", "CHANGE_DEVICES_TITLE":"📱更改设备数量", "CHANNEL_CHECK_BUTTON":"✅我已订阅", -"CHANNEL_REQUIRED_TEXT":"🔒要使用机器人，请订阅新闻频道，然后点击下方按钮。", +"CHANNEL_REQUIRED_TEXT":"请订阅所有必需频道，然后点击下方按钮。", "CHANNEL_SUBSCRIBE_BUTTON":"🔗订阅", -"CHANNEL_SUBSCRIBE_REQUIRED_ALERT":"❌您没有订阅该频道！", +"CHANNEL_SUBSCRIBE_REQUIRED_ALERT":"请先订阅所有必需频道！", "CHANNEL_SUBSCRIBE_THANKS":"✅感谢您的订阅", "CHECK_STATUS_BUTTON":"📊检查状态", "CHECK_STATUS_NO_CHANGES":"状态未更改", @@ -1360,10 +1360,12 @@ "SUBSCRIPTION_NO_SERVERS":"没有服务器", "SUBSCRIPTION_OVERVIEW_TEMPLATE":"👤{full_name}\n💰余额：{balance}\n📱订阅：{status_emoji}{status_display}{warning}{tariff_info_block}\n\n📱订阅信息\n🎭类型：{subscription_type}\n📅有效期至：{end_date}\n⏰剩余时间：{time_left}\n📈流量：{traffic}\n🌍服务器：{servers}\n📱设备：{devices_used}/{device_limit}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE":"👤{full_name}\n💰余额：{balance}\n📱订阅：{status_emoji}{status_display}{warning}{tariff_info_block}\n\n📱订阅信息\n🎭类型：{subscription_type}\n📈流量：{traffic}\n🌍服务器：{servers}\n📱设备：{devices_used}/{device_limit}", +"SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE":"由于您退出了必需频道，您的订阅已暂停。\n\n请订阅所有频道以恢复VPN访问。", "SUBSCRIPTION_PROMO_DISCOUNT_HINT":"⚡已激活额外{percent}%折扣。\n\n可与其他折扣叠加！", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE":"⚡额外{percent}%折扣：-{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER":"⏳折扣剩余时间：{time_left}\n{bar}", "SUBSCRIPTION_PURCHASED":"🎉订阅已成功购买！", +"SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE":"您的订阅已恢复！\n\n感谢您订阅频道。VPN已重新激活。", "SUBSCRIPTION_SETTINGS_BUTTON":"⚙️订阅设置", "SUBSCRIPTION_SETTINGS_OVERVIEW":"⚙️订阅设置\n\n📊当前参数：\n🌐国家：{countries_count}\n📈流量：{traffic_used}/{traffic_limit}\n📱设备：{devices_used}/{devices_limit}\n\n请选择要更改的内容：", "SUBSCRIPTION_SETTINGS_PAID_ONLY":"⚠️设置仅适用于付费订阅", @@ -1693,10 +1695,12 @@ "SUBSCRIPTION_NO_SERVERS":"没有服务器", "SUBSCRIPTION_OVERVIEW_TEMPLATE":"👤{full_name}\n💰余额：{balance}\n📱订阅：{status_emoji}{status_display}{warning}{tariff_info_block}\n\n📱订阅信息\n🎭类型：{subscription_type}\n📅有效期至：{end_date}\n⏰剩余时间：{time_left}\n📈流量：{traffic}\n🌍服务器：{servers}\n📱设备：{devices_used}/{device_limit}", "SUBSCRIPTION_DAILY_OVERVIEW_TEMPLATE":"👤{full_name}\n💰余额：{balance}\n📱订阅：{status_emoji}{status_display}{warning}{tariff_info_block}\n\n📱订阅信息\n🎭类型：{subscription_type}\n📈流量：{traffic}\n🌍服务器：{servers}\n📱设备：{devices_used}/{device_limit}", +"SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE":"由于您退出了必需频道，您的订阅已暂停。\n\n请订阅所有频道以恢复VPN访问。", "SUBSCRIPTION_PROMO_DISCOUNT_HINT":"⚡已激活额外{percent}%折扣。\n\n可与其他折扣叠加！", "SUBSCRIPTION_PROMO_DISCOUNT_NOTE":"⚡额外{percent}%折扣：-{amount}", "SUBSCRIPTION_PROMO_DISCOUNT_TIMER":"⏳折扣剩余时间：{time_left}\n{bar}", "SUBSCRIPTION_PURCHASED":"🎉订阅已成功购买！", +"SUBSCRIPTION_REACTIVATED_CHANNEL_SUBSCRIBE":"您的订阅已恢复！\n\n感谢您订阅频道。VPN已重新激活。", "SUBSCRIPTION_SETTINGS_BUTTON":"⚙️订阅设置", "SUBSCRIPTION_SETTINGS_OVERVIEW":"⚙️订阅设置\n\n📊当前参数：\n🌐国家：{countries_count}\n📈流量：{traffic_used}/{traffic_limit}\n📱设备：{devices_used}/{devices_limit}\n\n请选择要更改的内容：", "SUBSCRIPTION_SETTINGS_PAID_ONLY":"⚠️设置仅适用于付费订阅", diff --git a/app/middlewares/channel_checker.py b/app/middlewares/channel_checker.py index 88704a32..fb2a55e6 100644 --- a/app/middlewares/channel_checker.py +++ b/app/middlewares/channel_checker.py @@ -2,11 +2,9 @@ from collections.abc import Awaitable, Callable from datetime import UTC, datetime from typing import Any -import redis.asyncio as aioredis import structlog from aiogram import BaseMiddleware, Bot, types -from aiogram.enums import ChatMemberStatus -from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramNetworkError +from aiogram.exceptions import TelegramBadRequest from aiogram.fsm.context import FSMContext from aiogram.types import CallbackQuery, Message, TelegramObject, Update @@ -20,69 +18,62 @@ from app.keyboards.inline import get_channel_sub_keyboard from app.localization.loader import DEFAULT_LANGUAGE from app.localization.texts import get_texts from app.services.admin_notification_service import AdminNotificationService +from app.services.channel_subscription_service import channel_subscription_service from app.services.subscription_service import SubscriptionService +from app.utils.cache import cache from app.utils.check_reg_process import is_registration_process logger = structlog.get_logger(__name__) -# Ключ для хранения pending_start_payload в Redis (резервный механизм) +# Redis key prefix and TTL for pending /start payload backup REDIS_PAYLOAD_KEY_PREFIX = 'pending_start_payload:' -REDIS_PAYLOAD_TTL = 3600 # 1 час +REDIS_PAYLOAD_TTL = 3600 # 1 hour async def save_pending_payload_to_redis(telegram_id: int, payload: str) -> bool: - """Сохраняет pending_start_payload в Redis напрямую (резервный механизм).""" + """Save pending_start_payload to Redis via the shared cache singleton.""" try: - redis_client = aioredis.from_url(settings.REDIS_URL) key = f'{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}' - await redis_client.set(key, payload, ex=REDIS_PAYLOAD_TTL) - await redis_client.aclose() - logger.info( - "💾 [Redis fallback] Сохранен payload '' для пользователя", payload=payload, telegram_id=telegram_id - ) - return True + result = await cache.set(key, payload, expire=REDIS_PAYLOAD_TTL) + if result: + logger.info('Saved pending payload to Redis', payload=payload, telegram_id=telegram_id) + return result except Exception as e: - logger.error('❌ [Redis fallback] Ошибка сохранения payload для', telegram_id=telegram_id, e=e) + logger.error('Failed to save payload to Redis', telegram_id=telegram_id, error=e) return False async def get_pending_payload_from_redis(telegram_id: int) -> str | None: - """Получает pending_start_payload из Redis (резервный механизм).""" + """Get pending_start_payload from Redis via the shared cache singleton.""" try: - redis_client = aioredis.from_url(settings.REDIS_URL) key = f'{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}' - payload = await redis_client.get(key) - await redis_client.aclose() - if payload: - return payload.decode('utf-8') if isinstance(payload, bytes) else payload - return None + return await cache.get(key) except Exception as e: - logger.debug('❌ [Redis fallback] Ошибка получения payload для', telegram_id=telegram_id, e=e) + logger.debug('Failed to get payload from Redis', telegram_id=telegram_id, error=e) return None async def delete_pending_payload_from_redis(telegram_id: int) -> None: - """Удаляет pending_start_payload из Redis.""" + """Delete pending_start_payload from Redis via the shared cache singleton.""" try: - redis_client = aioredis.from_url(settings.REDIS_URL) key = f'{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}' - await redis_client.delete(key) - await redis_client.aclose() + await cache.delete(key) except Exception: pass class ChannelCheckerMiddleware(BaseMiddleware): - """ - Middleware для проверки подписки на канал. - ОПТИМИЗИРОВАНО: создаёт максимум одну сессию БД на запрос. + """Middleware for checking required channel subscriptions. + + OPTIMIZED FOR 100k+ USERS: + - Does NOT call Telegram API directly in the hot path + - Reads from Redis cache (TTL 600s) -> PostgreSQL -> rate-limited API fallback + - Updated in real-time via ChatMemberUpdated events """ def __init__(self): - self.BAD_MEMBER_STATUS = (ChatMemberStatus.LEFT, ChatMemberStatus.KICKED, ChatMemberStatus.RESTRICTED) - self.GOOD_MEMBER_STATUS = (ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.CREATOR) - logger.info('🔧 ChannelCheckerMiddleware инициализирован') + logger.info('ChannelCheckerMiddleware initialized (multi-channel mode)') async def __call__( self, @@ -90,6 +81,11 @@ class ChannelCheckerMiddleware(BaseMiddleware): event: TelegramObject, data: dict[str, Any], ) -> Any: + # Runtime check (supports toggling without restart) + if not settings.CHANNEL_IS_REQUIRED_SUB: + return await handler(event, data) + + # Fast-path bypasses telegram_id = None if isinstance(event, (Message, CallbackQuery)): telegram_id = event.from_user.id @@ -100,7 +96,6 @@ class ChannelCheckerMiddleware(BaseMiddleware): telegram_id = event.callback_query.from_user.id if telegram_id is None: - logger.debug('❌ telegram_id не найден, пропускаем') return await handler(event, data) # Skip channel check for lightweight UI callbacks (close/delete notifications) @@ -112,110 +107,123 @@ class ChannelCheckerMiddleware(BaseMiddleware): ): return await handler(event, data) - # Админам разрешаем пропускать проверку подписки if settings.is_admin(telegram_id): - logger.debug( - '✅ Пользователь является администратором — пропускаем проверку подписки', telegram_id=telegram_id - ) return await handler(event, data) state: FSMContext = data.get('state') - current_state = None - - if state: - current_state = await state.get_state() - - is_reg_process = is_registration_process(event, current_state) - - if is_reg_process: - logger.debug('✅ Событие разрешено (процесс регистрации), пропускаем проверку') + current_state = await state.get_state() if state else None + if is_registration_process(event, current_state): return await handler(event, data) + # Ensure service has bot reference for API fallback bot: Bot = data['bot'] + if not channel_subscription_service.bot: + channel_subscription_service.bot = bot - channel_id = settings.CHANNEL_SUB_ID + # Multi-channel check (Redis -> DB -> API) + unsubscribed = await channel_subscription_service.get_unsubscribed_channels(telegram_id) - if not channel_id: - logger.warning('⚠️ CHANNEL_SUB_ID не установлен, пропускаем проверку') + if not unsubscribed: + # All subscribed -- reactivate if needed + if settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE or settings.CHANNEL_REQUIRED_FOR_ALL: + await self._reactivate_subscription_on_subscribe(telegram_id, bot) return await handler(event, data) - is_required = settings.CHANNEL_IS_REQUIRED_SUB + # User is NOT subscribed to all channels + if settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE or settings.CHANNEL_REQUIRED_FOR_ALL: + await self._deactivate_subscription_on_unsubscribe(telegram_id, bot, unsubscribed) - if not is_required: - logger.debug('⚠️ Обязательная подписка отключена, пропускаем проверку') - return await handler(event, data) + await self._capture_start_payload(state, event, bot) - channel_link = self._normalize_channel_link(settings.CHANNEL_LINK, channel_id) + if isinstance(event, CallbackQuery) and event.data == 'sub_channel_check': + # Rate limit: max 1 check per 5 seconds per user + rate_key = f'sub_check_rate:{telegram_id}' + if await cache.exists(rate_key): + await event.answer() + return None + await cache.set(rate_key, 1, expire=5) - if not channel_link: - logger.warning('⚠️ CHANNEL_LINK не задан или невалиден, кнопка подписки будет скрыта') + # Re-check via API for immediate feedback (invalidate cache first) + await channel_subscription_service.invalidate_user_cache(telegram_id) - try: - member = await bot.get_chat_member(chat_id=channel_id, user_id=telegram_id) - - if member.status in self.GOOD_MEMBER_STATUS: - # Реактивируем подписку если была отключена из-за отписки от канала - if telegram_id and (settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE or settings.CHANNEL_REQUIRED_FOR_ALL): + unsubscribed_fresh = await channel_subscription_service.get_unsubscribed_channels(telegram_id) + if not unsubscribed_fresh: + # Now subscribed to all channels + if settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE or settings.CHANNEL_REQUIRED_FOR_ALL: await self._reactivate_subscription_on_subscribe(telegram_id, bot) return await handler(event, data) - if member.status in self.BAD_MEMBER_STATUS: - logger.info( - '❌ Пользователь не подписан на канал (статус: )', telegram_id=telegram_id, status=member.status - ) - if telegram_id and (settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE or settings.CHANNEL_REQUIRED_FOR_ALL): - await self._deactivate_subscription_on_unsubscribe(telegram_id, bot, channel_link) + user_lang = ( + event.from_user.language_code.split('-')[0] + if event.from_user and event.from_user.language_code + else DEFAULT_LANGUAGE + ) + texts = get_texts(user_lang) + await event.answer( + texts.t( + 'CHANNEL_CHECK_NOT_SUBSCRIBED', + 'You are not subscribed to all required channels. Please subscribe and try again.', + ), + show_alert=True, + ) + return None - await self._capture_start_payload(state, event, bot) + return await self._deny_message(event, bot, unsubscribed) - if isinstance(event, CallbackQuery) and event.data == 'sub_channel_check': - await event.answer( - '❌ Вы еще не подписались на канал! Подпишитесь и попробуйте снова.', show_alert=True - ) - return None - - return await self._deny_message(event, bot, channel_link, channel_id) - logger.warning('⚠️ Неожиданный статус пользователя', telegram_id=telegram_id, status=member.status) - await self._capture_start_payload(state, event, bot) - return await self._deny_message(event, bot, channel_link, channel_id) - - except TelegramForbiddenError as e: - logger.error('❌ Бот заблокирован в канале', channel_id=channel_id, error=e) - await self._capture_start_payload(state, event, bot) - return await self._deny_message(event, bot, channel_link, channel_id) - except TelegramBadRequest as e: - if 'chat not found' in str(e).lower(): - logger.error('❌ Канал не найден', channel_id=channel_id, error=e) - elif 'user not found' in str(e).lower(): - logger.error('❌ Пользователь не найден', telegram_id=telegram_id, error=e) - else: - logger.error('❌ Ошибка запроса к каналу', channel_id=channel_id, error=e) - await self._capture_start_payload(state, event, bot) - return await self._deny_message(event, bot, channel_link, channel_id) - except TelegramNetworkError as e: - logger.warning('⚠️ Таймаут при проверке подписки на канал', error=e) - return await handler(event, data) - except Exception as e: - logger.error('❌ Неожиданная ошибка при проверке подписки', error=e) - return await handler(event, data) + # -- _deny_message (multi-channel) ----------------------------------------- @staticmethod - def _normalize_channel_link(channel_link: str | None, channel_id: str | None) -> str | None: - link = (channel_link or '').strip() + async def _deny_message( + event: TelegramObject, + bot: Bot, + unsubscribed_channels: list[dict], + ): + user = None + if isinstance(event, (Message, CallbackQuery)): + user = getattr(event, 'from_user', None) + elif isinstance(event, Update): + if event.message and event.message.from_user: + user = event.message.from_user + elif event.callback_query and event.callback_query.from_user: + user = event.callback_query.from_user - if link.startswith('@'): # raw username - return f'https://t.me/{link.lstrip("@")}' + language = DEFAULT_LANGUAGE + if user and user.language_code: + language = user.language_code.split('-')[0] - if link and not link.lower().startswith(('http://', 'https://', 'tg://')): - return f'https://{link}' + # Normalize channel links (convert @username -> https://t.me/username) + normalized = [] + for ch in unsubscribed_channels: + ch_copy = dict(ch) + link = ch_copy.get('channel_link') + if link: + ch_copy['channel_link'] = _normalize_channel_link(link) + normalized.append(ch_copy) - if link: - return link + texts = get_texts(language) + channel_sub_kb = get_channel_sub_keyboard(normalized, language=language) + text = texts.t( + 'CHANNEL_REQUIRED_TEXT', + '🔒 Для использования бота подпишитесь на новостной канал, ' + 'чтобы получать уведомления о новых возможностях и обновлениях бота. Спасибо!', + ) - if channel_id and str(channel_id).startswith('@'): - return f'https://t.me/{str(channel_id).lstrip("@")}' + try: + if isinstance(event, Message): + return await event.answer(text, reply_markup=channel_sub_kb) + if isinstance(event, CallbackQuery): + try: + return await event.message.edit_text(text, reply_markup=channel_sub_kb) + except TelegramBadRequest as e: + if 'message is not modified' in str(e).lower(): + return await event.answer(text, show_alert=True) + raise + elif isinstance(event, Update) and event.message: + return await bot.send_message(event.message.chat.id, text, reply_markup=channel_sub_kb) + except Exception as e: + logger.error('Error sending subscription prompt', error=e) - return None + # -- _capture_start_payload ------------------------------------------------ async def _capture_start_payload( self, @@ -223,6 +231,11 @@ class ChannelCheckerMiddleware(BaseMiddleware): event: TelegramObject, bot: Bot | None = None, ) -> None: + """Save /start payload to FSM + Redis so it can be restored after subscription. + + This preserves referral codes, deep links, and other start parameters + when a user is blocked by the channel subscription requirement. + """ telegram_id = None if isinstance(event, (Message, CallbackQuery)): telegram_id = event.from_user.id if event.from_user else None @@ -246,23 +259,21 @@ class ChannelCheckerMiddleware(BaseMiddleware): payload = parts[1] - # Сохраняем в FSM state + # Save to FSM state if state: state_data = await state.get_data() or {} if state_data.get('pending_start_payload') != payload: state_data['pending_start_payload'] = payload await state.set_data(state_data) - logger.info( - "💾 Сохранен start payload '' для пользователя (FSM)", payload=payload, telegram_id=telegram_id - ) + logger.info('Saved start payload for user (FSM)', payload=payload, telegram_id=telegram_id) else: - logger.warning('⚠️ _capture_start_payload: state=None для пользователя', telegram_id=telegram_id) + logger.warning('_capture_start_payload: state=None for user', telegram_id=telegram_id) - # Также сохраняем в Redis как резерв (на случай потери FSM state) + # Also save to Redis as backup (in case FSM state is lost) if telegram_id: await save_pending_payload_to_redis(telegram_id, payload) - if bot and message.from_user: + if bot and message.from_user and state: await self._try_send_campaign_visit_notification( bot, message.from_user, @@ -280,9 +291,7 @@ class ChannelCheckerMiddleware(BaseMiddleware): try: state_data = await state.get_data() or {} except Exception as error: - logger.error( - '❌ Не удалось получить данные состояния для уведомления по кампании', payload=payload, error=error - ) + logger.error('Failed to get state data for campaign notification', payload=payload, error=error) return if state_data.get('campaign_notification_sent'): @@ -311,13 +320,18 @@ class ChannelCheckerMiddleware(BaseMiddleware): await state.update_data(campaign_notification_sent=True) await db.commit() except Exception as error: - logger.error('❌ Ошибка отправки уведомления о переходе по кампании', payload=payload, error=error) + logger.error('Error sending campaign visit notification', payload=payload, error=error) await db.rollback() + # -- _deactivate (multi-channel) ------------------------------------------- + async def _deactivate_subscription_on_unsubscribe( - self, telegram_id: int, bot: Bot, channel_link: str | None + self, + telegram_id: int, + bot: Bot, + unsubscribed_channels: list[dict], ) -> None: - """Деактивация подписки при отписке от канала.""" + """Deactivate subscription when user unsubscribes from required channels.""" if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE and not settings.CHANNEL_REQUIRED_FOR_ALL: return @@ -341,15 +355,15 @@ class ChannelCheckerMiddleware(BaseMiddleware): if is_active_paid_subscription(subscription): logger.info( - '⏭️ Пропуск отключения: у пользователя активная оплаченная подписка', + 'Skipping deactivation: user has active paid subscription', telegram_id=telegram_id, ) return await deactivate_subscription(db, subscription) - sub_type = 'Триальная' if subscription.is_trial else 'Платная' + sub_type = 'trial' if subscription.is_trial else 'paid' logger.info( - '🚫 подписка пользователя отключена после отписки от канала', + 'Subscription deactivated after channel unsubscribe', sub_type=sub_type, telegram_id=telegram_id, ) @@ -360,38 +374,49 @@ class ChannelCheckerMiddleware(BaseMiddleware): await service.disable_remnawave_user(user.remnawave_uuid) except Exception as api_error: logger.error( - '❌ Не удалось отключить пользователя RemnaWave', + 'Failed to disable RemnaWave user', remnawave_uuid=user.remnawave_uuid, api_error=api_error, ) - # Уведомляем пользователя о деактивации + # Notify user about deactivation try: + # Normalize links for keyboard + normalized = [] + for ch in unsubscribed_channels: + ch_copy = dict(ch) + link = ch_copy.get('channel_link') + if link: + ch_copy['channel_link'] = _normalize_channel_link(link) + normalized.append(ch_copy) + texts = get_texts(user.language if user.language else DEFAULT_LANGUAGE) notification_text = texts.t( 'SUBSCRIPTION_DEACTIVATED_CHANNEL_UNSUBSCRIBE', '🚫 Ваша подписка приостановлена, так как вы отписались от канала.\n\n' 'Подпишитесь на канал снова, чтобы восстановить доступ к VPN.', ) - channel_kb = get_channel_sub_keyboard(channel_link, language=user.language) + channel_kb = get_channel_sub_keyboard(normalized, language=user.language) await bot.send_message(telegram_id, notification_text, reply_markup=channel_kb) except Exception as notify_error: logger.error( - '❌ Не удалось отправить уведомление о деактивации пользователю', + 'Failed to send deactivation notification to user', telegram_id=telegram_id, notify_error=notify_error, ) await db.commit() except Exception as db_error: logger.error( - '❌ Ошибка деактивации подписки пользователя после отписки', + 'Error deactivating subscription after channel unsubscribe', telegram_id=telegram_id, db_error=db_error, ) await db.rollback() + # -- _reactivate ----------------------------------------------------------- + async def _reactivate_subscription_on_subscribe(self, telegram_id: int, bot: Bot) -> None: - """Реактивация подписки после повторной подписки на канал.""" + """Reactivate subscription after user subscribes to all required channels.""" if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE and not settings.CHANNEL_REQUIRED_FOR_ALL: return @@ -401,43 +426,42 @@ class ChannelCheckerMiddleware(BaseMiddleware): if not user or not user.subscription: return - # НЕ реактивируем подписку заблокированным пользователям + # Do NOT reactivate for blocked users if user.status == UserStatus.BLOCKED.value: - logger.info('🚫 Пропуск реактивации для заблокированного пользователя', telegram_id=telegram_id) + logger.info('Skipping reactivation for blocked user', telegram_id=telegram_id) return subscription = user.subscription - # Реактивируем только DISABLED подписки + # Only reactivate DISABLED subscriptions if subscription.status != SubscriptionStatus.DISABLED.value: return - # Проверяем что подписка ещё не истекла + # Check subscription has not expired if subscription.end_date and subscription.end_date <= datetime.now(UTC): return - # Реактивируем в БД await reactivate_subscription(db, subscription) - sub_type = 'Триальная' if subscription.is_trial else 'Платная' + sub_type = 'trial' if subscription.is_trial else 'paid' logger.info( - '✅ подписка пользователя реактивирована после подписки на канал', + 'Subscription reactivated after channel subscribe', sub_type=sub_type, telegram_id=telegram_id, ) - # Включаем в RemnaWave + # Enable in RemnaWave if user.remnawave_uuid: service = SubscriptionService() try: await service.enable_remnawave_user(user.remnawave_uuid) except Exception as api_error: logger.error( - '❌ Не удалось включить пользователя RemnaWave', + 'Failed to enable RemnaWave user', remnawave_uuid=user.remnawave_uuid, api_error=api_error, ) - # Уведомляем пользователя о реактивации + # Notify user about reactivation try: texts = get_texts(user.language if user.language else DEFAULT_LANGUAGE) notification_text = texts.t( @@ -447,65 +471,21 @@ class ChannelCheckerMiddleware(BaseMiddleware): await bot.send_message(telegram_id, notification_text) except Exception as notify_error: logger.warning( - 'Не удалось отправить уведомление о реактивации пользователю', + 'Failed to send reactivation notification to user', telegram_id=telegram_id, notify_error=notify_error, ) await db.commit() except Exception as db_error: - logger.error('❌ Ошибка реактивации подписки пользователя', telegram_id=telegram_id, db_error=db_error) + logger.error('Error reactivating subscription', telegram_id=telegram_id, db_error=db_error) await db.rollback() - @staticmethod - async def _deny_message( - event: TelegramObject, - bot: Bot, - channel_link: str | None, - channel_id: str | None, - ): - logger.debug('🚫 Отправляем сообщение о необходимости подписки') - user = None - if isinstance(event, (Message, CallbackQuery)): - user = getattr(event, 'from_user', None) - elif isinstance(event, Update): - if event.message and event.message.from_user: - user = event.message.from_user - elif event.callback_query and event.callback_query.from_user: - user = event.callback_query.from_user - - language = DEFAULT_LANGUAGE - if user and user.language_code: - language = user.language_code.split('-')[0] - - texts = get_texts(language) - channel_sub_kb = get_channel_sub_keyboard(channel_link, language=language) - text = texts.t( - 'CHANNEL_REQUIRED_TEXT', - '🔒 Для использования бота подпишитесь на новостной канал, чтобы получать уведомления о новых возможностях и обновлениях бота. Спасибо!', - ) - - if not channel_link and channel_id: - channel_hint = None - - if str(channel_id).startswith('@'): # username-based channel id - channel_hint = f'@{str(channel_id).lstrip("@")}' - - if channel_hint: - text = f'{text}\n\n{channel_hint}' - - try: - if isinstance(event, Message): - return await event.answer(text, reply_markup=channel_sub_kb) - if isinstance(event, CallbackQuery): - try: - return await event.message.edit_text(text, reply_markup=channel_sub_kb) - except TelegramBadRequest as e: - if 'message is not modified' in str(e).lower(): - logger.debug('ℹ️ Сообщение уже содержит текст проверки подписки, пропускаем редактирование') - return await event.answer(text, show_alert=True) - raise - elif isinstance(event, Update) and event.message: - return await bot.send_message(event.message.chat.id, text, reply_markup=channel_sub_kb) - except Exception as e: - logger.error('❌ Ошибка при отправке сообщения о подписке', error=e) +def _normalize_channel_link(link: str) -> str: + """Normalize channel link: convert @username to https://t.me/username.""" + if not link: + return link + link = link.strip() + if link.startswith('@'): + return f'https://t.me/{link[1:]}' + return link diff --git a/app/services/channel_subscription_service.py b/app/services/channel_subscription_service.py new file mode 100644 index 00000000..00059492 --- /dev/null +++ b/app/services/channel_subscription_service.py @@ -0,0 +1,272 @@ +"""Channel subscription verification service. + +Architecture for 100k+ users: +1. ChatMemberUpdated events -> update PostgreSQL (source of truth) + Redis in real-time +2. Middleware reads ONLY from Redis/PostgreSQL (never calls Telegram API directly) +3. Background reconciliation (~10 req/sec) corrects drift +""" + +import asyncio +from datetime import UTC, datetime + +import structlog +from aiogram import Bot +from aiogram.enums import ChatMemberStatus +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramNetworkError, TelegramRetryAfter + +from app.database.crud.required_channel import ( + get_active_channels, + get_user_channel_subs, + upsert_user_channel_sub, +) +from app.database.database import AsyncSessionLocal +from app.utils.cache import ChannelSubCache + + +logger = structlog.get_logger(__name__) + +# Rate limiting for Telegram API calls +_API_SEMAPHORE = asyncio.Semaphore(20) # max 20 concurrent getChatMember calls +_API_DELAY = 0.05 # 50ms between calls -> ~20/sec safe rate + +GOOD_STATUSES = (ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.CREATOR) + +# How long a DB record is considered fresh (no API call needed) +DB_FRESHNESS_SECONDS = 1800 # 30 min + + +class ChannelSubscriptionService: + """Centralized service for channel subscription verification.""" + + def __init__(self, bot: Bot | None = None): + self.bot = bot + + # -- Public API --------------------------------------------------------------- + + async def get_required_channels(self) -> list[dict]: + """Get the list of active required channels (cached).""" + cached = await ChannelSubCache.get_required_channels() + if cached is not None: + return cached + + async with AsyncSessionLocal() as db: + channels = await get_active_channels(db) + result = [ + { + 'id': ch.id, + 'channel_id': ch.channel_id, + 'channel_link': ch.channel_link, + 'title': ch.title, + 'sort_order': ch.sort_order, + } + for ch in channels + ] + await ChannelSubCache.set_required_channels(result) + return result + + async def get_required_channel_ids(self) -> set[str]: + """Get the set of active required channel_ids (for event filtering).""" + channels = await self.get_required_channels() + return {ch['channel_id'] for ch in channels} + + async def check_user_subscriptions(self, telegram_id: int) -> dict[str, bool]: + """Check user subscriptions to all required channels. + + Returns {channel_id: is_member}. + Does NOT call Telegram API unless cache miss + stale DB. + Uses a SINGLE DB session for all channels (no N+1). + """ + channels = await self.get_required_channels() + return await self._check_user_subscriptions_for_channels(telegram_id, channels) + + async def _check_user_subscriptions_for_channels( + self, + telegram_id: int, + channels: list[dict], + ) -> dict[str, bool]: + """Internal: check subscriptions for a given list of channels. + + Avoids double-fetching required_channels when called from + get_unsubscribed_channels or get_channels_with_status. + """ + if not channels: + return {} + + result: dict[str, bool] = {} + channels_needing_db: list[dict] = [] + + # Layer 1: Redis cache (single MGET round-trip) + all_channel_ids = [ch['channel_id'] for ch in channels] + cached_statuses = await ChannelSubCache.get_sub_statuses(telegram_id, all_channel_ids) + for ch in channels: + channel_id = ch['channel_id'] + cached = cached_statuses.get(channel_id) + if cached is not None: + result[channel_id] = cached + else: + channels_needing_db.append(ch) + + # Layer 2: PostgreSQL (single session for all channels) + channels_needing_api: list[dict] = [] + if channels_needing_db: + async with AsyncSessionLocal() as db: + subs = await get_user_channel_subs(db, telegram_id) + sub_map = {s.channel_id: s for s in subs} + + for ch in channels_needing_db: + channel_id = ch['channel_id'] + sub = sub_map.get(channel_id) + + if sub and sub.checked_at: + age = (datetime.now(UTC) - sub.checked_at).total_seconds() + if age < DB_FRESHNESS_SECONDS: + result[channel_id] = sub.is_member + await ChannelSubCache.set_sub_status(telegram_id, channel_id, sub.is_member) + continue + + channels_needing_api.append(ch) + + # Layer 3: Rate-limited API calls for channels without fresh data + if channels_needing_api and self.bot: + async with AsyncSessionLocal() as db: + for ch in channels_needing_api: + is_member = await self._rate_limited_check(telegram_id, ch['channel_id']) + result[ch['channel_id']] = is_member + # Write DB first (source of truth), then cache + await upsert_user_channel_sub(db, telegram_id, ch['channel_id'], is_member) + await ChannelSubCache.set_sub_status(telegram_id, ch['channel_id'], is_member) + await db.commit() + elif channels_needing_api: + # No bot available (e.g., cabinet API context) -- fail-closed + logger.warning( + 'No bot instance for API check -- failing closed', + telegram_id=telegram_id, + channels=[ch['channel_id'] for ch in channels_needing_api], + ) + for ch in channels_needing_api: + result[ch['channel_id']] = False + + return result + + async def is_user_subscribed_to_all(self, telegram_id: int) -> bool: + """Quick check: is user subscribed to ALL required channels?""" + subs = await self.check_user_subscriptions(telegram_id) + if not subs: + return True # No required channels = subscribed + return all(subs.values()) + + async def get_unsubscribed_channels(self, telegram_id: int) -> list[dict]: + """Get the list of channels the user is NOT subscribed to.""" + channels = await self.get_required_channels() + subs = await self._check_user_subscriptions_for_channels(telegram_id, channels) + + unsubscribed = [] + for ch in channels: + if not subs.get(ch['channel_id'], False): + unsubscribed.append(ch) + return unsubscribed + + async def get_channels_with_status(self, telegram_id: int) -> list[dict]: + """Get all required channels with per-channel subscription status (for cabinet API).""" + channels = await self.get_required_channels() + subs = await self._check_user_subscriptions_for_channels(telegram_id, channels) + + result = [] + for ch in channels: + result.append( + { + 'channel_id': ch['channel_id'], + 'channel_link': ch.get('channel_link'), + 'title': ch.get('title'), + 'is_subscribed': subs.get(ch['channel_id'], False), + } + ) + return result + + async def get_first_channel_id(self) -> str | None: + """Get the first active channel ID (for announcements, contest posts, etc.). + + Channel IDs are always stored as strings in the DB. + Telegram API accepts string channel_id in chat_id parameters. + """ + channels = await self.get_required_channels() + if not channels: + return None + return channels[0]['channel_id'] + + # -- Event handlers (called from ChatMemberUpdated router) -------------------- + + async def on_user_joined(self, telegram_id: int, channel_id: str) -> None: + """Called when ChatMemberUpdated fires: user subscribed.""" + logger.info('Channel join event', telegram_id=telegram_id, channel_id=channel_id) + # Write DB first (source of truth), then cache + async with AsyncSessionLocal() as db: + await upsert_user_channel_sub(db, telegram_id, channel_id, True) + await db.commit() + await ChannelSubCache.set_sub_status(telegram_id, channel_id, True) + + async def on_user_left(self, telegram_id: int, channel_id: str) -> None: + """Called when ChatMemberUpdated fires: user unsubscribed.""" + logger.info('Channel leave event', telegram_id=telegram_id, channel_id=channel_id) + # Write DB first (source of truth), then cache + async with AsyncSessionLocal() as db: + await upsert_user_channel_sub(db, telegram_id, channel_id, False) + await db.commit() + await ChannelSubCache.set_sub_status(telegram_id, channel_id, False) + + # -- Channel list management -------------------------------------------------- + + async def invalidate_channels_cache(self) -> None: + """Invalidate the channels list cache (call after CRUD).""" + await ChannelSubCache.invalidate_channels() + + async def invalidate_user_cache(self, telegram_id: int) -> None: + """Invalidate all cached subscription statuses for a user.""" + channels = await self.get_required_channels() + channel_ids = [ch['channel_id'] for ch in channels] + await ChannelSubCache.invalidate_user_channels(telegram_id, channel_ids) + + # -- Rate-limited Telegram API ------------------------------------------------ + + async def _rate_limited_check(self, telegram_id: int, channel_id: str) -> bool: + """Check subscription via Telegram API with rate-limiting. + + SECURITY: Fail-closed -- any error returns False (not subscribed). + For a VPN access control system, false negatives (temporary denial) + are preferable to false positives (unauthorized access). + """ + async with _API_SEMAPHORE: + try: + member = await self.bot.get_chat_member(chat_id=channel_id, user_id=telegram_id) + await asyncio.sleep(_API_DELAY) + return member.status in GOOD_STATUSES + except TelegramRetryAfter as e: + logger.warning('Rate limited by Telegram', retry_after=e.retry_after, channel_id=channel_id) + await asyncio.sleep(e.retry_after) + try: + member = await self.bot.get_chat_member(chat_id=channel_id, user_id=telegram_id) + return member.status in GOOD_STATUSES + except Exception: + logger.error('Double failure after rate-limit retry', channel_id=channel_id) + return False # Fail-closed on double failure + except TelegramForbiddenError: + logger.critical( + 'Bot removed/blocked from channel -- all checks will fail-closed', + channel_id=channel_id, + ) + return False # Fail-closed -- bot cannot verify membership + except TelegramBadRequest as e: + if 'user not found' in str(e).lower(): + return False # User never interacted with bot in that context + logger.error('Bad request checking channel', channel_id=channel_id, error=str(e)) + return False # Fail-closed + except TelegramNetworkError: + logger.warning('Network error checking channel', channel_id=channel_id) + return False # Fail-closed + except Exception as e: + logger.error('Unexpected error checking channel', channel_id=channel_id, error=str(e)) + return False # Fail-closed + + +# Singleton instance (bot is set at startup) +channel_subscription_service = ChannelSubscriptionService() diff --git a/app/services/contest_rotation_service.py b/app/services/contest_rotation_service.py index 460da7d0..70843396 100644 --- a/app/services/contest_rotation_service.py +++ b/app/services/contest_rotation_service.py @@ -295,13 +295,11 @@ class ContestRotationService: async def _send_channel_announce(self, text: str) -> None: if not self.bot: return - channel_id_raw = settings.CHANNEL_SUB_ID - if not channel_id_raw: + from app.services.channel_subscription_service import channel_subscription_service + + channel_id = await channel_subscription_service.get_first_channel_id() + if not channel_id: return - try: - channel_id = int(channel_id_raw) - except Exception: - channel_id = channel_id_raw keyboard = InlineKeyboardMarkup( inline_keyboard=[[InlineKeyboardButton(text='🎲 Играть', callback_data='contests_menu')]] @@ -315,7 +313,7 @@ class ContestRotationService: reply_markup=keyboard, ) except Exception as exc: - logger.error('Не удалось отправить анонс в канал', channel_id_raw=channel_id_raw, exc=exc) + logger.error('Не удалось отправить анонс в канал', channel_id=channel_id, exc=exc) async def _broadcast_to_users(self, text: str) -> None: """Отправляет анонс всем пользователям с активной/триальной подпиской.""" diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py index 7e72b78f..8e7e1124 100644 --- a/app/services/monitoring_service.py +++ b/app/services/monitoring_service.py @@ -4,7 +4,6 @@ from pathlib import Path from typing import Any import structlog -from aiogram.enums import ChatMemberStatus from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramNetworkError from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession @@ -72,6 +71,9 @@ from app.utils.timezone import format_local_datetime # Кулдаун между повторными уведомлениями об автоплатеже с недостаточным балансом (6 часов) AUTOPAY_INSUFFICIENT_BALANCE_COOLDOWN_SECONDS: int = 21600 +# Размер батча для проверки подписок на каналы (keyset pagination) +_CHANNEL_CHECK_BATCH_SIZE: int = 100 + logger = structlog.get_logger(__name__) @@ -522,214 +524,254 @@ class MonitoringService: logger.error('Ошибка проверки истекающих тестовых подписок', error=e) async def _check_trial_channel_subscriptions(self, db: AsyncSession): - from app.database.crud.subscription import is_recently_updated_by_webhook + """Background reconciliation of channel subscriptions (rate-limited). + + Processes subscriptions in batches using keyset pagination to avoid + loading all trial subscriptions into memory at once. Each batch gets + a fresh DB session to avoid holding a connection pool slot for hours. + + When CHANNEL_REQUIRED_FOR_ALL is True, checks ALL active subscriptions + (not just trials). Otherwise only checks trial subscriptions. + """ + from app.database.crud.subscription import is_active_paid_subscription, is_recently_updated_by_webhook if not settings.CHANNEL_IS_REQUIRED_SUB: return - if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE: - logger.debug('ℹ️ Проверка отписок от канала отключена — деактивация триальных подписок не требуется') - return - - channel_id = settings.CHANNEL_SUB_ID - if not channel_id: + if not settings.CHANNEL_DISABLE_TRIAL_ON_UNSUBSCRIBE and not settings.CHANNEL_REQUIRED_FOR_ALL: + logger.debug('Channel unsubscribe check disabled') return if not self.bot: - logger.debug('⚠️ Пропускаем проверку подписки на канал — бот недоступен') + logger.debug('Skipping channel subscription check - bot unavailable') return + from app.database.crud.required_channel import upsert_user_channel_sub + from app.services.channel_subscription_service import channel_subscription_service + from app.utils.cache import ChannelSubCache + + channels = await channel_subscription_service.get_required_channels() + if not channels: + return + + # Ensure bot is set on service + if not channel_subscription_service.bot: + channel_subscription_service.bot = self.bot + try: now = datetime.now(UTC) notifications_allowed = ( NotificationSettingsService.are_notifications_globally_enabled() and NotificationSettingsService.is_trial_channel_unsubscribed_enabled() ) - result = await db.execute( - select(Subscription) - .join(Subscription.user) - .options( - selectinload(Subscription.user), - selectinload(Subscription.tariff), - ) - .where( - and_( - Subscription.is_trial.is_(True), - Subscription.end_date > now, - Subscription.status.in_( - [ - SubscriptionStatus.ACTIVE.value, - SubscriptionStatus.DISABLED.value, - ] - ), - User.status == UserStatus.ACTIVE.value, - ) - ) - ) - - subscriptions = result.scalars().all() - if not subscriptions: - return disabled_count = 0 restored_count = 0 + checked_count = 0 + last_id = 0 - for subscription in subscriptions: - user = subscription.user - if not user or not user.telegram_id: - continue + # Build the trial/all filter based on CHANNEL_REQUIRED_FOR_ALL setting + from sqlalchemy import true as sa_true - try: - member = await self.bot.get_chat_member(channel_id, user.telegram_id) - member_status = member.status - is_member = member_status in ( - ChatMemberStatus.MEMBER, - ChatMemberStatus.ADMINISTRATOR, - ChatMemberStatus.CREATOR, - ) - except TelegramForbiddenError as error: - logger.error( - '❌ Не удалось проверить подписку пользователя на канал : бот заблокирован', - telegram_id=user.telegram_id, - channel_id=channel_id, - error=error, - ) - continue - except TelegramBadRequest as error: - # PARTICIPANT_ID_INVALID - пользователь никогда не был в канале, это нормально - logger.warning( - '⚠️ Ошибка Telegram при проверке подписки пользователя', - telegram_id=user.telegram_id, - error=error, - ) - continue - except Exception as error: - logger.error( - '❌ Неожиданная ошибка при проверке подписки пользователя', - telegram_id=user.telegram_id, - error=error, - ) - continue + is_trial_filter = ( + sa_true() if settings.CHANNEL_REQUIRED_FOR_ALL else Subscription.is_trial.is_(True) + ) - if subscription.status == SubscriptionStatus.ACTIVE.value and subscription.is_trial and not is_member: - if is_recently_updated_by_webhook(subscription): - logger.debug( - 'Пропуск деактивации trial подписки : обновлена вебхуком недавно', - subscription_id=subscription.id, + while True: + # Fresh session per batch to avoid long-running connections + async with AsyncSessionLocal() as batch_db: + result = await batch_db.execute( + select(Subscription) + .join(Subscription.user) + .options( + selectinload(Subscription.user), + selectinload(Subscription.tariff), ) - continue - subscription = await deactivate_subscription(db, subscription) - disabled_count += 1 - logger.info( - '🚫 Триальная подписка пользователя (ID) отключена из-за отписки от канала', - telegram_id=user.telegram_id, - subscription_id=subscription.id, + .where( + and_( + Subscription.id > last_id, + is_trial_filter, + Subscription.end_date > now, + Subscription.status.in_( + [ + SubscriptionStatus.ACTIVE.value, + SubscriptionStatus.DISABLED.value, + ] + ), + User.status == UserStatus.ACTIVE.value, + ) + ) + .order_by(Subscription.id) + .limit(_CHANNEL_CHECK_BATCH_SIZE) ) - if user.remnawave_uuid: - try: - await self.subscription_service.disable_remnawave_user(user.remnawave_uuid) - except Exception as api_error: - logger.error( - '❌ Не удалось отключить пользователя RemnaWave', - remnawave_uuid=user.remnawave_uuid, - api_error=api_error, + subscriptions = result.scalars().all() + if not subscriptions: + break + + last_id = subscriptions[-1].id + + for subscription in subscriptions: + user = subscription.user + if not user or not user.telegram_id: + continue + + # Existing guard: skip if recently updated by webhook + if is_recently_updated_by_webhook(subscription): + logger.debug( + 'Skipping subscription: recently updated by webhook', + subscription_id=subscription.id, + ) + continue + + checked_count += 1 + + # Rate-limited check for ALL channels + all_subscribed = True + for ch in channels: + is_member = await channel_subscription_service._rate_limited_check( + user.telegram_id, ch['channel_id'] + ) + # Update DB + cache + await upsert_user_channel_sub( + batch_db, user.telegram_id, ch['channel_id'], is_member + ) + await ChannelSubCache.set_sub_status( + user.telegram_id, ch['channel_id'], is_member ) - if notifications_allowed: - if not await notification_sent( - db, - user.id, - subscription.id, - 'trial_channel_unsubscribed', - ): - sent = await self._send_trial_channel_unsubscribed_notification(user) - if sent: - await record_notification( - db, + if not is_member: + all_subscribed = False + + # DEACTIVATE: was active, now not subscribed to all + if subscription.status == SubscriptionStatus.ACTIVE.value and not all_subscribed: + # Guard: always skip paid subscriptions (user paid money) + if is_active_paid_subscription(subscription): + continue + + subscription = await deactivate_subscription(batch_db, subscription) + disabled_count += 1 + logger.info( + 'Subscription deactivated (channel unsubscribe)', + telegram_id=user.telegram_id, + subscription_id=subscription.id, + is_trial=subscription.is_trial, + ) + + if user.remnawave_uuid: + try: + await self.subscription_service.disable_remnawave_user( + user.remnawave_uuid + ) + except Exception as api_error: + logger.error( + 'Failed to disable RemnaWave user', + remnawave_uuid=user.remnawave_uuid, + api_error=api_error, + ) + + if notifications_allowed: + if not await notification_sent( + batch_db, user.id, subscription.id, 'trial_channel_unsubscribed', + ): + sent = await self._send_trial_channel_unsubscribed_notification(user) + if sent: + await record_notification( + batch_db, + user.id, + subscription.id, + 'trial_channel_unsubscribed', + ) + + # REACTIVATE: was disabled, now subscribed to all + elif subscription.status == SubscriptionStatus.DISABLED.value and all_subscribed: + # Guard: traffic limit exhausted + if ( + subscription.traffic_limit_gb + and subscription.traffic_used_gb is not None + and subscription.traffic_used_gb >= subscription.traffic_limit_gb + ): + logger.debug( + 'Skipping reactivation: traffic exhausted', + subscription_id=subscription.id, + traffic_used=subscription.traffic_used_gb, + traffic_limit=subscription.traffic_limit_gb, ) - elif subscription.status == SubscriptionStatus.DISABLED.value and subscription.is_trial and is_member: - # Don't reactivate if traffic limit is exhausted (RemnaWave will just disable again) - if ( - subscription.traffic_limit_gb - and subscription.traffic_used_gb is not None - and subscription.traffic_used_gb >= subscription.traffic_limit_gb - ): - logger.debug( - 'Пропуск реактивации trial подписки: трафик исчерпан', - subscription_id=subscription.id, - traffic_used=subscription.traffic_used_gb, - traffic_limit=subscription.traffic_limit_gb, - ) - continue + continue - # Don't reactivate if subscription was disabled by RemnaWave (webhook) - # rather than by monitoring (channel unsubscribe). - # When webhook disables: last_webhook_update_at ≈ updated_at (both set to now()) - # When monitoring disables: updated_at is set, last_webhook_update_at stays old - if ( - subscription.last_webhook_update_at - and subscription.updated_at - and subscription.last_webhook_update_at >= subscription.updated_at - timedelta(seconds=10) - ): - logger.debug( - 'Пропуск реактивации trial подписки: отключена RemnaWave панелью', - subscription_id=subscription.id, - last_webhook_at=subscription.last_webhook_update_at, - updated_at=subscription.updated_at, - ) - continue + # Guard: disabled by webhook, not by monitoring + if ( + subscription.last_webhook_update_at + and subscription.updated_at + and subscription.last_webhook_update_at + >= subscription.updated_at - timedelta(seconds=10) + ): + logger.debug( + 'Skipping reactivation: disabled by RemnaWave panel', + subscription_id=subscription.id, + last_webhook_at=subscription.last_webhook_update_at, + updated_at=subscription.updated_at, + ) + continue - subscription.status = SubscriptionStatus.ACTIVE.value - subscription.updated_at = datetime.now(UTC) - await db.commit() - await db.refresh(subscription) - restored_count += 1 + subscription.status = SubscriptionStatus.ACTIVE.value + subscription.updated_at = datetime.now(UTC) + restored_count += 1 - logger.info( - '✅ Триальная подписка пользователя (ID) восстановлена после повторной подписки на канал', - telegram_id=user.telegram_id, - subscription_id=subscription.id, - ) + logger.info( + 'Subscription restored (channel resubscribe)', + telegram_id=user.telegram_id, + subscription_id=subscription.id, + is_trial=subscription.is_trial, + ) - try: - if user.remnawave_uuid: - await self.subscription_service.update_remnawave_user(db, subscription) - else: - await self.subscription_service.create_remnawave_user(db, subscription) - except Exception as api_error: - logger.error( - '❌ Не удалось обновить RemnaWave пользователя', - telegram_id=user.telegram_id, - api_error=api_error, - ) + try: + if user.remnawave_uuid: + await self.subscription_service.update_remnawave_user( + batch_db, subscription + ) + else: + await self.subscription_service.create_remnawave_user( + batch_db, subscription + ) + except Exception as api_error: + logger.error( + 'Failed to update RemnaWave user', + telegram_id=user.telegram_id, + api_error=api_error, + ) - await clear_notification_by_type( - db, - subscription.id, - 'trial_channel_unsubscribed', - ) + await clear_notification_by_type( + batch_db, + subscription.id, + 'trial_channel_unsubscribed', + ) + + # Commit all changes for this batch + await batch_db.commit() if disabled_count or restored_count: + check_scope = 'all' if settings.CHANNEL_REQUIRED_FOR_ALL else 'trial' await self._log_monitoring_event( db, 'trial_channel_subscription_check', ( - f'Проверено {len(subscriptions)} триальных подписок: отключено {disabled_count}, ' - f'восстановлено {restored_count}' + f'Checked {checked_count} {check_scope} subscriptions: ' + f'disabled {disabled_count}, restored {restored_count}' ), { - 'checked': len(subscriptions), + 'checked': checked_count, 'disabled': disabled_count, 'restored': restored_count, + 'scope': check_scope, }, ) except Exception as error: - logger.error('Ошибка проверки подписки на канал для триальных пользователей', error=error) + logger.error('Error checking channel subscriptions', error=error) async def _check_expired_subscription_followups(self, db: AsyncSession): if not NotificationSettingsService.are_notifications_globally_enabled(): @@ -1336,16 +1378,16 @@ class MonitoringService: from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup + from app.services.channel_subscription_service import channel_subscription_service + + unsubscribed = await channel_subscription_service.get_unsubscribed_channels(user.telegram_id) + buttons = [] - if settings.CHANNEL_LINK: - buttons.append( - [ - InlineKeyboardButton( - text=texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться'), - url=settings.CHANNEL_LINK, - ) - ] - ) + for ch in unsubscribed: + link = ch.get('channel_link') + if link: + title = ch.get('title') or texts.t('CHANNEL_SUBSCRIBE_BUTTON', '🔗 Подписаться') + buttons.append([InlineKeyboardButton(text=f'🔗 {title}', url=link)]) buttons.append( [ InlineKeyboardButton( diff --git a/app/services/payment/yookassa.py b/app/services/payment/yookassa.py index 7fa7811d..cb25ac55 100644 --- a/app/services/payment/yookassa.py +++ b/app/services/payment/yookassa.py @@ -258,7 +258,7 @@ class YooKassaPaymentMixin: status=yookassa_response['status'], confirmation_url=yookassa_response.get('confirmation_url'), # Используем confirmation URL metadata_json=payment_metadata, - payment_method_type='bank_card', + payment_method_type='sbp', yookassa_created_at=None, test_mode=yookassa_response.get('test_mode', False), ) @@ -897,6 +897,7 @@ class YooKassaPaymentMixin: f'🔄 При наличии сохранённой корзины подписки и включенной автопокупке, ' f'подписка будет приобретена автоматически после пополнения баланса.\n\n{cart_message}', reply_markup=keyboard, + parse_mode='HTML', ) logger.info( 'Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю', @@ -1079,14 +1080,14 @@ class YooKassaPaymentMixin: 'Успешно обработан платеж YooKassa как покупка подписки: пользователь , сумма ₽', yookassa_payment_id=payment.yookassa_payment_id, user_id=payment.user_id, - amount_kopeks=payment.amount_kopeks / 100, + amount_rubles=payment.amount_kopeks / 100, ) else: logger.info( 'Успешно обработан платеж YooKassa : пользователь пополнил баланс на ₽', yookassa_payment_id=payment.yookassa_payment_id, user_id=payment.user_id, - amount_kopeks=payment.amount_kopeks / 100, + amount_rubles=payment.amount_kopeks / 100, ) # Создаем чек через NaloGO (если NALOGO_ENABLED=true) diff --git a/app/services/referral_contest_service.py b/app/services/referral_contest_service.py index 4474345c..a1d5253e 100644 --- a/app/services/referral_contest_service.py +++ b/app/services/referral_contest_service.py @@ -323,14 +323,11 @@ class ReferralContestService: if not self.bot: return - channel_id_raw = settings.CHANNEL_SUB_ID - if not channel_id_raw: - return + from app.services.channel_subscription_service import channel_subscription_service - try: - channel_id = int(channel_id_raw) - except Exception: - channel_id = channel_id_raw + channel_id = await channel_subscription_service.get_first_channel_id() + if not channel_id: + return lines = [ f'🏆 {contest.title}', @@ -358,9 +355,9 @@ class ReferralContestService: disable_web_page_preview=True, ) except (TelegramForbiddenError, TelegramNotFound): - logger.info('Не удалось отправить сводку конкурса в канал', channel_id_raw=channel_id_raw) + logger.info('Не удалось отправить сводку конкурса в канал', channel_id=channel_id) except Exception as exc: - logger.error('Ошибка отправки сводки конкурса в канал', channel_id_raw=channel_id_raw, exc=exc) + logger.error('Ошибка отправки сводки конкурса в канал', channel_id=channel_id, exc=exc) def _build_participant_message( self, diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py index 829bd383..f8eb7eca 100644 --- a/app/services/system_settings_service.py +++ b/app/services/system_settings_service.py @@ -205,8 +205,6 @@ class BotConfigurationService: 'DATABASE_URL': 'DATABASE', 'DATABASE_MODE': 'DATABASE', 'LOCALES_PATH': 'LOCALIZATION', - 'CHANNEL_SUB_ID': 'CHANNEL', - 'CHANNEL_LINK': 'CHANNEL', 'CHANNEL_IS_REQUIRED_SUB': 'CHANNEL', 'BOT_USERNAME': 'CORE', 'DEFAULT_LANGUAGE': 'LOCALIZATION', diff --git a/app/services/yookassa_service.py b/app/services/yookassa_service.py index 8403b89f..68f7435b 100644 --- a/app/services/yookassa_service.py +++ b/app/services/yookassa_service.py @@ -102,7 +102,7 @@ class YooKassaService: 'description': description[:128], 'quantity': '1.00', 'amount': {'value': str(round(amount, 2)), 'currency': currency.upper()}, - 'vat_code': str(getattr(settings, 'YOOKASSA_VAT_CODE', 1)), + 'vat_code': int(getattr(settings, 'YOOKASSA_VAT_CODE', 1)), 'payment_mode': getattr(settings, 'YOOKASSA_PAYMENT_MODE', 'full_payment'), 'payment_subject': getattr(settings, 'YOOKASSA_PAYMENT_SUBJECT', 'service'), } @@ -208,7 +208,7 @@ class YooKassaService: 'description': description[:128], 'quantity': '1.00', 'amount': {'value': str(round(amount, 2)), 'currency': currency.upper()}, - 'vat_code': str(getattr(settings, 'YOOKASSA_VAT_CODE', 1)), + 'vat_code': int(getattr(settings, 'YOOKASSA_VAT_CODE', 1)), 'payment_mode': getattr(settings, 'YOOKASSA_PAYMENT_MODE', 'full_payment'), 'payment_subject': getattr(settings, 'YOOKASSA_PAYMENT_SUBJECT', 'service'), } @@ -223,7 +223,7 @@ class YooKassaService: payment_request = builder.build() logger.info( - "Создание платежа YooKassa СБП с подтверждением 'qr' (Idempotence-Key: ). Сумма: . Метаданные: . Чек", + 'Создание платежа YooKassa СБП с подтверждением redirect (Idempotence-Key: ). Сумма: . Метаданные: . Чек', idempotence_key=idempotence_key, amount=amount, currency=currency, @@ -237,14 +237,14 @@ class YooKassaService: ) logger.info( - 'Ответ YooKassa Payment.create (СБП, qr): ID=, Status=, Paid', + 'Ответ YooKassa Payment.create (СБП, redirect): ID=, Status=, Paid', response_id=response.id, status=response.status, paid=response.paid, ) - # Возвращаем данные платежа с QR-подтверждением - # Пользователь может использовать QR-код или оплатить через приложение банка по ID платежа + # Возвращаем данные платежа с redirect-подтверждением + # YooKassa покажет QR на десктопе или список банков на мобильном return { 'id': response.id, 'qr_confirmation_data': response.confirmation.confirmation_data @@ -270,109 +270,6 @@ class YooKassaService: logger.error('Ошибка создания платежа YooKassa СБП', error=e, exc_info=True) return None - async def _create_sbp_payment_with_confirmation_type( - self, - amount: float, - currency: str, - description: str, - metadata: dict[str, Any], - customer_contact_for_receipt: dict[str, str], - confirmation_type: str, - ) -> dict[str, Any] | None: - """Создает SBP платеж с указанным типом подтверждения""" - try: - builder = PaymentRequestBuilder() - - builder.set_amount({'value': str(round(amount, 2)), 'currency': currency.upper()}) - - builder.set_capture(True) - - if confirmation_type == 'qr': - builder.set_confirmation({'type': 'qr'}) - else: # redirect - builder.set_confirmation({'type': 'redirect', 'return_url': self.return_url}) - - builder.set_description(description) - - builder.set_metadata(metadata) - - builder.set_payment_method_data({'type': 'sbp'}) - - receipt_items_list: list[dict[str, Any]] = [ - { - 'description': description[:128], - 'quantity': '1.00', - 'amount': {'value': str(round(amount, 2)), 'currency': currency.upper()}, - 'vat_code': str(getattr(settings, 'YOOKASSA_VAT_CODE', 1)), - 'payment_mode': getattr(settings, 'YOOKASSA_PAYMENT_MODE', 'full_payment'), - 'payment_subject': getattr(settings, 'YOOKASSA_PAYMENT_SUBJECT', 'service'), - } - ] - - receipt_data_dict: dict[str, Any] = {'customer': customer_contact_for_receipt, 'items': receipt_items_list} - - builder.set_receipt(receipt_data_dict) - - idempotence_key = str(uuid.uuid4()) - - payment_request = builder.build() - - logger.info( - 'Создание платежа YooKassa СБП с подтверждением (Idempotence-Key: ). Сумма: . Метаданные: . Чек', - confirmation_type=confirmation_type, - idempotence_key=idempotence_key, - amount=amount, - currency=currency, - metadata=metadata, - receipt_data_dict=receipt_data_dict, - ) - - loop = asyncio.get_running_loop() - response = await loop.run_in_executor( - None, lambda: YooKassaPayment.create(payment_request, idempotence_key) - ) - - logger.info( - 'Ответ YooKassa Payment.create (СБП, ): ID=, Status=, Paid', - confirmation_type=confirmation_type, - response_id=response.id, - status=response.status, - paid=response.paid, - ) - - result = { - 'id': response.id, - 'status': response.status, - 'metadata': response.metadata, - 'amount_value': float(response.amount.value), - 'amount_currency': response.amount.currency, - 'idempotence_key_used': idempotence_key, - 'paid': response.paid, - 'refundable': response.refundable, - 'created_at': response.created_at.isoformat() - if hasattr(response.created_at, 'isoformat') - else str(response.created_at), - 'description_from_yk': response.description, - 'test_mode': response.test if hasattr(response, 'test') else None, - } - - # Добавляем данные подтверждения в зависимости от типа - if confirmation_type == 'qr': - if response.confirmation and hasattr(response.confirmation, 'confirmation_data'): - result['confirmation_data'] = response.confirmation.confirmation_data - elif response.confirmation and hasattr(response.confirmation, 'confirmation_url'): - result['confirmation_url'] = response.confirmation.confirmation_url - - return result - except Exception as e: - logger.error( - 'Ошибка создания платежа YooKassa СБП с подтверждением', - confirmation_type=confirmation_type, - error=e, - exc_info=True, - ) - return None - async def get_payment_info(self, payment_id_in_yookassa: str) -> dict[str, Any] | None: if not self.configured: logger.error('YooKassa не сконфигурирован. Невозможно получить информацию о платеже.') diff --git a/app/utils/cache.py b/app/utils/cache.py index 86e6ed6a..ab0de3a2 100644 --- a/app/utils/cache.py +++ b/app/utils/cache.py @@ -342,3 +342,97 @@ class RateLimitCache: async def reset_rate_limit(user_id: int, action: str) -> bool: key = cache_key('rate_limit', user_id, action) return await cache.delete(key) + + +class ChannelSubCache: + """Cache for user channel subscription statuses. + + Redis keys: + - channel_sub:{telegram_id}:{channel_id} -> "1" or "0" (TTL 600s) + - required_channels:active -> JSON list of active channels (TTL 60s) + """ + + SUB_TTL = 600 # 10 min -- individual user subscription status + CHANNELS_TTL = 60 # 1 min -- list of required channels + + @staticmethod + async def get_sub_status(telegram_id: int, channel_id: str) -> bool | None: + """Get subscription status from cache. None = cache miss.""" + key = cache_key('channel_sub', telegram_id, channel_id) + result = await cache.get(key) + if result is None: + return None + return result == 1 + + @staticmethod + async def get_sub_statuses(telegram_id: int, channel_ids: list[str]) -> dict[str, bool | None]: + """Batch-fetch subscription statuses via Redis MGET (single round-trip). + + Returns {channel_id: True/False/None} where None = cache miss. + Falls back to sequential gets if Redis pipeline is unavailable. + """ + if not channel_ids: + return {} + + if not cache._connected or cache.redis_client is None: + return dict.fromkeys(channel_ids, None) + + keys = [cache_key('channel_sub', telegram_id, ch_id) for ch_id in channel_ids] + try: + raw_values = await cache.redis_client.mget(keys) + except Exception as e: + logger.warning('Redis MGET failed, falling back to sequential', error=str(e)) + result: dict[str, bool | None] = {} + for ch_id in channel_ids: + result[ch_id] = await ChannelSubCache.get_sub_status(telegram_id, ch_id) + return result + + statuses: dict[str, bool | None] = {} + for ch_id, raw in zip(channel_ids, raw_values, strict=True): + if raw is None: + statuses[ch_id] = None + else: + try: + parsed = json.loads(raw) + statuses[ch_id] = parsed == 1 + except (ValueError, TypeError): + statuses[ch_id] = None + return statuses + + @staticmethod + async def set_sub_status(telegram_id: int, channel_id: str, is_member: bool) -> None: + key = cache_key('channel_sub', telegram_id, channel_id) + await cache.set(key, 1 if is_member else 0, expire=ChannelSubCache.SUB_TTL) + + @staticmethod + async def invalidate_sub(telegram_id: int, channel_id: str) -> None: + key = cache_key('channel_sub', telegram_id, channel_id) + await cache.delete(key) + + @staticmethod + async def invalidate_user_channels(telegram_id: int, channel_ids: list[str]) -> None: + """Invalidate specific channel keys for a user using single Redis DELETE. + + Uses multi-key DELETE (O(K)) instead of delete_pattern() which uses KEYS (O(N)). + At 100k users * 5 channels = 500k keys, KEYS would block Redis for seconds. + """ + if not channel_ids or not cache._connected or not cache.redis_client: + return + keys = [cache_key('channel_sub', telegram_id, ch_id) for ch_id in channel_ids] + try: + await cache.redis_client.delete(*keys) + except Exception as e: + logger.warning('Failed to invalidate user channel cache', telegram_id=telegram_id, error=e) + + @staticmethod + async def get_required_channels() -> list[dict] | None: + """Get the list of required channels from cache.""" + return await cache.get('required_channels:active') + + @staticmethod + async def set_required_channels(channels: list[dict]) -> None: + await cache.set('required_channels:active', channels, expire=ChannelSubCache.CHANNELS_TTL) + + @staticmethod + async def invalidate_channels() -> None: + await cache.delete('required_channels:active') diff --git a/app/webapi/routes/miniapp.py b/app/webapi/routes/miniapp.py index d92c7e5b..9a5331b9 100644 --- a/app/webapi/routes/miniapp.py +++ b/app/webapi/routes/miniapp.py @@ -205,17 +205,6 @@ router = APIRouter() promo_code_service = PromoCodeService() renewal_service = SubscriptionRenewalService() -# Кешированный Bot для проверки подписки на канал (снижает нагрузку) -_channel_check_bot: Bot | None = None - - -def _get_channel_check_bot() -> Bot: - """Получить или создать Bot для проверки подписки на канал.""" - global _channel_check_bot - if _channel_check_bot is None: - _channel_check_bot = Bot(token=settings.BOT_TOKEN) - return _channel_check_bot - _CRYPTOBOT_MIN_USD = 1.0 _CRYPTOBOT_MAX_USD = 1000.0 @@ -3100,26 +3089,21 @@ async def get_subscription_details( ) from None # Check required channel subscription - if settings.CHANNEL_IS_REQUIRED_SUB and settings.CHANNEL_SUB_ID: - try: - bot = _get_channel_check_bot() - chat_member = await bot.get_chat_member(chat_id=settings.CHANNEL_SUB_ID, user_id=telegram_id) - # Не закрываем сессию - бот переиспользуется + if settings.CHANNEL_IS_REQUIRED_SUB: + from app.services.channel_subscription_service import channel_subscription_service - if chat_member.status not in ['member', 'administrator', 'creator']: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail={ - 'code': 'channel_subscription_required', - 'message': 'Please subscribe to our channel to continue', - 'channel_link': settings.CHANNEL_LINK, - }, - ) - except HTTPException: - raise - except Exception as e: - logger.warning('Failed to check channel subscription for user', telegram_id=telegram_id, error=e) - # Don't block user if check fails + channels_with_status = await channel_subscription_service.get_channels_with_status(telegram_id) + is_subscribed = all(ch['is_subscribed'] for ch in channels_with_status) if channels_with_status else True + + if not is_subscribed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + 'code': 'channel_subscription_required', + 'message': 'Please subscribe to the required channels to continue', + 'channels': channels_with_status, + }, + ) user = await get_user_by_telegram_id(db, telegram_id) purchase_url = (settings.MINIAPP_PURCHASE_URL or '').strip() diff --git a/main.py b/main.py index dcc10173..866c578b 100644 --- a/main.py +++ b/main.py @@ -287,6 +287,10 @@ async def main(): daily_subscription_service.set_bot(bot) telegram_notifier.set_bot(bot) + from app.services.channel_subscription_service import channel_subscription_service + + channel_subscription_service.bot = bot + # Initialize email broadcast service from app.cabinet.services.email_service import email_service from app.services.broadcast_service import email_broadcast_service diff --git a/migrations/alembic/versions/0008_add_required_channels.py b/migrations/alembic/versions/0008_add_required_channels.py new file mode 100644 index 00000000..22d2431e --- /dev/null +++ b/migrations/alembic/versions/0008_add_required_channels.py @@ -0,0 +1,61 @@ +"""add required_channels and user_channel_subscriptions tables + +Revision ID: 0008 +Revises: 0007 +Create Date: 2026-02-24 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = '0008' +down_revision: Union[str, None] = '0007' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _has_table(table_name: str) -> bool: + """Check if table already exists (idempotency guard).""" + conn = op.get_bind() + result = conn.execute( + sa.text( + 'SELECT EXISTS (SELECT 1 FROM information_schema.tables ' + "WHERE table_schema = 'public' AND table_name = :name)" + ), + {'name': table_name}, + ) + return result.scalar() + + +def upgrade() -> None: + if not _has_table('required_channels'): + op.create_table( + 'required_channels', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('channel_id', sa.String(100), unique=True, nullable=False), + sa.Column('channel_link', sa.String(500), nullable=True), + sa.Column('title', sa.String(255), nullable=True), + sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'), + sa.Column('sort_order', sa.Integer(), nullable=False, server_default='0'), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), + ) + + if not _has_table('user_channel_subscriptions'): + op.create_table( + 'user_channel_subscriptions', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('telegram_id', sa.BigInteger(), nullable=False), + sa.Column('channel_id', sa.String(100), nullable=False), + sa.Column('is_member', sa.Boolean(), nullable=False, server_default='false'), + sa.Column('checked_at', sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.UniqueConstraint('telegram_id', 'channel_id', name='uq_user_channel_sub'), + sa.Index('ix_user_channel_sub_telegram_id', 'telegram_id'), + ) + + +def downgrade() -> None: + op.drop_table('user_channel_subscriptions') + op.drop_table('required_channels') diff --git a/migrations/alembic/versions/0009_add_channel_id_index.py b/migrations/alembic/versions/0009_add_channel_id_index.py new file mode 100644 index 00000000..8d4c30c5 --- /dev/null +++ b/migrations/alembic/versions/0009_add_channel_id_index.py @@ -0,0 +1,28 @@ +"""add channel_id index for user_channel_subscriptions + +Revision ID: 0009 +Revises: 0008 +Create Date: 2026-02-24 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = '0009' +down_revision: Union[str, None] = '0008' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_index( + 'ix_user_channel_sub_channel_id', + 'user_channel_subscriptions', + ['channel_id'], + if_not_exists=True, + ) + + +def downgrade() -> None: + op.drop_index('ix_user_channel_sub_channel_id', table_name='user_channel_subscriptions') diff --git a/tests/conftest.py b/tests/conftest.py index 0e17fae0..ccd4bef9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -146,11 +146,22 @@ if 'yookassa' not in sys.modules: payment_builder_module.PaymentRequestBuilder = _FakePaymentRequestBuilder confirmation_module.ConfirmationType = _FakeConfirmationType + exceptions_module = types.ModuleType('yookassa.domain.exceptions') + not_found_module = types.ModuleType('yookassa.domain.exceptions.not_found_error') + + class _FakeNotFoundError(Exception): + pass + + not_found_module.NotFoundError = _FakeNotFoundError + exceptions_module.not_found_error = not_found_module + sys.modules['yookassa.domain'] = domain_module sys.modules['yookassa.domain.request'] = request_module sys.modules['yookassa.domain.request.payment_request_builder'] = payment_builder_module sys.modules['yookassa.domain.common'] = common_module sys.modules['yookassa.domain.common.confirmation_type'] = confirmation_module + sys.modules['yookassa.domain.exceptions'] = exceptions_module + sys.modules['yookassa.domain.exceptions.not_found_error'] = not_found_module @pytest.fixture diff --git a/tests/services/test_payment_service_yookassa.py b/tests/services/test_payment_service_yookassa.py index 1f372530..535fa5bb 100644 --- a/tests/services/test_payment_service_yookassa.py +++ b/tests/services/test_payment_service_yookassa.py @@ -212,7 +212,7 @@ async def test_create_yookassa_sbp_payment_success(monkeypatch: pytest.MonkeyPat assert result is not None assert result['confirmation_token'] == 'token123' - assert captured_args['payment_method_type'] == 'bank_card' + assert captured_args['payment_method_type'] == 'sbp' assert captured_args['metadata_json']['type'] == 'balance_topup_sbp'