From e9bb3999a078dcf4b618d53c635cc95ced29e73a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9ophile=20Bastian?= Date: Tue, 23 Nov 2021 00:55:51 +0100 Subject: [PATCH] Draft of a gdpr export command --- kfet/management/commands/gdprexport.py | 208 +++++++++++++++++++++++++ 1 file changed, 208 insertions(+) create mode 100644 kfet/management/commands/gdprexport.py diff --git a/kfet/management/commands/gdprexport.py b/kfet/management/commands/gdprexport.py new file mode 100644 index 00000000..2cb7553e --- /dev/null +++ b/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)