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"], ), )
[docs] class SettingsForm(forms.ModelForm):
[docs] class Meta: model = Settings fields = [ "import_file_dir", ]