8de85ca87f
- Add SchemaVerifier class for automatic database schema verification and repair - Implement warehouse_manager (Level 75) and warehouse_worker (Level 35) roles - Add zone-based access control for warehouse workers - Implement worker-manager binding system with zone filtering - Add comprehensive database auto-repair on Docker initialization - Remove Module Access section from user form (role-based access only) - Add autocomplete attributes to password fields for better UX - Include detailed documentation for warehouse implementation - Update initialize_db.py with schema verification as Step 0
520 lines
20 KiB
Python
520 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Comprehensive Database Initialization Script
|
||
Creates all required tables and initializes default data
|
||
Includes schema verification to check existing databases for correctness
|
||
This script should be run once when the application starts
|
||
"""
|
||
|
||
import pymysql
|
||
import os
|
||
import sys
|
||
import logging
|
||
import hashlib
|
||
from pathlib import Path
|
||
from app.db_schema_verifier import SchemaVerifier
|
||
|
||
# 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 check_and_repair_database():
|
||
"""
|
||
Check existing database for correct structure
|
||
Repair any missing tables, columns, or reference data
|
||
"""
|
||
logger.info("Step 0: Checking existing database structure...")
|
||
|
||
try:
|
||
# First check if database exists
|
||
conn = pymysql.connect(
|
||
host=DB_HOST,
|
||
port=DB_PORT,
|
||
user=DB_USER,
|
||
password=DB_PASSWORD
|
||
)
|
||
|
||
cursor = conn.cursor()
|
||
cursor.execute(f"SHOW DATABASES LIKE %s", (DB_NAME,))
|
||
|
||
if not cursor.fetchone():
|
||
# Database doesn't exist, skip verification
|
||
logger.info(" ℹ Database doesn't exist yet, skipping structure check")
|
||
conn.close()
|
||
return True
|
||
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
# Database exists, now connect to it and verify/repair structure
|
||
conn = pymysql.connect(
|
||
host=DB_HOST,
|
||
port=DB_PORT,
|
||
user=DB_USER,
|
||
password=DB_PASSWORD,
|
||
database=DB_NAME
|
||
)
|
||
|
||
# Run schema verification and repair
|
||
verifier = SchemaVerifier(conn)
|
||
success, summary = verifier.verify_and_repair()
|
||
|
||
# Log the summary
|
||
for line in summary.split('\n'):
|
||
if line.strip():
|
||
logger.info(f" {line}")
|
||
|
||
conn.close()
|
||
return success
|
||
|
||
except pymysql.Error as e:
|
||
if "Unknown database" in str(e):
|
||
logger.info(" ℹ Database doesn't exist yet, skipping structure check")
|
||
return True
|
||
logger.error(f"✗ Database check failed: {e}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"✗ Database check 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'")
|
||
|
||
# Worker-Manager bindings (for warehouse module hierarchy)
|
||
execute_sql(conn, """
|
||
CREATE TABLE IF NOT EXISTS worker_manager_bindings (
|
||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||
manager_id INT NOT NULL,
|
||
worker_id INT NOT NULL,
|
||
warehouse_zone VARCHAR(100),
|
||
is_active TINYINT(1) DEFAULT 1,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY unique_binding (manager_id, worker_id),
|
||
FOREIGN KEY (manager_id) REFERENCES users(id) ON DELETE CASCADE,
|
||
FOREIGN KEY (worker_id) REFERENCES users(id) ON DELETE CASCADE,
|
||
CHECK (manager_id != worker_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
|
||
""", description="Table 'worker_manager_bindings'")
|
||
|
||
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 - Quality - Full access to assigned modules', 70),
|
||
('warehouse_manager', 'Manager - Warehouse - Full warehouse module access', 75),
|
||
('worker', 'Worker - Quality - Limited access', 50),
|
||
('warehouse_worker', 'Worker - Warehouse - Input-only warehouse access', 35),
|
||
]
|
||
|
||
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 = [
|
||
("Check/repair existing database", check_and_repair_database),
|
||
("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())
|