# (c) cavaliba.com - data - forms.py from django import forms from django.conf import settings from django.utils.translation import gettext as _ from app_data.loader import CSV_SNIFF_LINES, detect_csv_delimiter CSV_SEPARATOR_AUTO = "" CSV_SEPARATOR_CHOICES = ( (CSV_SEPARATOR_AUTO, _("Auto-detect")), (",", ","), (";", ";"), ("|", "|"), ) # enough bytes to hold CSV_SNIFF_LINES lines of any realistic header/row; the # upload is rewound afterwards, so this never costs the import anything CSV_SNIFF_BYTES = 64 * 1024 BLANK_CHOICE = ("", "-----------") class ImportForm(forms.Form): """Unified Import Tool form: file upload plus Common / CSV-specific / Advanced options. force_action/force_schema are only declared on the form when is_admin=True is passed to __init__ - a non-admin's bound instance never has these fields at all, so a crafted POST value for either one can never reach cleaned_data no matter what is submitted.""" file = forms.FileField( widget=forms.ClearableFileInput(attrs={"class": "form-control cv-file-input"}), label=_("File"), help_text=_( "Choose a file (.csv, .yml, .yaml, .json). After Check or Import " "you must re-select the file." ), ) # --- common options --- sync_mode = forms.BooleanField( required=False, initial=True, label=_("Touch last_sync"), help_text=_( "Update last_sync on every written object, same as ?sync=true on /api/load/ and /api/rawfile/." ), # noqa: E501 widget=forms.CheckboxInput(attrs={"class": "form-check-input"}), ) pipeline = forms.ChoiceField( required=False, label=_("Pipeline (optional)"), widget=forms.Select(attrs={"class": "form-select"}), ) first = forms.IntegerField( required=False, min_value=1, label=_("First row"), widget=forms.NumberInput(attrs={"class": "form-control"}), ) last = forms.IntegerField( required=False, min_value=1, label=_("Last row"), widget=forms.NumberInput(attrs={"class": "form-control"}), ) # --- CSV-specific options (ignored server-side for non-.csv uploads) --- encoding = forms.ChoiceField( choices=[(e, e) for e in settings.SUPPORTED_ENCODING], required=False, initial="utf-8", label=_("Encoding"), widget=forms.Select(attrs={"class": "form-select"}), ) separator = forms.ChoiceField( choices=CSV_SEPARATOR_CHOICES, required=False, initial=CSV_SEPARATOR_AUTO, label=_("Delimiter"), help_text=_( "Auto-detect reads the first few lines and keeps the character that " "separates the same number of columns on each - it checks |, then ;, then ," ), widget=forms.Select(attrs={"class": "form-select"}), ) split_multivalue = forms.BooleanField( required=False, initial=True, label=_("Split multi-value fields (^)"), widget=forms.CheckboxInput(attrs={"class": "form-check-input"}), ) # --- p_data_admin only (see __init__); force_schema renders in the CSV # --- section of the template, force_action in the advanced one --- force_action = forms.ChoiceField( required=False, label=_("Force action"), widget=forms.Select(attrs={"class": "form-select"}), ) force_schema = forms.ChoiceField( required=False, label=_("Schema"), help_text=_( "Schema applied to every row - needed when the file carries no schema " "column (a CSV header named _schema or classname)." ), widget=forms.Select(attrs={"class": "form-select"}), ) def __init__(self, *args, pipelines=None, schemas=None, is_admin=False, **kwargs): """pipelines/schemas: Instance/Schema objects from the view (schemas already permission-filtered). is_admin: gates whether force_action/ force_schema are even declared on this form instance.""" super().__init__(*args, **kwargs) # set by _clean_separator() when "Auto-detect" resolved to a delimiter, # so the view can name it in its report self.detected_separator = None self.fields["pipeline"].choices = [BLANK_CHOICE] + [ (p.keyname, p.displayname) for p in (pipelines or []) ] if is_admin: self.fields["force_action"].choices = [BLANK_CHOICE] + [ (a, a) for a in settings.ACTION_LIST ] self.fields["force_schema"].choices = [BLANK_CHOICE] + [ (s.classname, s.displayname) for s in (schemas or []) ] else: del self.fields["force_action"] del self.fields["force_schema"] def clean(self): """An inverted range is rejected; either bound alone is valid (first alone skips the rows before it, last alone stops after it). An auto-detect delimiter is resolved here too, see _clean_separator().""" cleaned_data = super().clean() first = cleaned_data.get("first") last = cleaned_data.get("last") if first and last and last < first: self.add_error("last", _("Last row must not be lower than First row")) self._clean_separator(cleaned_data) return cleaned_data def _clean_separator(self, cleaned_data): """Replace an "Auto-detect" delimiter by the detected one, or add a field error. Resolved here rather than at import time on purpose: a file nothing lines up on has to come back as an error on the field, with the form still filled in, so the user can simply pick a delimiter. Guessing a default instead would import the whole file as a single column and report rows that look fine. Only .csv uploads are concerned - the delimiter is ignored for YAML/JSON, so an undetectable one must not block them. """ if cleaned_data.get("separator") != CSV_SEPARATOR_AUTO: return postfile = cleaned_data.get("file") if not postfile or not postfile.name.lower().endswith(".csv"): return # errors="replace": the candidates are all ASCII, so a wrong encoding # must not be what defeats detection - a real encoding problem is # reported by the parse itself, with a message that names it encoding = cleaned_data.get("encoding") or "utf-8" try: postfile.seek(0) head = postfile.read(CSV_SNIFF_BYTES).decode(encoding, errors="replace") except (OSError, LookupError): head = "" finally: # the view reads the upload again to stage it on disk postfile.seek(0) detected = detect_csv_delimiter(head.splitlines()[:CSV_SNIFF_LINES]) if detected: cleaned_data["separator"] = detected self.detected_separator = detected return self.add_error( "separator", _("Could not detect the delimiter - select one (|, ; or ,)"), )