Source code for ferrosoft.apps.dataimport.forms
from django import forms
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.dataimport.models import (
MappingCollection,
ImportJob,
Settings,
UndefinedSettings,
FormatParameters,
DateFormatType,
ImportJobTemplate,
Processors,
)
from ferrosoft.apps.ferrobase.encodings import SupportedEncodings
[docs]
class ImportJobForm(forms.Form):
collection = forms.ModelChoiceField(
queryset=MappingCollection.objects.order_by("name"),
label=_("Mapping collection"),
widget=forms.HiddenInput(),
)
import_file = forms.FileField(
label=_("Import file"),
widget=forms.FileInput(
attrs={
"accept": "text/csv",
}
),
)
text_encoding = forms.ChoiceField(
choices=lambda: [("", _("Auto-detect"))] + SupportedEncodings.choices(),
required=False,
label=_("Text encoding"),
help_text=_(
"Specify text encoding of import file or use auto-detection (which may fail)"
),
)
field_separator = forms.CharField(
label=_("CSV field separator"),
max_length=1,
)
decimal_separator = forms.CharField(
label=_("Decimal separator"),
max_length=1,
)
date_format_type = forms.ChoiceField(
choices=DateFormatType.choices,
label=_("Date format"),
)
date_format_codes = forms.CharField(
label=_("Date format codes"),
required=False,
)
date_time_format_codes = forms.CharField(
label=_("Date time format codes"),
required=False,
)
processor_before_job = forms.CharField(
label=_("Processor before job"),
required=False,
widget=forms.HiddenInput(),
)
processor_after_job = forms.CharField(
label=_("Processor after job"),
required=False,
widget=forms.HiddenInput(),
)
model_processor = forms.CharField(
label=_("Model processor"),
required=False,
widget=forms.HiddenInput(),
)
def __init__(self, *args, **kwargs):
self.user = kwargs.pop("user")
template = kwargs.pop("template", None)
if isinstance(template, ImportJobTemplate):
initial = {
"collection": template.collection,
"text_encoding": template.text_encoding,
"field_separator": template.field_separator,
"decimal_separator": template.decimal_separator,
"date_format_type": template.date_format,
"date_format_codes": template.date_format_codes,
"date_time_format_codes": template.date_time_format_codes,
"processor_before_job": template.processor_before_job,
"processor_after_job": template.processor_after_job,
"model_processor": template.model_processor,
} | (kwargs.pop("initial", {}) or {})
else:
initial = kwargs.pop("initial", None)
super().__init__(*args, initial=initial, **kwargs)
[docs]
def clean(self):
super().clean()
try:
Settings.get_instance().get_import_file_dir()
except UndefinedSettings:
raise ValidationError("Import file directory is not configured")
[docs]
def save(self):
data = self.cleaned_data
return ImportJob.create(
self.files["import_file"],
self.files["import_file"].name,
data["text_encoding"] or None,
self.user,
data["collection"],
FormatParameters(
data["field_separator"],
data["decimal_separator"],
DateFormatType(data["date_format_type"]),
data["date_format_codes"] or "",
data["date_time_format_codes"] or "",
),
Processors(
data["processor_before_job"],
data["processor_after_job"],
data["model_processor"],
),
)