companies: add logs to move_company_data & allow its use in the admin

itou/companies/admin.py

@@ -1,18 +1,23 @@
 from django.contrib import admin, messages
 from django.contrib.admin.utils import display_for_value
-from django.urls import reverse
+from django.core.exceptions import PermissionDenied
+from django.shortcuts import get_object_or_404, redirect
+from django.template.response import TemplateResponse
+from django.urls import path, reverse
 from django.utils import timezone
 from django.utils.html import format_html
 
 from itou.approvals.models import Approval
 from itou.common_apps.organizations.admin import HasMembersFilter, MembersInline, OrganizationAdmin
-from itou.companies import enums, models
-from itou.companies.admin_forms import CompanyAdminForm
+from itou.companies import enums, models, transfer
+from itou.companies.admin_forms import CompanyAdminForm, CompanyChooseFieldsToTransfer, SelectTargetCompanyForm
+from itou.siae_evaluations.models import EvaluatedSiae
 from itou.utils.admin import (
     ItouGISMixin,
     ItouModelAdmin,
     ItouTabularInline,
     PkSupportRemarkInline,
+    add_support_remark_to_obj,
     get_admin_view_link,
 )
 from itou.utils.apis.exceptions import GeocodingDataError
@@ -132,6 +137,7 @@ def export(self, request, queryset):
         )
 
     form = CompanyAdminForm
+    change_form_template = "admin/companies/change_company_form.html"
     list_display = ("pk", "siret", "kind", "name", "department", "geocoding_score", "member_count", "created_at")
     list_filter = (HasMembersFilter, "kind", "block_job_applications", "source", "department")
     raw_id_fields = ("created_by", "convention")
@@ -260,6 +266,146 @@ def approvals_list(self, obj):
         valid_count = Approval.objects.is_assigned_to(obj.id).valid().count()
         return format_html('<a href="{}">Liste des {} Pass IAE (dont {} valides)</a>', url, count, valid_count)
 
+    def get_urls(self):
+        urls = super().get_urls()
+        return [
+            path(
+                "transfer/<int:from_company_pk>",
+                self.admin_site.admin_view(self.transfer_view),
+                name="transfer_company_data",
+            ),
+            path(
+                "transfer/<int:from_company_pk>/<int:to_company_pk>",
+                self.admin_site.admin_view(self.transfer_view),
+                name="transfer_company_data",
+            ),
+        ] + urls
+
+    def transfer_view(self, request, from_company_pk, to_company_pk=None):
+        if not self.has_change_permission(request):
+            raise PermissionDenied
+
+        from_company = get_object_or_404(models.Company.objects, pk=from_company_pk)
+        to_company = get_object_or_404(models.Company, pk=to_company_pk) if to_company_pk is not None else None
+
+        transfer_data = {}
+        for transfer_field in transfer.TransferField:
+            spec = transfer.TRANSFER_SPECS[transfer_field]
+            if model_field := spec.get("model_field"):
+                from_data = [getattr(from_company, model_field.name)]
+                transfer_data[transfer_field] = {
+                    "data": from_data,
+                }
+            else:
+                from_data = transfer.get_transfer_queryset(from_company, to_company, spec)
+                transfer_data[transfer_field] = {
+                    "data": from_data,
+                }
+
+        if not to_company:
+            form = SelectTargetCompanyForm(
+                from_company=from_company,
+                admin_site=self.admin_site,
+                data=request.POST or None,
+            )
+            if request.POST and form.is_valid():
+                return redirect(
+                    reverse(
+                        "admin:transfer_company_data",
+                        kwargs={
+                            "from_company_pk": from_company.pk,
+                            "to_company_pk": form.cleaned_data["to_company"].pk,
+                        },
+                    )
+                )
+        else:
+            fields_choices = []
+            for transfer_field in transfer.TransferField:
+                spec = transfer.TRANSFER_SPECS[transfer_field]
+                if model_field := spec.get("model_field"):
+                    if transfer_data[transfer_field]["data"] == [getattr(to_company, model_field.name)]:
+                        transfer_data[transfer_field]["data"] = None
+                    else:
+                        fields_choices.append((transfer_field.value, transfer_field.label))
+                elif from_data := transfer_data[transfer_field]["data"]:
+                    plural = "s" if len(from_data) > 1 else ""
+                    fields_choices.append(
+                        (
+                            transfer_field.value,
+                            f"{transfer_field.label} ({len(from_data)} objet{plural} à transférer)",
+                        )
+                    )
+
+            siae_evaluations = EvaluatedSiae.objects.filter(siae=from_company).exists()
+            form = CompanyChooseFieldsToTransfer(
+                fields_choices=sorted(fields_choices, key=lambda field: field[1]),
+                siae_evaluations=siae_evaluations,
+                data=request.POST or None,
+            )
+            if request.POST and form.is_valid():
+                if siae_evaluations and not form.cleaned_data["ignore_siae_evaluations"]:
+                    messages.error(
+                        request,
+                        (
+                            f"Impossible de transférer les objets de l'entreprise ID={from_company.pk}: "
+                            "il y a un contrôle a posteriori lié. Vérifiez avec l'équipe support."
+                        ),
+                    )
+                else:
+                    try:
+                        reporter = transfer.transfer_company_data(
+                            from_company,
+                            to_company,
+                            form.cleaned_data["fields_to_transfer"],
+                            ignore_siae_evaluations=form.cleaned_data.get("ignore_siae_evaluations", False),
+                        )
+                    except transfer.TransferError as e:
+                        messages.error(request, e.args[0])
+                    else:
+                        summary_lines = [
+                            "-" * 20,
+                            f"Transfert du {timezone.now():%Y-%m-%d %H:%M:%S} effectué par {request.user} ",
+                            f"de l'entreprise {from_company.pk} vers {to_company.pk}:",
+                        ]
+                        for report_field, items in reporter.changes.items():
+                            if items:
+                                summary_lines.extend([f"- {report_field.label}:"] + [f"  * {item}" for item in items])
+                        summary_lines += ["-" * 20]
+                        summary_text = "\n".join(summary_lines)
+                        add_support_remark_to_obj(from_company, summary_text)
+                        add_support_remark_to_obj(to_company, summary_text)
+                        message = format_html(
+                            "Transfert effectué avec succès de l’entreprise {from_company} vers {to_company}.",
+                            from_company=from_company,
+                            to_company=to_company,
+                        )
+                        messages.info(request, message)
+
+                        return redirect(
+                            reverse(
+                                "admin:companies_company_change",
+                                kwargs={"object_id": from_company.pk},
+                            )
+                        )
+        title = f"Transfert des données de '{ from_company }' [{from_company.kind}]"
+        if to_company:
+            title += f" vers '{ to_company}' [{to_company.kind}]"
+        context = self.admin_site.each_context(request) | {
+            "media": self.media,
+            "opts": self.opts,
+            "form": form,
+            "from_company": from_company,
+            "to_company": to_company,
+            "transfer_data": transfer_data,
+            "title": title,
+        }
+
+        return TemplateResponse(
+            request,
+            "admin/companies/transfer_company.html",
+            context,
+        )
+
 
 @admin.register(models.JobDescription)
 class JobDescriptionAdmin(ItouModelAdmin):

itou/companies/admin_forms.py

@@ -1,6 +1,11 @@
+from collections import namedtuple
+
 from django import forms
+from django.contrib.admin import widgets
+from django.core.exceptions import ValidationError
 
 from itou.companies.models import Company
+from itou.utils.admin import ChooseFieldsToTransfer
 
 
 class CompanyAdminForm(forms.ModelForm):
@@ -17,3 +22,40 @@ class CompanyAdminForm(forms.ModelForm):
     class Meta:
         model = Company
         fields = "__all__"
+
+
+FakeField = namedtuple("FakeField", ("name",))
+
+
+class FakeRelForToCompanyRawIdWidget:
+    model = Company
+    limit_choices_to = {}
+
+    def get_related_field(self):
+        # This must return something that has the name of an existing field
+        return FakeField("id")
+
+
+class SelectTargetCompanyForm(forms.Form):
+    to_company = forms.ModelChoiceField(Company.objects.all(), required=True, label="Choisissez l’entreprise cible")
+
+    def __init__(self, *args, from_company, admin_site, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields["to_company"].widget = widgets.ForeignKeyRawIdWidget(FakeRelForToCompanyRawIdWidget(), admin_site)
+        self.from_company = from_company
+
+    def clean_to_company(self):
+        to_company = self.cleaned_data["to_company"]
+        if to_company.pk == self.from_company.pk:
+            raise ValidationError("L’entreprise cible doit être différente de celle d’origine")
+        return to_company
+
+
+class CompanyChooseFieldsToTransfer(ChooseFieldsToTransfer):
+    def __init__(self, *args, fields_choices, siae_evaluations, **kwargs):
+        super().__init__(*args, fields_choices=fields_choices, **kwargs)
+        if siae_evaluations:
+            self.fields["ignore_siae_evaluations"] = forms.BooleanField(
+                label="Ignorer la présence d'un contrôle a posteriori.",
+                required=True,
+            )