database.py - main - Paff/declarant

database.py
228 строк9 KB
1pash1985-gif
Декларант - веб-версия для Timeweb Cloud
15 июл 2026, 09:36
15 июл 2026, 09:36a6b08b8
"""
Менеджер базы данных SQLite для хранения сессий и конфигурации.
"""
import sqlite3
import json
import os
from datetime import datetime
from typing import Optional

# Путь к базе данных
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "declarant.db")


class DatabaseManager:
    """Менеджер для работы с SQLite базой данных."""

    def __init__(self, db_path: str = DB_PATH):
        self.db_path = db_path
        self.conn: Optional[sqlite3.Connection] = None

    def initialize(self):
        """Инициализировать базу данных (создать таблицы)."""
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA foreign_keys = ON")
        
        # Создать таблицы
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS base_config (
                id INTEGER PRIMARY KEY DEFAULT 1,
                xlsm_path TEXT NOT NULL,
                flowers TEXT NOT NULL,
                clients TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            
            CREATE TABLE IF NOT EXISTS sessions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                date TEXT NOT NULL,
                invoice_folder TEXT,
                status TEXT DEFAULT 'draft',
                total_boxes INTEGER DEFAULT 0
            );
            
            CREATE TABLE IF NOT EXISTS session_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id INTEGER NOT NULL,
                flower_name TEXT NOT NULL,
                client_name TEXT NOT NULL,
                boxes_count INTEGER DEFAULT 0,
                FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
            );
        """)
        self.conn.commit()

    def close(self):
        """Закрыть соединение с базой данных."""
        if self.conn:
            self.conn.close()

    # === Base Config ===

    def has_base_config(self) -> bool:
        """Проверить, есть ли базовая конфигурация."""
        cursor = self.conn.execute("SELECT COUNT(*) FROM base_config WHERE id = 1")
        return cursor.fetchone()[0] > 0

    def get_base_config(self) -> Optional[dict]:
        """Получить базовую конфигурацию."""
        cursor = self.conn.execute(
            "SELECT xlsm_path, flowers, clients, created_at FROM base_config WHERE id = 1"
        )
        row = cursor.fetchone()
        if row:
            return {
                "xlsm_path": row["xlsm_path"],
                "flowers": json.loads(row["flowers"]),
                "clients": json.loads(row["clients"]),
                "created_at": row["created_at"]
            }
        return None

    def save_base_config(self, xlsm_path: str, flowers: list, clients: list):
        """Сохранить базовую конфигурацию."""
        now = datetime.now().isoformat()
        flowers_json = json.dumps(flowers, ensure_ascii=False)
        clients_json = json.dumps(clients, ensure_ascii=False)
        
        with self.conn:
            self.conn.execute("""
                INSERT OR REPLACE INTO base_config (id, xlsm_path, flowers, clients, created_at)
                VALUES (1, ?, ?, ?, ?)
            """, (xlsm_path, flowers_json, clients_json, now))

    # === Sessions ===

    def create_session(self, name: str, invoice_folder: str = "") -> int:
        """Создать новую сессию. Возвращает ID сессии."""
        now = datetime.now().isoformat()
        with self.conn:
            cursor = self.conn.execute("""
                INSERT INTO sessions (name, date, invoice_folder, status, total_boxes)
                VALUES (?, ?, ?, 'draft', 0)
            """, (name, now, invoice_folder))
            return cursor.lastrowid

    def get_session(self, session_id: int) -> Optional[dict]:
        """Получить сессию по ID."""
        cursor = self.conn.execute(
            "SELECT * FROM sessions WHERE id = ?", (session_id,)
        )
        row = cursor.fetchone()
        if row:
            return dict(row)
        return None

    def get_all_sessions(self) -> list:
        """Получить все сессии (сортировка по дате, новые первые)."""
        cursor = self.conn.execute(
            "SELECT * FROM sessions ORDER BY date DESC"
        )
        return [dict(row) for row in cursor.fetchall()]

    def update_session_status(self, session_id: int, status: str):
        """Обновить статус сессии."""
        with self.conn:
            self.conn.execute(
                "UPDATE sessions SET status = ? WHERE id = ?",
                (status, session_id)
            )

    def update_session_total_boxes(self, session_id: int, total: int):
        """Обновить общее количество коробок в сессии."""
        with self.conn:
            self.conn.execute(
                "UPDATE sessions SET total_boxes = ? WHERE id = ?",
                (total, session_id)
            )

    def update_session_folder(self, session_id: int, folder: str):
        """Обновить папку инвойсов в сессии."""
        with self.conn:
            self.conn.execute(
                "UPDATE sessions SET invoice_folder = ? WHERE id = ?",
                (folder, session_id)
            )

    def delete_session(self, session_id: int):
        """Удалить сессию (включая данные)."""
        with self.conn:
            self.conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))

    # === Session Data ===

    def save_session_data(self, session_id: int, data: list):
        """
        Сохранить данные сессии (цветок, клиент, коробки).
        data: список словарей [{flower_name, client_name, boxes_count}]
        """
        with self.conn:
            # Удалить старые данные
            self.conn.execute(
                "DELETE FROM session_data WHERE session_id = ?", (session_id,)
            )
            # Вставить новые
            for entry in data:
                self.conn.execute("""
                    INSERT INTO session_data (session_id, flower_name, client_name, boxes_count)
                    VALUES (?, ?, ?, ?)
                """, (session_id, entry["flower_name"], entry["client_name"], entry["boxes_count"]))

    def get_session_data(self, session_id: int) -> list:
        """Получить данные сессии."""
        cursor = self.conn.execute("""
            SELECT flower_name, client_name, boxes_count 
            FROM session_data 
            WHERE session_id = ?
            ORDER BY flower_name, client_name
        """, (session_id,))
        return [dict(row) for row in cursor.fetchall()]

    def clear_session_data(self, session_id: int):
        """Очистить данные сессии."""
        with self.conn:
            self.conn.execute(
                "DELETE FROM session_data WHERE session_id = ?", (session_id,)
            )


# Тест при запуске модуля
if __name__ == "__main__":
    print("Тестирование DatabaseManager...")
    
    # Использовать тестовую базу
    test_db = DatabaseManager("test_declarant.db")
    test_db.initialize()
    
    # Тест base_config
    print("\n1. Base Config:")
    test_db.save_base_config("/path/to/table.xlsm", ["Rose", "Hydrangea"], ["CLIENT1", "CLIENT2"])
    config = test_db.get_base_config()
    print(f"   Config: {config}")
    
    # Тест sessions
    print("\n2. Sessions:")
    session_id = test_db.create_session("Test Session", "/path/to/invoices")
    print(f"   Created session ID: {session_id}")
    
    sessions = test_db.get_all_sessions()
    print(f"   All sessions: {len(sessions)}")
    
    # Тест session_data
    print("\n3. Session Data:")
    test_data = [
        {"flower_name": "Rose", "client_name": "CLIENT1", "boxes_count": 5},
        {"flower_name": "Hydrangea", "client_name": "CLIENT2", "boxes_count": 3},
    ]
    test_db.save_session_data(session_id, test_data)
    
    loaded_data = test_db.get_session_data(session_id)
    print(f"   Loaded data: {loaded_data}")
    
    # Очистка
    test_db.close()
    os.remove("test_declarant.db")
    
    print("\n✅ Все тесты пройдены!")