""" 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)