mirror of
https://gitlab.com/animath/si/plateforme.git
synced 2024-11-30 07:33:09 +00:00
273 lines
10 KiB
Python
273 lines
10 KiB
Python
from django.core.management import BaseCommand, CommandError
|
|
from django.db import transaction
|
|
|
|
from member.models import TFJMUser, Document, Solution, Synthesis, Authorization, MotivationLetter
|
|
from tournament.models import Team, Tournament
|
|
|
|
|
|
class Command(BaseCommand):
|
|
def add_arguments(self, parser):
|
|
parser.add_argument('--tournaments', '-t', action="store_true", help="Import tournaments")
|
|
parser.add_argument('--teams', '-T', action="store_true", help="Import teams")
|
|
parser.add_argument('--users', '-u', action="store_true", help="Import users")
|
|
parser.add_argument('--documents', '-d', action="store_true", help="Import all documents")
|
|
|
|
def handle(self, *args, **options):
|
|
if "tournaments" in options:
|
|
self.import_tournaments()
|
|
|
|
if "teams" in options:
|
|
self.import_teams()
|
|
|
|
if "users" in options:
|
|
self.import_users()
|
|
|
|
if "documents" in options:
|
|
self.import_documents()
|
|
|
|
@transaction.atomic
|
|
def import_tournaments(self):
|
|
print("Importing tournaments...")
|
|
with open("import_olddb/tournaments.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if Tournament.objects.filter(pk=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"id": args[0],
|
|
"name": args[1],
|
|
"size": args[2],
|
|
"place": args[3],
|
|
"price": args[4],
|
|
"description": args[5],
|
|
"date_start": args[6],
|
|
"date_end": args[7],
|
|
"date_inscription": args[8],
|
|
"date_solutions": args[9],
|
|
"date_syntheses": args[10],
|
|
"final": args[11],
|
|
"year": args[12],
|
|
}
|
|
with transaction.atomic():
|
|
Tournament.objects.create(**obj_dict)
|
|
print(self.style.SUCCESS("Tournaments imported"))
|
|
|
|
@staticmethod
|
|
def validation_status(status):
|
|
if status == "NOT_READY":
|
|
return "0invalid"
|
|
elif status == "WAITING":
|
|
return "1waiting"
|
|
elif status == "VALIDATED":
|
|
return "2valid"
|
|
else:
|
|
raise CommandError("Unknown status: {}".format(status))
|
|
|
|
@transaction.atomic
|
|
def import_teams(self):
|
|
self.stdout.write("Importing teams...")
|
|
with open("import_olddb/teams.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if Team.objects.filter(pk=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"id": args[0],
|
|
"name": args[1],
|
|
"trigram": args[2],
|
|
"tournament": Tournament.objects.get(pk=args[3]),
|
|
"inscription_date": args[13],
|
|
"validation_status": Command.validation_status(args[14]),
|
|
"selected_for_final": args[15],
|
|
"access_code": args[16],
|
|
"year": args[17],
|
|
}
|
|
with transaction.atomic():
|
|
Team.objects.create(**obj_dict)
|
|
print(self.style.SUCCESS("Teams imported"))
|
|
|
|
@staticmethod
|
|
def role(role):
|
|
if role == "ADMIN":
|
|
return "0admin"
|
|
elif role == "ORGANIZER":
|
|
return "1volunteer"
|
|
elif role == "ENCADRANT":
|
|
return "2coach"
|
|
elif role == "PARTICIPANT":
|
|
return "3participant"
|
|
else:
|
|
raise CommandError("Unknown role: {}".format(role))
|
|
|
|
@transaction.atomic
|
|
def import_users(self):
|
|
self.stdout.write("Importing users...")
|
|
with open("import_olddb/users.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if TFJMUser.objects.filter(pk=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"id": args[0],
|
|
"email": args[1],
|
|
"username": args[1],
|
|
"password": "bcrypt$" + args[2],
|
|
"last_name": args[3],
|
|
"first_name": args[4],
|
|
"birth_date": args[5],
|
|
"gender": "male" if args[6] == "M" else "female",
|
|
"address": args[7],
|
|
"postal_code": args[8],
|
|
"city": args[9],
|
|
"country": args[10],
|
|
"phone_number": args[11],
|
|
"school": args[12],
|
|
"student_class": args[13],
|
|
"responsible_name": args[14],
|
|
"responsible_phone": args[15],
|
|
"responsible_email": args[16],
|
|
"description": args[17].replace("\\n", "\n") if args[17] else None,
|
|
"role": Command.role(args[18]),
|
|
"team": Team.objects.get(pk=args[19]) if args[19] else None,
|
|
"year": args[20],
|
|
"date_joined": args[23],
|
|
"is_active": args[18] == "ADMIN", # TODO Replace it with "True"
|
|
"is_staff": args[18] == "ADMIN",
|
|
"is_superuser": args[18] == "ADMIN",
|
|
}
|
|
with transaction.atomic():
|
|
TFJMUser.objects.create(**obj_dict)
|
|
self.stdout.write(self.style.SUCCESS("Users imported"))
|
|
|
|
self.stdout.write("Importing organizers...")
|
|
with open("import_olddb/organizers.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
with transaction.atomic():
|
|
tournament = Tournament.objects.get(pk=args[2])
|
|
organizer = TFJMUser.objects.get(pk=args[1])
|
|
tournament.organizers.add(organizer)
|
|
tournament.save()
|
|
self.stdout.write(self.style.SUCCESS("Organizers imported"))
|
|
|
|
@transaction.atomic
|
|
def import_documents(self):
|
|
self.stdout.write("Importing documents...")
|
|
with open("import_olddb/documents.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if Document.objects.filter(file=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"file": args[0],
|
|
"uploaded_at": args[5],
|
|
}
|
|
if args[4] != "MOTIVATION_LETTER":
|
|
obj_dict["user"] = TFJMUser.objects.get(args[1]),
|
|
obj_dict["type"] = args[4].lower()
|
|
else:
|
|
try:
|
|
obj_dict["team"] = Team.objects.get(pk=args[2])
|
|
except Team.DoesNotExist:
|
|
print("Team with pk {} does not exist, ignoring".format(args[2]))
|
|
continue
|
|
with transaction.atomic():
|
|
if args[4] != "MOTIVATION_LETTER":
|
|
Authorization.objects.create(**obj_dict)
|
|
else:
|
|
MotivationLetter.objects.create(**obj_dict)
|
|
self.stdout.write(self.style.SUCCESS("Authorizations imported"))
|
|
|
|
with open("import_olddb/solutions.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if Document.objects.filter(file=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"file": args[0],
|
|
"team": Team.objects.get(pk=args[1]),
|
|
"problem": args[3],
|
|
"uploaded_at": args[4],
|
|
}
|
|
with transaction.atomic():
|
|
Solution.objects.create(**obj_dict)
|
|
self.stdout.write(self.style.SUCCESS("Solutions imported"))
|
|
|
|
with open("import_olddb/syntheses.csv") as f:
|
|
first_line = True
|
|
for line in f:
|
|
if first_line:
|
|
first_line = False
|
|
continue
|
|
|
|
line = line[:-1].replace("\"", "")
|
|
args = line.split(";")
|
|
args = [arg if arg and arg != "NULL" else None for arg in args]
|
|
|
|
if Document.objects.filter(file=args[0]).exists():
|
|
continue
|
|
|
|
obj_dict = {
|
|
"file": args[0],
|
|
"team": Team.objects.get(pk=args[1]),
|
|
"dest": "defender" if args[3] == "DEFENDER" else "opponent"
|
|
if args[4] == "OPPOSANT" else "rapporteur",
|
|
"uploaded_at": args[4],
|
|
}
|
|
with transaction.atomic():
|
|
Synthesis.objects.create(**obj_dict)
|
|
self.stdout.write(self.style.SUCCESS("Syntheses imported"))
|