telegram-helper-bot/docs/IMPROVEMENTS.md

План улучшений проекта

Этот документ содержит список рекомендаций по улучшению кодовой базы проекта Telegram Helper Bot. Пункты отсортированы по приоритетам и могут быть использованы для планирования работ.

Статус задач

• ⬜ Не начато
• 🟡 В работе
• ✅ Выполнено
• ❌ Отложено

---

🔴 Высокий приоритет

1. Стандартизация Dependency Injection

Статус: ⬜

Проблема: В проекте используется смешанный подход к dependency injection:

• В некоторых местах используется MagicData("bot_db") и MagicData("settings")
• В других местах используется **kwargs и получение из data
• В сервисах напрямую вызывается get_global_instance()

Текущее состояние:

# callback_handlers.py - смешанный подход
async def handler(call: CallbackQuery, settings: MagicData("settings")):
    publish_service = get_post_publish_service()  # Прямой вызов фабрики

async def handler(call: CallbackQuery, **kwargs):
    ban_service = get_ban_service()  # Прямой вызов фабрики

Рекомендация: Стандартизировать на использование MagicData и Annotated везде:

from typing import Annotated
from aiogram.filters import MagicData
from helper_bot.handlers.admin.dependencies import BotDB, Settings

async def handler(
    call: CallbackQuery,
    bot_db: Annotated[AsyncBotDB, BotDB],
    settings: Annotated[dict, Settings],
    service: Annotated[PostPublishService, get_post_publish_service()]
):
    # Использовать зависимости напрямую
    ...

Файлы для изменения:

• helper_bot/handlers/callback/callback_handlers.py (строки 47, 80, 109, 131, 182)
• helper_bot/handlers/private/private_handlers.py
• Все сервисы, которые используют get_global_instance()

Оценка: Средняя сложность, требует рефакторинга нескольких файлов

---

2. Удаление import *

Статус: ⬜

Проблема: В voice_handler.py используется импорт всех констант через import *, что затрудняет понимание зависимостей и может привести к конфликтам имен.

Текущее состояние:

# helper_bot/handlers/voice/voice_handler.py
from helper_bot.handlers.voice.constants import *

Рекомендация: Заменить на явные импорты:

from helper_bot.handlers.voice.constants import (
    CONSTANT1,
    CONSTANT2,
    CONSTANT3,
    # ... все используемые константы
)

Файлы для изменения:

• helper_bot/handlers/voice/voice_handler.py (строка 17)

Оценка: Низкая сложность, быстрое исправление

---

3. Закрытие критичных TODO

Статус: ⬜

Проблема: В коде есть несколько TODO комментариев, указывающих на технический долг и места, требующие рефакторинга.

Список TODO:

3.1. Callback handlers - переход на MagicData

Файл: helper_bot/handlers/callback/callback_handlers.py

• Строка 47: # TODO: переделать на MagicData
• Строка 80: # TODO: переделать на MagicData
• Строка 109: # TODO: переделать на MagicData
• Строка 131: # TODO: переделать на MagicData
• Строка 182: # TODO: переделать на MagicData

Решение: Связано с задачей #1 (стандартизация DI)

3.2. Metrics middleware - подключение к БД

Файл: helper_bot/middlewares/metrics_middleware.py

• Строка 153: #TODO: Должна подключаться к базе данных, а не к глобальному экземпляру

Решение:

# Вместо
bdf = get_global_instance()
bot_db = bdf.get_db()

# Использовать dependency injection через MagicData
async def _update_active_users_metric(
    self,
    bot_db: Annotated[AsyncBotDB, BotDB]
):
    ...

3.3. Voice handler - вынос логики

Файл: helper_bot/handlers/voice/voice_handler.py

• Строка 354: #TODO: удалить логику из хендлера

Решение: Переместить бизнес-логику в VoiceBotService

3.4. Helper functions - архитектура

Файл: helper_bot/utils/helper_func.py

• Строка 35: #TODO: поменять архитектуру и подключить правильный BotDB
• Строка 145: #TODO: Уверен можно укоротить

Решение: Рефакторинг функций для использования dependency injection

3.5. Group handlers - архитектура

Файл: helper_bot/handlers/group/group_handlers.py

• Строка 109: #TODO: поменять архитектуру и подключить правильный BotDB

Решение: Использовать dependency injection вместо прямого доступа к БД

Оценка: Средняя-высокая сложность, требует анализа каждого случая

---

🟡 Средний приоритет

4. Оптимизация работы с БД - Connection Pooling

Статус: ⬜

Проблема: Каждый запрос к БД открывает новое соединение и закрывает его. При высокой нагрузке это неэффективно и может привести к проблемам с производительностью.

Текущее состояние:

# database/base.py
async def _get_connection(self):
    conn = await aiosqlite.connect(self.db_path)
    # Настройка PRAGMA каждый раз
    await conn.execute("PRAGMA foreign_keys = ON")
    await conn.execute("PRAGMA journal_mode = WAL")
    # ...
    return conn

async def _execute_query(self, query: str, params: tuple = ()):
    conn = None
    try:
        conn = await self._get_connection()  # Новое соединение каждый раз
        result = await conn.execute(query, params)
        await conn.commit()
        return result
    finally:
        if conn:
            await conn.close()  # Закрытие после каждого запроса

Рекомендация: Реализовать переиспользование соединений или connection pool:

Вариант 1: Переиспользование соединения в рамках транзакции

class DatabaseConnection:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._connection: Optional[aiosqlite.Connection] = None
    
    async def _get_connection(self):
        if self._connection is None:
            self._connection = await aiosqlite.connect(self.db_path)
            # Настройка PRAGMA один раз
            await self._connection.execute("PRAGMA foreign_keys = ON")
            # ...
        return self._connection
    
    async def close(self):
        if self._connection:
            await self._connection.close()
            self._connection = None

Вариант 2: Использование async context manager

async def _execute_query(self, query: str, params: tuple = ()):
    async with aiosqlite.connect(self.db_path) as conn:
        await conn.execute("PRAGMA foreign_keys = ON")
        result = await conn.execute(query, params)
        await conn.commit()
        return result

Файлы для изменения:

• database/base.py
• database/repository_factory.py (добавить метод close())
• helper_bot/utils/base_dependency_factory.py (закрытие соединений при shutdown)

Оценка: Средняя сложность, требует тестирования на производительность

---

5. Улучшение обработки ошибок - декораторы

Статус: ⬜

Проблема: В callback_handlers.py повторяется один и тот же блок обработки ошибок в каждом handler:

try:
    # Бизнес-логика
except UserBlockedBotError:
    await call.answer(text=MESSAGE_ERROR, show_alert=True, cache_time=3)
except (PostNotFoundError, PublishError) as e:
    logger.error(f'Ошибка: {str(e)}')
    await call.answer(text=MESSAGE_ERROR, show_alert=True, cache_time=3)
except Exception as e:
    if str(e) == ERROR_BOT_BLOCKED:
        await call.answer(text=MESSAGE_ERROR, show_alert=True, cache_time=3)
    else:
        important_logs = settings['Telegram']['important_logs']
        await call.bot.send_message(
            chat_id=important_logs,
            text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
        )
        logger.error(f'Неожиданная ошибка: {str(e)}')
        await call.answer(text=MESSAGE_ERROR, show_alert=True, cache_time=3)

Рекомендация: Создать декоратор для централизованной обработки ошибок:

# helper_bot/handlers/callback/decorators.py
from functools import wraps
from typing import Callable, Any
from aiogram.types import CallbackQuery
from logs.custom_logger import logger
import traceback

def handle_callback_errors(func: Callable[..., Any]) -> Callable[..., Any]:
    """Декоратор для обработки ошибок в callback handlers."""
    @wraps(func)
    async def wrapper(call: CallbackQuery, *args, **kwargs):
        try:
            return await func(call, *args, **kwargs)
        except UserBlockedBotError:
            await call.answer(
                text=MESSAGE_ERROR, 
                show_alert=True, 
                cache_time=3
            )
        except (PostNotFoundError, PublishError) as e:
            logger.error(f'Ошибка в {func.__name__}: {str(e)}')
            await call.answer(
                text=MESSAGE_ERROR, 
                show_alert=True, 
                cache_time=3
            )
        except Exception as e:
            if str(e) == ERROR_BOT_BLOCKED:
                await call.answer(
                    text=MESSAGE_ERROR, 
                    show_alert=True, 
                    cache_time=3
                )
            else:
                # Получить settings из kwargs или через dependency injection
                settings = kwargs.get('settings')
                if settings:
                    important_logs = settings['Telegram']['important_logs']
                    await call.bot.send_message(
                        chat_id=important_logs,
                        text=f"Произошла ошибка в {func.__name__}: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
                    )
                logger.error(f'Неожиданная ошибка в {func.__name__}: {str(e)}')
                await call.answer(
                    text=MESSAGE_ERROR, 
                    show_alert=True, 
                    cache_time=3
                )
    return wrapper

Использование:

@callback_router.callback_query(F.data == CALLBACK_APPROVE)
@handle_callback_errors
@track_time("post_for_group", "callback_handlers")
@track_errors("callback_handlers", "post_for_group")
async def post_for_group(call: CallbackQuery, ...):
    # Только бизнес-логика, без try-except
    publish_service = get_post_publish_service()
    await publish_service.publish_post(call)
    await call.answer(text=MESSAGE_PUBLISHED, cache_time=3)

Файлы для изменения:

• Создать helper_bot/handlers/callback/decorators.py
• Рефакторинг helper_bot/handlers/callback/callback_handlers.py

Оценка: Средняя сложность, требует тестирования всех сценариев

---

6. Валидация настроек при старте

Статус: ⬜

Проблема: Настройки загружаются из .env без валидации. Отсутствие обязательных настроек обнаруживается только во время выполнения, что затрудняет отладку.

Текущее состояние:

# helper_bot/utils/base_dependency_factory.py
def _load_settings_from_env(self):
    self.settings['Telegram'] = {
        'bot_token': os.getenv('BOT_TOKEN', ''),  # Может быть пустой строкой
        # ...
    }

Рекомендация: Добавить валидацию обязательных настроек:

class BaseDependencyFactory:
    REQUIRED_SETTINGS = {
        'Telegram': ['bot_token'],
        'S3': ['endpoint_url', 'access_key', 'secret_key', 'bucket_name']  # Если S3 включен
    }
    
    def _validate_settings(self):
        """Валидирует обязательные настройки."""
        errors = []
        
        # Проверка Telegram настроек
        for key in self.REQUIRED_SETTINGS['Telegram']:
            value = self.settings['Telegram'].get(key)
            if not value:
                errors.append(f"Telegram.{key} is required but not set")
        
        # Проверка S3 настроек (если включен)
        if self.settings['S3']['enabled']:
            for key in self.REQUIRED_SETTINGS['S3']:
                value = self.settings['S3'].get(key)
                if not value:
                    errors.append(f"S3.{key} is required when S3 is enabled but not set")
        
        if errors:
            error_msg = "Configuration errors:\n" + "\n".join(f"  - {e}" for e in errors)
            raise ValueError(error_msg)
    
    def __init__(self):
        # ... существующий код ...
        self._load_settings_from_env()
        self._validate_settings()  # Добавить валидацию
        self._init_s3_storage()

Файлы для изменения:

• helper_bot/utils/base_dependency_factory.py

Оценка: Низкая сложность, быстрое добавление

---

7. Исправление RepositoryFactory

Статус: ⬜

Проблема: Методы check_database_integrity() и cleanup_wal_files() в RepositoryFactory вызываются только для репозитория users, хотя должны применяться ко всем репозиториям или к базе данных в целом.

Текущее состояние:

# database/repository_factory.py
async def check_database_integrity(self):
    """Проверяет целостность базы данных."""
    await self.users.check_database_integrity()  # Только users?

async def cleanup_wal_files(self):
    """Очищает WAL файлы."""
    await self.users.cleanup_wal_files()  # Только users?

Рекомендация: Проверка целостности и очистка WAL должны выполняться один раз для всей БД, а не для каждого репозитория:

async def check_database_integrity(self):
    """Проверяет целостность базы данных."""
    # Использовать любой репозиторий для доступа к БД
    await self.users.check_database_integrity()

async def cleanup_wal_files(self):
    """Очищает WAL файлы."""
    # Использовать любой репозиторий для доступа к БД
    await self.users.cleanup_wal_files()

Или лучше - вынести эти методы в DatabaseConnection и вызывать через любой репозиторий (текущая реализация уже правильная, но можно улучшить документацию).

Альтернатива: Создать отдельный класс DatabaseManager для операций на уровне БД.

Файлы для изменения:

• database/repository_factory.py (улучшить документацию)
• Возможно создать database/database_manager.py

Оценка: Низкая сложность, в основном документация

---

🟢 Низкий приоритет

8. Добавление кэширования (Redis)

Статус: ⬜

Проблема: Часто запрашиваемые данные (например, список администраторов, настройки пользователей) загружаются из БД при каждом запросе, что создает лишнюю нагрузку на базу данных.

Рекомендация: Добавить Redis для кэширования часто используемых данных:

# helper_bot/utils/cache.py
import redis.asyncio as redis
from typing import Optional, Any
import json
from helper_bot.utils.base_dependency_factory import get_global_instance

class CacheService:
    def __init__(self):
        bdf = get_global_instance()
        settings = bdf.get_settings()
        self.redis_client = None
        
        if settings.get('Redis', {}).get('enabled', False):
            self.redis_client = redis.from_url(
                settings['Redis']['url'],
                decode_responses=True
            )
    
    async def get(self, key: str) -> Optional[Any]:
        """Получить значение из кэша."""
        if not self.redis_client:
            return None
        
        try:
            value = await self.redis_client.get(key)
            if value:
                return json.loads(value)
        except Exception as e:
            logger.error(f"Ошибка получения из кэша: {e}")
        return None
    
    async def set(self, key: str, value: Any, ttl: int = 3600):
        """Установить значение в кэш."""
        if not self.redis_client:
            return
        
        try:
            await self.redis_client.setex(
                key,
                ttl,
                json.dumps(value)
            )
        except Exception as e:
            logger.error(f"Ошибка записи в кэш: {e}")
    
    async def delete(self, key: str):
        """Удалить значение из кэша."""
        if not self.redis_client:
            return
        
        try:
            await self.redis_client.delete(key)
        except Exception as e:
            logger.error(f"Ошибка удаления из кэша: {e}")

Использование:

# В репозиториях или сервисах
cache = CacheService()

# Получение с кэшированием
async def get_admin_list(self):
    cache_key = "admin_list"
    cached = await cache.get(cache_key)
    if cached:
        return cached
    
    # Загрузка из БД
    admins = await self._load_from_db()
    
    # Сохранение в кэш на 1 час
    await cache.set(cache_key, admins, ttl=3600)
    return admins

Данные для кэширования:

• Список администраторов
• Настройки пользователей (если редко меняются)
• Статистика (активные пользователи за день)
• Черный список (с коротким TTL)

Файлы для изменения:

• Создать helper_bot/utils/cache.py
• Добавить настройки Redis в BaseDependencyFactory
• Обновить репозитории для использования кэша

Оценка: Средняя сложность, требует настройки Redis инфраструктуры

---

9. Улучшение Type Hints

Статус: ⬜

Проблема: Некоторые методы возвращают dict без указания структуры, что затрудняет понимание API и использование IDE.

Пример:

def get_settings(self):
    return self.settings  # Какой тип? Dict[str, Any]?

Рекомендация: Использовать TypedDict для структурированных словарей:

from typing import TypedDict, Dict, Any

class TelegramSettings(TypedDict):
    bot_token: str
    listen_bot_token: str
    preview_link: bool
    main_public: str
    group_for_posts: int
    # ...

class SettingsDict(TypedDict):
    Telegram: TelegramSettings
    Settings: Dict[str, bool]
    Metrics: Dict[str, Any]
    S3: Dict[str, Any]

class BaseDependencyFactory:
    def get_settings(self) -> SettingsDict:
        return self.settings

Файлы для изменения:

• helper_bot/utils/base_dependency_factory.py
• Создать helper_bot/utils/types.py для типов

Оценка: Средняя сложность, требует обновления всех мест использования

---

10. Расширение тестового покрытия

Статус: ⬜

Проблема: Некоторые компоненты не покрыты тестами или имеют недостаточное покрытие.

Рекомендация: Добавить тесты для:

1. Middleware:

   • DependenciesMiddleware - проверка внедрения зависимостей
   • BlacklistMiddleware - проверка блокировки пользователей
   • RateLimitMiddleware - проверка ограничений

2. BaseDependencyFactory:

   • Инициализация с валидными настройками
   • Инициализация с невалидными настройками
   • Получение зависимостей

3. Интеграционные тесты:

   • Полные сценарии обработки сообщений
   • Сценарии с ошибками
   • Сценарии с rate limiting

Файлы для создания:

• tests/test_dependencies_middleware.py
• tests/test_base_dependency_factory.py
• tests/test_integration_handlers.py

Оценка: Высокая сложность, требует времени на написание тестов

---

11. Улучшение логирования

Статус: ⬜

Проблема: В коде много logger.info() там, где можно использовать logger.debug() для детальной отладки. Это приводит к засорению логов в production.

Рекомендация: Пересмотреть уровни логирования:

• logger.debug() - детальная отладочная информация (шаги выполнения, промежуточные значения)
• logger.info() - важные события (старт/остановка бота, критические действия пользователей)
• logger.warning() - предупреждения (нестандартные ситуации, которые не критичны)
• logger.error() - ошибки (исключения, сбои)

Примеры для изменения:

# Было
logger.info(f"DependenciesMiddleware: внедрены зависимости для {type(event).__name__}")

# Стало
logger.debug(f"DependenciesMiddleware: внедрены зависимости для {type(event).__name__}")

Файлы для изменения:

• Все файлы с избыточным logger.info()

Оценка: Низкая сложность, но требует времени на ревью всех логов

---

12. Документация проекта

Статус: ⬜

Проблема: Отсутствует общая документация проекта, что затрудняет onboarding новых разработчиков.

Рекомендация: Создать следующие документы:

1. README.md (в корне проекта):

   • Описание проекта
   • Требования
   • Установка и настройка
   • Запуск
   • Структура проекта

2. docs/ARCHITECTURE.md:

   • Детальное описание архитектуры
   • Диаграммы компонентов
   • Паттерны проектирования

3. docs/DEPLOYMENT.md:

   • Инструкции по развертыванию
   • Настройка окружения
   • Мониторинг

4. docs/DEVELOPMENT.md:

   • Руководство для разработчиков
   • Процесс разработки
   • Code style guide (ссылка на .cursor/rules)

Оценка: Средняя сложность, требует времени на написание

---

📊 Статистика

• Всего задач: 12
• Высокий приоритет: 3
• Средний приоритет: 4
• Низкий приоритет: 5

📝 Заметки

• Большинство задач высокого приоритета связаны между собой (стандартизация DI решит несколько TODO)
• Задачи среднего приоритета улучшают производительность и качество кода
• Задачи низкого приоритета улучшают developer experience и поддерживаемость

🔄 Обновления

• 2026-01-25: Создан первоначальный список улучшений на основе анализа кодовой базы
• 2026-01-25: Добавлена задача #8 по кэшированию (Redis)
• 2026-01-25: Создан документ PYTHON_VERSION_MANAGEMENT.md с рекомендациями по унификации версий Python