286 lines
10 KiB
Python
286 lines
10 KiB
Python
from flask import render_template, request, session, redirect, url_for, flash, current_app
|
|
from .models import User
|
|
from . import db
|
|
import mariadb
|
|
import os
|
|
|
|
# Settings module logic
|
|
import sqlite3
|
|
import os
|
|
def ensure_roles_table():
|
|
instance_folder = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../instance'))
|
|
if not os.path.exists(instance_folder):
|
|
os.makedirs(instance_folder)
|
|
db_path = os.path.join(instance_folder, 'users.db')
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS roles (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT UNIQUE NOT NULL,
|
|
access_level TEXT NOT NULL,
|
|
description TEXT
|
|
)
|
|
""")
|
|
cursor.execute("""
|
|
INSERT OR IGNORE INTO roles (name, access_level, description)
|
|
VALUES (?, ?, ?)
|
|
""", ('superadmin', 'full', 'Full access to all app areas and functions'))
|
|
conn.commit()
|
|
conn.close()
|
|
# List of roles (should match your app's roles)
|
|
ROLES = [
|
|
'superadmin', 'admin', 'manager', 'warehouse_manager', 'warehouse_worker', 'quality_manager', 'quality_worker'
|
|
]
|
|
|
|
# Helper to check if current user is superadmin
|
|
def is_superadmin():
|
|
return session.get('role') == 'superadmin'
|
|
|
|
# Route handler for editing access roles
|
|
def edit_access_roles_handler():
|
|
if not is_superadmin():
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.dashboard'))
|
|
ensure_roles_table()
|
|
return render_template('edit_access_roles.html', roles=ROLES)
|
|
|
|
# Handler for updating role access (stub, to be implemented)
|
|
def update_role_access_handler(role):
|
|
if not is_superadmin():
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.dashboard'))
|
|
if role == 'superadmin':
|
|
flash('Superadmin access cannot be changed.')
|
|
return redirect(url_for('main.edit_access_roles'))
|
|
access_level = request.form.get('access_level')
|
|
# TODO: Save access_level for the role in the database or config
|
|
flash(f'Access for role {role} updated to {access_level}.')
|
|
return redirect(url_for('main.edit_access_roles'))
|
|
|
|
def settings_handler():
|
|
if 'role' not in session or session['role'] != 'superadmin':
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
# Get users from external MariaDB database
|
|
users = []
|
|
try:
|
|
conn = get_external_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Create users table if it doesn't exist
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(50) UNIQUE NOT NULL,
|
|
password VARCHAR(255) NOT NULL,
|
|
role VARCHAR(50) NOT NULL
|
|
)
|
|
''')
|
|
|
|
# Get all users from external database
|
|
cursor.execute("SELECT id, username, password, role FROM users")
|
|
users_data = cursor.fetchall()
|
|
|
|
# Convert to list of dictionaries for template compatibility
|
|
users = []
|
|
for user_data in users_data:
|
|
users.append({
|
|
'id': user_data[0],
|
|
'username': user_data[1],
|
|
'password': user_data[2],
|
|
'role': user_data[3]
|
|
})
|
|
|
|
conn.close()
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching users from external database: {e}")
|
|
flash(f'Error loading users: {e}')
|
|
|
|
# Load external database settings from the instance folder
|
|
external_settings = {}
|
|
settings_file = os.path.join(current_app.instance_path, 'external_server.conf')
|
|
if os.path.exists(settings_file):
|
|
with open(settings_file, 'r') as f:
|
|
for line in f:
|
|
key, value = line.strip().split('=', 1)
|
|
external_settings[key] = value
|
|
|
|
return render_template('settings.html', users=users, external_settings=external_settings)
|
|
|
|
# Helper function to get external database connection
|
|
def get_external_db_connection():
|
|
"""Reads the external_server.conf file and returns a MariaDB database connection."""
|
|
settings_file = os.path.join(current_app.instance_path, 'external_server.conf')
|
|
if not os.path.exists(settings_file):
|
|
raise FileNotFoundError("The external_server.conf file is missing in the instance folder.")
|
|
|
|
# Read settings from the configuration file
|
|
settings = {}
|
|
with open(settings_file, 'r') as f:
|
|
for line in f:
|
|
key, value = line.strip().split('=', 1)
|
|
settings[key] = value
|
|
|
|
# Create a database connection
|
|
return mariadb.connect(
|
|
user=settings['username'],
|
|
password=settings['password'],
|
|
host=settings['server_domain'],
|
|
port=int(settings['port']),
|
|
database=settings['database_name']
|
|
)
|
|
|
|
# User management handlers
|
|
def create_user_handler():
|
|
if 'role' not in session or session['role'] != 'superadmin':
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.settings'))
|
|
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
role = request.form['role']
|
|
|
|
try:
|
|
# Connect to external MariaDB database
|
|
conn = get_external_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Create users table if it doesn't exist
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(50) UNIQUE NOT NULL,
|
|
password VARCHAR(255) NOT NULL,
|
|
role VARCHAR(50) NOT NULL
|
|
)
|
|
''')
|
|
|
|
# Check if the username already exists
|
|
cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
|
|
if cursor.fetchone():
|
|
flash('User already exists.')
|
|
conn.close()
|
|
return redirect(url_for('main.settings'))
|
|
|
|
# Create a new user in external MariaDB
|
|
cursor.execute("""
|
|
INSERT INTO users (username, password, role)
|
|
VALUES (%s, %s, %s)
|
|
""", (username, password, role))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
flash('User created successfully in external database.')
|
|
|
|
except Exception as e:
|
|
print(f"Error creating user in external database: {e}")
|
|
flash(f'Error creating user: {e}')
|
|
|
|
return redirect(url_for('main.settings'))
|
|
|
|
def edit_user_handler():
|
|
if 'role' not in session or session['role'] != 'superadmin':
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.settings'))
|
|
|
|
user_id = request.form.get('user_id')
|
|
password = request.form.get('password', '').strip()
|
|
role = request.form.get('role')
|
|
|
|
if not user_id or not role:
|
|
flash('Missing required fields.')
|
|
return redirect(url_for('main.settings'))
|
|
|
|
try:
|
|
# Connect to external MariaDB database
|
|
conn = get_external_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Check if the user exists
|
|
cursor.execute("SELECT id FROM users WHERE id = %s", (user_id,))
|
|
if not cursor.fetchone():
|
|
flash('User not found.')
|
|
conn.close()
|
|
return redirect(url_for('main.settings'))
|
|
|
|
# Update the user's details in external MariaDB
|
|
if password: # Only update password if provided
|
|
cursor.execute("""
|
|
UPDATE users SET password = %s, role = %s WHERE id = %s
|
|
""", (password, role, user_id))
|
|
flash('User updated successfully (including password).')
|
|
else: # Just update role if no password provided
|
|
cursor.execute("""
|
|
UPDATE users SET role = %s WHERE id = %s
|
|
""", (role, user_id))
|
|
flash('User role updated successfully.')
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
except Exception as e:
|
|
print(f"Error updating user in external database: {e}")
|
|
flash(f'Error updating user: {e}')
|
|
|
|
return redirect(url_for('main.settings'))
|
|
|
|
def delete_user_handler():
|
|
if 'role' not in session or session['role'] != 'superadmin':
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.settings'))
|
|
|
|
user_id = request.form['user_id']
|
|
|
|
try:
|
|
# Connect to external MariaDB database
|
|
conn = get_external_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Check if the user exists
|
|
cursor.execute("SELECT id FROM users WHERE id = %s", (user_id,))
|
|
if not cursor.fetchone():
|
|
flash('User not found.')
|
|
conn.close()
|
|
return redirect(url_for('main.settings'))
|
|
|
|
# Delete the user from external MariaDB
|
|
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
flash('User deleted successfully from external database.')
|
|
|
|
except Exception as e:
|
|
print(f"Error deleting user from external database: {e}")
|
|
flash(f'Error deleting user: {e}')
|
|
|
|
return redirect(url_for('main.settings'))
|
|
|
|
def save_external_db_handler():
|
|
if 'role' not in session or session['role'] != 'superadmin':
|
|
flash('Access denied: Superadmin only.')
|
|
return redirect(url_for('main.settings'))
|
|
|
|
# Get form data
|
|
server_domain = request.form['server_domain']
|
|
port = request.form['port']
|
|
database_name = request.form['database_name']
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
|
|
# Save data to a file in the instance folder
|
|
settings_file = os.path.join(current_app.instance_path, 'external_server.conf')
|
|
os.makedirs(os.path.dirname(settings_file), exist_ok=True)
|
|
with open(settings_file, 'w') as f:
|
|
f.write(f"server_domain={server_domain}\n")
|
|
f.write(f"port={port}\n")
|
|
f.write(f"database_name={database_name}\n")
|
|
f.write(f"username={username}\n")
|
|
f.write(f"password={password}\n")
|
|
|
|
flash('External database settings saved/updated successfully.')
|
|
return redirect(url_for('main.settings'))
|