scripts/create_database.py - master - angrybricks/dementor

scripts/create_database.py
356 строк17 KB
MaksPV
Убрал опасные данные и сделал скрипт для развёртывания БД
30 сен 2025, 20:52
30 сен 2025, 20:521adc81f
import os
import mysql.connector
from mysql.connector import Error
from dotenv import load_dotenv
import logging


# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class DatabaseDeployer:
    def __init__(self):
        load_dotenv()  # Загружаем переменные окружения из .env файла

        self.db_config = {
            'host': os.getenv('DB_HOST', 'localhost'),
            'port': int(os.getenv('DB_PORT', 3306)),
            'user': os.getenv('DB_USER', 'dement'),
            'password': os.getenv('DB_PASSWORD', ''),
            'database': os.getenv('DB_NAME', 'dement')
        }

        # Конфигурация для подключения без выбора базы данных (для создания БД)
        self.root_config = {
            'host': self.db_config['host'],
            'port': self.db_config['port'],
            'user': self.db_config['user'],
            'password': self.db_config['password']
        }

    def get_connection(self, use_database=True):
        """Создает соединение с MySQL"""
        try:
            config = self.db_config if use_database else self.root_config
            connection = mysql.connector.connect(**config)
            logger.info("Успешное подключение к MySQL")
            return connection
        except Error as e:
            logger.error(f"Ошибка подключения к MySQL: {e}")
            return None

    def create_database(self):
        """Создает базу данных если она не существует"""
        connection = self.get_connection(use_database=False)
        if not connection:
            return False

        try:
            cursor = connection.cursor()
            database_name = self.db_config['database']

            # Создаем базу данных если она не существует
            cursor.execute(
                f"CREATE DATABASE IF NOT EXISTS {database_name} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
            logger.info(f"База данных '{database_name}' создана или уже существует")

            # Выбираем созданную базу данных
            cursor.execute(f"USE {database_name}")

            cursor.close()
            connection.close()
            return True

        except Error as e:
            logger.error(f"Ошибка создания базы данных: {e}")
            if connection.is_connected():
                cursor.close()
                connection.close()
            return False

    def create_tables(self):
        """Создает все таблицы в базе данных"""
        connection = self.get_connection()
        if not connection:
            return False

        try:
            cursor = connection.cursor()

            # Таблица групп
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS `group` (
                    id INT NOT NULL AUTO_INCREMENT,
                    group_name VARCHAR(100) NOT NULL,
                    course VARCHAR(100) NOT NULL,
                    institution VARCHAR(200) NOT NULL,
                    PRIMARY KEY (id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'group' создана или уже существует")

            # Таблица учебных материалов
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS educational_material (
                    id INT NOT NULL AUTO_INCREMENT,
                    class VARCHAR(50) NOT NULL,
                    pdf_link VARCHAR(200) NOT NULL,
                    pdf_first_page SMALLINT NOT NULL,
                    pdf_last_page INT NOT NULL,
                    edu_start DATE NOT NULL,
                    edu_end DATE NOT NULL,
                    PRIMARY KEY (id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'educational_material' создана или уже существует")

            # Таблица событий
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS events (
                    id INT NOT NULL AUTO_INCREMENT,
                    event_name VARCHAR(50) NOT NULL,
                    event_date DATE NOT NULL,
                    event_description VARCHAR(200) NOT NULL,
                    PRIMARY KEY (id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'events' создана или уже существует")

            # Таблица вопросов и ответов
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS faq_and_answers (
                    id INT NOT NULL AUTO_INCREMENT,
                    class VARCHAR(50) NOT NULL,
                    question VARCHAR(300) NOT NULL,
                    answer VARCHAR(500) NOT NULL,
                    PRIMARY KEY (id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'faq_and_answers' создана или уже существует")

            # Таблица студентов
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS student (
                    id INT NOT NULL AUTO_INCREMENT,
                    full_name VARCHAR(200) NOT NULL,
                    group_id INT NOT NULL,
                    integration_id VARCHAR(100) NOT NULL,
                    current_work_type VARCHAR(100) NOT NULL,
                    PRIMARY KEY (id),
                    FOREIGN KEY (group_id) REFERENCES `group`(id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'student' создана или уже существует")

            # Таблица успеваемости студентов
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS student_performance (
                    id INT NOT NULL AUTO_INCREMENT,
                    student_id INT NOT NULL,
                    grade SMALLINT NOT NULL,
                    comment VARCHAR(300) NOT NULL,
                    completeness TINYINT(1),
                    edmat_id INT NOT NULL,
                    PRIMARY KEY (id),
                    FOREIGN KEY (student_id) REFERENCES student(id),
                    FOREIGN KEY (edmat_id) REFERENCES educational_material(id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'student_performance' создана или уже существует")

            # Таблица связи учебных материалов с группами
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS edu_material_for_group (
                    id INT NOT NULL AUTO_INCREMENT,
                    edmat_id INT NOT NULL,
                    group_id INT NOT NULL,
                    PRIMARY KEY (id),
                    FOREIGN KEY (edmat_id) REFERENCES educational_material(id),
                    FOREIGN KEY (group_id) REFERENCES `group`(id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """)
            logger.info("Таблица 'edu_material_for_group' создана или уже существует")

            connection.commit()
            cursor.close()
            connection.close()

            logger.info("Все таблицы успешно созданы")
            return True

        except Error as e:
            logger.error(f"Ошибка создания таблиц: {e}")
            if connection.is_connected():
                cursor.close()
                connection.close()
            return False

    def insert_sample_data(self):
        """Вставляет тестовые данные в таблицы"""
        connection = self.get_connection()
        if not connection:
            return False

        try:
            cursor = connection.cursor()

            # Проверяем, есть ли уже данные в таблицах
            cursor.execute("SELECT COUNT(*) FROM `group`")
            if cursor.fetchone()[0] == 0:
                # Вставляем тестовые группы
                groups = [
                    ('Группа 1А', 'Математика', 'Университет Дементьева'),
                    ('Группа 2Б', 'Физика', 'Университет Дементьева'),
                    ('Группа 3В', 'Программирование', 'Университет Дементьева')
                ]
                cursor.executemany(
                    "INSERT INTO `group` (group_name, course, institution) VALUES (%s, %s, %s)",
                    groups
                )
                logger.info("Добавлены тестовые группы")

            # Проверяем студентов
            cursor.execute("SELECT COUNT(*) FROM student")
            if cursor.fetchone()[0] == 0:
                # Вставляем тестовых студентов
                students = [
                    ('Иванов Иван Иванович', 1, 'STU001', 'Лекция'),
                    ('Петров Петр Петрович', 1, 'STU002', 'Практика'),
                    ('Сидорова Мария Сергеевна', 2, 'STU003', 'Лекция'),
                    ('Кузнецов Алексей Владимирович', 3, 'STU004', 'Проект')
                ]
                cursor.executemany(
                    "INSERT INTO student (full_name, group_id, integration_id, current_work_type) VALUES (%s, %s, %s, %s)",
                    students
                )
                logger.info("Добавлены тестовые студенты")

            # Проверяем учебные материалы
            cursor.execute("SELECT COUNT(*) FROM educational_material")
            if cursor.fetchone()[0] == 0:
                # Вставляем тестовые учебные материалы
                materials = [
                    ('Математика', 'https://example.com/math1.pdf', 1, 10, '2024-01-15', '2024-06-15'),
                    ('Физика', 'https://example.com/physics1.pdf', 1, 8, '2024-02-01', '2024-07-01'),
                    ('Программирование', 'https://example.com/programming1.pdf', 1, 12, '2024-01-20', '2024-06-30')
                ]
                cursor.executemany(
                    "INSERT INTO educational_material (class, pdf_link, pdf_first_page, pdf_last_page, edu_start, edu_end) VALUES (%s, %s, %s, %s, %s, %s)",
                    materials
                )
                logger.info("Добавлены тестовые учебные материалы")

            # Проверяем связи материалов с группами
            cursor.execute("SELECT COUNT(*) FROM edu_material_for_group")
            if cursor.fetchone()[0] == 0:
                # Связываем материалы с группами
                material_groups = [
                    (1, 1),  # Математика для группы 1
                    (2, 2),  # Физика для группы 2
                    (3, 3)  # Программирование для группы 3
                ]
                cursor.executemany(
                    "INSERT INTO edu_material_for_group (edmat_id, group_id) VALUES (%s, %s)",
                    material_groups
                )
                logger.info("Добавлены связи материалов с группами")

            # Проверяем FAQ
            cursor.execute("SELECT COUNT(*) FROM faq_and_answers")
            if cursor.fetchone()[0] == 0:
                # Вставляем тестовые FAQ
                faqs = [
                    ('Математика', 'Как решить квадратное уравнение?',
                     'Квадратное уравнение решается через дискриминант: D = b² - 4ac'),
                    ('Физика', 'Что такое закон Ома?',
                     'Закон Ома: I = U/R, где I - сила тока, U - напряжение, R - сопротивление'),
                    ('Программирование', 'Что такое переменная?',
                     'Переменная - это именованная область памяти для хранения данных')
                ]
                cursor.executemany(
                    "INSERT INTO faq_and_answers (class, question, answer) VALUES (%s, %s, %s)",
                    faqs
                )
                logger.info("Добавлены тестовые FAQ")

            # Проверяем события
            cursor.execute("SELECT COUNT(*) FROM events")
            if cursor.fetchone()[0] == 0:
                # Вставляем тестовые события
                events = [
                    ('Экзамен по математике', '2024-06-20', 'Итоговый экзамен по курсу математики'),
                    ('Практика по программированию', '2024-06-25', 'Практическое занятие по написанию кода'),
                    ('Лекция по физике', '2024-06-18', 'Заключительная лекция по курсу физики')
                ]
                cursor.executemany(
                    "INSERT INTO events (event_name, event_date, event_description) VALUES (%s, %s, %s)",
                    events
                )
                logger.info("Добавлены тестовые события")

            connection.commit()
            cursor.close()
            connection.close()

            logger.info("Тестовые данные успешно добавлены")
            return True

        except Error as e:
            logger.error(f"Ошибка добавления тестовых данных: {e}")
            if connection.is_connected():
                cursor.close()
                connection.close()
            return False

    def deploy(self):
        """Основной метод развертывания базы данных"""
        logger.info("Начало развертывания базы данных...")

        if not self.create_database():
            logger.error("Не удалось создать базу данных")
            return False

        if not self.create_tables():
            logger.error("Не удалось создать таблицы")
            return False

        if not self.insert_sample_data():
            logger.warning("Не удалось добавить тестовые данные, но база данных развернута")

        logger.info("Развертывание базы данных завершено успешно!")
        return True

    def test_connection(self):
        """Тестирует подключение к базе данных"""
        connection = self.get_connection()
        if connection:
            cursor = connection.cursor()
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()
            logger.info(f"Подключение к MySQL установлено. Версия сервера: {version[0]}")
            cursor.close()
            connection.close()
            return True
        return False


def main():
    """Основная функция"""
    deployer = DatabaseDeployer()

    # Тестируем подключение
    if not deployer.test_connection():
        logger.error("Не удалось подключиться к MySQL серверу")
        return

    # Развертываем базу данных
    if deployer.deploy():
        logger.info("База данных успешно развернута и готова к использованию")
    else:
        logger.error("Произошла ошибка при развертывании базы данных")


if __name__ == "__main__":
    main()