system-pri/backend/app/coordinator/query/enrollments.py

from datetime import datetime
from typing import List

from flask import abort
from sqlalchemy import and_, or_

from ...base.mode import EnrollmentsMode
from ...dependencies import db
from ...examination_schedule.models import ExaminationSchedule, TermOfDefence
from ...project_supervisor.models import ProjectSupervisor
from ...students.models import Group


def get_term_of_defence_by_id_and_examination_schedule_id(
    examination_schedule_id: int, term_of_defence_id: int
) -> ExaminationSchedule:
    td = TermOfDefence.query.filter(
        TermOfDefence.id == term_of_defence_id,
        TermOfDefence.examination_schedule_id == examination_schedule_id,
    ).first()
    if td is None:
        abort(404, "Not found examination schedule or term of defence!")
    return td


def get_group_by_id(group_id: int) -> Group:
    group = Group.query.filter(Group.id == group_id).first()
    if group is None:
        abort(404, "Not found group!")
    return group


def check_the_group_has_assigned_to_term_of_defence(group_id: int) -> TermOfDefence:
    td = TermOfDefence.query.filter(TermOfDefence.group_id == group_id).first()
    if td is not None:
        abort(400, "Group has already assigned to term of defence!")
    return td


def set_new_group_to_term_of_defence(
    examination_schedule_id: int, term_of_defence_id: int, group_id: int
) -> TermOfDefence:
    td = get_term_of_defence_by_id_and_examination_schedule_id(
        examination_schedule_id, term_of_defence_id
    )
    get_group_by_id(group_id)
    check_the_group_has_assigned_to_term_of_defence(group_id)
    td.group_id = group_id
    return td


def get_examination_schedule_by_id(examination_schedule_id: int) -> ExaminationSchedule:
    ex = ExaminationSchedule.query.filter(
        ExaminationSchedule.id == examination_schedule_id
    ).first()
    if ex is None:
        abort(404, "Not found examination schedule!")
    return ex


def get_and_check_the_project_supervisors_exists_in_db(
    year_group_id: int, project_supervisors_ids: List[int]
) -> List[ProjectSupervisor]:
    project_supervisors = (
        ProjectSupervisor.query.filter(
            or_(*[ProjectSupervisor.id == i for i in project_supervisors_ids])
        )
        .filter(ProjectSupervisor.year_group_id == year_group_id)
        .all()
    )

    if len(project_supervisors) != len(project_supervisors_ids):
        abort(404, "Not found project supervisors!")

    return project_supervisors


def validate_enrollments_date(
    ex_start_date: datetime, ex_end_date: datetime, duration_time: int, data: dict
) -> None:
    start_date = data.get("start_date")
    end_date = data.get("end_date")

    if not (
        ex_start_date.timestamp() <= start_date.timestamp()
        and ex_end_date.timestamp() >= end_date.timestamp()
    ):
        abort(400, "Invalid date range!")

    if end_date <= start_date:
        abort(400, "End date must be greater than start date!")

    delta_time = end_date - start_date
    delta_time_in_minutes = delta_time.total_seconds() / 60
    if delta_time_in_minutes % duration_time != 0:
        abort(400, "Invalid duration time!")


def check_the_term_of_defence_not_exists_in_chosen_date_range(
    examination_schedule_id: int, data: dict
) -> None:
    start_date = data.get("start_date")
    end_date = data.get("end_date")
    td = (
        TermOfDefence.query.filter(
            TermOfDefence.examination_schedule_id == examination_schedule_id
        )
        .filter(
            or_(
                and_(
                    TermOfDefence.start_date >= start_date,
                    TermOfDefence.start_date < end_date,
                    TermOfDefence.end_date >= end_date,
                ),
                and_(
                    TermOfDefence.start_date <= start_date,
                    TermOfDefence.end_date > start_date,
                    TermOfDefence.end_date <= end_date,
                ),
            )
        )
        .first()
    )

    if td is not None:
        abort(400, "This term of defence is taken! You choose other date!")


def set_enrollments_mode(
    examination_schedule_id: int, mode: EnrollmentsMode, action_name: str
) -> dict:
    examination_schedule = (
        db.session.query(ExaminationSchedule)
        .filter(ExaminationSchedule.id == examination_schedule_id)
        .first()
    )

    if examination_schedule is None:
        abort(404, "Not found examination schedule!")

    examination_schedule.open_enrollments = mode.value
    db.session.commit()
    return {"message": f"You {action_name} enrollments for this examination schedule!"}