manager.py - master - ring0/support-system

manager.py
186 строк8 KB
ring0
Init project
22 июн 2026, 17:13
22 июн 2026, 17:139df53b7
from datetime import datetime

from sqlalchemy import and_
from sqlalchemy.orm import Session

from auth import hash_password, verify_password
from models import User, Category, Priority, Ticket, Comment


class UserManager:
    """Менеджер пользователей: регистрация, аутентификация, хэширование паролей"""

    def register(self, db: Session, data):
        """Регистрация нового пользователя с хэшированием пароля"""
        exists = db.query(User).filter(
            (User.username == data.username) | (User.email == data.email)
        ).first()
        if exists:
            return {"error": "Пользователь с таким логином или email уже существует"}

        user = User(
            username=data.username,
            password_hash=hash_password(data.password),
            full_name=data.full_name,
            email=data.email,
            role=data.role,
        )
        db.add(user)
        db.commit()
        db.refresh(user)
        return user.to_dict()

    def authenticate(self, db: Session, username: str, password: str):
        """Проверка учётных данных пользователя"""
        user = db.query(User).filter(User.username == username).first()
        if not user or not verify_password(password, user.password_hash):
            return None
        return user

    def get_users(self, db: Session, role: str = None):
        """Получение списка пользователей (опционально по роли)"""
        query = db.query(User)
        if role:
            query = query.filter(User.role == role)
        return query.all()


class TicketManager:
    """Менеджер заявок: создание с проверкой дубликатов, фильтрация, назначение"""

    def create_ticket(self, db: Session, data, author: User):
        """Создание заявки с проверкой существования справочников и дубликатов"""
        # Проверка существования категории и приоритета
        category = db.query(Category).filter(Category.id == data.category_id).first()
        if not category:
            return {"error": "Указанная категория не существует", "code": 400}

        priority = db.query(Priority).filter(Priority.id == data.priority_id).first()
        if not priority:
            return {"error": "Указанный приоритет не существует", "code": 400}

        # Защита от дублирования: открытая заявка того же автора с той же темой
        duplicate = db.query(Ticket).filter(
            and_(
                Ticket.author_id == author.id,
                Ticket.title == data.title,
                Ticket.status.in_(["open", "in_progress"]),
            )
        ).first()
        if duplicate:
            return {
                "error": "Уже существует открытая заявка с такой темой",
                "code": 409,
                "existing_ticket_id": duplicate.id,
            }

        ticket = Ticket(
            title=data.title,
            description=data.description,
            category_id=data.category_id,
            priority_id=data.priority_id,
            author_id=author.id,
            status="open",
        )
        db.add(ticket)
        db.commit()
        db.refresh(ticket)
        return ticket.to_dict()

    def get_tickets(self, db: Session, current_user: User, filters: dict):
        """Получение заявок с учётом роли пользователя и фильтров"""
        query = db.query(Ticket)

        # Разграничение видимости по роли
        if current_user.role == "user":
            query = query.filter(Ticket.author_id == current_user.id)
        elif current_user.role == "specialist":
            query = query.filter(Ticket.assignee_id == current_user.id)
        # администратор видит все заявки — без ограничения

        # Опциональные фильтры (комбинируются)
        if filters.get("status"):
            query = query.filter(Ticket.status == filters["status"])
        if filters.get("category_id"):
            query = query.filter(Ticket.category_id == filters["category_id"])
        if filters.get("priority_id"):
            query = query.filter(Ticket.priority_id == filters["priority_id"])
        if filters.get("assignee_id"):
            query = query.filter(Ticket.assignee_id == filters["assignee_id"])
        if filters.get("author_id"):
            query = query.filter(Ticket.author_id == filters["author_id"])

        # Сортировка по весу приоритета и дате создания
        query = query.join(Priority).order_by(
            Priority.weight.desc(), Ticket.created_at.desc()
        )
        return query.all()

    def get_ticket(self, db: Session, ticket_id: int):
        return db.query(Ticket).filter(Ticket.id == ticket_id).first()

    def update_ticket(self, db: Session, ticket_id: int, data, current_user: User):
        """Обновление заявки с учётом прав роли"""
        ticket = db.query(Ticket).filter(Ticket.id == ticket_id).first()
        if not ticket:
            return {"error": "Заявка не найдена", "code": 404}

        # Специалист может менять только статус назначенных ему заявок
        if current_user.role == "specialist":
            if ticket.assignee_id != current_user.id:
                return {"error": "Заявка не назначена данному специалисту", "code": 403}
            if data.status is not None:
                ticket.status = data.status

        # Администратор может изменять любые поля
        elif current_user.role == "admin":
            if data.status is not None:
                ticket.status = data.status
            if data.assignee_id is not None:
                assignee = db.query(User).filter(User.id == data.assignee_id).first()
                if not assignee:
                    return {"error": "Назначаемый исполнитель не найден", "code": 400}
                ticket.assignee_id = data.assignee_id
                # При назначении исполнителя открытая заявка переходит в работу
                if ticket.status == "open":
                    ticket.status = "in_progress"
            if data.category_id is not None:
                ticket.category_id = data.category_id
            if data.priority_id is not None:
                ticket.priority_id = data.priority_id
        else:
            return {"error": "Недостаточно прав для обновления заявки", "code": 403}

        ticket.updated_at = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
        db.commit()
        db.refresh(ticket)
        return ticket.to_dict()


class CommentManager:
    """Менеджер комментариев: добавление и получение сообщений к заявкам"""

    def add_comment(self, db: Session, ticket_id: int, data, author: User):
        """Добавление комментария к заявке"""
        ticket = db.query(Ticket).filter(Ticket.id == ticket_id).first()
        if not ticket:
            return {"error": "Заявка не найдена", "code": 404}

        comment = Comment(
            text=data.text,
            ticket_id=ticket_id,
            author_id=author.id,
        )
        db.add(comment)
        db.commit()
        db.refresh(comment)
        return comment.to_dict()

    def get_comments(self, db: Session, ticket_id: int):
        """Получение всех комментариев заявки в хронологическом порядке"""
        return (
            db.query(Comment)
            .filter(Comment.ticket_id == ticket_id)
            .order_by(Comment.created_at.asc())
            .all()
        )