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

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)
        )
    )
    by_email = {v.email.lower(): v for v in existing if v.email}
    by_name: dict[tuple[str, str], Volunteer] = {}
    for volunteer in existing:
        if not volunteer.email:
            key = (normalize(volunteer.last_name), normalize(volunteer.first_name))
            by_name.setdefault(key, volunteer)

    report = ImportReport(dry_run=mapping.dry_run, total_rows=len(table.rows))
    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

        if email:
            if email in email_seen:
                entry.issues.append(
                    Issue(
                        code="import.row.duplicate_email_in_file",
                        severity="ERROR",
                        field="email",
                        params={"value": email_raw, "first_line": email_seen[email]},
                    )
                )
                report.rows.append(entry)
                report.skipped += 1
                continue
            email_seen[email] = line
        else:
            key = (normalize(last_name), normalize(first_name))
            if key in name_seen:
                entry.issues.append(
                    Issue(
                        code="import.row.duplicate_name_in_file",
                        severity="WARNING",
                        params={"first_line": name_seen[key]},
                    )
                )
            else:
                name_seen[key] = line

        volunteer = by_email.get(email) 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((normalize(last_name), normalize(first_name)))
                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