services/database.py - dev - nedfid16/kktd_keeper_bot

services/database.py
297 строк12 KB
nedfid16
upd
08 окт 2025, 18:40
08 окт 2025, 18:402c7abc7
import os
from pathlib import Path
import sqlite3
from datetime import date
from models import Student, Group, Teacher, Absence

class Database:
    def __init__(self, 
                #  db_path="attendance.db"
                 ):
        self.db_path = os.getenv('DATABASE_PATH', 'data/database.db')
        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
        self.init_tables()

    def init_tables(self):
        with sqlite3.connect(self.db_path) as conn:
            # Таблица преподавателей
            conn.execute("""
                CREATE TABLE IF NOT EXISTS teachers (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    first_name TEXT NOT NULL,
                    last_name TEXT NOT NULL,
                    patronymic TEXT,
                    tutor BOOLEAN NOT NULL DEFAULT FALSE
                )
            """)
            
            # Таблица групп
            conn.execute("""
                CREATE TABLE IF NOT EXISTS groups (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    tutor_id INTEGER,
                    FOREIGN KEY (tutor_id) REFERENCES teachers (id)
                )
            """)
            
            # Таблица студентов
            conn.execute("""
                CREATE TABLE IF NOT EXISTS students (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    first_name TEXT NOT NULL,
                    last_name TEXT NOT NULL,
                    patronymic TEXT,
                    elder BOOLEAN NOT NULL DEFAULT FALSE,
                    telegram_id INTEGER UNIQUE,
                    group_id INTEGER NOT NULL,
                    FOREIGN KEY (group_id) REFERENCES groups (id)
                )
            """)
            
            # Таблица отсутствий
            conn.execute("""
                CREATE TABLE IF NOT EXISTS absences (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    student_id INTEGER,
                    date TEXT NOT NULL,
                    lesson_number INTEGER,
                    reason TEXT CHECK(reason IN ('valid', 'invalid')),
                    FOREIGN KEY (student_id) REFERENCES students (id)
                )
            """)
            conn.commit()

    # Методы для работы со студентами
    def add_student(self, first_name: str, last_name: str, patronymic: str, 
                   elder: bool, telegram_id: int, group_id: int):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """INSERT INTO students (first_name, last_name, patronymic, elder, telegram_id, group_id)
                 VALUES (?, ?, ?, ?, ?, ?)""",
                (first_name, last_name, patronymic, elder, telegram_id, group_id)
            )
            conn.commit()

    def get_student_by_telegram_id(self, telegram_id: int) -> Student:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                """SELECT id, first_name, last_name, patronymic, elder, telegram_id, group_id 
                FROM students WHERE telegram_id = ?""",
                (telegram_id,)
            )
            row = cursor.fetchone()
            return Student(*row) if row else None

    def get_student_by_id(self, student_id: int) -> Student:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                """SELECT id, first_name, last_name, patronymic, elder, telegram_id, group_id 
                FROM students WHERE id = ?""",
                (student_id,)
            )
            row = cursor.fetchone()
            return Student(*row) if row else None

    def is_elder(self, telegram_id: int) -> bool:
        student = self.get_student_by_telegram_id(telegram_id)
        return student.elder if student else False

    def get_group_students(self, group_id: int):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                """SELECT id, first_name, last_name, patronymic, elder, telegram_id, group_id
                FROM students WHERE group_id = ? ORDER BY last_name, first_name""",
                (group_id,)
            )
            return [Student(*row) for row in cursor.fetchall()]

    # Методы для обновления данных студентов
    def update_student_first_name(self, student_id: int, first_name: str):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE students SET first_name = ? WHERE id = ?",
                (first_name, student_id)
            )
            conn.commit()

    def update_student_last_name(self, student_id: int, last_name: str):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE students SET last_name = ? WHERE id = ?",
                (last_name, student_id)
            )
            conn.commit()

    def update_student_patronymic(self, student_id: int, patronymic: str):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE students SET patronymic = ? WHERE id = ?",
                (patronymic, student_id)
            )
            conn.commit()

    def update_student_elder_status(self, student_id: int, elder: bool):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE students SET elder = ? WHERE id = ?",
                (elder, student_id)
            )
            conn.commit()

    def delete_student(self, student_id: int):
        with sqlite3.connect(self.db_path) as conn:
            # Сначала удаляем все отсутствия студента
            conn.execute("DELETE FROM absences WHERE student_id = ?", (student_id,))
            # Затем удаляем самого студента
            conn.execute("DELETE FROM students WHERE id = ?", (student_id,))
            conn.commit()

    # Методы для работы с отсутствиями
    def record_absence(self, student_id: int, date: str, lesson_number: int = None, reason: str = None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """INSERT INTO absences (student_id, date, lesson_number, reason)
                 VALUES (?, ?, ?, ?)""",
                (student_id, date, lesson_number, reason)
            )
            conn.commit()

    def get_student_absences(self, student_id: int):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT a.date, a.lesson_number, a.reason 
                FROM absences a
                WHERE a.student_id = ?
                ORDER BY a.date DESC
            """, (student_id,))
            return cursor.fetchall()

    def get_student_absences_detailed(self, student_id: int):
        """Получить детальную информацию об отсутствиях студента"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT date, lesson_number, reason 
                FROM absences 
                WHERE student_id = ?
                ORDER BY date DESC, lesson_number
            """, (student_id,))
            return cursor.fetchall()

    def get_group_absences(self, group_id: int, target_date: str = None):
        with sqlite3.connect(self.db_path) as conn:
            query = """
                SELECT s.first_name, s.last_name, a.date, a.lesson_number, a.reason 
                FROM absences a
                JOIN students s ON a.student_id = s.id
                WHERE s.group_id = ?
            """
            params = [group_id]
            
            if target_date:
                query += " AND a.date = ?"
                params.append(target_date)
            
            query += " ORDER BY a.date DESC, s.last_name, s.first_name"
            
            cursor = conn.execute(query, params)
            return cursor.fetchall()

    # Методы для работы с группами
    def get_group_by_id(self, group_id: int):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT id, name, tutor_id FROM groups WHERE id = ?",
                (group_id,)
            )
            row = cursor.fetchone()
            if row:
                return Group(*row)
            return None

    # Методы для добавления тестовых данных
    def add_teacher(self, first_name: str, last_name: str, patronymic: str = None, tutor: bool = False):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO teachers (first_name, last_name, patronymic, tutor) VALUES (?, ?, ?, ?)",
                (first_name, last_name, patronymic, tutor)
            )
            conn.commit()

    def add_group(self, name: str, tutor_id: int):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO groups (name, tutor_id) VALUES (?, ?)",
                (name, tutor_id)
            )
            conn.commit()
            
    def get_group_absences(self, group_id: int, target_date: str = None):
        with sqlite3.connect(self.db_path) as conn:
            query = """
                SELECT s.first_name, s.last_name, a.date, a.lesson_number, a.reason 
                FROM absences a
                JOIN students s ON a.student_id = s.id
                WHERE s.group_id = ?
            """
            params = [group_id]
            
            if target_date:
                query += " AND a.date = ?"
                params.append(target_date)
            
            # Добавляем сортировку по фамилии, имени и дате
            query += " ORDER BY s.last_name, s.first_name, a.lesson_number"
            
            cursor = conn.execute(query, params)
            return cursor.fetchall()
        
    
    def get_absence_by_id(self, absence_id: int):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT id, student_id, date, lesson_number, reason 
                FROM absences 
                WHERE id = ?
            """, (absence_id,))
            row = cursor.fetchone()
            if row:
                return Absence(*row)
            return None

    def update_absence(self, absence_id: int, date: str = None, lesson_number: int = None, reason: str = None):
        with sqlite3.connect(self.db_path) as conn:
            update_fields = []
            params = []
            
            if date is not None:
                update_fields.append("date = ?")
                params.append(date)
            if lesson_number is not None:
                update_fields.append("lesson_number = ?")
                params.append(lesson_number)
            if reason is not None:
                update_fields.append("reason = ?")
                params.append(reason)
                
            if update_fields:
                params.append(absence_id)
                query = f"UPDATE absences SET {', '.join(update_fields)} WHERE id = ?"
                conn.execute(query, params)
                conn.commit()

    def delete_absence(self, absence_id: int):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM absences WHERE id = ?", (absence_id,))
            conn.commit()

    def get_student_absences_with_ids(self, student_id: int):
        """Получить все отсутствия студента с ID записей"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT id, date, lesson_number, reason 
                FROM absences 
                WHERE student_id = ?
                ORDER BY date DESC, lesson_number
            """, (student_id,))
            return cursor.fetchall()