vbv/server/vbv_lernwelt/importer/services.py

103 lines
2.9 KiB
Python

from typing import Dict, Any
from openpyxl.reader.excel import load_workbook
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.models import Course, CourseSession
from vbv_lernwelt.importer.utils import try_parse_datetime
from vbv_lernwelt.learnpath.models import LearningContentAttendanceCourse
def create_or_update_user(
email: str, first_name: str = "", last_name: str = "", sso_id: str = None
):
user = None
if sso_id:
user_qs = User.objects.filter(sso_id=sso_id)
if user_qs.exists():
user = user_qs.first()
if not user:
user_qs = User.objects.filter(email=email)
if user_qs.exists():
user = user_qs.first()
if not user:
# create user
user = User(sso_id=sso_id, email=email, username=email)
user.email = email
user.sso_id = sso_id
user.username = email
user.first_name = first_name
user.last_name = last_name
user.set_unusable_password()
user.save()
return user
def import_course_sessions_from_excel(course: Course, filename: str):
workbook = load_workbook(filename=filename)
sheet = workbook["Schulungen Durchführung"]
header = [cell.value for cell in sheet[1]]
for row in sheet.iter_rows(min_row=2, values_only=True):
row_with_header = list(zip(header, row))
cs = create_or_update_course_session(
course, dict(row_with_header), circles=["Fahrzeug"]
)
def create_or_update_course_session(course: Course, data: Dict[str, Any], circles=None):
"""
:param data: the following keys are required to process the data: Generation, Region, Klasse
:return:
"""
if circles is None:
circles = []
# TODO: validation
group = data["Klasse"].strip()
import_id = data["ID"].strip()
generation = str(data["Generation"]).strip()
region = data["Region"].strip()
title = f"{region} {generation} {group}"
cs, _created = CourseSession.objects.get_or_create(
import_id=import_id, group=group, course=course
)
cs.additional_json_data["import_data"] = data
cs.save()
cs.title = title
cs.generation = generation
cs.region = region
cs.group = group
cs.import_id = import_id
cs.save()
for circle in circles:
attendance_course_lp_qs = LearningContentAttendanceCourse.objects.filter(
slug=f"{course.slug}-lp-circle-{circle.lower()}-lc-präsenzkurs-{circle.lower()}"
)
if attendance_course_lp_qs.exists():
cs.attendance_courses.append(
{
"learningContentId": attendance_course_lp_qs.first().id,
"start": try_parse_datetime(data[f"{circle} Start"])[1].isoformat(),
"end": try_parse_datetime(data[f"{circle} Ende"])[1].isoformat(),
"location": data[f"{circle} Raum"],
"trainer": "",
}
)
cs.save()
return cs