feature: create a feature to allow easy anonymization of donors

- create a feature to allow the anonymization of all the "old" donations (>1/2 years old)
- allow anonymization of a donation in different contexts in the admin

Author: tudoramariei

.idea/runConfigurations/compilemessages.xml

@@ -1,12 +1,17 @@
 import logging
+from datetime import timedelta
 
+from django import forms
+from django.conf import settings
 from django.contrib import admin, messages
-from django.core.management import call_command
+from django.contrib.messages import DEFAULT_LEVELS
+from django.core.management import CommandError, call_command
 from django.core.validators import EMPTY_VALUES
 from django.db.models import QuerySet
 from django.http import HttpRequest
-from django.shortcuts import redirect
-from django.urls import reverse
+from django.shortcuts import redirect, render
+from django.urls import reverse, reverse_lazy
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from django.utils.translation import ngettext_lazy
 from unfold.admin import ModelAdmin
@@ -15,13 +20,78 @@
 
 from donations.models.donors import Donor
 from redirectioneaza.common.app_url import build_uri
+from redirectioneaza.common.async_wrapper import async_wrapper
 from redirectioneaza.common.messaging import extend_email_context, send_email
 from utils.common.admin import HasNgoFilter
+from utils.constants.time import YEAR
 
 logger = logging.getLogger(__name__)
 
-REMOVE_DONATIONS_SUCCESS_FLAG = "success"
-REMOVE_DONATIONS_FAILURE_FLAG = "failure"
+TASK_SUCCESS_FLAG = "SUCCESS"
+TASK_FAILURE_FLAG = "ERROR"
+TASK_SCHEDULED_FLAG = "SCHEDULED"
+
+
+class AnonymizeDonorForm(forms.Form):
+    confirmation = forms.MultipleChoiceField(
+        label=_("Please confirm"),
+        help_text=_(
+            "This action will PERMANENTLY anonymize the selected donor's personal data, "
+            "removing any personally identifiable information and the form itself. "
+            "THIS ACTION CANNOT BE UNDONE."
+        ),
+        choices=[("yes", _("Yes, I want to anonymize this donor"))],
+        widget=forms.CheckboxSelectMultiple,
+        required=True,
+    )
+
+
+def _print_task_result(process: str, task_results: dict[str, int], queryset_size: int) -> str:
+    message_parts: list[str] = [
+        ngettext_lazy(
+            singular="Out of %(queryset_size)d donor, the %(process)s results were:",
+            plural="Out of %(queryset_size)d donors, the %(process)s results were:",
+            number=queryset_size,
+        )
+        % {
+            "queryset_size": queryset_size,
+            "process": process.upper(),
+        }
+    ]
+
+    if task_results.get(TASK_SUCCESS_FLAG):
+        message_parts.append(
+            ngettext_lazy(
+                singular="%(success)d was processed",
+                plural="%(success)d were processed",
+                number=task_results[TASK_SUCCESS_FLAG],
+            )
+            % {"success": task_results[TASK_SUCCESS_FLAG]}
+        )
+
+    if task_results.get(TASK_FAILURE_FLAG):
+        message_parts.append(
+            ngettext_lazy(
+                singular="%(failure)d was not processed",
+                plural="%(failure)d were not processed",
+                number=task_results[TASK_FAILURE_FLAG],
+            )
+            % {"failure": task_results[TASK_FAILURE_FLAG]}
+        )
+
+    if task_results.get(TASK_SCHEDULED_FLAG):
+        message_parts.append(
+            ngettext_lazy(
+                singular="%(scheduled)d was scheduled for processing",
+                plural="%(scheduled)d were scheduled for processing",
+                number=task_results[TASK_SCHEDULED_FLAG],
+            )
+            % {"scheduled": task_results[TASK_SCHEDULED_FLAG]}
+        )
+
+    message_parts[-1] += "."
+
+    return ", ".join(message_parts)
 
 
 def soft_delete_donor(donor_pk: int):
@@ -43,10 +113,26 @@ def soft_delete_donor(donor_pk: int):
             html_template="emails/donor/removed-redirection/main.html",
             context=mail_context,
         )
-        return REMOVE_DONATIONS_SUCCESS_FLAG
+        return TASK_SUCCESS_FLAG
     except Exception as e:
         logger.error(f"Failed to delete donor {donor_pk}: {e}")
-        return REMOVE_DONATIONS_FAILURE_FLAG
+        return TASK_FAILURE_FLAG
+
+
+def anonymize_donor(donor: Donor):
+    logger.info(f"Anonymizing donor {donor.pk}")
+
+    try:
+        if settings.USER_ANONIMIZATION_METHOD == "async":
+            async_wrapper(donor.anonymize)
+            return TASK_SCHEDULED_FLAG
+
+        donor.anonymize()
+    except Exception as e:
+        logger.error(f"Failed to anonymize donor {donor.pk}: {e}")
+        return TASK_FAILURE_FLAG
+
+    return TASK_SUCCESS_FLAG
 
 
 class CommonNumericFilter(TextFilter):
@@ -140,8 +226,13 @@ class DonorAdmin(ModelAdmin):
         ),
     )
 
-    actions = ("remove_donations",)
-    actions_list = ("run_redirections_stats_generator", "run_redirections_stats_generator_force")
+    actions = ("remove_donations", "anonymize_donations")
+    actions_list = (
+        "run_redirections_stats_generator",
+        "run_redirections_stats_generator_force",
+        "anonymize_old_donations",
+    )
+    actions_detail = ("anonymize_donation",)
 
     def has_change_permission(self, request, obj=None):
         return False
@@ -155,37 +246,131 @@ def get_form_url(self, obj: Donor):
 
     @action(description=_("Remove donations and notify donors"), url_path="remove-donations")
     def remove_donations(self, request, queryset: QuerySet[Donor]):
-        task_results = {REMOVE_DONATIONS_SUCCESS_FLAG: 0, REMOVE_DONATIONS_FAILURE_FLAG: 0}
+        task_results = {TASK_SUCCESS_FLAG: 0, TASK_FAILURE_FLAG: 0}
 
         queryset_size = queryset.count()
         for donor in queryset:
             task_result = soft_delete_donor(donor.pk)
             task_results[task_result] += 1
 
-        part_1 = ngettext_lazy(
-            singular="Out of %(queryset_size)d donor",
-            plural="Out of %(queryset_size)d donors",
-            number=queryset_size,
-        ) % {"queryset_size": queryset_size}
+        message: str = _print_task_result(
+            process="removal",
+            task_results=task_results,
+            queryset_size=queryset_size,
+        )
+
+        self.message_user(request, message)
 
-        part_2 = ngettext_lazy(
-            singular="%(success)d was deleted",
-            plural="%(success)d were deleted",
-            number=task_results[REMOVE_DONATIONS_SUCCESS_FLAG],
-        ) % {"success": task_results[REMOVE_DONATIONS_SUCCESS_FLAG]}
+    @action(description=_("Anonymize public data of donations"), url_path="anonymize-donations")
+    def anonymize_donations(self, request, queryset: QuerySet[Donor]):
+        task_results = {TASK_SUCCESS_FLAG: 0, TASK_FAILURE_FLAG: 0, TASK_SCHEDULED_FLAG: 0}
 
-        part_3 = ngettext_lazy(
-            singular="%(failure)d was not deleted",
-            plural="%(failure)d were not deleted",
-            number=task_results[REMOVE_DONATIONS_FAILURE_FLAG],
-        ) % {"failure": task_results[REMOVE_DONATIONS_FAILURE_FLAG]}
+        queryset_size = queryset.count()
+        for donor in queryset:
+            task_result = anonymize_donor(donor)
+            task_results[task_result] += 1
 
-        self.message_user(request, ", ".join([part_1, part_2, part_3 + "."]))
+        message: str = _print_task_result(
+            process="anonymization",
+            task_results=task_results,
+            queryset_size=queryset_size,
+        )
+
+        self.message_user(request, message)
+
+    def _anonymize_donation(self, donor: Donor) -> dict[str, str]:
+        task_result = anonymize_donor(donor)
+
+        if task_result == TASK_SUCCESS_FLAG:
+            message = _("The donor was successfully anonymized.")
+        elif task_result == TASK_SCHEDULED_FLAG:
+            message = _("The donor anonymization was scheduled for processing.")
+        else:
+            message = _("Failed to anonymize the donor. Check the logs for more details.")
+
+        return {"status": task_result, "message": message}
+
+    @action(description=_("Anonymize public data of the donation"), url_path="anonymize-donation")
+    def anonymize_donation(self, request: HttpRequest, object_id: int):
+        donor = Donor.objects.get(pk=object_id)
+
+        if request.method == "POST":
+            form = AnonymizeDonorForm(request.POST)
+            if form.is_valid():
+                result: dict = self._anonymize_donation(donor)
+
+                self.message_user(request, result["message"], level=DEFAULT_LEVELS.get(result["status"], messages.INFO))
+
+                return redirect(reverse_lazy("admin:donations_donor_change", args=[donor.pk]))
+            else:
+                messages.error(request, _("Please confirm the anonymization by checking the box."))
+        else:
+            form = AnonymizeDonorForm()
+
+        return render(
+            request,
+            "admin/forms/action.html",
+            context={
+                "form": form,
+                "object": donor,
+                "title": _("Anonymize donor"),
+                **self.admin_site.each_context(request),
+            },
+        )
+
+        return redirect(reverse_lazy("admin:donations_donor_change", args=[donor.pk]))
 
     @action(description=_("Schedule redirections stats"), url_path="schedule-redirections-stats-generator")
-    def run_redirections_stats_generator(self, request, queryset: QuerySet[Donor]):
-        call_command("generate_redirections_stats")
+    def run_redirections_stats_generator(self, request):
+        try:
+            call_command("generate_redirections_stats")
+        except CommandError:
+            self.message_user(request, _("Error calling the command"))
+
+        self.message_user(request, _("Succesfully scheduled"))
 
     @action(description=_("Schedule redirections stats [FORCE]"), url_path="schedule-redirections-stats-generator-f")
-    def run_redirections_stats_generator_force(self, request, queryset: QuerySet[Donor]):
-        call_command("generate_redirections_stats", "--force")
+    def run_redirections_stats_generator_force(self, request):
+        try:
+            call_command("generate_redirections_stats", "--force")
+        except CommandError:
+            self.message_user(request, _("Error calling the command"))
+
+        self.message_user(request, _("Succesfully scheduled"))
+
+    def _anonymize_old_donations(self):
+        try:
+            now = timezone.now()
+            one_year_ago_date = now - timedelta(days=YEAR)
+            two_years_ago_date = now - timedelta(days=2 * YEAR)
+
+            base_donor_qs = Donor.objects.exclude(
+                l_name="",
+                f_name="",
+            )
+
+            one_year_donations_qs: QuerySet[Donor] = base_donor_qs.filter(
+                two_years=False,
+                date_created__lte=one_year_ago_date,
+            )
+            two_year_donations_qs: QuerySet[Donor] = base_donor_qs.filter(
+                date_created__lte=two_years_ago_date,
+            )
+
+            for donor in two_year_donations_qs:
+                anonymize_donor(donor)
+            for donor in one_year_donations_qs:
+                anonymize_donor(donor)
+
+        except CommandError:
+            logger.error("Error calling the anonymize_old_donations command")
+            raise
+
+    @action(description=_("Anonymize old donations"), url_path="anonymize-old-donations")
+    def anonymize_old_donations(self, request):
+        try:
+            self._anonymize_old_donations()
+        except CommandError:
+            self.message_user(request, _("Error calling the command"))
+
+        self.message_user(request, _("Succesfully scheduled"))