1120 lines
43 KiB
Python
1120 lines
43 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'))
|
|
|
|
|
|
@admin_bp.route('/users/delete/<int:id>', methods=['POST'])
|
|
@admin_required
|
|
def user_delete(id):
|
|
"""Delete a user account"""
|
|
if id == current_user.id:
|
|
flash(_('No puedes eliminar tu propia cuenta mientras estás en sesión activa.'), 'error')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
user = User.query.get_or_404(id)
|
|
|
|
# Protect sole system administrator
|
|
if user.is_admin():
|
|
admin_count = User.query.filter(User.role.ilike('%ADMIN%'), User.is_active == True).count()
|
|
if admin_count <= 1:
|
|
flash(_('No es posible eliminar al único administrador activo del sistema.'), 'error')
|
|
return redirect(url_for('admin.users_list'))
|
|
|
|
# Unassign user from commissions as teacher
|
|
from app.models.subject import Commission
|
|
Commission.query.filter_by(teacher_id=id).update({'teacher_id': None})
|
|
|
|
name = user.name
|
|
email = user.email
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
|
|
flash(_('El usuario "%(name)s" (%(email)s) ha sido eliminado del sistema.', name=name, email=email), '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('/milestone-types/delete/<int:id>', methods=['POST'])
|
|
@permission_required('milestone_types', 'read_write')
|
|
def milestone_type_delete(id):
|
|
"""Delete a milestone type if unused, or deactivate with notice"""
|
|
item = MilestoneType.query.get_or_404(id)
|
|
from app.models.milestone import AcademicMilestone
|
|
milestone_count = AcademicMilestone.query.filter_by(milestone_type_id=id).count()
|
|
|
|
if milestone_count > 0:
|
|
item.is_active = False
|
|
db.session.commit()
|
|
flash(_('La tipificación "%(name)s" tiene %(count)d hitos registrados. Se ha desactivado para conservar el historial académico.', name=item.name, count=milestone_count), 'warning')
|
|
else:
|
|
name = item.name
|
|
db.session.delete(item)
|
|
db.session.commit()
|
|
flash(_('Tipificación "%(name)s" eliminada permanentemente.', name=name), '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])
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# 1. GESTIÓN DE CARRERAS (CAREERS)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/careers')
|
|
@login_required
|
|
def careers_list():
|
|
"""List and manage academic careers"""
|
|
if not (current_user.has_permission('academic', 'read') or current_user.can_manage()):
|
|
abort(403)
|
|
|
|
from app.models.career import Career
|
|
from app.models.subject import Subject
|
|
search = request.args.get('search', '').strip()
|
|
|
|
query = Career.query
|
|
if search:
|
|
query = query.filter(
|
|
or_(
|
|
Career.name.ilike(f'%{search}%'),
|
|
Career.code.ilike(f'%{search}%'),
|
|
Career.description.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
careers = query.order_by(Career.name.asc()).all()
|
|
|
|
# Calculate stats per career
|
|
stats = {}
|
|
for c in careers:
|
|
subj_count = len(c.subjects) if c.subjects else 0
|
|
comm_count = sum(len(s.commissions) for s in c.subjects) if c.subjects else 0
|
|
stats[c.id] = {
|
|
'subjects_count': subj_count,
|
|
'commissions_count': comm_count
|
|
}
|
|
|
|
can_edit = current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')
|
|
|
|
return render_template(
|
|
'admin/careers/list.html',
|
|
careers=careers,
|
|
stats=stats,
|
|
search=search,
|
|
can_edit=can_edit
|
|
)
|
|
|
|
|
|
@admin_bp.route('/careers/add', methods=['POST'])
|
|
@login_required
|
|
def career_add():
|
|
"""Create a new academic career"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.career import Career
|
|
name = request.form.get('name', '').strip()
|
|
code = request.form.get('code', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
|
|
if not name:
|
|
flash(_('El nombre de la carrera es obligatorio.'), 'error')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
existing = Career.query.filter(Career.name.ilike(name)).first()
|
|
if existing:
|
|
flash(_('Ya existe una carrera registrada con el nombre "%(name)s".', name=name), 'error')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
career = Career(name=name, code=code or None, description=description or None)
|
|
db.session.add(career)
|
|
db.session.commit()
|
|
|
|
flash(_('Carrera "%(name)s" creada exitosamente.', name=career.name), 'success')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
|
|
@admin_bp.route('/careers/<int:id>/edit', methods=['POST'])
|
|
@login_required
|
|
def career_edit(id):
|
|
"""Edit an academic career"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.career import Career
|
|
career = Career.query.get_or_404(id)
|
|
name = request.form.get('name', '').strip()
|
|
code = request.form.get('code', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
|
|
if not name:
|
|
flash(_('El nombre de la carrera es obligatorio.'), 'error')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
existing = Career.query.filter(Career.name.ilike(name), Career.id != id).first()
|
|
if existing:
|
|
flash(_('Ya existe otra carrera registrada con el nombre "%(name)s".', name=name), 'error')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
career.name = name
|
|
career.code = code or None
|
|
career.description = description or None
|
|
db.session.commit()
|
|
|
|
flash(_('Carrera "%(name)s" actualizada correctamente.', name=career.name), 'success')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
|
|
@admin_bp.route('/careers/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def career_delete(id):
|
|
"""Delete an academic career if it has no associated subjects"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.career import Career
|
|
career = Career.query.get_or_404(id)
|
|
if career.subjects and len(career.subjects) > 0:
|
|
flash(_('No se puede eliminar la carrera "%(name)s" porque tiene %(count)d asignaturas vinculadas.', name=career.name, count=len(career.subjects)), 'error')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
name = career.name
|
|
db.session.delete(career)
|
|
db.session.commit()
|
|
flash(_('Carrera "%(name)s" eliminada permanentemente.', name=name), 'success')
|
|
return redirect(url_for('admin.careers_list'))
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# 2. GESTIÓN DE ASIGNATURAS (SUBJECTS)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/subjects')
|
|
@login_required
|
|
def subjects_list():
|
|
"""List and manage subjects / materias"""
|
|
if not (current_user.has_permission('academic', 'read') or current_user.can_manage()):
|
|
abort(403)
|
|
|
|
from app.models.subject import Subject
|
|
from app.models.career import Career
|
|
|
|
search = request.args.get('search', '').strip()
|
|
career_filter = request.args.get('career_id', type=int)
|
|
active_filter = request.args.get('active', '').strip()
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = 20
|
|
|
|
query = Subject.query
|
|
if search:
|
|
query = query.filter(
|
|
or_(
|
|
Subject.name.ilike(f'%{search}%'),
|
|
Subject.code.ilike(f'%{search}%'),
|
|
Subject.department.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
if career_filter:
|
|
query = query.filter(Subject.career_id == career_filter)
|
|
if active_filter in ['true', 'false']:
|
|
query = query.filter(Subject.is_active == (active_filter == 'true'))
|
|
|
|
pagination = query.order_by(Subject.code.asc(), Subject.name.asc()).paginate(page=page, per_page=per_page, error_out=False)
|
|
careers = Career.query.order_by(Career.name.asc()).all()
|
|
can_edit = current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')
|
|
|
|
return render_template(
|
|
'admin/subjects/list.html',
|
|
subjects=pagination.items,
|
|
pagination=pagination,
|
|
careers=careers,
|
|
search=search,
|
|
career_filter=career_filter,
|
|
active_filter=active_filter,
|
|
can_edit=can_edit
|
|
)
|
|
|
|
|
|
@admin_bp.route('/subjects/add', methods=['POST'])
|
|
@login_required
|
|
def subject_add():
|
|
"""Create a new subject"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.subject import Subject
|
|
from app.models.career import Career
|
|
code = request.form.get('code', '').strip().upper()
|
|
name = request.form.get('name', '').strip()
|
|
department = request.form.get('department', '').strip()
|
|
career_id = request.form.get('career_id', type=int)
|
|
credits = request.form.get('credits', 4, type=int)
|
|
description = request.form.get('description', '').strip()
|
|
|
|
if not code or not name:
|
|
flash(_('El código y el nombre de la asignatura son obligatorios.'), 'error')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
existing = Subject.query.filter_by(code=code).first()
|
|
if existing:
|
|
flash(_('Ya existe una asignatura con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
career_name = None
|
|
if career_id:
|
|
c = Career.query.get(career_id)
|
|
if c:
|
|
career_name = c.name
|
|
|
|
subject = Subject(
|
|
code=code,
|
|
name=name,
|
|
department=department or career_name or 'General',
|
|
career_id=career_id if career_id else None,
|
|
credits=credits or 4,
|
|
description=description or None,
|
|
is_active=True
|
|
)
|
|
db.session.add(subject)
|
|
db.session.commit()
|
|
|
|
flash(_('Asignatura "%(name)s" (%(code)s) creada exitosamente.', name=subject.name, code=subject.code), 'success')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
|
|
@admin_bp.route('/subjects/<int:id>/edit', methods=['POST'])
|
|
@login_required
|
|
def subject_edit(id):
|
|
"""Edit an existing subject"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.subject import Subject
|
|
from app.models.career import Career
|
|
subject = Subject.query.get_or_404(id)
|
|
|
|
code = request.form.get('code', '').strip().upper()
|
|
name = request.form.get('name', '').strip()
|
|
department = request.form.get('department', '').strip()
|
|
career_id = request.form.get('career_id', type=int)
|
|
credits = request.form.get('credits', 4, type=int)
|
|
description = request.form.get('description', '').strip()
|
|
is_active = request.form.get('is_active') == 'on'
|
|
|
|
if not code or not name:
|
|
flash(_('El código y el nombre de la asignatura son obligatorios.'), 'error')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
existing = Subject.query.filter(Subject.code == code, Subject.id != id).first()
|
|
if existing:
|
|
flash(_('Ya existe otra asignatura con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
career_name = None
|
|
if career_id:
|
|
c = Career.query.get(career_id)
|
|
if c:
|
|
career_name = c.name
|
|
|
|
subject.code = code
|
|
subject.name = name
|
|
subject.department = department or career_name or subject.department
|
|
subject.career_id = career_id if career_id else None
|
|
subject.credits = credits or 4
|
|
subject.description = description or None
|
|
subject.is_active = is_active
|
|
db.session.commit()
|
|
|
|
flash(_('Asignatura "%(name)s" actualizada exitosamente.', name=subject.name), 'success')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
|
|
@admin_bp.route('/subjects/<int:id>/toggle', methods=['POST'])
|
|
@login_required
|
|
def subject_toggle(id):
|
|
"""Toggle active status of a subject"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.subject import Subject
|
|
subject = Subject.query.get_or_404(id)
|
|
subject.is_active = not subject.is_active
|
|
db.session.commit()
|
|
|
|
status_str = _('activada') if subject.is_active else _('desactivada')
|
|
flash(_('Asignatura "%(name)s" %(status)s.', name=subject.name, status=status_str), 'success')
|
|
return redirect(url_for('admin.subjects_list'))
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# 3. GESTIÓN DE CICLO LECTIVO (ACADEMIC TERMS)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/academic-terms')
|
|
@login_required
|
|
def academic_terms_list():
|
|
"""List and manage academic cycles / terms"""
|
|
if not (current_user.has_permission('academic', 'read') or current_user.can_manage()):
|
|
abort(403)
|
|
|
|
from app.models.academic_term import AcademicTerm
|
|
from app.models.subject import Commission
|
|
from app.models.reservation import Reservation
|
|
|
|
terms = AcademicTerm.query.order_by(AcademicTerm.year.desc(), AcademicTerm.code.desc()).all()
|
|
|
|
# Calculate associated commissions and reservations
|
|
stats = {}
|
|
for t in terms:
|
|
comm_count = Commission.query.filter(
|
|
or_(Commission.semester == t.code, db.and_(Commission.year == t.year, Commission.semester.ilike(f'%{t.code}%')))
|
|
).count()
|
|
stats[t.id] = {
|
|
'commissions_count': comm_count
|
|
}
|
|
|
|
can_edit = current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')
|
|
|
|
return render_template(
|
|
'admin/academic_terms/list.html',
|
|
terms=terms,
|
|
stats=stats,
|
|
can_edit=can_edit
|
|
)
|
|
|
|
|
|
@admin_bp.route('/academic-terms/add', methods=['POST'])
|
|
@login_required
|
|
def academic_term_add():
|
|
"""Create a new academic cycle"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.academic_term import AcademicTerm
|
|
from datetime import datetime
|
|
code = request.form.get('code', '').strip()
|
|
name = request.form.get('name', '').strip()
|
|
year = request.form.get('year', type=int)
|
|
period_type = request.form.get('period_type', '').strip()
|
|
start_date_str = request.form.get('start_date', '').strip()
|
|
end_date_str = request.form.get('end_date', '').strip()
|
|
is_current = request.form.get('is_current') == 'on'
|
|
|
|
if not code or not name or not year:
|
|
flash(_('El código, nombre y año del ciclo lectivo son obligatorios.'), 'error')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
existing = AcademicTerm.query.filter_by(code=code).first()
|
|
if existing:
|
|
flash(_('Ya existe un ciclo lectivo con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date() if start_date_str else None
|
|
end_date = datetime.strptime(end_date_str, '%Y-%m-%d').date() if end_date_str else None
|
|
|
|
if is_current:
|
|
# Unset previous current terms
|
|
AcademicTerm.query.update({'is_current': False})
|
|
|
|
term = AcademicTerm(
|
|
code=code,
|
|
name=name,
|
|
year=year,
|
|
period_type=period_type or 'Segundo Cuatrimestre',
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
is_current=is_current,
|
|
is_active=True
|
|
)
|
|
db.session.add(term)
|
|
db.session.commit()
|
|
|
|
flash(_('Ciclo lectivo "%(name)s" creado exitosamente.', name=term.name), 'success')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
|
|
@admin_bp.route('/academic-terms/<int:id>/edit', methods=['POST'])
|
|
@login_required
|
|
def academic_term_edit(id):
|
|
"""Edit an academic cycle"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.academic_term import AcademicTerm
|
|
from datetime import datetime
|
|
term = AcademicTerm.query.get_or_404(id)
|
|
|
|
code = request.form.get('code', '').strip()
|
|
name = request.form.get('name', '').strip()
|
|
year = request.form.get('year', type=int)
|
|
period_type = request.form.get('period_type', '').strip()
|
|
start_date_str = request.form.get('start_date', '').strip()
|
|
end_date_str = request.form.get('end_date', '').strip()
|
|
is_current = request.form.get('is_current') == 'on'
|
|
is_active = request.form.get('is_active') == 'on'
|
|
|
|
if not code or not name or not year:
|
|
flash(_('El código, nombre y año del ciclo lectivo son obligatorios.'), 'error')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
existing = AcademicTerm.query.filter(AcademicTerm.code == code, AcademicTerm.id != id).first()
|
|
if existing:
|
|
flash(_('Ya existe otro ciclo lectivo con el código "%(code)s".', code=code), 'error')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
if is_current and not term.is_current:
|
|
AcademicTerm.query.update({'is_current': False})
|
|
|
|
term.code = code
|
|
term.name = name
|
|
term.year = year
|
|
term.period_type = period_type or term.period_type
|
|
term.start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date() if start_date_str else None
|
|
term.end_date = datetime.strptime(end_date_str, '%Y-%m-%d').date() if end_date_str else None
|
|
term.is_current = is_current
|
|
term.is_active = is_active
|
|
db.session.commit()
|
|
|
|
flash(_('Ciclo lectivo "%(name)s" actualizado correctamente.', name=term.name), 'success')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
|
|
@admin_bp.route('/academic-terms/<int:id>/set-current', methods=['POST'])
|
|
@login_required
|
|
def academic_term_set_current(id):
|
|
"""Set an academic cycle as the currently active one"""
|
|
if not (current_user.has_permission('academic', 'read_write') or (current_user.role and current_user.role.upper() == 'ADMIN')):
|
|
abort(403)
|
|
|
|
from app.models.academic_term import AcademicTerm
|
|
term = AcademicTerm.query.get_or_404(id)
|
|
AcademicTerm.query.update({'is_current': False})
|
|
term.is_current = True
|
|
term.is_active = True
|
|
db.session.commit()
|
|
|
|
flash(_('El ciclo lectivo "%(name)s" fue establecido como Ciclo Activo del sistema.', name=term.name), 'success')
|
|
return redirect(url_for('admin.academic_terms_list'))
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
# 4. GESTIÓN DE OCUPACIÓN Y MÉTRICAS (OCCUPANCY DASHBOARD)
|
|
# ---------------------------------------------------------
|
|
|
|
@admin_bp.route('/occupancy-metrics')
|
|
@login_required
|
|
def occupancy_metrics():
|
|
"""Consolidated institutional dashboard of classroom occupancy, shifts, and space usage"""
|
|
if not (current_user.has_permission('metrics', 'read') or current_user.can_manage()):
|
|
abort(403)
|
|
|
|
from app.models.classroom import Classroom
|
|
from app.models.building import Building
|
|
from app.models.reservation import Reservation, ReservationStatus
|
|
import calendar
|
|
from datetime import datetime, date, timedelta
|
|
|
|
today = datetime.utcnow().date()
|
|
today_start = datetime.combine(today, datetime.min.time())
|
|
today_end = datetime.combine(today, datetime.max.time())
|
|
|
|
week_start = datetime.combine(today - timedelta(days=today.weekday()), datetime.min.time())
|
|
week_end = datetime.combine(week_start.date() + timedelta(days=6), datetime.max.time())
|
|
|
|
_, last_day = calendar.monthrange(today.year, today.month)
|
|
month_start = datetime(today.year, today.month, 1, 0, 0, 0)
|
|
month_end = datetime(today.year, today.month, last_day, 23, 59, 59)
|
|
|
|
# Buildings & Classrooms
|
|
buildings = Building.query.filter_by(is_active=True).all()
|
|
vb = next((b for b in buildings if b.code == 'VIRTUAL' or 'virtual' in b.name.lower()), None)
|
|
vb_id = vb.id if vb else 0
|
|
|
|
physical_rooms = Classroom.query.filter(
|
|
Classroom.is_active == True,
|
|
Classroom.building_id != vb_id if vb_id else True,
|
|
~Classroom.code.ilike('%VIRTUAL%')
|
|
).all()
|
|
|
|
vc = Classroom.query.filter(
|
|
Classroom.building_id == vb_id if vb_id else Classroom.code.ilike('%VIRTUAL%')
|
|
).first()
|
|
|
|
# Reservations in month
|
|
month_res = Reservation.query.filter(
|
|
Reservation.start_time >= month_start,
|
|
Reservation.start_time <= month_end,
|
|
Reservation.status.in_([ReservationStatus.CONFIRMED.value, ReservationStatus.PENDING.value])
|
|
).all()
|
|
|
|
# Calculate physical stats
|
|
DAILY_OP_MINS = 15 * 60
|
|
WEEKLY_OP_MINS = 6 * 15 * 60
|
|
MONTHLY_OP_MINS = 24 * 15 * 60
|
|
|
|
room_metrics = []
|
|
for c in physical_rooms:
|
|
c_res = [r for r in month_res if r.classroom_id == c.id]
|
|
day_res = [r for r in c_res if today_start <= r.start_time <= today_end]
|
|
day_mins = sum((min(r.end_time, today_end) - max(r.start_time, today_start)).total_seconds() / 60 for r in day_res)
|
|
day_pct = min(100, int(round((day_mins / DAILY_OP_MINS) * 100)))
|
|
|
|
week_res = [r for r in c_res if week_start <= r.start_time <= week_end]
|
|
week_mins = sum((min(r.end_time, week_end) - max(r.start_time, week_start)).total_seconds() / 60 for r in week_res)
|
|
week_pct = min(100, int(round((week_mins / WEEKLY_OP_MINS) * 100)))
|
|
|
|
month_mins = sum((min(r.end_time, month_end) - max(r.start_time, month_start)).total_seconds() / 60 for r in c_res)
|
|
month_pct = min(100, int(round((month_mins / MONTHLY_OP_MINS) * 100)))
|
|
|
|
room_metrics.append({
|
|
'classroom': c,
|
|
'day_pct': day_pct,
|
|
'day_count': len(day_res),
|
|
'week_pct': week_pct,
|
|
'week_count': len(week_res),
|
|
'month_pct': month_pct,
|
|
'month_count': len(c_res)
|
|
})
|
|
|
|
room_metrics.sort(key=lambda x: x['month_pct'], reverse=True)
|
|
top_rooms = room_metrics[:5]
|
|
low_rooms = list(reversed(room_metrics))[:5]
|
|
|
|
# Global averages
|
|
avg_day_pct = int(round(sum(rm['day_pct'] for rm in room_metrics) / len(room_metrics))) if room_metrics else 0
|
|
avg_week_pct = int(round(sum(rm['week_pct'] for rm in room_metrics) / len(room_metrics))) if room_metrics else 0
|
|
avg_month_pct = int(round(sum(rm['month_pct'] for rm in room_metrics) / len(room_metrics))) if room_metrics else 0
|
|
|
|
# Virtual classes
|
|
v_res = [r for r in month_res if vc and r.classroom_id == vc.id]
|
|
v_today = sum(1 for r in v_res if today_start <= r.start_time <= today_end)
|
|
v_week = sum(1 for r in v_res if week_start <= r.start_time <= week_end)
|
|
v_month = len(v_res)
|
|
|
|
# Shifts distribution
|
|
shift_counts = {'Mañana': 0, 'Tarde': 0, 'Vespertino': 0, 'Noche': 0, 'Otro': 0}
|
|
for r in month_res:
|
|
s = (r.shift or '').strip().capitalize()
|
|
if s in shift_counts:
|
|
shift_counts[s] += 1
|
|
elif 'Mañana' in s:
|
|
shift_counts['Mañana'] += 1
|
|
elif 'Tarde' in s:
|
|
shift_counts['Tarde'] += 1
|
|
elif 'Noche' in s:
|
|
shift_counts['Noche'] += 1
|
|
else:
|
|
shift_counts['Otro'] += 1
|
|
|
|
# Floors occupancy
|
|
floor_stats = {}
|
|
for rm in room_metrics:
|
|
flr = rm['classroom'].floor_display
|
|
if flr not in floor_stats:
|
|
floor_stats[flr] = {'rooms': 0, 'month_pct_sum': 0}
|
|
floor_stats[flr]['rooms'] += 1
|
|
floor_stats[flr]['month_pct_sum'] += rm['month_pct']
|
|
|
|
floor_summary = [
|
|
{'floor': f, 'avg_pct': int(round(data['month_pct_sum'] / data['rooms'])), 'count': data['rooms']}
|
|
for f, data in floor_stats.items()
|
|
]
|
|
floor_summary.sort(key=lambda x: x['avg_pct'], reverse=True)
|
|
|
|
date_filters = {
|
|
'today': today.strftime('%Y-%m-%d'),
|
|
'week_start': week_start.strftime('%Y-%m-%d'),
|
|
'week_end': week_end.strftime('%Y-%m-%d'),
|
|
'month_start': month_start.strftime('%Y-%m-%d'),
|
|
'month_end': month_end.strftime('%Y-%m-%d')
|
|
}
|
|
|
|
return render_template(
|
|
'admin/metrics/occupancy.html',
|
|
physical_rooms_count=len(physical_rooms),
|
|
avg_day_pct=avg_day_pct,
|
|
avg_week_pct=avg_week_pct,
|
|
avg_month_pct=avg_month_pct,
|
|
top_rooms=top_rooms,
|
|
low_rooms=low_rooms,
|
|
v_today=v_today,
|
|
v_week=v_week,
|
|
v_month=v_month,
|
|
shift_counts=shift_counts,
|
|
floor_summary=floor_summary,
|
|
total_month_classes=len(month_res),
|
|
date_filters=date_filters
|
|
)
|
|
|
|
|
|
|