vbv/server/vbv_lernwelt/importer/services.py

97 lines
3.0 KiB
Python

from typing import Dict, Any
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 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
generation = str(data["Generation"]).strip()
region = data["Region"].strip()
group = data["Klasse"].strip()
import_id = data["ID"].strip()
title = f"{region} {generation} {group}"
cs, _created = CourseSession.objects.get_or_create(
import_id=import_id, 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()
"""
("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),
"""
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