Реферальные и глубокие ссылки start

bot/middlewares/referal_mdw.py

@@ -0,0 +1,544 @@
+"""
+Middleware для обработки реферальных ссылок и deep links
+"""
+from typing import Callable, Awaitable, Any, Dict, Optional
+from dataclasses import dataclass, field
+from datetime import datetime
+from collections import defaultdict
+import re
+
+from aiogram import BaseMiddleware
+from aiogram.filters.command import CommandObject
+from aiogram.types import TelegramObject, Message, User
+
+from middleware.loggers import logger
+
+__all__ = (
+    'ReferralMiddleware',
+    'DeepLinkData',
+    'referral_stats',
+    'ReferralType'
+)
+
+
+class ReferralType:
+    """Типы реферальных ссылок"""
+    REFERRAL = 'ref'  # Обычная реферальная ссылка
+    PROMO = 'promo'  # Промокод
+    UTM = 'utm'  # UTM метки
+    INVITE = 'invite'  # Инвайт-ссылка
+    DEEPLINK = 'deeplink'  # Произвольный deep link
+    CUSTOM = 'custom'  # Кастомный тип
+
+
+@dataclass
+class DeepLinkData:
+    """
+    Данные deep link.
+
+    Attributes:
+        raw: Исходная строка (все после /start)
+        type: Тип ссылки (ref, promo, utm, и т.д.)
+        params: Распарсенные параметры
+        user_id: ID пользователя, перешедшего по ссылке
+        username: Username пользователя
+        timestamp: Время перехода
+        is_valid: Валидна ли ссылка
+    """
+    raw: str
+    type: str = ReferralType.DEEPLINK
+    params: Dict[str, Any] = field(default_factory=dict)
+    user_id: Optional[int] = None
+    username: Optional[str] = None
+    timestamp: datetime = field(default_factory=datetime.now)
+    is_valid: bool = True
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """Получает параметр по ключу"""
+        return self.params.get(key, default)
+
+    def __getitem__(self, key: str) -> Any:
+        """Позволяет использовать data['key']"""
+        return self.params[key]
+
+    def __contains__(self, key: str) -> bool:
+        """Позволяет использовать 'key' in data"""
+        return key in self.params
+
+
+class ReferralStatistics:
+    """
+    Статистика реферальных переходов.
+    """
+
+    def __init__(self):
+        # Счетчики переходов по типам: {type: count}
+        self.clicks_by_type: Dict[str, int] = defaultdict(int)
+
+        # Переходы по кодам: {ref_code: count}
+        self.clicks_by_code: Dict[str, int] = defaultdict(int)
+
+        # История переходов: [(timestamp, user_id, ref_code, type), ...]
+        self.history: list[tuple[datetime, int, str, str]] = []
+
+        # Уникальные пользователи: {ref_code: set(user_ids)}
+        self.unique_users: Dict[str, set[int]] = defaultdict(set)
+
+    def record(self, deep_link: DeepLinkData) -> None:
+        """Записывает переход"""
+        # Счетчик по типу
+        self.clicks_by_type[deep_link.type] += 1
+
+        # Счетчик по коду (если есть реферальный код)
+        ref_code = deep_link.get('ref_code') or deep_link.get('code') or deep_link.raw
+        if ref_code:
+            self.clicks_by_code[ref_code] += 1
+
+            # Уникальные пользователи
+            if deep_link.user_id:
+                self.unique_users[ref_code].add(deep_link.user_id)
+
+        # История
+        if deep_link.user_id:
+            self.history.append((
+                deep_link.timestamp,
+                deep_link.user_id,
+                ref_code,
+                deep_link.type
+            ))
+
+    def get_stats(self, ref_code: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Возвращает статистику.
+
+        Args:
+            ref_code: Код для фильтрации (если None, возвращает общую статистику)
+        """
+        if ref_code:
+            return {
+                'ref_code': ref_code,
+                'total_clicks': self.clicks_by_code.get(ref_code, 0),
+                'unique_users': len(self.unique_users.get(ref_code, set()))
+            }
+
+        return {
+            'total_clicks': sum(self.clicks_by_type.values()),
+            'clicks_by_type': dict(self.clicks_by_type),
+            'top_codes': self.get_top_codes(10),
+            'total_unique_users': sum(len(users) for users in self.unique_users.values())
+        }
+
+    def get_top_codes(self, limit: int = 10) -> list[tuple[str, int]]:
+        """Возвращает топ реферальных кодов"""
+        sorted_codes = sorted(
+            self.clicks_by_code.items(),
+            key=lambda x: x[1],
+            reverse=True
+        )
+        return sorted_codes[:limit]
+
+
+# Глобальная статистика
+referral_stats = ReferralStatistics()
+
+
+class ReferralMiddleware(BaseMiddleware):
+    """
+    Middleware для обработки реферальных ссылок и deep links.
+
+    Возможности:
+    - Парсинг различных форматов deep links
+    - Автоматическое определение типа ссылки
+    - Валидация параметров
+    - Сбор статистики
+    - Интеграция с базой данных через callback
+    - Поддержка сложных параметров (ref_123_promo_abc)
+
+    Поддерживаемые форматы:
+    - /start ref123              → {'ref_code': 'ref123'}
+    - /start promo_SUMMER2024    → {'type': 'promo', 'code': 'SUMMER2024'}
+    - /start ref_123_bonus_50    → {'ref_code': '123', 'bonus': '50'}
+    - /start utm_source_telegram → {'utm_source': 'telegram'}
+
+    Attributes:
+        on_referral: Callback функция для сохранения в БД
+        validator: Функция валидации кодов
+        parse_complex: Парсить ли сложные параметры
+        collect_stats: Собирать ли статистику
+
+    Example:
+        ```python
+        from middleware.referral import ReferralMiddleware, DeepLinkData
+
+        async def save_referral(deep_link: DeepLinkData):
+            # Сохранение в БД
+            await db.save_referral(
+                user_id=deep_link.user_id,
+                ref_code=deep_link.get('ref_code'),
+                timestamp=deep_link.timestamp
+            )
+
+        # Регистрация middleware
+        referral_mdw = ReferralMiddleware(
+            on_referral=save_referral,
+            parse_complex=True,
+            collect_stats=True
+        )
+
+        dp.message.middleware(referral_mdw)
+
+        # В хендлере
+        @router.message(CommandStart())
+        async def start(message: Message, deep_link: Optional[DeepLinkData] = None):
+            if deep_link:
+                ref_code = deep_link.get('ref_code')
+                await message.answer(f"Привет! Вы пришли по ссылке: {ref_code}")
+            else:
+                await message.answer("Привет!")
+        ```
+    """
+
+    # Паттерны для парсинга
+    PATTERNS = {
+        # ref_123 или ref123
+        ReferralType.REFERRAL: re.compile(r'^ref[_-]?(\w+)$', re.IGNORECASE),
+
+        # promo_SUMMER2024
+        ReferralType.PROMO: re.compile(r'^promo[_-]?(\w+)$', re.IGNORECASE),
+
+        # invite_abc123
+        ReferralType.INVITE: re.compile(r'^invite[_-]?(\w+)$', re.IGNORECASE),
+
+        # utm_source_telegram_campaign_ads
+        ReferralType.UTM: re.compile(r'^utm[_-]', re.IGNORECASE),
+    }
+
+    def __init__(
+            self,
+            on_referral: Optional[Callable[[DeepLinkData], Awaitable[None]]] = None,
+            validator: Optional[Callable[[str], bool]] = None,
+            parse_complex: bool = True,
+            collect_stats: bool = True,
+            max_length: int = 64
+    ):
+        """
+        Инициализация middleware.
+
+        Args:
+            on_referral: Callback для обработки реферала (сохранение в БД)
+            validator: Функция валидации кода (должна вернуть True если валиден)
+            parse_complex: Парсить ли сложные параметры (ref_123_bonus_50)
+            collect_stats: Собирать ли статистику
+            max_length: Максимальная длина deep link
+        """
+        super().__init__()
+        self.on_referral = on_referral
+        self.validator = validator
+        self.parse_complex = parse_complex
+        self.collect_stats = collect_stats
+        self.max_length = max_length
+
+    def _parse_simple(self, args: str) -> tuple[str, Dict[str, Any]]:
+        """
+        Парсит простые форматы deep links.
+
+        Args:
+            args: Аргументы команды /start
+
+        Returns:
+            tuple: (тип, параметры)
+        """
+        # Проверка по паттернам
+        for link_type, pattern in self.PATTERNS.items():
+            match = pattern.match(args)
+            if match:
+                if link_type == ReferralType.REFERRAL:
+                    return link_type, {'ref_code': match.group(1)}
+                elif link_type == ReferralType.PROMO:
+                    return link_type, {'code': match.group(1), 'promo_code': match.group(1)}
+                elif link_type == ReferralType.INVITE:
+                    return link_type, {'invite_code': match.group(1)}
+                elif link_type == ReferralType.UTM:
+                    # Парсим UTM параметры
+                    return link_type, self._parse_utm(args)
+
+        # Если не совпало ни с одним паттерном - просто код
+        return ReferralType.DEEPLINK, {'code': args}
+
+    def _parse_utm(self, args: str) -> Dict[str, Any]:
+        """
+        Парсит UTM параметры: utm_source_telegram_campaign_ads
+
+        Args:
+            args: Строка с UTM параметрами
+
+        Returns:
+            Dict с UTM параметрами
+        """
+        params = {}
+
+        # Удаляем префикс utm_
+        if args.lower().startswith('utm_'):
+            args = args[4:]
+
+        # Разбиваем по _ и парсим пары ключ-значение
+        parts = args.split('_')
+
+        i = 0
+        while i < len(parts) - 1:
+            key = f"utm_{parts[i]}"
+            value = parts[i + 1]
+            params[key] = value
+            i += 2
+
+        return params
+
+    def _parse_complex(self, args: str) -> tuple[str, Dict[str, Any]]:
+        """
+        Парсит сложные форматы: ref_123_bonus_50_promo_SUMMER
+
+        Args:
+            args: Аргументы команды
+
+        Returns:
+            tuple: (тип, параметры)
+        """
+        params = {}
+        parts = args.split('_')
+
+        # Определяем тип по первому элементу
+        link_type = ReferralType.DEEPLINK
+
+        if parts[0].lower() in ['ref', 'referral']:
+            link_type = ReferralType.REFERRAL
+            if len(parts) > 1:
+                params['ref_code'] = parts[1]
+            parts = parts[2:]  # Пропускаем первые 2 элемента
+        elif parts[0].lower() == 'promo':
+            link_type = ReferralType.PROMO
+            if len(parts) > 1:
+                params['promo_code'] = parts[1]
+            parts = parts[2:]
+        elif parts[0].lower() == 'invite':
+            link_type = ReferralType.INVITE
+            if len(parts) > 1:
+                params['invite_code'] = parts[1]
+            parts = parts[2:]
+
+        # Парсим остальные параметры как пары ключ-значение
+        i = 0
+        while i < len(parts) - 1:
+            key = parts[i]
+            value = parts[i + 1]
+
+            # Пытаемся преобразовать в число
+            try:
+                value = int(value)
+            except ValueError:
+                try:
+                    value = float(value)
+                except ValueError:
+                    pass  # Оставляем строкой
+
+            params[key] = value
+            i += 2
+
+        return link_type, params
+
+    def _validate_deep_link(self, args: str) -> bool:
+        """
+        Валидирует deep link.
+
+        Args:
+            args: Строка для валидации
+
+        Returns:
+            bool: True если валиден
+        """
+        # Проверка длины
+        if len(args) > self.max_length:
+            logger.warning(
+                f"Deep link слишком длинный: {len(args)} > {self.max_length}",
+                log_type='REFERRAL'
+            )
+            return False
+
+        # Проверка на запрещенные символы (только буквы, цифры, _ и -)
+        if not re.match(r'^[a-zA-Z0-9_-]+$', args):
+            logger.warning(
+                f"Deep link содержит недопустимые символы: {args}",
+                log_type='REFERRAL'
+            )
+            return False
+
+        # Кастомная валидация
+        if self.validator:
+            return self.validator(args)
+
+        return True
+
+    def _parse_deep_link(self, args: str, user: User) -> DeepLinkData:
+        """
+        Парсит deep link и создает объект DeepLinkData.
+
+        Args:
+            args: Аргументы команды /start
+            user: Пользователь, перешедший по ссылке
+
+        Returns:
+            DeepLinkData: Распарсенные данные
+        """
+        # Валидация
+        is_valid = self._validate_deep_link(args)
+
+        # Парсинг
+        if self.parse_complex and '_' in args:
+            link_type, params = self._parse_complex(args)
+        else:
+            link_type, params = self._parse_simple(args)
+
+        # Создаем объект
+        deep_link = DeepLinkData(
+            raw=args,
+            type=link_type,
+            params=params,
+            user_id=user.id,
+            username=user.username,
+            is_valid=is_valid
+        )
+
+        return deep_link
+
+    async def __call__(
+            self,
+            handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
+            event: TelegramObject,
+            data: Dict[str, Any]
+    ) -> Any:
+        """
+        Перехватывает команды /start с аргументами.
+
+        Args:
+            handler: Функция хендлера
+            event: Объект события
+            data: Дополнительные данные
+
+        Returns:
+            Результат хендлера
+        """
+        # Обрабатываем только сообщения
+        if not isinstance(event, Message):
+            return await handler(event, data)
+
+        # Извлекаем команду
+        command: Optional[CommandObject] = data.get('command')
+
+        # Проверяем, что это /start с аргументами
+        if not command or command.command.lower() != 'start' or not command.args:
+            return await handler(event, data)
+
+        user = event.from_user
+        args = command.args
+
+        # Парсим deep link
+        deep_link = self._parse_deep_link(args, user)
+
+        # Логирование
+        if deep_link.is_valid:
+            logger.info(
+                f"Deep link: type={deep_link.type}, params={deep_link.params}",
+                log_type='REFERRAL',
+                user=f"@{user.username}" if user.username else f"id{user.id}"
+            )
+        else:
+            logger.warning(
+                f"Невалидный deep link: {args}",
+                log_type='REFERRAL',
+                user=f"@{user.username}" if user.username else f"id{user.id}"
+            )
+
+        # Собираем статистику
+        if self.collect_stats and deep_link.is_valid:
+            referral_stats.record(deep_link)
+
+        # Вызываем callback для сохранения в БД
+        if self.on_referral and deep_link.is_valid:
+            try:
+                await self.on_referral(deep_link)
+            except Exception as e:
+                logger.error(
+                    f"Ошибка в on_referral callback: {e}",
+                    log_type='REFERRAL'
+                )
+
+        # Добавляем deep_link в data для хендлера
+        data['deep_link'] = deep_link
+        data['ref_code'] = deep_link.get('ref_code')  # Для обратной совместимости
+
+        # Выполняем хендлер
+        return await handler(event, data)
+
+
+# ================= ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =================
+
+def create_deep_link(bot_username: str, **params) -> str:
+    """
+    Создает deep link для бота.
+
+    Args:
+        bot_username: Username бота (без @)
+        **params: Параметры для ссылки
+
+    Returns:
+        str: Готовая ссылка
+
+    Example:
+        >>> create_deep_link('mybot', ref_code='123', bonus='50')
+        'https://t.me/mybot?start=ref_123_bonus_50'
+    """
+    # Формируем строку параметров
+    parts = []
+
+    for key, value in params.items():
+        parts.append(str(key))
+        parts.append(str(value))
+
+    param_string = '_'.join(parts)
+
+    return f"https://t.me/{bot_username}?start={param_string}"
+
+
+def create_referral_link(bot_username: str, ref_code: str) -> str:
+    """
+    Создает простую реферальную ссылку.
+
+    Args:
+        bot_username: Username бота
+        ref_code: Реферальный код
+
+    Returns:
+        str: Реферальная ссылка
+
+    Example:
+        >>> create_referral_link('mybot', '123')
+        'https://t.me/mybot?start=ref_123'
+    """
+    return f"https://t.me/{bot_username}?start=ref_{ref_code}"
+
+
+def create_promo_link(bot_username: str, promo_code: str) -> str:
+    """
+    Создает ссылку с промокодом.
+
+    Args:
+        bot_username: Username бота
+        promo_code: Промокод
+
+    Returns:
+        str: Ссылка с промокодом
+
+    Example:
+        >>> create_promo_link('mybot', 'SUMMER2024')
+        'https://t.me/mybot?start=promo_SUMMER2024'
+    """
+    return f"https://t.me/{bot_username}?start=promo_{promo_code}"