https://pastein.ru/t/_3A
Загрузка данных
# -*- coding: utf-8 -*-
# project_database.py
import os
import sqlite3
from pathlib import Path
from typing import Optional
# Константы для имен таблиц. Это удобно для рефакторинга.
TABLE_PROJECT_DATA = "project_data"
TABLE_EVENT_LOGS = "event_logs"
class ProjectDatabase:
"""
Класс для управления базой данных проекта.
Создает файл БД с именем на основе кода проекта в указанной директории.
"""
def __init__(self, project_code: str, base_dir: str = "project_dbs"):
"""
Инициализирует объект для работы с БД проекта.
:param project_code: Уникальный код проекта (например, 'PRJ-001').
:param base_dir: Директория, где будут храниться файлы баз данных.
"""
self.project_code = self._sanitize_filename(project_code)
self.base_dir = base_dir
# Формируем путь к файлу БД, используя pathlib для кроссплатформенности
self.db_path = Path(base_dir) / f"{self.project_code}.db"
# Создаем директорию, если ее нет. Блок try/except здесь не нужен,
# так как exist_ok=True подавляет ошибки, если директория уже есть.
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.connection: Optional[sqlite3.Connection] = None
@staticmethod
def _sanitize_filename(name: str) -> str:
"""Очищает строку от символов, недопустимых в именах файлов."""
# Оставляем только буквы, цифры, подчеркивание, дефис и точку.
return "".join(c for c in name if c.isalnum() or c in "_-.").rstrip()
def connect(self) -> sqlite3.Connection:
"""Устанавливает соединение с базой данных. Создает файл, если его нет."""
if self.connection is None:
try:
self.connection = sqlite3.connect(self.db_path)
# print(f"Подключено к базе данных проекта: {self.db_path}")
except sqlite3.Error as e:
# print(f"Ошибка при подключении к БД: {e}")
raise
return self.connection
def close(self):
"""Закрывает соединение с базой данных."""
if self.connection:
self.connection.close()
self.connection = None
print(f"Соединение с {self.db_path} закрыто.")
def create_tables(self):
"""
Создает все необходимые таблицы.
Гарантирует наличие базовых колонок и добавляет новые поля для текущей логики.
"""
conn = self.connect()
cursor = conn.cursor()
# --- 1. СОЗДАЕМ ТАБЛИЦУ event_logs ---
cursor.execute("""CREATE TABLE IF NOT EXISTS event_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_code TEXT NOT NULL,
event_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
event_text TEXT NOT NULL
)""")
# --- 2. СОЗДАЕМ/ОБНОВЛЯЕМ ТАБЛИЦУ project_data ---
cursor.execute("""CREATE TABLE IF NOT EXISTS project_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_code TEXT NOT NULL,
name TEXT,
created_at TEXT NOT NULL,
description TEXT DEFAULT '',
date_opened TEXT DEFAULT '',
responsible TEXT DEFAULT '',
group_name TEXT,
group_kov REAL,
indicator_name TEXT,
indicator_code TEXT,
ivent_data DATE,
ivent_sign INTEGER,
ivent_code TEXT,
event_data DATE,
event_sign INTEGER,
event_code TEXT,
event_text_content TEXT,
group_code TEXT
)""")
# --- 3. ГАРАНТИРУЕМ НАЛИЧИЕ НОВЫХ КОЛОНОК (МИГРАЦИЯ) ---
cursor.execute(f"PRAGMA table_info({TABLE_PROJECT_DATA})")
existing_columns = [col[1] for col in cursor.fetchall()]
new_columns = ["group_code"]
for col_name in new_columns:
if col_name not in existing_columns:
cursor.execute(
f"ALTER TABLE {TABLE_PROJECT_DATA} ADD COLUMN {col_name} TEXT"
)
self._rename_ivent_columns_to_event(conn)
conn.commit()
def _rename_ivent_columns_to_event(self, conn: sqlite3.Connection):
"""
Внутренний метод для переименования столбцов ivent_* в event_*.
"""
cursor = conn.cursor()
columns_to_rename = [
("ivent_data", "event_data"),
("ivent_sign", "event_sign"),
("ivent_code", "event_code"),
]
for old_name, new_name in columns_to_rename:
try:
cursor.execute(
f"ALTER TABLE {TABLE_PROJECT_DATA} ADD COLUMN {new_name} TEXT"
)
cursor.execute(
f"UPDATE {TABLE_PROJECT_DATA} SET {new_name} = {old_name}"
)
cursor.execute(
f"ALTER TABLE {TABLE_PROJECT_DATA} DROP COLUMN {old_name}"
)
conn.commit()
except sqlite3.Error:
conn.rollback()
def _rename_ivent_columns_to_event(self, conn: sqlite3.Connection):
"""
Внутренний метод для переименования столбцов ivent_* в event_*.
SQLite не поддерживает ALTER COLUMN, поэтому нужно создавать новую колонку,
копировать данные и удалять старую.
"""
cursor = conn.cursor()
# Список кортежей (старая_колонка, новая_колонка)
columns_to_rename = [
("ivent_data", "event_data"),
("ivent_sign", "event_sign"),
("ivent_code", "event_code"),
]
for old_name, new_name in columns_to_rename:
try:
# 1. Добавляем новую колонку
cursor.execute(
f"ALTER TABLE {TABLE_PROJECT_DATA} ADD COLUMN {new_name} {
self._get_column_type(
cursor, TABLE_PROJECT_DATA, old_name)}"
)
# 2. Копируем данные из старой в новую (если тип данных
# совпадает)
cursor.execute(
f"UPDATE {TABLE_PROJECT_DATA} SET {new_name} = {old_name}"
)
# 3. Удаляем старую колонку
cursor.execute(
f"ALTER TABLE {TABLE_PROJECT_DATA} DROP COLUMN {old_name}"
)
conn.commit()
print(
f"[РЕФАКТОРИНГ] Колонка '{old_name}' успешно переименована в '{new_name}'."
)
except sqlite3.Error as e:
# Если колонки нет или другая ошибка, просто продолжаем
print(
f"[ПРЕДУПРЕЖДЕНИЕ] Не удалось переименовать '{old_name}': {e}. Возможно, колонка уже была переименована или отсутствует."
)
conn.rollback()
@staticmethod
def _get_column_type(cursor, table_name, column_name):
"""Вспомогательный метод для получения типа колонки (в данном случае всегда TEXT/DATE/INTEGER)."""
# В этом конкретном случае мы знаем типы, но метод оставлен для примера.
# Для простоты возвращаем пустую строку, так как ALTER TABLE в SQLite
# не требует тип при ADD COLUMN.
return ""
def delete_database_file(self):
"""
Полностью удаляет файл базы данных с диска.
Используется при завершении или архивации проекта.
"""
self.close() # Сначала закрываем соединение
if self.db_path.exists():
try:
self.db_path.unlink() # Используем метод Path.unlink()
# print(f"Файл базы данных {self.db_path} удален.")
except OSError as e:
print(f"Ошибка при удалении файла БД: {e}")
