✨(backend) manage reconciliation requests for user accounts

For now, the reconciliation requests are imported through CSV in the
Django admin, which sends confirmation email to both addresses.
When both are checked, the actual reconciliation is processed, in a
threefold process (update document acess, update invitations, update
user status.)

Author: Ash-Crow

src/backend/core/admin.py

@@ -1,12 +1,17 @@
 """Admin classes and registrations for core app."""
 
-from django.contrib import admin
+from os import remove
+from django.contrib import admin, messages
 from django.contrib.auth import admin as auth_admin
+from django.forms import ModelForm
+from django.shortcuts import render, redirect
+from django.urls import path
 from django.utils.translation import gettext_lazy as _
 
 from treebeard.admin import TreeAdmin
 
-from . import models
+from core import models
+from core.tasks.user_reconciliation import user_reconciliation_csv_import_job
 
 
 class TemplateAccessInline(admin.TabularInline):
@@ -104,6 +109,87 @@ class UserAdmin(auth_admin.UserAdmin):
     search_fields = ("id", "sub", "admin_email", "email", "full_name")
 
 
+class UserReconciliationCsvImportForm(ModelForm):
+    class Meta:
+        model = models.UserReconciliationCsvImport
+        fields = ("file",)
+
+
+@admin.register(models.UserReconciliationCsvImport)
+class UserReconciliationCsvImportAdmin(admin.ModelAdmin):
+    list_display = ("id", "created_at", "status")
+
+    def save_model(self, request, obj, form, change):
+        super().save_model(request, obj, form, change)
+
+        if not change:
+            user_reconciliation_csv_import_job.delay(obj.pk)
+            messages.success(request, "Import job created and queued.")
+        return redirect("..")
+
+
+@admin.action(description="Process selected user reconciliations")
+def process_reconciliation(modeladmin, request, queryset):
+    """
+    Admin action to process selected user reconciliations.
+    The action will process only entries that are ready and have both emails checked.
+
+    Its action is threefold:
+    - Transfer document accesses from inactive to active user, updating roles as needed.
+    - Transfer invitations from inactive to active user, updating roles as needed.
+    - Activate the active user and deactivate the inactive user.
+    """
+    processable_entries = queryset.filter(
+        status="ready", active_email_checked=True, inactive_email_checked=True
+    )
+
+    # Prepare the bulk operations
+    updated_documentaccess = []
+    removed_documentaccess = []
+    updated_invitations = []
+    removed_invitations = []
+    update_users_active_status = []
+
+    for entry in processable_entries:
+        new_updated_documentaccess, new_removed_documentaccess = (
+            entry.process_documentaccess_reconciliation()
+        )
+        updated_documentaccess += new_updated_documentaccess
+        removed_documentaccess += new_removed_documentaccess
+
+        new_updated_invitations, new_removed_invitations = (
+            entry.process_invitation_reconciliation()
+        )
+        updated_invitations += new_updated_invitations
+        removed_invitations += new_removed_invitations
+
+        entry.active_user.is_active = True
+        entry.inactive_user.is_active = False
+        update_users_active_status.append(entry.active_user)
+        update_users_active_status.append(entry.inactive_user)
+
+    # Actually perform the bulk operations
+    models.DocumentAccess.objects.bulk_update(updated_documentaccess, ["user", "role"])
+
+    if len(removed_documentaccess):
+        ids_to_delete = [rd.id for rd in removed_documentaccess]
+        models.DocumentAccess.objects.filter(id__in=ids_to_delete).delete()
+
+    models.Invitation.objects.bulk_update(updated_invitations, ["email", "role"])
+
+    if len(removed_invitations):
+        ids_to_delete = [ri.id for ri in removed_invitations]
+        models.Invitation.objects.filter(id__in=ids_to_delete).delete()
+
+    models.User.objects.bulk_update(update_users_active_status, ["is_active"])
+
+
+@admin.register(models.UserReconciliation)
+class UserReconciliationAdmin(admin.ModelAdmin):
+    list_display = ["id", "created_at", "status"]
+    actions = [process_reconciliation]
+
+
 @admin.register(models.Template)
 class TemplateAdmin(admin.ModelAdmin):
     """Template admin interface declaration."""

src/backend/core/migrations/0027_userreconciliationcsvimport_userreconciliation.py

@@ -0,0 +1,49 @@
+# Generated by Django 5.2.8 on 2025-12-01 15:01
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0026_comments'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UserReconciliationCsvImport',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('file', models.FileField(upload_to='imports/')),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('done', 'Done'), ('error', 'Error')], default='pending', max_length=20)),
+                ('logs', models.TextField(blank=True)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='UserReconciliation',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('active_email', models.EmailField(max_length=254, verbose_name='Active email address')),
+                ('inactive_email', models.EmailField(max_length=254, verbose_name='Email address to deactivate')),
+                ('active_email_checked', models.BooleanField(default=False)),
+                ('inactive_email_checked', models.BooleanField(default=False)),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('ready', 'Ready'), ('done', 'Done'), ('error', 'Error')], default='pending', max_length=20)),
+                ('logs', models.TextField(blank=True)),
+                ('active_user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='active_user', to=settings.AUTH_USER_MODEL)),
+                ('inactive_user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='inactive_user', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

src/backend/core/models.py

@@ -1,6 +1,7 @@
 """
 Declare and configure the models for the impress core application
 """
+
 # pylint: disable=too-many-lines
 
 import hashlib
@@ -265,6 +266,174 @@ def teams(self):
         return []
 
 
+class UserReconciliation(BaseModel):
+    """Model to run batch jobs to replace an active user by another one"""
+
+    active_email = models.EmailField(_("Active email address"))
+    inactive_email = models.EmailField(_("Email address to deactivate"))
+    active_email_checked = models.BooleanField(default=False)
+    inactive_email_checked = models.BooleanField(default=False)
+    active_user = models.ForeignKey(
+        User,
+        on_delete=models.CASCADE,
+        null=True,
+        blank=True,
+        related_name="active_user",
+    )
+    inactive_user = models.ForeignKey(
+        User,
+        on_delete=models.CASCADE,
+        null=True,
+        blank=True,
+        related_name="inactive_user",
+    )
+
+    status = models.CharField(
+        max_length=20,
+        choices=[
+            ("pending", _("Pending")),
+            ("ready", _("Ready")),
+            ("done", _("Done")),
+            ("error", _("Error")),
+        ],
+        default="pending",
+    )
+    logs = models.TextField(blank=True)
+
+    class Meta:
+        verbose_name = _("user reconciliation")
+        verbose_name_plural = _("user reconciliations")
+
+    def process_documentaccess_reconciliation(self):
+        """
+        Process the reconciliation by transferring document accesses from the inactive user
+        to the active user.
+        """
+        updated_accesses = []
+        removed_accesses = []
+        inactive_accesses = DocumentAccess.objects.filter(user=self.inactive_user)
+
+        # Check documents where the active user already has access
+        documents_with_both_users = inactive_accesses.values_list("document", flat=True)
+        existing_accesses = DocumentAccess.objects.filter(user=self.active_user).filter(
+            document__in=documents_with_both_users
+        )
+        existing_roles_per_doc = {
+            x: y for (x, y) in existing_accesses.values_list("document", "role")
+        }
+
+        for entry in inactive_accesses:
+            if entry.document_id in existing_roles_per_doc:
+                # Update role if needed
+                existing_role = existing_roles_per_doc[entry.document_id]
+                max_role = RoleChoices.max(entry.role, existing_role)
+                if existing_role != max_role:
+                    existing_access = existing_accesses.get(document=entry.document)
+                    existing_access.role = max_role
+                    updated_accesses.append(existing_access)
+                removed_accesses.append(entry)
+            else:
+                entry.user = self.active_user
+                updated_accesses.append(entry)
+
+        self.logs += f"""Requested update for {len(updated_accesses)} DocumentAccess items
+            and deletion for {len(removed_accesses)} DocumentAccess items.\n"""
+        self.status = "done"
+        self.save()
+
+        return updated_accesses, removed_accesses
+
+    def process_invitation_reconciliation(self):
+        """
+        Process the reconciliation by transferring invitations from the inactive user
+        to the active user.
+        """
+
+        updated_invitations = []
+        removed_invitations = []
+        inactive_invitations = Invitation.objects.filter(email=self.inactive_email)
+
+        # Check documents where the active user already has access
+        documents_with_both_users = inactive_invitations.values_list(
+            "document", flat=True
+        )
+        existing_accesses = Invitation.objects.filter(email=self.active_email).filter(
+            document__in=documents_with_both_users
+        )
+        existing_roles_per_doc = {
+            x: y for (x, y) in existing_accesses.values_list("document", "role")
+        }
+
+        for entry in inactive_invitations:
+            if entry.document_id in existing_roles_per_doc:
+                # Update role if needed
+                existing_role = existing_roles_per_doc[entry.document_id]
+                max_role = RoleChoices.max(entry.role, existing_role)
+                if existing_role != max_role:
+                    existing_access = existing_accesses.get(document=entry.document)
+                    existing_access.role = max_role
+                    updated_invitations.append(existing_access)
+                removed_invitations.append(entry)
+            else:
+                entry.user = self.active_user
+                updated_invitations.append(entry)
+
+        self.logs += f"""Requested update for {len(updated_invitations)} Invitation items
+            and deletion for {len(removed_invitations)} Invitation items.\n"""
+        self.status = "done"
+        self.save()
+
+        return updated_invitations, removed_invitations
+
+    def save(self, *args, **kwargs):
+        """
+        For pending queries, identify the actual users and send validation emails
+        """
+        if self.status == "pending":
+            self.active_user = User.objects.filter(email=self.active_email).first()
+            self.inactive_user = User.objects.filter(email=self.inactive_email).first()
+
+            if self.active_user and self.inactive_user:
+                email_subject = _("Account reconciliation request")
+                email_content = _(
+                    """
+                    Please click here.
+                    """
+                )
+                if not self.active_email_checked:
+                    self.active_user.email_user(email_subject, email_content)
+                if not self.inactive_email_checked:
+                    self.inactive_user.email_user(email_subject, email_content)
+                self.status = "ready"
+            else:
+                self.status = "error"
+                self.logs = "Error: Both active and inactive users need to exist."
+
+        super().save(*args, **kwargs)
+
+
+class UserReconciliationCsvImport(BaseModel):
+    """Model to import reconciliations requests from an external source
+    (eg, )"""
+
+    file = models.FileField(upload_to="imports/")
+    status = models.CharField(
+        max_length=20,
+        choices=[
+            ("pending", _("Pending")),
+            ("running", _("Running")),
+            ("done", _("Done")),
+            ("error", _("Error")),
+        ],
+        default="pending",
+    )
+    logs = models.TextField(blank=True)
+
+    class Meta:
+        verbose_name = _("user reconciliation CSV import")
+        verbose_name_plural = _("user reconciliation CSV imports")
+
+
 class BaseAccess(BaseModel):
     """Base model for accesses to handle resources."""
 

src/backend/core/tasks/user_reconciliation.py

@@ -0,0 +1,37 @@
+from impress.celery_app import app
+
+from core.models import UserReconciliation, UserReconciliationCsvImport
+
+import csv
+
+
+@app.task
+def user_reconciliation_csv_import_job(job_id):
+    # Imports the CSV file, breaks it into UserReconciliation items
+    job = UserReconciliationCsvImport.objects.get(id=job_id)
+    job.status = "running"
+    job.save()
+
+    try:
+        with job.file.open(mode="r") as f:
+            reader = csv.DictReader(f)
+            for row in reader:
+                active_email_checked = row["active_email_checked"] == "1"
+                inactive_email_checked = row["inactive_email_checked"] == "1"
+
+                rec_entry = UserReconciliation.objects.create(
+                    active_email=row["active_email"],
+                    inactive_email=row["inactive_email"],
+                    active_email_checked=active_email_checked,
+                    inactive_email_checked=inactive_email_checked,
+                    status="pending",
+                )
+                rec_entry.save()
+
+        job.status = "done"
+        job.logs = f"Import completed successfully. {reader.line_num} rows processed."
+    except Exception as e:
+        job.status = "error"
+        job.logs = str(e)
+    finally:
+        job.save()

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

@@ -0,0 +1,6 @@
+active_email,inactive_email,active_email_checked,inactive_email_checked,
+"[email protected]","[email protected]",0,0
+"[email protected]","[email protected]",0,1
+"[email protected]","[email protected]",1,0
+"[email protected]","[email protected]",1,1
+"[email protected]","[email protected]",1,1