Draft of a gdpr export command

This commit is contained in:
Théophile Bastian 2021-11-23 00:55:51 +01:00
parent 1c880b265e
commit e9bb3999a0

View file

@ -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)