ynerant/nk20-scripts
1
0
Fork 0
mirror of https://gitlab.crans.org/bde/nk20-scripts synced 2025-10-30 22:59:52 +01:00
Code Issues Releases Wiki Activity

use fixture as references and display status

Browse Source
This commit is contained in:
Pierre-antoine Comby
2020-04-10 18:13:26 +02:00
parent b9fdced3c2
commit 5a1b36181a
1 changed files with 86 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
management/commands/import_nk15.py
View File

@@ -7,8 +7,9 @@ import psycopg2 as pg
import psycopg2.extras as pge
from django.db import transaction
import json
import datetime
import collections
from datetime import timedelta
from django.core.exceptions import ValidationError
from django.utils.timezone import make_aware
@@ -29,52 +30,34 @@ TODO: import activite
TODO: import ...
"""
M_DURATION = timedelta(days=396)
M_START = timedelta(days=213)
M_END = timedelta(days=273)
M_DURATION = 396
M_START = datetime.date(2019,8,31)
M_END = datetime.date(2020,9,30)
MAP_IDBDE={
-4: 2, # Carte Bancaire
-3: 4, # Virement
-2: 1, # Especes
-1: 3, # Chèque
0: 5, # BDE
}
def update_line(n,N, content):
n = str(n)
N = str(N)
n.rjust(len(N))
content.ljust(41)
content = content[:40]
print(f"({n}/{N}) {content}", end="\r")
@transaction.atomic
def import_special(cur):
cur.execute("SELECT * FROM comptes WHERE idbde <0 ORDER BY idbde;")
map_idbde = dict()
for row in cur:
obj,created = NoteSpecial.objects.get_or_create(special_type = row["pseudo"],
balance = row["solde"],
is_active =True)
if created:
obj.save()
map_idbde[row["idbde"]] = obj.pk
cur.execute("SELECT * FROM comptes WHERE idbde=0;")
res = cur.fetchone()
clubBde, c = Club.objects.get_or_create(pk = 1,
name = "Bde",
email = "bureau.bde@lists.crans.org",
membership_duration = M_DURATION,
membership_start = M_START,
membership_end = M_END,
membership_fee = 5,
)
clubKfet, c = Club.objects.get_or_create(pk = 2,
name = "Kfet",
email = "tresorerie.bde@lists.crans.org",
membership_duration = M_DURATION,
membership_start = M_START,
membership_end = M_END,
membership_fee = 35,
)
clubBde.save()
clubKfet.save()
clubBde.note.solde=res["solde"]
map_idbde[0] = clubKfet.note.pk
return map_idbde
@transaction.atomic
def import_comptes(cur,map_idbde):
def import_comptes(cur):
cur.execute("SELECT * FROM comptes WHERE idbde > 0 ORDER BY idbde;")
pkclub = 3
for row in cur:
N = cur.rowcount
for idx, row in enumerate(cur):
update_line(idx,N,row["pseudo"])
if row["type"] == "personne":
#sanitize password
if row["passwd"] != "*|*":
@@ -88,21 +71,26 @@ def import_comptes(cur,map_idbde):
"first_name": row["nom"],
"last_name": row["prenom"],
"email": row["mail"],
"is_active" : False, # temporary
"is_active" : True, # temporary
}
user = User.objects.create(**obj_dict)
#sanitize duplicate aliases (nk12)
profile = user.profile
profile.phone_number = row['tel']
profile.address = row['adresse']
profile.paid = row['normalien']
profile.registration_valid = True
profile.email_confirmed = True
user.save()
profile.save()
# sanitize duplicate aliases (nk12)
except ValidationError as e:
if e.code == 'same_alias':
obj_dict["username"] = row["pseudo"]+str(row["idbde"])
user = User.objects.create(**obj_dict)
user.username = row["pseudo"]+str(row["idbde"])
user.save()
else:
raise(e)
# profile and note created via signal.
profile = user.profile
profile.phone_number = row["tel"]
profile.address = row["adresse"]
profile.paid = row["normalien"]
note = user.note
date = row.get("last_negatif",None)
@@ -110,6 +98,7 @@ def import_comptes(cur,map_idbde):
note.last_negative = make_aware(date)
note.balance = row["solde"]
obj_list =[user, profile, note]
note.save()
else: # club
obj_dict = {
"pk":pkclub,
@@ -118,23 +107,24 @@ def import_comptes(cur,map_idbde):
"membership_duration": M_DURATION,
"membership_start": M_START,
"membership_end": M_END,
"membership_fee": 0,
"membership_fee_paid": 0,
"membership_fee_unpaid":0,
}
club,c = Club.objects.get_or_create(**obj_dict)
pkclub +=1
note = club.note
note.balance = row["solde"]
obj_list = [club,note]
for obj in obj_list:
obj.save()
map_idbde[row["idbde"]] = note.pk
return map_idbde
club.save()
note.save()
MAP_IDBDE[row["idbde"]] = note.pk
@transaction.atomic
def import_boutons(cur,map_idbde):
def import_boutons(cur):
cur.execute("SELECT * FROM boutons;")
for row in cur:
N = cur.rowcount
for idx, row in enumerate(cur):
update_line(idx,N,row["label"])
cat, created = TemplateCategory.objects.get_or_create(name=row["categorie"])
if created:
cat.save()
@@ -142,7 +132,7 @@ def import_boutons(cur,map_idbde):
"pk": row["id"],
"name": row["label"],
"amount": row["montant"],
"destination_id": map_idbde[row["destinataire"]],
"destination_id": MAP_IDBDE[row["destinataire"]],
"category": cat,
"display" : row["affiche"],
"description": row["description"],
@@ -153,7 +143,7 @@ def import_boutons(cur,map_idbde):
except IntegrityError as e:
# button with the same name is not possible in NK20.
if "unique" in e.args[0]:
qs = Club.objects.filter(note__id=map_idbde[row["destinataire"]]).values('name')
qs = Club.objects.filter(note__id=MAP_IDBDE[row["destinataire"]]).values('name')
note_name = qs[0]["name"]
#rename button name
obj_dict["name"] ="{} {}".format(obj_dict["name"],note_name)
@@ -162,15 +152,16 @@ def import_boutons(cur,map_idbde):
raise(e)
button.save()
@transaction.atomic
def import_transaction(cur, map_idbde):
def import_transaction(cur):
cur.execute("SELECT * FROM transactions LEFT JOIN adhesions ON transactions.id = adhesions.idtransaction ORDER BY -id;")
for row in cur:
N = cur.rowcount
for idx, row in enumerate(cur):
update_line(idx,N,row["label"])
obj_dict = {
# "pk": row["id"],
"destination_id" : map_idbde[row["destinataire"]],
"source_id": map_idbde[row["emetteur"]],
"destination_id" : MAP_IDBDE[row["destinataire"]],
"source_id": MAP_IDBDE[row["emetteur"]],
"created_at":make_aware(row["date"]),
"amount":row["montant"],
"quantity":row["quantite"],
@@ -191,26 +182,26 @@ def import_transaction(cur, map_idbde):
print("adhesion not supported yet")
else:
print("other type not supported yet")
@transaction.atomic
def import_aliases(cur,map_idbde):
def import_aliases(cur):
cur.execute("SELECT * FROM aliases ORDER by id")
for row in cur:
N = cur.rowcount
for idx, row in enumerate(cur):
update_line(idx,N,row["alias"])
alias_name = row["alias"]
alias_name_good = (alias_name[:252]+'...') if len(alias_name) > 255 else alias_name
obj_dict = {
"note_id":map_idbde[row["idbde"]],
"note_id":MAP_IDBDE[row["idbde"]],
"name":alias_name_good,
"normalized_name":Alias.normalize(alias_name_good)
}
try:
with transaction.atomic():
alias, created = Alias.objects.get_or_create(**obj_dict)
print(alias)
except IntegrityError as e:
if "unique" in e.args[0]:
print("error, {}".format(alias))
continue
else:
raise(e)
@@ -226,37 +217,45 @@ class Command(BaseCommand):
return self.stdout.write(self.style.SUCCESS(to_print))
def add_arguments(self,parser):
parser.add_argument('-s', '--special', action = 'store_true', help="create Minimum instance (special note and clubs)")
parser.add_argument('-c', '--comptes', action = 'store_true', help="import accounts")
parser.add_argument('-b', '--boutons', action = 'store_true', help="import boutons")
parser.add_argument('-t', '--transactions', action = 'store_true',help="import transaction")
parser.add_argument('-a', '--aliases', action = 'store_true',help="import aliases")
parser.add_argument('-s', '--save', action='store', help="save mapping of idbde")
parser.add_argument('-m', '--map', action='store', help="import mapping of idbde")
parser.add_argument('-d', '--nk15db', action='store', default='nk15', help='NK15 database name')
parser.add_argument('-u', '--nk15user', action='store', default='nk15_user', help='NK15 database owner')
def handle(self, *args, **kwargs):
global MAP_IDBDE
nk15db, nk15user = kwargs['nk15db'], kwargs['nk15user']
#reset database.
call_command("flush")
self.print_success("flush nk20 database")
call_command("migrate")
call_command("loaddata","initial")
self.print_success("reset nk20 database")
# connecting to nk15 database
conn = pg.connect(database=nk15db,user=nk15user)
cur = conn.cursor(cursor_factory = pge.DictCursor)
if kwargs["special"]:
map_idbde = import_special(cur)
self.print_success("Minimal setup created")
if kwargs["comptes"]:
map_idbde = import_comptes(cur,map_idbde)
import_comptes(cur)
self.print_success("comptes table imported")
elif kwargs["map"]:
filename = kwargs["map"]
with open(filename,'w') as fp:
MAP_IDBDE = json.load(fp)
if kwargs["save"]:
filename = kwargs["save"]
with open(filename,'w') as fp:
json.dump(MAP_IDBDE,fp,sort_keys=True, indent=2)
# /!\ need a prober MAP_IDBDE
if kwargs["boutons"]:
import_boutons(cur,map_idbde)
import_boutons(cur)
self.print_success("boutons table imported")
if kwargs["transactions"]:
import_transaction(cur,map_idbde)
import_transaction(cur)
self.print_success("transaction imported")
if kwargs["aliases"]:
import_aliases(cur,map_idbde)
import_aliases(cur)
self.print_success("aliases imported")