389 lines
15 KiB
Python
389 lines
15 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy.orm import joinedload
|
|
from sqlalchemy import func
|
|
from models import Grade, GradingElement, Exercise, Assessment, Student
|
|
from .base_repository import BaseRepository
|
|
from app_config import config_manager
|
|
|
|
|
|
class GradeRepository(BaseRepository[Grade]):
|
|
"""Repository pour les notes."""
|
|
|
|
def __init__(self):
|
|
super().__init__(Grade)
|
|
|
|
def find_by_student_and_element(self, student_id: int, grading_element_id: int) -> Optional[Grade]:
|
|
"""Trouve une note par étudiant et élément de notation."""
|
|
return Grade.query.filter_by(
|
|
student_id=student_id,
|
|
grading_element_id=grading_element_id
|
|
).first()
|
|
|
|
def find_or_create_by_student_and_element(self, student_id: int, grading_element_id: int) -> Grade:
|
|
"""Trouve ou crée une note par étudiant et élément de notation."""
|
|
grade = self.find_by_student_and_element(student_id, grading_element_id)
|
|
if not grade:
|
|
grade = Grade(
|
|
student_id=student_id,
|
|
grading_element_id=grading_element_id
|
|
)
|
|
self.save(grade)
|
|
self.flush() # Pour obtenir l'ID
|
|
return grade
|
|
|
|
def find_by_assessment(self, assessment_id: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'une évaluation."""
|
|
return Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).filter_by(
|
|
assessment_id=assessment_id
|
|
).all()
|
|
|
|
def find_by_student(self, student_id: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'un étudiant."""
|
|
return Grade.query.filter_by(
|
|
student_id=student_id
|
|
).all()
|
|
|
|
def delete_by_student(self, student_id: int) -> int:
|
|
"""Supprime toutes les notes d'un étudiant. Retourne le nombre supprimé."""
|
|
count = Grade.query.filter_by(student_id=student_id).count()
|
|
Grade.query.filter_by(student_id=student_id).delete()
|
|
return count
|
|
|
|
def find_existing_grades_for_assessment(self, assessment_id: int) -> Dict[str, Grade]:
|
|
"""
|
|
Trouve les notes existantes d'une évaluation indexées par clé.
|
|
Clé format: "{student_id}_{grading_element_id}"
|
|
"""
|
|
existing_grades = {}
|
|
for grade in self.find_by_assessment(assessment_id):
|
|
key = f"{grade.student_id}_{grade.grading_element_id}"
|
|
existing_grades[key] = grade
|
|
return existing_grades
|
|
|
|
def bulk_update_or_create_grades(self, grade_data: List[Dict[str, Any]]) -> int:
|
|
"""
|
|
Met à jour ou crée plusieurs notes en lot.
|
|
|
|
Args:
|
|
grade_data: Liste de dictionnaires avec student_id, grading_element_id, value, comment
|
|
|
|
Returns:
|
|
Nombre de notes traitées
|
|
"""
|
|
count = 0
|
|
for data in grade_data:
|
|
grade = self.find_by_student_and_element(
|
|
data['student_id'],
|
|
data['grading_element_id']
|
|
)
|
|
|
|
if data.get('value') or data.get('comment'):
|
|
if not grade:
|
|
grade = Grade(
|
|
student_id=data['student_id'],
|
|
grading_element_id=data['grading_element_id'],
|
|
value=data.get('value'),
|
|
comment=data.get('comment')
|
|
)
|
|
self.save(grade)
|
|
else:
|
|
grade.value = data.get('value')
|
|
grade.comment = data.get('comment')
|
|
count += 1
|
|
elif grade:
|
|
# Supprimer si valeur et commentaire vides
|
|
self.delete(grade)
|
|
count += 1
|
|
|
|
return count
|
|
|
|
def find_by_student_trimester_with_elements(self, student_id: int, trimester: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'un élève pour un trimestre avec les éléments de notation."""
|
|
return Grade.query.join(GradingElement).join(Exercise).join(Assessment).filter(
|
|
Grade.student_id == student_id,
|
|
Assessment.trimester == trimester
|
|
).options(
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise).joinedload(Exercise.assessment),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.domain)
|
|
).all()
|
|
|
|
def find_by_class_trimester(self, class_group_id: int, trimester: int) -> List[Grade]:
|
|
"""Trouve toutes les notes d'une classe pour un trimestre."""
|
|
return Grade.query.join(Student).join(GradingElement).join(Exercise).join(Assessment).filter(
|
|
Student.class_group_id == class_group_id,
|
|
Assessment.trimester == trimester
|
|
).options(
|
|
joinedload(Grade.student),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise).joinedload(Exercise.assessment),
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.domain)
|
|
).all()
|
|
|
|
def get_student_grades_by_assessment(self, student_id: int, assessment_id: int) -> List[Grade]:
|
|
"""Récupère toutes les notes d'un élève pour une évaluation spécifique."""
|
|
return Grade.query.join(GradingElement).join(Exercise).filter(
|
|
Grade.student_id == student_id,
|
|
Exercise.assessment_id == assessment_id
|
|
).options(
|
|
joinedload(Grade.grading_element).joinedload(GradingElement.exercise)
|
|
).all()
|
|
|
|
def get_special_values_counts_by_student_trimester(self, student_id: int, trimester: int) -> Dict[str, int]:
|
|
"""
|
|
Compte les valeurs spéciales pour un élève dans un trimestre donné.
|
|
|
|
Args:
|
|
student_id: ID de l'élève
|
|
trimester: Numéro du trimestre (1, 2, ou 3)
|
|
|
|
Returns:
|
|
Dictionnaire avec les comptes par valeur spéciale (ex: {'.': 3, 'd': 1, 'a': 0})
|
|
"""
|
|
# Récupération dynamique des valeurs spéciales configurées
|
|
special_values = config_manager.get_special_values()
|
|
|
|
# Requête pour compter les valeurs spéciales avec jointures optimisées
|
|
grade_values = Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).join(
|
|
Assessment
|
|
).filter(
|
|
Grade.student_id == student_id,
|
|
Assessment.trimester == trimester,
|
|
Grade.value.in_(list(special_values.keys()))
|
|
).with_entities(
|
|
Grade.value,
|
|
func.count(Grade.value).label('count')
|
|
).group_by(
|
|
Grade.value
|
|
).all()
|
|
|
|
# Initialiser le dictionnaire avec toutes les valeurs spéciales à 0
|
|
result = {value: 0 for value in special_values.keys()}
|
|
|
|
# Mettre à jour avec les comptes réels
|
|
for value, count in grade_values:
|
|
if value in result:
|
|
result[value] = count
|
|
|
|
return result
|
|
|
|
def get_special_values_counts_by_student_assessment(self, student_id: int, assessment_id: int) -> Dict[str, int]:
|
|
"""
|
|
Compte les valeurs spéciales pour un élève dans une évaluation donnée.
|
|
|
|
Args:
|
|
student_id: ID de l'élève
|
|
assessment_id: ID de l'évaluation
|
|
|
|
Returns:
|
|
Dictionnaire avec les comptes par valeur spéciale (ex: {'.': 2, 'd': 0, 'a': 1})
|
|
"""
|
|
# Récupération dynamique des valeurs spéciales configurées
|
|
special_values = config_manager.get_special_values()
|
|
|
|
# Requête pour compter les valeurs spéciales avec jointures optimisées
|
|
grade_values = Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).filter(
|
|
Grade.student_id == student_id,
|
|
Exercise.assessment_id == assessment_id,
|
|
Grade.value.in_(list(special_values.keys()))
|
|
).with_entities(
|
|
Grade.value,
|
|
func.count(Grade.value).label('count')
|
|
).group_by(
|
|
Grade.value
|
|
).all()
|
|
|
|
# Initialiser le dictionnaire avec toutes les valeurs spéciales à 0
|
|
result = {value: 0 for value in special_values.keys()}
|
|
|
|
# Mettre à jour avec les comptes réels
|
|
for value, count in grade_values:
|
|
if value in result:
|
|
result[value] = count
|
|
|
|
return result
|
|
|
|
def get_special_values_details_by_student_trimester(self, student_id: int, trimester: int) -> Dict[str, List[Dict[str, Any]]]:
|
|
"""
|
|
Récupère les détails des valeurs spéciales pour un élève dans un trimestre donné.
|
|
|
|
Args:
|
|
student_id: ID de l'élève
|
|
trimester: Numéro du trimestre (1, 2, ou 3)
|
|
|
|
Returns:
|
|
Dictionnaire avec les détails par valeur spéciale incluant les commentaires
|
|
Format: {'.': [{'element_name': str, 'comment': str, 'assessment_title': str}, ...]}
|
|
"""
|
|
# Récupération dynamique des valeurs spéciales configurées
|
|
special_values = config_manager.get_special_values()
|
|
|
|
# Requête pour récupérer les détails des valeurs spéciales
|
|
grade_details = Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).join(
|
|
Assessment
|
|
).filter(
|
|
Grade.student_id == student_id,
|
|
Assessment.trimester == trimester,
|
|
Grade.value.in_(list(special_values.keys()))
|
|
).with_entities(
|
|
Grade.value,
|
|
Grade.comment,
|
|
GradingElement.label.label('element_name'),
|
|
Assessment.title.label('assessment_title'),
|
|
Assessment.id.label('assessment_id')
|
|
).all()
|
|
|
|
# Organiser par valeur spéciale
|
|
result_details = {}
|
|
|
|
# Initialiser avec toutes les valeurs spéciales
|
|
for special_value in special_values.keys():
|
|
result_details[special_value] = []
|
|
|
|
# Ajouter les détails trouvés
|
|
for detail in grade_details:
|
|
result_details[detail.value].append({
|
|
'element_name': detail.element_name,
|
|
'comment': detail.comment,
|
|
'assessment_title': detail.assessment_title,
|
|
'assessment_id': detail.assessment_id
|
|
})
|
|
|
|
return result_details
|
|
|
|
def get_special_values_details_by_student_assessment(self, student_id: int, assessment_id: int) -> Dict[str, List[Dict[str, Any]]]:
|
|
"""
|
|
Récupère les détails des valeurs spéciales pour un élève dans une évaluation donnée.
|
|
|
|
Args:
|
|
student_id: ID de l'élève
|
|
assessment_id: ID de l'évaluation
|
|
|
|
Returns:
|
|
Dictionnaire avec les détails par valeur spéciale incluant les commentaires
|
|
Format: {'.': [{'element_name': str, 'comment': str}, ...]}
|
|
"""
|
|
# Récupération dynamique des valeurs spéciales configurées
|
|
special_values = config_manager.get_special_values()
|
|
|
|
# Requête pour récupérer les détails des valeurs spéciales
|
|
grade_details = Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).filter(
|
|
Grade.student_id == student_id,
|
|
Exercise.assessment_id == assessment_id,
|
|
Grade.value.in_(list(special_values.keys()))
|
|
).with_entities(
|
|
Grade.value,
|
|
Grade.comment,
|
|
GradingElement.label.label('element_name')
|
|
).all()
|
|
|
|
# Organiser par valeur spéciale
|
|
result_details = {}
|
|
|
|
# Initialiser avec toutes les valeurs spéciales
|
|
for special_value in special_values.keys():
|
|
result_details[special_value] = []
|
|
|
|
# Ajouter les détails trouvés
|
|
for detail in grade_details:
|
|
result_details[detail.value].append({
|
|
'element_name': detail.element_name,
|
|
'comment': detail.comment
|
|
})
|
|
|
|
return result_details
|
|
|
|
def get_all_comments_by_student_trimester(self, student_id: int, trimester: int) -> Dict[str, Any]:
|
|
"""
|
|
Récupère tous les commentaires regroupés par évaluations pour un élève dans un trimestre.
|
|
|
|
Args:
|
|
student_id: ID de l'élève
|
|
trimester: Numéro du trimestre (1, 2, ou 3)
|
|
|
|
Returns:
|
|
Dict avec commentaires organisés par évaluation avec métadonnées complètes
|
|
"""
|
|
# Requête pour récupérer tous les commentaires non vides avec toutes les informations
|
|
grade_comments = Grade.query.join(
|
|
GradingElement
|
|
).join(
|
|
Exercise
|
|
).join(
|
|
Assessment
|
|
).filter(
|
|
Grade.student_id == student_id,
|
|
Assessment.trimester == trimester,
|
|
Grade.comment.isnot(None),
|
|
Grade.comment != ''
|
|
).with_entities(
|
|
Grade.value,
|
|
Grade.comment,
|
|
GradingElement.label.label('element_label'),
|
|
GradingElement.description.label('element_description'),
|
|
Assessment.title.label('assessment_title'),
|
|
Assessment.id.label('assessment_id'),
|
|
Assessment.date.label('assessment_date'),
|
|
Exercise.title.label('exercise_title'),
|
|
Exercise.order.label('exercise_order')
|
|
).order_by(
|
|
Assessment.date.desc(),
|
|
Exercise.order,
|
|
GradingElement.id
|
|
).all()
|
|
|
|
# Organiser par évaluation
|
|
comments_by_assessment = {}
|
|
|
|
for comment_data in grade_comments:
|
|
assessment_id = comment_data.assessment_id
|
|
|
|
# Initialiser l'évaluation si pas encore présente
|
|
if assessment_id not in comments_by_assessment:
|
|
comments_by_assessment[assessment_id] = {
|
|
'id': assessment_id,
|
|
'title': comment_data.assessment_title,
|
|
'date': comment_data.assessment_date,
|
|
'comments': []
|
|
}
|
|
|
|
# Ajouter le commentaire
|
|
comments_by_assessment[assessment_id]['comments'].append({
|
|
'value': comment_data.value,
|
|
'comment': comment_data.comment,
|
|
'element_label': comment_data.element_label,
|
|
'element_description': comment_data.element_description,
|
|
'exercise_title': comment_data.exercise_title,
|
|
'exercise_order': comment_data.exercise_order
|
|
})
|
|
|
|
# Convertir en liste triée par date (plus récent en premier)
|
|
assessments_with_comments = list(comments_by_assessment.values())
|
|
assessments_with_comments.sort(key=lambda x: x['date'] if x['date'] else '', reverse=True)
|
|
|
|
# Calculer le total de commentaires
|
|
total_comments = sum(len(assessment['comments']) for assessment in assessments_with_comments)
|
|
|
|
return {
|
|
'assessments': assessments_with_comments,
|
|
'total_comments': total_comments,
|
|
'has_comments': total_comments > 0
|
|
} |