Подготовка к релизу

utils/helpers.py

@@ -0,0 +1,366 @@
+"""
+Вспомогательные функции для телеграм-бота.
+Содержит общие утилиты и хелперы для повседневных задач.
+"""
+
+import asyncio
+import functools
+import logging
+from typing import Any, Callable, Dict, List, Optional, Union
+from core.exceptions import BotException
+
+
+def safe_execute(func: Callable, *args, **kwargs) -> Any:
+    """
+    Безопасное выполнение функции с обработкой исключений.
+
+    Args:
+        func: Функция для выполнения
+        *args: Позиционные аргументы
+        **kwargs: Именованные аргументы
+
+    Returns:
+        Результат выполнения функции или None при ошибке
+    """
+    try:
+        return func(*args, **kwargs)
+    except Exception as e:
+        logging.error(f"Ошибка при выполнении функции {func.__name__}: {e}")
+        return None
+
+
+def chunk_text(text: str, chunk_size: int = 4000) -> List[str]:
+    """
+    Разбиение текста на части указанного размера.
+
+    Args:
+        text: Исходный текст
+        chunk_size: Максимальный размер части
+
+    Returns:
+        Список частей текста
+    """
+    if not text:
+        return []
+
+    chunks = []
+    for i in range(0, len(text), chunk_size):
+        chunks.append(text[i:i + chunk_size])
+
+    return chunks
+
+
+def escape_markdown(text: str) -> str:
+    """
+    Экранирование специальных символов Markdown.
+
+    Args:
+        text: Текст для экранирования
+
+    Returns:
+        Экранированный текст
+    """
+    if not text:
+        return ""
+
+    # Символы, которые нужно экранировать в Markdown
+    escape_chars = r'_*`['
+
+    result = text
+    for char in escape_chars:
+        result = result.replace(char, f'\\{char}')
+
+    return result
+
+
+def create_chunks(items: List[Any], chunk_size: int) -> List[List[Any]]:
+    """
+    Разбиение списка на части указанного размера.
+
+    Args:
+        items: Исходный список
+        chunk_size: Размер части
+
+    Returns:
+        Список частей
+    """
+    return [items[i:i + chunk_size] for i in range(0, len(items), chunk_size)]
+
+
+def format_number(num: Union[int, float]) -> str:
+    """
+    Форматирование числа с разделителями тысяч.
+
+    Args:
+        num: Число для форматирования
+
+    Returns:
+        Отформатированная строка
+    """
+    try:
+        if isinstance(num, float):
+            # Для float возвращаем как есть
+            return str(num)
+        else:
+            # Ручное форматирование для кроссплатформенности
+            s = str(int(num))
+            if len(s) <= 3:
+                return s
+            groups = []
+            while s:
+                groups.append(s[-3:])
+                s = s[:-3]
+            return ','.join(reversed(groups))
+    except (ValueError, TypeError):
+        return str(num)
+
+
+def get_nested_value(data: Dict, path: str, default: Any = None) -> Any:
+    """
+    Получение значения из вложенного словаря по пути.
+
+    Args:
+        data: Исходный словарь
+        path: Путь к значению (через точку)
+        default: Значение по умолчанию
+
+    Returns:
+        Значение или default
+
+    Example:
+        data = {'user': {'profile': {'name': 'John'}}}
+        get_nested_value(data, 'user.profile.name')  # 'John'
+    """
+    keys = path.split('.')
+    current = data
+
+    for key in keys:
+        if isinstance(current, dict) and key in current:
+            current = current[key]
+        else:
+            return default
+
+    return current
+
+
+def set_nested_value(data: Dict, path: str, value: Any) -> Dict:
+    """
+    Установка значения во вложенный словарь по пути.
+
+    Args:
+        data: Исходный словарь
+        path: Путь к значению (через точку)
+        value: Значение для установки
+
+    Returns:
+        Обновленный словарь
+
+    Example:
+        data = {}
+        set_nested_value(data, 'user.profile.name', 'John')
+        # data = {'user': {'profile': {'name': 'John'}}}
+    """
+    keys = path.split('.')
+    current = data
+
+    for key in keys[:-1]:
+        if key not in current:
+            current[key] = {}
+        current = current[key]
+
+    current[keys[-1]] = value
+    return data
+
+
+def retry_async(max_attempts: int = 3, delay: float = 1.0):
+    """
+    Декоратор для повторных попыток асинхронных функций.
+
+    Args:
+        max_attempts: Максимальное количество попыток
+        delay: Задержка между попытками в секундах
+    """
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            last_exception = None
+
+            for attempt in range(max_attempts):
+                try:
+                    return await func(*args, **kwargs)
+                except Exception as e:
+                    last_exception = e
+                    if attempt < max_attempts - 1:
+                        await asyncio.sleep(delay)
+
+            raise last_exception
+
+        return wrapper
+    return decorator
+
+
+def validate_required_fields(data: Dict, required_fields: List[str]) -> bool:
+    """
+    Проверка наличия обязательных полей в данных.
+
+    Args:
+        data: Словарь с данными
+        required_fields: Список обязательных полей
+
+    Returns:
+        True если все поля присутствуют
+
+    Raises:
+        ValidationError: Если отсутствуют обязательные поля
+    """
+    missing_fields = [field for field in required_fields if field not in data]
+
+    if missing_fields:
+        raise ValidationError(f"Отсутствуют обязательные поля: {', '.join(missing_fields)}")
+
+    return True
+
+
+def calculate_percentage(part: Union[int, float], total: Union[int, float]) -> float:
+    """
+    Вычисление процента.
+
+    Args:
+        part: Часть от целого
+        total: Целое
+
+    Returns:
+        Процент (0-100)
+    """
+    try:
+        if total == 0:
+            return 0.0
+        return (float(part) / float(total)) * 100
+    except (ValueError, TypeError):
+        return 0.0
+
+
+def format_duration(seconds: int) -> str:
+    """
+    Форматирование длительности в читаемый вид.
+
+    Args:
+        seconds: Длительность в секундах
+
+    Returns:
+        Отформатированная строка
+    """
+    if seconds < 60:
+        return f"{seconds} сек"
+    elif seconds < 3600:
+        minutes = seconds // 60
+        remaining_seconds = seconds % 60
+        return f"{minutes} мин {remaining_seconds} сек"
+    else:
+        hours = seconds // 3600
+        minutes = (seconds % 3600) // 60
+        return f"{hours} ч {minutes} мин"
+
+
+def is_empty(value: Any) -> bool:
+    """
+    Проверка значения на пустоту.
+
+    Args:
+        value: Значение для проверки
+
+    Returns:
+        True если значение пустое
+    """
+    if value is None:
+        return True
+
+    if isinstance(value, (str, list, dict, tuple)):
+        return len(value) == 0
+
+    return False
+
+
+def generate_user_mention(user_id: int, name: str) -> str:
+    """
+    Генерация упоминания пользователя.
+
+    Args:
+        user_id: ID пользователя
+        name: Имя пользователя
+
+    Returns:
+        Строка упоминания в формате Markdown
+    """
+    return f"[{name}](tg://user?id={user_id})"
+
+
+def clean_string(text: str) -> str:
+    """
+    Очистка строки от лишних пробелов и символов.
+
+    Args:
+        text: Исходная строка
+
+    Returns:
+        Очищенная строка
+    """
+    if not text:
+        return ""
+
+    # Удаляем лишние пробелы и переносы строк
+    cleaned = ' '.join(text.split())
+    return cleaned.strip()
+
+
+def merge_dicts(*dicts: Dict) -> Dict:
+    """
+    Объединение словарей с приоритетом последних.
+
+    Args:
+        *dicts: Словари для объединения
+
+    Returns:
+        Объединенный словарь
+    """
+    result = {}
+
+    for d in dicts:
+        if d:
+            result.update(d)
+
+    return result
+
+
+def filter_dict(data: Dict, keys: List[str]) -> Dict:
+    """
+    Фильтрация словаря по ключам.
+
+    Args:
+        data: Исходный словарь
+        keys: Ключи для сохранения
+
+    Returns:
+        Отфильтрованный словарь
+    """
+    return {k: v for k, v in data.items() if k in keys}
+
+
+def async_context_manager(func: Callable) -> Callable:
+    """
+    Декоратор для создания асинхронного контекстного менеджера.
+
+    Args:
+        func: Функция для декорирования
+
+    Returns:
+        Декорированная функция
+    """
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        if asyncio.iscoroutinefunction(func):
+            return await func(*args, **kwargs)
+        else:
+            return func(*args, **kwargs)
+
+    return wrapper