feat: add csv import
This commit is contained in:
328
services/csv_import_service.py
Normal file
328
services/csv_import_service.py
Normal file
@@ -0,0 +1,328 @@
|
||||
"""
|
||||
Service d'import CSV pour les élèves.
|
||||
Permet d'importer plusieurs élèves en une fois depuis un fichier CSV.
|
||||
"""
|
||||
|
||||
import csv
|
||||
import re
|
||||
from typing import List, Tuple, Dict, Any, Optional
|
||||
from dataclasses import dataclass
|
||||
from io import StringIO
|
||||
from flask import current_app
|
||||
from datetime import date
|
||||
|
||||
from models import db, Student, StudentEnrollment, ClassGroup
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImportedStudent:
|
||||
"""Représente un élève extrait du CSV."""
|
||||
first_name: str
|
||||
last_name: str
|
||||
email: Optional[str] = None
|
||||
line_number: int = 0
|
||||
raw_name: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImportResult:
|
||||
"""Résultat de l'import avec statistiques détaillées."""
|
||||
success: bool
|
||||
total_lines: int
|
||||
imported_count: int
|
||||
skipped_count: int
|
||||
error_count: int
|
||||
imported_students: List[ImportedStudent]
|
||||
skipped_students: List[Dict[str, Any]] # Doublons
|
||||
errors: List[Dict[str, str]] # Erreurs avec détails
|
||||
message: str
|
||||
|
||||
|
||||
class CSVImportService:
|
||||
"""Service pour l'import d'élèves depuis un fichier CSV."""
|
||||
|
||||
def __init__(self):
|
||||
self.delimiter = ';' # Séparateur CSV détecté
|
||||
|
||||
def extract_name_parts(self, full_name: str) -> Tuple[str, str]:
|
||||
"""
|
||||
Extrait le nom et prénom depuis un nom complet.
|
||||
|
||||
Stratégie pour "AABIDA LAHDILI Fatima Zahra":
|
||||
- Les mots en MAJUSCULES → nom de famille
|
||||
- Les autres mots → prénoms
|
||||
|
||||
Args:
|
||||
full_name: Nom complet de l'élève
|
||||
|
||||
Returns:
|
||||
Tuple (nom_de_famille, prenoms)
|
||||
"""
|
||||
if not full_name or not full_name.strip():
|
||||
return "", ""
|
||||
|
||||
# Nettoyer et séparer les mots
|
||||
parts = full_name.strip().split()
|
||||
|
||||
if not parts:
|
||||
return "", ""
|
||||
|
||||
# Séparer les mots en majuscules des autres
|
||||
uppercase_words = []
|
||||
other_words = []
|
||||
|
||||
for part in parts:
|
||||
# Un mot est considéré en majuscules s'il contient au moins 2 caractères
|
||||
# et que tous ses caractères alphabétiques sont en majuscules
|
||||
if len(part) >= 2 and part.isupper():
|
||||
uppercase_words.append(part)
|
||||
else:
|
||||
other_words.append(part)
|
||||
|
||||
# Le nom de famille = mots en majuscules
|
||||
last_name = " ".join(uppercase_words) if uppercase_words else parts[0]
|
||||
|
||||
# Les prénoms = autres mots, ou le reste si pas de majuscules détectées
|
||||
if uppercase_words and other_words:
|
||||
first_name = " ".join(other_words)
|
||||
elif len(parts) > 1:
|
||||
# Fallback: premier mot = nom, reste = prénoms
|
||||
last_name = parts[0]
|
||||
first_name = " ".join(parts[1:])
|
||||
else:
|
||||
# Un seul mot : considérer comme prénom
|
||||
first_name = parts[0]
|
||||
last_name = ""
|
||||
|
||||
return last_name.strip(), first_name.strip()
|
||||
|
||||
def parse_csv_content(self, csv_content: str) -> List[ImportedStudent]:
|
||||
"""
|
||||
Parse le contenu CSV et extrait les élèves.
|
||||
|
||||
Args:
|
||||
csv_content: Contenu du fichier CSV
|
||||
|
||||
Returns:
|
||||
Liste des élèves extraits
|
||||
"""
|
||||
students = []
|
||||
|
||||
try:
|
||||
# Lire le CSV avec le délimiteur détecté
|
||||
csv_reader = csv.reader(StringIO(csv_content), delimiter=self.delimiter)
|
||||
|
||||
# Ignorer la première ligne (headers)
|
||||
headers = next(csv_reader, None)
|
||||
if not headers:
|
||||
return students
|
||||
|
||||
for line_number, row in enumerate(csv_reader, start=2): # Start=2 car ligne 1 = headers
|
||||
if not row or len(row) < 1:
|
||||
continue
|
||||
|
||||
# Prendre la première colonne (nom complet)
|
||||
full_name = row[0].strip().strip('"')
|
||||
|
||||
if not full_name or full_name.lower() in ['undefined', 'null', '']:
|
||||
continue
|
||||
|
||||
# Extraire nom et prénom
|
||||
last_name, first_name = self.extract_name_parts(full_name)
|
||||
|
||||
if not first_name and not last_name:
|
||||
continue
|
||||
|
||||
student = ImportedStudent(
|
||||
first_name=first_name or "Prénom", # Fallback si pas de prénom
|
||||
last_name=last_name or "Nom", # Fallback si pas de nom
|
||||
line_number=line_number,
|
||||
raw_name=full_name
|
||||
)
|
||||
|
||||
students.append(student)
|
||||
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"Erreur parsing CSV: {e}")
|
||||
raise ValueError(f"Erreur lors de l'analyse du fichier CSV: {str(e)}")
|
||||
|
||||
return students
|
||||
|
||||
def find_existing_students(self, students: List[ImportedStudent]) -> List[Student]:
|
||||
"""
|
||||
Trouve les élèves existants par nom et prénom.
|
||||
|
||||
Args:
|
||||
students: Liste des élèves à vérifier
|
||||
|
||||
Returns:
|
||||
Liste des élèves existants en base
|
||||
"""
|
||||
existing = []
|
||||
|
||||
for student in students:
|
||||
existing_student = Student.query.filter(
|
||||
db.func.lower(Student.first_name) == student.first_name.lower(),
|
||||
db.func.lower(Student.last_name) == student.last_name.lower()
|
||||
).first()
|
||||
|
||||
if existing_student:
|
||||
existing.append(existing_student)
|
||||
|
||||
return existing
|
||||
|
||||
def import_students_to_class(
|
||||
self,
|
||||
csv_content: str,
|
||||
class_group_id: int,
|
||||
enrollment_date: date = None,
|
||||
skip_duplicates: bool = True
|
||||
) -> ImportResult:
|
||||
"""
|
||||
Importe les élèves du CSV dans une classe.
|
||||
|
||||
Args:
|
||||
csv_content: Contenu du fichier CSV
|
||||
class_group_id: ID de la classe de destination
|
||||
enrollment_date: Date d'inscription (par défaut: aujourd'hui)
|
||||
skip_duplicates: Ignorer les doublons (True) ou échouer (False)
|
||||
|
||||
Returns:
|
||||
Résultat de l'import avec statistiques
|
||||
"""
|
||||
if enrollment_date is None:
|
||||
enrollment_date = date.today()
|
||||
|
||||
# Vérifier que la classe existe
|
||||
class_group = ClassGroup.query.get(class_group_id)
|
||||
if not class_group:
|
||||
return ImportResult(
|
||||
success=False,
|
||||
total_lines=0,
|
||||
imported_count=0,
|
||||
skipped_count=0,
|
||||
error_count=1,
|
||||
imported_students=[],
|
||||
skipped_students=[],
|
||||
errors=[{"line": 0, "error": f"Classe ID {class_group_id} non trouvée"}],
|
||||
message="Classe non trouvée"
|
||||
)
|
||||
|
||||
try:
|
||||
# Parser le CSV
|
||||
parsed_students = self.parse_csv_content(csv_content)
|
||||
|
||||
if not parsed_students:
|
||||
return ImportResult(
|
||||
success=True,
|
||||
total_lines=0,
|
||||
imported_count=0,
|
||||
skipped_count=0,
|
||||
error_count=0,
|
||||
imported_students=[],
|
||||
skipped_students=[],
|
||||
errors=[],
|
||||
message="Aucun élève trouvé dans le fichier CSV"
|
||||
)
|
||||
|
||||
# Vérifier les doublons
|
||||
imported_students = []
|
||||
skipped_students = []
|
||||
errors = []
|
||||
|
||||
for student_data in parsed_students:
|
||||
try:
|
||||
# Vérifier si l'élève existe déjà
|
||||
existing_student = Student.query.filter(
|
||||
db.func.lower(Student.first_name) == student_data.first_name.lower(),
|
||||
db.func.lower(Student.last_name) == student_data.last_name.lower()
|
||||
).first()
|
||||
|
||||
if existing_student:
|
||||
if skip_duplicates:
|
||||
skipped_students.append({
|
||||
"line": student_data.line_number,
|
||||
"name": f"{student_data.first_name} {student_data.last_name}",
|
||||
"reason": "Élève déjà existant"
|
||||
})
|
||||
continue
|
||||
else:
|
||||
errors.append({
|
||||
"line": student_data.line_number,
|
||||
"error": f"Élève déjà existant: {student_data.first_name} {student_data.last_name}"
|
||||
})
|
||||
continue
|
||||
|
||||
# Créer le nouvel élève
|
||||
new_student = Student(
|
||||
first_name=student_data.first_name,
|
||||
last_name=student_data.last_name,
|
||||
email=student_data.email
|
||||
)
|
||||
|
||||
db.session.add(new_student)
|
||||
db.session.flush() # Pour obtenir l'ID
|
||||
|
||||
# Créer l'inscription dans la classe
|
||||
enrollment = StudentEnrollment(
|
||||
student_id=new_student.id,
|
||||
class_group_id=class_group_id,
|
||||
enrollment_date=enrollment_date,
|
||||
enrollment_reason="Import CSV"
|
||||
)
|
||||
|
||||
db.session.add(enrollment)
|
||||
imported_students.append(student_data)
|
||||
|
||||
current_app.logger.info(f"Élève importé: {new_student.full_name} en {class_group.name}")
|
||||
|
||||
except Exception as e:
|
||||
errors.append({
|
||||
"line": student_data.line_number,
|
||||
"error": f"Erreur création élève {student_data.first_name} {student_data.last_name}: {str(e)}"
|
||||
})
|
||||
current_app.logger.error(f"Erreur import élève ligne {student_data.line_number}: {e}")
|
||||
|
||||
# Valider la transaction
|
||||
if errors and not skip_duplicates:
|
||||
db.session.rollback()
|
||||
return ImportResult(
|
||||
success=False,
|
||||
total_lines=len(parsed_students),
|
||||
imported_count=0,
|
||||
skipped_count=len(skipped_students),
|
||||
error_count=len(errors),
|
||||
imported_students=[],
|
||||
skipped_students=skipped_students,
|
||||
errors=errors,
|
||||
message=f"Import échoué: {len(errors)} erreur(s)"
|
||||
)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return ImportResult(
|
||||
success=True,
|
||||
total_lines=len(parsed_students),
|
||||
imported_count=len(imported_students),
|
||||
skipped_count=len(skipped_students),
|
||||
error_count=len(errors),
|
||||
imported_students=imported_students,
|
||||
skipped_students=skipped_students,
|
||||
errors=errors,
|
||||
message=f"Import réussi: {len(imported_students)} élève(s) importé(s), {len(skipped_students)} ignoré(s)"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
current_app.logger.error(f"Erreur import CSV classe {class_group_id}: {e}")
|
||||
return ImportResult(
|
||||
success=False,
|
||||
total_lines=0,
|
||||
imported_count=0,
|
||||
skipped_count=0,
|
||||
error_count=1,
|
||||
imported_students=[],
|
||||
skipped_students=[],
|
||||
errors=[{"line": 0, "error": str(e)}],
|
||||
message=f"Erreur lors de l'import: {str(e)}"
|
||||
)
|
||||
Reference in New Issue
Block a user