Frontend/utils/database.py - master - Xton/Case_13_HTML_Newsletter_Builder

Frontend/utils/database.py
293 строки11 KB
sdankov37
Доработка UI, обновление взаимодействия UI и Data
15 май 2026, 17:38
15 май 2026, 17:38cda60a2
import sqlite3
import uuid
from datetime import datetime
from typing import List, Dict, Any, Optional

class Database:
    """Класс для управления базой данных"""
    
    def __init__(self, db_path: str = 'deadalus_sequencer.db'):
        self.db_path = db_path
        self._init_database()
    
    def _get_connection(self):
        """Получить соединение с БД"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn
    
    def _init_database(self):
        """Инициализация базы данных"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("PRAGMA foreign_keys = ON;")
            
            # Table: sessions
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS sessions (
                session_id TEXT PRIMARY KEY,
                current_template_name TEXT NOT NULL,
                started_at TIMESTAMP NOT NULL,
                canvas_width INTEGER NOT NULL DEFAULT 600,
                canvas_height INTEGER NOT NULL DEFAULT 400
            )
            ''')
            
            # Table: templates
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS templates (
                template_id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                name TEXT NOT NULL,
                html_cache TEXT,
                created_at TIMESTAMP NOT NULL,
                updated_at TIMESTAMP NOT NULL,
                FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
            )
            ''')
            
            # Table: blocks
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS blocks (
                block_id TEXT PRIMARY KEY,
                template_id TEXT NOT NULL,
                type TEXT NOT NULL,
                content TEXT,
                x INTEGER NOT NULL,
                y INTEGER NOT NULL,
                width INTEGER,
                height INTEGER,
                radius INTEGER,
                x1 INTEGER,
                y1 INTEGER,
                x2 INTEGER,
                y2 INTEGER,
                color TEXT DEFAULT '#000000',
                fill BOOLEAN DEFAULT FALSE,
                font_family TEXT,
                font_size INTEGER,
                FOREIGN KEY (template_id) REFERENCES templates(template_id) ON DELETE CASCADE
            )
            ''')
            
            # Table: metrics
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL UNIQUE,
                elements_count INTEGER DEFAULT 0,
                time_in_editor TEXT DEFAULT '00:00',
                saved_templates_count INTEGER DEFAULT 0,
                FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
            )
            ''')
            
            # Table: campaigns
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS campaigns (
                campaign_id TEXT PRIMARY KEY,
                recipient_email TEXT NOT NULL,
                status TEXT NOT NULL,
                queued_at TIMESTAMP NOT NULL,
                error_message TEXT,
                completed_at TIMESTAMP
            )
            ''')
            
            conn.commit()
            print("Database initialized successfully")
    
    def create_session(self, session_id: str = None) -> Dict[str, Any]:
        """Create a new session"""
        if session_id is None:
            session_id = str(uuid.uuid4())
        
        now = datetime.now().isoformat()
        
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
            INSERT INTO sessions (session_id, current_template_name, started_at, canvas_width, canvas_height)
            VALUES (?, ?, ?, ?, ?)
            ''', (session_id, 'New Template', now, 600, 400))
            
            cursor.execute('''
            INSERT INTO metrics (session_id, elements_count, time_in_editor, saved_templates_count)
            VALUES (?, ?, ?, ?)
            ''', (session_id, 0, '00:00', 0))
            
            conn.commit()
        
        return self.get_session(session_id)
    
    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        """Get session data by ID"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM sessions WHERE session_id = ?', (session_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all_templates(self, session_id: str) -> List[Dict[str, Any]]:
        """Get all templates for a session"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM templates WHERE session_id = ? ORDER BY created_at DESC', (session_id,))
            return [dict(row) for row in cursor.fetchall()]
    
    def get_template(self, template_id: str) -> Optional[Dict[str, Any]]:
        """Get template by ID"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM templates WHERE template_id = ?', (template_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_blocks(self, template_id: str) -> List[Dict[str, Any]]:
        """Get all blocks for a template"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM blocks WHERE template_id = ?', (template_id,))
            return [dict(row) for row in cursor.fetchall()]
    
    def create_template(self, session_id: str, name: str) -> str:
        """Create a new template"""
        template_id = str(uuid.uuid4())
        now = datetime.now().isoformat()
        
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
            INSERT INTO templates (template_id, session_id, name, html_cache, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
            ''', (template_id, session_id, name, None, now, now))
            
            self._update_metrics(conn, session_id)
            conn.commit()
        
        return template_id
    
    def add_block(self, template_id: str, block_data: Dict[str, Any]) -> str:
        """Add a block to a template"""
        block_id = str(uuid.uuid4())
        
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
            INSERT INTO blocks (
                block_id, template_id, type, content, x, y, 
                color, fill, font_family, font_size, width, height, radius, x1, y1, x2, y2
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                block_id, template_id,
                block_data.get('type'), block_data.get('content'),
                block_data.get('x', 0), block_data.get('y', 0),
                block_data.get('color', '#000000'), block_data.get('fill', False),
                block_data.get('font_family'), block_data.get('font_size'),
                block_data.get('width'), block_data.get('height'),
                block_data.get('radius'), block_data.get('x1'), block_data.get('y1'),
                block_data.get('x2'), block_data.get('y2')
            ))
            
            self._update_metrics_by_template(conn, template_id)
            conn.commit()
        
        return block_id
    
    def _update_metrics(self, conn, session_id: str):
        """Update session metrics"""
        cursor = conn.cursor()
        cursor.execute('SELECT COUNT(*) as count FROM templates WHERE session_id = ?', (session_id,))
        saved_count = cursor.fetchone()['count']
        
        cursor.execute('''
        UPDATE metrics SET saved_templates_count = ? WHERE session_id = ?
        ''', (saved_count, session_id))
    
    def _update_metrics_by_template(self, conn, template_id: str):
        """Update metrics by template ID"""
        cursor = conn.cursor()
        cursor.execute('SELECT session_id FROM templates WHERE template_id = ?', (template_id,))
        row = cursor.fetchone()
        if row:
            self._update_metrics(conn, row['session_id'])
    
    def create_campaign(self, recipient_email: str) -> Dict[str, Any]:
        """Create a new campaign"""
        import random
        
        campaign_id = str(uuid.uuid4())
        now = datetime.now().isoformat()
        
        is_valid = '@' in recipient_email and '.' in recipient_email
        
        if not is_valid:
            status = 'failed'
            error_message = 'Invalid email format'
        else:
            statuses = ['queued', 'sending', 'completed', 'failed']
            status = random.choice(statuses)
            error_message = None if status != 'failed' else 'Mock sending error'
        
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
            INSERT INTO campaigns (campaign_id, recipient_email, status, queued_at, error_message)
            VALUES (?, ?, ?, ?, ?)
            ''', (campaign_id, recipient_email, status, now, error_message))
            conn.commit()
        
        return {
            'campaign_id': campaign_id,
            'status': status,
            'queued_at': now,
            'error_message': error_message
        }
    
    def get_campaign_status(self, campaign_id: str) -> Optional[Dict[str, Any]]:
        """Get campaign status by ID"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM campaigns WHERE campaign_id = ?', (campaign_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all_campaigns(self) -> List[Dict[str, Any]]:
        """Get all campaigns"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM campaigns ORDER BY queued_at DESC')
            return [dict(row) for row in cursor.fetchall()]


# Create global database instance
db = Database()


if __name__ == '__main__':
    print("\n" + "="*50)
    print("Testing Database")
    print("="*50)
    
    session = db.create_session()
    print(f"\nSession created: {session['session_id']}")
    
    template_id = db.create_template(session['session_id'], 'Test Template')
    print(f"Template created: {template_id}")
    
    db.add_block(template_id, {
        'type': 'text',
        'content': 'Hello, world!',
        'x': 50, 'y': 50,
        'color': '#333333', 'font_family': 'Arial', 'font_size': 24
    })
    print("Text block added")
    
    blocks = db.get_blocks(template_id)
    print(f"Blocks in template: {len(blocks)}")
    
    campaign = db.create_campaign('test@example.com')
    print(f"Campaign created: status {campaign['status']}")
    
    print("\nTesting completed!")