577 lines
24 KiB
Python
577 lines
24 KiB
Python
import datetime
|
|
import inspect
|
|
import json
|
|
from rest_framework import serializers
|
|
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
|
|
from pruefplan_parser.models import JsonParser
|
|
from pruefplan_viewer.models import *
|
|
# from django.utils.encoding import smart_str
|
|
|
|
# Class to generate a UserToken with custom fields
|
|
class MyTokenObtainPairSerializer(TokenObtainPairSerializer):
|
|
@classmethod
|
|
def get_token(cls, user):
|
|
token = super().get_token(user) # Default Token
|
|
|
|
# Add custom informations to the Token
|
|
token["username"] = user.username
|
|
token["password"] = user.password
|
|
|
|
return token
|
|
|
|
|
|
class API_ExamDataSet_ForStudentSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = API_ExamDataSet_ForStudent
|
|
fields = '__all__'
|
|
|
|
|
|
|
|
class PruefPlan_JSON_Serializer(serializers.Serializer):
|
|
def create(self, validated_data):
|
|
return super().create(validated_data)
|
|
|
|
def validate(self, data):
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"Das 'array'-Feld ist erforderlich und muss als Python-Liste verfügbar sein."
|
|
)
|
|
|
|
return {"dataArray": data["array"]} # Rückgabe des validierten Datenarrays
|
|
|
|
|
|
class Lecturer_JSON_Serializer(serializers.Serializer):
|
|
# Definiert die Felder, die der Serializer erwartet
|
|
revision = serializers.CharField(required=True)
|
|
array = serializers.ListField()
|
|
|
|
def create(self, validated_data):
|
|
# Definieren der erwarteten Schlüssel im JSON-Daten
|
|
idKey = "kennung"
|
|
firstNameKey = "vorname"
|
|
lastNameKey = "nachname"
|
|
titleKey = "titel"
|
|
|
|
|
|
created_lecturers = [] # Liste zum Speichern der erstellten Lecturer-Objekte
|
|
try:
|
|
# Schleife über die Daten, um Lecturer-Objekte zu erstellen
|
|
for iteration_number, d in enumerate(validated_data["dataArray"]):
|
|
# Lecturer-Objekt erstellen und speichern
|
|
entry = Lecturer(
|
|
identification=d[idKey],
|
|
firstName=d[firstNameKey],
|
|
lastName=d[lastNameKey],
|
|
title=d[titleKey],
|
|
|
|
)
|
|
entry.save()
|
|
created_lecturers.append(entry) # Objekt der Liste hinzufügen
|
|
except KeyError as e:
|
|
# Fehlerbehandlung bei fehlenden Schlüsseln
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop.. Arguments: {3!r}. Message: [{4}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e,
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
except Exception as e:
|
|
# Allgemeine Fehlerbehandlung
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop. Arguments: {3!r}. Message: {4}."
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e.__str__(),
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
|
|
return {
|
|
"Serializer": "Lecturer",
|
|
"status": "success",
|
|
"created_count": len(created_lecturers),
|
|
} # Rückgabe des Erfolgsstatus
|
|
|
|
def validate(self, data):
|
|
# Validierung, um sicherzustellen, dass 'revision' und 'array' Felder vorhanden sind
|
|
if "revision" not in data.keys():
|
|
raise serializers.ValidationError("Das 'revision'-Feld ist erforderlich.")
|
|
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"Das 'array'-Feld ist erforderlich und muss als Python-Liste verfügbar sein."
|
|
)
|
|
|
|
return {"dataArray": data["array"]} # Rückgabe des validierten Datenarrays
|
|
|
|
|
|
class Attendance_JSON_Serializer(serializers.Serializer):
|
|
# Definiert die Felder, die der Serializer erwartet
|
|
revision = serializers.CharField(required=True)
|
|
array = serializers.ListField()
|
|
|
|
|
|
def create(self, validated_data):
|
|
# Definieren der erwarteten Schlüssel im JSON-Daten
|
|
idKey = "dozentenkennung"
|
|
yearKey = "jahr"
|
|
monthKey = "monat"
|
|
dayKey = "tag"
|
|
hourKey = "stunde"
|
|
minuteKey = "minute"
|
|
weekdayKey = "wochentag"
|
|
dateKey = "termin"
|
|
|
|
created_attendences = (
|
|
[]
|
|
) # Liste zum Speichern der erstellten Attendance-Objekte
|
|
try:
|
|
# Schleife über die Daten, um Attendance-Objekte zu erstellen
|
|
for iteration_number, d in enumerate(validated_data["dataArray"]):
|
|
# Überprüfung, ob das 'termin'-Feld vorhanden ist
|
|
keyAvailable = "termin" in d
|
|
if not keyAvailable:
|
|
# Warnung ausgeben, wenn kein Termin zugewiesen wurde
|
|
template = "Warning. Following Exam was not assigned any examination date ['termin']: [{1}]"
|
|
message = template.format(d[idKey])
|
|
JsonParser.warningMessages.append(message)
|
|
continue
|
|
|
|
# Datum und Uhrzeit aus den Daten extrahieren
|
|
date = datetime.date(
|
|
d[dateKey][yearKey], d[dateKey][monthKey], d[dateKey][dayKey]
|
|
)
|
|
time = datetime.time(d[dateKey][hourKey], d[dateKey][minuteKey])
|
|
|
|
# Attendance-Objekt erstellen und speichern
|
|
entry = Attendance(
|
|
identification=d[idKey],
|
|
time=time,
|
|
date=date,
|
|
dateTimeText=d[dateKey]["datumText"],
|
|
weekday=d[dateKey]["wochentag"],
|
|
)
|
|
entry.save()
|
|
created_attendences.append(entry) # Objekt der Liste hinzufügen
|
|
|
|
try:
|
|
# Dozent aus der Datenbank abrufen und mit der Anwesenheit verknüpfen
|
|
lecturer = Lecturer.objects.get(identification=d[idKey])
|
|
lecturer.attendance_set.add(
|
|
entry
|
|
) # Attendance dem Lecturer zuweisen
|
|
except Exception as e:
|
|
# Fehlerbehandlung, wenn der Dozent nicht gefunden wird
|
|
template = "Exception of type [{0}]. Arguments: {1!r}. Message: [{2}] entry might be missing"
|
|
message = template.format(type(e).__name__, e.args, idKey)
|
|
JsonParser.exceptionMessages.append(
|
|
message
|
|
) # Fehlermeldung speichern
|
|
|
|
except KeyError as e:
|
|
# Fehlerbehandlung bei fehlenden Schlüsseln
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop.. Arguments: {3!r}. Message: [{4}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e,
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
except Exception as e:
|
|
# Allgemeine Fehlerbehandlung
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop. Arguments: {3!r}. Message: {4}."
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e.__str__(),
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
|
|
return {
|
|
"Serializer": "Attendance",
|
|
"status": "success",
|
|
"created_count": len(created_attendences),
|
|
} # Rückgabe des Erfolgsstatus
|
|
|
|
def validate(self, data):
|
|
# Validierung, um sicherzustellen, dass 'revision' und 'array' Felder vorhanden sind
|
|
if "revision" not in data.keys():
|
|
raise serializers.ValidationError("Das 'revision'-Feld ist erforderlich.")
|
|
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"Das 'array'-Feld ist erforderlich und muss als Python-Liste verfügbar sein."
|
|
)
|
|
|
|
return {"dataArray": data["array"]} # Rückgabe des validierten Datenarrays
|
|
|
|
|
|
class Subject_JSON_Serializer(serializers.Serializer):
|
|
# Definiert die Felder, die der Serializer erwartet
|
|
revision = serializers.CharField(required=True)
|
|
array = serializers.ListField()
|
|
|
|
|
|
def create(self, validated_data):
|
|
# Definieren der erwarteten Schlüssel im JSON-Daten
|
|
idKey = "kennung"
|
|
initialsKey = "kurztext"
|
|
nameKey = "langtext"
|
|
ohmIdKey = "fachkennungOhm"
|
|
|
|
created_subjects = [] # Liste zum Speichern der erstellten Subject-Objekte
|
|
try:
|
|
# Schleife über die Daten, um Subject-Objekte zu erstellen
|
|
for iteration_number, d in enumerate(validated_data["dataArray"]):
|
|
# Subject-Objekt erstellen und speichern
|
|
entry = Subject(
|
|
identification=d[idKey],
|
|
initials=d[ohmIdKey][initialsKey],
|
|
name=d[ohmIdKey][nameKey],
|
|
)
|
|
entry.save()
|
|
created_subjects.append(entry) # Objekt der Liste hinzufügen
|
|
|
|
except KeyError as e:
|
|
# Fehlerbehandlung bei fehlenden Schlüsseln
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop.. Arguments: {3!r}. Message: [{4}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e,
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
except Exception as e:
|
|
# Allgemeine Fehlerbehandlung
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop. Arguments: {3!r}. Message: {4}."
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e.__str__(),
|
|
)
|
|
JsonParser.exceptionMessages.append(message) # Fehlermeldung speichern
|
|
|
|
return {
|
|
"Serializer": "Subject",
|
|
"status": "success",
|
|
"created_count": len(created_subjects),
|
|
} # Rückgabe des Erfolgsstatus
|
|
|
|
def validate(self, data):
|
|
# Validierung, um sicherzustellen, dass 'revision' und 'array' Felder vorhanden sind
|
|
if "revision" not in data.keys():
|
|
raise serializers.ValidationError("Das 'revision'-Feld ist erforderlich.")
|
|
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"Das 'array'-Feld ist erforderlich und muss als Python-Liste verfügbar sein."
|
|
)
|
|
|
|
return {"dataArray": data["array"]} # Rückgabe des validierten Datenarrays
|
|
|
|
|
|
class Exam_JSON_Serializer(serializers.Serializer):
|
|
# Define the fields expected by the serializer
|
|
revision = serializers.CharField(required=True)
|
|
array = serializers.ListField()
|
|
|
|
def create(self, validated_data):
|
|
# Define the expected keys in the JSON data
|
|
idKey = "kennung"
|
|
yearKey = "jahr"
|
|
monthKey = "monat"
|
|
dayKey = "tag"
|
|
hourKey = "stunde"
|
|
minuteKey = "minute"
|
|
weekdayKey = "wochentag"
|
|
dateKey = "termin"
|
|
partialExamsKey = "teilpruefungen"
|
|
partialExamIdKey = "kennung"
|
|
numRegStudKey = "anzahlAnmeldungen"
|
|
subjectIdsKey = "fachkennungen"
|
|
examExecutionsKey = "pruefungsdurchfuehrungen"
|
|
lecturerIdKey = "dozentenkennung"
|
|
supervisorTypeKey = "typ"
|
|
locationKey = "raum"
|
|
|
|
created_exams = [] # List to store created Exam objects
|
|
created_partialexams = [] # List to store created Exam objects
|
|
ignored_Vorgezogen_exams = 0 # List to ignored Exam objects
|
|
ignored_KeinTermin_exams = 0 # List to ignored Exam objects
|
|
try:
|
|
# Iterate over the data to create Exam objects
|
|
for iteration_number, d in enumerate(validated_data["dataArray"]):
|
|
|
|
# Certain exam entries are ignored
|
|
if d["pruefungsart"] == "LN_VORGEZOGEN": # These entries can be ignored
|
|
ignored_Vorgezogen_exams += 1
|
|
continue
|
|
|
|
keyAvailable = "termin" in d
|
|
if not keyAvailable:
|
|
# Warning if no examination date is assigned
|
|
template = "Warning. Following Exam was not assigned any examination date ['termin']: [{0}]"
|
|
message = template.format(d[idKey])
|
|
JsonParser.warningMessages.append(message)
|
|
ignored_KeinTermin_exams += 1
|
|
continue
|
|
|
|
# Extract date and time for the exam
|
|
date = datetime.date(
|
|
d[dateKey][yearKey], d[dateKey][monthKey], d[dateKey][dayKey]
|
|
)
|
|
time = datetime.time(d[dateKey][hourKey], d[dateKey][minuteKey])
|
|
|
|
# Create and save a new Exam entry
|
|
examEntry = Exam(
|
|
identification=d[idKey],
|
|
date=date,
|
|
time=time,
|
|
weekday=d[dateKey][weekdayKey],
|
|
)
|
|
examEntry.save()
|
|
created_exams.append(examEntry)
|
|
|
|
# Create PartialExam entries and link them with the main exam
|
|
for teilpruef in d[partialExamsKey]:
|
|
partialExamId = teilpruef[partialExamIdKey]
|
|
regStudCount = teilpruef[numRegStudKey]
|
|
|
|
# Create and save a new PartialExam entry
|
|
partialExam = PartialExam(
|
|
identification=partialExamId, regStudCount=regStudCount
|
|
)
|
|
partialExam.save()
|
|
created_partialexams.append(partialExam)
|
|
|
|
# Link Exam and PartialExam (One-to-Many relationship)
|
|
examEntry.partialexam_set.add(partialExam)
|
|
|
|
# Link subjects to the PartialExams
|
|
for subId in teilpruef[subjectIdsKey]:
|
|
try:
|
|
# Check if the subject exists and link it
|
|
subject = Subject.objects.get(identification=subId)
|
|
partialExam.subjectIds.add(
|
|
subject
|
|
) # Many-to-Many relationship
|
|
except Exception as e:
|
|
# Log an error if the subject is not found
|
|
template = "Exception of type [{0}]. Arguments: {1!r}. Message: [{2}] entry might be missing"
|
|
message = template.format(type(e).__name__, e.args, subId)
|
|
JsonParser.exceptionMessages.append(message)
|
|
|
|
# Link exam executions to PartialExams
|
|
for pruefdurch in teilpruef[examExecutionsKey]:
|
|
lecId = pruefdurch[lecturerIdKey]
|
|
supervisorType = pruefdurch[supervisorTypeKey]
|
|
location = pruefdurch[locationKey]
|
|
|
|
# Create and save a new ExamExecution entry
|
|
examExecution = ExamExecution(
|
|
supervisorType=supervisorType, location=location
|
|
)
|
|
examExecution.save()
|
|
|
|
# Check if the lecturer exists and link the exam execution
|
|
try:
|
|
lecturer = Lecturer.objects.get(pk=lecId)
|
|
lecturer.examexecution_set.add(
|
|
examExecution
|
|
) # One-to-Many relationship
|
|
except Exception as e:
|
|
# Log an error if the lecturer is not found
|
|
template = "Exception of type [{0}]. Arguments: {1!r}. Message: [{2}] entry might be missing"
|
|
message = template.format(type(e).__name__, e.args, lecId)
|
|
JsonParser.exceptionMessages.append(message)
|
|
|
|
# Link subjects to the ExamExecution
|
|
for subId in pruefdurch[subjectIdsKey]:
|
|
try:
|
|
# Check if the subject exists and link it
|
|
subject = Subject.objects.get(identification=subId)
|
|
examExecution.subjectIds.add(
|
|
subject
|
|
) # Many-to-Many relationship
|
|
except Exception as e:
|
|
# Log an error if the subject is not found
|
|
template = "Exception of type [{0}]. Arguments: {1!r}. Message: [{2}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__, e.args, subId
|
|
)
|
|
JsonParser.exceptionMessages.append(message)
|
|
|
|
# Link PartialExam and ExamExecution (One-to-Many relationship)
|
|
partialExam.examexecution_set.add(examExecution)
|
|
|
|
except KeyError as e:
|
|
# Error handling for missing keys in the dict
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop.. Arguments: {3!r}. Message: [{4}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e,
|
|
)
|
|
JsonParser.exceptionMessages.append(message)
|
|
except Exception as e:
|
|
# General error handling
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop. Arguments: {3!r}. Message: {4}."
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e.__str__(),
|
|
)
|
|
JsonParser.exceptionMessages.append(message)
|
|
|
|
return {
|
|
"Serializer": "Exam",
|
|
"status": "success",
|
|
"created_Exams": len(created_exams),
|
|
"created_PartialExams_count": len(created_partialexams),
|
|
"ignored_Exams -> Vorgezogen": ignored_Vorgezogen_exams,
|
|
"ignored_Exams -> Kein Termin": ignored_KeinTermin_exams,
|
|
}
|
|
|
|
def validate(self, data):
|
|
# Validation for the 'revision' field
|
|
if "revision" not in data.keys():
|
|
raise serializers.ValidationError("The 'revision' field is required.")
|
|
|
|
# Validation for the 'array' field
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"The 'array' field is required and must be a Python list."
|
|
)
|
|
|
|
return {"dataArray": data["array"]}
|
|
|
|
|
|
class RoomAllocation_JSON_Serializer(serializers.Serializer):
|
|
# Define the fields expected by the serializer
|
|
revision = serializers.CharField(required=True)
|
|
array = serializers.ListField()
|
|
|
|
def create(self, validated_data):
|
|
# Define the expected keys in the JSON data
|
|
idKey = "kennungPruefung"
|
|
yearKey = "jahr"
|
|
monthKey = "monat"
|
|
dayKey = "tag"
|
|
hourKey = "stunde"
|
|
minuteKey = "minute"
|
|
weekdayKey = "wochentag"
|
|
locationKey = "raum"
|
|
dateKey = "termin"
|
|
matrikelKey = "matrikel"
|
|
|
|
created_roomAllocations = [] # List to store created StudentExam objects
|
|
try:
|
|
# Iterate over the data to create StudentExam objects
|
|
for iteration_number, d in enumerate(validated_data["dataArray"]):
|
|
# Extract date and time
|
|
date = datetime.date(
|
|
d[dateKey][yearKey], d[dateKey][monthKey], d[dateKey][dayKey]
|
|
)
|
|
time = datetime.time(d[dateKey][hourKey], d[dateKey][minuteKey])
|
|
|
|
# Find the corresponding Exam entry
|
|
exam = Exam.objects.filter(
|
|
identification=d[idKey], date=date, time=time
|
|
).first()
|
|
|
|
# Iterate through partial exams and their executions
|
|
partialExams = exam.partialexam_set.all()
|
|
for partialExam in partialExams:
|
|
examExecutions = partialExam.examexecution_set.all()
|
|
for examExecution in examExecutions:
|
|
# Check if the location matches and create StudentExam entry
|
|
if examExecution.location == d[locationKey]:
|
|
studentExamEntry = StudentExam(
|
|
examIdentification=d[idKey],
|
|
date=date,
|
|
time=time,
|
|
weekday=d[dateKey][weekdayKey],
|
|
location=d[locationKey],
|
|
examExecution=examExecution,
|
|
)
|
|
studentExamEntry.save()
|
|
created_roomAllocations.append(studentExamEntry)
|
|
|
|
# Create or update Student entries
|
|
for matrikel in d[matrikelKey]:
|
|
studentEntry = Student.objects.filter(matrikel=matrikel).exists()
|
|
if not studentEntry:
|
|
# Create a new Student entry if it doesn't exist
|
|
entry = Student(matrikel=matrikel)
|
|
entry.save()
|
|
else:
|
|
entry = Student.objects.all().filter(matrikel=matrikel).first()
|
|
entry.exams.add(studentExamEntry)
|
|
except KeyError as e:
|
|
# Error handling for missing keys in the dict
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop.. Arguments: {3!r}. Message: [{4}] entry might be missing"
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e,
|
|
)
|
|
JsonParser.exceptionMessages.append(message)
|
|
except Exception as e:
|
|
# General error handling
|
|
calling_function = inspect.stack()[0][3]
|
|
template = "Exception of type [{0}] occurred in function {1} in iteration {2} of the loop. Arguments: {3!r}. Message: {4}."
|
|
message = template.format(
|
|
type(e).__name__,
|
|
calling_function,
|
|
iteration_number,
|
|
e.args,
|
|
e.__str__(),
|
|
)
|
|
JsonParser.exceptionMessages.append(message)
|
|
|
|
return {
|
|
"Serializer": "RoomAllocation",
|
|
"status": "success",
|
|
"created_RoomAllocations": len(created_roomAllocations),
|
|
}
|
|
|
|
def validate(self, data):
|
|
# Validation for the 'revision' field
|
|
if "revision" not in data.keys():
|
|
raise serializers.ValidationError("The 'revision' field is required.")
|
|
|
|
# Validation for the 'array' field
|
|
if "array" not in data or not isinstance(data["array"], list):
|
|
raise serializers.ValidationError(
|
|
"The 'array' field is required and must be a Python list."
|
|
)
|
|
|
|
return {"dataArray": data["array"]}
|