src/models/database.py - master - kurzoid/gaugematch

src/models/database.py
531 строка21 KB
kurzoid
окно редактирования сведений об уровнемерах
25 окт 2025, 20:01
25 окт 2025, 20:01f40293d
import sqlite3
import os
from typing import List, Dict, Any


class Database:
    def __init__(self, db_path=None):
        # Мультиплатформенный путь к базе данных
        if db_path is None:
            # Определяем базовую директорию проекта
            base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
            self.db_path = os.path.join(base_dir, 'data', 'database.db')
        else:
            self.db_path = db_path

        self._ensure_data_directory()
        self._verify_database()

    def _ensure_data_directory(self):
        """Проверяем существование папки data и создаем если нужно"""
        data_dir = os.path.dirname(self.db_path)
        if not os.path.exists(data_dir):
            os.makedirs(data_dir)
            print(f"Создана папка: {data_dir}")

    def _verify_database(self):
        """Проверяем существование базы данных и таблиц"""
        if not os.path.exists(self.db_path):
            raise FileNotFoundError(f"База данных не найдена: {self.db_path}")

        print(f"Подключение к базе данных: {self.db_path}")

        # Проверяем существование основных таблиц
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            tables_to_check = ['reservoirs', 'performers', 'control_device', 'level_meters']
            for table in tables_to_check:
                cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'")
                if not cursor.fetchone():
                    raise Exception(f"Таблица '{table}' не найдена в базе данных")
                print(f"Таблица '{table}' найдена")

        finally:
            conn.close()

    def _get_connection(self):
        """Получение соединения с базой данных"""
        return sqlite3.connect(self.db_path)

    # Методы для таблицы reservoirs
    def get_all_reservoirs(self) -> List[Dict[str, Any]]:
        """Получение всех записей из таблицы reservoirs"""
        conn = self._get_connection()
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        try:
            cursor.execute("SELECT * FROM reservoirs ORDER BY tech_number")
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            print(f"Загружено записей из reservoirs: {len(result)}")
            return result
        except Exception as e:
            print(f"Ошибка при загрузке данных reservoirs: {e}")
            raise
        finally:
            conn.close()

    def insert_reservoir(self, data: Dict[str, Any]) -> int:
        """Добавление новой записи в reservoirs"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                INSERT INTO reservoirs (name, model, factory_number, tech_number, base_height)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                data['tech_number'].strip(),
                float(data['base_height'])
            ))

            reservoir_id = cursor.lastrowid
            conn.commit()
            print(f"Добавлен резервуар ID: {reservoir_id}")
            return reservoir_id
        except sqlite3.IntegrityError as e:
            conn.rollback()
            print(f"Ошибка целостности при добавлении резервуара: {e}")
            raise
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при добавлении резервуара: {e}")
            raise
        finally:
            conn.close()

    def update_reservoir(self, reservoir_id: int, data: Dict[str, Any]) -> bool:
        """Обновление записи в reservoirs"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                UPDATE reservoirs 
                SET name = ?, model = ?, factory_number = ?, tech_number = ?, base_height = ?
                WHERE id = ?
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                data['tech_number'].strip(),
                float(data['base_height']),
                reservoir_id
            ))

            affected = cursor.rowcount
            conn.commit()
            print(f"Обновлен резервуар ID: {reservoir_id}, затронуто строк: {affected}")
            return affected > 0
        except sqlite3.IntegrityError as e:
            conn.rollback()
            print(f"Ошибка целостности при обновлении резервуара: {e}")
            raise
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при обновлении резервуара: {e}")
            raise
        finally:
            conn.close()

    def delete_reservoir(self, reservoir_id: int) -> bool:
        """Удаление записи из reservoirs"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute("DELETE FROM reservoirs WHERE id = ?", (reservoir_id,))
            affected = cursor.rowcount
            conn.commit()
            print(f"Удален резервуар ID: {reservoir_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при удалении резервуара: {e}")
            raise
        finally:
            conn.close()

    # Методы для таблицы performers
    def get_all_performers(self) -> List[Dict[str, Any]]:
        """Получение всех записей из таблицы performers"""
        conn = self._get_connection()
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        try:
            cursor.execute("SELECT * FROM performers ORDER BY surname, name")
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            print(f"Загружено записей из performers: {len(result)}")
            return result
        except Exception as e:
            print(f"Ошибка при загрузке данных performers: {e}")
            raise
        finally:
            conn.close()

    def insert_performer(self, data: Dict[str, Any]) -> int:
        """Добавление новой записи в performers"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                INSERT INTO performers (post, surname, name, patronymic)
                VALUES (?, ?, ?, ?)
            ''', (
                data['post'].strip(),
                data['surname'].strip(),
                data['name'].strip(),
                data.get('patronymic', '').strip()
            ))

            performer_id = cursor.lastrowid
            conn.commit()
            print(f"Добавлен исполнитель ID: {performer_id}")
            return performer_id
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при добавлении исполнителя: {e}")
            raise
        finally:
            conn.close()

    def update_performer(self, performer_id: int, data: Dict[str, Any]) -> bool:
        """Обновление записи в performers"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                UPDATE performers 
                SET post = ?, surname = ?, name = ?, patronymic = ?
                WHERE id = ?
            ''', (
                data['post'].strip(),
                data['surname'].strip(),
                data['name'].strip(),
                data.get('patronymic', '').strip(),
                performer_id
            ))

            affected = cursor.rowcount
            conn.commit()
            print(f"Обновлен исполнитель ID: {performer_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при обновлении исполнителя: {e}")
            raise
        finally:
            conn.close()

    def delete_performer(self, performer_id: int) -> bool:
        """Удаление записи из performers"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute("DELETE FROM performers WHERE id = ?", (performer_id,))
            affected = cursor.rowcount
            conn.commit()
            print(f"Удален исполнитель ID: {performer_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при удалении исполнителя: {e}")
            raise
        finally:
            conn.close()

    # Методы для таблицы control_device
    def get_all_control_devices(self) -> List[Dict[str, Any]]:
        """Получение всех записей из таблицы control_device"""
        conn = self._get_connection()
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        try:
            cursor.execute("SELECT * FROM control_device ORDER BY name")
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            print(f"Загружено записей из control_device: {len(result)}")
            return result
        except Exception as e:
            print(f"Ошибка при загрузке данных control_device: {e}")
            raise
        finally:
            conn.close()

    def insert_control_device(self, data: Dict[str, Any]) -> int:
        """Добавление новой записи в control_device"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                INSERT INTO control_device (name, model, factory_number, verification_date, 
                                         verification_interval, certificate_number, thermal_coefficient)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                data.get('verification_date', '').strip(),
                data.get('verification_interval', '').strip(),
                data.get('certificate_number', '').strip(),
                float(data.get('thermal_coefficient', 0))
            ))

            device_id = cursor.lastrowid
            conn.commit()
            print(f"Добавлено контрольное СИ ID: {device_id}")
            return device_id
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при добавлении контрольного СИ: {e}")
            raise
        finally:
            conn.close()

    def update_control_device(self, device_id: int, data: Dict[str, Any]) -> bool:
        """Обновление записи в control_device"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute('''
                UPDATE control_device 
                SET name = ?, model = ?, factory_number = ?, verification_date = ?,
                    verification_interval = ?, certificate_number = ?, thermal_coefficient = ?
                WHERE id = ?
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                data.get('verification_date', '').strip(),
                data.get('verification_interval', '').strip(),
                data.get('certificate_number', '').strip(),
                float(data.get('thermal_coefficient', 0)),
                device_id
            ))

            affected = cursor.rowcount
            conn.commit()
            print(f"Обновлено контрольное СИ ID: {device_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при обновлении контрольного СИ: {e}")
            raise
        finally:
            conn.close()

    def delete_control_device(self, device_id: int) -> bool:
        """Удаление записи из control_device"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute("DELETE FROM control_device WHERE id = ?", (device_id,))
            affected = cursor.rowcount
            conn.commit()
            print(f"Удалено контрольное СИ ID: {device_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при удалении контрольного СИ: {e}")
            raise
        finally:
            conn.close()

    # Методы для таблицы level_meters
    def get_all_level_meters(self) -> List[Dict[str, Any]]:
        """Получение всех записей из таблицы level_meters с данными о резервуарах"""
        conn = self._get_connection()
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        try:
            cursor.execute('''
                SELECT lm.*, r.tech_number as reservoir_tech_number
                FROM level_meters lm
                LEFT JOIN reservoirs r ON lm.reservoir_id = r.id
                ORDER BY lm.name
            ''')
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            print(f"Загружено записей из level_meters: {len(result)}")
            return result
        except Exception as e:
            print(f"Ошибка при загрузке данных level_meters: {e}")
            raise
        finally:
            conn.close()

    def get_all_reservoirs_for_combobox(self) -> List[Dict[str, Any]]:
        """Получение списка резервуаров для Combobox"""
        conn = self._get_connection()
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        try:
            cursor.execute("SELECT id, tech_number, name FROM reservoirs ORDER BY tech_number")
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            return result
        except Exception as e:
            print(f"Ошибка при загрузке данных резервуаров для Combobox: {e}")
            raise
        finally:
            conn.close()

    def insert_level_meter(self, data: Dict[str, Any]) -> int:
        """Добавление новой записи в level_meters"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            # Получаем reservoir_id по tech_number
            reservoir_id = None
            if data.get('reservoir_tech_number'):
                cursor.execute("SELECT id FROM reservoirs WHERE tech_number = ?",
                               (data['reservoir_tech_number'],))
                result = cursor.fetchone()
                if result:
                    reservoir_id = result[0]

            cursor.execute('''
                INSERT INTO level_meters (name, model, factory_number, reservoir_id)
                VALUES (?, ?, ?, ?)
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                reservoir_id
            ))

            device_id = cursor.lastrowid
            conn.commit()
            print(f"Добавлен уровнемер ID: {device_id}")
            return device_id
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при добавлении уровнемера: {e}")
            raise
        finally:
            conn.close()

    def update_level_meter(self, device_id: int, data: Dict[str, Any]) -> bool:
        """Обновление записи в level_meters"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            # Получаем reservoir_id по tech_number
            reservoir_id = None
            if data.get('reservoir_tech_number'):
                cursor.execute("SELECT id FROM reservoirs WHERE tech_number = ?",
                               (data['reservoir_tech_number'],))
                result = cursor.fetchone()
                if result:
                    reservoir_id = result[0]

            cursor.execute('''
                UPDATE level_meters 
                SET name = ?, model = ?, factory_number = ?, reservoir_id = ?
                WHERE id = ?
            ''', (
                data['name'].strip(),
                data.get('model', '').strip(),
                data.get('factory_number', '').strip(),
                reservoir_id,
                device_id
            ))

            affected = cursor.rowcount
            conn.commit()
            print(f"Обновлен уровнемер ID: {device_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при обновлении уровнемера: {e}")
            raise
        finally:
            conn.close()

    def delete_level_meter(self, device_id: int) -> bool:
        """Удаление записи из level_meters"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute("DELETE FROM level_meters WHERE id = ?", (device_id,))
            affected = cursor.rowcount
            conn.commit()
            print(f"Удален уровнемер ID: {device_id}, затронуто строк: {affected}")
            return affected > 0
        except Exception as e:
            conn.rollback()
            print(f"Ошибка при удалении уровнемера: {e}")
            raise
        finally:
            conn.close()

    def close(self):
        """Закрытие соединения с базой данных"""
        pass

    def get_table_structure(self, table_name: str) -> List[Dict[str, str]]:
        """Получение структуры таблицы"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            cursor.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()
            return [{'name': col[1], 'type': col[2]} for col in columns]
        finally:
            conn.close()

    def get_database_info(self) -> Dict[str, Any]:
        """Получение информации о базе данных"""
        conn = self._get_connection()
        cursor = conn.cursor()

        try:
            # Информация о таблицах
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
            tables = [row[0] for row in cursor.fetchall()]

            # Количество записей в таблицах
            cursor.execute("SELECT COUNT(*) FROM reservoirs")
            reservoirs_count = cursor.fetchone()[0]

            cursor.execute("SELECT COUNT(*) FROM performers")
            performers_count = cursor.fetchone()[0]

            cursor.execute("SELECT COUNT(*) FROM control_device")
            control_devices_count = cursor.fetchone()[0]

            cursor.execute("SELECT COUNT(*) FROM level_meters")
            level_meters_count = cursor.fetchone()[0]

            return {
                'path': self.db_path,
                'tables': tables,
                'reservoirs_count': reservoirs_count,
                'performers_count': performers_count,
                'control_devices_count': control_devices_count,
                'level_meters_count': level_meters_count,
                'file_size': os.path.getsize(self.db_path) if os.path.exists(self.db_path) else 0
            }
        finally:
            conn.close()