900 lines
36 KiB
Python
900 lines
36 KiB
Python
"""
|
|
Services découplés pour les opérations métier sur les évaluations.
|
|
|
|
Ce module applique les principes SOLID en séparant les responsabilités
|
|
de calcul, statistiques et progression qui étaient auparavant dans le modèle Assessment.
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, Any, List, Optional, Tuple, Protocol
|
|
from dataclasses import dataclass
|
|
from collections import defaultdict
|
|
import statistics
|
|
import math
|
|
|
|
# Type hints pour améliorer la lisibilité
|
|
StudentId = int
|
|
ExerciseId = int
|
|
GradingElementId = int
|
|
|
|
|
|
# =================== INTERFACES (Dependency Inversion Principle) ===================
|
|
|
|
class ConfigProvider(Protocol):
|
|
"""Interface pour l'accès à la configuration."""
|
|
|
|
def is_special_value(self, value: str) -> bool:
|
|
"""Vérifie si une valeur est spéciale (., d, etc.)"""
|
|
...
|
|
|
|
def get_special_values(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Retourne la configuration des valeurs spéciales."""
|
|
...
|
|
|
|
|
|
class DatabaseProvider(Protocol):
|
|
"""Interface pour l'accès aux données."""
|
|
|
|
def get_grades_for_assessment(self, assessment_id: int) -> List[Any]:
|
|
"""Récupère toutes les notes d'une évaluation en une seule requête."""
|
|
...
|
|
|
|
def get_grading_elements_with_students(self, assessment_id: int) -> List[Any]:
|
|
"""Récupère les éléments de notation avec les étudiants associés."""
|
|
...
|
|
|
|
|
|
# =================== DATA TRANSFER OBJECTS ===================
|
|
|
|
@dataclass
|
|
class ProgressResult:
|
|
"""Résultat du calcul de progression."""
|
|
percentage: int
|
|
completed: int
|
|
total: int
|
|
status: str
|
|
students_count: int
|
|
|
|
|
|
@dataclass
|
|
class StudentScore:
|
|
"""Score d'un étudiant pour une évaluation."""
|
|
student_id: int
|
|
student_name: str
|
|
total_score: float
|
|
total_max_points: float
|
|
exercises: Dict[ExerciseId, Dict[str, Any]]
|
|
|
|
|
|
@dataclass
|
|
class StatisticsResult:
|
|
"""Résultat des calculs statistiques."""
|
|
count: int
|
|
mean: float
|
|
median: float
|
|
min: float
|
|
max: float
|
|
std_dev: float
|
|
|
|
|
|
# =================== STRATEGY PATTERN pour les types de notation ===================
|
|
|
|
class GradingStrategy(ABC):
|
|
"""Interface Strategy pour les différents types de notation."""
|
|
|
|
@abstractmethod
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
"""Calcule le score selon le type de notation."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_grading_type(self) -> str:
|
|
"""Retourne le type de notation."""
|
|
pass
|
|
|
|
|
|
class NotesStrategy(GradingStrategy):
|
|
"""Strategy pour la notation en points (notes)."""
|
|
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
try:
|
|
return float(grade_value)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
def get_grading_type(self) -> str:
|
|
return 'notes'
|
|
|
|
|
|
class ScoreStrategy(GradingStrategy):
|
|
"""Strategy pour la notation par compétences (score 0-3)."""
|
|
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
try:
|
|
score_int = int(grade_value)
|
|
if 0 <= score_int <= 3:
|
|
return (score_int / 3) * max_points
|
|
return 0.0
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
def get_grading_type(self) -> str:
|
|
return 'score'
|
|
|
|
|
|
class GradingStrategyFactory:
|
|
"""Factory pour créer les strategies de notation."""
|
|
|
|
_strategies = {
|
|
'notes': NotesStrategy,
|
|
'score': ScoreStrategy
|
|
}
|
|
|
|
@classmethod
|
|
def create(cls, grading_type: str) -> GradingStrategy:
|
|
"""Crée une strategy selon le type."""
|
|
strategy_class = cls._strategies.get(grading_type)
|
|
if not strategy_class:
|
|
raise ValueError(f"Type de notation non supporté: {grading_type}")
|
|
return strategy_class()
|
|
|
|
@classmethod
|
|
def register_strategy(cls, grading_type: str, strategy_class: type):
|
|
"""Permet d'enregistrer de nouveaux types de notation."""
|
|
cls._strategies[grading_type] = strategy_class
|
|
|
|
|
|
# =================== SERVICES MÉTIER ===================
|
|
|
|
class UnifiedGradingCalculator:
|
|
"""
|
|
Calculateur unifié utilisant le pattern Strategy et l'injection de dépendances.
|
|
Remplace la classe GradingCalculator du modèle.
|
|
"""
|
|
|
|
def __init__(self, config_provider: ConfigProvider):
|
|
self.config_provider = config_provider
|
|
self._strategies = {}
|
|
|
|
def calculate_score(self, grade_value: str, grading_type: str, max_points: float) -> Optional[float]:
|
|
"""
|
|
Point d'entrée unifié pour tous les calculs de score.
|
|
Utilise l'injection de dépendances pour éviter les imports circulaires.
|
|
"""
|
|
# Valeurs spéciales en premier
|
|
if self.config_provider.is_special_value(grade_value):
|
|
special_config = self.config_provider.get_special_values()[grade_value]
|
|
special_value = special_config['value']
|
|
if special_value is None: # Dispensé
|
|
return None
|
|
return float(special_value) # 0 pour '.', etc.
|
|
|
|
# Utilisation du pattern Strategy
|
|
strategy = GradingStrategyFactory.create(grading_type)
|
|
return strategy.calculate_score(grade_value, max_points)
|
|
|
|
def is_counted_in_total(self, grade_value: str) -> bool:
|
|
"""Détermine si une note doit être comptée dans le total."""
|
|
if self.config_provider.is_special_value(grade_value):
|
|
special_config = self.config_provider.get_special_values()[grade_value]
|
|
return special_config['counts']
|
|
return True
|
|
|
|
|
|
class AssessmentProgressService:
|
|
"""
|
|
Service dédié au calcul de progression des notes.
|
|
Single Responsibility: calcul et formatage de la progression.
|
|
"""
|
|
|
|
def __init__(self, db_provider: DatabaseProvider):
|
|
self.db_provider = db_provider
|
|
|
|
def calculate_grading_progress(self, assessment) -> ProgressResult:
|
|
"""
|
|
Calcule la progression de saisie des notes pour une évaluation.
|
|
Optimisé pour éviter les requêtes N+1.
|
|
"""
|
|
total_students = len(assessment.class_group.students)
|
|
|
|
if total_students == 0:
|
|
return ProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_students',
|
|
students_count=0
|
|
)
|
|
|
|
# Requête optimisée : récupération en une seule fois
|
|
grading_elements_data = self.db_provider.get_grading_elements_with_students(assessment.id)
|
|
|
|
total_elements = 0
|
|
completed_elements = 0
|
|
|
|
for element_data in grading_elements_data:
|
|
total_elements += total_students
|
|
completed_elements += element_data['completed_grades_count']
|
|
|
|
if total_elements == 0:
|
|
return ProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_elements',
|
|
students_count=total_students
|
|
)
|
|
|
|
percentage = round((completed_elements / total_elements) * 100)
|
|
|
|
# Détermination du statut
|
|
status = self._determine_status(percentage)
|
|
|
|
return ProgressResult(
|
|
percentage=percentage,
|
|
completed=completed_elements,
|
|
total=total_elements,
|
|
status=status,
|
|
students_count=total_students
|
|
)
|
|
|
|
def _determine_status(self, percentage: int) -> str:
|
|
"""Détermine le statut basé sur le pourcentage."""
|
|
if percentage == 0:
|
|
return 'not_started'
|
|
elif percentage == 100:
|
|
return 'completed'
|
|
else:
|
|
return 'in_progress'
|
|
|
|
|
|
class StudentScoreCalculator:
|
|
"""
|
|
Service dédié au calcul des scores des étudiants.
|
|
Single Responsibility: calculs de notes avec logique métier.
|
|
"""
|
|
|
|
def __init__(self,
|
|
grading_calculator: UnifiedGradingCalculator,
|
|
db_provider: DatabaseProvider):
|
|
self.grading_calculator = grading_calculator
|
|
self.db_provider = db_provider
|
|
|
|
def calculate_student_scores(self, assessment) -> Tuple[Dict[StudentId, StudentScore], Dict[ExerciseId, Dict[StudentId, float]]]:
|
|
"""
|
|
Calcule les scores de tous les étudiants pour une évaluation.
|
|
Optimisé avec requête unique pour éviter N+1.
|
|
"""
|
|
# Requête optimisée : toutes les notes en une fois
|
|
grades_data = self.db_provider.get_grades_for_assessment(assessment.id)
|
|
|
|
# Organisation des données par étudiant et exercice
|
|
students_scores = {}
|
|
exercise_scores = defaultdict(lambda: defaultdict(float))
|
|
|
|
# Calcul pour chaque étudiant
|
|
for student in assessment.class_group.students:
|
|
student_score = self._calculate_single_student_score(
|
|
student, assessment, grades_data
|
|
)
|
|
students_scores[student.id] = student_score
|
|
|
|
# Mise à jour des scores par exercice
|
|
for exercise_id, exercise_data in student_score.exercises.items():
|
|
exercise_scores[exercise_id][student.id] = exercise_data['score']
|
|
|
|
return students_scores, dict(exercise_scores)
|
|
|
|
def _calculate_single_student_score(self, student, assessment, grades_data) -> StudentScore:
|
|
"""Calcule le score d'un seul étudiant."""
|
|
total_score = 0
|
|
total_max_points = 0
|
|
student_exercises = {}
|
|
|
|
# Filtrage des notes pour cet étudiant
|
|
student_grades = {
|
|
grade['grading_element_id']: grade
|
|
for grade in grades_data
|
|
if grade['student_id'] == student.id
|
|
}
|
|
|
|
for exercise in assessment.exercises:
|
|
exercise_result = self._calculate_exercise_score(
|
|
exercise, student_grades
|
|
)
|
|
|
|
student_exercises[exercise.id] = exercise_result
|
|
total_score += exercise_result['score']
|
|
total_max_points += exercise_result['max_points']
|
|
|
|
return StudentScore(
|
|
student_id=student.id,
|
|
student_name=f"{student.first_name} {student.last_name}",
|
|
total_score=round(total_score, 2),
|
|
total_max_points=total_max_points,
|
|
exercises=student_exercises
|
|
)
|
|
|
|
def _calculate_exercise_score(self, exercise, student_grades) -> Dict[str, Any]:
|
|
"""Calcule le score pour un exercice spécifique."""
|
|
exercise_score = 0
|
|
exercise_max_points = 0
|
|
|
|
for element in exercise.grading_elements:
|
|
grade_data = student_grades.get(element.id)
|
|
|
|
if grade_data and grade_data['value'] and grade_data['value'] != '':
|
|
calculated_score = self.grading_calculator.calculate_score(
|
|
grade_data['value'].strip(),
|
|
element.grading_type,
|
|
element.max_points
|
|
)
|
|
|
|
if self.grading_calculator.is_counted_in_total(grade_data['value'].strip()):
|
|
if calculated_score is not None: # Pas dispensé
|
|
exercise_score += calculated_score
|
|
exercise_max_points += element.max_points
|
|
|
|
return {
|
|
'score': exercise_score,
|
|
'max_points': exercise_max_points,
|
|
'title': exercise.title
|
|
}
|
|
|
|
|
|
class AssessmentStatisticsService:
|
|
"""
|
|
Service dédié aux calculs statistiques.
|
|
Single Responsibility: analyses statistiques des résultats.
|
|
"""
|
|
|
|
def __init__(self, score_calculator: StudentScoreCalculator):
|
|
self.score_calculator = score_calculator
|
|
|
|
def get_assessment_statistics(self, assessment) -> StatisticsResult:
|
|
"""Calcule les statistiques descriptives pour une évaluation."""
|
|
students_scores, _ = self.score_calculator.calculate_student_scores(assessment)
|
|
scores = [score.total_score for score in students_scores.values()]
|
|
|
|
if not scores:
|
|
return StatisticsResult(
|
|
count=0,
|
|
mean=0,
|
|
median=0,
|
|
min=0,
|
|
max=0,
|
|
std_dev=0
|
|
)
|
|
|
|
return StatisticsResult(
|
|
count=len(scores),
|
|
mean=round(statistics.mean(scores), 2),
|
|
median=round(statistics.median(scores), 2),
|
|
min=min(scores),
|
|
max=max(scores),
|
|
std_dev=round(statistics.stdev(scores) if len(scores) > 1 else 0, 2)
|
|
)
|
|
|
|
|
|
# =================== FACADE pour simplifier l'utilisation ===================
|
|
|
|
class AssessmentServicesFacade:
|
|
"""
|
|
Facade qui regroupe tous les services pour faciliter l'utilisation.
|
|
Point d'entrée unique avec injection de dépendances.
|
|
"""
|
|
|
|
def __init__(self,
|
|
config_provider: ConfigProvider,
|
|
db_provider: DatabaseProvider):
|
|
# Création des services avec injection de dépendances
|
|
self.grading_calculator = UnifiedGradingCalculator(config_provider)
|
|
self.progress_service = AssessmentProgressService(db_provider)
|
|
self.score_calculator = StudentScoreCalculator(self.grading_calculator, db_provider)
|
|
self.statistics_service = AssessmentStatisticsService(self.score_calculator)
|
|
|
|
def get_grading_progress(self, assessment) -> ProgressResult:
|
|
"""Point d'entrée pour la progression."""
|
|
return self.progress_service.calculate_grading_progress(assessment)
|
|
|
|
def calculate_student_scores(self, assessment) -> Tuple[Dict[StudentId, StudentScore], Dict[ExerciseId, Dict[StudentId, float]]]:
|
|
"""Point d'entrée pour les scores étudiants."""
|
|
return self.score_calculator.calculate_student_scores(assessment)
|
|
|
|
def get_statistics(self, assessment) -> StatisticsResult:
|
|
"""Point d'entrée pour les statistiques."""
|
|
return self.statistics_service.get_assessment_statistics(assessment)
|
|
|
|
|
|
# =================== SERVICES pour ClassGroup ===================
|
|
|
|
class ClassStatisticsService:
|
|
"""
|
|
Service dédié aux statistiques de classe (get_trimester_statistics, get_class_results).
|
|
Single Responsibility: calculs statistiques au niveau classe.
|
|
"""
|
|
|
|
def __init__(self, db_provider: DatabaseProvider):
|
|
self.db_provider = db_provider
|
|
|
|
def get_trimester_statistics(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""
|
|
Retourne les statistiques globales pour un trimestre ou toutes les évaluations.
|
|
|
|
Args:
|
|
class_group: Instance de ClassGroup
|
|
trimester: Trimestre à filtrer (1, 2, 3) ou None pour toutes les évaluations
|
|
|
|
Returns:
|
|
Dict avec nombre total, répartition par statut (terminées/en cours/non commencées)
|
|
"""
|
|
try:
|
|
# Utiliser les évaluations filtrées si disponibles depuis le repository
|
|
if hasattr(class_group, '_filtered_assessments'):
|
|
assessments = class_group._filtered_assessments
|
|
else:
|
|
# Import ici pour éviter la dépendance circulaire
|
|
from models import Assessment, db
|
|
|
|
# Construire la requête de base avec jointures optimisées
|
|
query = Assessment.query.filter(Assessment.class_group_id == class_group.id)
|
|
|
|
# Filtrage par trimestre si spécifié
|
|
if trimester is not None:
|
|
query = query.filter(Assessment.trimester == trimester)
|
|
|
|
# Récupérer toutes les évaluations avec leurs exercices et éléments
|
|
assessments = query.options(
|
|
db.joinedload(Assessment.exercises).joinedload('grading_elements')
|
|
).all()
|
|
|
|
# Compter le nombre d'élèves dans la classe
|
|
students_count = len(class_group.students)
|
|
|
|
# Initialiser les compteurs
|
|
total_assessments = len(assessments)
|
|
completed_count = 0
|
|
in_progress_count = 0
|
|
not_started_count = 0
|
|
|
|
# Analyser le statut de chaque évaluation
|
|
for assessment in assessments:
|
|
# Utiliser la propriété grading_progress existante
|
|
progress = assessment.grading_progress
|
|
status = progress['status']
|
|
|
|
if status == 'completed':
|
|
completed_count += 1
|
|
elif status in ['in_progress']:
|
|
in_progress_count += 1
|
|
else: # not_started, no_students, no_elements
|
|
not_started_count += 1
|
|
|
|
return {
|
|
'total': total_assessments,
|
|
'completed': completed_count,
|
|
'in_progress': in_progress_count,
|
|
'not_started': not_started_count,
|
|
'students_count': students_count,
|
|
'trimester': trimester
|
|
}
|
|
|
|
except Exception as e:
|
|
from flask import current_app
|
|
current_app.logger.error(f"Erreur dans get_trimester_statistics: {e}", exc_info=True)
|
|
return {
|
|
'total': 0,
|
|
'completed': 0,
|
|
'in_progress': 0,
|
|
'not_started': 0,
|
|
'students_count': 0,
|
|
'trimester': trimester
|
|
}
|
|
|
|
def get_class_results(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""
|
|
Statistiques de résultats pour la classe sur un trimestre.
|
|
|
|
Args:
|
|
class_group: Instance de ClassGroup
|
|
trimester: Trimestre à filtrer (1, 2, 3) ou None pour toutes les évaluations
|
|
|
|
Returns:
|
|
Dict avec moyennes, distribution des notes et métriques statistiques
|
|
"""
|
|
try:
|
|
# Utiliser les évaluations filtrées si disponibles
|
|
if hasattr(class_group, '_filtered_assessments'):
|
|
assessments = class_group._filtered_assessments
|
|
else:
|
|
# Import ici pour éviter la dépendance circulaire
|
|
from models import Assessment
|
|
|
|
# Construire la requête des évaluations avec filtres
|
|
assessments_query = Assessment.query.filter(Assessment.class_group_id == class_group.id)
|
|
|
|
if trimester is not None:
|
|
assessments_query = assessments_query.filter(Assessment.trimester == trimester)
|
|
|
|
assessments = assessments_query.all()
|
|
|
|
if not assessments:
|
|
return self._empty_class_results(class_group, trimester)
|
|
|
|
# Calculer les moyennes par évaluation et par élève
|
|
class_averages = []
|
|
all_individual_scores = [] # Toutes les notes individuelles pour statistiques globales
|
|
student_averages = {} # Moyennes par élève {student_id: [scores]}
|
|
|
|
for assessment in assessments:
|
|
# Utiliser la méthode existante calculate_student_scores
|
|
students_scores, _ = assessment.calculate_student_scores()
|
|
|
|
# Extraire les scores individuels
|
|
individual_scores = []
|
|
for student_id, student_data in students_scores.items():
|
|
score = student_data['total_score']
|
|
max_points = student_data['total_max_points']
|
|
|
|
if max_points > 0: # Éviter la division par zéro
|
|
# Normaliser sur 20 pour comparaison
|
|
normalized_score = (score / max_points) * 20
|
|
individual_scores.append(normalized_score)
|
|
all_individual_scores.append(normalized_score)
|
|
|
|
# Ajouter à la moyenne de l'élève
|
|
if student_id not in student_averages:
|
|
student_averages[student_id] = []
|
|
student_averages[student_id].append(normalized_score)
|
|
|
|
# Calculer la moyenne de classe pour cette évaluation
|
|
if individual_scores:
|
|
import statistics
|
|
class_average = statistics.mean(individual_scores)
|
|
class_averages.append({
|
|
'assessment_id': assessment.id,
|
|
'assessment_title': assessment.title,
|
|
'date': assessment.date.isoformat() if assessment.date else None,
|
|
'class_average': round(class_average, 2),
|
|
'students_evaluated': len(individual_scores),
|
|
'max_possible': 20 # Normalisé sur 20
|
|
})
|
|
|
|
# Calculer les moyennes finales des élèves
|
|
student_final_averages = []
|
|
for student_id, scores in student_averages.items():
|
|
if scores:
|
|
import statistics
|
|
avg = statistics.mean(scores)
|
|
student_final_averages.append(round(avg, 2))
|
|
|
|
# Statistiques globales et distributions
|
|
overall_stats, distribution, student_averages_distribution = self._calculate_statistics_and_distribution(
|
|
student_final_averages
|
|
)
|
|
|
|
return {
|
|
'trimester': trimester,
|
|
'assessments_count': len(assessments),
|
|
'students_count': len(class_group.students),
|
|
'class_averages': class_averages,
|
|
'student_averages': student_final_averages,
|
|
'overall_statistics': overall_stats,
|
|
'distribution': distribution,
|
|
'student_averages_distribution': student_averages_distribution
|
|
}
|
|
|
|
except Exception as e:
|
|
from flask import current_app
|
|
current_app.logger.error(f"Erreur dans get_class_results: {e}", exc_info=True)
|
|
return self._empty_class_results(class_group, trimester)
|
|
|
|
def _empty_class_results(self, class_group, trimester) -> Dict[str, Any]:
|
|
"""Retourne un résultat vide pour get_class_results."""
|
|
return {
|
|
'trimester': trimester,
|
|
'assessments_count': 0,
|
|
'students_count': len(class_group.students),
|
|
'class_averages': [],
|
|
'student_averages': [],
|
|
'overall_statistics': {
|
|
'count': 0,
|
|
'mean': 0,
|
|
'median': 0,
|
|
'min': 0,
|
|
'max': 0,
|
|
'std_dev': 0
|
|
},
|
|
'distribution': [],
|
|
'student_averages_distribution': []
|
|
}
|
|
|
|
def _calculate_statistics_and_distribution(self, student_final_averages) -> Tuple[Dict[str, Any], List[Dict], List[Dict]]:
|
|
"""Calcule les statistiques et la distribution des moyennes."""
|
|
overall_stats = {
|
|
'count': 0,
|
|
'mean': 0,
|
|
'median': 0,
|
|
'min': 0,
|
|
'max': 0,
|
|
'std_dev': 0
|
|
}
|
|
|
|
distribution = []
|
|
student_averages_distribution = []
|
|
|
|
# Utiliser les moyennes des élèves pour les statistiques (cohérent avec l'histogramme)
|
|
if student_final_averages:
|
|
import statistics
|
|
|
|
overall_stats = {
|
|
'count': len(student_final_averages),
|
|
'mean': round(statistics.mean(student_final_averages), 2),
|
|
'median': round(statistics.median(student_final_averages), 2),
|
|
'min': round(min(student_final_averages), 2),
|
|
'max': round(max(student_final_averages), 2),
|
|
'std_dev': round(statistics.stdev(student_final_averages) if len(student_final_averages) > 1 else 0, 2)
|
|
}
|
|
|
|
# Créer l'histogramme des moyennes des élèves (distribution principale)
|
|
if student_final_averages:
|
|
# Bins pour les moyennes des élèves (de 0 à 20)
|
|
avg_bins = list(range(0, 22))
|
|
avg_bin_counts = [0] * (len(avg_bins) - 1)
|
|
|
|
for avg in student_final_averages:
|
|
# Trouver le bon bin
|
|
bin_index = min(int(avg), len(avg_bin_counts) - 1)
|
|
avg_bin_counts[bin_index] += 1
|
|
|
|
# Formatage pour Chart.js
|
|
for i in range(len(avg_bin_counts)):
|
|
if i == len(avg_bin_counts) - 1:
|
|
label = f"{avg_bins[i]}+"
|
|
else:
|
|
label = f"{avg_bins[i]}-{avg_bins[i+1]}"
|
|
|
|
bin_data = {
|
|
'range': label,
|
|
'count': avg_bin_counts[i]
|
|
}
|
|
student_averages_distribution.append(bin_data)
|
|
# Maintenir la compatibilité avec distribution (même données maintenant)
|
|
distribution.append(bin_data.copy())
|
|
|
|
return overall_stats, distribution, student_averages_distribution
|
|
|
|
|
|
class ClassAnalysisService:
|
|
"""
|
|
Service dédié aux analyses de classe (get_domain_analysis, get_competence_analysis).
|
|
Single Responsibility: analyses métier des domaines et compétences.
|
|
"""
|
|
|
|
def __init__(self, db_provider: DatabaseProvider):
|
|
self.db_provider = db_provider
|
|
|
|
def get_domain_analysis(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""
|
|
Analyse les domaines couverts dans les évaluations d'un trimestre.
|
|
|
|
Args:
|
|
class_group: Instance de ClassGroup
|
|
trimester: Trimestre à filtrer (1, 2, 3) ou None pour toutes les évaluations
|
|
|
|
Returns:
|
|
Dict avec liste des domaines, points totaux et nombre d'éléments par domaine
|
|
"""
|
|
try:
|
|
# Import ici pour éviter la dépendance circulaire
|
|
from models import db, GradingElement, Exercise, Assessment, Domain
|
|
|
|
# Utiliser les évaluations filtrées si disponibles
|
|
if hasattr(class_group, '_filtered_assessments'):
|
|
assessment_ids = [a.id for a in class_group._filtered_assessments]
|
|
if not assessment_ids:
|
|
return {'domains': [], 'trimester': trimester}
|
|
|
|
query = db.session.query(
|
|
GradingElement.domain_id,
|
|
Domain.name.label('domain_name'),
|
|
Domain.color.label('domain_color'),
|
|
db.func.sum(GradingElement.max_points).label('total_points'),
|
|
db.func.count(GradingElement.id).label('elements_count')
|
|
).select_from(GradingElement)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.outerjoin(Domain, GradingElement.domain_id == Domain.id)\
|
|
.filter(Exercise.assessment_id.in_(assessment_ids))
|
|
else:
|
|
# Requête originale avec toutes les jointures nécessaires
|
|
query = db.session.query(
|
|
GradingElement.domain_id,
|
|
Domain.name.label('domain_name'),
|
|
Domain.color.label('domain_color'),
|
|
db.func.sum(GradingElement.max_points).label('total_points'),
|
|
db.func.count(GradingElement.id).label('elements_count')
|
|
).select_from(GradingElement)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.join(Assessment, Exercise.assessment_id == Assessment.id)\
|
|
.outerjoin(Domain, GradingElement.domain_id == Domain.id)\
|
|
.filter(Assessment.class_group_id == class_group.id)
|
|
|
|
# Filtrage par trimestre si spécifié
|
|
if trimester is not None:
|
|
query = query.filter(Assessment.trimester == trimester)
|
|
|
|
# Grouper par domaine (y compris les éléments sans domaine)
|
|
query = query.group_by(
|
|
GradingElement.domain_id,
|
|
Domain.name,
|
|
Domain.color
|
|
)
|
|
|
|
results = query.all()
|
|
domains = []
|
|
|
|
for result in results:
|
|
if result.domain_id is not None:
|
|
# Domaine défini
|
|
domains.append({
|
|
'id': result.domain_id,
|
|
'name': result.domain_name,
|
|
'color': result.domain_color,
|
|
'total_points': float(result.total_points) if result.total_points else 0.0,
|
|
'elements_count': result.elements_count
|
|
})
|
|
else:
|
|
# Éléments sans domaine assigné
|
|
domains.append({
|
|
'id': None,
|
|
'name': 'Sans domaine',
|
|
'color': '#6B7280', # Gris neutre
|
|
'total_points': float(result.total_points) if result.total_points else 0.0,
|
|
'elements_count': result.elements_count
|
|
})
|
|
|
|
# Trier par ordre alphabétique, avec "Sans domaine" en dernier
|
|
domains.sort(key=lambda x: (x['name'] == 'Sans domaine', x['name'].lower()))
|
|
|
|
return {
|
|
'domains': domains,
|
|
'trimester': trimester
|
|
}
|
|
|
|
except Exception as e:
|
|
from flask import current_app
|
|
current_app.logger.error(f"Erreur dans get_domain_analysis: {e}", exc_info=True)
|
|
return {
|
|
'domains': [],
|
|
'trimester': trimester
|
|
}
|
|
|
|
def get_competence_analysis(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""
|
|
Analyse les compétences évaluées dans un trimestre.
|
|
|
|
Args:
|
|
class_group: Instance de ClassGroup
|
|
trimester: Trimestre à filtrer (1, 2, 3) ou None pour toutes les évaluations
|
|
|
|
Returns:
|
|
Dict avec liste des compétences, points totaux et nombre d'éléments par compétence
|
|
"""
|
|
try:
|
|
# Import ici pour éviter la dépendance circulaire
|
|
from models import db, GradingElement, Exercise, Assessment
|
|
|
|
# Utiliser les évaluations filtrées si disponibles
|
|
if hasattr(class_group, '_filtered_assessments'):
|
|
assessment_ids = [a.id for a in class_group._filtered_assessments]
|
|
if not assessment_ids:
|
|
return {'competences': [], 'trimester': trimester}
|
|
|
|
query = db.session.query(
|
|
GradingElement.skill.label('skill_name'),
|
|
db.func.sum(GradingElement.max_points).label('total_points'),
|
|
db.func.count(GradingElement.id).label('elements_count')
|
|
).select_from(GradingElement)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.filter(Exercise.assessment_id.in_(assessment_ids))\
|
|
.filter(GradingElement.skill.isnot(None))\
|
|
.filter(GradingElement.skill != '')
|
|
else:
|
|
# Requête optimisée pour analyser les compétences
|
|
query = db.session.query(
|
|
GradingElement.skill.label('skill_name'),
|
|
db.func.sum(GradingElement.max_points).label('total_points'),
|
|
db.func.count(GradingElement.id).label('elements_count')
|
|
).select_from(GradingElement)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.join(Assessment, Exercise.assessment_id == Assessment.id)\
|
|
.filter(Assessment.class_group_id == class_group.id)\
|
|
.filter(GradingElement.skill.isnot(None))\
|
|
.filter(GradingElement.skill != '')
|
|
|
|
# Filtrage par trimestre si spécifié
|
|
if trimester is not None:
|
|
query = query.filter(Assessment.trimester == trimester)
|
|
|
|
# Grouper par compétence
|
|
query = query.group_by(GradingElement.skill)
|
|
|
|
results = query.all()
|
|
|
|
# Récupérer la configuration des compétences pour les couleurs
|
|
from app_config import config_manager
|
|
competences_config = {comp['name']: comp for comp in config_manager.get_competences_list()}
|
|
|
|
competences = []
|
|
for result in results:
|
|
skill_name = result.skill_name
|
|
# Récupérer la couleur depuis la configuration ou utiliser une couleur par défaut
|
|
config = competences_config.get(skill_name, {})
|
|
color = config.get('color', '#6B7280') # Gris neutre par défaut
|
|
|
|
competences.append({
|
|
'name': skill_name,
|
|
'color': color,
|
|
'total_points': float(result.total_points) if result.total_points else 0.0,
|
|
'elements_count': result.elements_count
|
|
})
|
|
|
|
# Trier par ordre alphabétique
|
|
competences.sort(key=lambda x: x['name'].lower())
|
|
|
|
return {
|
|
'competences': competences,
|
|
'trimester': trimester
|
|
}
|
|
|
|
except Exception as e:
|
|
from flask import current_app
|
|
current_app.logger.error(f"Erreur dans get_competence_analysis: {e}", exc_info=True)
|
|
return {
|
|
'competences': [],
|
|
'trimester': trimester
|
|
}
|
|
|
|
|
|
# =================== FACADE ÉTENDUE ===================
|
|
|
|
class ClassServicesFacade:
|
|
"""
|
|
Facade qui regroupe tous les services pour les classes.
|
|
Point d'entrée unique pour les méthodes de ClassGroup.
|
|
"""
|
|
|
|
def __init__(self, db_provider: DatabaseProvider):
|
|
self.statistics_service = ClassStatisticsService(db_provider)
|
|
self.analysis_service = ClassAnalysisService(db_provider)
|
|
|
|
def get_trimester_statistics(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""Point d'entrée pour les statistiques trimestrielles."""
|
|
return self.statistics_service.get_trimester_statistics(class_group, trimester)
|
|
|
|
def get_class_results(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""Point d'entrée pour les résultats de classe."""
|
|
return self.statistics_service.get_class_results(class_group, trimester)
|
|
|
|
def get_domain_analysis(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""Point d'entrée pour l'analyse des domaines."""
|
|
return self.analysis_service.get_domain_analysis(class_group, trimester)
|
|
|
|
def get_competence_analysis(self, class_group, trimester=None) -> Dict[str, Any]:
|
|
"""Point d'entrée pour l'analyse des compétences."""
|
|
return self.analysis_service.get_competence_analysis(class_group, trimester)
|
|
|
|
|
|
# =================== FACTORY FUNCTION ===================
|
|
|
|
def create_assessment_services() -> AssessmentServicesFacade:
|
|
"""
|
|
Factory function pour créer une instance configurée de AssessmentServicesFacade.
|
|
Point d'entrée standard pour l'utilisation des services refactorisés.
|
|
"""
|
|
from app_config import config_manager
|
|
from models import db
|
|
|
|
config_provider = ConfigProvider(config_manager)
|
|
db_provider = DatabaseProvider(db)
|
|
|
|
return AssessmentServicesFacade(config_provider, db_provider) |