benevoles/backend/app/importing/service.py

"""Cœur de l'import : suggestion de mapping, validation, exécution.

L'import ne porte que l'identité des bénévoles (nom, email, téléphone,
commentaire). Les disponibilités se déclarent dans l'application.

Détection des doublons : upsert sur (email, nom + prénom) — l'email seul ne
suffit pas, plusieurs bénévoles peuvent le partager (couple, famille) ; sans
email, détection sur (nom, prénom) signalée sans upsert automatique. Un même
email revu avec un nom différent crée un bénévole de plus (signalé), jamais
ignoré.
"""

import re

from sqlmodel import Session, select

from app.importing.parser import ParsedTable
from app.importing.schemas import (
    ImportMapping,
    ImportReport,
    Issue,
    RowReport,
)
from app.importing.text import normalize
from app.models import Edition, Volunteer

_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

_FIELD_HINTS: dict[str, tuple[str, ...]] = {
    "email": ("mail", "courriel"),
    "first_name": ("prenom", "first name", "firstname", "given name"),
    "last_name": ("nom", "last name", "lastname", "surname", "famille"),
    "phone": ("tel", "phone", "portable", "mobile", "gsm"),
    "comment": ("commentaire", "remarque", "note", "precision"),
}


def suggest_mapping(headers: list[str]) -> dict[str, str]:
    """Association approximative en-tête → champ, à confirmer côté client."""
    result: dict[str, str] = {}
    used: set[str] = set()
    for field, hints in _FIELD_HINTS.items():
        for header in headers:
            if header in used:
                continue
            if any(hint in normalize(header) for hint in hints):
                result[field] = header
                used.add(header)
                break
    return result


def validate_mapping(
    table: ParsedTable, mapping: ImportMapping
) -> list[Issue]:
    """Contrôles préalables. Retourne une liste vide si le mapping est correct."""
    errors: list[Issue] = []
    headers = set(table.headers)
    columns = mapping.columns

    for field in ("first_name", "last_name"):
        header = getattr(columns, field)
        if not header:
            errors.append(
                Issue(
                    code="import.mapping.required_field_missing",
                    severity="ERROR",
                    field=field,
                )
            )
        elif header not in headers:
            errors.append(
                Issue(
                    code="import.mapping.column_not_found",
                    severity="ERROR",
                    field=field,
                    params={"column": header},
                )
            )

    for field in ("email", "phone", "comment"):
        header = getattr(columns, field)
        if header and header not in headers:
            errors.append(
                Issue(
                    code="import.mapping.column_not_found",
                    severity="ERROR",
                    field=field,
                    params={"column": header},
                )
            )
    return errors


def run_import(
    *,
    session: Session,
    edition: Edition,
    table: ParsedTable,
    mapping: ImportMapping,
) -> ImportReport:
    """Exécute l'import. `mapping.dry_run` → aucune persistance (rollback final).

    Le mapping est supposé déjà validé par `validate_mapping`.
    """
    columns = mapping.columns

    existing = list(
        session.exec(
            select(Volunteer).where(Volunteer.edition_id == edition.id)
        )
    )
    # Plusieurs bénévoles peuvent partager un même email (couple, famille) :
    # l'email seul ne suffit pas à identifier un bénévole existant, il faut
    # l'associer au nom. `by_name` ne sert qu'aux bénévoles sans email.
    by_email_and_name: dict[tuple[str, tuple[str, str]], Volunteer] = {}
    by_name: dict[tuple[str, str], Volunteer] = {}
    for volunteer in existing:
        key = (normalize(volunteer.last_name), normalize(volunteer.first_name))
        if volunteer.email:
            by_email_and_name[(volunteer.email.lower(), key)] = volunteer
        else:
            by_name.setdefault(key, volunteer)

    report = ImportReport(dry_run=mapping.dry_run, total_rows=len(table.rows))
    row_seen: dict[tuple[str, tuple[str, str]], int] = {}
    email_seen: dict[str, int] = {}
    name_seen: dict[tuple[str, str], int] = {}

    for index, row in enumerate(table.as_dicts()):
        line = index + 2  # +1 en-tête, +1 pour une numérotation à partir de 1
        entry = RowReport(line=line, action="SKIPPED")

        first_name = (row.get(columns.first_name) or "").strip()
        last_name = (row.get(columns.last_name) or "").strip()
        email_raw = (
            (row.get(columns.email) or "").strip() if columns.email else ""
        )
        email = email_raw.lower() or None

        if not first_name or not last_name:
            entry.issues.append(
                Issue(
                    code="import.row.missing_required_field",
                    severity="ERROR",
                    field="first_name" if not first_name else "last_name",
                )
            )
            report.rows.append(entry)
            report.skipped += 1
            continue

        if email and not _EMAIL_RE.match(email):
            entry.issues.append(
                Issue(
                    code="import.row.invalid_email",
                    severity="ERROR",
                    field="email",
                    params={"value": email_raw},
                )
            )
            report.rows.append(entry)
            report.skipped += 1
            continue

        name_key = (normalize(last_name), normalize(first_name))
        if email:
            row_key = (email, name_key)
            if row_key in row_seen:
                # Même email ET même nom déjà vus : la ligne est répétée dans
                # le fichier (copier-coller), pas deux bénévoles distincts.
                entry.issues.append(
                    Issue(
                        code="import.row.duplicate_email_in_file",
                        severity="ERROR",
                        field="email",
                        params={"value": email_raw, "first_line": row_seen[row_key]},
                    )
                )
                report.rows.append(entry)
                report.skipped += 1
                continue
            if email in email_seen:
                # Même email, nom différent : deux bénévoles peuvent partager
                # une adresse (couple, famille) — on les crée tous les deux,
                # simplement signalé pour la coordination.
                entry.issues.append(
                    Issue(
                        code="import.row.shared_email",
                        severity="WARNING",
                        field="email",
                        params={"value": email_raw, "first_line": email_seen[email]},
                    )
                )
            row_seen[row_key] = line
            email_seen.setdefault(email, line)
        else:
            if name_key in name_seen:
                entry.issues.append(
                    Issue(
                        code="import.row.duplicate_name_in_file",
                        severity="WARNING",
                        params={"first_line": name_seen[name_key]},
                    )
                )
            else:
                name_seen[name_key] = line

        volunteer = by_email_and_name.get((email, name_key)) if email else None
        if volunteer is not None:
            entry.action = "UPDATED"
            entry.issues.append(
                Issue(
                    code="import.row.updated_existing",
                    severity="WARNING",
                    params={"volunteer_id": str(volunteer.id)},
                )
            )
        else:
            if not email:
                match = by_name.get(name_key)
                if match is not None:
                    entry.issues.append(
                        Issue(
                            code="import.row.possible_duplicate_name",
                            severity="WARNING",
                            params={"volunteer_id": str(match.id)},
                        )
                    )
            volunteer = Volunteer(
                edition_id=edition.id,
                first_name=first_name,
                last_name=last_name,
            )
            session.add(volunteer)
            entry.action = "CREATED"

        volunteer.first_name = first_name
        volunteer.last_name = last_name
        if columns.email:
            volunteer.email = email_raw or None
        if columns.phone:
            volunteer.phone = (row.get(columns.phone) or "").strip() or None
        if columns.comment:
            volunteer.comment = (row.get(columns.comment) or "").strip() or None

        session.flush()

        entry.volunteer_id = volunteer.id
        report.rows.append(entry)
        if entry.action == "CREATED":
            report.created += 1
        else:
            report.updated += 1

    if mapping.dry_run:
        session.rollback()
    else:
        session.commit()
    return report