438 lines
16 KiB
Python
438 lines
16 KiB
Python
#!/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())
|