343 lines
13 KiB
Python
343 lines
13 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, abort
|
|
from flask_login import login_required, current_user
|
|
from flask_babel import gettext as _
|
|
from app import db
|
|
from app.models.user import User
|
|
from app.models.role import Role, Permission, SYSTEM_MODULES, ACCESS_LEVELS
|
|
from app.utils.decorators import admin_required, permission_required
|
|
from sqlalchemy import or_
|
|
|
|
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
|
|
|
|
# ---------------------------------------------------------
|
|
# USERS MANAGEMENT
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/users')
|
|
@admin_required
|
|
def users_list():
|
|
"""List system users with filtering and role information"""
|
|
search = request.args.get('search', '').strip()
|
|
role_filter = request.args.get('role', '').strip()
|
|
status_filter = request.args.get('status', '').strip()
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = 15
|
|
|
|
query = User.query
|
|
|
|
if search:
|
|
query = query.filter(
|
|
or_(
|
|
User.name.ilike(f'%{search}%'),
|
|
User.email.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
if role_filter:
|
|
if role_filter.isdigit():
|
|
query = query.filter(User.role_id == int(role_filter))
|
|
else:
|
|
query = query.filter(User.role.ilike(role_filter))
|
|
|
|
if status_filter in ['active', 'inactive']:
|
|
query = query.filter(User.is_active == (status_filter == 'active'))
|
|
|
|
pagination = query.order_by(User.id.asc()).paginate(page=page, per_page=per_page, error_out=False)
|
|
roles = Role.query.order_by(Role.name.asc()).all()
|
|
|
|
return render_template(
|
|
'admin/users/list.html',
|
|
users=pagination.items,
|
|
pagination=pagination,
|
|
roles=roles,
|
|
search=search,
|
|
role_filter=role_filter,
|
|
status_filter=status_filter
|
|
)
|
|
|
|
|
|
@admin_bp.route('/users/add', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def user_add():
|
|
"""Create a new user and assign a role"""
|
|
roles = Role.query.order_by(Role.name.asc()).all()
|
|
|
|
if request.method == 'POST':
|
|
name = request.form.get('name', '').strip()
|
|
email = request.form.get('email', '').strip().lower()
|
|
password = request.form.get('password', '')
|
|
role_id = request.form.get('role_id', type=int)
|
|
is_active = request.form.get('is_active') == 'on'
|
|
|
|
errors = []
|
|
if not name:
|
|
errors.append(_('El nombre es requerido.'))
|
|
if not email:
|
|
errors.append(_('El correo electrónico es requerido.'))
|
|
elif User.query.filter_by(email=email).first():
|
|
errors.append(_('Ya existe una cuenta con este correo electrónico.'))
|
|
if not password or len(password) < 6:
|
|
errors.append(_('La contraseña debe tener al menos 6 caracteres.'))
|
|
|
|
role_obj = Role.query.get(role_id) if role_id else None
|
|
if not role_obj:
|
|
errors.append(_('Debes seleccionar un rol válido.'))
|
|
|
|
if errors:
|
|
for err in errors:
|
|
flash(err, 'error')
|
|
return render_template('admin/users/form.html', user=None, roles=roles)
|
|
|
|
user = User(
|
|
name=name,
|
|
email=email,
|
|
role_id=role_obj.id,
|
|
role=role_obj.name,
|
|
is_active=is_active
|
|
)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
flash(_('Usuario "%(name)s" creado exitosamente.', name=user.name), 'success')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
return render_template('admin/users/form.html', user=None, roles=roles)
|
|
|
|
|
|
@admin_bp.route('/users/edit/<int:id>', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def user_edit(id):
|
|
"""Edit user details, role and status"""
|
|
user = User.query.get_or_404(id)
|
|
roles = Role.query.order_by(Role.name.asc()).all()
|
|
|
|
if request.method == 'POST':
|
|
name = request.form.get('name', '').strip()
|
|
email = request.form.get('email', '').strip().lower()
|
|
password = request.form.get('password', '')
|
|
role_id = request.form.get('role_id', type=int)
|
|
is_active = request.form.get('is_active') == 'on'
|
|
|
|
errors = []
|
|
if not name:
|
|
errors.append(_('El nombre es requerido.'))
|
|
if not email:
|
|
errors.append(_('El correo electrónico es requerido.'))
|
|
else:
|
|
existing = User.query.filter_by(email=email).first()
|
|
if existing and existing.id != user.id:
|
|
errors.append(_('Ya existe otra cuenta con este correo electrónico.'))
|
|
|
|
role_obj = Role.query.get(role_id) if role_id else None
|
|
if not role_obj:
|
|
errors.append(_('Debes seleccionar un rol válido.'))
|
|
|
|
# Do not allow deactivating the currently logged in admin user
|
|
if user.id == current_user.id and not is_active:
|
|
errors.append(_('No puedes desactivar tu propia cuenta mientras estás en sesión.'))
|
|
|
|
if errors:
|
|
for err in errors:
|
|
flash(err, 'error')
|
|
return render_template('admin/users/form.html', user=user, roles=roles)
|
|
|
|
user.name = name
|
|
user.email = email
|
|
user.role_id = role_obj.id
|
|
user.role = role_obj.name
|
|
user.is_active = is_active
|
|
|
|
if password:
|
|
if len(password) < 6:
|
|
flash(_('La nueva contraseña debe tener al menos 6 caracteres.'), 'error')
|
|
return render_template('admin/users/form.html', user=user, roles=roles)
|
|
user.set_password(password)
|
|
|
|
db.session.commit()
|
|
flash(_('Usuario "%(name)s" actualizado correctamente.', name=user.name), 'success')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
return render_template('admin/users/form.html', user=user, roles=roles)
|
|
|
|
|
|
@admin_bp.route('/users/toggle/<int:id>', methods=['POST'])
|
|
@admin_required
|
|
def user_toggle(id):
|
|
"""Toggle user active status"""
|
|
user = User.query.get_or_404(id)
|
|
if user.id == current_user.id:
|
|
flash(_('No puedes desactivar tu propia cuenta.'), 'error')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
user.is_active = not user.is_active
|
|
db.session.commit()
|
|
status_str = _('activada') if user.is_active else _('desactivada')
|
|
flash(_('La cuenta de %(name)s ha sido %(status)s.', name=user.name, status=status_str), 'success')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# ROLES & PERMISSIONS MANAGEMENT (FortiGate Matrix)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/roles')
|
|
@admin_required
|
|
def roles_list():
|
|
"""List all roles and permission summaries"""
|
|
roles = Role.query.order_by(Role.id.asc()).all()
|
|
return render_template(
|
|
'admin/roles/list.html',
|
|
roles=roles,
|
|
modules=SYSTEM_MODULES
|
|
)
|
|
|
|
|
|
@admin_bp.route('/roles/add', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def role_add():
|
|
"""Create a new role with the FortiGate permission matrix"""
|
|
if request.method == 'POST':
|
|
name = request.form.get('name', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
|
|
if not name:
|
|
flash(_('El nombre del rol es requerido.'), 'error')
|
|
return render_template('admin/roles/edit.html', role=None, modules=SYSTEM_MODULES, current_perms={})
|
|
|
|
if Role.query.filter_by(name=name).first():
|
|
flash(_('Ya existe un rol con ese nombre.'), 'error')
|
|
return render_template('admin/roles/edit.html', role=None, modules=SYSTEM_MODULES, current_perms={})
|
|
|
|
new_role = Role(name=name, description=description, is_system=False)
|
|
db.session.add(new_role)
|
|
db.session.flush() # to obtain new_role.id
|
|
|
|
# Save permissions from matrix
|
|
for mod in SYSTEM_MODULES:
|
|
mod_id = mod['id']
|
|
level = request.form.get(f'perm_{mod_id}', 'none')
|
|
if level not in ACCESS_LEVELS:
|
|
level = 'none'
|
|
new_role.set_permission(mod_id, level)
|
|
|
|
db.session.commit()
|
|
flash(_('Rol "%(name)s" creado exitosamente con sus permisos.', name=new_role.name), 'success')
|
|
return redirect(url_for('admin.roles_list'))
|
|
|
|
return render_template('admin/roles/edit.html', role=None, modules=SYSTEM_MODULES, current_perms={})
|
|
|
|
|
|
@admin_bp.route('/roles/edit/<int:id>', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def role_edit(id):
|
|
"""Edit an existing role and its FortiGate permission matrix"""
|
|
role = Role.query.get_or_404(id)
|
|
|
|
if request.method == 'POST':
|
|
name = request.form.get('name', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
|
|
if not name:
|
|
flash(_('El nombre del rol es requerido.'), 'error')
|
|
return render_template('admin/roles/edit.html', role=role, modules=SYSTEM_MODULES, current_perms={p.module: p.access_level for p in role.permissions})
|
|
|
|
existing = Role.query.filter_by(name=name).first()
|
|
if existing and existing.id != role.id:
|
|
flash(_('Ya existe otro rol con ese nombre.'), 'error')
|
|
return render_template('admin/roles/edit.html', role=role, modules=SYSTEM_MODULES, current_perms={p.module: p.access_level for p in role.permissions})
|
|
|
|
# If it's a system role, keep name unchanged or protected
|
|
if not role.is_system:
|
|
role.name = name
|
|
role.description = description
|
|
|
|
# Update permissions from the FortiGate matrix
|
|
for mod in SYSTEM_MODULES:
|
|
mod_id = mod['id']
|
|
level = request.form.get(f'perm_{mod_id}', 'none')
|
|
if level not in ACCESS_LEVELS:
|
|
level = 'none'
|
|
role.set_permission(mod_id, level)
|
|
|
|
db.session.commit()
|
|
flash(_('Rol "%(name)s" y permisos actualizados correctamente.', name=role.name), 'success')
|
|
return redirect(url_for('admin.roles_list'))
|
|
|
|
current_perms = {p.module: p.access_level for p in role.permissions}
|
|
return render_template(
|
|
'admin/roles/edit.html',
|
|
role=role,
|
|
modules=SYSTEM_MODULES,
|
|
current_perms=current_perms
|
|
)
|
|
|
|
|
|
@admin_bp.route('/roles/delete/<int:id>', methods=['POST'])
|
|
@admin_required
|
|
def role_delete(id):
|
|
"""Delete a non-system role if it has no assigned users"""
|
|
role = Role.query.get_or_404(id)
|
|
if role.is_system:
|
|
flash(_('No se pueden eliminar roles protegidos del sistema.'), 'error')
|
|
return redirect(url_for('admin.roles_list'))
|
|
|
|
user_count = len(role.users) if role.users else 0
|
|
if user_count > 0:
|
|
flash(_('No se puede eliminar el rol "%(name)s" porque tiene %(count)d usuario(s) asignado(s). Reasígnalos primero.', name=role.name, count=user_count), 'error')
|
|
return redirect(url_for('admin.roles_list'))
|
|
|
|
role_name = role.name
|
|
db.session.delete(role)
|
|
db.session.commit()
|
|
flash(_('El rol "%(name)s" ha sido eliminado.', name=role_name), 'success')
|
|
return redirect(url_for('admin.roles_list'))
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# GOOGLE SHEETS IMPORT & SYNCHRONIZATION
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/import', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def google_sheets_import():
|
|
"""Manual sync and status page for Google Sheets academic schedule data"""
|
|
from app.services.sheets_importer import GoogleSheetsImporter, SHEETS_CONFIG
|
|
from app.models.classroom import Classroom
|
|
from app.models.career import Career
|
|
from app.models.subject import Subject, Commission
|
|
from app.models.reservation import Reservation
|
|
|
|
stats = None
|
|
if request.method == 'POST':
|
|
try:
|
|
importer = GoogleSheetsImporter()
|
|
stats = importer.sync()
|
|
if stats['errors']:
|
|
flash(_('La sincronización se completó con algunas advertencias.'), 'warning')
|
|
else:
|
|
flash(_('Sincronización completada exitosamente. %(rows)d registros procesados.', rows=stats['total_rows_processed']), 'success')
|
|
except Exception as e:
|
|
flash(_('Error al sincronizar con Google Sheets: %(err)s', err=str(e)), 'error')
|
|
|
|
# Calculate current system metrics
|
|
metrics = {
|
|
'total_classrooms': Classroom.query.count(),
|
|
'total_careers': Career.query.count(),
|
|
'total_subjects': Subject.query.count(),
|
|
'total_commissions': Commission.query.count(),
|
|
'total_reservations': Reservation.query.count(),
|
|
'sheets': SHEETS_CONFIG
|
|
}
|
|
|
|
# Recent samples from DB if available
|
|
recent_subjects = Subject.query.order_by(Subject.id.desc()).limit(10).all()
|
|
|
|
return render_template(
|
|
'admin/import.html',
|
|
metrics=metrics,
|
|
stats=stats,
|
|
recent_subjects=recent_subjects
|
|
)
|
|
|