database.py - master - Befun/FamalyControlle

database.py
197 строк8 KB
BeFun
Init project
07 май 2026, 10:43
07 май 2026, 10:437e96900
"""
Менеджер БД — высокоуровневые CRUD-операции.
"""

from datetime import datetime, timedelta
from typing import Optional

from models.models import (
    User, Restriction, Schedule, HistoryEntry,
    DatabaseManager as _DBBase,
)


class Database:
    """
    Фасад над DatabaseManager + Session.
    Предоставляет удобные методы для бизнес-логики.
    """

    def __init__(self, database_url: str):
        self._db = _DBBase(database_url)

    # ── Сессии ────────────────────────────────────────────────

    def get_session(self):
        return self._db.get_session()

    # ── Users ─────────────────────────────────────────────────

    def get_user_by_id(self, user_id: int) -> Optional[User]:
        with self.get_session() as s:
            return s.query(User).filter(User.id == user_id).first()

    def get_user_by_max_id(self, max_user_id: str) -> Optional[User]:
        with self.get_session() as s:
            return s.query(User).filter(User.max_user_id == max_user_id).first()

    def get_all_users(self) -> list[User]:
        with self.get_session() as s:
            return s.query(User).filter(User.is_active == True).all()  # noqa: E712

    def create_user(self, max_user_id: str, display_name: str = "",
                    role: str = "child") -> User:
        with self.get_session() as s:
            user = User(
                max_user_id=max_user_id,
                display_name=display_name or "Пользователь",
                role=role,
            )
            s.add(user)
            s.commit()
            s.refresh(user)
            return user

    def update_user_role(self, user_id: int, new_role: str) -> Optional[User]:
        with self.get_session() as s:
            user = s.query(User).filter(User.id == user_id).first()
            if user:
                user.role = new_role
                s.commit()
                s.refresh(user)
            return user

    def deactivate_user(self, user_id: int) -> bool:
        with self.get_session() as s:
            user = s.query(User).filter(User.id == user_id).first()
            if user:
                user.is_active = False
                s.commit()
                return True
            return False

    # ── Restrictions ──────────────────────────────────────────

    def get_active_restrictions(self) -> list[Restriction]:
        """Получить все активные ограничения (с учётом срока действия)."""
        now = datetime.utcnow()
        with self.get_session() as s:
            result = s.query(Restriction).filter(
                Restriction.is_active == True,  # noqa: E712
                (Restriction.expires_at.is_(None)) | (Restriction.expires_at > now),
            ).all()
            # Проверяем просроченные и деактивируем
            expired = [r for r in result if r.expires_at and r.expires_at <= now]
            for r in expired:
                r.is_active = False
            if expired:
                s.commit()
            return [r for r in result if r.is_active]

    def create_restriction(self, category: str, target: str,
                           created_by: int,
                           expires_minutes: Optional[int] = None) -> Restriction:
        expires_at = None
        if expires_minutes:
            expires_at = datetime.utcnow() + timedelta(minutes=expires_minutes)

        with self.get_session() as s:
            r = Restriction(
                category=category,
                target=target,
                created_by=created_by,
                expires_at=expires_at,
            )
            s.add(r)
            s.commit()
            s.refresh(r)

            # Логируем
            self._log(s, "block", f"{category}:{target}", created_by,
                      f"Срок: {expires_minutes} мин" if expires_minutes else "Бессрочно")
            return r

    def remove_restriction(self, restriction_id: int,
                           removed_by: int) -> Optional[Restriction]:
        with self.get_session() as s:
            r = s.query(Restriction).filter(Restriction.id == restriction_id).first()
            if r:
                r.is_active = False
                s.commit()
                self._log(s, "unblock", f"{r.category}:{r.target}", removed_by)
                s.refresh(r)
            return r

    def get_restriction(self, restriction_id: int) -> Optional[Restriction]:
        with self.get_session() as s:
            return s.query(Restriction).filter(Restriction.id == restriction_id).first()

    def get_history(self, limit: int = 50, offset: int = 0) -> list[HistoryEntry]:
        with self.get_session() as s:
            return (s.query(HistoryEntry)
                    .order_by(HistoryEntry.created_at.desc())
                    .offset(offset)
                    .limit(limit)
                    .all())

    # ── Schedules ─────────────────────────────────────────────

    def get_schedules(self) -> list[Schedule]:
        with self.get_session() as s:
            return s.query(Schedule).filter(Schedule.is_active == True).all()  # noqa: E712

    def create_schedule(self, name: str, schedule_type: str,
                        start_time: str, end_time: str,
                        days: list[int], created_by: int) -> Schedule:
        with self.get_session() as s:
            sch = Schedule(
                name=name,
                schedule_type=schedule_type,
                start_time=start_time,
                end_time=end_time,
                days=days,
                created_by=created_by,
            )
            s.add(sch)
            s.commit()
            s.refresh(sch)
            self._log(s, "schedule_create", name, created_by,
                      f"{start_time}-{end_time} дни={days}")
            return sch

    def update_schedule(self, schedule_id: int, **kwargs) -> Optional[Schedule]:
        with self.get_session() as s:
            sch = s.query(Schedule).filter(Schedule.id == schedule_id).first()
            if sch:
                for key, val in kwargs.items():
                    if hasattr(sch, key):
                        setattr(sch, key, val)
                s.commit()
                s.refresh(sch)
            return sch

    def remove_schedule(self, schedule_id: int, removed_by: int) -> bool:
        with self.get_session() as s:
            sch = s.query(Schedule).filter(Schedule.id == schedule_id).first()
            if sch:
                s.delete(sch)
                s.commit()
                self._log(s, "schedule_remove", sch.name, removed_by)
                return True
            return False

    def get_schedule(self, schedule_id: int) -> Optional[Schedule]:
        with self.get_session() as s:
            return s.query(Schedule).filter(Schedule.id == schedule_id).first()

    # ── Logging ───────────────────────────────────────────────

    def _log(self, session, action: str, target: str,
             user_id: int, details: str = "") -> None:
        entry = HistoryEntry(
            action=action,
            target=target,
            details=details,
            user_id=user_id,
        )
        session.add(entry)