new_lottery_bot/docs/USER_MANAGEMENT_GUIDE.md

Система управления пользователями

Обзор

Система управления пользователями предоставляет администраторам инструменты для:

• Поиска пользователей по различным критериям
• Просмотра детальной информации о пользователях
• Блокировки/разблокировки пользователей в чате
• Управления большими базами пользователей (1000+)

Функциональность

1. Блокировка в чате

Назначение: Запрет отправки сообщений в чат бота для конкретного пользователя.

Как работает:

• Заблокированный пользователь не может отправлять сообщения в P2P чат
• Блокировка не затрагивает другие функции бота (участие в розыгрышах, просмотр статистики)
• Блокировка фиксируется в поле is_chat_banned в базе данных

Применение:

1. Найти пользователя через поиск или список
2. Открыть карточку пользователя
3. Нажать кнопку "🚫 Заблокировать в чате"

2. Поиск пользователей

Критерии поиска:

• Username (с @ или без)
• Имя или фамилия
• Telegram ID
• Номер клубной карты
• Никнейм

Пример запросов:

• @username - поиск по username
• Иван - поиск по имени/фамилии
• 123456789 - поиск по Telegram ID или номеру карты
• nickname - поиск по никнейму

Особенности:

• Поиск работает по принципу "ИЛИ" (OR) - ищет во всех полях одновременно
• Поддержка частичного совпадения (ILIKE)
• Автоматическая пагинация (15 пользователей на странице)

3. Списки пользователей

Все пользователи

• Отображаются все зарегистрированные в боте
• Сортировка по дате создания (новые первые)
• Пагинация по 15 пользователей

Заблокированные пользователи

• Отображаются только пользователи с блокировкой в чате
• Быстрый доступ для управления блокировками

4. Карточка пользователя

Отображаемая информация:

• Основное:

  • Имя и фамилия
  • Username (если есть)
  • Telegram ID

• Статистика:

  • Дата регистрации
  • Последняя активность
  • Статус регистрации
  • Статус администратора

• Дополнительно:

  • Никнейм (если установлен)
  • Номер клубной карты (если привязан)
  • Телефон (если указан)
  • Статус блокировки в чате

Архитектура

Компоненты системы

1. UserManagementService

Файл: src/core/user_management.py

Методы:

# Поиск с фильтрами и пагинацией
async def search_users(
    session: AsyncSession,
    query: str = None,      # Поисковый запрос
    page: int = 1,          # Номер страницы
    per_page: int = 15,     # Размер страницы
    filters: Dict = None    # Фильтры (is_registered, is_admin, is_chat_banned)
) -> Tuple[List[User], int]:
    """Возвращает (список пользователей, общее количество)"""

# Получение по ID
async def get_user_by_id(session: AsyncSession, user_id: int) -> Optional[User]:
    """Получить пользователя по внутреннему ID"""

# Блокировка в чате
async def ban_user_in_chat(session: AsyncSession, user_id: int) -> bool:
    """Заблокировать пользователя в чате"""

# Разблокировка в чате
async def unban_user_in_chat(session: AsyncSession, user_id: int) -> bool:
    """Разблокировать пользователя в чате"""

# Статистика
async def get_user_stats(session: AsyncSession) -> Dict[str, int]:
    """Получить статистику: total, registered, admins, chat_banned"""

# Форматирование
def format_user_info(user: User, detailed: bool = False) -> str:
    """Красивый HTML вывод информации о пользователе"""

Константы:

• USERS_PER_PAGE = 15 - количество пользователей на странице

2. Обработчики админ-панели

Файл: src/handlers/admin_panel.py

Основные обработчики:

# Главное меню управления пользователями
@admin_router.callback_query(F.data == "admin_users")
async def admin_users_menu(callback: CallbackQuery):
    """Показывает статистику и главное меню"""

# Запрос поискового запроса
@admin_router.callback_query(F.data == "admin_users_search")
async def admin_users_search_prompt(callback: CallbackQuery, state: FSMContext):
    """Переводит в режим ожидания поискового запроса"""

# Обработка поиска
@admin_router.message(AdminStates.user_management_search)
async def admin_users_search_process(message: Message, state: FSMContext):
    """Выполняет поиск и показывает результаты"""

# Список всех пользователей
@admin_router.callback_query(F.data.startswith("admin_users_list:"))
async def admin_users_list(callback: CallbackQuery):
    """Постраничный список всех пользователей"""

# Список заблокированных
@admin_router.callback_query(F.data.startswith("admin_users_banned:"))
async def admin_users_banned_list(callback: CallbackQuery):
    """Список только заблокированных пользователей"""

# Просмотр пользователя
@admin_router.callback_query(F.data.startswith("admin_user_view:"))
async def admin_user_view(callback: CallbackQuery):
    """Детальная карточка пользователя"""

# Блокировка
@admin_router.callback_query(F.data.startswith("admin_user_ban:"))
async def admin_user_ban(callback: CallbackQuery):
    """Заблокировать пользователя"""

# Разблокировка
@admin_router.callback_query(F.data.startswith("admin_user_unban:"))
async def admin_user_unban(callback: CallbackQuery):
    """Разблокировать пользователя"""

3. FSM состояния

Файл: src/handlers/admin_panel.py

class AdminStates(StatesGroup):
    # ... другие состояния ...
    user_management_search = State()  # Ожидание поискового запроса
    user_management_view = State()    # Просмотр пользователя

Модель данных

Поле is_chat_banned

Файл: src/core/models.py

class User(Base):
    # ... другие поля ...
    is_chat_banned: Mapped[bool] = mapped_column(Boolean, default=False)
    last_activity: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)

Миграция: migrations/versions/20260215_0403_00_b4c435a7dc5f_add_is_chat_banned_to_users.py

Интеграция с чатом

Файл: src/core/chat_services.py

class ChatPermissionService:
    @staticmethod
    async def can_send_message(session: AsyncSession, telegram_id: int) -> Tuple[bool, str]:
        """
        Проверяет, может ли пользователь отправлять сообщения в чат
        
        Порядок проверок:
        1. Глобальная блокировка (BanService)
        2. Блокировка в чате (is_chat_banned)
        3. Регистрация в боте
        
        Returns:
            (может_отправить: bool, причина_если_нет: str)
        """

Проверка в обработчиках P2P сообщений:

can_send, reason = await ChatPermissionService.can_send_message(session, message.from_user.id)
if not can_send:
    await message.answer(f"❌ {reason}")
    return

Производительность

Оптимизация для больших баз (1000+ пользователей)

1. Пагинация

• Все списки ограничены 15 пользователями на странице
• SQL LIMIT/OFFSET для эффективных запросов
• Подсчет общего количества отдельным запросом

2. Индексы (рекомендуется добавить)

-- Для поиска по username
CREATE INDEX idx_users_username ON users(username);

-- Для поиска по имени/фамилии
CREATE INDEX idx_users_names ON users(first_name, last_name);

-- Для поиска по клубной карте
CREATE INDEX idx_users_card ON users(club_card_number);

-- Для фильтрации заблокированных
CREATE INDEX idx_users_chat_banned ON users(is_chat_banned) WHERE is_chat_banned = true;

3. ILIKE оптимизация

Для больших баз рекомендуется добавить индекс с расширением pg_trgm:

CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX idx_users_username_trgm ON users USING gin(username gin_trgm_ops);
CREATE INDEX idx_users_names_trgm ON users USING gin((first_name || ' ' || last_name) gin_trgm_ops);

4. Кеширование (будущее улучшение)

• Кеш для статистики (Redis)
• Кеш для часто просматриваемых пользователей
• TTL: 5 минут

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

Для администраторов

Шаг 1: Доступ к управлению

1. Открыть админ-панель: /admin
2. Выбрать "👤 Управление пользователями"

Шаг 2: Поиск пользователя

Вариант A: Быстрый поиск

1. Нажать "🔍 Поиск пользователей"
2. Ввести запрос (username, имя, ID)
3. Выбрать пользователя из результатов

Вариант B: Просмотр списка

1. Нажать "📋 Все пользователи"
2. Перемещаться по страницам (⬅️ Назад / ➡️ Далее)
3. Выбрать пользователя

Вариант C: Только заблокированные

1. Нажать "🚫 Заблокированные"
2. Просмотреть список заблокированных
3. Выбрать для разблокировки

Шаг 3: Управление блокировкой

1. В карточке пользователя нажать:
   • "🚫 Заблокировать в чате" - для блокировки
   • "✅ Разблокировать в чате" - для разблокировки
2. Подтвердить действие

Для разработчиков

Добавление новых фильтров

# В UserManagementService.search_users()
if filters:
    if 'custom_field' in filters:
        conditions.append(User.custom_field == filters['custom_field'])

Добавление новых полей в карточку

# В UserManagementService.format_user_info()
if detailed:
    text += f"🆕 Кастомное поле: {user.custom_field}\n"

Создание нового списка

@admin_router.callback_query(F.data.startswith("admin_users_custom:"))
async def admin_users_custom_list(callback: CallbackQuery):
    page = int(callback.data.split(":")[1])
    
    async with async_session_maker() as session:
        users, total = await UserManagementService.search_users(
            session,
            page=page,
            filters={'custom_field': True}
        )
    
    # ... рендер списка ...

Мониторинг

Логирование

logger.info(f"Пользователь {user.telegram_id} заблокирован в чате")
logger.error(f"Ошибка блокировки пользователя {user_id}: {e}")

Метрики для мониторинга

• Количество поисковых запросов
• Среднее время ответа на поиск
• Количество блокировок/разблокировок
• Топ-10 самых активных поисковых запросов

Безопасность

Проверка прав доступа

Все обработчики проверяют права администратора:

if not is_admin(callback.from_user.id):
    await callback.answer("❌ Доступ запрещен", show_alert=True)
    return

Защита от SQL-инъекций

• Все запросы используют параметризованные запросы SQLAlchemy
• Нет прямой конкатенации SQL

Аудит действий (рекомендуется добавить)

CREATE TABLE admin_actions (
    id SERIAL PRIMARY KEY,
    admin_telegram_id BIGINT NOT NULL,
    action VARCHAR(50) NOT NULL,  -- 'ban_user', 'unban_user', 'search_user'
    target_user_id INTEGER,
    details JSONB,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

Расширения

Планируемые улучшения

1. Массовые операции

   • Блокировка нескольких пользователей одновременно
   • Экспорт списка в CSV/JSON

2. Расширенные фильтры

   • Диапазон дат регистрации
   • Неактивные пользователи (по last_activity)
   • Пользователи без клубной карты

3. История блокировок

   • Журнал всех блокировок/разблокировок
   • Кто и когда заблокировал
   • Причина блокировки (опциональное поле)

4. Автоматическая блокировка

   • При определенном количестве жалоб от других пользователей
   • При обнаружении спам-активности

5. Уведомления

   • Уведомление пользователя о блокировке (опционально)
   • Уведомление администраторов о подозрительной активности

Troubleshooting

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

Решение:

• Убедиться, что пользователь существует в базе
• Проверить правильность написания (регистр не важен)
• Попробовать поиск по Telegram ID

Проблема: Блокировка не применяется

Решение:

• Проверить логи: docker logs lottery_bot | grep "блокирован"
• Убедиться, что транзакция закоммитилась
• Проверить поле в БД: SELECT telegram_id, is_chat_banned FROM users WHERE id = ?

Проблема: Медленный поиск (>2 секунд)

Решение:

• Добавить индексы (см. раздел "Производительность")
• Проверить EXPLAIN ANALYZE для поискового запроса
• Рассмотреть использование полнотекстового поиска PostgreSQL

Тестирование

Ручное тестирование

1. Тест поиска:

   - Поиск по существующему username
   - Поиск по несуществующему username
   - Поиск с частичным совпадением
   - Поиск по Telegram ID
   

2. Тест блокировки:

   - Заблокировать пользователя
   - Попытаться отправить сообщение от заблокированного пользователя
   - Разблокировать пользователя
   - Убедиться, что сообщения проходят
   

3. Тест пагинации:

   - Создать >15 пользователей
   - Проверить навигацию вперед/назад
   - Проверить корректность подсчета страниц
   

Автоматическое тестирование

# tests/test_user_management.py
import pytest
from src.core.user_management import UserManagementService

@pytest.mark.asyncio
async def test_search_users_by_username(session):
    users, total = await UserManagementService.search_users(
        session, query="@testuser"
    )
    assert len(users) > 0
    assert users[0].username == "testuser"

@pytest.mark.asyncio
async def test_ban_user(session):
    success = await UserManagementService.ban_user_in_chat(session, 1)
    assert success == True
    
    user = await UserManagementService.get_user_by_id(session, 1)
    assert user.is_chat_banned == True

Заключение

Система управления пользователями предоставляет:

• ✅ Быстрый и удобный поиск среди больших баз данных
• ✅ Простое управление блокировками в чате
• ✅ Масштабируемость для 1000+ пользователей
• ✅ Интуитивный интерфейс для администраторов
• ✅ Интеграцию с системой разрешений чата

Система готова к использованию и может быть расширена дополнительными функциями по мере необходимости.