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.models.audit_log import AuditLog 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/', 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/', 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/', 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/', 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/', 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/', 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/', 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/', 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//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//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//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//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//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//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 ) # --------------------------------------------------------- # AUDIT LOGS # --------------------------------------------------------- @admin_bp.route('/audit-logs') @admin_required def audit_logs(): """Vista de bitácora de auditoría del sistema""" page = request.args.get('page', 1, type=int) module_filter = request.args.get('module', '').strip() action_filter = request.args.get('action', '').strip() search = request.args.get('search', '').strip() per_page = 20 query = AuditLog.query if module_filter: query = query.filter(AuditLog.module == module_filter) if action_filter: query = query.filter(AuditLog.action == action_filter) if search: query = query.filter( or_( AuditLog.user_email.ilike(f'%{search}%'), AuditLog.details.ilike(f'%{search}%'), AuditLog.ip_address.ilike(f'%{search}%') ) ) pagination = query.order_by(AuditLog.created_at.desc()).paginate(page=page, per_page=per_page, error_out=False) # Módulos y acciones únicas para los selectores de filtro modules = [r[0] for r in db.session.query(AuditLog.module).distinct().all() if r[0]] actions = [r[0] for r in db.session.query(AuditLog.action).distinct().all() if r[0]] return render_template( 'admin/audit_logs.html', pagination=pagination, audit_logs=pagination.items, modules=modules, actions=actions, current_module=module_filter, current_action=action_filter, search=search ) # --------------------------------------------------------- # 5. GESTION DE COMISIONES (BEDELIA / ADMIN) # --------------------------------------------------------- @admin_bp.route('/commissions') @login_required def commissions_list(): """List all commissions with filters - accessible by Bedelia and Admin""" if not (current_user.has_permission('academic', 'read') or current_user.can_manage()): abort(403) from app.models.subject import Subject, Commission from app.models.user import User search = request.args.get('search', '').strip() subject_filter = request.args.get('subject_id', type=int) teacher_filter = request.args.get('teacher_id', type=int) shift_filter = request.args.get('shift', '').strip() active_filter = request.args.get('active', 'true').strip() page = request.args.get('page', 1, type=int) per_page = 20 query = Commission.query.join(Subject, Commission.subject_id == Subject.id) if search: query = query.filter( or_( Commission.code.ilike(f'%{search}%'), Subject.name.ilike(f'%{search}%'), Subject.code.ilike(f'%{search}%'), ) ) if subject_filter: query = query.filter(Commission.subject_id == subject_filter) if teacher_filter: query = query.filter(Commission.teacher_id == teacher_filter) if shift_filter: query = query.filter(Commission.shift.ilike(f'%{shift_filter}%')) if active_filter in ['true', 'false']: query = query.filter(Commission.active == (active_filter == 'true')) pagination = query.order_by(Subject.name.asc(), Commission.code.asc()).paginate( page=page, per_page=per_page, error_out=False ) subjects = Subject.query.filter_by(is_active=True).order_by(Subject.name.asc()).all() teachers = User.query.filter( or_(User.role.ilike('%DOCENTE%'), User.role.ilike('%docent%'), User.role.ilike('%ADMIN%')) ).filter_by(is_active=True).order_by(User.name.asc()).all() can_edit = ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ) return render_template( 'admin/commissions/list.html', commissions=pagination.items, pagination=pagination, subjects=subjects, teachers=teachers, search=search, subject_filter=subject_filter, teacher_filter=teacher_filter, shift_filter=shift_filter, active_filter=active_filter, can_edit=can_edit ) @admin_bp.route('/commissions/') @login_required def commission_detail(id): """Detail view of a commission: enrolled students and teacher assignment""" if not (current_user.has_permission('academic', 'read') or current_user.can_manage()): abort(403) from app.models.subject import Commission from app.models.enrollment import StudentEnrollment from app.models.user import User commission = Commission.query.get_or_404(id) enrollments = ( StudentEnrollment.query .filter_by(commission_id=id) .order_by(StudentEnrollment.enrolled_at.asc()) .all() ) teachers = User.query.filter( or_(User.role.ilike('%DOCENTE%'), User.role.ilike('%docent%')) ).filter_by(is_active=True).order_by(User.name.asc()).all() students = User.query.filter( or_(User.role.ilike('%ALUMNO%'), User.role.ilike('%alumn%'), User.role.ilike('%estud%')) ).filter_by(is_active=True).order_by(User.name.asc()).all() enrolled_ids = {e.student_id for e in enrollments} can_edit = ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ) return render_template( 'admin/commissions/detail.html', commission=commission, enrollments=enrollments, teachers=teachers, students=students, enrolled_ids=enrolled_ids, can_edit=can_edit ) @admin_bp.route('/commissions//assign-teacher', methods=['POST']) @login_required def commission_assign_teacher(id): """Assign or change the teacher of a commission""" if not ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ): abort(403) from app.models.subject import Commission from app.models.user import User commission = Commission.query.get_or_404(id) teacher_id = request.form.get('teacher_id', type=int) if teacher_id: teacher = User.query.get(teacher_id) if not teacher or not teacher.is_active: flash(_('El docente seleccionado no existe o esta inactivo.'), 'error') return redirect(url_for('admin.commission_detail', id=id)) commission.teacher_id = teacher_id db.session.commit() flash(_('Docente asignado exitosamente a la comision.'), 'success') else: commission.teacher_id = None db.session.commit() flash(_('Se removio la asignacion de docente.'), 'warning') return redirect(url_for('admin.commission_detail', id=id)) @admin_bp.route('/commissions//enroll-student', methods=['POST']) @login_required def commission_enroll_student(id): """Enroll a student into a commission""" if not ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ): abort(403) from app.models.subject import Commission from app.models.enrollment import StudentEnrollment from app.models.user import User commission = Commission.query.get_or_404(id) student_id = request.form.get('student_id', type=int) notes = request.form.get('notes', '').strip() if not student_id: flash(_('Debes seleccionar un alumno.'), 'error') return redirect(url_for('admin.commission_detail', id=id)) student = User.query.get(student_id) if not student or not student.is_active: flash(_('El alumno seleccionado no existe o esta inactivo.'), 'error') return redirect(url_for('admin.commission_detail', id=id)) existing = StudentEnrollment.query.filter_by(student_id=student_id, commission_id=id).first() if existing: flash(_('El alumno ya esta inscripto en esta comision.'), 'warning') return redirect(url_for('admin.commission_detail', id=id)) enrollment = StudentEnrollment( student_id=student_id, commission_id=id, status='activo', notes=notes or None ) db.session.add(enrollment) commission.current_students = (commission.current_students or 0) + 1 db.session.commit() flash(_('Alumno inscripto exitosamente en la comision.'), 'success') return redirect(url_for('admin.commission_detail', id=id)) @admin_bp.route('/commissions//unenroll-student/', methods=['POST']) @login_required def commission_unenroll_student(id, student_id): """Remove a student from a commission""" if not ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ): abort(403) from app.models.subject import Commission from app.models.enrollment import StudentEnrollment commission = Commission.query.get_or_404(id) enrollment = StudentEnrollment.query.filter_by( student_id=student_id, commission_id=id ).first_or_404() db.session.delete(enrollment) if commission.current_students and commission.current_students > 0: commission.current_students -= 1 db.session.commit() flash(_('Alumno dado de baja de la comision.'), 'success') return redirect(url_for('admin.commission_detail', id=id)) @admin_bp.route('/commissions//update-enrollment/', methods=['POST']) @login_required def commission_update_enrollment(id, student_id): """Update enrollment status for a student""" if not ( current_user.has_permission('academic', 'read_write') or current_user.is_admin() or current_user.is_bedelia() ): abort(403) from app.models.enrollment import StudentEnrollment enrollment = StudentEnrollment.query.filter_by( student_id=student_id, commission_id=id ).first_or_404() new_status = request.form.get('status', 'activo') if new_status not in ['activo', 'retirado', 'condicional']: new_status = 'activo' enrollment.status = new_status enrollment.notes = request.form.get('notes', enrollment.notes or '') db.session.commit() flash(_('Estado de inscripcion actualizado.'), 'success') return redirect(url_for('admin.commission_detail', id=id))