feat: completar fases 3 y 4 del roadmap mvp (gradebook, drag-and-drop, impersonacion) y actualizar README

This commit is contained in:
2026-09-23 12:10:31 -03:00
parent 4ebc5df4fd
commit 7ad4e3f099
38 changed files with 3804 additions and 390 deletions
+2
View File
@@ -30,6 +30,7 @@ def create_app(config_class=Config):
from app.routes.api.dashboard import api_dashboard_bp
from app.routes.api.admin import api_admin_bp
from app.routes.api.openapi import api_openapi_bp
from app.routes.api.students import api_students_bp
from app.routes.genetic_algorithm import genetic_bp
from app.security import SecurityFilterChain
@@ -41,6 +42,7 @@ def create_app(config_class=Config):
app.register_blueprint(api_dashboard_bp)
app.register_blueprint(api_openapi_bp)
app.register_blueprint(genetic_bp)
app.register_blueprint(api_students_bp, url_prefix='/api/v1')
app.register_blueprint(api_admin_bp, url_prefix='/api/v1')
app.register_blueprint(api_admin_bp, url_prefix='/api/v1/admin', name='api_admin_prefixed')
+3 -1
View File
@@ -11,6 +11,7 @@ from .milestone import MilestoneType, AcademicMilestone, BASE_MILESTONE_TYPES
from .audit_log import AuditLog
from .enrollment import StudentEnrollment
from .setting import SystemSetting
from .grade import MilestoneGrade
__all__ = [
'User',
@@ -34,5 +35,6 @@ __all__ = [
'BASE_MILESTONE_TYPES',
'AuditLog',
'StudentEnrollment',
'SystemSetting'
'SystemSetting',
'MilestoneGrade'
]
+50 -12
View File
@@ -390,28 +390,61 @@ class ReservationOptimizer:
}
def _calculate_preferred_time(self, commission: Commission) -> Dict:
"""Calculate preferred time slots for a commission"""
# This is a simplified version - would need to be based on actual semester schedule
"""Calculate preferred time slots for a commission based on shift and schedule"""
current_date = datetime.utcnow()
# Default to weekday mornings for regular classes
days_ahead = (0 - current_date.weekday()) % 7 # Next Monday
if days_ahead == 0: # If today is Monday, use next week
# Determine day of week from commission schedule or default to Monday
target_weekday = 0 # Monday
sched = (commission.schedule or '').lower()
if 'martes' in sched:
target_weekday = 1
elif 'miercoles' in sched or 'miércoles' in sched:
target_weekday = 2
elif 'jueves' in sched:
target_weekday = 3
elif 'viernes' in sched:
target_weekday = 4
elif 'sabado' in sched or 'sábado' in sched:
target_weekday = 5
days_ahead = (target_weekday - current_date.weekday()) % 7
if days_ahead == 0:
days_ahead = 7
preferred_date = current_date + timedelta(days=days_ahead)
# Determine hours based on shift
shift = (commission.shift or '').strip().lower()
if 'mañana' in shift or 'manana' in shift:
start_h, start_m = 8, 0
end_h, end_m = 12, 0
elif 'tarde' in shift:
start_h, start_m = 14, 0
end_h, end_m = 18, 0
elif 'noche' in shift or 'vespertino' in shift:
start_h, start_m = 18, 30
end_h, end_m = 22, 30
else:
start_h, start_m = 9, 0
end_h, end_m = 13, 0
return {
'start': preferred_date.replace(hour=9, minute=0, second=0, microsecond=0),
'end': preferred_date.replace(hour=11, minute=0, second=0, microsecond=0)
'start': preferred_date.replace(hour=start_h, minute=start_m, second=0, microsecond=0),
'end': preferred_date.replace(hour=end_h, minute=end_m, second=0, microsecond=0)
}
def apply_optimized_reservations(self, reservation_data: List[Dict]) -> List[Reservation]:
"""Apply the optimized reservations to the database"""
"""Apply the optimized reservations to the database with ISO string parsing and conflict avoidance"""
applied_reservations = []
for data in reservation_data:
# Check for conflicts before creating
for item in reservation_data:
data = dict(item)
if isinstance(data.get('start_time'), str):
data['start_time'] = datetime.fromisoformat(data['start_time'].replace('Z', '+00:00'))
if isinstance(data.get('end_time'), str):
data['end_time'] = datetime.fromisoformat(data['end_time'].replace('Z', '+00:00'))
# Check for physical room conflicts before creating
conflicts = Reservation.find_conflicts(
data['classroom_id'],
data['start_time'],
@@ -419,12 +452,17 @@ class ReservationOptimizer:
)
if not conflicts:
# Ensure status enum value compatibility
if 'status' in data and isinstance(data['status'], str):
try:
data['status'] = ReservationStatus(data['status']).value
except ValueError:
data['status'] = ReservationStatus.CONFIRMED.value
reservation = Reservation(**data)
db.session.add(reservation)
applied_reservations.append(reservation)
else:
# Log conflict - could be handled differently
print(f"Conflict detected for reservation data: {data}")
current_app.logger.warning(f"Conflict detected for reservation during genetic apply: {data}")
try:
db.session.commit()
+92
View File
@@ -0,0 +1,92 @@
from datetime import datetime
from app import db
class MilestoneGrade(db.Model):
"""
Calificación y seguimiento pedagógico individual de un alumno en un hito evaluativo (Fase 3 MVP).
Vincula un AcademicMilestone con un estudiante matriculado (StudentEnrollment / User).
"""
__tablename__ = 'milestone_grades'
id = db.Column(db.Integer, primary_key=True)
milestone_id = db.Column(
db.Integer,
db.ForeignKey('academic_milestones.id', ondelete='CASCADE'),
nullable=False,
index=True
)
enrollment_id = db.Column(
db.Integer,
db.ForeignKey('student_enrollments.id', ondelete='CASCADE'),
nullable=False,
index=True
)
student_id = db.Column(
db.Integer,
db.ForeignKey('users.id', ondelete='CASCADE'),
nullable=False,
index=True
)
score = db.Column(db.Float, nullable=True) # Calificación numérica de 1.0 a 10.0
is_absent = db.Column(db.Boolean, default=False) # Flag de Ausente
feedback = db.Column(db.Text, nullable=True) # Devolución formativa/pedagógica
graded_by_user_id = db.Column(
db.Integer,
db.ForeignKey('users.id', ondelete='SET NULL'),
nullable=True
)
graded_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Restricción de unicidad: un estudiante tiene una única nota por hito
__table_args__ = (
db.UniqueConstraint('milestone_id', 'student_id', name='uq_milestone_student'),
)
# Relaciones
milestone = db.relationship('AcademicMilestone', backref=db.backref('grades', lazy='dynamic', cascade='all, delete-orphan'))
enrollment = db.relationship('StudentEnrollment', backref=db.backref('milestone_grades', lazy='dynamic', cascade='all, delete-orphan'))
student = db.relationship('User', foreign_keys=[student_id], backref=db.backref('received_grades', lazy='dynamic'))
graded_by = db.relationship('User', foreign_keys=[graded_by_user_id])
@property
def is_passed(self) -> bool:
"""Determina si la nota está aprobada (>= 4.00 y no ausente)."""
if self.is_absent or self.score is None:
return False
return self.score >= 4.0
@property
def display_score(self) -> str:
"""Representación textual estándar: 'Ausente', '7.5' o '-'."""
if self.is_absent:
return 'Ausente'
if self.score is not None:
return f'{self.score:.1f}' if self.score % 1 != 0 else f'{int(self.score)}'
return '-'
def to_dict(self):
return {
'id': self.id,
'milestone_id': self.milestone_id,
'milestone_title': self.milestone.title if self.milestone else None,
'milestone_type': self.milestone.tipo if self.milestone else None,
'milestone_color': self.milestone.color if self.milestone else None,
'enrollment_id': self.enrollment_id,
'student_id': self.student_id,
'student_name': self.student.name if self.student else None,
'student_email': self.student.email if self.student else None,
'score': self.score,
'display_score': self.display_score,
'is_absent': bool(self.is_absent),
'is_passed': self.is_passed,
'feedback': self.feedback or '',
'graded_by_user_id': self.graded_by_user_id,
'graded_by_name': self.graded_by.name if self.graded_by else None,
'graded_at': self.graded_at.isoformat() if self.graded_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f'<MilestoneGrade milestone={self.milestone_id} student={self.student_id} score={self.display_score}>'
+13
View File
@@ -54,6 +54,13 @@ class Commission(db.Model):
shift = db.Column(db.String(30), nullable=True) # Mañana, Tarde, Vespertino, Noche
virtual_link = db.Column(db.String(500), nullable=True) # Link Zoom, Meet, Teams, Moodle
active = db.Column(db.Boolean, default=True) # Changed from is_active to active
# Cierre de Actas y Libro de Calificaciones (Fase 3 MVP)
grades_closed = db.Column(db.Boolean, default=False, nullable=True)
grades_closed_at = db.Column(db.DateTime, nullable=True)
grades_closed_by = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='SET NULL'), nullable=True)
acta_number = db.Column(db.String(50), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@@ -69,6 +76,7 @@ class Commission(db.Model):
# Relationships
pass # Reservation already defines the relationship back to commission
teacher = db.relationship('User', backref='teaching_commissions', foreign_keys=[teacher_id])
closed_by_user = db.relationship('User', foreign_keys=[grades_closed_by])
# Virtual field for teacher_name
@property
@@ -125,6 +133,11 @@ class Commission(db.Model):
'shift': self.shift,
'virtual_link': self.virtual_link,
'active': self.active,
'grades_closed': bool(self.grades_closed),
'grades_closed_at': self.grades_closed_at.isoformat() if self.grades_closed_at else None,
'grades_closed_by': self.grades_closed_by,
'closed_by_name': self.closed_by_user.name if self.closed_by_user else None,
'acta_number': self.acta_number,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'full_code': self.get_full_code(),
+266 -1
View File
@@ -1,4 +1,4 @@
from flask import Blueprint, jsonify, request
from flask import Blueprint, jsonify, request, g
from datetime import datetime
import re
from app.utils.jwt_decorators import jwt_required
@@ -6,11 +6,14 @@ from app.models.user import User
from app.models.role import Role, Permission, SYSTEM_MODULES
from app.models.subject import Subject, Commission, CommissionTeacher
from app.models.career import Career
from app.models.classroom import Classroom
from app.models.reservation import Reservation, ReservationStatus
from app.models.academic_term import AcademicTerm
from app.models.milestone import MilestoneType, AcademicMilestone
from app.models.audit_log import AuditLog
from app.constants.document_types import DOCUMENT_TYPES, validate_document, format_document
from app.services.enrollment_service import EnrollmentService
from app.services.gradebook_service import GradebookService
from app import db
api_admin_bp = Blueprint('api_admin', __name__)
@@ -1032,6 +1035,95 @@ def update_commission_enrollment(id, student_id):
'enrollment': enrollment.to_dict()
}), 200
# ---------------------------------------------------------
# COMMISSION GRADEBOOK & ACTAS (FASE 3 MVP)
# ---------------------------------------------------------
@api_admin_bp.route('/commissions/<int:id>/gradebook', methods=['GET'])
@jwt_required
def get_commission_gradebook(id):
"""Obtiene la matriz completa de calificaciones de la comisión."""
try:
data = GradebookService.get_commission_gradebook(id)
return jsonify({'status': 'success', 'data': data}), 200
except ValueError as e:
return jsonify({'error': 'NotFound', 'message': str(e)}), 404
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
@api_admin_bp.route('/commissions/<int:id>/gradebook/grades', methods=['POST', 'PUT'])
@jwt_required
def save_commission_grades(id):
"""Carga masiva o individual de notas con autoguardado asíncrono."""
data = request.get_json(silent=True) or {}
grader_id = getattr(request, 'current_user_id', None)
try:
if 'grades' in data and isinstance(data['grades'], list):
res = GradebookService.bulk_save_grades(
commission_id=id,
grades_data=data['grades'],
grader_id=grader_id
)
return jsonify({'status': 'success', 'message': f"{res['saved_count']} calificaciones guardadas.", 'data': res}), 200
else:
milestone_id = data.get('milestone_id')
student_id = data.get('student_id')
score = data.get('score')
is_absent = bool(data.get('is_absent', False))
feedback = data.get('feedback', '')
if not milestone_id or not student_id:
return jsonify({'error': 'BadRequest', 'message': 'milestone_id y student_id son obligatorios.'}), 400
grade = GradebookService.save_single_grade(
commission_id=id,
milestone_id=int(milestone_id),
student_id=int(student_id),
score=float(score) if score is not None and str(score).strip() != '' else None,
is_absent=is_absent,
feedback=feedback,
grader_id=grader_id
)
return jsonify({'status': 'success', 'message': 'Calificación guardada.', 'grade': grade.to_dict()}), 200
except ValueError as e:
return jsonify({'error': 'ValidationError', 'message': str(e)}), 400
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
@api_admin_bp.route('/commissions/<int:id>/gradebook/close', methods=['POST'])
@jwt_required
def close_commission_gradebook(id):
"""Cierre formal del acta de regularidad y promoción."""
data = request.get_json(silent=True) or {}
user_id = getattr(request, 'current_user_id', None) or 1
notes = data.get('notes', '')
try:
res = GradebookService.close_gradebook(commission_id=id, user_id=user_id, notes=notes)
return jsonify({'status': 'success', **res}), 200
except ValueError as e:
return jsonify({'error': 'BadRequest', 'message': str(e)}), 400
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
@api_admin_bp.route('/commissions/<int:id>/gradebook/reopen', methods=['POST'])
@jwt_required
def reopen_commission_gradebook(id):
"""Reapertura administrativa del acta de calificaciones."""
data = request.get_json(silent=True) or {}
user_id = getattr(request, 'current_user_id', None) or 1
reason = data.get('reason', '')
try:
res = GradebookService.reopen_gradebook(commission_id=id, user_id=user_id, reason=reason)
return jsonify({'status': 'success', **res}), 200
except ValueError as e:
return jsonify({'error': 'BadRequest', 'message': str(e)}), 400
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
# ---------------------------------------------------------
# ACADEMIC TERMS CRUD
# ---------------------------------------------------------
@@ -1376,4 +1468,177 @@ def reset_academic_data_api():
'message': result.get('error', 'Error durante la purga de datos.')
}), 500
# ---------------------------------------------------------
# FASE 4: MODO IMPERSONACIÓN & DRAG-AND-DROP (ESTADIO 4.1 & 4.2)
# ---------------------------------------------------------
@api_admin_bp.route('/admin/impersonate', methods=['POST'])
@jwt_required
def impersonate_user():
"""Iniciar sesión bajo la identidad de otro usuario (Superadmin / Bedelía)."""
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
if not admin_user.is_admin():
return jsonify({'error': 'Forbidden', 'message': 'Solo administradores pueden impersonar usuarios.'}), 403
data = request.get_json(silent=True) or {}
target_id = data.get('target_user_id') or data.get('user_id')
target_email = data.get('target_email') or data.get('email')
target = None
if target_id:
target = db.session.get(User, int(target_id))
elif target_email:
target = User.query.filter_by(email=target_email.strip()).first()
if not target or not target.is_active:
return jsonify({'error': 'NotFound', 'message': 'Usuario objetivo no encontrado o inactivo.'}), 404
audit = AuditLog(
user_id=admin_user.id,
user_email=admin_user.email,
action='IMPERSONATE_START',
module='auth',
entity_id=target.id,
details=f"Administrador {admin_user.email} inició sesión temporal como {target.email} ({target.role})"
)
db.session.add(audit)
db.session.commit()
return jsonify({
'status': 'success',
'message': f"Impersonando exitosamente a {target.name or target.email}",
'target_user': target.to_dict(),
'real_admin_id': admin_user.id
}), 200
@api_admin_bp.route('/admin/stop-impersonating', methods=['POST'])
@jwt_required
def stop_impersonating():
"""Finalizar sesión de impersonación y restaurar cuenta de administrador."""
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
current_target = g.jwt_user
audit = AuditLog(
user_id=admin_user.id,
user_email=admin_user.email,
action='IMPERSONATE_END',
module='auth',
entity_id=current_target.id if current_target else None,
details=f"Administrador {admin_user.email} finalizó la sesión de impersonación."
)
db.session.add(audit)
db.session.commit()
return jsonify({
'status': 'success',
'message': 'Sesión de impersonación finalizada.'
}), 200
@api_admin_bp.route('/reservations/drag-update', methods=['POST'])
@jwt_required
def drag_update_reservation():
"""Actualización o creación rápida de reserva mediante arrastre en la grilla semanal."""
data = request.get_json(silent=True) or {}
reservation_id = data.get('reservation_id')
commission_id = data.get('commission_id')
classroom_id = data.get('classroom_id')
start_time_str = data.get('start_time')
end_time_str = data.get('end_time')
if not classroom_id or not start_time_str or not end_time_str:
return jsonify({'error': 'ValidationError', 'message': 'Faltan parámetros obligatorios (aula, inicio, fin).'}), 400
try:
if isinstance(start_time_str, str):
start_dt = datetime.fromisoformat(start_time_str.replace('Z', '+00:00'))
else:
start_dt = start_time_str
if isinstance(end_time_str, str):
end_dt = datetime.fromisoformat(end_time_str.replace('Z', '+00:00'))
else:
end_dt = end_time_str
except Exception as e:
return jsonify({'error': 'ValidationError', 'message': f'Formato de fecha inválido: {str(e)}'}), 400
classroom = db.session.get(Classroom, classroom_id)
if not classroom:
return jsonify({'error': 'NotFound', 'message': 'Aula no encontrada.'}), 404
# Pre-validación de conflictos (físico, capacidad, docente)
from app.services.reservation_service import ReservationService
effective_comm_id = commission_id
if not effective_comm_id and reservation_id:
existing_res = db.session.get(Reservation, reservation_id)
if existing_res:
effective_comm_id = existing_res.commission_id
res_service = ReservationService()
matrix = res_service.check_full_conflicts_matrix(
classroom_id=classroom_id,
start_time=start_dt,
end_time=end_dt,
commission_id=effective_comm_id,
exclude_id=reservation_id
)
if matrix.get('is_blocked'):
return jsonify({
'status': 'conflict',
'error': 'ConflictDetected',
'message': ' '.join(matrix.get('block_reasons', ['Conflicto detectado en la asignación.'])),
'conflicts': matrix.get('block_reasons', [])
}), 409
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
acting_user = g.jwt_user
if reservation_id:
res = db.session.get(Reservation, reservation_id)
if not res:
return jsonify({'error': 'NotFound', 'message': 'Reserva no encontrada.'}), 404
res.classroom_id = classroom_id
res.start_time = start_dt
res.end_time = end_dt
action = 'UPDATE_RESERVATION_DRAG'
msg = f"Reserva #{res.id} reprogramada al aula {classroom.code} ({start_dt.strftime('%H:%M')} a {end_dt.strftime('%H:%M')})"
else:
comm = db.session.get(Commission, commission_id) if commission_id else None
res = Reservation(
classroom_id=classroom_id,
commission_id=commission_id,
user_id=acting_user.id,
start_time=start_dt,
end_time=end_dt,
purpose=f"Cursada regular - {comm.subject.name if comm and comm.subject else 'Comisión'}",
expected_attendees=comm.max_students if comm else 30,
status=ReservationStatus.CONFIRMED.value
)
db.session.add(res)
action = 'CREATE_RESERVATION_DRAG'
msg = f"Comisión {comm.code if comm else ''} asignada al aula {classroom.code}"
details_str = msg
if getattr(g, 'is_impersonating', False):
details_str += f" [Ejecutado bajo impersonación por Admin #{admin_user.id}]"
audit = AuditLog(
user_id=admin_user.id,
user_email=admin_user.email,
action=action,
module='reservations',
entity_id=res.id,
details=details_str
)
db.session.add(audit)
db.session.commit()
return jsonify({
'status': 'success',
'message': msg,
'reservation': res.to_dict()
}), 200
+43
View File
@@ -0,0 +1,43 @@
from flask import Blueprint, jsonify, request, g
from app.utils.jwt_decorators import jwt_required
from app.services.gradebook_service import GradebookService
api_students_bp = Blueprint('api_students', __name__)
@api_students_bp.route('/students/my-grades', methods=['GET'])
@jwt_required
def get_my_grades():
"""Consulta individual de calificaciones y estado académico del alumno logueado."""
user = getattr(g, 'jwt_user', None)
user_id = user.id if user else getattr(request, 'current_user_id', None)
if not user_id:
return jsonify({'error': 'Unauthorized', 'message': 'Usuario no autenticado.'}), 401
try:
grades = GradebookService.get_student_grades(user_id)
return jsonify({
'status': 'success',
'student_id': user_id,
'is_impersonating': getattr(g, 'is_impersonating', False),
'data': grades
}), 200
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
@api_students_bp.route('/students/my-upcoming-milestones', methods=['GET'])
@jwt_required
def get_my_upcoming_milestones():
"""Próximas evaluaciones e hitos académicos con cuenta regresiva para el dashboard."""
user = getattr(g, 'jwt_user', None)
user_id = user.id if user else getattr(request, 'current_user_id', None)
if not user_id:
return jsonify({'error': 'Unauthorized', 'message': 'Usuario no autenticado.'}), 401
limit = request.args.get('limit', default=5, type=int)
try:
milestones = GradebookService.get_student_upcoming_milestones(user_id, limit=limit)
return jsonify({'status': 'success', 'data': milestones}), 200
except Exception as e:
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
+436
View File
@@ -0,0 +1,436 @@
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any
from app import db
from app.models.subject import Commission, Subject
from app.models.milestone import AcademicMilestone, MilestoneType
from app.models.enrollment import StudentEnrollment
from app.models.grade import MilestoneGrade
from app.models.user import User
from app.models.audit_log import AuditLog
class GradebookService:
"""
Servicio de Libro de Calificaciones, Hitos Evaluativos y Cierre de Actas (Fase 3 MVP).
"""
@classmethod
def get_commission_gradebook(cls, commission_id: int) -> Dict[str, Any]:
"""Obtiene la matriz completa de calificaciones de una comisión."""
comm = Commission.query.get(commission_id)
if not comm:
raise ValueError(f'Comisión con ID {commission_id} no encontrada.')
# 1. Hitos evaluativos de la comisión
milestones = AcademicMilestone.query.filter_by(
commission_id=comm.id,
is_active=True
).order_by(AcademicMilestone.date.asc()).all()
# 2. Estudiantes inscriptos
enrollments = StudentEnrollment.query.filter_by(
commission_id=comm.id
).join(User).order_by(User.name.asc()).all()
# 3. Mapa de notas por (student_id, milestone_id)
grades = MilestoneGrade.query.filter(
MilestoneGrade.milestone_id.in_([m.id for m in milestones]) if milestones else False
).all() if milestones else []
grades_map: Dict[tuple, MilestoneGrade] = {
(g.student_id, g.milestone_id): g for g in grades
}
# 4. Construcción de filas por estudiante
rows = []
for enr in enrollments:
student = enr.student
student_grades = []
valid_scores = []
has_failed = False
total_milestones = len(milestones)
graded_count = 0
for m in milestones:
g = grades_map.get((student.id, m.id))
if g:
graded_count += 1
g_dict = g.to_dict()
student_grades.append(g_dict)
if g.is_absent:
has_failed = True
elif g.score is not None:
valid_scores.append(g.score)
if g.score < 4.0:
has_failed = True
else:
student_grades.append({
'id': None,
'milestone_id': m.id,
'student_id': student.id,
'score': None,
'display_score': '-',
'is_absent': False,
'is_passed': False,
'feedback': ''
})
# Cálculo de promedio y condición
avg_score = round(sum(valid_scores) / len(valid_scores), 2) if valid_scores else None
if graded_count == 0:
condition = 'Sin Calificar'
condition_badge = 'bg-secondary'
elif has_failed:
condition = 'Desaprobado'
condition_badge = 'bg-danger'
elif avg_score is not None and avg_score >= 7.0 and graded_count == total_milestones:
condition = 'Promociona'
condition_badge = 'bg-success'
elif avg_score is not None and avg_score >= 4.0:
condition = 'Regular'
condition_badge = 'bg-primary'
else:
condition = 'Pendiente'
condition_badge = 'bg-warning text-dark'
rows.append({
'enrollment_id': enr.id,
'student_id': student.id,
'student_name': student.name,
'student_email': student.email,
'enrollment_status': enr.status,
'grades': student_grades,
'average_score': avg_score,
'graded_count': graded_count,
'total_milestones': total_milestones,
'condition': condition,
'condition_badge': condition_badge
})
return {
'commission': comm.to_dict(),
'milestones': [m.to_dict() for m in milestones],
'students': rows,
'is_closed': bool(comm.grades_closed),
'acta_number': comm.acta_number,
'closed_at': comm.grades_closed_at.isoformat() if comm.grades_closed_at else None,
'closed_by_name': comm.closed_by_user.name if comm.closed_by_user else None,
'can_modify': not bool(comm.grades_closed)
}
@classmethod
def save_single_grade(
cls,
commission_id: int,
milestone_id: int,
student_id: int,
score: Optional[float],
is_absent: bool = False,
feedback: str = '',
grader_id: Optional[int] = None
) -> MilestoneGrade:
"""Guarda o actualiza una calificación individual."""
comm = Commission.query.get(commission_id)
if not comm:
raise ValueError(f'Comisión con ID {commission_id} no encontrada.')
if comm.grades_closed:
raise ValueError('El acta de calificaciones de esta comisión se encuentra cerrada y no admite modificaciones.')
milestone = AcademicMilestone.query.filter_by(id=milestone_id, commission_id=comm.id).first()
if not milestone:
raise ValueError(f'Hito evaluativo {milestone_id} no pertenece a la comisión {commission_id}.')
enrollment = StudentEnrollment.query.filter_by(commission_id=comm.id, student_id=student_id).first()
if not enrollment:
raise ValueError(f'El estudiante {student_id} no está matriculado en la comisión {commission_id}.')
# Validación de nota numérica
if is_absent:
final_score = None
final_absent = True
else:
final_absent = False
if score is not None:
try:
f_score = float(score)
except (ValueError, TypeError):
raise ValueError('La calificación debe ser un valor numérico válido entre 1.0 y 10.0.')
if f_score < 1.0 or f_score > 10.0:
raise ValueError('La calificación debe estar comprendida entre 1.0 y 10.0.')
final_score = round(f_score, 2)
else:
final_score = None
grade = MilestoneGrade.query.filter_by(
milestone_id=milestone.id,
student_id=student_id
).first()
if not grade:
grade = MilestoneGrade(
milestone_id=milestone.id,
enrollment_id=enrollment.id,
student_id=student_id,
score=final_score,
is_absent=final_absent,
feedback=feedback.strip() if feedback else '',
graded_by_user_id=grader_id,
graded_at=datetime.utcnow()
)
db.session.add(grade)
else:
grade.score = final_score
grade.is_absent = final_absent
if feedback is not None:
grade.feedback = feedback.strip()
grade.graded_by_user_id = grader_id
grade.updated_at = datetime.utcnow()
db.session.commit()
return grade
@classmethod
def bulk_save_grades(
cls,
commission_id: int,
grades_data: List[Dict[str, Any]],
grader_id: Optional[int] = None
) -> Dict[str, Any]:
"""Carga masiva / autoguardado asíncrono de calificaciones."""
comm = Commission.query.get(commission_id)
if not comm:
raise ValueError(f'Comisión {commission_id} no encontrada.')
if comm.grades_closed:
raise ValueError('El acta de calificaciones se encuentra cerrada y no permite modificaciones.')
saved_count = 0
for item in grades_data:
m_id = item.get('milestone_id')
s_id = item.get('student_id')
score = item.get('score')
is_absent = bool(item.get('is_absent', False))
feedback = item.get('feedback', '')
if m_id and s_id:
cls.save_single_grade(
commission_id=comm.id,
milestone_id=int(m_id),
student_id=int(s_id),
score=float(score) if score is not None and str(score).strip() != '' else None,
is_absent=is_absent,
feedback=feedback,
grader_id=grader_id
)
saved_count += 1
return {'success': True, 'saved_count': saved_count}
@classmethod
def close_gradebook(cls, commission_id: int, user_id: int, notes: str = '') -> Dict[str, Any]:
"""Cierre formal del acta de regularidad y promoción para Bedelía/Titular."""
comm = Commission.query.get(commission_id)
if not comm:
raise ValueError(f'Comisión {commission_id} no encontrada.')
if comm.grades_closed:
raise ValueError(f'El acta de la comisión {commission_id} ya fue cerrada anteriormente.')
acta_num = f'ACTA-{comm.year}-{comm.semester}-{comm.id:04d}'
comm.grades_closed = True
comm.grades_closed_at = datetime.utcnow()
comm.grades_closed_by = user_id
comm.acta_number = acta_num
# Auditoría
audit = AuditLog(
user_id=user_id,
action='CLOSE_ACTA',
module='gradebook',
entity_id=comm.id,
details=f'Cierre formal de acta de calificaciones. Nro: {acta_num}. Notas: {notes}'
)
db.session.add(audit)
db.session.commit()
return {
'success': True,
'acta_number': acta_num,
'closed_at': comm.grades_closed_at.isoformat(),
'message': f'Acta {acta_num} cerrada exitosamente. Las calificaciones han quedado selladas.'
}
@classmethod
def reopen_gradebook(cls, commission_id: int, user_id: int, reason: str) -> Dict[str, Any]:
"""Reapertura excepcional de acta con justificación administrativa."""
if not reason or len(reason.strip()) < 5:
raise ValueError('Debe especificar un motivo formal para reabrir el acta de calificaciones.')
comm = Commission.query.get(commission_id)
if not comm:
raise ValueError(f'Comisión {commission_id} no encontrada.')
if not comm.grades_closed:
raise ValueError('El acta no se encuentra cerrada.')
old_acta = comm.acta_number
comm.grades_closed = False
comm.grades_closed_at = None
comm.grades_closed_by = None
audit = AuditLog(
user_id=user_id,
action='REOPEN_ACTA',
module='gradebook',
entity_id=comm.id,
details=f'Reapertura excepcional de acta {old_acta}. Motivo: {reason.strip()}'
)
db.session.add(audit)
db.session.commit()
return {
'success': True,
'message': f'Acta {old_acta} reabierta para rectificación de calificaciones.'
}
@classmethod
def get_student_grades(cls, student_id: int) -> List[Dict[str, Any]]:
"""Consulta individual de calificaciones de todas las materias del alumno."""
enrollments = StudentEnrollment.query.filter_by(
student_id=student_id
).all()
results = []
for enr in enrollments:
comm = enr.commission
if not comm:
continue
milestones = AcademicMilestone.query.filter_by(
commission_id=comm.id,
is_active=True
).order_by(AcademicMilestone.date.asc()).all()
grades = MilestoneGrade.query.filter_by(
enrollment_id=enr.id
).all()
grades_by_milestone = {g.milestone_id: g for g in grades}
ms_results = []
valid_scores = []
has_failed = False
for m in milestones:
g = grades_by_milestone.get(m.id)
if g:
ms_results.append(g.to_dict())
if g.is_absent:
has_failed = True
elif g.score is not None:
valid_scores.append(g.score)
if g.score < 4.0:
has_failed = True
else:
ms_results.append({
'milestone_id': m.id,
'milestone_title': m.title or m.tipo,
'milestone_type': m.tipo,
'milestone_color': m.color,
'display_score': '-',
'score': None,
'is_absent': False,
'feedback': ''
})
avg_score = round(sum(valid_scores) / len(valid_scores), 2) if valid_scores else None
if not grades:
condition = 'Cursando'
condition_badge = 'bg-secondary'
elif has_failed:
condition = 'Desaprobado'
condition_badge = 'bg-danger'
elif avg_score is not None and avg_score >= 7.0 and len(grades) == len(milestones):
condition = 'Promociona'
condition_badge = 'bg-success'
elif avg_score is not None and avg_score >= 4.0:
condition = 'Regular'
condition_badge = 'bg-primary'
else:
condition = 'En curso'
condition_badge = 'bg-info text-dark'
results.append({
'enrollment_id': enr.id,
'commission_id': comm.id,
'commission_code': comm.get_full_code(),
'subject_name': comm.subject.name if comm.subject else 'Materia',
'subject_code': comm.subject.code if comm.subject else '',
'semester': comm.semester,
'year': comm.year,
'teacher_name': comm.teacher_name,
'milestones': ms_results,
'average_score': avg_score,
'condition': condition,
'condition_badge': condition_badge,
'is_acta_closed': bool(comm.grades_closed),
'acta_number': comm.acta_number
})
return results
@classmethod
def get_student_upcoming_milestones(cls, student_id: int, limit: int = 5) -> List[Dict[str, Any]]:
"""Obtiene las próximas evaluaciones con cálculo de cuenta regresiva para el dashboard."""
now = datetime.utcnow()
enrollments = StudentEnrollment.query.filter_by(student_id=student_id).all()
commission_ids = [e.commission_id for e in enrollments]
if not commission_ids:
return []
upcoming = AcademicMilestone.query.filter(
AcademicMilestone.commission_id.in_(commission_ids),
AcademicMilestone.date >= now - timedelta(hours=2),
AcademicMilestone.is_active == True
).order_by(AcademicMilestone.date.asc()).limit(limit).all()
results = []
for m in upcoming:
delta = m.date - now
days = delta.days
hours = int(delta.seconds // 3600)
mins = int((delta.seconds % 3600) // 60)
if days > 1:
countdown_str = f'Faltan {days} días'
badge_urgency = 'bg-info'
elif days == 1:
countdown_str = 'Mañana'
badge_urgency = 'bg-warning text-dark'
elif hours > 0:
countdown_str = f'Hoy en {hours} h {mins} min'
badge_urgency = 'bg-danger'
else:
countdown_str = 'En curso'
badge_urgency = 'bg-danger'
results.append({
'id': m.id,
'title': m.title or m.tipo,
'tipo': m.tipo,
'color': m.color,
'date': m.date.isoformat(),
'date_formatted': m.date.strftime('%d/%m/%Y %H:%M hs'),
'classroom': m.classroom or 'Campus Virtual',
'link': m.link,
'subject_name': m.subject.name if m.subject else m.materia_codigo,
'commission_code': m.commission.get_full_code() if m.commission else '',
'days_remaining': max(0, days),
'hours_remaining': max(0, hours),
'countdown_display': countdown_str,
'badge_urgency': badge_urgency
})
return results
+14
View File
@@ -40,6 +40,20 @@ def jwt_required(f):
g.jwt_user = user
g.jwt_payload = payload
g.jwt_token = token
g.real_admin_user = user
g.is_impersonating = False
impersonate_header = request.headers.get('X-Impersonate-User')
if impersonate_header and user.is_admin():
target_user = None
if impersonate_header.strip().isdigit():
target_user = db.session.get(User, int(impersonate_header.strip()))
else:
target_user = User.query.filter_by(email=impersonate_header.strip()).first()
if target_user and target_user.is_active:
g.jwt_user = target_user
g.is_impersonating = True
except jwt.ExpiredSignatureError:
return jsonify({