src/db_manager.py - master - kuzminaa/PDM

src/db_manager.py
185 строк8 KB
kuzminaa
refactor: migrate modules to src/, add logging system, add assembly workshop
29 июл 2026, 23:52
29 июл 2026, 23:52a5323aa
import os
import sqlite3
import datetime
import hashlib
import shutil
import logging
from .paths import DB_NAME, STORAGE_DIR, ensure_storage_dir

logger = logging.getLogger(__name__)

def init_db():
    """Инициализирует таблицы базы данных."""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            filename TEXT, filepath TEXT, checksum TEXT,
            dev_fio TEXT, dev_date TEXT, chk_fio TEXT, chk_date TEXT,
            ctl_fio TEXT, ctl_date TEXT, app_fio TEXT, app_date TEXT,
            designation TEXT, title TEXT, version INTEGER, last_change INTEGER
        )
    """)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS signings (
            id INTEGER PRIMARY KEY AUTOINCREMENT, doc_id INTEGER,
            agr_fio TEXT, agr_date TEXT,
            FOREIGN KEY (doc_id) REFERENCES documents(id) ON DELETE CASCADE
        )
    """)
    conn.commit()
    conn.close()

def generate_paths(designation, source_pdf_path):
    if not designation or not source_pdf_path: return "", ""
    clean_des = "".join(c for c in designation if c.isalnum() or c in ".-_ ")
    filename = f"{clean_des}.pdf"
    ensure_storage_dir()
    return filename, os.path.join(STORAGE_DIR, filename)

def calculate_md5(source_path):
    if not os.path.exists(source_path): return ""
    md5_hash = hashlib.md5()
    with open(source_path, "rb") as f:
        for byte_block in iter(lambda: f.read(4096), b""):
            md5_hash.update(byte_block)
    return md5_hash.hexdigest()

def check_duplicates(designation, checksum):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("SELECT designation, checksum FROM documents WHERE designation = ? OR checksum = ?", (designation, checksum))
    row = cursor.fetchone()
    conn.close()
    if row:
        if row[0] == designation:
            return f"Документ с Обозначением '{designation}' уже зарегистрирован!"
        else:
            return "Этот файл уже добавлялся под другим Обозначением (совпадает MD5)!"
    return None
def check_duplicates_v2(designation, change_number, checksum):
    """
    Проверяет дубликаты с учетом контроля версий.
    Разрешает одинаковые Обозначения, если у них разные Изменения.
    """
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    
    # 1. Проверка: нет ли точно такого же файла (по хэшу)
    cursor.execute("SELECT designation, last_change FROM documents WHERE checksum = ?", (checksum,))
    row_hash = cursor.fetchone()
    if row_hash:
        conn.close()
        return f"Этот файл уже добавлен в базу под обозначением {row_hash[0]} (Изм. {row_hash[1]})"
        
    # 2. Проверка: нет ли уже документа с ТАКИМ ЖЕ Обозначением и ТАКИМ ЖЕ номером Изменения
    cursor.execute("SELECT id FROM documents WHERE designation = ? AND last_change = ?", (designation, change_number))
    row_ver = cursor.fetchone()
    conn.close()
    
    if row_ver:
        return f"Документ {designation} с Изменением {change_number} уже существует! Используйте замещение или выпустите следующее изменение."
        
    return None

def insert_document(main_data, agreements):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    try:
        cursor.execute("""
            INSERT INTO documents (
                filename, filepath, checksum, dev_fio, dev_date, chk_fio, chk_date,
                ctl_fio, ctl_date, app_fio, app_date, designation, title, version, last_change
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            main_data["filename"], main_data["filepath"], main_data["checksum"],
            main_data["dev_fio"], main_data["dev_date"], main_data["chk_fio"], main_data["chk_date"],
            main_data["ctl_fio"], main_data["ctl_date"], main_data["app_fio"], main_data["app_date"],
            main_data["designation"], main_data["title"], main_data["version"], main_data["last_change"]
        ))
        doc_id = cursor.lastrowid
        logger.info(f"Запись документа: id={doc_id} '{main_data['designation']}' v{main_data['version']}")
        for agr in agreements:
            cursor.execute("INSERT INTO signings (doc_id, agr_fio, agr_date) VALUES (?, ?, ?)", 
                           (doc_id, agr["fio"], agr["date"]))
        conn.commit()
        try:
            shutil.copy2(main_data["source_path"], main_data["filepath"])
        except Exception as copy_err:
            conn.rollback()
            raise RuntimeError(f"Ошибка копирования файла: {copy_err}") from copy_err
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

def validate_date(date_str):
    try:
        datetime.datetime.strptime(date_str, "%Y-%m-%d")
        return True
    except ValueError: return False

def get_all_documents():
    """Возвращает список всех документов для таблицы главного окна."""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    # Выбираем ключевые поля для вывода в архив
    cursor.execute("""
        SELECT id, designation, title, version, last_change, filename, filepath 
        FROM documents 
        ORDER BY designation ASC, last_change DESC
    """)
    rows = cursor.fetchall()
    conn.close()
    return rows


def get_document_full_info(doc_id):
    """Возвращает полную информацию о документе и список согласований."""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT id, filename, filepath, checksum, dev_fio, dev_date, chk_fio, chk_date,
               ctl_fio, ctl_date, app_fio, app_date, designation, title, version, last_change
        FROM documents WHERE id = ?
    """, (doc_id,))
    row = cursor.fetchone()
    conn.close()
    
    if not row:
        return None, []
    
    main_data = {
        "id": row[0], "filename": row[1], "filepath": row[2], "checksum": row[3],
        "dev_fio": row[4], "dev_date": row[5], "chk_fio": row[6], "chk_date": row[7],
        "ctl_fio": row[8], "ctl_date": row[9], "app_fio": row[10], "app_date": row[11],
        "designation": row[12], "title": row[13], "version": row[14], "last_change": row[15]
    }
    
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("SELECT agr_fio, agr_date FROM signings WHERE doc_id = ?", (doc_id,))
    agreements = [{"fio": r[0], "date": r[1]} for r in cursor.fetchall()]
    conn.close()
    
    return main_data, agreements


def extract_file_to_user(source_path, filename):
    """Экспортирует PDF файл из архива в выбранный пользователем каталог."""
    import tkinter.filedialog as fd
    import tkinter.messagebox as mb
    
    output_dir = fd.askdirectory(title="Выберите каталог для сохранения файла")
    if not output_dir:
        return False, "Операция отменена"
    
    dest_path = os.path.join(output_dir, filename)
    try:
        shutil.copy2(source_path, dest_path)
        return True, f"Файл успешно экспортирован:\n{dest_path}"
    except Exception as e:
        return False, f"Ошибка при экспорте файла: {e}"