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