#!/usr/bin/env python3 """ Comprehensive Database Initialization Script Creates all required tables and initializes default data This script should be run once when the application starts """ import pymysql import os import sys import logging import hashlib from pathlib import Path # Setup logging logging.basicConfig( level=logging.INFO, format='[%(asctime)s] %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Database configuration from environment DB_HOST = os.getenv('DB_HOST', 'mariadb') DB_PORT = int(os.getenv('DB_PORT', '3306')) DB_USER = os.getenv('DB_USER', 'quality_user') DB_PASSWORD = os.getenv('DB_PASSWORD', 'quality_pass') DB_NAME = os.getenv('DB_NAME', 'quality_db') def hash_password(password): """Hash password using SHA256""" return hashlib.sha256(password.encode()).hexdigest() def execute_sql(conn, sql, params=None, description=""): """Execute SQL statement and log result""" try: cursor = conn.cursor() if params: cursor.execute(sql, params) else: cursor.execute(sql) if description: logger.info(f"✓ {description}") cursor.close() return True except pymysql.Error as e: if "already exists" in str(e).lower() or "duplicate" in str(e).lower(): if description: logger.info(f"✓ {description} (already exists)") return True logger.error(f"✗ SQL Error: {e}") return False except Exception as e: logger.error(f"✗ Unexpected Error: {e}") return False def create_database(): """Create the database if it doesn't exist""" logger.info("Step 1: Creating database...") try: conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD ) cursor = conn.cursor() cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}`") cursor.close() conn.close() logger.info(f"✓ Database '{DB_NAME}' created or already exists") return True except Exception as e: logger.error(f"✗ Failed to create database: {e}") return False def create_tables(): """Create all application tables""" logger.info("\nStep 2: Creating tables...") try: conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) # Users table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(255) UNIQUE NOT NULL, email VARCHAR(255), full_name VARCHAR(255), role VARCHAR(50) DEFAULT 'user', is_active TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'users'") # User credentials table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS user_credentials ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, password_hash VARCHAR(255) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'user_credentials'") # Quality inspections table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS quality_inspections ( id INT AUTO_INCREMENT PRIMARY KEY, inspection_type VARCHAR(100), status VARCHAR(50), inspector_id INT, inspection_date DATETIME DEFAULT CURRENT_TIMESTAMP, notes TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (inspector_id) REFERENCES users(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'quality_inspections'") # Settings table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS application_settings ( id INT AUTO_INCREMENT PRIMARY KEY, setting_key VARCHAR(255) UNIQUE NOT NULL, setting_value LONGTEXT, setting_type VARCHAR(50), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'application_settings'") # QZ Tray Pairing Keys table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS qz_pairing_keys ( id INT AUTO_INCREMENT PRIMARY KEY, printer_name VARCHAR(255) NOT NULL, pairing_key VARCHAR(255) UNIQUE NOT NULL, valid_until DATE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'qz_pairing_keys'") # API Keys table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS api_keys ( id INT AUTO_INCREMENT PRIMARY KEY, key_name VARCHAR(255) NOT NULL, key_type VARCHAR(100) NOT NULL, api_key VARCHAR(255) UNIQUE NOT NULL, is_active TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'api_keys'") # Backup Schedules table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS backup_schedules ( id INT AUTO_INCREMENT PRIMARY KEY, schedule_name VARCHAR(255) NOT NULL, frequency VARCHAR(50) NOT NULL COMMENT 'daily or weekly', day_of_week VARCHAR(20) COMMENT 'Monday, Tuesday, etc for weekly schedules', time_of_day TIME NOT NULL COMMENT 'HH:MM format', backup_type VARCHAR(50) DEFAULT 'full' COMMENT 'full or data_only', is_active TINYINT(1) DEFAULT 1, last_run DATETIME, next_run DATETIME, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'backup_schedules'") # Roles table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS roles ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) UNIQUE NOT NULL, description TEXT, level INT DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'roles'") # User modules table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS user_modules ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, module_name VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY unique_user_module (user_id, module_name), FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'user_modules'") # User permissions table execute_sql(conn, """ CREATE TABLE IF NOT EXISTS user_permissions ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, module_name VARCHAR(100) NOT NULL, section_name VARCHAR(100) NOT NULL, action_name VARCHAR(100) NOT NULL, granted TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY unique_permission (user_id, module_name, section_name, action_name), FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """, description="Table 'user_permissions'") conn.commit() conn.close() logger.info("✓ All tables created successfully") return True except Exception as e: logger.error(f"✗ Failed to create tables: {e}") return False def insert_default_data(): """Insert default roles and admin user""" logger.info("\nStep 3: Inserting default data...") try: conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) cursor = conn.cursor() # Insert default roles if they don't exist roles = [ ('superadmin', 'Super Administrator - Full system access', 100), ('admin', 'Administrator - Administrative access', 90), ('manager', 'Manager - Full access to assigned modules', 70), ('worker', 'Worker - Limited access', 50), ] logger.info(" Creating roles...") for role_name, role_desc, role_level in roles: try: cursor.execute( "INSERT INTO roles (name, description, level) VALUES (%s, %s, %s)", (role_name, role_desc, role_level) ) logger.info(f" ✓ Role '{role_name}' created") except pymysql.Error as e: if "duplicate" in str(e).lower(): logger.info(f" ✓ Role '{role_name}' already exists") else: logger.warning(f" ⚠ Role '{role_name}': {e}") # Check if admin user exists cursor.execute("SELECT id FROM users WHERE username = 'admin'") admin_result = cursor.fetchone() if not admin_result: logger.info(" Creating default admin user...") cursor.execute( "INSERT INTO users (username, email, full_name, role, is_active) VALUES (%s, %s, %s, %s, 1)", ('admin', 'admin@quality-app.local', 'Administrator', 'admin') ) # Get admin user ID cursor.execute("SELECT id FROM users WHERE username = 'admin'") admin_id = cursor.fetchone()[0] # Insert admin password password_hash = hash_password('admin123') cursor.execute( "INSERT INTO user_credentials (user_id, password_hash) VALUES (%s, %s)", (admin_id, password_hash) ) logger.info(" ✓ Admin user created (username: admin, password: admin123)") # Grant admin user access to all modules logger.info(" Granting module access to admin user...") modules = ['quality', 'settings'] for module in modules: try: cursor.execute( "INSERT IGNORE INTO user_modules (user_id, module_name) VALUES (%s, %s)", (admin_id, module) ) logger.info(f" ✓ Module '{module}' granted to admin") except pymysql.Error as e: logger.warning(f" ⚠ Module '{module}': {e}") else: logger.info(" ✓ Admin user already exists") # Insert default application settings logger.info(" Creating default application settings...") default_settings = [ ('app_name', 'Quality App v2', 'string'), ('app_version', '2.0.0', 'string'), ('session_timeout', '480', 'integer'), ('backup_retention_days', '30', 'integer'), ('backup_auto_cleanup', '0', 'boolean'), ] for setting_key, setting_value, setting_type in default_settings: try: cursor.execute( "INSERT IGNORE INTO application_settings (setting_key, setting_value, setting_type) VALUES (%s, %s, %s)", (setting_key, setting_value, setting_type) ) logger.info(f" ✓ Setting '{setting_key}' initialized") except pymysql.Error as e: logger.warning(f" ⚠ Setting '{setting_key}': {e}") conn.commit() conn.close() logger.info("✓ Default data inserted successfully") return True except Exception as e: logger.error(f"✗ Failed to insert default data: {e}") return False def verify_database(): """Verify all tables were created""" logger.info("\nStep 4: Verifying database...") try: conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) cursor = conn.cursor() cursor.execute("SHOW TABLES") tables = [row[0] for row in cursor.fetchall()] required_tables = [ 'users', 'user_credentials', 'quality_inspections', 'application_settings', 'roles', 'user_modules', 'user_permissions' ] logger.info(f" Database tables: {', '.join(tables)}") missing = [t for t in required_tables if t not in tables] if missing: logger.error(f" ✗ Missing tables: {', '.join(missing)}") conn.close() return False # Count records cursor.execute("SELECT COUNT(*) FROM roles") role_count = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM users") user_count = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM user_credentials") cred_count = cursor.fetchone()[0] logger.info(f" ✓ All {len(required_tables)} required tables exist") logger.info(f" ✓ Roles: {role_count}") logger.info(f" ✓ Users: {user_count}") logger.info(f" ✓ User credentials: {cred_count}") conn.close() return True except Exception as e: logger.error(f"✗ Verification failed: {e}") return False def main(): """Main initialization flow""" logger.info("=" * 60) logger.info("Database Initialization Script") logger.info("=" * 60) logger.info(f"Target: {DB_USER}@{DB_HOST}:{DB_PORT}/{DB_NAME}\n") steps = [ ("Create database", create_database), ("Create tables", create_tables), ("Insert default data", insert_default_data), ("Verify database", verify_database), ] failed = [] for step_name, step_func in steps: try: if not step_func(): failed.append(step_name) except Exception as e: logger.error(f"✗ {step_name} failed: {e}") failed.append(step_name) logger.info("\n" + "=" * 60) if failed: logger.error(f"✗ FAILED: {', '.join(failed)}") logger.info("=" * 60) return 1 else: logger.info("✓ Database initialization completed successfully!") logger.info("=" * 60) return 0 if __name__ == '__main__': sys.exit(main())