https://pastein.ru/t/eEJ
Загрузка данных
# -*- coding: utf-8 -*-
# -*- database_module.py -*-
import logging
import numpy as np
from functools import wraps
from sqlalchemy import (
Column,
Float,
ForeignKey,
Integer,
String,
create_engine,
func,
inspect,
)
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import (
declarative_base,
relationship,
sessionmaker,
backref,
selectinload # <-- Добавлена необходимая функция
)
from sqlalchemy.sql.expression import text
# Настройки логирования
logging.basicConfig(
level=logging.INFO, # Меняем здесь
format="%(asctime)s [%(levelname)s]: %(message)s"
)
logger = logging.getLogger(__name__)
# Заглушаем чрезмерные логи от сторонних библиотек
logging.getLogger("sqlalchemy").propagate = False
# дополнительно отключаем NumExpr
logging.getLogger("numexpr").propagate = False
# Настройка движка базы данных
DATABASE_PATH = "database.db"
engine = create_engine(f"sqlite:///{DATABASE_PATH}")
Base = declarative_base()
Session = sessionmaker(bind=engine)
# Определение моделей
class Group(Base):
__tablename__ = "groups"
id = Column(Integer, primary_key=True)
name = Column(String)
code = Column(String, unique=True) # Уникальный код группы
kov = Column(Float, default=None) # Коэффициент KOV группы
class Indicator(Base):
__tablename__ = "indicators"
id = Column(Integer, primary_key=True)
name = Column(String)
code = Column(String, unique=True) # Код индикатора
group_code = Column(String, ForeignKey("groups.code"))
# Связь с родительской группой
group = relationship("Group", backref="indicators", foreign_keys=[group_code])
# --- ДОБАВЬТЕ ЭТУ СТРОКУ ---
matrices = relationship("MatrixEntry", back_populates="indicator")
kov = Column(Float, default=None)
class MatrixEntry(Base):
__tablename__ = "matrices"
id = Column(Integer, primary_key=True)
# Связь ТОЛЬКО с индикатором
indicator_id = Column(Integer, ForeignKey("indicators.id"), nullable=False)
indicator = relationship("Indicator", back_populates="matrices")
row = Column(Integer, nullable=False)
col = Column(Integer, nullable=False)
value = Column(Float, nullable=True)
def __repr__(self):
return f"<MatrixEntry({self.row}, {self.col}, {self.value})>"
# Инициализация базы данных
def initialize_database():
logger.info("Инициализация базы данных...")
# --- ДОБАВЬТЕ ЭТИ СТРОКИ ---
print("[DEBUG] Вызвана функция initialize_database()")
print(f"[DEBUG] Путь к движку: {engine.url}")
# Проверим, видит ли Python наши классы моделей перед созданием
print(f"[DEBUG] Зарегистрированные таблицы: {Base.metadata.tables.keys()}")
# ------------------------------------
Base.metadata.create_all(engine)
logger.info("База данных создана успешно.")
# Функция миграции (добавляет недостающие колонки)
def migrate_database():
logger.info("Начало миграции базы данных...")
inspector = inspect(engine)
tables = inspector.get_table_names()
# Если таблица 'groups' отсутствует, создаём её
if "groups" not in tables:
logger.warning("'groups' table does not exist. Creating it now.")
Base.metadata.tables["groups"].create(engine)
# Проверка наличия колонки 'kov'
columns = inspector.get_columns("groups")
column_names = [column["name"] for column in columns]
if "kov" not in column_names:
logger.warning("Column 'kov' doesn't exist. Adding it now.")
with engine.connect() as conn:
conn.execute(text("ALTER TABLE groups ADD COLUMN kov FLOAT DEFAULT NULL;"))
logger.info("Migration completed successfully.")
else:
logger.info("Column 'kov' already exists. Skipping migration.")
# CRUD операции для групп
def add_group(name, code, kov=None):
"""Добавление новой группы"""
session = Session()
new_group = Group(name=name, code=code, kov=kov)
session.add(new_group)
session.commit()
return new_group
def get_group(code):
"""Получение группы по коду"""
session = Session()
return session.query(Group).filter_by(code=code).first()
def update_group(code, new_name=None, new_kov=None):
"""Обновление группы"""
session = Session()
group = session.query(Group).filter_by(code=code).first()
if group:
if new_name:
group.name = new_name
if new_kov is not None:
group.kov = new_kov
session.commit()
return group
return None
def delete_group(code):
"""Удаление группы"""
session = Session()
group = session.query(Group).filter_by(code=code).first()
if group:
session.delete(group)
session.commit()
return True
return False
# CRUD операции для индикаторов
def insert_indicator(name, code, group_code):
"""Вставка нового индикатора"""
session = Session()
new_indicator = Indicator(name=name, code=code, group_code=group_code)
session.add(new_indicator)
session.commit()
return new_indicator
def get_indicator(code):
"""Получение индикатора по коду"""
session = Session()
return session.query(Indicator).filter_by(code=code).first()
def update_indicator(code, new_name=None, new_group_code=None):
"""Обновление индикатора"""
session = Session()
indicator = session.query(Indicator).filter_by(code=code).first()
if indicator:
if new_name:
indicator.name = new_name
if new_group_code:
indicator.group_code = new_group_code
session.commit()
return indicator
return None
def delete_indicator_by_code(code):
"""Удаление индикатора по его уникальному коду"""
session = Session()
try:
indicator = session.query(Indicator).filter_by(code=code).first()
if indicator:
session.delete(indicator)
session.commit()
return True
return False
except Exception as e:
# Если что-то пошло не так, откатываем изменения
session.rollback()
print(f"Ошибка при удалении: {e}") # Или logger.error()
return False
finally:
# Эта строка выполнится ВСЕГДА, даже если была ошибка
session.close()
# Функция для получения всех индикаторов по группе (по уникальному коду)
def get_all_indicators_by_group(group_code, session=None):
"""Возвращает все индикаторы для группы по её коду."""
close_session = False
if session is None:
session = Session()
close_session = True
try:
indicators = session.query(Indicator).filter_by(group_code=group_code).all()
return indicators
finally:
if close_session:
session.close()
# Функция для получения всех индикаторов по группе (по внутреннему ID)
# modules/database_module.py
def list_groups():
"""
Получает все группы вместе с индикаторами и матрицами за один запрос.
Использует готовую фабрику Session из этого модуля.
"""
db = Session() # Создаём сессию через нашу фабрику
try:
groups = (
db.query(Group) # Запрашиваем таблицу Group
.options(
selectinload(Group.indicators) # Говорим: "загрузи сразу индикаторы"
.selectinload(Indicator.matrices) # И для каждого индикатора загрузи матрицы
)
.all()
)
return groups
except Exception as e:
logging.error(f"[DATABASE] Ошибка загрузки групп: {e}")
return []
finally:
db.close() # Обязательно закрываем сессию после работы
def list_indicators(group_code):
"""
Возвращает индикаторы конкретной группы со всеми связями.
"""
session = Session()
try:
indicators = session.query(Indicator).filter_by(group_code=group_code).options(
selectinload(Indicator.group),
selectinload(Indicator.matrices)
).all()
return indicators
finally:
session.close()
# Получение матричных записей
def get_matrix_entries(session, group_id):
return session.query(MatrixEntry).filter_by(group_id=group_id).all()
def update_matrix_value(session, group_id, row_index, col_index, value):
"""
Обновляем значение ячейки матрицы.
:param session: Текущая сессия SQLAlchemy
:param group_id: Идентификатор группы
:param row_index: Индекс строки
:param col_index: Индекс столбца
:param value: Новое значение ячейки
"""
entry = (
session.query(MatrixEntry)
.filter_by(group_id=group_id, row=row_index, col=col_index)
.first()
)
if entry:
entry.value = value
else:
new_entry = MatrixEntry(
group_id=group_id, row=row_index, col=col_index, value=value
)
session.add(new_entry)
session.commit()
def calculate_indicator_weights(matrix_entries):
"""
Рассчитывает вектор коэффициентов относительной важности (КОВ/весов)
для каждого индикатора на основе матрицы парных сравнений.
:param matrix_entries: Список объектов MatrixEntry для одной группы.
:return: Список весов (float), где индекс соответствует индексу индикатора.
"""
if not matrix_entries:
return None
# Определяем размерность матрицы (n x n)
max_row = max(entry.row for entry in matrix_entries)
max_col = max(entry.col for entry in matrix_entries)
n = max(max_row, max_col) + 1
# Создаем матрицу, заполненную нулями
matrix = np.zeros((n, n))
# Заполняем матрицу значениями из базы
for entry in matrix_entries:
i, j, value = entry.row, entry.col, entry.value
if value is not None and value > 0:
matrix[i][j] = value
# Для матрицы сравнений добавляем обратное значение
if i != j:
matrix[j][i] = 1 / value
# Нормируем столбцы
col_sums = matrix.sum(axis=0)
# Избегаем деления на ноль
col_sums[col_sums == 0] = 1
normalized_matrix = matrix / col_sums
# Вычисляем среднее по строкам - это и есть веса индикаторов
weights_vector = normalized_matrix.mean(axis=1)
return weights_vector.tolist()
# --- НАЧАЛО БЛОКА ДЛЯ ДОБАВЛЕНИЯ ТЕСТОВЫХ ДАННЫХ ---
def populate_test_data():
"""
Заполняет базу данных тестовыми данными, если она пуста.
"""
session = Session()
try:
# Проверяем наличие групп напрямую через таблицу
group_count = session.query(Group).count()
if group_count == 0:
print("[INFO] База данных пуста. Начинаем добавление...")
# Создаем объекты в памяти
g1 = Group(name="Фанаты", code="ФАН", kov=None)
g2 = Group(name="Болельщики", code="БОЛ", kov=None)
i1 = Indicator(name="Прибыль", code="IND-PROFIT-1", group_code="ФАН", kov=None)
i2 = Indicator(name="Риск", code="IND-RISK-1", group_code="ФАН", kov=None)
i3 = Indicator(name="Лояльность", code="IND-LOYALTY-1", group_code="БОЛ", kov=None)
i4 = Indicator(name="Охват", code="IND-REACH-1", group_code="БОЛ", kov=None)
# Добавляем всё в очередь сессии
session.add_all([g1, g2, i1, i2, i3, i4])
# ОДИН ЕДИНСТВЕННЫЙ КОМИТ В КОНЦЕ
session.commit()
print("[SUCCESS] Тестовые данные успешно записаны в базу.")
else:
print("[INFO] Данные уже есть в базе. Пропускаем вставку.")
except Exception as e:
# Если хоть одна строчка упала — отменяем все изменения
session.rollback()
print(f"[ERROR] Критическая ошибка базы данных: {e}")
finally:
# Всегда закрываем соединение
session.close()
if __name__ == "__main__":
from modules.database_module import populate_test_data, Session, Group
# Сначала принудительно заполняем базу
print("[ACTION] Ручной запуск наполнения данными...")
populate_test_data()
# А теперь проверяем, что там есть
session = Session()
count = session.query(Group).count()
print(f"[RESULT] После ручного запуска в базе найдено групп: {count}")
session.close()
# И только потом открываем интерфейс
root = tk.Tk()
app = MainWindow(root)
root.mainloop()