ramon
/
vbv
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merged in feature/VBV-464-excel-importer (pull request #160) 

Feature/VBV-464 excel importer

Approved-by: Daniel Egger
This commit is contained in:
Christian Cueni 2023-07-21 10:07:50 +00:00 committed by Daniel Egger
parent 4d9aa07b2d 842b0b4334
commit a1403f8a54
14 changed files with 697 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
server/config/settings/base.py
View File

@ -122,6 +122,7 @@ LOCAL_APPS = [
"vbv_lernwelt.notify",
"vbv_lernwelt.assignment",
"vbv_lernwelt.duedate",
"vbv_lernwelt.importer",
]
# https://docs.djangoproject.com/en/dev/ref/settings/#installed-apps
INSTALLED_APPS = DJANGO_APPS + THIRD_PARTY_APPS + LOCAL_APPS

22
server/config/urls.py
View File

@ -42,6 +42,11 @@ from vbv_lernwelt.feedback.views import (
get_expert_feedbacks_for_course,
get_feedback_for_circle,
)
from vbv_lernwelt.importer.views import (
coursesessions_students_import,
coursesessions_trainers_import,
t2l_sync,
)
from vbv_lernwelt.notify.views import email_notification_settings
from wagtail import urls as wagtail_urls
from wagtail.admin import urls as wagtailadmin_urls
@ -143,6 +148,23 @@ urlpatterns = [
# edoniq test
path(r'api/core/edoniq-test/export-users/', export_students, name='edoniq_export_students'),
# importer
path(
r"server/importer/coursesession-trainer-import/",
coursesessions_trainers_import,
name="coursesessions_trainers_import",
),
path(
r"server/importer/coursesession-students-import/",
coursesessions_students_import,
name="coursesessions_students_import",
),
path(
r"server/importer/t2l-sync/",
t2l_sync,
name="t2l_sync",
),
# testing and debug
path('server/raise_error/',
user_passes_test(lambda u: u.is_superuser, login_url='/login/')(

1
server/vbv_lernwelt/core/admin.py
View File

@ -23,6 +23,7 @@ class UserAdmin(auth_admin.UserAdmin):
},
),
(_("Important dates"), {"fields": ("last_login", "date_joined")}),
(_("Additional data"), {"fields": ("additional_json_data",)}),
)
list_display = [
"username",

25
server/vbv_lernwelt/course/management/commands/create_default_courses.py
View File

@ -79,7 +79,7 @@ from vbv_lernwelt.feedback.creators.create_demo_feedback import create_feedback
from vbv_lernwelt.importer.services import (
import_course_sessions_from_excel,
import_students_from_excel,
import_trainers_from_excel,
import_trainers_from_excel_for_training,
)
from vbv_lernwelt.learnpath.create_vv_new_learning_path import (
create_vv_new_learning_path,
@ -561,12 +561,13 @@ def create_course_training_de():
print(current_dir)
course = Course.objects.get(id=COURSE_UK_TRAINING)
import_course_sessions_from_excel(
course,
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
course=course,
restrict_language="de",
)
import_trainers_from_excel(
course,
import_trainers_from_excel_for_training(
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
course=course,
)
import_students_from_excel(
f"{current_dir}/../../../importer/tests/Schulungen_Teilnehmende.xlsx",
@ -650,13 +651,13 @@ def create_course_training_fr():
print(current_dir)
course = Course.objects.get(id=COURSE_UK_TRAINING_FR)
import_course_sessions_from_excel(
course,
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
language="fr",
restrict_language="fr",
course=course,
)
import_trainers_from_excel(
course,
import_trainers_from_excel_for_training(
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
course=course,
language="fr",
)
import_students_from_excel(
@ -737,13 +738,13 @@ def create_course_training_it():
print(current_dir)
course = Course.objects.get(id=COURSE_UK_TRAINING_IT)
import_course_sessions_from_excel(
course,
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
language="it",
restrict_language="it",
course=course,
)
import_trainers_from_excel(
course,
import_trainers_from_excel_for_training(
f"{current_dir}/../../../importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx",
course=course,
language="it",
)
import_students_from_excel(

313
server/vbv_lernwelt/importer/services.py
View File

@ -1,9 +1,11 @@
from typing import Any, Dict
from datetime import date, datetime, time
from typing import Any, Dict, List
import structlog
from openpyxl.reader.excel import load_workbook
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.consts import COURSE_UK, COURSE_UK_FR, COURSE_UK_IT
from vbv_lernwelt.course.models import Course, CourseSession, CourseSessionUser
from vbv_lernwelt.course_session.models import CourseSessionAttendanceCourse
from vbv_lernwelt.importer.utils import (
@ -15,9 +17,27 @@ from vbv_lernwelt.learnpath.models import Circle, LearningContentAttendanceCours
logger = structlog.get_logger(__name__)
CIRCLE_NAMES = {
"Kickoff": {"de": "Kickoff", "fr": "Lancement", "it": "Introduzione"},
"Basis": {"de": "Basis", "fr": "Base", "it": "Base"},
"Fahrzeug": {"de": "Fahrzeug", "fr": "Véhicule", "it": "Veicolo"},
"Haushalt Teil 1": {"de": "Haushalt Teil 1", "fr": "Habitat partie 1", "it": ""},
"Haushalt Teil 2": {"de": "Haushalt Teil 2", "fr": "Haushalt Teil 2", "it": ""},
}
T2L_IGNORE_FIELDS = ["Vorname", "Name", "Email", "Sprache", "Durchführungen"]
class DataImportError(Exception):
pass
def create_or_update_user(
email: str, first_name: str = "", last_name: str = "", sso_id: str = None
email: str,
first_name: str = "",
last_name: str = "",
sso_id: str = None,
contract_number: str = "",
):
logger.debug(
"create_or_update_user",
@ -33,6 +53,13 @@ def create_or_update_user(
if user_qs.exists():
user = user_qs.first()
if not user and contract_number:
user_qs = User.objects.filter(
additional_json_data__Lehrvertragsnummer=contract_number
)
if user_qs.exists():
user = user_qs.first()
if not user:
user_qs = User.objects.filter(email=email)
if user_qs.exists():
@ -53,25 +80,41 @@ def create_or_update_user(
return user
def import_course_sessions_from_excel(course: Course, filename: str, language="de"):
def import_course_sessions_from_excel(
filename: str, course: Course = None, restrict_language=None
):
workbook = load_workbook(filename=filename)
sheet = workbook["Schulungen Durchführung"]
tuple_list = calc_header_tuple_list_from_pyxl_sheet(sheet)
for row in tuple_list:
data = dict(row)
validate_row_data(data, ["Klasse", "ID", "Generation", "Region", "Sprache"])
language = data["Sprache"].strip()
# this can be removed when the training import (create_course_training_xx) is no longer used
if restrict_language and language != restrict_language:
continue
if not course:
course = get_uk_course(language)
create_or_update_course_session(
course, dict(row), language=language, circles=["Fahrzeug"]
course, data, language, circle_keys=["Kickoff", "Basis", "Fahrzeug"]
)
def create_or_update_course_session(
course: Course, data: Dict[str, Any], language="de", circles=None
course: Course,
data: Dict[str, Any],
language: str,
circle_keys=None,
):
"""
:param data: the following keys are required to process the data: Generation, Region, Klasse
:return:
"""
logger.debug(
"create_or_update_course_session",
course=course.title,
@ -79,10 +122,9 @@ def create_or_update_course_session(
label="import",
)
if circles is None:
circles = []
if not circle_keys:
circle_keys = []
# TODO: validation
group = data["Klasse"].strip()
import_id = data["ID"].strip()
@ -91,12 +133,8 @@ def create_or_update_course_session(
title = f"{region} {generation} {group}"
if not import_id.lower().startswith(language.lower()):
# FIXME: language check depends on import_id format for now...
return None
cs, _created = CourseSession.objects.get_or_create(
import_id=import_id, group=group, course=course
title=title, course=course, import_id=import_id
)
cs.additional_json_data["import_data"] = data
@ -109,31 +147,23 @@ def create_or_update_course_session(
cs.import_id = import_id
cs.save()
for circle in circles:
attendance_course_lp_qs = None
if language == "de":
attendance_course_lp_qs = LearningContentAttendanceCourse.objects.filter(
slug=f"{course.slug}-lp-circle-{circle.lower()}-lc-präsenzkurs-{circle.lower()}"
for circle in circle_keys:
circle_name = CIRCLE_NAMES[circle][language]
attendance_course_lc = LearningContentAttendanceCourse.objects.filter(
slug=f"{course.slug}-lp-circle-{circle_name.lower()}-lc-präsenzkurs-{circle_name.lower()}"
).first()
if attendance_course_lc:
# update existing data
csa, _created = CourseSessionAttendanceCourse.objects.get_or_create(
course_session=cs, learning_content=attendance_course_lc
)
elif language == "fr":
# todo: this is a hack remove me
attendance_course_lp_qs = LearningContentAttendanceCourse.objects.filter(
slug=f"{course.slug}-lp-circle-véhicule-lc-cours-de-présence-véhicule-à-moteur"
)
elif language == "it":
# todo: this is a hack remove me
attendance_course_lp_qs = LearningContentAttendanceCourse.objects.filter(
slug=f"{course.slug}-lp-circle-veicolo-lc-corso-di-presenza-veicolo"
)
location = f"{data[f'{circle} Raum']}, {data[f'{circle} Standort']}, {data[f'{circle} Adresse']}"
csa.location = location
if attendance_course_lp_qs and attendance_course_lp_qs.exists():
csa = CourseSessionAttendanceCourse.objects.create(
course_session=cs,
learning_content=attendance_course_lp_qs.first(),
location=data[f"{circle} Raum"],
trainer="",
)
csa.trainer = ""
csa.due_date.start = try_parse_datetime(data[f"{circle} Start"])[1]
csa.due_date.end = try_parse_datetime(data[f"{circle} Ende"])[1]
csa.due_date.save()
@ -141,19 +171,50 @@ def create_or_update_course_session(
return cs
def import_trainers_from_excel(course: Course, filename: str, language="de"):
def validate_row_data(data: Dict[str, any], required_headers: List[str]):
for header in required_headers:
some = str(data.get(header, "")).strip()
if str(data.get(header, "")).strip() in ["", "None"]:
logger.debug(
"validate_row_data_missing_header",
data={"data": data, "header": header},
label="import",
)
raise DataImportError(f"Missing or empty value for header {header}")
def get_uk_course(language: str) -> Course:
if language == "fr":
course_id = COURSE_UK_FR
elif language == "it":
course_id = COURSE_UK_IT
else:
course_id = COURSE_UK
return Course.objects.get(id=course_id)
def import_trainers_from_excel_for_training(
filename: str, language="de", course: Course = None
):
workbook = load_workbook(filename=filename)
sheet = workbook["Schulungen Trainer"]
tuple_list = calc_header_tuple_list_from_pyxl_sheet(sheet)
for row in tuple_list:
create_or_update_trainer(course, dict(row), language=language)
data = dict(row)
validate_row_data(
data, ["Email", "Vorname", "Name", "Sprache", "Klasse", "Generation"]
)
create_or_update_trainer(course, data, language=language)
def create_or_update_trainer(course: Course, data: Dict[str, Any], language="de"):
course_title = course.title if course else "None"
logger.debug(
"create_or_update_trainer",
course=course.title,
course=course_title,
data=data,
label="import",
)
@ -163,62 +224,64 @@ def create_or_update_trainer(course: Course, data: Dict[str, Any], language="de"
first_name=data["Vorname"],
last_name=data["Name"],
)
user.language = data["Sprache"]
user.save()
groups = [g.strip() for g in data["Klasse"].strip().split(",")]
group = data["Klasse"].strip()
# general expert handling
for group in groups:
import_id = f"{data['Generation'].strip()} {group}"
course_session = CourseSession.objects.filter(
import_id = f"{data['Generation'].strip()} {group}"
course_session = CourseSession.objects.filter(
import_id=import_id,
group=group,
).first()
if course_session:
csu, _created = CourseSessionUser.objects.get_or_create(
course_session_id=course_session.id, user_id=user.id
)
csu.role = CourseSessionUser.Role.EXPERT
csu.save()
else:
logger.warning(
"create_or_update_trainer: course_session not found",
import_id=import_id,
group=group,
).first()
if course_session:
csu, _created = CourseSessionUser.objects.get_or_create(
course_session_id=course_session.id, user_id=user.id
)
csu.role = CourseSessionUser.Role.EXPERT
csu.save()
else:
logger.warning(
"create_or_update_trainer: course_session not found",
import_id=import_id,
group=group,
label="import",
)
label="import",
)
if not course and not course_session:
logger.warning(
"create_or_update_trainer: course_session and course are None",
import_id=import_id,
group=group,
label="import",
)
return
if not course:
return
course = course_session.course
# circle expert handling
circle_data = parse_circle_group_string(data["Circles"])
for circle_string in circle_data:
parts = circle_string.split("(", 1)
circle_name = parts[0].strip()
groups = [g.strip() for g in parts[1].rstrip(")").strip().split(",")]
# FIXME: hardcoded translation
if language == "fr" and circle_name == "Fahrzeug":
circle_name = "Véhicule"
for circle_key in circle_data:
circle_name = CIRCLE_NAMES[circle_key][language]
# print(circle_name, groups)
for group in groups:
import_id = f"{data['Generation'].strip()} {group}"
course_session = CourseSession.objects.filter(
import_id=import_id, group=group
).first()
circle = Circle.objects.filter(
slug=f"{course.slug}-lp-circle-{circle_name.lower()}"
).first()
import_id = f"{data['Generation'].strip()} {group}"
course_session = CourseSession.objects.filter(
import_id=import_id, group=group
).first()
circle = Circle.objects.filter(
slug=f"{course.slug}-lp-circle-{circle_name.lower()}"
).first()
if course_session and circle:
csu = CourseSessionUser.objects.filter(
course_session_id=course_session.id, user_id=user.id
).first()
if csu:
csu.expert.add(circle)
csu.save()
if course_session and circle:
csu = CourseSessionUser.objects.filter(
course_session_id=course_session.id, user_id=user.id
).first()
if csu:
csu.expert.add(circle)
csu.save()
def import_students_from_excel(filename: str):
@ -227,7 +290,18 @@ def import_students_from_excel(filename: str):
tuple_list = calc_header_tuple_list_from_pyxl_sheet(sheet)
for row in tuple_list:
create_or_update_student(dict(row))
data = dict(row)
validate_row_data(
data,
[
"Email",
"Vorname",
"Name",
"Sprache",
"Durchführungen",
],
)
create_or_update_student(data)
def create_or_update_student(data: Dict[str, Any]):
@ -241,16 +315,71 @@ def create_or_update_student(data: Dict[str, Any]):
email=data["Email"].lower(),
first_name=data["Vorname"],
last_name=data["Name"],
contract_number=data.get("Lehrvertragsnummer", ""),
)
# TODO: handle language
user.language = data["Sprache"]
update_user_json_data(user, data)
user.save()
# general expert handling
import_ids = [i.strip() for i in data["Durchführungen"].split(",")]
for import_id in import_ids:
course_session = CourseSession.objects.filter(import_id=import_id).first()
if course_session:
csu, _created = CourseSessionUser.objects.get_or_create(
course_session_id=course_session.id, user_id=user.id
)
csu.save()
import_id = data["Durchführungen"]
course_session = CourseSession.objects.filter(import_id=import_id).first()
if course_session:
csu, _created = CourseSessionUser.objects.get_or_create(
course_session_id=course_session.id, user_id=user.id
)
csu.save()
def sync_students_from_t2l_excel(filename: str):
workbook = load_workbook(filename=filename)
sheet = workbook.active
tuple_list = calc_header_tuple_list_from_pyxl_sheet(sheet)
for row in tuple_list:
data = dict(row)
sync_students_from_t2l(data | {})
def sync_students_from_t2l(data):
# ignore errors
try:
user = User.objects.get(
additional_json_data__Lehrvertragsnummer=data["Lehrvertragsnummer"]
)
except User.DoesNotExist:
return
# only sync data that is not in our user model
for field in T2L_IGNORE_FIELDS:
try:
del data[field]
except KeyError:
pass
update_user_json_data(user, data)
user.save()
def update_user_json_data(user: User, data: Dict[str, Any]):
user.additional_json_data = user.additional_json_data | sanitize_json_data_input(
data
)
def sanitize_json_data_input(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Saving additional_json_data fails if the data contains datetime objects.
This is a quick and dirty fix to convert datetime objects to iso strings.
"""
for key, value in data.items():
if isinstance(value, datetime):
data[key] = value.isoformat()
elif isinstance(value, date):
data[key] = value.isoformat()
elif isinstance(value, time):
data[key] = value.isoformat()
else:
data[key] = value
return data

12
server/vbv_lernwelt/importer/templates/admin/importer/error.html Normal file
View File

@ -0,0 +1,12 @@
{% extends "admin/index.html" %}
{% block content %}
<div id="content-main">
<div class="content">
<h1>Die Daten konnten nicht importiert werden</h1>
<p>Folgender Fehler ist aufgetreten:</p>
<pre>{{ error }}</pre>
</div>
</div>
{% endblock %}

BIN
server/vbv_lernwelt/importer/tests/Schulungen_Durchfuehrung_Trainer.xlsx
View File

Binary file not shown.

BIN
server/vbv_lernwelt/importer/tests/Schulungen_Teilnehmende.xlsx
View File

Binary file not shown.

100
server/vbv_lernwelt/importer/tests/test_import_course_sessions.py
View File

@ -6,7 +6,11 @@ from openpyxl.reader.excel import load_workbook
from vbv_lernwelt.course.creators.test_course import create_test_course
from vbv_lernwelt.course.models import CourseSession
from vbv_lernwelt.course_session.models import CourseSessionAttendanceCourse
from vbv_lernwelt.importer.services import create_or_update_course_session
from vbv_lernwelt.importer.services import (
create_or_update_course_session,
DataImportError,
validate_row_data,
)
from vbv_lernwelt.importer.utils import calc_header_tuple_list_from_pyxl_sheet
test_dir = os.path.dirname(os.path.abspath(__file__))
@ -26,10 +30,10 @@ class ImportCourseSessionTestCase(TestCase):
for row in tuple_list:
print(row)
create_or_update_course_session(
self.course, dict(row), language="de", circles=["Fahrzeug"]
self.course, dict(row), language="de", circle_keys=["Fahrzeug"]
)
self.assertEqual(CourseSession.objects.count(), 3)
self.assertEqual(CourseSession.objects.count(), 6)
class CreateOrUpdateCourseSessionTestCase(TestCase):
@ -41,6 +45,7 @@ class CreateOrUpdateCourseSessionTestCase(TestCase):
("ID", "DE 2023 A"),
("Generation", 2023),
("Region", "Deutschschweiz"),
("Sprache", "de"),
("Klasse", "A"),
("Fahrzeug Start", "06.06.2023, 13:30"),
("Fahrzeug Ende", "06.06.2023, 15:00"),
@ -54,7 +59,9 @@ class CreateOrUpdateCourseSessionTestCase(TestCase):
data = dict(row)
cs = create_or_update_course_session(self.course, data, circles=["Fahrzeug"])
cs = create_or_update_course_session(
self.course, data, language="de", circle_keys=["Fahrzeug"]
)
self.assertEqual(cs.import_id, "DE 2023 A")
self.assertEqual(cs.title, "Deutschschweiz 2023 A")
@ -82,6 +89,7 @@ class CreateOrUpdateCourseSessionTestCase(TestCase):
("ID", "DE 2023"),
("Generation", 2023),
("Region", "Deutschschweiz"),
("Sprache", "de"),
("Klasse", "A"),
("Fahrzeug Start", "06.06.2023, 13:30"),
("Fahrzeug Ende", "06.06.2023, 15:00"),
@ -95,7 +103,9 @@ class CreateOrUpdateCourseSessionTestCase(TestCase):
data = dict(row)
cs = create_or_update_course_session(self.course, data, circles=["Fahrzeug"])
cs = create_or_update_course_session(
self.course, data, language="de", circle_keys=["Fahrzeug"]
)
self.assertEqual(1, CourseSession.objects.count())
@ -112,3 +122,83 @@ class CreateOrUpdateCourseSessionTestCase(TestCase):
self.assertEqual(
attendance_course.due_date.end.isoformat(), "2023-06-06T13:00:00+00:00"
)
def test_import_course_session_twice(self):
"""
importing the course session twice should create the
`CourseSessionAttendanceCourse` only once
"""
row = [
("ID", "DE 2023 A"),
("Generation", 2023),
("Region", "Deutschschweiz"),
("Sprache", "de"),
("Klasse", "A"),
("Fahrzeug Start", "06.06.2023, 13:30"),
("Fahrzeug Ende", "06.06.2023, 15:00"),
(
"Fahrzeug Raum",
"https://teams.microsoft.com/l/meetup-join/19%3ameeting_N2I5YzViZTQtYTM2Ny00OTYwLTgzNzAtYWI4OTQzODcxNTlj%40thread.v2/0?context=%7b%22Tid%22%3a%22fedd03c8-a756-4803-8f27-0db8f7c488f2%22%2c%22Oid%22%3a%22f92e6382-3884-4e71-a2fd-b305a75d9812%22%7d",
),
("Fahrzeug Standort", None),
("Fahrzeug Adresse", None),
]
data = dict(row)
cs1 = create_or_update_course_session(
self.course, data, language="de", circle_keys=["Fahrzeug"]
)
attendance_course1 = CourseSessionAttendanceCourse.objects.first()
self.assertEqual(
attendance_course1.due_date.start.isoformat(), "2023-06-06T11:30:00+00:00"
)
self.assertEqual(
attendance_course1.due_date.end.isoformat(), "2023-06-06T13:00:00+00:00"
)
# import a second time
data["Fahrzeug Start"] = "06.06.2023, 13:15"
data["Fahrzeug Ende"] = "06.06.2023, 14:45"
cs2 = create_or_update_course_session(
self.course, data, language="de", circle_keys=["Fahrzeug"]
)
self.assertEqual(cs1.id, cs2.id)
attendance_course2 = CourseSessionAttendanceCourse.objects.first()
self.assertEqual(
attendance_course2.due_date.start.isoformat(), "2023-06-06T11:15:00+00:00"
)
self.assertEqual(
attendance_course2.due_date.end.isoformat(), "2023-06-06T12:45:00+00:00"
)
self.assertEqual(attendance_course1.id, attendance_course2.id)
def test_raise_exception_if_header_is_missing(self):
data = [
("ID", "DE 2023"),
]
with self.assertRaises(DataImportError):
validate_row_data(
dict(data),
[
"ID",
"Generation",
],
)
def test_raise_exception_if_required_field_is_empty(self):
data = [
("ID", "DE 2023"),
("Generation", ""),
]
with self.assertRaises(DataImportError):
validate_row_data(
dict(data),
[
"ID",
"Generation",
],
)

64
server/vbv_lernwelt/importer/tests/test_import_students.py
View File

@ -1,5 +1,4 @@
import os
from datetime import datetime
from django.test import TestCase
from openpyxl.reader.excel import load_workbook
@ -44,33 +43,58 @@ class CreateOrUpdateStudentTestCase(TestCase):
group="A",
)
def test_create_student(self):
row = [
("Name", "Rascher"),
("Vorname", "Barbara"),
("Email", "barbara.rascher@vbv-afa.ch"),
("Sprache", "de"),
("Durchführungen", "DE 2023 A"),
("Datum", datetime(2023, 9, 6, 0, 0)),
(None, "VBV"),
(None, None),
(None, None),
(None, None),
(None, None),
]
self.user_dict = {
"Name": "Rascher",
"Vorname": "Barbara",
"Email": "barbara.rascher@vbv-afa.ch",
"Sprache": "de",
"Durchführungen": "DE 2023 A",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 43",
}
create_or_update_student(dict(row))
def test_create_student(self):
create_or_update_student(self.user_dict)
self.assertEqual(
CourseSessionUser.objects.filter(
user__email=self.user_dict["Email"]
).count(),
1,
)
csu = CourseSessionUser.objects.filter(
course_session=self.course_session_a,
).first()
self.assertEqual(csu.role, CourseSessionUser.Role.MEMBER)
self.assertEqual(csu.user.email, self.user_dict["Email"])
self.assertEqual(csu.user.additional_json_data, self.user_dict)
def test_update_student(self):
create_or_update_student(self.user_dict)
self.user_dict["Email"] = "br@vbv.ch"
create_or_update_student(self.user_dict)
self.assertEqual(
CourseSessionUser.objects.filter(
user__email="barbara.rascher@vbv-afa.ch"
).count(),
0,
)
self.assertEqual(
CourseSessionUser.objects.filter(
user__additional_json_data__Lehrvertragsnummer=self.user_dict[
"Lehrvertragsnummer"
]
).count(),
1,
)
csu = CourseSessionUser.objects.get(
csu = CourseSessionUser.objects.filter(
course_session=self.course_session_a,
)
).first()
self.assertEqual(csu.role, CourseSessionUser.Role.MEMBER)
self.assertEqual(csu.user.email, "barbara.rascher@vbv-afa.ch")
self.assertEqual(csu.user.additional_json_data, self.user_dict)

32
server/vbv_lernwelt/importer/tests/test_import_trainers.py
View File

@ -58,19 +58,29 @@ class CreateOrUpdateTrainerTestCase(TestCase):
)
def test_create_trainer(self):
row = [
("Name", "Hänni"),
("Vorname", "Fabienne"),
("Email", "fabienne.haenni@vbv-afa.ch"),
("Sprache", "de"),
("Generation", "DE 2023"),
("Klasse", "A, B"),
("Circles", "Fahrzeug (A, B), Reisen (A), KMU (B)"),
("Status Referenten", "ok"),
(None, "Schulung D"),
rows = [
[
("Name", "Hänni"),
("Vorname", "Fabienne"),
("Email", "fabienne.haenni@vbv-afa.ch"),
("Sprache", "de"),
("Generation", "DE 2023"),
("Klasse", "A"),
("Circles", "Fahrzeug, Haushalt Teil 1"),
],
[
("Name", "Hänni"),
("Vorname", "Fabienne"),
("Email", "fabienne.haenni@vbv-afa.ch"),
("Sprache", "de"),
("Generation", "DE 2023"),
("Klasse", "B"),
("Circles", "Fahrzeug, Haushalt Teil 2"),
],
]
create_or_update_trainer(self.course, dict(row))
for row in rows:
create_or_update_trainer(self.course, dict(row))
self.assertEqual(
CourseSessionUser.objects.filter(

178
server/vbv_lernwelt/importer/tests/test_t2l_sync.py Normal file
View File

@ -0,0 +1,178 @@
import os
from datetime import date, datetime, time
from django.test import TestCase
from vbv_lernwelt.course.creators.test_course import create_test_course
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.importer.services import (
create_or_update_student,
sanitize_json_data_input,
sync_students_from_t2l,
)
test_dir = os.path.dirname(os.path.abspath(__file__))
class SyncT2lTestCase(TestCase):
def setUp(self):
self.course = create_test_course(include_vv=False)
self.course_session_a = CourseSession.objects.create(
course=self.course,
title="Deutschschweiz 2023 A",
import_id="DE 2023 A",
group="A",
)
self.user_dict = {
"Name": "Rascher",
"Vorname": "Barbara",
"Email": "barbara.rascher@vbv-afa.ch",
"Sprache": "de",
"Durchführungen": "DE 2023 A",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 43",
}
create_or_update_student(self.user_dict)
def test_updates_field(self):
user_dict = {
"Name": "Rascher",
"Vorname": "Barbara",
"Email": "barbara.rascher@vbv-afa.ch",
"Sprache": "de",
"Durchführungen": "DE 2023 A",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 65",
}
sync_students_from_t2l(user_dict)
updated_user = CourseSessionUser.objects.get(
user__email=self.user_dict["Email"]
)
self.assertEqual(
updated_user.user.additional_json_data["Tel. Privat"],
user_dict["Tel. Privat"],
)
def test_adds_field(self):
user_dict = {
"Name": "Rascher",
"Vorname": "Barbara",
"Email": "barbara.rascher@vbv-afa.ch",
"Sprache": "de",
"Durchführungen": "DE 2023 A",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 43",
"Firma": "VBV",
}
sync_students_from_t2l(user_dict)
updated_user = CourseSessionUser.objects.get(
user__email=self.user_dict["Email"]
)
self.assertEqual(
updated_user.user.additional_json_data["Firma"], user_dict["Firma"]
)
def test_ignors_defined_field(self):
user_dict = {
"Name": "Rascher2",
"Vorname": "Barbara2",
"Email": "barbara.rascher2@vbv-afa.ch",
"Sprache": "fr",
"Durchführungen": "DE 2023 B",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 43",
}
sync_students_from_t2l(user_dict)
updated_user = CourseSessionUser.objects.get(
user__email=self.user_dict["Email"]
)
self.assertEqual(
updated_user.user.additional_json_data["Name"], self.user_dict["Name"]
)
self.assertEqual(
updated_user.user.additional_json_data["Vorname"], self.user_dict["Vorname"]
)
self.assertEqual(
updated_user.user.additional_json_data["Durchführungen"],
self.user_dict["Durchführungen"],
)
self.assertEqual(
updated_user.user.additional_json_data["Sprache"], self.user_dict["Sprache"]
)
def test_ignors_missing_defined_field(self):
user_dict = {
"Name": "Rascher2",
"Vorname": "Barbara2",
"Email": "barbara.rascher2@vbv-afa.ch",
"Sprache": "fr",
"Lehrvertragsnummer": "1234",
"Tel. Privat": "079 593 83 43",
}
sync_students_from_t2l(user_dict)
updated_user = CourseSessionUser.objects.get(
user__email=self.user_dict["Email"]
)
self.assertEqual(
updated_user.user.additional_json_data["Name"], self.user_dict["Name"]
)
self.assertEqual(
updated_user.user.additional_json_data["Vorname"], self.user_dict["Vorname"]
)
self.assertEqual(
updated_user.user.additional_json_data["Sprache"], self.user_dict["Sprache"]
)
def test_ignors_wrong_contract_nummer(self):
user_dict = {
"Name": "Rascher2",
"Vorname": "Barbara2",
"Email": "barbara.rascher2@vbv-afa.ch",
"Sprache": "fr",
"Lehrvertragsnummer": "12345",
"Tel. Privat": "079 593 83 43",
}
try:
sync_students_from_t2l(user_dict)
updated_user = CourseSessionUser.objects.get(
user__email=self.user_dict["Email"]
)
except Exception as e:
self.fail(
f"SyncT2lTestCase.test_ignors_wrong_contract_number: An exception was unexpectedly raised: {str(e)}"
)
class SanitizerTestCase(TestCase):
def test_date(self):
a_date = date(2021, 1, 1)
user_dict = {"Name": "Rascher", "Datum": a_date}
expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
sanitized_data = sanitize_json_data_input(user_dict)
self.assertEqual(sanitized_data, expected_sanitized_data)
def test_datetime(self):
a_date = datetime(2021, 1, 1)
user_dict = {"Name": "Rascher", "Datum": a_date}
expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
sanitized_data = sanitize_json_data_input(user_dict)
self.assertEqual(sanitized_data, expected_sanitized_data)
def test_time(self):
a_date = time(23, 59, 59)
user_dict = {"Name": "Rascher", "Datum": a_date}
expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
sanitized_data = sanitize_json_data_input(user_dict)
self.assertEqual(sanitized_data, expected_sanitized_data)

64
server/vbv_lernwelt/importer/views.py
View File

@ -0,0 +1,64 @@
import traceback
from typing import Callable
from django.contrib import messages
from django.contrib.admin.views.decorators import staff_member_required
from django.shortcuts import redirect, render
from vbv_lernwelt.importer.services import (
import_course_sessions_from_excel,
import_students_from_excel,
import_trainers_from_excel_for_training,
sync_students_from_t2l_excel,
)
@staff_member_required
def coursesessions_trainers_import(request):
return handle_import(
request,
"Die Durchführungen und Trainer wurden erfolgreich importiert!",
import_training_and_trainer,
)
def import_training_and_trainer(excel_file: str):
import_course_sessions_from_excel(
excel_file,
)
import_trainers_from_excel_for_training(excel_file)
@staff_member_required
def coursesessions_students_import(request):
return handle_import(
request,
"Die Teilnehmer wurden erflogreich importiert!",
import_students_from_excel,
)
@staff_member_required
def t2l_sync(request):
return handle_import(
request,
"Die Daten wurden erflogreich synchronisiert!",
sync_students_from_t2l_excel,
)
def handle_import(request, success_msg: str, importer: Callable[[str], None]):
if request.method == "POST" and request.FILES["excel_file"]:
excel_file = request.FILES["excel_file"]
try:
importer(excel_file)
except Exception as e:
return render(
# it is a "power" feature, so we will output the traceback on error
request,
"admin/importer/error.html",
{"error": traceback.format_exc()},
)
messages.info(request, success_msg)
return redirect("admin:index")

29
server/vbv_lernwelt/templates/admin/index.html
View File

@ -5,12 +5,33 @@
{% include "admin/app_list.html" with app_list=app_list show_changelinks=True %}
<div class="content">
<h1>Export üK</h1>
<h2>Edoniq Teilnehmer</h2>
<h1>üK</h1>
<h1>Excel Import üK</h1>
<h2>Durchführungen und Trainer</h2>
<form method="post" enctype="multipart/form-data" action="/server/importer/coursesession-trainer-import/">
{% csrf_token %}
<input type="file" name="excel_file" accept=".xlsx, .xls">
<input type="submit" value="Durchführungen und Trainer importieren">
</form>
<h2>Teilnehmer</h2>
<form method="post" enctype="multipart/form-data" action="/server/importer/coursesession-students-import/">
{% csrf_token %}
<input type="file" name="excel_file" accept=".xlsx, .xls">
<input type="submit" value="Teilnehmer importieren">
</form>
<h2>Sync mit T2L Daten</h2>
<form method="post" enctype="multipart/form-data" action="/server/importer/t2l-sync/">
{% csrf_token %}
<input type="file" name="excel_file" accept=".xlsx, .xls">
<input type="submit" value="Teilnehmer mit T2L-Daten Synchronisiern">
</form>
<h2>Export Edoniq Teilnehmer</h2>
<a href="{% url 'edoniq_export_students' %}" class="btn btn-primary">Teilnehmer exportieren</a>
<h2>Reset</h2>
<h1>Reset</h1>
<form action="/api/core/cypressreset/" method="post">
{% csrf_token %}
<button class="btn" name="">Testdaten zurück setzen</button>