Skip to content
Snippets Groups Projects
Verified Commit 41ec87fc authored by Nik | Klampfradler's avatar Nik | Klampfradler
Browse files

[Reformat} Run isort and black

parent d3503372
No related branches found
No related tags found
No related merge requests found
import pkg_resources import pkg_resources
try: try:
__version__ = pkg_resources.get_distribution('BiscuIT-App-SchILD-NRW').version __version__ = pkg_resources.get_distribution("BiscuIT-App-SchILD-NRW").version
except Exception: except Exception:
__version__ = 'unknown' __version__ = "unknown"
default_app_config = 'biscuit.apps.schild_nrw.apps.SchILDNRWConfig' default_app_config = "biscuit.apps.schild_nrw.apps.SchILDNRWConfig"
...@@ -2,5 +2,5 @@ from biscuit.core.util.apps import AppConfig ...@@ -2,5 +2,5 @@ from biscuit.core.util.apps import AppConfig
class SchILDNRWConfig(AppConfig): class SchILDNRWConfig(AppConfig):
name = 'biscuit.apps.schild_nrw' name = "biscuit.apps.schild_nrw"
verbose_name = 'BiscuIT - SchILD-NRW interface' verbose_name = "BiscuIT - SchILD-NRW interface"
...@@ -3,6 +3,6 @@ from django.utils.translation import ugettext_lazy as _ ...@@ -3,6 +3,6 @@ from django.utils.translation import ugettext_lazy as _
class SchILDNRWUploadForm(forms.Form): class SchILDNRWUploadForm(forms.Form):
teachers_csv = forms.FileField(label=_('CSV export of teachers')) teachers_csv = forms.FileField(label=_("CSV export of teachers"))
students_csv = forms.FileField(label=_('CSV export of students')) students_csv = forms.FileField(label=_("CSV export of students"))
guardians_csv = forms.FileField(label=_('CSV export of guardians/parents')) guardians_csv = forms.FileField(label=_("CSV export of guardians/parents"))
...@@ -6,16 +6,19 @@ from biscuit.apps.schild_nrw.util import schild_import_csv ...@@ -6,16 +6,19 @@ from biscuit.apps.schild_nrw.util import schild_import_csv
class Command(BaseCommand): class Command(BaseCommand):
def add_arguments(self, parser): def add_arguments(self, parser):
parser.add_argument('teachers_csv_path', help=_( parser.add_argument(
'Path to CSV file with exported teachers')) "teachers_csv_path", help=_("Path to CSV file with exported teachers")
parser.add_argument('students_csv_path', help=_( )
'Path to CSV file with exported students')) parser.add_argument(
parser.add_argument('guardians_csv_path', help=_( "students_csv_path", help=_("Path to CSV file with exported students")
'Path to CSV file with exported guardians')) )
parser.add_argument(
"guardians_csv_path", help=_("Path to CSV file with exported guardians")
)
def handle(self, *args, **options): def handle(self, *args, **options):
teachers_csv = open(options['teachers_csv_path'], 'rb') teachers_csv = open(options["teachers_csv_path"], "rb")
students_csv = open(options['students_csv_path'], 'rb') students_csv = open(options["students_csv_path"], "rb")
guardians_csv = open(options['guardians_csv_path'], 'rb') guardians_csv = open(options["guardians_csv_path"], "rb")
schild_import_csv(None, teachers_csv, students_csv, guardians_csv) schild_import_csv(None, teachers_csv, students_csv, guardians_csv)
from django.utils.translation import ugettext_lazy as _ from django.utils.translation import ugettext_lazy as _
MENUS = { MENUS = {
'DATA_MANAGEMENT_MENU': [ "DATA_MANAGEMENT_MENU": [
{ {
'name': _('SchILD-NRW import'), "name": _("SchILD-NRW import"),
'url': 'schild_import', "url": "schild_import",
'validators': ['menu_generator.validators.is_authenticated', 'menu_generator.validators.is_superuser', 'biscuit.core.util.core_helpers.has_person'], "validators": [
"menu_generator.validators.is_authenticated",
"menu_generator.validators.is_superuser",
"biscuit.core.util.core_helpers.has_person",
],
} }
] ]
} }
...@@ -2,7 +2,6 @@ from django.urls import path ...@@ -2,7 +2,6 @@ from django.urls import path
from . import views from . import views
urlpatterns = [ urlpatterns = [
path('import', views.schild_import, name='schild_import'), path("import", views.schild_import, name="schild_import"),
] ]
...@@ -10,7 +10,6 @@ import phonenumbers ...@@ -10,7 +10,6 @@ import phonenumbers
from biscuit.core.models import Person from biscuit.core.models import Person
from biscuit.core.util import messages from biscuit.core.util import messages
SCHILD_STATE_ACTIVE = (True, 2) SCHILD_STATE_ACTIVE = (True, 2)
...@@ -19,15 +18,32 @@ def is_active(person_row: dict) -> bool: ...@@ -19,15 +18,32 @@ def is_active(person_row: dict) -> bool:
at different attributes. at different attributes.
""" """
if 'is_active' in person_row: if "is_active" in person_row:
return person_row['is_active'] in SCHILD_STATE_ACTIVE return person_row["is_active"] in SCHILD_STATE_ACTIVE
return True return True
def schild_import_csv_single(request: HttpRequest, csv: Union[BinaryIO, str], cols: Dict[str, Any], converters: Dict[str, Callable[[Optional[str]], Any]]) -> None: def schild_import_csv_single(
persons = pandas.read_csv(csv, sep=';', names=cols.keys(), dtype=cols, usecols=lambda k: not k.startswith('_'), keep_default_na=False, request: HttpRequest,
converters=converters, parse_dates=['date_of_birth'], quotechar='"', encoding='utf-8-sig', true_values=['+', 'Ja'], false_values=['-', 'Nein']) csv: Union[BinaryIO, str],
cols: Dict[str, Any],
converters: Dict[str, Callable[[Optional[str]], Any]],
) -> None:
persons = pandas.read_csv(
csv,
sep=";",
names=cols.keys(),
dtype=cols,
usecols=lambda k: not k.startswith("_"),
keep_default_na=False,
converters=converters,
parse_dates=["date_of_birth"],
quotechar='"',
encoding="utf-8-sig",
true_values=["+", "Ja"],
false_values=["-", "Nein"],
)
# Clean up invalid date values # Clean up invalid date values
persons.date_of_birth = persons.date_of_birth.astype(object) persons.date_of_birth = persons.date_of_birth.astype(object)
...@@ -39,19 +55,33 @@ def schild_import_csv_single(request: HttpRequest, csv: Union[BinaryIO, str], co ...@@ -39,19 +55,33 @@ def schild_import_csv_single(request: HttpRequest, csv: Union[BinaryIO, str], co
for person_row in persons.transpose().to_dict().values(): for person_row in persons.transpose().to_dict().values():
# Fill the is_active field from other fields if necessary # Fill the is_active field from other fields if necessary
person_row['is_active'] = is_active(person_row) person_row["is_active"] = is_active(person_row)
if person_row['is_active']: if person_row["is_active"]:
try: try:
person, created = Person.objects.update_or_create( person, created = Person.objects.update_or_create(
import_ref=person_row['import_ref'], defaults=person_row) import_ref=person_row["import_ref"], defaults=person_row
)
except (ValueError, phonenumbers.NumberParseException) as err: except (ValueError, phonenumbers.NumberParseException) as err:
messages.error(request, _( messages.error(
'Failed to import person %s' % ('%s, %s' % (person_row['last_name'], person_row['first_name']))) + ': %s' % err, fail_silently=True) request,
_(
"Failed to import person %s"
% (
"%s, %s"
% (person_row["last_name"], person_row["first_name"])
)
)
+ ": %s" % err,
fail_silently=True,
)
all_ok = False all_ok = False
# Ensure that newly set primary group is also in member_of # Ensure that newly set primary group is also in member_of
if person.primary_group and person.primary_group not in person.member_of.all(): if (
person.primary_group
and person.primary_group not in person.member_of.all()
):
person.member_of.add(person.primary_group) person.member_of.add(person.primary_group)
person.save() person.save()
...@@ -59,56 +89,80 @@ def schild_import_csv_single(request: HttpRequest, csv: Union[BinaryIO, str], co ...@@ -59,56 +89,80 @@ def schild_import_csv_single(request: HttpRequest, csv: Union[BinaryIO, str], co
created_count += 1 created_count += 1
else: else:
# Store import refs to deactivate later # Store import refs to deactivate later
inactive_refs.append(person_row['import_ref']) inactive_refs.append(person_row["import_ref"])
# Deactivate all persons that existed but are now inactive # Deactivate all persons that existed but are now inactive
if inactive_refs: if inactive_refs:
affected = Person.objects.filter( affected = Person.objects.filter(
import_ref__in=inactive_refs, import_ref__in=inactive_refs, is_active=True
is_active=True ).update(is_active=False)
).update(
is_active=False
)
if affected: if affected:
messages.warning(request, _('%d existing persons were deactivated.') % affected) messages.warning(
request, _("%d existing persons were deactivated.") % affected
)
if created_count: if created_count:
messages.success(request, _('%d persons were newly created.') % created_count) messages.success(request, _("%d persons were newly created.") % created_count)
if all_ok: if all_ok:
messages.success(request, _( messages.success(request, _("All persons were imported successfully."))
'All persons were imported successfully.'))
else: else:
messages.warning(request, _( messages.warning(request, _("Some persons failed to be imported."))
'Some persons failed to be imported.'))
def schild_import_csv(
def schild_import_csv(request: HttpRequest, teachers_csv: Union[BinaryIO, str], students_csv: Union[BinaryIO, str], guardians_csv: Union[BinaryIO, str]) -> None: request: HttpRequest,
csv_converters = {'phone_number': lambda v: phonenumbers.parse(v, 'DE') if v else '', teachers_csv: Union[BinaryIO, str],
'mobile_number': lambda v: phonenumbers.parse(v, 'DE') if v else '', students_csv: Union[BinaryIO, str],
'sex': lambda v: 'f' if v == 'w' else v} guardians_csv: Union[BinaryIO, str],
) -> None:
teachers_csv_cols = OrderedDict([('import_ref', str), ('email', str), ('_email_business', str), csv_converters = {
('date_of_birth', str), ('sex', "phone_number": lambda v: phonenumbers.parse(v, "DE") if v else "",
str), ('short_name', str), "mobile_number": lambda v: phonenumbers.parse(v, "DE") if v else "",
('last_name', str), ('first_name', "sex": lambda v: "f" if v == "w" else v,
str), ('street', str), }
('postal_code', str), ('place',
str), ('phone_number', str), teachers_csv_cols = OrderedDict(
('mobile_number', str), ('is_active', 'bool')]) [
("import_ref", str),
schild_import_csv_single( ("email", str),
request, teachers_csv, teachers_csv_cols, csv_converters) ("_email_business", str),
("date_of_birth", str),
students_csv_cols = OrderedDict([('import_ref', str), ('_internal_id', int), ('primary_group_short_name', str), ("sex", str),
('last_name', str), ('first_name', ("short_name", str),
str), ('additional_name', str), ("last_name", str),
('date_of_birth', str), ('email', ("first_name", str),
str), ('_email_business', str), ("street", str),
('sex', str), ('street', ("postal_code", str),
str), ('housenumber', str), ("place", str),
('postal_code', str), ('place', str), ('phone_number', str), ('is_active', int)]) ("phone_number", str),
("mobile_number", str),
schild_import_csv_single( ("is_active", "bool"),
request, students_csv, students_csv_cols, csv_converters) ]
)
schild_import_csv_single(request, teachers_csv, teachers_csv_cols, csv_converters)
students_csv_cols = OrderedDict(
[
("import_ref", str),
("_internal_id", int),
("primary_group_short_name", str),
("last_name", str),
("first_name", str),
("additional_name", str),
("date_of_birth", str),
("email", str),
("_email_business", str),
("sex", str),
("street", str),
("housenumber", str),
("postal_code", str),
("place", str),
("phone_number", str),
("is_active", int),
]
)
schild_import_csv_single(request, students_csv, students_csv_cols, csv_converters)
...@@ -2,11 +2,11 @@ from django.contrib.auth.decorators import login_required ...@@ -2,11 +2,11 @@ from django.contrib.auth.decorators import login_required
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse
from django.shortcuts import render from django.shortcuts import render
from biscuit.core.decorators import admin_required
from .forms import SchILDNRWUploadForm from .forms import SchILDNRWUploadForm
from .util import schild_import_csv from .util import schild_import_csv
from biscuit.core.decorators import admin_required
@login_required @login_required
@admin_required @admin_required
...@@ -15,13 +15,17 @@ def schild_import(request: HttpRequest) -> HttpResponse: ...@@ -15,13 +15,17 @@ def schild_import(request: HttpRequest) -> HttpResponse:
upload_form = SchILDNRWUploadForm() upload_form = SchILDNRWUploadForm()
if request.method == 'POST': if request.method == "POST":
upload_form = SchILDNRWUploadForm(request.POST, request.FILES) upload_form = SchILDNRWUploadForm(request.POST, request.FILES)
if upload_form.is_valid(): if upload_form.is_valid():
schild_import_csv( schild_import_csv(
request, request.FILES['teachers_csv'], request.FILES['students_csv'], request.FILES['guardians_csv']) request,
request.FILES["teachers_csv"],
request.FILES["students_csv"],
request.FILES["guardians_csv"],
)
context['upload_form'] = upload_form context["upload_form"] = upload_form
return render(request, 'schild_nrw/schild_import.html', context) return render(request, "schild_nrw/schild_import.html", context)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment