478 lines
18 KiB
Python
478 lines
18 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.models.milestone import MilestoneType, AcademicMilestone, BASE_MILESTONE_TYPES
|
|
from app.utils.decorators import admin_required, permission_required
|
|
from sqlalchemy import or_
|
|
import re
|
|
|
|
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
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# MILESTONE TYPES & ACADEMIC CATEGORIES (ABM / CRUD)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/milestone-types')
|
|
@permission_required('milestone_types', 'read')
|
|
def milestone_types_list():
|
|
"""List all milestone categories / types"""
|
|
# Preload default milestone types if table is empty
|
|
if MilestoneType.query.count() == 0:
|
|
for item in BASE_MILESTONE_TYPES:
|
|
mt = MilestoneType(
|
|
name=item['name'],
|
|
code=item['code'],
|
|
color=item['color'],
|
|
description=item['description'],
|
|
sort_order=item['sort_order'],
|
|
is_active=True
|
|
)
|
|
db.session.add(mt)
|
|
db.session.commit()
|
|
|
|
types = MilestoneType.query.order_by(MilestoneType.sort_order.asc(), MilestoneType.id.asc()).all()
|
|
can_edit = current_user.has_permission('milestone_types', 'read_write')
|
|
return render_template('admin/milestone_types/list.html', types=types, can_edit=can_edit)
|
|
|
|
|
|
@admin_bp.route('/milestone-types/add', methods=['POST'])
|
|
@permission_required('milestone_types', 'read_write')
|
|
def milestone_type_add():
|
|
"""Create a new milestone type"""
|
|
name = request.form.get('name', '').strip()
|
|
code = request.form.get('code', '').strip().lower()
|
|
color = request.form.get('color', '#0d6efd').strip()
|
|
description = request.form.get('description', '').strip()
|
|
sort_order = request.form.get('sort_order', 0, type=int)
|
|
is_active = True if request.form.get('is_active') in ['1', 'true', 'True', 'on'] else False
|
|
|
|
if not name:
|
|
flash(_('El nombre de la tipificación es obligatorio.'), 'error')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
# Auto-generate code slug if not provided
|
|
if not code:
|
|
code = re.sub(r'[^a-zA-Z0-9_]+', '_', name.lower()).strip('_')
|
|
|
|
# Check code uniqueness
|
|
existing = MilestoneType.query.filter_by(code=code).first()
|
|
if existing:
|
|
flash(_('Ya existe una tipificación con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
new_type = MilestoneType(
|
|
name=name,
|
|
code=code,
|
|
color=color or '#0d6efd',
|
|
description=description,
|
|
sort_order=sort_order,
|
|
is_active=is_active
|
|
)
|
|
db.session.add(new_type)
|
|
db.session.commit()
|
|
flash(_('Tipificación "%(name)s" creada exitosamente.', name=new_type.name), 'success')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
|
|
@admin_bp.route('/milestone-types/edit/<int:id>', methods=['POST'])
|
|
@permission_required('milestone_types', 'read_write')
|
|
def milestone_type_edit(id):
|
|
"""Edit an existing milestone type"""
|
|
item = MilestoneType.query.get_or_404(id)
|
|
name = request.form.get('name', '').strip()
|
|
code = request.form.get('code', '').strip().lower()
|
|
color = request.form.get('color', '#0d6efd').strip()
|
|
description = request.form.get('description', '').strip()
|
|
sort_order = request.form.get('sort_order', item.sort_order, type=int)
|
|
is_active = True if request.form.get('is_active') in ['1', 'true', 'True', 'on'] else False
|
|
|
|
if not name:
|
|
flash(_('El nombre de la tipificación es obligatorio.'), 'error')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
if not code:
|
|
code = re.sub(r'[^a-zA-Z0-9_]+', '_', name.lower()).strip('_')
|
|
|
|
# Check code uniqueness against other records
|
|
existing = MilestoneType.query.filter(MilestoneType.code == code, MilestoneType.id != id).first()
|
|
if existing:
|
|
flash(_('Ya existe otra tipificación con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
item.name = name
|
|
item.code = code
|
|
item.color = color or item.color
|
|
item.description = description
|
|
item.sort_order = sort_order
|
|
item.is_active = is_active
|
|
|
|
db.session.commit()
|
|
flash(_('Tipificación "%(name)s" actualizada correctamente.', name=item.name), 'success')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
|
|
@admin_bp.route('/milestone-types/toggle/<int:id>', methods=['POST'])
|
|
@permission_required('milestone_types', 'read_write')
|
|
def milestone_type_toggle(id):
|
|
"""Toggle active status of a milestone type"""
|
|
item = MilestoneType.query.get_or_404(id)
|
|
item.is_active = not item.is_active
|
|
db.session.commit()
|
|
|
|
if request.is_json or request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
return jsonify({
|
|
'success': True,
|
|
'id': item.id,
|
|
'is_active': item.is_active,
|
|
'message': _('Tipificación "%(name)s" %(status)s.', name=item.name, status=_('activada') if item.is_active else _('desactivada'))
|
|
})
|
|
|
|
status_str = _('activada') if item.is_active else _('desactivada')
|
|
flash(_('Tipificación "%(name)s" ha sido %(status)s.', name=item.name, status=status_str), 'success')
|
|
return redirect(url_for('admin.milestone_types_list'))
|
|
|
|
|
|
@admin_bp.route('/api/milestone-types')
|
|
@login_required
|
|
def api_milestone_types():
|
|
"""API endpoint to get active milestone types for forms and dropdowns"""
|
|
types = MilestoneType.query.filter_by(is_active=True).order_by(MilestoneType.sort_order.asc(), MilestoneType.id.asc()).all()
|
|
return jsonify([t.to_dict() for t in types])
|
|
|
|
|