storage.py - master - TurGG/AI-Assistant

storage.py
887 строк38 KB
den
Админ-логика: мероприятия, баллы, база знаний + Django + Docker
23 мар 2026, 16:32
23 мар 2026, 16:32617abaa
# ===========================================================
# storage.py — Слой доступа к данным (Data Access Layer)
# ===========================================================
# Рефакторинг с соблюдением ООП: классы вместо функций
# ===========================================================

import json
import logging
import os
import tempfile
import shutil
from pathlib import Path
from typing import Optional, List, Dict, Any, Tuple
from abc import ABC, abstractmethod

from config import DATABASE_FILE, CACHE_TTL_SECONDS, LOG_LEVEL, LOG_FORMAT
from cache import Cache
from models import User, TeamStats, AchievementInfo, Event

# ===========================================================
# ЛОГИРОВАНИЕ
# ===========================================================
logging.basicConfig(
    level=getattr(logging, LOG_LEVEL),
    format=LOG_FORMAT,
    handlers=[
        logging.FileHandler("okd.log", encoding="utf-8"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)


# ===========================================================
# АБСТРАКЦИЯ ХРАНИЛИЩА
# ===========================================================
class Storage(ABC):
    """
    Абстрактный базовый класс хранилища.
    
    Определяет интерфейс для всех реализаций хранилищ.
    При миграции на БД создаётся новый класс, наследующий этот.
    """

    @abstractmethod
    def get_all_users(self) -> List[User]:
        """Получить всех пользователей."""
        pass

    @abstractmethod
    def get_user_by_id(self, user_id: int) -> Optional[User]:
        """Получить пользователя по ID."""
        pass

    @abstractmethod
    def save_user(self, user: User) -> bool:
        """Сохранить пользователя."""
        pass

    @abstractmethod
    def get_users_sorted_by_points(self, limit: Optional[int] = None) -> List[User]:
        """Получить пользователей, отсортированных по баллам."""
        pass

    @abstractmethod
    def get_teams_with_scores(self) -> List[Tuple[str, int]]:
        """Получить команды с суммой баллов."""
        pass

    @abstractmethod
    def get_team_members(self, team_name: str) -> List[User]:
        """Получить участников команды."""
        pass

    @abstractmethod
    def get_all_achievements(self) -> List[AchievementInfo]:
        """Получить все достижения."""
        pass

    @abstractmethod
    def get_statistics(self) -> Dict[str, Any]:
        """Получить общую статистику."""
        pass

    @abstractmethod
    def get_user_rank(self, user_id: int) -> Optional[int]:
        """Получить место пользователя в рейтинге."""
        pass

    @abstractmethod
    def get_team_stats(self, team_name: str) -> Dict[str, Any]:
        """Получить детальную статистику команды."""
        pass

    @abstractmethod
    def mark_attendance(self, user_id: int, event_name: str, attended: bool) -> Tuple[bool, str]:
        """Отметить посещаемость мероприятия."""
        pass

    @abstractmethod
    def update_project_stage(self, user_id: int, project_name: str, stage: str) -> Tuple[bool, str]:
        """Обновить стадию проекта."""
        pass

    @abstractmethod
    def get_worst_test_takers(self, limit: int) -> List[Dict[str, Any]]:
        """Получить пользователей с наименьшими баллами за тесты."""
        pass

    @abstractmethod
    def get_non_attendees(self) -> List[Dict[str, Any]]:
        """Получить пользователей, не посещавших мероприятия."""
        pass

    @abstractmethod
    def add_pending_achievement(self, user_id: int, achievement_name: str, description: str) -> Tuple[bool, str]:
        """Добавить заявку на достижение."""
        pass

    @abstractmethod
    def get_pending_achievements(self) -> List[Dict[str, Any]]:
        """Получить заявки на достижения."""
        pass

    @abstractmethod
    def approve_pending_achievement(self, user_id: int, achievement_name: str) -> Tuple[bool, str]:
        """Одобрить достижение."""
        pass

    @abstractmethod
    def reject_pending_achievement(self, user_id: int, achievement_name: str, reason: str) -> Tuple[bool, str]:
        """Отклонить достижение."""
        pass

    @abstractmethod
    def get_team_leads(self, team_name: Optional[str] = None) -> List[Dict[str, Any]]:
        """Получить тимлидов."""
        pass

    @abstractmethod
    def get_users_by_stage(self, stage: str) -> List[Dict[str, Any]]:
        """Получить пользователей по стадии проекта."""
        pass

    # ==================== АДМИН-МЕТОДЫ: МЕРОПРИЯТИЯ ====================

    @abstractmethod
    def get_all_events(self) -> List[Dict[str, Any]]:
        """Получить все глобальные мероприятия."""
        pass

    @abstractmethod
    def create_event(self, name: str, date: str, description: str = "") -> Tuple[bool, str]:
        """Создать новое мероприятие."""
        pass

    @abstractmethod
    def update_event(self, event_name: str, new_name: str = None, new_date: str = None, new_description: str = None) -> Tuple[bool, str]:
        """Обновить мероприятие."""
        pass

    @abstractmethod
    def delete_event(self, event_name: str) -> Tuple[bool, str]:
        """Удалить мероприятие."""
        pass

    # ==================== АДМИН-МЕТОДЫ: БАЛЛЫ ====================

    @abstractmethod
    def add_points_to_user(self, user_id: int, points: int, reason: str = "") -> Tuple[bool, str]:
        """Начислить баллы пользователю."""
        pass

    @abstractmethod
    def deduct_points_from_user(self, user_id: int, points: int, reason: str = "") -> Tuple[bool, str]:
        """Списать баллы у пользователя."""
        pass

    # ==================== БАЗА ЗНАНИЙ ====================

    @abstractmethod
    def add_knowledge_item(self, title: str, content: str, category: str = "") -> Tuple[bool, str]:
        """Добавить материал в базу знаний."""
        pass

    @abstractmethod
    def get_all_knowledge_items(self, category: str = None) -> List[Dict[str, Any]]:
        """Получить все материалы из базы знаний."""
        pass

    @abstractmethod
    def delete_knowledge_item(self, item_id: str) -> Tuple[bool, str]:
        """Удалить материал из базы знаний."""
        pass


# ===========================================================
# JSON-РЕАЛИЗАЦИЯ ХРАНИЛИЩА
# ===========================================================
class JsonStorage(Storage):
    """
    JSON-реализация хранилища данных.
    
    Использует кэширование для чтения и атомарную запись.
    """

    VALID_STAGES = [
        "анализ и синтез",
        "генерация идей",
        "прототипирование",
        "тестирование",
        "сторителлинг",
    ]

    def __init__(self, db_path: Path, cache_ttl: int = 30):
        self._db_path = db_path
        self._cache = Cache(cache_ttl)
        # Путь к базе знаний
        from config import BASE_DIR
        self._knowledge_base_path = BASE_DIR / "knowledge_base"
        self._knowledge_base_path.mkdir(exist_ok=True)
        logger.info(f"JsonStorage инициализирован: {db_path}")
        logger.info(f"База знаний: {self._knowledge_base_path}")

    def _load_raw_data(self) -> Dict[str, Any]:
        """Сырая загрузка данных из JSON."""
        with open(self._db_path, "r", encoding="utf-8") as f:
            return json.load(f)

    def _save_raw_data(self, data: Dict[str, Any]) -> None:
        """Атомарное сохранение данных в JSON."""
        temp_fd, temp_path = tempfile.mkstemp(suffix=".json", dir=self._db_path.parent)

        try:
            with os.fdopen(temp_fd, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)

            shutil.move(temp_path, self._db_path)
            self._cache.invalidate()
            logger.info(f"Данные атомарно сохранены в {self._db_path}")

        except Exception as e:
            logger.exception(f"Ошибка сохранения данных: {e}")
            if os.path.exists(temp_path):
                os.remove(temp_path)
            raise

    def _validate_user(self, user: Dict[str, Any]) -> bool:
        """Проверка пользователя на наличие обязательных полей."""
        required_fields = {"id", "name", "points"}
        return isinstance(user, dict) and all(field in user for field in required_fields)

    def _validate_data(self, data: Dict[str, Any]) -> List[User]:
        """Валидация всех данных."""
        if "users" not in data:
            logger.error("Отсутствует ключ 'users' в базе данных")
            raise ValueError("Некорректная структура базы данных: отсутствует 'users'")

        valid_users = []
        for user in data["users"]:
            if self._validate_user(user):
                valid_users.append(User.from_dict(user))
            else:
                logger.warning(f"Пользователь не прошёл валидацию и будет пропущен: {user}")

        if len(valid_users) != len(data["users"]):
            logger.warning(
                f"Отфильтровано {len(data['users']) - len(valid_users)} некорректных записей"
            )

        return valid_users

    # ==================== БАЗОВЫЕ МЕТОДЫ ====================

    def get_all_users(self) -> List[User]:
        """Получение всех пользователей с кэшированием."""
        cached = self._cache.get()
        if cached is not None:
            logger.debug("Данные загружены из кэша")
            return [User.from_dict(u) for u in cached]

        logger.debug("Загрузка данных из файла")
        raw_data = self._load_raw_data()
        valid_users = self._validate_data(raw_data)

        self._cache.set([u.to_dict() for u in valid_users])
        logger.info(f"Загружено {len(valid_users)} пользователей")
        return valid_users

    def get_user_by_id(self, user_id: int) -> Optional[User]:
        """Поиск пользователя по ID."""
        users = self.get_all_users()
        user = next((u for u in users if u.id == user_id), None)

        if user is None:
            logger.warning(f"Пользователь с ID={user_id} не найден")

        return user

    def save_user(self, user: User) -> bool:
        """Сохранение данных пользователя."""
        users = self.get_all_users()

        user_index = next((i for i, u in enumerate(users) if u.id == user.id), None)

        if user_index is None:
            logger.warning(f"Пользователь {user.id} не найден для сохранения")
            return False

        users[user_index] = user
        self._save_raw_data({"users": [u.to_dict() for u in users]})

        logger.info(f"Пользователь {user.id} сохранён")
        return True

    def get_users_sorted_by_points(self, limit: Optional[int] = None) -> List[User]:
        """Получение пользователей, отсортированных по баллам."""
        users = self.get_all_users()
        sorted_users = sorted(users, key=lambda x: x.points, reverse=True)

        if limit is not None:
            limit = max(1, min(limit, len(sorted_users)))
            sorted_users = sorted_users[:limit]
            logger.debug(f"Возвращаем топ-{limit} пользователей")

        return sorted_users

    def get_teams_with_scores(self) -> List[Tuple[str, int]]:
        """Группировка по командам с подсчётом суммы баллов."""
        users = self.get_all_users()
        team_scores: Dict[str, int] = {}

        for user in users:
            team = user.team or "Без команды"
            team_scores[team] = team_scores.get(team, 0) + user.points

        sorted_teams = sorted(team_scores.items(), key=lambda x: x[1], reverse=True)
        logger.info(f"Посчитано {len(sorted_teams)} команд")
        return sorted_teams

    def get_team_members(self, team_name: str) -> List[User]:
        """Получение списка участников команды."""
        users = self.get_all_users()
        team_users = [u for u in users if (u.team or "").lower() == team_name.lower()]
        team_users.sort(key=lambda x: x.points, reverse=True)

        logger.debug(f"Найдено {len(team_users)} участников в команде '{team_name}'")
        return team_users

    def get_all_achievements(self) -> List[AchievementInfo]:
        """Получение всех уникальных достижений."""
        users = self.get_all_users()
        achievement_owners: Dict[str, int] = {}

        for user in users:
            for achievement in user.achievements:
                achievement_owners[achievement] = achievement_owners.get(achievement, 0) + 1

        result = [
            AchievementInfo(name=name, count=count)
            for name, count in sorted(achievement_owners.items())
        ]

        logger.info(f"Найдено {len(result)} уникальных достижений")
        return result

    def get_statistics(self) -> Dict[str, Any]:
        """Общая статистика по базе данных."""
        users = self.get_all_users()

        if not users:
            return {
                "total_users": 0,
                "total_points": 0,
                "average_points": 0,
                "teams_count": 0,
                "top_scorer": None,
            }

        total_points = sum(u.points for u in users)
        teams = set(u.team for u in users if u.team)
        top_scorer = max(users, key=lambda x: x.points)

        return {
            "total_users": len(users),
            "total_points": total_points,
            "average_points": total_points / len(users),
            "teams_count": len(teams),
            "top_scorer": {"name": top_scorer.name, "points": top_scorer.points, "team": top_scorer.team},
        }

    def get_user_rank(self, user_id: int) -> Optional[int]:
        """Получение места пользователя в рейтинге."""
        sorted_users = self.get_users_sorted_by_points()

        for rank, user in enumerate(sorted_users, 1):
            if user.id == user_id:
                logger.debug(f"Пользователь {user_id} на позиции #{rank}")
                return rank

        return None

    def get_team_stats(self, team_name: str) -> Dict[str, Any]:
        """Детальная статистика команды."""
        members = self.get_team_members(team_name)

        if not members:
            return {"exists": False}

        total_points = sum(m.points for m in members)
        total_tests = sum(sum(t.score for t in m.tests) for m in members)
        total_projects = sum(sum(p.points for p in m.projects) for m in members)
        total_events = sum(len(m.events) for m in members)

        return {
            "exists": True,
            "name": members[0].team,
            "members_count": len(members),
            "total_points": total_points,
            "average_points": total_points / len(members),
            "tests_score": total_tests,
            "projects_score": total_projects,
            "events_count": total_events,
            "members": members,
        }

    # ==================== АДМИН-МЕТОДЫ ====================

    def mark_attendance(self, user_id: int, event_name: str, attended: bool) -> Tuple[bool, str]:
        """Отметка посещаемости мероприятия."""
        from datetime import datetime

        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"

        event_found = False
        for event in user.events:
            if event.name.lower() == event_name.lower():
                event.attended = attended
                event_found = True
                break

        if not event_found:
            user.events.append(
                Event(name=event_name, date=datetime.now().strftime("%Y-%m-%d"), attended=attended)
            )

        if self.save_user(user):
            status = "посещено" if attended else "пропущено"
            logger.info(f"Отмечено {status} мероприятия '{event_name}' для пользователя {user_id}")
            return True, f"Посещаемость обновлена: {event_name} — {'Был' if attended else 'Не был'}"

        return False, "Ошибка сохранения"

    def update_project_stage(self, user_id: int, project_name: str, stage: str) -> Tuple[bool, str]:
        """Обновление стадии проекта."""
        if stage.lower() not in self.VALID_STAGES:
            return False, f"Некорректная стадия. Доступные: {', '.join(self.VALID_STAGES)}"

        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"

        project_found = False
        for project in user.projects:
            if project.name.lower() == project_name.lower():
                project.current_stage = stage.lower()
                project_found = True
                break

        if not project_found:
            return False, f"Проект '{project_name}' не найден у пользователя"

        if self.save_user(user):
            logger.info(f"Обновлена стадия проекта '{project_name}' до '{stage}'")
            return True, f"Стадия проекта '{project_name}' обновлена на '{stage}'"

        return False, "Ошибка сохранения"

    def get_worst_test_takers(self, limit: int) -> List[Dict[str, Any]]:
        """Получение пользователей с наименьшими баллами за тесты."""
        users = self.get_all_users()

        users_with_test_scores = []
        for user in users:
            test_score = sum(t.score for t in user.tests)
            users_with_test_scores.append(
                {
                    "id": user.id,
                    "name": user.name,
                    "team": user.team or "Без команды",
                    "test_score": test_score,
                    "tests_count": len(user.tests),
                    "total_points": user.points,
                }
            )

        sorted_users = sorted(users_with_test_scores, key=lambda x: x["test_score"])
        return sorted_users[:limit]

    def get_non_attendees(self) -> List[Dict[str, Any]]:
        """Получение пользователей, не посещавших мероприятия."""
        users = self.get_all_users()

        non_attendees = []
        for user in users:
            attended_events = [e for e in user.events if e.attended]

            if not attended_events:
                non_attendees.append(
                    {
                        "id": user.id,
                        "name": user.name,
                        "team": user.team or "Без команды",
                        "total_events": len(user.events),
                        "attended_count": 0,
                        "total_points": user.points,
                    }
                )

        logger.info(f"Найдено {len(non_attendees)} пользователей без посещённых мероприятий")
        return non_attendees

    def add_pending_achievement(
        self, user_id: int, achievement_name: str, description: str
    ) -> Tuple[bool, str]:
        """Добавление заявки на достижение."""
        from datetime import datetime

        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"

        # Проверяем существующие заявки
        for pending in user.pending_admin_review:
            name = pending.get("achievement_name") if isinstance(pending, dict) else pending.achievement_name
            if name.lower() == achievement_name.lower():
                return False, f"Заявка на '{achievement_name}' уже находится на рассмотрении"

        user.pending_admin_review.append(
            {
                "achievement_name": achievement_name,
                "description": description,
                "submitted_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "status": "pending",
            }
        )

        if self.save_user(user):
            logger.info(f"Добавлена заявка на '{achievement_name}' для пользователя {user_id}")
            return True, f"Заявка на '{achievement_name}' отправлена на проверку администратору"

        return False, "Ошибка сохранения"

    def get_pending_achievements(self) -> List[Dict[str, Any]]:
        """Получение заявок на достижения."""
        users = self.get_all_users()

        pending_list = []
        for user in users:
            for pending in user.pending_admin_review:
                # Обрабатываем как dict, так и PendingAchievement
                if isinstance(pending, dict):
                    status = pending.get("status")
                else:
                    status = pending.status

                if status == "pending":
                    name = pending.get("achievement_name") if isinstance(pending, dict) else pending.achievement_name
                    desc = pending.get("description") if isinstance(pending, dict) else pending.description
                    submitted = pending.get("submitted_at") if isinstance(pending, dict) else pending.submitted_at

                    pending_list.append(
                        {
                            "user_id": user.id,
                            "user_name": user.name,
                            "team": user.team or "Без команды",
                            "achievement_name": name,
                            "description": desc,
                            "submitted_at": submitted,
                        }
                    )

        logger.info(f"Найдено {len(pending_list)} достижений на проверке")
        return pending_list

    def approve_pending_achievement(self, user_id: int, achievement_name: str) -> Tuple[bool, str]:
        """Одобрение достижения."""
        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"

        pending_index = None
        achievement_to_add = None
        for i, pending in enumerate(user.pending_admin_review):
            if isinstance(pending, dict) and pending.get("achievement_name", "").lower() == achievement_name.lower():
                pending_index = i
                achievement_to_add = pending.get("achievement_name")
                break
            elif hasattr(pending, "achievement_name") and pending.achievement_name.lower() == achievement_name.lower():
                pending_index = i
                achievement_to_add = pending.achievement_name
                break

        if pending_index is None:
            return False, f"Заявка на '{achievement_name}' не найдена"

        if achievement_to_add in user.achievements:
            user.pending_admin_review.pop(pending_index)
            self.save_user(user)
            return True, f"Достижение '{achievement_to_add}' уже получено пользователем"

        user.achievements.append(achievement_to_add)
        user.pending_admin_review.pop(pending_index)

        if self.save_user(user):
            logger.info(f"Одобрено достижение '{achievement_to_add}'")
            return True, f"Достижение '{achievement_to_add}' одобрено и добавлено!"

        return False, "Ошибка сохранения"

    def reject_pending_achievement(
        self, user_id: int, achievement_name: str, reason: str = ""
    ) -> Tuple[bool, str]:
        """Отклонение достижения."""
        from datetime import datetime

        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"

        pending_index = None
        for i, pending in enumerate(user.pending_admin_review):
            if isinstance(pending, dict) and pending.get("achievement_name", "").lower() == achievement_name.lower():
                pending_index = i
                break
            elif hasattr(pending, "achievement_name") and pending.achievement_name.lower() == achievement_name.lower():
                pending_index = i
                break

        if pending_index is None:
            return False, f"Заявка на '{achievement_name}' не найдена"

        if isinstance(user.pending_admin_review[pending_index], dict):
            user.pending_admin_review[pending_index]["status"] = "rejected"
            user.pending_admin_review[pending_index]["rejection_reason"] = reason
            user.pending_admin_review[pending_index]["reviewed_at"] = datetime.now().strftime(
                "%Y-%m-%d %H:%M:%S"
            )
        else:
            user.pending_admin_review[pending_index].status = "rejected"
            user.pending_admin_review[pending_index].rejection_reason = reason
            user.pending_admin_review[pending_index].reviewed_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        if self.save_user(user):
            logger.info(f"Отклонено достижение '{achievement_name}'")
            return True, f"Заявка отклонена: {reason}" if reason else "Заявка отклонена"

        return False, "Ошибка сохранения"

    def get_team_leads(self, team_name: Optional[str] = None) -> List[Dict[str, Any]]:
        """Получение тимлидов."""
        users = self.get_all_users()

        leads = []
        for user in users:
            if user.is_team_lead:
                if team_name is None or (user.team or "").lower() == team_name.lower():
                    leads.append({"id": user.id, "name": user.name, "team": user.team, "points": user.points})

        logger.info(f"Найдено {len(leads)} тимлидов" + (f" в команде '{team_name}'" if team_name else ""))
        return leads

    def get_users_by_stage(self, stage: str) -> List[Dict[str, Any]]:
        """Получение пользователей по стадии проекта."""
        users = self.get_all_users()

        if stage.lower() not in self.VALID_STAGES:
            logger.warning(f"Некорректная стадия: {stage}")
            return []

        result = []
        for user in users:
            for project in user.projects:
                if project.current_stage.lower() == stage.lower():
                    result.append(
                        {
                            "id": user.id,
                            "name": user.name,
                            "team": user.team or "Без команды",
                            "project_name": project.name,
                            "project_status": project.status,
                            "current_stage": project.current_stage,
                            "points": user.points,
                        }
                    )
                    break

        logger.info(f"Найдено {len(result)} пользователей на стадии '{stage}'")
        return result

    # ==================== МЕРОПРИЯТИЯ: РЕАЛИЗАЦИЯ ====================

    def _load_events(self) -> List[Dict[str, Any]]:
        """Загрузить список мероприятий из JSON."""
        data = self._load_raw_data()
        return data.get("events", [])

    def _save_events(self, events: List[Dict[str, Any]]) -> None:
        """Сохранить список мероприятий в JSON."""
        data = self._load_raw_data()
        data["events"] = events
        self._save_raw_data(data)

    def get_all_events(self) -> List[Dict[str, Any]]:
        """Получить все глобальные мероприятия."""
        events = self._load_events()
        logger.info(f"Загружено {len(events)} мероприятий")
        return events

    def create_event(self, name: str, date: str, description: str = "") -> Tuple[bool, str]:
        """Создать новое мероприятие."""
        events = self._load_events()
        for event in events:
            if event["name"].lower() == name.lower():
                return False, f"Мероприятие '{name}' уже существует"
        events.append({"name": name, "date": date, "description": description})
        self._save_events(events)
        logger.info(f"Создано мероприятие: {name}")
        return True, f"Мероприятие '{name}' создано"

    def update_event(self, event_name: str, new_name: str = None, new_date: str = None, new_description: str = None) -> Tuple[bool, str]:
        """Обновить мероприятие."""
        events = self._load_events()
        event_found = None
        for event in events:
            if event["name"].lower() == event_name.lower():
                event_found = event
                break
        if not event_found:
            return False, f"Мероприятие '{event_name}' не найдено"
        if new_name:
            event_found["name"] = new_name
        if new_date:
            event_found["date"] = new_date
        if new_description is not None:
            event_found["description"] = new_description
        self._save_events(events)
        logger.info(f"Обновлено мероприятие: {event_name}")
        return True, f"Мероприятие '{event_name}' обновлено"

    def delete_event(self, event_name: str) -> Tuple[bool, str]:
        """Удалить мероприятие."""
        events = self._load_events()
        event_index = None
        for i, event in enumerate(events):
            if event["name"].lower() == event_name.lower():
                event_index = i
                break
        if event_index is None:
            return False, f"Мероприятие '{event_name}' не найдено"
        events.pop(event_index)
        self._save_events(events)
        logger.info(f"Удалено мероприятие: {event_name}")
        return True, f"Мероприятие '{event_name}' удалено"

    # ==================== БАЛЛЫ: РЕАЛИЗАЦИЯ ====================

    def add_points_to_user(self, user_id: int, points: int, reason: str = "") -> Tuple[bool, str]:
        """Начислить баллы пользователю."""
        if points <= 0:
            return False, "Баллы должны быть положительным числом"
        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"
        user.points += points
        if self.save_user(user):
            logger.info(f"Начислено {points} баллов пользователю {user_id}")
            reason_text = f": {reason}" if reason else ""
            return True, f"Начислено {points} баллов{reason_text}"
        return False, "Ошибка сохранения"

    def deduct_points_from_user(self, user_id: int, points: int, reason: str = "") -> Tuple[bool, str]:
        """Списать баллы у пользователя."""
        if points <= 0:
            return False, "Баллы должны быть положительным числом"
        user = self.get_user_by_id(user_id)
        if user is None:
            return False, f"Пользователь с ID={user_id} не найден"
        if user.points < points:
            return False, f"Недостаточно баллов (текущий баланс: {user.points})"
        user.points -= points
        if self.save_user(user):
            logger.info(f"Списано {points} баллов у пользователя {user_id}")
            reason_text = f": {reason}" if reason else ""
            return True, f"Списано {points} баллов{reason_text}"
        return False, "Ошибка сохранения"

    # ==================== БАЗА ЗНАНИЙ: РЕАЛИЗАЦИЯ ====================

    def _get_knowledge_index_path(self) -> Path:
        """Получить путь к индексу базы знаний."""
        return self._knowledge_base_path / "index.json"

    def _load_knowledge_index(self) -> List[Dict[str, Any]]:
        """Загрузить индекс базы знаний."""
        index_path = self._get_knowledge_index_path()
        if not index_path.exists():
            return []
        with open(index_path, "r", encoding="utf-8") as f:
            return json.load(f)

    def _save_knowledge_index(self, index: List[Dict[str, Any]]) -> None:
        """Сохранить индекс базы знаний."""
        index_path = self._get_knowledge_index_path()
        with open(index_path, "w", encoding="utf-8") as f:
            json.dump(index, f, ensure_ascii=False, indent=2)

    def add_knowledge_item(self, title: str, content: str, category: str = "") -> Tuple[bool, str]:
        """Добавить материал в базу знаний."""
        import uuid
        from datetime import datetime
        index = self._load_knowledge_index()
        item_id = str(uuid.uuid4())[:8]
        content_path = self._knowledge_base_path / f"{item_id}.md"
        with open(content_path, "w", encoding="utf-8") as f:
            f.write(content)
        index.append({
            "id": item_id,
            "title": title,
            "category": category,
            "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "content_file": f"{item_id}.md",
        })
        self._save_knowledge_index(index)
        logger.info(f"Добавлен материал в базу знаний: {title}")
        return True, f"Материал '{title}' добавлен в базу знаний"

    def get_all_knowledge_items(self, category: str = None) -> List[Dict[str, Any]]:
        """Получить все материалы из базы знаний."""
        index = self._load_knowledge_index()
        if category:
            index = [item for item in index if item.get("category", "").lower() == category.lower()]
        result = []
        for item in index:
            content_path = self._knowledge_base_path / item["content_file"]
            if content_path.exists():
                with open(content_path, "r", encoding="utf-8") as f:
                    content = f.read()
                result.append({
                    "id": item["id"],
                    "title": item["title"],
                    "category": item.get("category", ""),
                    "content": content,
                    "created_at": item.get("created_at", ""),
                })
            else:
                logger.warning(f"Файл контента не найден: {content_path}")
        logger.info(f"Загружено {len(result)} материалов из базы знаний")
        return result

    def delete_knowledge_item(self, item_id: str) -> Tuple[bool, str]:
        """Удалить материал из базы знаний."""
        index = self._load_knowledge_index()
        item_index = None
        item_to_delete = None
        for i, item in enumerate(index):
            if item["id"] == item_id:
                item_index = i
                item_to_delete = item
                break
        if item_index is None:
            return False, f"Материал с ID={item_id} не найден"
        content_path = self._knowledge_base_path / item_to_delete["content_file"]
        if content_path.exists():
            content_path.unlink()
        index.pop(item_index)
        self._save_knowledge_index(index)
        logger.info(f"Удален материал из базы знаний: {item_to_delete['title']}")
        return True, f"Материал '{item_to_delete['title']}' удален"


# ===========================================================
# ГЛОБАЛЬНЫЙ ЭКЗЕМПЛЯР ХРАНИЛИЩА
# ===========================================================
storage = JsonStorage(DATABASE_FILE, CACHE_TTL_SECONDS)