✨(backend) add user reconciliation csv import task

Let an admin bulk-create reconciliation requests from an external CSV
export instead of entering each pair by hand.

src/backend/core/tasks/user_reconciliation.py

@@ -0,0 +1,141 @@
+"""Processing tasks for user reconciliation CSV imports."""
+
+import csv
+import logging
+import traceback
+import uuid
+
+from django.core.exceptions import ValidationError
+from django.core.validators import validate_email
+from django.db import IntegrityError
+
+from botocore.exceptions import ClientError
+
+from core.models import UserReconciliation, UserReconciliationCsvImport
+
+from drive.celery_app import app
+
+logger = logging.getLogger(__name__)
+
+
+def _process_row(row, job, counters):
+    """Process a single row from the CSV file."""
+
+    source_unique_id = row["id"].strip()
+
+    # Skip entries if they already exist with this source_unique_id
+    if UserReconciliation.objects.filter(source_unique_id=source_unique_id).exists():
+        counters["already_processed_source_ids"] += 1
+        return counters
+
+    active_email_checked = row.get("active_email_checked", "0") == "1"
+    inactive_email_checked = row.get("inactive_email_checked", "0") == "1"
+
+    active_email = row["active_email"]
+    inactive_emails = row["inactive_email"].split("|")
+    try:
+        validate_email(active_email)
+    except ValidationError:
+        job.send_reconciliation_error_email(
+            recipient_email=inactive_emails[0], other_email=active_email
+        )
+        job.logs += f"Invalid active email address on row {source_unique_id}."
+        counters["rows_with_errors"] += 1
+        return counters
+
+    for inactive_email in inactive_emails:
+        try:
+            validate_email(inactive_email)
+        except (ValidationError, ValueError):
+            job.send_reconciliation_error_email(
+                recipient_email=active_email, other_email=inactive_email
+            )
+            job.logs += f"Invalid inactive email address on row {source_unique_id}.\n"
+            counters["rows_with_errors"] += 1
+            continue
+
+        if inactive_email == active_email:
+            job.send_reconciliation_error_email(
+                recipient_email=active_email, other_email=inactive_email
+            )
+            job.logs += (
+                f"Error on row {source_unique_id}: "
+                f"{active_email} set as both active and inactive email.\n"
+            )
+            counters["rows_with_errors"] += 1
+            continue
+
+        UserReconciliation.objects.create(
+            active_email=active_email,
+            inactive_email=inactive_email,
+            active_email_checked=active_email_checked,
+            inactive_email_checked=inactive_email_checked,
+            active_email_confirmation_id=uuid.uuid4(),
+            inactive_email_confirmation_id=uuid.uuid4(),
+            source_unique_id=source_unique_id,
+            status="pending",
+        )
+        counters["rec_entries_created"] += 1
+
+    return counters
+
+
+@app.task
+def user_reconciliation_csv_import_job(job_id):
+    """Process a UserReconciliationCsvImport job.
+
+    Creates UserReconciliation entries from the CSV file.
+
+    Does some sanity checks on the data:
+    - active_email and inactive_email must be valid email addresses
+    - active_email and inactive_email cannot be the same
+
+    Rows with errors are logged in the job logs and skipped, but do not cause
+    the entire job to fail or prevent the next rows from being processed.
+    """
+    try:
+        job = UserReconciliationCsvImport.objects.get(id=job_id)
+    except UserReconciliationCsvImport.DoesNotExist:
+        logger.warning("CSV import job %s no longer exists; skipping.", job_id)
+        return
+
+    job.status = "running"
+    job.save()
+
+    counters = {
+        "rec_entries_created": 0,
+        "rows_with_errors": 0,
+        "already_processed_source_ids": 0,
+    }
+
+    try:
+        with job.file.open(mode="r") as file:
+            reader = csv.DictReader(file)
+
+            if not {"active_email", "inactive_email", "id"}.issubset(reader.fieldnames or []):
+                raise KeyError("CSV is missing mandatory columns: active_email, inactive_email, id")
+
+            for row in reader:
+                counters = _process_row(row, job, counters)
+
+        job.status = "done"
+        job.logs += (
+            f"Import completed successfully. {reader.line_num} rows processed."
+            f" {counters['rec_entries_created']} reconciliation entries created."
+            f" {counters['already_processed_source_ids']} rows were already processed."
+            f" {counters['rows_with_errors']} rows had errors."
+        )
+    except (
+        csv.Error,
+        KeyError,
+        ValidationError,
+        ValueError,
+        IntegrityError,
+        OSError,
+        ClientError,
+    ) as exception:
+        # Catch expected I/O/CSV/model errors and record traceback in logs for debugging
+        job.status = "error"
+        job.logs += f"{exception!s}\n{traceback.format_exc()}"
+    finally:
+        job.save()

src/backend/core/tests/data/example_reconciliation_basic.csv

@@ -0,0 +1,3 @@
+active_email,inactive_email,active_email_checked,inactive_email_checked,id
+active1@example.com,inactive1@example.com,1,1,1
+active2@example.com,inactive2@example.com,1,1,2

src/backend/core/tests/data/example_reconciliation_error.csv

@@ -0,0 +1,2 @@
+active_email,inactive_email,active_email_checked,inactive_email_checked,id
+active1@example.com,,1,1,40

src/backend/core/tests/data/example_reconciliation_grist_form.csv

@@ -0,0 +1,2 @@
+active_email,inactive_email,active_email_checked,inactive_email_checked,id
+active1@example.com,inactive1@example.com|inactive2@example.com,1,1,10

src/backend/core/tests/data/example_reconciliation_missing_column.csv

@@ -0,0 +1,2 @@
+active_email,inactive_email
+active1@example.com,inactive1@example.com

src/backend/core/tests/tasks/test_user_reconciliation.py

@@ -0,0 +1,120 @@
+"""Tests for the user reconciliation CSV import task."""
+
+import logging
+import uuid
+from pathlib import Path
+
+from django.core import mail
+from django.core.files.base import ContentFile
+
+import pytest
+
+from core import factories, models
+from core.tasks.user_reconciliation import user_reconciliation_csv_import_job
+
+pytestmark = pytest.mark.django_db
+
+DATA_DIR = Path(__file__).parent.parent / "data"
+
+
+def make_import(filename):
+    """Create a UserReconciliationCsvImport from a fixture CSV file."""
+    with open(DATA_DIR / filename, "rb") as file:
+        csv_file = ContentFile(file.read(), name=filename)
+    return models.UserReconciliationCsvImport.objects.create(file=csv_file)
+
+
+def test_user_reconciliation_csv_import_missing_job(caplog):
+    """A missing import job logs a warning and returns without raising."""
+    with caplog.at_level(logging.WARNING):
+        user_reconciliation_csv_import_job(uuid.uuid4())
+
+    assert "no longer exists" in caplog.text
+
+
+def test_user_reconciliation_csv_import_creates_entries():
+    """A well-formed CSV creates one reconciliation entry per row."""
+    for email in ["active1", "inactive1", "active2", "inactive2"]:
+        factories.UserFactory(email=f"{email}@example.com")
+
+    csv_import = make_import("example_reconciliation_basic.csv")
+    user_reconciliation_csv_import_job(csv_import.id)
+    csv_import.refresh_from_db()
+
+    assert csv_import.status == "done"
+    assert models.UserReconciliation.objects.count() == 2
+    assert models.UserReconciliation.objects.filter(status="ready").count() == 2
+
+
+def test_user_reconciliation_csv_import_empty_file():
+    """An empty CSV fails into the error status instead of hanging in running."""
+    csv_import = models.UserReconciliationCsvImport.objects.create(
+        file=ContentFile(b"", name="empty.csv")
+    )
+    user_reconciliation_csv_import_job(csv_import.id)
+    csv_import.refresh_from_db()
+
+    assert csv_import.status == "error"
+    assert "missing mandatory columns" in csv_import.logs
+
+
+def test_user_reconciliation_csv_import_missing_column():
+    """A CSV missing a mandatory column fails into the error status."""
+    csv_import = make_import("example_reconciliation_missing_column.csv")
+    user_reconciliation_csv_import_job(csv_import.id)
+    csv_import.refresh_from_db()
+
+    assert csv_import.status == "error"
+    assert "missing mandatory columns" in csv_import.logs
+    assert models.UserReconciliation.objects.count() == 0
+
+
+def test_user_reconciliation_csv_import_invalid_email():
+    """An invalid email is logged and triggers an error email, without failing the job."""
+    factories.UserFactory(email="active1@example.com")
+
+    csv_import = make_import("example_reconciliation_error.csv")
+    user_reconciliation_csv_import_job(csv_import.id)
+    csv_import.refresh_from_db()
+
+    assert csv_import.status == "done"
+    assert "Invalid inactive email address on row 40" in csv_import.logs
+    assert models.UserReconciliation.objects.count() == 0
+
+    # pylint: disable-next=no-member
+    assert len(mail.outbox) == 1
+    # pylint: disable-next=no-member
+    assert mail.outbox[0].to == ["active1@example.com"]
+
+
+def test_user_reconciliation_csv_import_multiple_inactive_emails():
+    """A single row may list several inactive emails separated by a pipe."""
+    for email in ["active1", "inactive1", "inactive2"]:
+        factories.UserFactory(email=f"{email}@example.com")
+
+    csv_import = make_import("example_reconciliation_grist_form.csv")
+    user_reconciliation_csv_import_job(csv_import.id)
+    csv_import.refresh_from_db()
+
+    assert csv_import.status == "done"
+    assert models.UserReconciliation.objects.count() == 2
+    assert set(models.UserReconciliation.objects.values_list("inactive_email", flat=True)) == {
+        "inactive1@example.com",
+        "inactive2@example.com",
+    }
+
+
+def test_user_reconciliation_csv_import_is_idempotent():
+    """Re-importing the same source ids does not create duplicate entries."""
+    for email in ["active1", "inactive1", "active2", "inactive2"]:
+        factories.UserFactory(email=f"{email}@example.com")
+
+    csv_import = make_import("example_reconciliation_basic.csv")
+    user_reconciliation_csv_import_job(csv_import.id)
+
+    second_import = make_import("example_reconciliation_basic.csv")
+    user_reconciliation_csv_import_job(second_import.id)
+    second_import.refresh_from_db()
+
+    assert models.UserReconciliation.objects.count() == 2
+    assert "already processed" in second_import.logs