Draft of a gdpr export command

kfet/management/commands/gdprexport.py

@@ -0,0 +1,208 @@
+import sys
+import typing as ty
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db.models import Prefetch, Q
+
+from kfet.models import (
+    Account,
+    AccountNegative,
+    Operation,
+    OperationGroup,
+    Transfer,
+    TransferGroup,
+)
+
+
+class Command(BaseCommand):
+    help = "Exports all personal user data for a given trigramme"
+
+    def add_arguments(self, parser):
+        parser.add_argument("trigramme", help="the trigramme to)export")
+        parser.add_argument(
+            "-o",
+            "--output",
+            default="-",
+            help="File in which the results should be written (defaults to stdout)",
+        )
+
+    def _do_handle_history(self, account: Account, stream: ty.TextIO):
+        transfer_queryset_prefetch = Transfer.objects.select_related(
+            "from_acc", "to_acc", "canceled_by"
+        )
+
+        # Le check sur les comptes est dans le prefetch pour les transferts
+        transfer_queryset_prefetch = transfer_queryset_prefetch.filter(
+            Q(from_acc=account) | Q(to_acc=account)
+        )
+
+        transfer_prefetch = Prefetch(
+            "transfers",
+            queryset=transfer_queryset_prefetch,
+            to_attr="filtered_transfers",
+        )
+
+        # Construction de la requête (sur les opérations) pour le prefetch
+        ope_queryset_prefetch = Operation.objects.select_related(
+            "article", "canceled_by", "addcost_for"
+        )
+        ope_prefetch = Prefetch("opes", queryset=ope_queryset_prefetch)
+
+        # Construction de la requête principale
+        opegroups = (
+            OperationGroup.objects.prefetch_related(ope_prefetch)
+            .select_related("on_acc", "valid_by")
+            .filter(on_acc=account)
+            .order_by("at")
+        )
+        transfergroups = (
+            TransferGroup.objects.prefetch_related(transfer_prefetch)
+            .select_related("valid_by")
+            .order_by("at")
+        )
+
+        # Construction de la réponse
+        history_groups = []
+        for opegroup in opegroups:
+            opegroup_dict = {
+                "type": "operation",
+                "id": opegroup.id,
+                "amount": opegroup.amount,
+                "at": opegroup.at,
+                "checkout_id": opegroup.checkout_id,
+                "is_cof": opegroup.is_cof,
+                "comment": opegroup.comment,
+                "entries": [],
+                "on_acc__trigramme": opegroup.on_acc
+                and opegroup.on_acc.trigramme
+                or None,
+            }
+            for ope in opegroup.opes.all():
+                ope_dict = {
+                    "id": ope.id,
+                    "type": ope.type,
+                    "amount": ope.amount,
+                    "article_nb": ope.article_nb,
+                    "addcost_amount": ope.addcost_amount,
+                    "canceled_at": ope.canceled_at,
+                    "article__name": ope.article and ope.article.name or None,
+                    "addcost_for__trigramme": ope.addcost_for
+                    and ope.addcost_for.trigramme
+                    or None,
+                }
+                opegroup_dict["entries"].append(ope_dict)
+            history_groups.append(opegroup_dict)
+        for transfergroup in transfergroups:
+            if transfergroup.filtered_transfers:
+                transfergroup_dict = {
+                    "type": "transfer",
+                    "id": transfergroup.id,
+                    "at": transfergroup.at,
+                    "comment": transfergroup.comment,
+                    "entries": [],
+                }
+
+                for transfer in transfergroup.filtered_transfers:
+                    transfer_dict = {
+                        "id": transfer.id,
+                        "amount": transfer.amount,
+                        "canceled_at": transfer.canceled_at,
+                        "from_acc": transfer.from_acc.trigramme,
+                        "to_acc": transfer.to_acc.trigramme,
+                    }
+                    transfergroup_dict["entries"].append(transfer_dict)
+                history_groups.append(transfergroup_dict)
+
+        history_groups.sort(key=lambda group: group["at"])
+
+        stream.write("## Historique\n\n")
+        for group in history_groups:
+            if group["type"] == "operation":
+                stream.write(
+                    f"* {group['at']} : opération (valeur {group['amount']} EUR)\n"
+                )
+                if group["comment"]:
+                    stream.write(f"  Commentaire : {group['comment']}\n")
+                for op in group["entries"]:
+                    stream.write(f"  - {op['type']} : ")
+                    if op["type"] == "purchase":
+                        stream.write(f"{op['article_nb']}x {op['article__name']}")
+                    stream.write(f" ({op['amount']} EUR)")
+                    if op["canceled_at"]:
+                        stream.write(" [ANNULÉ]")
+                    stream.write("\n")
+            elif group["type"] == "transfer":
+                stream.write(f"* {group['at']} : transferts d'UKF\n")
+                if group["comment"]:
+                    stream.write(f"  Commentaire : {group['comment']}\n")
+                for op in group["entries"]:
+                    stream.write(
+                        f"  - {op['from_acc']} -> {op['to_acc']} : "
+                        f"{op['amount']} EUR"
+                    )
+                    if op["canceled_at"]:
+                        stream.wrte(" [ANNULÉ]")
+                    stream.write("\n")
+
+    def _do_handle(self, trigramme: str, stream: ty.TextIO):
+        """Actually write the personal user data to `stream`"""
+
+        try:
+            kfet_account = Account.objects.get(trigramme=trigramme)
+        except Account.DoesNotExist as exn:
+            raise CommandError("No such trigramme: {}", trigramme) from exn
+
+        try:
+            cof_account = kfet_account.cofprofile
+        except Account.DoesNotExist as exn:
+            raise CommandError(
+                "No cof profile with PK={}", kfet_account.cofprofile
+            ) from exn
+
+        stream.write(
+            "# Export de données personnelles pour le trigramme {}\n\n".format(
+                trigramme
+            )
+        )
+
+        # Generic data
+        stream.write("## Informations générales (COF)\n\n")
+        for field in cof_account._meta.get_fields():
+            fname = field.name
+            if not hasattr(field, "attname"):
+                continue
+            if hasattr(field, "verbose_name"):
+                fname = field.verbose_name
+            stream.write(f"{fname}: {getattr(cof_account, field.attname)}\n")
+        stream.write("\n")
+
+        stream.write("## Informations générales (K-Fêt)\n\n")
+        stream.write(
+            f"Balance : {kfet_account.balance}\n"
+            f"Compte figé : {kfet_account.is_frozen}\n"
+            f"Créé le : {kfet_account.created_at}\n"
+            f"Promotion : {kfet_account.promo}\n"
+            f"Pseudo : {kfet_account.nickname}\n"
+        )
+
+        # Negative
+        try:
+            neg_profile = kfet_account.negative
+            stream.write("Balance négative : oui\n")
+            stream.write(f"Depuis : f{neg_profile.start}\n")
+            stream.write(f"Jusqu'à : f{neg_profile.end}\n")
+            stream.write(f"Rappel le : f{neg_profile.last_rappel}\n\n")
+        except AccountNegative.DoesNotExist:
+            stream.write("Balance négative : non\n\n")
+
+        self._do_handle_history(kfet_account, stream)
+
+    def handle(self, *args, **options):
+        trigramme = options["trigramme"]
+        out_stream_path = options["output"]
+
+        if out_stream_path == "-":
+            self._do_handle(trigramme, sys.stdout)
+        else:
+            with open(out_stream_path, "w") as handle:
+                self._do_handle(trigramme, handle)