437 lines
17 KiB
Python
437 lines
17 KiB
Python
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
|