# Copyright (C) 2028-2024 by BDE ENS Paris-Saclay
# SPDX-License-Identifier: GPL-3.0-or-later
import json
from argparse import ArgumentParser
from django.core.management import BaseCommand
from django.db.models import Q
from note.models import Note, Transaction
from member.models import User, Club, Membership
from activity.models import Activity, Entry
from wei.models import WEIClub
from ...models import Bde, Wrapped
class Command(BaseCommand):
help = "Generate wrapper for the annual BDE change"
def add_arguments(self, parser: ArgumentParser):
parser.add_argument(
'-b', '--bde',
type=str,
required=False,
help="A list of BDE name, BDE1,BDE2,... (a BDE name cannot have ',')",
dest='bde',
)
parser.add_argument(
'-i', '--id',
type=str,
required=False,
help="A list of BDE id, id1,id2,...",
dest='bde_id',
)
parser.add_argument(
'-u', '--users',
type=str,
required=False,
help="""User will have their(s) wrapped generated,
all = all users
adh = all users who have a valid memberships to BDE during the BDE considered
supersuser = all superusers
custom user1,user2,... = a list of username,
custom_id id1,id2,... = a list of user id""",
dest='user',
)
parser.add_argument(
'-c', '--club',
type=str,
required=False,
help="""Club will have their(s) wrapped generated,
all = all clubs,
active = all clubs with at least one transaction during the BDE mandate considered,
custom club1,club2,... = a list of club name,
custom_id id1,id2,... = a list of club id""",
dest='club',
)
parser.add_argument(
'-f', '--force-change',
required=False,
action='store_true',
help="if wrapped already exist change data_json",
dest='change',
)
parser.add_argument(
'-n', '--no-creation',
required=False,
action='store_false',
help="if wrapped don't already exist, don't generate it",
dest='create',
)
def handle(self, *args, **options):
# useful string for output
red = '\033[31;1m'
yellow = '\033[33;1m'
green = '\033[32;1m'
abort = red + 'ABORT'
warning = yellow + 'WARNING'
success = green + 'SUCCESS'
# Traitement des paramètres
verb = options['verbosity']
bde = []
if options['bde']:
bde_list = options['bde'].split(',')
bde = [Bde.objects.get(name=bde_name) for bde_name in bde_list]
if options['bde_id']:
if bde:
if verb >= 1:
print(warning)
print(yellow + 'You already defined bde with their name !')
if verb >= 0:
print(abort)
return
bde_id = options['bde_id'].split(',')
bde = [Bde.objects.get(pk=i) for i in bde_id]
user = []
if options['user']:
if options['user'] == 'all':
user = ['all', None]
elif options['user'] == 'adh':
user = ['adh', None]
elif options['user'] == 'superuser':
user = ['superuser', None]
elif options['user'].split(' ')[0] == 'custom':
user_list = options['user'].split(' ')[1].split(',')
user = ['custom', [User.objects.get(username=u) for u in user_list]]
elif options['user'].split(' ')[0] == 'custom_id':
user_id = options['user'].split(' ')[1].split(',')
user = ['custom_id', [User.objects.get(pk=u) for u in user_id]]
else:
if verb >= 1:
print(warning)
print(yellow + 'You user option is not recognized')
if verb >= 0:
print(abort)
return
club = []
if options['club']:
if options['club'] == 'all':
club = ['all', None]
elif options['club'] == 'active':
club = ['active', None]
elif options['club'].split(' ')[0] == 'custom':
club_list = options['club'].split(' ')[1].split(',')
club = ['custom', [Club.objects.get(name=club_name) for club_name in club_list]]
elif options['club'].split(' ')[0] == 'custom_id':
club_id = options['club'].split(' ')[1].split(',')
club = ['custom_id', [Club.objects.get(pk=c) for c in club_id]]
else:
if verb >= 1:
print(warning)
print(yellow + 'You club option is not recognized')
if verb >= 0:
print(abort)
return
change = options['change']
create = options['create']
# check if parameters are sufficient for generate wrapped with the desired option
if not bde:
if verb >= 1:
print(warning)
print(yellow + 'You have not selectionned a BDE !')
if verb >= 0:
print(abort)
return
if not (user or club):
if verb >= 1:
print(warning)
print(yellow + 'No club or user selected !')
if verb >= 0:
print(abort)
return
if verb >= 3:
print('\033[1mOptions:\033[m')
bde_str = ''
for b in bde:
bde_str += str(b)
print('BDE: ' + bde_str)
if user:
print('User: ' + user[0])
if club:
print('Club: ' + club[0])
print('change: ' + str(change))
print('create: ' + str(create))
print('')
if not (change or create):
if verb >= 1:
print(warning)
print(yellow + 'change and create is set to false, none wrapped will be created')
if verb >= 0:
print(abort)
return
if verb >= 1 and change:
print(warning)
print(yellow + 'change is set to true, some wrapped may be replaced !')
if verb >= 1 and not create:
print(warning)
print(yellow + 'create is set to false, wrapped will not be created !')
if verb >= 3 or change or not create:
a = str(input('\033[mContinue ? (y/n) ')).lower()
if a in ['n', 'no', 'non', '0']:
if verb >= 0:
print(abort)
return
note = self.convert_to_note(change, create, bde=bde, user=user, club=club, verb=verb)
if verb >= 1:
print("\033[32mUser and/or Club given has successfully convert in their note\033[m")
global_data = self.global_data(bde, verb=verb)
if verb >= 1:
print("\033[32mGlobal data has been successfully generated\033[m")
unique_data = self.unique_data(bde, note, global_data=global_data, verb=verb)
if verb >= 1:
print("\033[32mUnique data has been successfully generated\033[m")
self.make_wrapped(unique_data, note, bde, change, create, verb=verb)
if verb >= 1:
print(green + "The wrapped has been generated !")
if verb >= 0:
print(success)
return
def convert_to_note(self, change, create, bde=None, user=None, club=None, verb=1):
notes = []
for b in bde:
note_for_bde = Note.objects.filter(pk__lte=-1)
if user:
if 'custom' in user[0]:
for u in user[1]:
query = Q(noteuser__user=u)
note_for_bde |= Note.objects.filter(query)
elif user[0] == 'all':
query = Q(noteuser__user__pk__gte=-1)
note_for_bde |= Note.objects.filter(query)
elif user[0] == 'adh':
m = Membership.objects.filter(club=1,
date_start__lt=b.date_end,
date_end__gt=b.date_start,
).distinct('user')
for membership in m:
note_for_bde |= Note.objects.filter(noteuser__user=membership.user)
elif user[0] == 'superuser':
query |= Q(noteuser__user__is_superuser=True)
note_for_bde |= Note.objects.filter(query)
if club:
if 'custom' in club[0]:
for c in club[1]:
query = Q(noteclub__club=c)
note_for_bde |= Note.objects.filter(query)
elif club[0] == 'all':
query = Q(noteclub__club__pk__gte=-1)
note_for_bde |= Note.objects.filter(query)
elif club[0] == 'active':
nc = Note.objects.filter(noteclub__club__pk__gte=-1)
for noteclub in nc:
if Transaction.objects.filter(
Q(created_at__gte=b.date_start,
created_at__lte=b.date_end) & (Q(source=noteclub) | Q(destination=noteclub))):
note_for_bde |= Note.objects.filter(pk=noteclub.pk)
note_for_bde = self.filter_note(b, note_for_bde, change, create, verb=verb)
notes.append(note_for_bde)
if verb >= 2:
print("\033[m{nb} note selectionned for bde {bde}".format(nb=len(note_for_bde), bde=b.name))
return notes
def global_data(self, bde, verb=1):
data = {}
for b in bde:
if b.name == 'Rave Part[list]':
if verb >= 2:
print("Begin to make global data")
if verb >= 3:
print('nb_transaction')
# nb total de transactions
data['nb_transaction'] = Transaction.objects.filter(
created_at__gte=b.date_start,
created_at__lte=b.date_end,
valid=True).count()
if verb >= 3:
print('nb_vieux_con')
# nb total de vielleux con·ne·s derrière le bar
button_id = [2884, 2585]
transactions = Transaction.objects.filter(
created_at__gte=b.date_start,
created_at__lte=b.date_end,
valid=True,
recurrenttransaction__template__pk__in=button_id)
q = 0
for t in transactions:
q += t.quantity
data['nb_vieux_con'] = q
if verb >= 3:
print('nb_soiree')
# nb total de soirée
a_type_id = [1, 2, 4, 5, 7, 10]
data['nb_soiree'] = Activity.objects.filter(
date_end__gte=b.date_start,
date_start__lte=b.date_end,
valid=True,
activity_type__pk__in=a_type_id).count()
if verb >= 3:
print('pots, nb_entree_pot')
# nb d'entrée totale aux pots
pot_id = [1, 4, 10]
pots = Activity.objects.filter(
date_end__gte=b.date_start,
date_start__lte=b.date_end,
valid=True,
activity_type__pk__in=pot_id)
data['pots'] = pots # utile dans unique_data
data['nb_entree_pot'] = 0
for pot in pots:
data['nb_entree_pot'] += Entry.objects.filter(activity=pot).count()
if verb >= 3:
print('top3_buttons')
# top 3 des boutons les plus cliqués
transactions = Transaction.objects.filter(
created_at__gte=b.date_start,
created_at__lte=b.date_end,
valid=True,
amount__gt=0,
recurrenttransaction__template__pk__gte=-1)
d = {}
for t in transactions:
if t.recurrenttransaction.template.name in d:
d[t.recurrenttransaction.template.name] += t.quantity
else:
d[t.recurrenttransaction.template.name] = t.quantity
data['top3_buttons'] = list(sorted(d.items(), key=lambda item: item[1], reverse=True))[:3]
if verb >= 3:
print('class_conso_all')
# le classement des plus gros consommateurs (BDE + club)
transactions = Transaction.objects.filter(
created_at__gte=b.date_start,
created_at__lte=b.date_end,
valid=True,
source__noteuser__user__pk__gte=-1,
destination__noteclub__club__pk__gte=-1)
d = {}
for t in transactions:
if t.source in d:
d[t.source] += t.total
else:
d[t.source] = t.total
data['class_conso_all'] = dict(sorted(d.items(), key=lambda item: item[1], reverse=True))
if verb >= 3:
print('class_conso_bde')
# le classement des plus gros consommateurs BDE
transactions = Transaction.objects.filter(
created_at__gte=b.date_start,
created_at__lte=b.date_end,
valid=True,
source__noteuser__user__pk__gte=-1,
destination=5)
d = {}
for t in transactions:
if t.source in d:
d[t.source] += t.total
else:
d[t.source] = t.total
data['class_conso_bde'] = dict(sorted(d.items(), key=lambda item: item[1], reverse=True))
else:
# make your wrapped or reuse previous wrapped
raise NotImplementedError("The BDE: {bde_name} has not personalized wrapped, make it !"
.format(bde_name=b.name))
return data
def unique_data(self, bde, note, global_data=None, verb=1):
data = []
for i in range(len(bde)):
data_bde = []
if bde[i].name == 'Rave Part[list]':
if verb >= 3:
total = len(note[i])
current = 0
print('Make {nb} data for wrapped sponsored by {bde}'
.format(nb=total, bde=bde[i].name))
for n in note[i]:
d = {}
if 'user' in n.__dir__():
# première conso du mandat
transactions = Transaction.objects.filter(
valid=True,
recurrenttransaction__template__id__gte=-1,
created_at__gte=bde[i].date_start,
created_at__lte=bde[i].date_end,
source=n,
destination=5).order_by('created_at')
if transactions:
d['first_conso'] = transactions[0].template.name
else:
d['first_conso'] = ''
# Wei + bus
wei = WEIClub.objects.filter(
date_start__lte=bde[i].date_end,
date_end__gte=bde[i].date_start)
if not wei:
d['wei'] = ''
d['bus'] = ''
else:
w = wei[0]
memberships = Membership.objects.filter(club=w, user=n.user)
if not memberships:
d['wei'] = ''
d['bus'] = ''
else:
alias = []
for a in w.note.alias.iterator():
alias.append(str(a))
d['wei'] = alias[-1]
d['bus'] = memberships[0].weimembership.bus.name
# top3 conso
transactions = Transaction.objects.filter(
valid=True,
created_at__gte=bde[i].date_start,
created_at__lte=bde[i].date_end,
source=n,
amount__gt=0,
recurrenttransaction__template__id__gte=-1)
dt = {}
dc = {}
for t in transactions:
if t.template.name in dt:
dt[t.template.name] += t.quantity
else:
dt[t.template.name] = t.quantity
if t.template.category.name in dc:
dc[t.template.category.name] += t.quantity
else:
dc[t.template.category.name] = t.quantity
d['top3_conso'] = list(sorted(dt.items(), key=lambda item: item[1], reverse=True))[:3]
# catégorie de bouton préférée
if dc:
d['top_category'] = list(sorted(dc.items(), key=lambda item: item[1], reverse=True))[0][0]
else:
d['top_category'] = ''
# nombre de pot, et nombre d'entrée pot
pots = global_data['pots']
d['nb_pots'] = pots.count()
p = 0
for pot in pots:
if Entry.objects.filter(activity=pot, note=n):
p += 1
d['nb_pot_entry'] = p
# ton nombre de rechargement
d['nb_rechargement'] = Transaction.objects.filter(
valid=True,
created_at__gte=bde[i].date_start,
created_at__lte=bde[i].date_end,
destination=n,
source__pk__in=[1, 2, 3, 4]).count()
# ajout info globale spécifique user
# classement et montant conso all
d['class_part_all'] = len(global_data['class_conso_all'])
if n in global_data['class_conso_all']:
d['class_conso_all'] = list(global_data['class_conso_all']).index(n) + 1
d['amount_conso_all'] = global_data['class_conso_all'][n] / 100
else:
d['class_conso_all'] = 0
d['amount_conso_all'] = 0
# classement et montant conso bde
d['class_part_bde'] = len(global_data['class_conso_bde'])
if n in global_data['class_conso_bde']:
d['class_conso_bde'] = list(global_data['class_conso_bde']).index(n) + 1
d['amount_conso_bde'] = global_data['class_conso_bde'][n] / 100
else:
d['class_conso_bde'] = 0
d['amount_conso_bde'] = 0
if 'club' in n.__dir__():
# plus gros consommateur
transactions = Transaction.objects.filter(
valid=True,
created_at__lte=bde[i].date_end,
created_at__gte=bde[i].date_start,
destination=n,
source__noteuser__user__pk__gte=-1)
dt = {}
for t in transactions:
if t.source.user.username in dt:
dt[t.source.user.username] += t.total
else:
dt[t.source.user.username] = t.total
if dt:
d['big_consumer'] = list(sorted(dt.items(), key=lambda item: item[1], reverse=True))[0]
d['big_consumer'] = (d['big_consumer'][0], d['big_consumer'][1] / 100)
else:
d['big_consumer'] = ''
# plus gros créancier
transactions = Transaction.objects.filter(
valid=True,
created_at__lte=bde[i].date_end,
created_at__gte=bde[i].date_start,
source=n,
destination__noteuser__user__pk__gte=-1)
dt = {}
for t in transactions:
if t.destination.user.username in dt:
dt[t.destination.user.username] += t.total
else:
dt[t.destination.user.username] = t.total
if dt:
d['big_creancier'] = list(sorted(dt.items(), key=lambda item: item[1], reverse=True))[0]
d['big_creancier'] = (d['big_creancier'][0], d['big_creancier'][1] / 100)
else:
d['big_creancier'] = ''
# nb de soirée organisée
d['nb_soiree_orga'] = Activity.objects.filter(
valid=True,
date_start__lte=bde[i].date_end,
date_end__gte=bde[i].date_start,
organizer=n.club).count()
# nb de membres cumulé
d['nb_member'] = Membership.objects.filter(
date_start__lte=bde[i].date_end,
date_end__gte=bde[i].date_start,
club=n.club).distinct('user').count()
# ajout info globale
# top3 button
d['glob_top3_conso'] = global_data['top3_buttons']
# nb entree pot
d['glob_nb_entree_pot'] = global_data['nb_entree_pot']
# nb soiree
d['glob_nb_soiree'] = global_data['nb_soiree']
# nb vieux con
d['glob_nb_vieux_con'] = global_data['nb_vieux_con']
# nb transaction
d['glob_nb_transaction'] = global_data['nb_transaction']
data_bde.append(json.dumps(d))
if verb >= 3:
current += 1
print('\033[2K' + '({c}/{t})'.format(c=current, t=total) + '\033[1A')
else:
# make your wrapped or reuse previous wrapped
raise NotImplementedError("The BDE: {bde_name} has not personalized wrapped, make it !"
.format(bde_name=bde[i].name))
data.append(data_bde)
return data
def make_wrapped(self, unique_data, note, bde, change, create, verb=1):
if verb >= 3:
current = 0
total = 0
for n in note:
total += len(n)
print('\033[mMake {nb} wrapped'.format(nb=total))
for i in range(len(bde)):
for j in range(len(note[i])):
if create and not Wrapped.objects.filter(bde=bde[i], note=note[i][j]):
Wrapped(bde=bde[i],
note=note[i][j],
data_json=unique_data[i][j],
public=False,
generated=True).save()
elif change:
w = Wrapped.objects.get(bde=bde[i], note=note[i][j])
w.data_json = unique_data[i][j]
w.save()
if verb >= 3:
current += 1
print('\033[2K' + '({c}/{t})'.format(c=current, t=total) + '\033[1A')
return
def filter_note(self, bde, note, change, create, verb=1):
if change and create:
return list(note)
if change and not create:
note_new = []
for n in note:
if Wrapped.objects.filter(bde=bde, note=n):
note_new.append(n)
return note_new
if not change and create:
note_new = []
for n in note:
if not Wrapped.objects.filter(bde=bde, note=n):
note_new.append(n)
return note_new